Old-hardware training through emulated GPU logic

.gitignore

@@ -0,0 +1,9 @@
+__pycache__/
+*.pyc
+*.pt
+!daisychain/verified/weights/*.pt
+*.egg-info/
+.venv/
+.DS_Store
+status.json
+daisychain_model.pt

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Dean Byrne (Quazim0t0) / DaisyChainAI
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,144 @@
+# 🌼 DaisyChain — Old Hardware Training Pipeline
+
+> **In plain terms:** DaisyChain lets you use **old / spare machines** to train
+> neural networks. The training runs through **emulated GPU logic** — verified
+> INT8 units (GUDA-style) that stand in for a GPU's math — so machines *without*
+> a modern GPU can still do the work. Chain several together and they train one
+> shared model as a cluster.
+> Before you rely on it, see what it **can't** do → [Limitations](docs/LIMITS.md).
+
+**Use the hardware you already have to train.** Each machine runs the emulated
+GPU logic (verified INT8 units — multiply / requantize / ReLU) to compute the
+model, and DaisyChain pools the machines data-parallel: device selection,
+capacity-weighted sharding, gradient sync, a P2P setup, and a live dashboard.
+Two ways to run — **Docker** or **Python**.
+
+> Built by **DaisyChainAI**. Point it at your model + data and it trains across
+> whatever old machines you have, through the emulated GPU logic.
+
+---
+
+## ⚠️ Read this first
+DaisyChain is for **small models on spare hardware**. It **pools compute, not
+memory** (the model must fit on one node), scaling is **sublinear**, and it is
+**not** a substitute for a real GPU on real models. Full envelope in
+**[docs/LIMITS.md](docs/LIMITS.md)** — please read it before relying on it.
+
+---
+
+## Quick start
+
+### Docker (most reliable — one command)
+```bash
+docker compose -f docker/docker-compose.yml up --build
+# open http://localhost:8080
+```
+Brings up a 3-node demo cluster + dashboard on one machine.
+
+### Python (real machines)
+On every machine (`pip install daisychain` or `pip install -e .`):
+```bash
+export MASTER_ADDR=100.101.102.10   # coordinator IP (Tailscale 100.x recommended)
+export MASTER_PORT=29560
+export WORLD_SIZE=3
+export RANK=0                        # 1, 2, ... on the others
+export GLOO_SOCKET_IFNAME=tailscale0 # your mesh / LAN NIC
+daisychain-train
+```
+
+### Windows helper
+```bat
+scripts\setup.bat
+```
+An interactive menu: Docker, Python node, or just install deps.
+
+Full walkthrough: **[docs/QUICKSTART.md](docs/QUICKSTART.md)**.
+
+---
+
+## How it works
+
+Each machine runs the **same** command; they form a cluster and train one shared
+model. Two things happen:
+
+1. **The compute runs through the emulated GPU logic.** By default the model is
+   built from `VerifiedLinear` layers, so every forward multiply / requantize /
+   ReLU is done by the **bundled verified INT8 units** (`daisychain/verified/`)
+   — the emulated GPU math. Rank 0 prints **cluster-wide unit-invocation counts**
+   so you can see the emulated logic doing the work.
+2. **The machines are pooled data-parallel.** Each node trains on its own shard;
+   gradients are capacity-weighted and combined into the exact full-batch
+   gradient, so replicas stay **bit-identical**. Faster machines automatically
+   take a bigger share.
+
+```
+  old machine A ─┐
+  old machine B ─┼─►  each runs the emulated GPU logic on its shard  ─►  one model
+  old machine C ─┘        (gradients combined across the cluster)
+```
+
+## Bring your own model
+DaisyChain trains any **Task** (`build_model` / `sample` / `loss`). Copy
+`examples/my_task_template.py`, set `DAISY_TASK=your_module:YourTask`. Use
+`VerifiedLinear` (see `daisychain/verified_task.py`) to run your model's compute
+through the emulated units. See **[docs/CUSTOM_TASK.md](docs/CUSTOM_TASK.md)**.
+
+## Plain-float alternative
+If you'd rather skip the emulated units and just train with normal float math on
+each machine, set `DAISY_TASK=daisychain.example_task:ExampleTask`. Same cluster,
+same pooling — the model math just runs as ordinary float instead of through the
+verified units.
+
+## The dashboard
+`daisychain-dashboard` (or the Docker service) serves a Tailwind page at
+`:8080` — readiness banner, P2P connectivity scan, pooled cores/RAM + capacity
+plan (per-node device, weight, batch), and live training loss.
+
+## Networking
+Use **Tailscale** for a P2P mesh so machines on different networks get stable
+IPs on one interface — **[docs/TAILSCALE.md](docs/TAILSCALE.md)**.
+
+---
+
+## Layout
+```
+daisychain/cluster.py        capacity-weighted CPU/GPU data-parallel trainer
+daisychain/task.py           the Task interface + loader
+daisychain/train.py          entry point (daisychain-train)
+daisychain/example_task.py   default runnable task (plain float)
+daisychain/verified/         bundled trained N/N units + VerifiedLinear (train through them)
+daisychain/verified_task.py  example task whose forward runs on the verified units
+daisychain/dashboard/        agent + P2P scanner + Tailwind server
+docker/                      Dockerfile, dashboard image, compose (demo cluster)
+scripts/setup.bat / setup.sh interactive setup helpers
+config/                      nodes + cluster env examples
+examples/my_task_template.py starting point for your own model
+docs/                        QUICKSTART, LIMITS, CUSTOM_TASK, TAILSCALE
+```
+
+## Install
+```bash
+pip install torch numpy psutil
+pip install -e .          # exposes: daisychain-train, daisychain-agent, daisychain-dashboard
+```
+
+Requires Python ≥ 3.9, PyTorch ≥ 2.0. Multi-node is reliable on **Linux/macOS**;
+on **Windows use Docker/WSL** (see LIMITS).
+
+---
+
+**License:** MIT · **Author:** Dean Byrne (Quazim0t0) · **Org:** DaisyChainAI
+
+## Citation
+
+```bibtex
+@misc{byrne2026daisychain,
+  title        = {DaisyChain: An Old Hardware Training Pipeline},
+  author       = {Byrne, Dean (Quazim0t0)},
+  year         = {2026},
+  howpublished = {\url{https://huggingface.co/DaisyChainAI/old-hw-train}},
+  note         = {Chain spare/old machines into a data-parallel training cluster}
+}
+```
+
+**Dean Byrne (Quazim0t0)** · 2026

config/cluster.example.env

@@ -0,0 +1,18 @@
+# Copy to cluster.env and set on EACH machine (change only RANK per machine).
+# Then: source it and run `daisychain-train`  (or use the scripts/ helpers).
+
+MASTER_ADDR=100.101.102.10      # the coordinator's IP (Tailscale 100.x recommended)
+MASTER_PORT=29560
+WORLD_SIZE=3
+RANK=0                          # 0 on the coordinator, 1 / 2 / ... on the others
+GLOO_SOCKET_IFNAME=tailscale0   # the NIC to use (tailscale0, or eth0 / your LAN NIC)
+USE_LIBUV=0
+
+# Task + training
+DAISY_TASK=daisychain.example_task:ExampleTask   # swap for "your_module:YourTask"
+DAISY_STEPS=300
+DAISY_LR=0.05
+DAISY_OPTIMIZER=sgd
+DAISY_BASE_BATCH=32
+DAISY_STATUS_FILE=status.json
+DAISY_SAVE=daisychain_model.pt

config/nodes.example.json

@@ -0,0 +1,5 @@
+[
+  { "name": "rank0 (coordinator)", "host": "rank0", "agent_port": 8900, "gloo_port": 29560 },
+  { "name": "rank1",               "host": "rank1", "agent_port": 8900, "gloo_port": 29560 },
+  { "name": "rank2",               "host": "rank2", "agent_port": 8900, "gloo_port": 29560 }
+]

daisychain/__init__.py

@@ -0,0 +1,13 @@
+"""DaisyChain — Old Hardware Training Pipeline.
+
+Chain spare/old machines into a data-parallel training cluster in minutes.
+Bring your own model+data as a Task; DaisyChain handles device selection,
+capacity-weighted sharding, gradient sync, and a live dashboard.
+"""
+from .cluster import (DaisyCluster, survey_node, cluster_plan, pick_device,
+                      capacity_score, configure_cpu)
+from .task import Task, load_task
+
+__version__ = "0.1.0"
+__all__ = ["DaisyCluster", "survey_node", "cluster_plan", "pick_device",
+           "capacity_score", "configure_cpu", "Task", "load_task"]

daisychain/cluster.py

@@ -0,0 +1,211 @@
+"""DaisyChain cluster core — data-parallel CPU/GPU training across spare machines.
+
+Design: distribute the *parallel* axis (the batch) across nodes; keep each node's
+work local. Every node holds a full model replica and trains on its shard; a
+capacity-weighted gradient all-reduce combines them into the exact full-batch
+gradient, so replicas stay bit-identical.
+
+  - Each node uses ~90% of its cores (and its GPU if it has one).
+  - Capacity is MEASURED (matmuls/sec) so a strong node auto-takes a bigger
+    batch. Gradients are reduced on CPU copies, so CPU and GPU nodes mix.
+
+Pools compute, not memory: the model must fit on one node. Honest limits are in
+docs/LIMITS.md.
+"""
+from __future__ import annotations
+
+import json
+import os
+import socket
+import time
+
+import torch
+import torch.distributed as dist
+
+
+# ---------------------------------------------------------------- resources ---
+def configure_cpu(fraction: float = 0.9) -> int:
+    cores = os.cpu_count() or 1
+    n = max(1, int(round(cores * fraction)))
+    torch.set_num_threads(n)
+    return n
+
+
+def pick_device() -> "torch.device":
+    if os.environ.get("DAISY_FORCE_CPU") == "1":
+        return torch.device("cpu")
+    return torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+
+def _gpu_info():
+    if not torch.cuda.is_available():
+        return None
+    p = torch.cuda.get_device_properties(0)
+    return {"name": p.name, "vram_gb": round(p.total_memory / 1e9, 1),
+            "capability": f"{p.major}.{p.minor}"}
+
+
+def _available_ram_gb():
+    try:
+        import psutil
+        return round(psutil.virtual_memory().available / 1e9, 1)
+    except Exception:
+        return None
+
+
+def capacity_score(device=None, secs: float = 0.3) -> float:
+    """Measured throughput: fixed matmuls/sec on the local device. Self-calibrating
+    (a GPU scores far higher), so capacity weighting hands it a bigger batch."""
+    dev = device or pick_device()
+    try:
+        a = torch.randn(512, 512, device=dev)
+        b = torch.randn(512, 512, device=dev)
+        _ = a @ b
+        if dev.type == "cuda":
+            torch.cuda.synchronize()
+        t0, it = time.time(), 0
+        while time.time() - t0 < secs:
+            a = a @ b
+            it += 1
+        if dev.type == "cuda":
+            torch.cuda.synchronize()
+        return it / (time.time() - t0)
+    except Exception:
+        return float(os.cpu_count() or 1)
+
+
+def survey_node(cpu_fraction: float = 0.9, measure: bool = True) -> dict:
+    cores = int(os.environ.get("DAISY_CORES", os.cpu_count() or 1))
+    dev = pick_device()
+    gpu = _gpu_info() if dev.type == "cuda" else None
+    if "DAISY_CAPACITY" in os.environ:
+        cap = float(os.environ["DAISY_CAPACITY"])
+    elif measure:
+        cap = capacity_score(dev)
+    else:
+        cap = float(cores)
+    return {"host": socket.gethostname(), "cores": cores,
+            "threads": max(1, int(round(cores * cpu_fraction))),
+            "ram_gb": _available_ram_gb(), "device": dev.type,
+            "gpu": gpu, "capacity": cap}
+
+
+# ------------------------------------------------------------------ cluster ---
+def init_cluster(backend: str = "gloo"):
+    os.environ.setdefault("USE_LIBUV", "0")
+    if not dist.is_initialized():
+        dist.init_process_group(backend=backend)
+    return dist.get_rank(), dist.get_world_size()
+
+
+def cluster_plan(cpu_fraction: float = 0.9, base_batch: int = 32) -> dict:
+    rank, world = dist.get_rank(), dist.get_world_size()
+    me = survey_node(cpu_fraction)
+    gathered = [None] * world
+    dist.all_gather_object(gathered, me)
+
+    caps = [float(g.get("capacity") or g["cores"]) for g in gathered]
+    total_cap = sum(caps) or world
+    global_batch = base_batch * world
+    batches = [max(1, round(global_batch * c / total_cap)) for c in caps]
+    total_batch = sum(batches)
+    weights = [b / total_batch for b in batches]
+    rams = [g["ram_gb"] for g in gathered if g["ram_gb"] is not None]
+    return {"rank": rank, "world": world, "nodes": gathered,
+            "weights": weights, "local_batches": batches,
+            "my_weight": weights[rank], "my_local_batch": batches[rank],
+            "total_cores": sum(g["cores"] for g in gathered),
+            "total_ram_gb": (sum(rams) if rams else None),
+            "global_batch": sum(batches),
+            "devices": [g.get("device", "cpu") for g in gathered],
+            "capacities": [round(c, 1) for c in caps],
+            "gpus": [g.get("gpu") for g in gathered]}
+
+
+@torch.no_grad()
+def broadcast_params(model, src: int = 0):
+    for p in model.parameters():
+        cpu = p.data.detach().to("cpu")
+        dist.broadcast(cpu, src=src)
+        p.data.copy_(cpu.to(p.data.device))
+
+
+@torch.no_grad()
+def capacity_weighted_allreduce_grads(model, weight: float):
+    """Σ_i w_i g_i with w_i = n_i/Σn_j == the true full-batch mean gradient.
+    Reduced on CPU copies so mixed CPU/GPU nodes interoperate over gloo."""
+    for p in model.parameters():
+        if p.grad is None:
+            p.grad = torch.zeros_like(p.data)
+        g = p.grad.detach().to("cpu").mul_(weight)
+        dist.all_reduce(g, op=dist.ReduceOp.SUM)
+        p.grad.copy_(g.to(p.grad.device))
+
+
+class DaisyCluster:
+    """One node's handle on the cluster. Same code runs on every machine."""
+
+    def __init__(self, cpu_fraction: float = 0.9, base_batch: int = 32):
+        self.threads = configure_cpu(cpu_fraction)
+        self.device = pick_device()
+        self.rank, self.world = init_cluster()
+        self.plan = cluster_plan(cpu_fraction, base_batch)
+
+    def is_master(self):
+        return self.rank == 0
+
+    def _write_status(self, path, **kw):
+        payload = {"rank": self.rank, "world": self.world,
+                   "plan": {"total_cores": self.plan["total_cores"],
+                            "total_ram_gb": self.plan["total_ram_gb"],
+                            "weights": self.plan["weights"],
+                            "devices": self.plan["devices"],
+                            "local_batches": self.plan["local_batches"]}, **kw}
+        try:
+            with open(path, "w") as f:
+                json.dump(payload, f)
+        except Exception:
+            pass
+
+    def fit(self, model, task, steps=500, lr=1e-2, optimizer="sgd",
+            status_path=None, step_delay=0.0):
+        """Train `model` on `task` (build_model already called). task.sample(n)
+        draws this node's shard; task.loss(model, X, y) returns a scalar."""
+        model.to(self.device)
+        broadcast_params(model)
+        if optimizer == "adam":
+            opt = torch.optim.Adam(model.parameters(), lr=lr)
+        else:
+            opt = torch.optim.SGD(model.parameters(), lr=lr)
+        w, nb = self.plan["my_weight"], self.plan["my_local_batch"]
+        for s in range(steps):
+            X, y = task.sample(nb)
+            X, y = X.to(self.device), y.to(self.device)
+            opt.zero_grad(set_to_none=False)
+            loss = task.loss(model, X, y)
+            loss.backward()
+            capacity_weighted_allreduce_grads(model, w)
+            opt.step()
+            if step_delay:
+                time.sleep(step_delay)
+            if s % max(1, steps // 20) == 0 or s == steps - 1:
+                lt = loss.detach().to("cpu").clone()
+                dist.all_reduce(lt, op=dist.ReduceOp.SUM)
+                if self.is_master():
+                    cl = lt.item() / self.world
+                    print(f"  step {s:5d}  cluster-avg loss {cl:.6f}", flush=True)
+                    if status_path:
+                        self._write_status(status_path, step=s, total_steps=steps,
+                                            cluster_avg_loss=cl, done=(s == steps - 1))
+        return model
+
+    def replica_diff(self, model):
+        vec = torch.cat([p.data.reshape(-1).to("cpu") for p in model.parameters()])
+        bucket = [torch.zeros_like(vec) for _ in range(self.world)]
+        dist.all_gather(bucket, vec)
+        return max((bucket[i] - bucket[0]).abs().max().item() for i in range(self.world))
+
+    def shutdown(self):
+        if dist.is_initialized():
+            dist.barrier()
+            dist.destroy_process_group()

daisychain/dashboard/agent.py

@@ -0,0 +1,60 @@
+"""Per-node agent (CLI: `daisychain-agent`). Serves health/resources/status so
+the dashboard can scan this machine. Pure stdlib."""
+import json
+import os
+import socket
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+try:
+    from daisychain.cluster import survey_node
+except Exception:
+    def survey_node(cpu_fraction=0.9):
+        cores = int(os.environ.get("DAISY_CORES", os.cpu_count() or 1))
+        return {"host": socket.gethostname(), "cores": cores,
+                "threads": max(1, int(cores * cpu_fraction)),
+                "ram_gb": None, "device": "cpu", "gpu": None, "capacity": cores}
+
+RANK = int(os.environ.get("RANK", "0"))
+STATUS_FILE = os.environ.get("DAISY_STATUS_FILE", "status.json")
+PORT = int(os.environ.get("DAISY_AGENT_PORT", "8900"))
+
+
+def _status():
+    try:
+        with open(STATUS_FILE) as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+class Handler(BaseHTTPRequestHandler):
+    def _send(self, obj, code=200):
+        body = json.dumps(obj).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        if self.path.startswith("/health"):
+            self._send({"ok": True, "rank": RANK, "host": socket.gethostname()})
+        elif self.path.startswith("/resources"):
+            r = survey_node(); r["rank"] = RANK; self._send(r)
+        elif self.path.startswith("/status"):
+            self._send(_status())
+        else:
+            self._send({"error": "not found"}, 404)
+
+    def log_message(self, *a):
+        pass
+
+
+def main():
+    print(f"[agent] rank {RANK} on :{PORT}", flush=True)
+    ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    main()

daisychain/dashboard/scanner.py

@@ -0,0 +1,85 @@
+"""P2P cluster scanner: probe each node's agent, measure latency, gather
+resources + live status, compute the capacity plan and readiness verdict."""
+import json
+import socket
+import time
+import urllib.request
+
+
+def _get(url, timeout=2.0):
+    with urllib.request.urlopen(url, timeout=timeout) as r:
+        return json.load(r)
+
+
+def _latency_ms(host, port, timeout=2.0):
+    t = time.time()
+    try:
+        with socket.create_connection((host, int(port)), timeout=timeout):
+            return (time.time() - t) * 1000.0
+    except Exception:
+        return None
+
+
+def scan_node(node):
+    host, ap = node["host"], node.get("agent_port", 8900)
+    base = f"http://{host}:{ap}"
+    out = {"name": node["name"], "host": host, "reachable": False,
+           "latency_ms": None, "resources": None, "status": None,
+           "gloo_port_open": None}
+    lat = _latency_ms(host, ap)
+    out["latency_ms"] = round(lat, 1) if lat is not None else None
+    if lat is None:
+        return out
+    try:
+        h = _get(f"{base}/health")
+        out["reachable"] = bool(h.get("ok"))
+        out["rank"] = h.get("rank")
+        out["resources"] = _get(f"{base}/resources")
+        out["status"] = _get(f"{base}/status")
+    except Exception as e:
+        out["error"] = str(e)
+    gp = node.get("gloo_port")
+    if gp:
+        out["gloo_port_open"] = _latency_ms(host, gp) is not None
+    return out
+
+
+def capacity_plan(scans, base_batch=32):
+    live = [n for n in scans if n.get("resources")]
+    caps = [float(n["resources"].get("capacity") or n["resources"].get("cores", 1))
+            for n in live]
+    total = sum(caps) or 1
+    world = len(live)
+    gb = base_batch * max(1, world)
+    batches = [max(1, round(gb * c / total)) for c in caps]
+    tb = sum(batches) or 1
+    weights = [b / tb for b in batches]
+    rams = [n["resources"].get("ram_gb") for n in live
+            if n["resources"].get("ram_gb") is not None]
+    return {"world": world, "total_cores": sum(n["resources"].get("cores", 0) for n in live),
+            "total_ram_gb": round(sum(rams), 1) if rams else None,
+            "per_node": [{"name": n["name"], "device": n["resources"].get("device", "cpu"),
+                          "cores": n["resources"].get("cores"),
+                          "gpu": (n["resources"].get("gpu") or {}).get("name"),
+                          "capacity": round(c, 1), "weight": round(w, 3), "batch": b,
+                          "ram_gb": n["resources"].get("ram_gb")}
+                         for n, c, w, b in zip(live, caps, weights, batches)],
+            "global_batch": sum(batches)}
+
+
+def scan_cluster(nodes, expected_world=None, base_batch=32):
+    scans = [scan_node(n) for n in nodes]
+    plan = capacity_plan(scans, base_batch)
+    reachable = [s for s in scans if s["reachable"]]
+    want = expected_world if expected_world is not None else len(nodes)
+    ready = (len(reachable) == len(nodes) == want)
+    train = None
+    for s in scans:
+        st = s.get("status") or {}
+        if st.get("rank") == 0 and st:
+            train = st
+            break
+    return {"nodes": scans, "plan": plan, "ready": ready,
+            "reachable": len(reachable), "total": len(nodes),
+            "expected_world": want, "training": train,
+            "scanned_at": time.strftime("%H:%M:%S")}

daisychain/dashboard/server.py

@@ -0,0 +1,127 @@
+"""DaisyChain dashboard (CLI: `daisychain-dashboard`). Tailwind-styled page with
+a readiness banner, P2P connectivity scan, pooled resource + capacity plan, and
+live training status. Config via DAISY_NODES_FILE. Serves on :8080."""
+import json
+import os
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+try:
+    from daisychain.dashboard.scanner import scan_cluster
+except Exception:
+    from scanner import scan_cluster  # running from the folder directly
+
+NODES_FILE = os.environ.get("DAISY_NODES_FILE", "config/nodes.example.json")
+PORT = int(os.environ.get("DAISY_DASH_PORT", "8080"))
+EXPECTED_WORLD = os.environ.get("DAISY_EXPECTED_WORLD")
+BASE_BATCH = int(os.environ.get("DAISY_BASE_BATCH", "32"))
+
+
+def load_nodes():
+    with open(NODES_FILE) as f:
+        return json.load(f)
+
+
+def _chip(ok, yes="OK", no="DOWN"):
+    cls = ("bg-emerald-500/20 text-emerald-600 dark:text-emerald-400" if ok
+           else "bg-rose-500/20 text-rose-600 dark:text-rose-400")
+    return f'<span class="px-2 py-0.5 rounded text-xs font-semibold {cls}">{yes if ok else no}</span>'
+
+
+def render(d):
+    ready = d["ready"]
+    banner = ("bg-emerald-500", "✓ CLUSTER READY — all nodes connected") if ready \
+        else ("bg-rose-500", f"✗ NOT READY — {d['reachable']}/{d['total']} nodes reachable")
+    rows = ""
+    for n in d["nodes"]:
+        lat = f'{n["latency_ms"]} ms' if n["latency_ms"] is not None else "—"
+        res = n.get("resources") or {}
+        dev = res.get("device", "—")
+        gpu = (res.get("gpu") or {}).get("name", "")
+        devlabel = f"{dev}" + (f" ({gpu})" if gpu else "")
+        rows += f"""<tr class="border-b border-slate-100 dark:border-slate-800">
+          <td class="py-2 px-3 font-medium">{n['name']}</td>
+          <td class="py-2 px-3 text-slate-500">{n['host']}</td>
+          <td class="py-2 px-3">{_chip(n['reachable'],'reachable','unreachable')}</td>
+          <td class="py-2 px-3">{devlabel}</td>
+          <td class="py-2 px-3 tabular-nums">{lat}</td></tr>"""
+    plan = d["plan"]
+    pr = ""
+    for p in plan["per_node"]:
+        dv = p["device"] + (f" ({p['gpu']})" if p.get("gpu") else "")
+        ram = f'{p["ram_gb"]} GB' if p.get("ram_gb") is not None else "—"
+        pr += f"""<tr class="border-b border-slate-100 dark:border-slate-800">
+          <td class="py-2 px-3 font-medium">{p['name']}</td>
+          <td class="py-2 px-3">{dv}</td>
+          <td class="py-2 px-3 tabular-nums">{ram}</td>
+          <td class="py-2 px-3 tabular-nums">{p['capacity']}</td>
+          <td class="py-2 px-3 tabular-nums">{p['weight']}</td>
+          <td class="py-2 px-3 tabular-nums">{p['batch']}</td></tr>"""
+    tot_ram = f'{plan["total_ram_gb"]} GB' if plan["total_ram_gb"] is not None else "—"
+    t = d["training"]
+    if t:
+        step, total = t.get("step", 0), t.get("total_steps", 1)
+        pct = int(100 * (step + 1) / max(1, total)); loss = t.get("cluster_avg_loss")
+        badge = _chip(True, "DONE") if t.get("done") else '<span class="text-xs text-sky-500 animate-pulse">training…</span>'
+        train = f"""<div class="flex items-center justify-between mb-2">
+          <span class="text-sm text-slate-500">step {step} / {total}</span>{badge}</div>
+        <div class="w-full h-3 rounded-full bg-slate-200 dark:bg-slate-700 overflow-hidden">
+          <div class="h-3 bg-sky-500" style="width:{pct}%"></div></div>
+        <p class="mt-3 text-2xl font-semibold tabular-nums">{loss:.5f}
+          <span class="text-sm font-normal text-slate-500">cluster-avg loss</span></p>"""
+    else:
+        train = '<p class="text-slate-400 text-sm">No active run detected (waiting for rank 0 status)…</p>'
+    return f"""<!doctype html><html><head><meta charset="utf-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1">
+    <meta http-equiv="refresh" content="3"><title>DaisyChain — Cluster</title>
+    <script src="https://cdn.tailwindcss.com"></script></head>
+    <body class="bg-slate-50 dark:bg-slate-900 text-slate-800 dark:text-slate-100 min-h-screen">
+    <div class="max-w-5xl mx-auto p-6 space-y-6">
+      <header class="flex items-center justify-between">
+        <div><h1 class="text-2xl font-bold">\U0001f33c DaisyChain</h1>
+        <p class="text-sm text-slate-500">Old Hardware Training Pipeline · scanned {d['scanned_at']}</p></div>
+        <span class="text-xs text-slate-400">auto-refresh 3s</span></header>
+      <div class="{banner[0]} text-white rounded-xl px-5 py-4 font-semibold text-lg shadow">{banner[1]}</div>
+      <div class="grid md:grid-cols-3 gap-4">
+        <div class="rounded-xl border border-slate-200 dark:border-slate-700 p-4"><p class="text-xs uppercase tracking-wide text-slate-400">Nodes</p><p class="text-3xl font-bold tabular-nums">{d['reachable']}/{d['total']}</p></div>
+        <div class="rounded-xl border border-slate-200 dark:border-slate-700 p-4"><p class="text-xs uppercase tracking-wide text-slate-400">Total cores</p><p class="text-3xl font-bold tabular-nums">{plan['total_cores']}</p></div>
+        <div class="rounded-xl border border-slate-200 dark:border-slate-700 p-4"><p class="text-xs uppercase tracking-wide text-slate-400">Total RAM</p><p class="text-3xl font-bold tabular-nums">{tot_ram}</p></div></div>
+      <section class="rounded-xl border border-slate-200 dark:border-slate-700 overflow-hidden">
+        <h2 class="px-4 py-3 font-semibold border-b border-slate-100 dark:border-slate-800">Connectivity scan</h2>
+        <table class="w-full text-sm"><thead class="text-left text-slate-400"><tr>
+          <th class="py-2 px-3">Node</th><th class="py-2 px-3">Host</th><th class="py-2 px-3">Agent</th>
+          <th class="py-2 px-3">Device</th><th class="py-2 px-3">Latency</th></tr></thead><tbody>{rows}</tbody></table></section>
+      <div class="grid md:grid-cols-2 gap-6">
+        <section class="rounded-xl border border-slate-200 dark:border-slate-700 overflow-hidden">
+          <h2 class="px-4 py-3 font-semibold border-b border-slate-100 dark:border-slate-800">Capacity plan · global batch {plan['global_batch']}</h2>
+          <table class="w-full text-sm"><thead class="text-left text-slate-400"><tr>
+            <th class="py-2 px-3">Node</th><th class="py-2 px-3">Device</th><th class="py-2 px-3">RAM</th>
+            <th class="py-2 px-3">Capacity</th><th class="py-2 px-3">Weight</th><th class="py-2 px-3">Batch</th></tr></thead><tbody>{pr}</tbody></table></section>
+        <section class="rounded-xl border border-slate-200 dark:border-slate-700 p-4">
+          <h2 class="font-semibold mb-3">Live training</h2>{train}</section></div>
+      <footer class="text-center text-xs text-slate-400 pt-2">DaisyChainAI · pools compute, not memory · small models on spare hardware</footer>
+    </div></body></html>"""
+
+
+class Handler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        data = scan_cluster(load_nodes(),
+                            int(EXPECTED_WORLD) if EXPECTED_WORLD else None, BASE_BATCH)
+        if self.path.startswith("/api"):
+            body = json.dumps(data).encode(); ctype = "application/json"
+        else:
+            body = render(data).encode(); ctype = "text/html; charset=utf-8"
+        self.send_response(200); self.send_header("Content-Type", ctype)
+        self.send_header("Content-Length", str(len(body))); self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *a):
+        pass
+
+
+def main():
+    print(f"[dashboard] :{PORT} nodes={NODES_FILE}", flush=True)
+    ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    main()

daisychain/example_task.py

@@ -0,0 +1,26 @@
+"""The default example task: fit a small MLP to a synthetic function.
+
+It exists so `daisychain-train` runs out of the box and you can confirm the
+cluster works end to end. Replace it with your own task (see docs/CUSTOM_TASK.md)
+-- copy this file, change build_model / sample / loss, and set DAISY_TASK.
+"""
+import torch
+import torch.nn as nn
+
+
+class ExampleTask:
+    def __init__(self):
+        # fixed target so every node's shard is consistent
+        g = torch.Generator().manual_seed(1234)
+        self.W = torch.randn(8, 1, generator=g)
+
+    def build_model(self):
+        torch.manual_seed(0)                 # identical init on every node
+        return nn.Sequential(nn.Linear(8, 32), nn.ReLU(), nn.Linear(32, 1))
+
+    def sample(self, n):
+        X = torch.randn(n, 8)
+        return X, X @ self.W + 0.05 * torch.randn(n, 1)
+
+    def loss(self, model, X, y):
+        return nn.functional.mse_loss(model(X), y)

daisychain/task.py

@@ -0,0 +1,33 @@
+"""The Task interface — bring your own model + data.
+
+A DaisyChain task is any object with three methods:
+
+    build_model() -> torch.nn.Module          # the model to train (identical on every node)
+    sample(n)     -> (X, y)                    # draw n training samples (this node's shard)
+    loss(model, X, y) -> scalar tensor         # mean loss over the batch
+
+Point DaisyChain at your task with  DAISY_TASK="my_module:MyTask"  (or --task).
+An example lives in examples/example_task.py. Keep build_model deterministic
+(seed it) so every node starts from the same weights.
+"""
+from __future__ import annotations
+
+import importlib
+from typing import Protocol, Tuple
+
+import torch
+
+
+class Task(Protocol):
+    def build_model(self) -> torch.nn.Module: ...
+    def sample(self, n: int) -> Tuple[torch.Tensor, torch.Tensor]: ...
+    def loss(self, model: torch.nn.Module, X: torch.Tensor, y: torch.Tensor) -> torch.Tensor: ...
+
+
+def load_task(spec: str):
+    """spec = 'package.module:ClassName' -> instantiated task object."""
+    if ":" not in spec:
+        raise ValueError(f"task spec must be 'module:Class', got {spec!r}")
+    mod_name, cls_name = spec.split(":", 1)
+    mod = importlib.import_module(mod_name)
+    return getattr(mod, cls_name)()

daisychain/train.py

@@ -0,0 +1,85 @@
+"""DaisyChain training entry point (CLI: `daisychain-train`).
+
+Reads cluster settings from env (set on each machine, changing only RANK):
+
+  MASTER_ADDR, MASTER_PORT, WORLD_SIZE, RANK   -- standard torch.distributed
+  GLOO_SOCKET_IFNAME                            -- the NIC to use (e.g. tailscale0)
+  DAISY_TASK        = "module:Class"            -- your task (default: example)
+  DAISY_STEPS       = 300
+  DAISY_LR          = 0.05
+  DAISY_OPTIMIZER   = sgd | adam
+  DAISY_BASE_BATCH  = 32
+  DAISY_STATUS_FILE = status.json               -- rank 0 writes live status here
+  DAISY_STEP_SLEEP  = 0                          -- demo pacing
+  DAISY_SAVE        = daisychain_model.pt        -- rank 0 saves here
+"""
+import os
+
+import torch
+
+from .cluster import DaisyCluster
+from .task import load_task
+
+
+def _report_verified_counts(cluster):
+    """All-reduce verified-unit invocation counts across nodes (if any fired)."""
+    try:
+        from .verified import instrument
+        import torch.distributed as dist
+        counts = instrument.report()
+        if not counts:
+            return
+        keys = sorted(counts)
+        t = torch.tensor([counts[k] for k in keys], dtype=torch.float64)
+        dist.all_reduce(t, op=dist.ReduceOp.SUM)
+        if cluster.is_master():
+            print("[verified] CLUSTER-WIDE verified-unit invocations (trained through them):")
+            for k, v in zip(keys, t.tolist()):
+                print(f"[verified]   {k:34s} {int(v):,}")
+    except Exception:
+        pass
+
+
+def main():
+    # Default: train THROUGH the emulated GPU logic (verified INT8 units).
+    # Set DAISY_TASK=daisychain.example_task:ExampleTask for a plain-float run.
+    task_spec = os.environ.get("DAISY_TASK", "daisychain.verified_task:VerifiedTask")
+    task = load_task(task_spec)
+
+    cluster = DaisyCluster(
+        cpu_fraction=float(os.environ.get("DAISY_CPU_FRACTION", "0.9")),
+        base_batch=int(os.environ.get("DAISY_BASE_BATCH", "32")),
+    )
+
+    if cluster.is_master():
+        p = cluster.plan
+        print(f"[daisychain] task={task_spec}")
+        print(f"[plan] world={p['world']} devices={p['devices']} "
+              f"total_cores={p['total_cores']} total_ram_gb={p['total_ram_gb']}")
+        print(f"[plan] capacities={p['capacities']} weights={[round(w,3) for w in p['weights']]}")
+        print(f"[plan] local_batches={p['local_batches']} global_batch={p['global_batch']}")
+
+    model = task.build_model()
+    cluster.fit(
+        model, task,
+        steps=int(os.environ.get("DAISY_STEPS", "300")),
+        lr=float(os.environ.get("DAISY_LR", "0.05")),
+        optimizer=os.environ.get("DAISY_OPTIMIZER", "sgd"),
+        status_path=os.environ.get("DAISY_STATUS_FILE", "status.json"),
+        step_delay=float(os.environ.get("DAISY_STEP_SLEEP", "0")),
+    )
+
+    # if the task trained THROUGH the verified units, report cluster-wide counts
+    _report_verified_counts(cluster)
+
+    diff = cluster.replica_diff(model)
+    if cluster.is_master():
+        print(f"[check] replica max param diff across nodes: {diff:.2e}")
+        save = os.environ.get("DAISY_SAVE", "daisychain_model.pt")
+        torch.save({"state_dict": model.state_dict()}, save)
+        print(f"[save] {save}")
+    cluster.shutdown()
+
+
+if __name__ == "__main__":
+    main()

daisychain/verified/__init__.py

@@ -0,0 +1,14 @@
+"""Verified units — the trained, N/N bit-exact INT8 building blocks bundled with
+DaisyChain (from the neural-gpu-verify project). Build models with VerifiedLinear
+and their FORWARD compute runs through these verified units (multiply / requant /
+ReLU), materialized as lookup tables for practical speed (fast=True).
+
+These are pre-trained; users do NOT train new units — they train THROUGH them.
+"""
+from .qat import VerifiedLinear, load_units, build_luts
+from .mul8 import NeuralMul8
+from .ops import NeuralReLU8, NeuralRequant16
+from . import instrument
+
+__all__ = ["VerifiedLinear", "load_units", "build_luts",
+           "NeuralMul8", "NeuralReLU8", "NeuralRequant16", "instrument"]

daisychain/verified/backends.py

@@ -0,0 +1,61 @@
+"""Compute backends for the bridge -- pure Python, no external toolchain.
+
+  NumpyBackend  -- int32 GEMM on CPU SIMD (numpy). Real throughput, CPU-capped.
+  NeuralBackend -- GEMM where every multiply is the N/N-verified neural unit and
+                   accumulation is exact integer sum. The compute path is itself
+                   a net: bit-exact, but functional (slow), not a speed path.
+
+Whatever backend runs, the bridge's self-certify gate proves its output bit-exact
+to the verified op before any result is trusted.
+
+(An optional Go/GUDA backend lives in backends_go.py; it is not imported here and
+not required -- this package stands alone without Go.)
+"""
+from __future__ import annotations
+
+import numpy as np
+
+from .kernel import gemm_int8
+
+
+class NumpyBackend:
+    name = "numpy-int8"
+
+    def available(self) -> bool:
+        return True
+
+    def gemm(self, A: np.ndarray, B: np.ndarray) -> np.ndarray:
+        return gemm_int8(A, B)
+
+
+class NeuralBackend:
+    """GEMM computed entirely by the verified neural multiply (+ exact sum).
+
+    Pass a trained NeuralMul8 (or anything with `mul_array(a, b)`), e.g. loaded
+    from mul8.pt. Products come from the net; accumulation is exact int64.
+    """
+    name = "neural-mul"
+
+    def __init__(self, mul):
+        self.mul = mul
+
+    def available(self) -> bool:
+        return hasattr(self.mul, "mul_array")
+
+    def gemm(self, A: np.ndarray, B: np.ndarray) -> np.ndarray:
+        A = np.asarray(A).astype(np.int64)
+        B = np.asarray(B).astype(np.int64)
+        m, k = A.shape
+        _, n = B.shape
+        # all (A[i,t], B[t,j]) pairs -> one batched neural multiply -> sum over k
+        Ai = np.broadcast_to(A[:, None, :], (m, n, k))          # (m,n,k)
+        Bj = np.broadcast_to(B.T[None, :, :], (m, n, k))        # (m,n,k)
+        prod = self.mul.mul_array(Ai.reshape(-1), Bj.reshape(-1)).reshape(m, n, k)
+        return prod.sum(axis=2).astype(np.int64)
+
+
+def pick_backend(neural=None):
+    """NeuralBackend(mul) if a multiplier is given, else the numpy throughput path."""
+    if neural is not None:
+        return NeuralBackend(neural)
+    return NumpyBackend()

daisychain/verified/common.py

@@ -0,0 +1,63 @@
+"""Shared helpers for verified neural units.
+
+Same discipline as the neural-aarch64 / neural-photonic / neural-ddr units:
+a small MLP is trained until it is *bit-identical to a golden reference over its
+entire finite input domain* (N/N verification).
+"""
+from __future__ import annotations
+
+import torch
+import torch.nn as nn
+
+
+def bits_of(v: int, n: int) -> torch.Tensor:
+    """LSB-first bit vector of length n."""
+    return torch.tensor([(v >> i) & 1 for i in range(n)], dtype=torch.float32)
+
+
+def int_of(bits: torch.Tensor) -> int:
+    """Inverse of bits_of: LSB-first bit vector -> int."""
+    v = 0
+    for i, b in enumerate(bits.tolist()):
+        if b >= 0.5:
+            v |= (1 << i)
+    return v
+
+
+def pm(bits: torch.Tensor) -> torch.Tensor:
+    """Map {0,1} bits to {-1,+1} for a friendlier input scale."""
+    return bits * 2.0 - 1.0
+
+
+def mlp(inp: int, out: int, h: int = 128, layers: int = 3) -> nn.Sequential:
+    mods: list[nn.Module] = [nn.Linear(inp, h), nn.ReLU()]
+    for _ in range(layers - 1):
+        mods += [nn.Linear(h, h), nn.ReLU()]
+    mods += [nn.Linear(h, out)]
+    return nn.Sequential(*mods)
+
+
+@torch.no_grad()
+def verify(net: nn.Module, X: torch.Tensor, Ybits: torch.Tensor) -> tuple[int, int]:
+    """Return (n_correct, n_total) over the full enumerated domain."""
+    net.eval()
+    pred = (net(X) > 0).float()
+    ok = (pred == Ybits).all(dim=1).sum().item()
+    return int(ok), X.shape[0]
+
+
+def train(net: nn.Module, X: torch.Tensor, Ybits: torch.Tensor,
+          steps: int = 4000, lr: float = 2e-3, tag: str = "") -> nn.Module:
+    opt = torch.optim.Adam(net.parameters(), lr=lr)
+    lossf = nn.BCEWithLogitsLoss()
+    net.train()
+    for s in range(steps):
+        opt.zero_grad()
+        loss = lossf(net(X), Ybits)
+        loss.backward()
+        opt.step()
+        if tag and (s % 1000 == 0 or s == steps - 1):
+            ok, tot = verify(net, X, Ybits)
+            net.train()
+            print(f"  [{tag}] step {s:5d}  loss {loss.item():.5f}  verify {ok}/{tot}")
+    return net

daisychain/verified/instrument.py

@@ -0,0 +1,34 @@
+"""Invocation counters for the verified units.
+
+Turn it on and every verified unit records how many times its neural forward
+actually ran (and how many scalar ops it produced). This is the evidence that a
+training/inference pass genuinely computed *through* the verified GUDA logic --
+not around it.
+"""
+from collections import Counter
+
+COUNTS = Counter()
+_ENABLED = False
+
+
+def enable():
+    global _ENABLED
+    _ENABLED = True
+
+
+def disable():
+    global _ENABLED
+    _ENABLED = False
+
+
+def reset():
+    COUNTS.clear()
+
+
+def bump(key: str, n: int = 1):
+    if _ENABLED:
+        COUNTS[key] += int(n)
+
+
+def report() -> dict:
+    return dict(COUNTS)

daisychain/verified/kernel.py

@@ -0,0 +1,22 @@
+"""The GUDA-role path: a fast, vectorized INT8 GEMM on real CPU silicon.
+
+This is where the *throughput* comes from -- real CPU SIMD (via numpy/torch
+int32 accumulation), exactly GUDA's job. It is NOT a neural net and NOT
+GPU-fast; its ceiling is the CPU. The neural verified op certifies that this
+kernel is bit-faithful to the reference (verify_kernel.py) -- turning GUDA-style
+"IEEE parity within tolerance" into "provably equal on this finite integer op".
+"""
+from __future__ import annotations
+
+import numpy as np
+
+
+def gemm_int8(A: np.ndarray, B: np.ndarray) -> np.ndarray:
+    """Signed INT8 GEMM with exact int32 accumulation (tensor-core semantics)."""
+    a = A.astype(np.int32)
+    b = B.astype(np.int32)
+    return a @ b            # exact integer matmul, no overflow for our sizes
+
+
+def random_int8(shape, rng) -> np.ndarray:
+    return rng.integers(-128, 128, size=shape, dtype=np.int16).astype(np.int8)

daisychain/verified/lut.py

@@ -0,0 +1,55 @@
+"""Native-speed deployment of the verified units, without losing the guarantee.
+
+A verified unit is a finite function. Its neural net is only needed to *prove*
+correctness (N/N). For SPEED you materialize the proven function as a lookup
+table -- run the net once over its whole (small) domain -- then every later call
+is an array index at native memory speed. Because the net is N/N-verified, the
+LUT is bit-identical to the net, which is bit-identical to the true op. So:
+
+    neural forward  ==  LUT  ==  native integer op      (all bit-exact)
+
+That's the "freeze the mesh to its matrix" lesson: verify once (slow, offline),
+deploy native (fast). The LUTs are tiny: mul 256x256, requant 65536, relu 256.
+"""
+from __future__ import annotations
+
+import numpy as np
+
+
+def build_mul8_lut(mul) -> np.ndarray:
+    """[256,256] signed-product table, indexed by unsigned bytes. Net runs once."""
+    a = np.repeat(np.arange(256), 256)
+    b = np.tile(np.arange(256), 256)
+    prod = mul.mul_array(a, b)                     # verified neural multiply, ONCE
+    return prod.reshape(256, 256).astype(np.int64)
+
+
+def build_requant16_lut(rq) -> np.ndarray:
+    """[65536] int16->int8 table, indexed by acc & 0xFFFF."""
+    return rq.requant_array(np.arange(65536)).astype(np.int64)
+
+
+def build_relu8_lut(relu) -> np.ndarray:
+    """[256] int8 ReLU table, indexed by unsigned byte."""
+    return relu.relu_array(np.arange(256)).astype(np.int64)
+
+
+class LUTBackend:
+    """GEMM via the materialized (verified) multiply table + integer accumulate."""
+    name = "lut"
+
+    def __init__(self, mul):
+        self.mul_lut = build_mul8_lut(mul)
+
+    def available(self):
+        return True
+
+    def gemm(self, A: np.ndarray, B: np.ndarray) -> np.ndarray:
+        au = (A.astype(np.int64) & 0xFF)
+        bu = (B.astype(np.int64) & 0xFF)
+        # products via table lookup, then sum over the contraction axis
+        prod = self.mul_lut[au[:, None, :], bu.T[None, :, :]]   # (m, n, k)
+        from . import instrument
+        instrument.bump("VerifiedMul(LUT).gemms", 1)
+        instrument.bump("VerifiedMul(LUT).products", prod.size)
+        return prod.sum(axis=2)

daisychain/verified/mul8.py

@@ -0,0 +1,132 @@
+"""Verified neural INT8 multiplier -- the atom of a GPU tensor core / VNNI lane.
+
+A monolithic MLP cannot learn a bit-exact 8x8 multiply (the high-order product
+bits are too nonlinear). So -- exactly like the byte-slice / ripple trick the
+other projects use for hard functions -- we shrink the *verified atom* to a
+4-bit unsigned multiply and compose everything exactly:
+
+  atom:   NeuralMul4  -- unsigned 4x4 -> 8, domain 16*16 = 256, verified N/N.
+  8x8:    a*b = ah*bh<<8 + (ah*bl + al*bh)<<4 + al*bl   (unsigned), exact.
+  signed: a_s = a_u - 256*a7 ;  Baugh-Wooley correction, exact integer glue.
+
+Only the 4x4 multiply is neural (and N/N-proven); the shifts, adds and sign
+correction are exact composition. The parallel *throughput* of thousands of such
+lanes is NOT here -- that needs real silicon (kernel.py, the GUDA-role path).
+"""
+from __future__ import annotations
+
+import numpy as np
+import torch
+
+from .common import bits_of, int_of, pm, mlp, verify, train
+
+
+class NeuralMul4:
+    """N/N-verified unsigned 4x4 -> 8 multiplier (the finite atom)."""
+
+    def __init__(self, h: int = 128, layers: int = 3):
+        self.net = mlp(8, 8, h=h, layers=layers)
+
+    def dataset(self) -> tuple[torch.Tensor, torch.Tensor]:
+        X, Y = [], []
+        for a in range(16):
+            ab = bits_of(a, 4)
+            for b in range(16):
+                X.append(pm(torch.cat([ab, bits_of(b, 4)])))
+                Y.append(bits_of(a * b, 8))
+        return torch.stack(X), torch.stack(Y)
+
+    def fit(self, steps: int = 4000, lr: float = 2e-3, tag: str = "mul4"):
+        X, Y = self.dataset()
+        train(self.net, X, Y, steps=steps, lr=lr, tag=tag)
+        return self
+
+    def verify(self) -> tuple[int, int]:
+        X, Y = self.dataset()
+        return verify(self.net, X, Y)
+
+    @torch.no_grad()
+    def mul(self, a: int, b: int) -> int:
+        self.net.eval()
+        x = pm(torch.cat([bits_of(a & 0xF, 4), bits_of(b & 0xF, 4)])).unsqueeze(0)
+        return int_of((self.net(x)[0] > 0).float())
+
+    @torch.no_grad()
+    def mul_array(self, a: np.ndarray, b: np.ndarray) -> np.ndarray:
+        """Batched unsigned 4x4 -> 8 over arrays (one neural forward for all)."""
+        self.net.eval()
+        a = (np.asarray(a).astype(np.int64) & 0xF)
+        b = (np.asarray(b).astype(np.int64) & 0xF)
+        idx = np.arange(4)
+        bits_a = (a[:, None] >> idx) & 1
+        bits_b = (b[:, None] >> idx) & 1
+        x = np.concatenate([bits_a, bits_b], axis=1).astype(np.float32) * 2.0 - 1.0
+        out = (self.net(torch.from_numpy(x)) > 0).to(torch.int64).numpy()
+        from . import instrument
+        instrument.bump("NeuralMul4.forward_calls", 1)
+        instrument.bump("NeuralMul4.products", a.shape[0])
+        return (out * (1 << np.arange(8))).sum(axis=1)
+
+
+class NeuralMul8:
+    """Signed 8x8 -> 16 multiply, composed exactly from the verified 4x4 atom."""
+
+    def __init__(self, h: int = 128, layers: int = 3):
+        self.atom = NeuralMul4(h=h, layers=layers)
+
+    def fit(self, steps: int = 4000, lr: float = 2e-3, tag: str = "mul4"):
+        self.atom.fit(steps=steps, lr=lr, tag=tag)
+        return self
+
+    def verify_atom(self) -> tuple[int, int]:
+        return self.atom.verify()
+
+    def _umul8(self, a_u: int, b_u: int) -> int:
+        """Unsigned 8x8 -> 16 via four verified 4x4 sub-products (exact glue)."""
+        al, ah = a_u & 0xF, (a_u >> 4) & 0xF
+        bl, bh = b_u & 0xF, (b_u >> 4) & 0xF
+        ll = self.atom.mul(al, bl)
+        lh = self.atom.mul(al, bh)
+        hl = self.atom.mul(ah, bl)
+        hh = self.atom.mul(ah, bh)
+        return ll + ((lh + hl) << 4) + (hh << 8)
+
+    def mul(self, a: int, b: int) -> int:
+        """Signed product via the verified atom + exact Baugh-Wooley correction."""
+        a_u, b_u = a & 0xFF, b & 0xFF
+        a7, b7 = (a_u >> 7) & 1, (b_u >> 7) & 1
+        prod = self._umul8(a_u, b_u) - (a7 * b_u << 8) - (b7 * a_u << 8) + (a7 * b7 << 16)
+        prod &= 0xFFFF                      # 16-bit two's complement
+        return prod - 65536 if prod >= 32768 else prod
+
+    @torch.no_grad()
+    def mul_array(self, a: np.ndarray, b: np.ndarray) -> np.ndarray:
+        """Batched signed 8x8 -> 16 over arrays, via the verified 4x4 atom.
+
+        Four nibble sub-products are batched into ONE neural forward, then the
+        exact shift/add glue and Baugh-Wooley sign correction are applied in
+        numpy. Result is bit-exact (the atom is N/N-verified)."""
+        a = np.asarray(a).astype(np.int64).ravel()
+        b = np.asarray(b).astype(np.int64).ravel()
+        au, bu = a & 0xFF, b & 0xFF
+        al, ah = au & 0xF, (au >> 4) & 0xF
+        bl, bh = bu & 0xF, (bu >> 4) & 0xF
+        n = au.shape[0]
+        pa = np.concatenate([al, al, ah, ah])
+        pb = np.concatenate([bl, bh, bl, bh])
+        prod = self.atom.mul_array(pa, pb)
+        ll, lh, hl, hh = prod[:n], prod[n:2*n], prod[2*n:3*n], prod[3*n:4*n]
+        u = ll + ((lh + hl) << 4) + (hh << 8)
+        a7, b7 = (au >> 7) & 1, (bu >> 7) & 1
+        p = (u - (a7 * bu << 8) - (b7 * au << 8) + (a7 * b7 << 16)) & 0xFFFF
+        return np.where(p >= 32768, p - 65536, p)
+
+    @torch.no_grad()
+    def verify(self, full: bool = True) -> tuple[int, int]:
+        """Exhaustively check the composed signed multiply over all 65536 inputs."""
+        ok = 0
+        for a in range(-128, 128):
+            for b in range(-128, 128):
+                if self.mul(a, b) == a * b:
+                    ok += 1
+        return ok, 256 * 256

daisychain/verified/mul8_tied.py

@@ -0,0 +1,106 @@
+"""TIED neural multiplier -- one small cell, iterated (a la neural-raytracing).
+
+Multiply is inherently iterative (shift-and-add), so instead of four separate
+4x4 atoms we use a SINGLE weight-tied cell applied across the 8 steps -- exactly
+the raytracing pattern of marching one shared cell rather than stacking many.
+
+The tied cell is the conditional-add step of a shift-add multiplier:
+
+    cell(acc, a, enable) = acc + (a if enable else 0)      # 8+8+1 -> 9 bits
+
+Its domain is 2^8 * 2^8 * 2 = 131072 -- enumerable, so the cell is verified
+BIT-EXACT over its whole domain (N/N). The shift and re-assembly between steps
+are exact wiring, not neural. Looping the one tied cell 8x yields the full
+unsigned 8x8 -> 16 product; signed uses the exact Baugh-Wooley correction.
+
+One verified cell, reused 8 times -- the smallest possible learned core.
+"""
+from __future__ import annotations
+
+import torch
+
+from .common import bits_of, int_of, pm, mlp, verify, train
+
+
+class NeuralMACStep:
+    """N/N-verified tied cell: acc8 + (a8 if enable else 0) -> 9-bit sum."""
+
+    def __init__(self, h: int = 128, layers: int = 3):
+        self.net = mlp(17, 9, h=h, layers=layers)   # acc(8) + a(8) + enable(1) -> 9
+
+    def dataset(self) -> tuple[torch.Tensor, torch.Tensor]:
+        X, Y = [], []
+        for acc in range(256):
+            ab = bits_of(acc, 8)
+            for a in range(256):
+                aa = bits_of(a, 8)
+                for en in (0, 1):
+                    X.append(pm(torch.cat([ab, aa, torch.tensor([float(en)])])))
+                    Y.append(bits_of(acc + (a if en else 0), 9))
+        return torch.stack(X), torch.stack(Y)
+
+    def fit(self, steps: int = 5000, lr: float = 2e-3, tag: str = "macstep"):
+        X, Y = self.dataset()
+        train(self.net, X, Y, steps=steps, lr=lr, tag=tag)
+        return self
+
+    def verify(self) -> tuple[int, int]:
+        X, Y = self.dataset()
+        return verify(self.net, X, Y)
+
+    @torch.no_grad()
+    def step(self, acc: int, a: int, enable: int) -> int:
+        self.net.eval()
+        x = pm(torch.cat([bits_of(acc & 0xFF, 8), bits_of(a & 0xFF, 8),
+                          torch.tensor([float(enable)])])).unsqueeze(0)
+        return int_of((self.net(x)[0] > 0).float())
+
+
+class TiedMul8:
+    """Signed 8x8 -> 16 multiply from ONE tied MAC-step cell, iterated 8x."""
+
+    def __init__(self, h: int = 128, layers: int = 3):
+        self.cell = NeuralMACStep(h=h, layers=layers)
+
+    def fit(self, steps: int = 5000, lr: float = 2e-3, tag: str = "macstep"):
+        self.cell.fit(steps=steps, lr=lr, tag=tag)
+        return self
+
+    def verify_cell(self) -> tuple[int, int]:
+        return self.cell.verify()
+
+    def _umul8(self, a_u: int, b_u: int) -> int:
+        """Unsigned product via the tied shift-add loop (cell reused 8x)."""
+        combined = b_u & 0xFF                     # low byte holds b, high byte acc
+        for _ in range(8):
+            enable = combined & 1                 # current LSB of b
+            hi = (combined >> 8) & 0xFF
+            s = self.cell.step(hi, a_u, enable)   # 9-bit: hi + (a if enable)
+            combined = (combined & 0xFF) | (s << 8)
+            combined >>= 1                        # shift right one place
+        return combined & 0xFFFF
+
+    def mul(self, a: int, b: int) -> int:
+        a_u, b_u = a & 0xFF, b & 0xFF
+        a7, b7 = (a_u >> 7) & 1, (b_u >> 7) & 1
+        prod = self._umul8(a_u, b_u) - (a7 * b_u << 8) - (b7 * a_u << 8) + (a7 * b7 << 16)
+        prod &= 0xFFFF
+        return prod - 65536 if prod >= 32768 else prod
+
+    @torch.no_grad()
+    def verify_unsigned(self) -> tuple[int, int]:
+        ok = 0
+        for a in range(256):
+            for b in range(256):
+                if self._umul8(a, b) == a * b:
+                    ok += 1
+        return ok, 256 * 256
+
+    @torch.no_grad()
+    def verify(self) -> tuple[int, int]:
+        ok = 0
+        for a in range(-128, 128):
+            for b in range(-128, 128):
+                if self.mul(a, b) == a * b:
+                    ok += 1
+        return ok, 256 * 256

daisychain/verified/ops.py

@@ -0,0 +1,111 @@
+"""Verified INT8 activation / requantize units -- the rest of a quantized layer.
+
+A quantized linear layer is:  int8 x int8 -> int32 accumulate -> requantize to
+int8 -> activation. The multiply/accumulate is covered by mul8 + exact int32
+sum. These two units cover the tail, each over a FINITE, enumerable domain so
+they are verified BIT-EXACT (N/N):
+
+  NeuralReLU8      int8 -> int8,  y = max(0, x)                 domain 256
+  NeuralRequant16  int16 -> int8, y = sat_int8(x >> shift)      domain 65536
+
+The int32 -> int16 narrowing that precedes requantize is an EXACT saturating
+clamp (integer wiring, not neural). So a full qlinear is exact/verified through
+every stage.
+"""
+from __future__ import annotations
+
+import numpy as np
+import torch
+
+from .common import bits_of, int_of, pm, mlp, verify, train
+from . import instrument
+
+
+def _s(v: int, bits: int) -> int:
+    """two's-complement raw -> signed."""
+    m = 1 << (bits - 1)
+    return v - (1 << bits) if v >= m else v
+
+
+def sat_int8(x: int) -> int:
+    return -128 if x < -128 else (127 if x > 127 else x)
+
+
+class NeuralReLU8:
+    """N/N-verified INT8 ReLU: y = max(0, x)."""
+
+    def __init__(self, h: int = 64, layers: int = 2):
+        self.net = mlp(8, 8, h=h, layers=layers)
+
+    def dataset(self):
+        X, Y = [], []
+        for b in range(256):
+            X.append(pm(bits_of(b, 8)))
+            Y.append(bits_of(max(0, _s(b, 8)) & 0xFF, 8))
+        return torch.stack(X), torch.stack(Y)
+
+    def fit(self, steps: int = 3000, lr: float = 2e-3, tag: str = "relu8"):
+        X, Y = self.dataset(); train(self.net, X, Y, steps=steps, lr=lr, tag=tag); return self
+
+    def verify(self):
+        X, Y = self.dataset(); return verify(self.net, X, Y)
+
+    @torch.no_grad()
+    def relu(self, x: int) -> int:
+        self.net.eval()
+        raw = int_of((self.net(pm(bits_of(x & 0xFF, 8)).unsqueeze(0))[0] > 0).float())
+        return _s(raw, 8)
+
+    @torch.no_grad()
+    def relu_array(self, arr: np.ndarray) -> np.ndarray:
+        """Batched int8 ReLU over an array (one neural forward)."""
+        self.net.eval()
+        a = np.asarray(arr).astype(np.int64).ravel() & 0xFF
+        bits = ((a[:, None] >> np.arange(8)) & 1).astype(np.float32) * 2.0 - 1.0
+        out = (self.net(torch.from_numpy(bits)) > 0).to(torch.int64).numpy()
+        raw = (out * (1 << np.arange(8))).sum(axis=1)
+        instrument.bump("NeuralReLU8.forward_calls", 1)
+        instrument.bump("NeuralReLU8.elements", a.shape[0])
+        return np.where(raw >= 128, raw - 256, raw).reshape(np.asarray(arr).shape)
+
+
+class NeuralRequant16:
+    """N/N-verified requantize: int16 -> int8, y = sat_int8(x >> shift)."""
+
+    def __init__(self, shift: int = 8, h: int = 256, layers: int = 3):
+        self.shift = int(shift)
+        self.net = mlp(16, 8, h=h, layers=layers)
+
+    def _ref(self, x_signed: int) -> int:
+        return sat_int8(x_signed >> self.shift)      # arithmetic shift, saturate
+
+    def dataset(self):
+        X, Y = [], []
+        for b in range(65536):
+            X.append(pm(bits_of(b, 16)))
+            Y.append(bits_of(self._ref(_s(b, 16)) & 0xFF, 8))
+        return torch.stack(X), torch.stack(Y)
+
+    def fit(self, steps: int = 6000, lr: float = 2e-3, tag: str = "requant16"):
+        X, Y = self.dataset(); train(self.net, X, Y, steps=steps, lr=lr, tag=tag); return self
+
+    def verify(self):
+        X, Y = self.dataset(); return verify(self.net, X, Y)
+
+    @torch.no_grad()
+    def requant(self, x: int) -> int:
+        self.net.eval()
+        raw = int_of((self.net(pm(bits_of(x & 0xFFFF, 16)).unsqueeze(0))[0] > 0).float())
+        return _s(raw, 8)
+
+    @torch.no_grad()
+    def requant_array(self, arr: np.ndarray) -> np.ndarray:
+        """Batched int16->int8 requantize over an array (one neural forward)."""
+        self.net.eval()
+        a = np.asarray(arr).astype(np.int64).ravel() & 0xFFFF
+        bits = ((a[:, None] >> np.arange(16)) & 1).astype(np.float32) * 2.0 - 1.0
+        out = (self.net(torch.from_numpy(bits)) > 0).to(torch.int64).numpy()
+        raw = (out * (1 << np.arange(8))).sum(axis=1)
+        instrument.bump("NeuralRequant16.forward_calls", 1)
+        instrument.bump("NeuralRequant16.elements", a.shape[0])
+        return np.where(raw >= 128, raw - 256, raw).reshape(np.asarray(arr).shape)

daisychain/verified/qat.py

@@ -0,0 +1,116 @@
+"""Quantization-aware training THROUGH the verified units.
+
+This is the piece that was missing: a trainable layer whose FORWARD compute
+actually runs on the verified GUDA logic --
+
+    quantize -> NeuralMul (verified INT8 multiply) GEMM -> NeuralRequant16 ->
+    NeuralReLU8 -> dequantize
+
+-- while the BACKWARD uses a straight-through estimator (the integer path has no
+gradient), so ordinary float weights still learn. With instrument.enable(), each
+unit records how many times its neural forward ran, so a training run leaves
+hard evidence (call counts) that it computed through the units, not around them.
+
+Honest cost: every forward multiply is a neural forward pass -> this is SLOW
+(functional, not fast). It is a correctness/《evidence》demo, not a speed path.
+"""
+from __future__ import annotations
+
+import numpy as np
+import torch
+import torch.nn as nn
+
+from .backends import NeuralBackend
+
+
+class _VerifiedQGEMM(torch.autograd.Function):
+    @staticmethod
+    def forward(ctx, x, w, mul, requant, relu_unit, use_relu, luts):
+        ctx.save_for_backward(x, w)
+        ctx.device = x.device                       # verified units run on CPU;
+        xnp, wnp = x.detach().cpu().numpy(), w.detach().cpu().numpy()
+        sx = max(float(np.abs(xnp).max()) / 127.0, 1e-8)
+        sw = max(float(np.abs(wnp).max()) / 127.0, 1e-8)
+        xq = np.clip(np.round(xnp / sx), -128, 127).astype(np.int8)
+        wq = np.clip(np.round(wnp / sw), -128, 127).astype(np.int8)
+
+        if luts is not None:
+            # FAST path: the verified units, materialized as lookup tables
+            # (bit-identical to the neural forward, ~500x faster).
+            from . import instrument
+            acc = luts["backend"].gemm(xq, wq)         # counts LUT products
+            acc16 = np.clip(acc, -32768, 32767).astype(np.int64)
+            yq = luts["requant"][acc16 & 0xFFFF]
+            instrument.bump("VerifiedRequant16(LUT).elements", acc16.size)
+            if use_relu:
+                yq = luts["relu"][yq & 0xFF]
+                instrument.bump("VerifiedReLU8(LUT).elements", yq.size)
+        else:
+            acc = NeuralBackend(mul).gemm(xq, wq)      # verified multiply fires
+            acc16 = np.clip(acc, -32768, 32767).astype(np.int64)
+            yq = requant.requant_array(acc16)          # requant16 fires
+            if use_relu:
+                yq = relu_unit.relu_array(yq)          # relu8 fires
+        dequant = sx * sw * 256.0                       # undo requant's >>8
+        return torch.from_numpy(yq.astype(np.float32) * dequant).to(ctx.device)
+
+    @staticmethod
+    def backward(ctx, gy):
+        # straight-through: treat the quantized path as y ≈ x @ w
+        x, w = ctx.saved_tensors
+        return gy @ w.t(), x.t() @ gy, None, None, None, None, None
+
+
+def build_luts(mul, requant, relu_unit):
+    """Materialize the verified units as lookup tables (one-time). The result is
+    bit-identical to the neural forward but ~500x faster to run."""
+    from .lut import LUTBackend, build_requant16_lut, build_relu8_lut
+    return {"backend": LUTBackend(mul),
+            "requant": build_requant16_lut(requant),
+            "relu": build_relu8_lut(relu_unit)}
+
+
+class VerifiedLinear(nn.Module):
+    """Linear layer whose forward is computed by the verified units.
+
+    fast=True materializes the units as LUTs (bit-identical, ~500x faster) so
+    verified training is practical; fast=False runs the neural forward (proof).
+    """
+
+    def __init__(self, in_f, out_f, mul, requant, relu_unit, use_relu=True,
+                 fast=False):
+        super().__init__()
+        self.weight = nn.Parameter(torch.randn(in_f, out_f) * 0.3)
+        self.bias = nn.Parameter(torch.zeros(out_f))
+        self.mul, self.requant, self.relu_unit = mul, requant, relu_unit
+        self.use_relu = use_relu
+        self.luts = build_luts(mul, requant, relu_unit) if fast else None
+
+    def forward(self, x):
+        y = _VerifiedQGEMM.apply(x, self.weight, self.mul, self.requant,
+                                 self.relu_unit, self.use_relu, self.luts)
+        return y + self.bias
+
+
+def _weights_dir():
+    import os
+    return os.path.join(os.path.dirname(os.path.abspath(__file__)), "weights")
+
+
+def load_units(mul_pt=None, requant_pt=None, relu_pt=None):
+    """Load the TRAINED, N/N-verified units bundled with DaisyChain."""
+    import os
+    wd = _weights_dir()
+    mul_pt = mul_pt or os.path.join(wd, "mul8.pt")
+    requant_pt = requant_pt or os.path.join(wd, "requant16.pt")
+    relu_pt = relu_pt or os.path.join(wd, "relu8.pt")
+    from .mul8 import NeuralMul8
+    from .ops import NeuralReLU8, NeuralRequant16
+    mul = NeuralMul8()
+    mul.atom.net.load_state_dict(torch.load(mul_pt)["state_dict"]); mul.atom.net.eval()
+    relu = NeuralReLU8()
+    relu.net.load_state_dict(torch.load(relu_pt)["state_dict"]); relu.net.eval()
+    ck = torch.load(requant_pt)
+    rq = NeuralRequant16(shift=ck["shift"])
+    rq.net.load_state_dict(ck["state_dict"]); rq.net.eval()
+    return mul, rq, relu

daisychain/verified/weights/macstep.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:2a0e771445a7bac7295af9156880e33693b6c5cf6ea0f1e24ae2daad3241c005
+size 148608

daisychain/verified/weights/mul8.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:be360811bddcda90e7ffa8e6cb338673ab7c0902c5882f12bc0c89d84b05aef9
+size 143452

daisychain/verified/weights/relu8.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:b6806712b0302c003029b26c658c9a005850116308bc225d0a9a7e26c73e07ba
+size 23284

daisychain/verified/weights/requant16.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:0cca8f64e40b4969395b588d781a4ae49bca0ede8d13576ea03dbe0fe1cb2ea2
+size 555160

daisychain/verified_task.py

@@ -0,0 +1,36 @@
+"""Example task that trains THROUGH the bundled verified units.
+
+Same shape as ExampleTask, but the model is built from VerifiedLinear layers, so
+every forward multiply/requant/ReLU runs on the trained N/N-verified INT8 units
+(materialized as lookup tables for speed). Backward uses a straight-through
+estimator, so ordinary weights still learn.
+
+Run it with:  DAISY_TASK=daisychain.verified_task:VerifiedTask  daisychain-train
+"""
+import torch
+import torch.nn as nn
+
+from .verified import VerifiedLinear, load_units, instrument
+
+
+class VerifiedTask:
+    def __init__(self, fast: bool = True):
+        self.mul, self.rq, self.relu = load_units()   # bundled trained weights
+        instrument.enable()                            # count unit invocations
+        self._fast = fast
+        g = torch.Generator().manual_seed(1234)
+        self.W = torch.randn(8, 1, generator=g)
+
+    def build_model(self):
+        torch.manual_seed(0)
+        return nn.Sequential(
+            VerifiedLinear(8, 8, self.mul, self.rq, self.relu, use_relu=True, fast=self._fast),
+            VerifiedLinear(8, 1, self.mul, self.rq, self.relu, use_relu=False, fast=self._fast),
+        )
+
+    def sample(self, n):
+        X = torch.randn(n, 8)
+        return X, X @ self.W
+
+    def loss(self, model, X, y):
+        return nn.functional.mse_loss(model(X), y)

docker/Dockerfile

@@ -0,0 +1,15 @@
+# DaisyChain node image (CPU torch keeps it lean; GPU nodes use a real machine).
+FROM python:3.11-slim
+RUN pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu \
+ && pip install --no-cache-dir numpy psutil
+WORKDIR /app
+COPY daisychain/ ./daisychain/
+COPY examples/ ./examples/
+COPY config/ ./config/
+COPY docker/node_entrypoint.sh ./node_entrypoint.sh
+RUN chmod +x ./node_entrypoint.sh
+ENV GLOO_SOCKET_IFNAME=eth0
+ENV USE_LIBUV=0
+ENV PYTHONUNBUFFERED=1
+ENV DAISY_STATUS_FILE=/app/status.json
+CMD ["./node_entrypoint.sh"]

docker/dashboard.Dockerfile

@@ -0,0 +1,10 @@
+# DaisyChain dashboard image — pure stdlib, no torch. Runs server.py standalone
+# (its scanner import falls back to a flat import, so no torch package init).
+FROM python:3.11-slim
+WORKDIR /app
+COPY daisychain/dashboard/server.py daisychain/dashboard/scanner.py ./
+COPY config/ ./config/
+ENV PYTHONUNBUFFERED=1
+ENV DAISY_NODES_FILE=config/nodes.example.json
+EXPOSE 8080
+CMD ["python", "server.py"]

docker/docker-compose.yml

@@ -0,0 +1,45 @@
+# DaisyChain demo cluster: 3 node containers + a dashboard on one Docker network.
+# In production each node is a real machine (see docs/TAILSCALE.md); this brings
+# the whole thing up on one box so you can see it work.
+#
+#   docker compose -f docker/docker-compose.yml up --build
+#   open http://localhost:8080
+name: daisychain
+x-node: &node
+  build: { context: .., dockerfile: docker/Dockerfile }
+  image: daisychain:latest
+  environment: &env
+    MASTER_ADDR: rank0
+    MASTER_PORT: "29560"
+    WORLD_SIZE: "3"
+    DAISY_STEPS: "300"
+    DAISY_STEP_SLEEP: "0.1"      # demo pacing so the dashboard shows it live
+    # train through the emulated GPU logic (verified INT8 units) by default
+    DAISY_TASK: "daisychain.verified_task:VerifiedTask"
+  networks: [cluster]
+
+services:
+  rank0:
+    <<: *node
+    environment: { <<: *env, RANK: "0", DAISY_CAPACITY: "8000" }
+  rank1:
+    <<: *node
+    depends_on: [rank0]
+    environment: { <<: *env, RANK: "1", DAISY_CAPACITY: "4000" }
+  rank2:
+    <<: *node
+    depends_on: [rank0]
+    environment: { <<: *env, RANK: "2", DAISY_CAPACITY: "2000" }
+  dashboard:
+    build: { context: .., dockerfile: docker/dashboard.Dockerfile }
+    image: daisychain-dashboard:latest
+    depends_on: [rank0, rank1, rank2]
+    environment:
+      DAISY_NODES_FILE: config/nodes.example.json
+      DAISY_EXPECTED_WORLD: "3"
+      DAISY_BASE_BATCH: "32"
+    ports: ["8080:8080"]
+    networks: [cluster]
+
+networks:
+  cluster: { driver: bridge }

docker/node_entrypoint.sh

@@ -0,0 +1,8 @@
+#!/bin/sh
+# A DaisyChain node: run the dashboard agent in the background, then train.
+# After training, keep the agent alive so the dashboard can still show status.
+set -e
+python -m daisychain.dashboard.agent &
+python -m daisychain.train || echo "[node] training exited"
+echo "[node] training done — agent still serving status"
+wait

docs/CUSTOM_TASK.md

@@ -0,0 +1,52 @@
+# Training your own model
+
+DaisyChain trains any **Task** — an object with three methods:
+
+```python
+import torch, torch.nn as nn
+
+class MyTask:
+    def build_model(self) -> nn.Module:
+        torch.manual_seed(0)              # deterministic -> identical on every node
+        return nn.Sequential(nn.Linear(16, 64), nn.ReLU(), nn.Linear(64, 10))
+
+    def sample(self, n):                  # this node's data shard
+        X = torch.randn(n, 16)
+        y = torch.randint(0, 10, (n,))
+        return X, y
+
+    def loss(self, model, X, y):          # mean loss over the batch
+        return nn.functional.cross_entropy(model(X), y)
+```
+
+## Point DaisyChain at it
+
+```bash
+export DAISY_TASK="my_task:MyTask"        # module:Class, must be importable
+daisychain-train
+```
+
+Copy `examples/my_task_template.py` to start.
+
+## Rules that matter
+
+1. **`build_model` must be deterministic** (seed it). Every node builds the model
+   independently, then rank 0's weights are broadcast — but seeding keeps shapes
+   and buffers consistent.
+2. **`sample(n)` should return this node's shard.** For real datasets, split by
+   `RANK` (e.g. different files/row-ranges per rank) so nodes don't all train on
+   the same rows. Read `os.environ["RANK"]` / `WORLD_SIZE`.
+3. **The model must fit on one node.** DaisyChain pools compute, not memory.
+4. Keep it **small.** See [LIMITS.md](LIMITS.md).
+
+## Knobs (env)
+
+| var | default | meaning |
+|-----|---------|---------|
+| `DAISY_TASK` | example | `module:Class` |
+| `DAISY_STEPS` | 300 | training steps |
+| `DAISY_LR` | 0.05 | learning rate |
+| `DAISY_OPTIMIZER` | sgd | `sgd` or `adam` |
+| `DAISY_BASE_BATCH` | 32 | per-node base batch (scaled by capacity) |
+| `DAISY_SAVE` | daisychain_model.pt | where rank 0 saves |
+| `DAISY_FORCE_CPU` | – | set `1` to ignore a local GPU |

docs/LIMITS.md

@@ -0,0 +1,44 @@
+# Honest limits — read before you rely on DaisyChain
+
+DaisyChain is a real tool for a **specific** job: chaining spare/old machines to
+train **small** models faster by pooling their compute. It is not magic, and it
+is easy to misapply. Here's the honest envelope.
+
+## What it does
+- **Data-parallel** training: every node holds a full copy of the model and
+  trains on its slice of the data; gradients are averaged across the network.
+- **Capacity-weighted**: each node measures its own speed (CPU or GPU) and takes
+  a proportional batch, so a faster machine does more. GPU nodes auto-join and
+  carry more load; CPU-only nodes still participate.
+
+## What it does NOT do (expect these)
+- ❌ **Not GPU-class training.** N pooled old CPUs/GPUs are still that class of
+  hardware. A single modern GPU will beat the whole cluster for real models.
+- ❌ **Pools compute, not memory.** Every node needs the *whole* model in RAM
+  (or VRAM). You **cannot** train a model bigger than a single node can hold.
+  Chaining 5 laptops does not give you one big machine.
+- ❌ **Sublinear scaling.** Gradient sync over WiFi is bandwidth-bound, and the
+  **slowest node paces every step** (synchronous). Each added node helps less
+  than the last; a slow link can erase the benefit.
+- ❌ **Not for huge/modern models.** Transformers/large CNNs at CPU (or old-GPU)
+  speed are impractically slow.
+
+## The sweet spot
+Small models (small MLPs, tabular/classifier tasks, small fine-tunes),
+a handful of machines, model fits on each, you want more throughput and don't
+have a GPU handy. Education, research, retro/old-hardware hobby clusters.
+
+## Hardware / OS
+- **Multi-node reliably runs on Linux/macOS.** On **Windows**, gloo tensor
+  collectives are unstable — use **Docker** or **WSL**. Single-machine use is
+  fine on Windows.
+- **Old GPUs** must be new enough for current PyTorch (compute capability ≳ 5.0,
+  roughly GTX 900 / Maxwell and up). Kepler/Fermi-era cards won't work with
+  modern torch.
+- Put all nodes on the same **interface** (Tailscale `tailscale0`, or a real LAN
+  NIC) via `GLOO_SOCKET_IFNAME` — not a VPN/Docker/WSL virtual NIC.
+
+## The one-line version
+You can chain old machines to train a **small** model together, faster by
+throughput, with honest sublinear scaling — and DaisyChain won't pretend to be
+more than that.

docs/QUICKSTART.md

@@ -0,0 +1,57 @@
+# DaisyChain Quickstart
+
+Two ways to run. **Docker** is the most reliable (especially on Windows).
+
+## A. Docker — one command (demo cluster on one machine)
+
+```bash
+docker compose -f docker/docker-compose.yml up --build
+# open http://localhost:8080
+```
+
+This starts 3 node containers + the dashboard on a Docker network — the whole
+pipeline, so you can see connectivity, the capacity plan, and live training.
+Stop with `docker compose -f docker/docker-compose.yml down`.
+
+Windows: just run `scripts\setup.bat` and pick **[1] Docker**.
+
+## B. Python — real machines
+
+On **every** machine:
+
+```bash
+pip install torch numpy psutil
+pip install -e .            # from the repo, or `pip install daisychain`
+```
+
+Set the cluster env (copy `config/cluster.example.env`), changing only `RANK`
+per machine, then run:
+
+```bash
+export MASTER_ADDR=100.101.102.10   # coordinator IP (Tailscale 100.x recommended)
+export MASTER_PORT=29560
+export WORLD_SIZE=3
+export RANK=0                        # 1, 2, ... on the others
+export GLOO_SOCKET_IFNAME=tailscale0 # your mesh/LAN NIC
+daisychain-train
+```
+
+Windows: run `scripts\setup.bat` and pick **[2] Python** (it prompts for these).
+
+## Watch it
+
+Run the dashboard anywhere that can reach the nodes:
+
+```bash
+# edit config/nodes.example.json with your node hosts, then:
+daisychain-dashboard        # http://localhost:8080
+```
+
+## Train your own model
+
+The default is a tiny example task. To train **your** model, see
+[CUSTOM_TASK.md](CUSTOM_TASK.md) — copy `examples/my_task_template.py`, fill in
+`build_model` / `sample` / `loss`, and set `DAISY_TASK=your_module:YourTask`.
+
+**Before you rely on it, read [LIMITS.md](LIMITS.md).** DaisyChain pools compute,
+not memory, and is for *small* models on spare hardware — not GPU-class training.

docs/TAILSCALE.md

@@ -0,0 +1,41 @@
+# Connecting machines with Tailscale (recommended)
+
+DaisyChain needs every node reachable by a **stable IP on one interface**.
+Tailscale gives you exactly that: a private mesh where each machine gets a
+`100.x.y.z` address on the `tailscale0` interface, reachable P2P across NATs and
+different networks — no port-forwarding.
+
+## 1. Install on every machine
+- Linux: `curl -fsSL https://tailscale.com/install.sh | sh`
+- Windows / macOS: <https://tailscale.com/download>
+
+Bring each up (same tailnet / account) and note its IP:
+
+```bash
+sudo tailscale up
+tailscale ip -4          # e.g. 100.101.102.10
+tailscale status         # see every machine + IP
+```
+
+## 2. Configure the cluster
+Pick one machine as **rank 0** and use its Tailscale IP as `MASTER_ADDR`. On
+every machine:
+
+```bash
+export MASTER_ADDR=100.101.102.10
+export MASTER_PORT=29560
+export WORLD_SIZE=3
+export GLOO_SOCKET_IFNAME=tailscale0   # pin gloo to the mesh NIC
+export RANK=0                           # 1, 2, ... on the others
+daisychain-train
+```
+
+`GLOO_SOCKET_IFNAME=tailscale0` is the important line — it stops gloo from
+wandering onto a LAN/VPN/Docker interface.
+
+## 3. Verify
+Run `daisychain-dashboard` (point `config/nodes.example.json` at the Tailscale
+IPs). Green banner = every node reachable and ports open → launch training.
+
+> On Windows the Tailscale interface name differs and gloo tensor collectives are
+> unstable anyway — prefer Linux nodes or Docker for the actual training.

examples/my_task_template.py

@@ -0,0 +1,28 @@
+"""Template: copy this, fill in your model + data, then set
+
+    DAISY_TASK=my_task_template:MyTask
+
+(make sure the file is importable, e.g. run daisychain-train from this folder or
+pip-install your package). Keep build_model deterministic so every node starts
+from identical weights.
+"""
+import torch
+import torch.nn as nn
+
+
+class MyTask:
+    def build_model(self) -> nn.Module:
+        torch.manual_seed(0)                       # identical init on every node
+        # TODO: return YOUR model
+        return nn.Sequential(nn.Linear(16, 64), nn.ReLU(), nn.Linear(64, 10))
+
+    def sample(self, n: int):
+        # TODO: return n training samples from THIS node's data shard as (X, y).
+        # For real data, shard by rank (e.g. different files/rows per RANK).
+        X = torch.randn(n, 16)
+        y = torch.randint(0, 10, (n,))
+        return X, y
+
+    def loss(self, model, X, y):
+        # TODO: your loss (mean over the batch)
+        return nn.functional.cross_entropy(model(X), y)

pyproject.toml

@@ -0,0 +1,32 @@
+[build-system]
+requires = ["setuptools>=61"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "daisychain"
+version = "0.1.0"
+description = "DaisyChain — Old Hardware Training Pipeline: chain spare machines into a data-parallel training cluster."
+readme = "README.md"
+requires-python = ">=3.9"
+license = { text = "MIT" }
+authors = [{ name = "Dean Byrne (Quazim0t0)" }]
+keywords = ["distributed-training", "old-hardware", "cluster", "cpu", "gpu", "data-parallel"]
+dependencies = ["torch>=2.0", "numpy>=1.23"]
+
+[project.optional-dependencies]
+dashboard = ["psutil>=5.9"]
+
+[project.scripts]
+daisychain-train = "daisychain.train:main"
+daisychain-agent = "daisychain.dashboard.agent:main"
+daisychain-dashboard = "daisychain.dashboard.server:main"
+
+[project.urls]
+Homepage = "https://github.com/quzi93/daisychain"
+HuggingFace = "https://huggingface.co/DaisyChainAI/old-hw-train"
+
+[tool.setuptools]
+packages = ["daisychain", "daisychain.dashboard", "daisychain.verified"]
+
+[tool.setuptools.package-data]
+"daisychain.verified" = ["weights/*.pt"]

requirements.txt

@@ -0,0 +1,3 @@
+torch>=2.0
+numpy>=1.23
+psutil>=5.9

scripts/setup.bat

@@ -0,0 +1,78 @@
+@echo off
+REM ============================================================
+REM  DaisyChain - Old Hardware Training Pipeline : setup helper
+REM ============================================================
+setlocal enabledelayedexpansion
+cd /d "%~dp0\.."
+
+echo.
+echo   ____       _            ____ _           _
+echo  ^|  _ \ __ _^(_^)___ _   _^/ ___^| ^|__   __ _^(_^)_ __
+echo  ^| ^| ^| / _` ^| / __^| ^| ^| ^| ^|   ^| '_ \ / _` ^| ^| '_ \
+echo  ^| ^|_^| ^(_^| ^| \__ \ ^|_^| ^| ^|___^| ^| ^| ^| ^(_^| ^| ^| ^| ^| ^|
+echo  ^|____/ \__,_^|_^|___/\__, ^|\____^|_^| ^|_^|\__,_^|_^|_^| ^|_^|
+echo                     ^|___/   Old Hardware Training Pipeline
+echo.
+echo  Choose how to run DaisyChain on this machine:
+echo    [1] Docker      (recommended on Windows - most reliable)
+echo    [2] Python      (native; multi-node gloo is unstable on Windows - use WSL/Linux)
+echo    [3] Just install Python deps
+echo    [Q] Quit
+echo.
+set /p choice="  Your choice: "
+
+if /i "%choice%"=="1" goto docker
+if /i "%choice%"=="2" goto python
+if /i "%choice%"=="3" goto deps
+goto end
+
+:docker
+where docker >nul 2>nul
+if errorlevel 1 (
+  echo  [!] Docker not found. Install Docker Desktop from https://docker.com and re-run.
+  goto end
+)
+echo  Building and starting the demo cluster ^(3 nodes + dashboard^)...
+docker compose -f docker/docker-compose.yml up --build -d
+echo.
+echo  [OK] Cluster starting. Opening the dashboard at http://localhost:8080
+timeout /t 4 >nul
+start "" http://localhost:8080
+echo  To stop:  docker compose -f docker/docker-compose.yml down
+goto end
+
+:deps
+where python >nul 2>nul
+if errorlevel 1 ( echo  [!] Python not found. Install Python 3.9+ first. & goto end )
+echo  Installing dependencies...
+python -m pip install --upgrade pip
+python -m pip install torch numpy psutil
+python -m pip install -e .
+echo  [OK] Installed. You can now run:  daisychain-train
+goto end
+
+:python
+call :deps
+echo.
+echo  ---- Configure this node ----
+set /p master="  Coordinator IP (MASTER_ADDR, e.g. Tailscale 100.x): "
+set /p world="  Total number of machines (WORLD_SIZE): "
+set /p rank="  This machine's RANK (0 = coordinator): "
+set /p iface="  Network interface (GLOO_SOCKET_IFNAME, e.g. tailscale0): "
+set MASTER_ADDR=%master%
+set MASTER_PORT=29560
+set WORLD_SIZE=%world%
+set RANK=%rank%
+set GLOO_SOCKET_IFNAME=%iface%
+set USE_LIBUV=0
+echo.
+echo  [!] Note: native multi-node training over gloo is unstable on Windows.
+echo      If it hangs, use the Docker option or run the nodes on Linux/WSL.
+echo  Launching node RANK=%rank% ...
+python -m daisychain.train
+goto end
+
+:end
+echo.
+pause
+endlocal

scripts/setup.sh

@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+# DaisyChain setup helper (Linux/macOS).
+set -e
+cd "$(dirname "$0")/.."
+
+echo "🌼 DaisyChain — Old Hardware Training Pipeline"
+echo "  [1] Docker (demo cluster on this box)"
+echo "  [2] Python node (join a real cluster)"
+echo "  [3] Install deps only"
+read -rp "  Choice: " c
+
+case "$c" in
+  1)
+    command -v docker >/dev/null || { echo "Install Docker first."; exit 1; }
+    docker compose -f docker/docker-compose.yml up --build -d
+    echo "Dashboard: http://localhost:8080   (stop: docker compose -f docker/docker-compose.yml down)"
+    ;;
+  3|2)
+    python3 -m pip install --upgrade pip
+    python3 -m pip install torch numpy psutil
+    python3 -m pip install -e .
+    echo "Installed."
+    if [ "$c" = "2" ]; then
+      read -rp "  MASTER_ADDR (coordinator IP): " MA
+      read -rp "  WORLD_SIZE: " WS
+      read -rp "  RANK (0=coordinator): " RK
+      read -rp "  GLOO_SOCKET_IFNAME (e.g. tailscale0): " IF
+      export MASTER_ADDR="$MA" MASTER_PORT=29560 WORLD_SIZE="$WS" RANK="$RK"
+      export GLOO_SOCKET_IFNAME="$IF" USE_LIBUV=0
+      python3 -m daisychain.train
+    fi
+    ;;
+  *) echo "bye" ;;
+esac