Import from Quazim0t0/neural-storage; repoint refs to NeuralVerified

GF256.pt

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

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Dean Byrne (Quazim0t0)
+
+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,79 @@
+---
+license: mit
+library_name: pytorch
+tags:
+  - verified-units
+  - reed-solomon
+  - erasure-coding
+  - deduplication
+  - storage
+---
+
+# neural-storage — self-healing vault with a verified GF(256) core
+
+**Repositories:** [GitHub](https://github.com/quzi93/neural-storage) · [🤗 HuggingFace](https://huggingface.co/NeuralVerified/neural-storage)
+
+A content-addressed, deduplicating, **self-healing** storage vault. Any *k* of *n*
+Reed-Solomon shards reconstruct the whole, so pieces can be lost or corrupted and
+the data survives. The erasure-coding arithmetic — GF(2⁸) multiply — is provided
+by neural `LOG`/`EXP` units verified **bit-exact over all 65,536 (a,b) pairs**,
+the same N/N discipline as
+[neural-aarch64-units](https://huggingface.co/NeuralVerified/neural-aarch64-units).
+
+> **Honest by design:** dedup removes *redundant* data; RS *adds* redundancy for
+> resilience. Incompressible data never shrinks — you cannot beat entropy. Chunk
+> hashes are real SHA-256, not a neural emulation.
+
+## What's verified
+
+- `GF256` (`LOG`/`EXP`): GF(2⁸) multiply — **composed multiply 65536/65536**
+- Reed-Solomon any-*k*-of-*n* recovery (every loss pattern)
+- `Vault`: dedup + RS shards on disk, survives deleted/corrupted shards, `heal()`s
+- `VaultFS` + optional WinFsp drive-letter mount
+
+## Use
+
+```bash
+pip install torch
+python step1_storage.py   # verified GF(256) + dedup chunk store
+python step2_rs.py        # Reed-Solomon recovery
+python step3_vault.py     # self-healing vault
+
+python cli.py store  <vault> <folder>
+python cli.py export <vault> <folder>   # RS-healed reconstruct
+python cli.py mount  <vault> X:          # requires WinFsp
+
+# image a whole drive/partition into a self-healing .pt (may need admin):
+python cli.py image         \\.\C: diskC.pt
+python cli.py image-verify  diskC.pt
+python cli.py image-restore diskC.pt out.img
+```
+
+Weights: `GF256.pt`.
+
+**Create your own verified unit** (template: `storage/gf256.py`): write the exact
+golden finite function → enumerate the domain (decompose big/linear ones into
+bit/byte slices, see `storage/rs.py`) → `common.train` → `common.verify` must be
+bit-exact on 100% of inputs → compose. `step1_storage.py` shows the full loop.
+
+## Citation
+
+```bibtex
+@misc{byrne2026neuralstorage,
+  title  = {neural-storage: Self-Healing Erasure-Coded Vault with a Verified GF(256) Core},
+  author = {Byrne, Dean (Quazim0t0)},
+  year   = {2026},
+  howpublished = {\url{https://huggingface.co/NeuralVerified/neural-storage}}
+}
+```
+
+**Dean Byrne (Quazim0t0)** · 2026
+
+
+---
+
+<!-- neuralverified-relocation-note -->
+> **Now hosted by [NeuralVerified](https://huggingface.co/NeuralVerified).**
+>
+> This repo was moved into the NeuralVerified organization to help organize my profile.
+> Originally published at [`Quazim0t0/neural-storage`](https://huggingface.co/Quazim0t0/neural-storage).

cli.py

@@ -0,0 +1,93 @@
+"""
+neural-vault CLI. Dependency-free except `mount` (needs WinFsp + winfspy).
+
+  store  <vault> <src_folder> [--k K --m M]   ingest a folder (dedup + RS)
+  ls     <vault>                              list stored files
+  stat   <vault>                              dedup / redundancy stats
+  export <vault> <dst_folder>                 reconstruct all files (RS-healed)
+  heal   <vault>                              regenerate any missing shards
+  mount  <vault> <mountpoint>                 expose as a drive (WinFsp)
+
+  image         <device> <out.pt> [--k --m --part-mb]   scan a whole drive/partition
+                (e.g. \\\\.\\C: or an .img, may need admin) into a self-healing .pt
+  image-verify  <out.pt>                      report archive health
+  image-heal    <out.pt>                      repair detected bit-rot
+  image-restore <out.pt> <out.img>            reconstruct the image bit-exact
+"""
+import argparse
+from storage.vault import Vault
+from storage.vaultfs import winfsp_available
+from storage import diskimage
+
+
+def main():
+    ap = argparse.ArgumentParser(prog="neural-vault")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    p = sub.add_parser("store"); p.add_argument("vault"); p.add_argument("src")
+    p.add_argument("--k", type=int, default=4); p.add_argument("--m", type=int, default=2)
+    for name in ("ls", "stat", "heal"):
+        q = sub.add_parser(name); q.add_argument("vault")
+    e = sub.add_parser("export"); e.add_argument("vault"); e.add_argument("dst")
+    mo = sub.add_parser("mount"); mo.add_argument("vault"); mo.add_argument("mountpoint")
+    im = sub.add_parser("image"); im.add_argument("device"); im.add_argument("out")
+    im.add_argument("--k", type=int, default=4); im.add_argument("--m", type=int, default=2)
+    im.add_argument("--part-mb", type=int, default=64); im.add_argument("--label")
+    iv = sub.add_parser("image-verify"); iv.add_argument("out")
+    ih = sub.add_parser("image-heal"); ih.add_argument("out")
+    ir = sub.add_parser("image-restore"); ir.add_argument("out"); ir.add_argument("dst")
+    args = ap.parse_args()
+
+    if args.cmd == "store":
+        v = Vault(args.vault, args.k, args.m)
+        v.store_folder(args.src)
+        s = v.stats()
+        print(f"stored {len(v.list_files())} files, {s['unique_chunks']} unique chunks, "
+              f"{s['stored']//1024} KB on disk ({s['overhead']:.2f}x, RS {args.k}+{args.m})")
+    elif args.cmd == "ls":
+        for rel, sz in Vault(args.vault).list_files():
+            print(f"{sz:>10}  {rel}")
+    elif args.cmd == "stat":
+        s = Vault(args.vault).stats()
+        print(f"logical {s['logical']//1024} KB | stored {s['stored']//1024} KB | "
+              f"{s['unique_chunks']} chunks | overhead {s['overhead']:.2f}x")
+    elif args.cmd == "heal":
+        print(f"regenerated {Vault(args.vault).heal()} missing shard(s)")
+    elif args.cmd == "export":
+        print(f"exported {Vault(args.vault).export(args.dst)} files")
+    elif args.cmd == "mount":
+        if not winfsp_available():
+            print("WinFsp/winfspy not installed. Install WinFsp (https://winfsp.dev) "
+                  "and `pip install winfspy`, or use `export` for now.")
+            return
+        from storage.vaultfs import mount
+        fs = mount(Vault(args.vault), args.mountpoint)
+        print(f"mounted vault at {args.mountpoint} (Ctrl+C to unmount)")
+        try:
+            import time
+            while True:
+                time.sleep(1)
+        except KeyboardInterrupt:
+            fs.stop()
+            print("unmounted")
+    elif args.cmd == "image":
+        r = diskimage.create(args.device, args.out, k=args.k, m=args.m,
+                             part_mb=args.part_mb, label=args.label)
+        red = (args.k + args.m) / args.k
+        print(f"imaged {r['bytes']//1024} KB -> {r['out']} ({r['mode']}, {r['parts']} part(s), "
+              f"RS {args.k}+{args.m} ~{red:.2f}x). bad sectors: {r['bad_sectors']}")
+        if r["bad_sectors"]:
+            print(f"  WARNING: {r['bad_sectors']} sectors unreadable at scan (zero-filled).")
+    elif args.cmd == "image-verify":
+        h = diskimage.verify(args.out)
+        print(f"health: {'HEALTHY' if h['healthy'] else 'DAMAGED'}  "
+              f"chunks={h['chunks']} corrupted_shards={h['corrupted_shards']} "
+              f"repairable={h['repairable_chunks']} lost={h['lost_chunks']}")
+    elif args.cmd == "image-heal":
+        print(f"repaired {diskimage.heal(args.out)} shard(s)")
+    elif args.cmd == "image-restore":
+        n = diskimage.restore(args.out, args.dst)
+        print(f"restored {n//1024} KB -> {args.dst}")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1 @@
+torch

step1_storage.py

@@ -0,0 +1,73 @@
+"""
+Step 1: verified GF(256) multiply (RS keystone) + content-addressed chunk store.
+
+  (a) LOG / EXP verified N/N (255/255 each)
+  (b) COMPOSED GF(256) multiply verified over all 65,536 (a,b) pairs
+  (c) chunk store reconstructs a 'drive image' bit-exact
+  (d) dedup removes redundancy (real savings), and is bit-verified by SHA-256
+  (e) honesty: incompressible data yields ~no savings (no beating entropy)
+"""
+import os
+import torch
+from storage import gf256
+from storage.common import verify
+from storage.chunkstore import ChunkStore
+
+torch.manual_seed(0)
+
+print("=" * 62)
+print("STEP 1  --  verified GF(256) + content-addressed chunk store")
+print("=" * 62)
+
+# ---- verified coding keystone ----
+net_log, net_exp = gf256.train_units()
+lok, ltot = verify(net_log, *gf256.log_domain())
+eok, etot = verify(net_exp, *gf256.exp_domain())
+mok, mtot = gf256.verify_mul(net_log, net_exp)
+print("-" * 62)
+print(f"(a) LOG verified {lok}/{ltot}  |  EXP verified {eok}/{etot}  -> "
+      f"{'PASS' if lok==ltot and eok==etot else 'FAIL'}")
+print(f"(b) COMPOSED GF(256) multiply verified {mok}/{mtot}  -> "
+      f"{'PASS' if mok==mtot else 'FAIL'}")
+
+# ---- content-addressed chunk store on a synthetic 'drive image' ----
+fileA = os.urandom(64 * 1024)
+fileB = os.urandom(64 * 1024)
+fileC = os.urandom(64 * 1024)
+tail = os.urandom(64 * 1024)
+# a drive with duplicated files (A and B appear twice) + unique C + unique tail
+image = fileA + fileB + fileA + fileC + fileB + tail
+
+cs = ChunkStore()
+manifest = cs.put_stream(image)
+rebuilt = cs.reconstruct(manifest)
+
+exact = rebuilt == image
+print("-" * 62)
+print(f"(c) reconstruct drive image: {'bit-exact PASS' if exact else 'FAIL'} "
+      f"({len(image)} bytes)")
+print(f"(d) integrity (all chunks hash-verified): "
+      f"{'PASS' if cs.integrity_ok() else 'FAIL'}")
+st = cs.stats()
+print(f"    logical {st['logical_bytes']//1024} KB  ->  stored {st['stored_bytes']//1024} KB   "
+      f"dedup {st['dedup_ratio']:.2f}x   "
+      f"({st['unique_chunks']}/{st['total_chunks']} chunks unique)")
+
+# ---- honesty check: incompressible random data must NOT shrink ----
+rnd = os.urandom(384 * 1024)
+cs2 = ChunkStore()
+cs2.reconstruct(cs2.put_stream(rnd))  # store it
+st2 = cs2.stats()
+print(f"(e) incompressible 384 KB: dedup {st2['dedup_ratio']:.2f}x  "
+      f"-> {'PASS (no magic)' if st2['dedup_ratio'] < 1.05 else 'unexpected'}")
+
+allpass = (lok == ltot and eok == etot and mok == mtot and exact
+           and cs.integrity_ok() and st['dedup_ratio'] > 1.2 and st2['dedup_ratio'] < 1.05)
+print("=" * 62)
+print(f"OVERALL: {'ALL PASS' if allpass else 'some checks failed'}")
+
+if lok == ltot and eok == etot and mok == mtot:
+    torch.save({"log": net_log.state_dict(), "exp": net_exp.state_dict(),
+                "meta": {"unit": "GF256_MUL(log/exp)", "poly": "0x11D",
+                         "multiply_verified": f"{mok}/{mtot}"}}, "GF256.pt")
+    print("saved verified units -> GF256.pt")

step2_rs.py

@@ -0,0 +1,68 @@
+"""
+Step 2: Reed-Solomon erasure coding on the verified GF(256) core.
+
+  (a) neural coding equivalence -> parity computed via the NEURAL gf_mul equals
+      the golden RS parity (the verified units drive the code)
+  (b) any-k-of-n recovery       -> RS(k=4, m=2): reconstruct from EVERY possible
+      2-shard-loss pattern, bit-exact
+  (c) shards are bit-verified    -> each shard addressed by SHA-256
+  (d) honesty                    -> n shards total (k+m)/k x the data (redundancy,
+      not shrinkage)
+"""
+import os
+import hashlib
+import itertools
+import torch
+from storage import gf256, rs
+from storage.common import verify
+
+torch.manual_seed(0)
+
+print("=" * 62)
+print("STEP 2  --  Reed-Solomon erasure coding (verified GF core)")
+print("=" * 62)
+
+# load or train the verified GF units
+try:
+    ck = torch.load("GF256.pt", map_location="cpu")
+    from storage.common import mlp
+    net_log = mlp(8, 8, 256, 2); net_log.load_state_dict(ck["log"])
+    net_exp = mlp(8, 8, 256, 2); net_exp.load_state_dict(ck["exp"])
+    print("loaded verified GF units from GF256.pt")
+except Exception:
+    net_log, net_exp = gf256.train_units()
+
+# (a) neural coding equivalence
+neq = rs.neural_parity_equiv(net_log, net_exp)
+print(f"(a) neural gf_mul parity == golden RS parity  -> {'PASS' if neq else 'FAIL'}")
+
+# (b) any-k-of-n recovery over a real 'drive image'
+K, M = 4, 2
+N = K + M
+image = os.urandom(300 * 1024)               # 300 KB payload
+shards, L, orig = rs.encode(image, K, M)
+print(f"    RS({K}+{M}): {N} shards of {L//1024} KB each  (tolerates up to {M} lost)")
+
+fails = 0
+patterns = list(itertools.combinations(range(N), M))    # every 2-shard-loss case
+for lost in patterns:
+    present = {i: bytes(shards[i]) for i in range(N) if i not in lost}
+    if rs.decode(present, K, M, L, orig) != image:
+        fails += 1
+print(f"(b) recover from every {M}-loss pattern ({len(patterns)} of them): "
+      f"{'PASS' if fails == 0 else f'FAIL({fails})'}")
+
+# (c) shards bit-verified by SHA-256
+ids = [hashlib.sha256(bytes(s)).hexdigest()[:12] for s in shards]
+verified = all(hashlib.sha256(bytes(shards[i])).hexdigest()[:12] == ids[i] for i in range(N))
+print(f"(c) shard hashes (bit-verified): {'PASS' if verified else 'FAIL'}  "
+      f"e.g. shard0={ids[0]}")
+
+# (d) honesty: storage overhead is exactly the redundancy, nothing shrinks
+overhead = N * L / len(image)
+print(f"(d) storage: {len(image)//1024} KB -> {N*L//1024} KB ({overhead:.2f}x)  "
+      f"-> {'PASS (redundancy, no shrink)' if overhead >= 1.0 else 'FAIL'}")
+
+allpass = neq and fails == 0 and verified and overhead >= 1.0
+print("=" * 62)
+print(f"OVERALL: {'ALL PASS' if allpass else 'some checks failed'}")

step3_vault.py

@@ -0,0 +1,87 @@
+"""
+Step 3: the Vault -- dedup + RS-protected store that reconstructs files on demand
+and survives shard loss. Runs with no external dependencies (WinFsp only needed
+for the live drive-letter mount).
+
+  (a) store a folder (with duplicate files) -> dedup
+  (b) reconstruct every file bit-exact via the read layer (VaultFS)
+  (c) DAMAGE: delete up to m shards from chunks -> files still read (RS-heal)
+  (d) heal() regenerates the missing shards; integrity restored
+"""
+import os
+import random
+import shutil
+import tempfile
+from storage.vault import Vault
+from storage.vaultfs import VaultFS, winfsp_available
+
+random.seed(0)
+root = tempfile.mkdtemp(prefix="nvault_")
+src = os.path.join(root, "src"); os.makedirs(os.path.join(src, "sub"))
+vault_dir = os.path.join(root, "vault")
+dst = os.path.join(root, "restored")
+
+print("=" * 62)
+print("STEP 3  --  Vault: dedup + RS-protected virtual storage")
+print("=" * 62)
+
+# build a source folder with a duplicated file (photo.bin appears twice)
+photo = os.urandom(120 * 1024)
+files = {
+    "photo.bin": photo,
+    "sub/photo_copy.bin": photo,               # exact duplicate -> dedup
+    "notes.txt": ("neural vault demo\n" * 4000).encode(),
+    "random.bin": os.urandom(80 * 1024),
+}
+for rel, data in files.items():
+    p = os.path.join(src, rel)
+    os.makedirs(os.path.dirname(p), exist_ok=True)
+    open(p, "wb").write(data)
+
+# (a) store
+v = Vault(vault_dir, k=4, m=2)
+v.store_folder(src)
+s = v.stats()
+logical_files = sum(len(d) for d in files.values())
+print(f"(a) stored {len(v.list_files())} files, {s['unique_chunks']} unique chunks; "
+      f"logical {logical_files//1024} KB, on-disk {s['stored']//1024} KB "
+      f"(RS 4+2 => ~1.5x of DEDUPED data)")
+
+# (b) reconstruct via read layer, compare to originals
+fs = VaultFS(v)
+ok = all(fs.read(rel) == files[rel] for rel in files)
+print(f"(b) reconstruct every file bit-exact (VaultFS.read): {'PASS' if ok else 'FAIL'}")
+
+# (c) DAMAGE: delete up to m=2 shards from each chunk, then read anyway
+deleted = 0
+for h in os.listdir(v.cdir):
+    d = v._cd(h)
+    if not os.path.isfile(os.path.join(d, "meta.json")):
+        continue
+    shard_files = [f for f in os.listdir(d) if f.startswith("shard_")]
+    for f in random.sample(shard_files, k=random.randint(1, 2)):   # <= m
+        os.remove(os.path.join(d, f)); deleted += 1
+fs2 = VaultFS(v)                                    # fresh (no cache)
+survived = all(fs2.read(rel) == files[rel] for rel in files)
+print(f"(c) deleted {deleted} shard files (<=2/chunk); files still read: "
+      f"{'PASS (RS-healed on read)' if survived else 'FAIL'}")
+
+# (d) heal regenerates missing shards
+healed = v.heal()
+full = v.integrity_ok()
+print(f"(d) heal() regenerated {healed} shard(s); integrity restored: "
+      f"{'PASS' if full else 'FAIL'}")
+
+# export as final proof
+n = v.export(dst)
+export_ok = all(open(os.path.join(dst, rel), "rb").read() == files[rel] for rel in files)
+print(f"    export {n} files -> {'bit-exact PASS' if export_ok else 'FAIL'}")
+
+print("-" * 62)
+print(f"WinFsp drive-letter mount available on this machine: {winfsp_available()}  "
+      f"({'ready: python cli.py mount <vault> X:' if winfsp_available() else 'use: python cli.py export <vault> <folder>'})")
+
+allpass = ok and survived and full and export_ok
+print("=" * 62)
+print(f"OVERALL: {'ALL PASS' if allpass else 'some checks failed'}")
+shutil.rmtree(root, ignore_errors=True)

storage/archive.py

@@ -0,0 +1,120 @@
+"""
+Self-healing .pt disc archive.
+
+A disc image is chunked (fixed size), duplicate chunks deduped, and each unique
+chunk Reed-Solomon split into n = k+m shards (verified GF(256) core). Every shard
+carries its own SHA-256, so SILENT corruption (bit-rot flipping bytes) is
+DETECTED, marked as an erasure, and REPAIRED from the surviving shards -- as long
+as at least k of the n shards per chunk are still intact.
+
+The whole thing serializes to a single .pt (torch.save):
+  { meta, manifest:[chunk_hash...], chunks:{ hash: {shards, shard_hashes, L, orig} } }
+
+Honesty: RS adds redundancy (archive is ~(k+m)/k x the deduped image). It repairs
+up to m damaged shards per chunk; beyond that a chunk is flagged as lost, never
+silently wrong. No entropy is beaten.
+"""
+from __future__ import annotations
+import hashlib
+import time
+import torch
+from . import rs
+
+CHUNK = 4096
+
+
+def _h(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()
+
+
+def build(data: bytes, label="disc", k=4, m=2, chunk=CHUNK, bad_sectors=None) -> dict:
+    chunks = {}
+    manifest = []
+    for off in range(0, len(data), chunk):
+        ch = data[off:off + chunk]
+        h = _h(ch)
+        manifest.append(h)
+        if h in chunks:
+            continue
+        shards, L, orig = rs.encode(ch, k, m)
+        shards = [bytes(s) for s in shards]
+        chunks[h] = {"shards": shards, "shard_hashes": [_h(s) for s in shards],
+                     "L": L, "orig": orig, "k": k, "m": m}
+    return {
+        "meta": {"label": label, "size": len(data), "chunk": chunk, "k": k, "m": m,
+                 "bad_sectors": list(bad_sectors or []), "created": time.time(),
+                 "format": "neural-cd-preserve/1"},
+        "manifest": manifest, "chunks": chunks,
+    }
+
+
+def save(arc: dict, path: str):
+    torch.save(arc, path)
+
+
+def load(path: str) -> dict:
+    return torch.load(path, map_location="cpu", weights_only=False)
+
+
+def _good_shards(rec: dict) -> dict:
+    """{index: bytes} for shards whose hash still matches (uncorrupted)."""
+    out = {}
+    for i, s in enumerate(rec["shards"]):
+        if _h(s) == rec["shard_hashes"][i]:
+            out[i] = s
+    return out
+
+
+def verify(arc: dict) -> dict:
+    """Report health without modifying. -> {chunks, corrupted_shards, repairable,
+    lost} where lost chunks have < k intact shards (unrecoverable)."""
+    corrupted = repairable = lost = 0
+    for h, rec in arc["chunks"].items():
+        good = _good_shards(rec)
+        nbad = len(rec["shards"]) - len(good)
+        corrupted += nbad
+        if nbad and len(good) >= rec["k"]:
+            repairable += 1
+        if len(good) < rec["k"]:
+            lost += 1
+    return {"chunks": len(arc["chunks"]), "corrupted_shards": corrupted,
+            "repairable_chunks": repairable, "lost_chunks": lost,
+            "healthy": corrupted == 0 and lost == 0}
+
+
+def heal(arc: dict) -> int:
+    """Detect corrupted shards and regenerate them from survivors. Returns the
+    number of shards repaired. Chunks with < k intact shards cannot be healed."""
+    repaired = 0
+    for h, rec in arc["chunks"].items():
+        good = _good_shards(rec)
+        if len(good) == len(rec["shards"]):
+            continue
+        if len(good) < rec["k"]:
+            continue                                    # unrecoverable, leave as-is
+        ch = rs.decode(good, rec["k"], rec["m"], rec["L"], rec["orig"])
+        if _h(ch) != h:
+            continue                                    # decoded wrong -> don't trust
+        fresh, _, _ = rs.encode(ch, rec["k"], rec["m"])
+        for i in range(len(rec["shards"])):
+            if i not in good:
+                rec["shards"][i] = bytes(fresh[i])
+                rec["shard_hashes"][i] = _h(rec["shards"][i])
+                repaired += 1
+    return repaired
+
+
+def restore(arc: dict) -> bytes:
+    """Reconstruct the disc image, RS-correcting any detected corruption."""
+    out = bytearray()
+    for h in arc["manifest"]:
+        rec = arc["chunks"][h]
+        good = _good_shards(rec)
+        if len(good) < rec["k"]:
+            raise IOError(f"chunk {h[:8]}: only {len(good)} intact shards, "
+                          f"need {rec['k']} -- unrecoverable")
+        ch = rs.decode(good, rec["k"], rec["m"], rec["L"], rec["orig"])
+        if _h(ch) != h:
+            raise IOError(f"chunk {h[:8]}: hash mismatch after decode")
+        out += ch
+    return bytes(out[:arc["meta"]["size"]])

storage/chunkstore.py

@@ -0,0 +1,79 @@
+"""
+Content-addressed, deduplicating chunk store.
+
+  * content-defined chunking (Gear-hash CDC) splits a byte stream into
+    variable-length chunks at data-dependent boundaries -- so duplicate regions
+    chunk identically even if shifted.
+  * each chunk's SHA-256 is its address and its bit-verification.
+  * identical chunks are stored once (dedup). A manifest (list of hashes)
+    reconstructs the original stream exactly.
+
+Honest note: dedup + (optional) compression only removes REDUNDANT/compressible
+data. Incompressible or unique data is stored in full -- you cannot beat entropy.
+The chunk hashes are real crypto (SHA-256), not a neural emulation.
+"""
+from __future__ import annotations
+import hashlib
+import random
+
+# Gear table: 256 fixed 64-bit values (deterministic).
+_rng = random.Random(0xC0FFEE)
+GEAR = [_rng.getrandbits(64) for _ in range(256)]
+MASK64 = (1 << 64) - 1
+
+
+def chunk_stream(data: bytes, min_sz=2048, avg_bits=13, max_sz=65536):
+    """Yield (offset, length) chunk boundaries via Gear-hash CDC.
+    avg_bits sets average chunk size ~ 2**avg_bits bytes."""
+    mask = (1 << avg_bits) - 1
+    n = len(data)
+    start = 0
+    while start < n:
+        h = 0
+        i = start
+        end = min(start + max_sz, n)
+        while i < end:
+            h = ((h << 1) + GEAR[data[i]]) & MASK64
+            i += 1
+            if i - start >= min_sz and (h & mask) == 0:
+                break
+        yield start, i - start
+        start = i
+
+
+class ChunkStore:
+    def __init__(self):
+        self.store: dict[str, bytes] = {}     # sha256 -> chunk bytes
+        self.logical_bytes = 0
+        self.total_chunks = 0
+
+    def put_stream(self, data: bytes) -> list[str]:
+        manifest = []
+        self.logical_bytes += len(data)
+        for off, ln in chunk_stream(data):
+            ch = data[off:off + ln]
+            h = hashlib.sha256(ch).hexdigest()
+            if h not in self.store:
+                self.store[h] = ch
+            manifest.append(h)
+            self.total_chunks += 1
+        return manifest
+
+    def reconstruct(self, manifest: list[str]) -> bytes:
+        return b"".join(self.store[h] for h in manifest)
+
+    def integrity_ok(self) -> bool:
+        """Every stored chunk must still hash to its address (bit-verified)."""
+        return all(hashlib.sha256(v).hexdigest() == k for k, v in self.store.items())
+
+    def stored_bytes(self) -> int:
+        return sum(len(v) for v in self.store.values())
+
+    def stats(self) -> dict:
+        return {
+            "logical_bytes": self.logical_bytes,
+            "stored_bytes": self.stored_bytes(),
+            "total_chunks": self.total_chunks,
+            "unique_chunks": len(self.store),
+            "dedup_ratio": self.logical_bytes / max(1, self.stored_bytes()),
+        }

storage/common.py

@@ -0,0 +1,56 @@
+"""Shared helpers (same as the neural-DDR project): bit<->int, MLP, verify, train."""
+from __future__ import annotations
+import torch
+import torch.nn as nn
+
+DEV = "cuda" if torch.cuda.is_available() else "cpu"
+
+
+def bits_of(v: int, n: int) -> list[int]:
+    return [(v >> k) & 1 for k in range(n)]
+
+
+def int_of(bits) -> int:
+    return sum((1 << k) for k, b in enumerate(bits) if b > 0)
+
+
+def pm(bits) -> torch.Tensor:
+    return torch.tensor([1.0 if b else -1.0 for b in bits], dtype=torch.float32)
+
+
+def mlp(inp: int, out: int, h: int = 256, layers: int = 2) -> nn.Sequential:
+    mods = [nn.Linear(inp, h), nn.GELU()]
+    for _ in range(layers - 1):
+        mods += [nn.Linear(h, h), nn.GELU()]
+    mods += [nn.Linear(h, out)]
+    return nn.Sequential(*mods)
+
+
+@torch.no_grad()
+def verify(net, X, Ybits) -> tuple[int, int]:
+    net = net.to("cpu")
+    pred = (net(X) > 0).int()
+    ok = (pred == Ybits.int()).all(dim=1).sum().item()
+    return ok, X.shape[0]
+
+
+def train(net, X, Ybits, steps=8000, lr=2e-3, tag="", report=2000):
+    net = net.to(DEV)
+    Xd, Yd = X.to(DEV), Ybits.to(DEV)
+    opt = torch.optim.Adam(net.parameters(), lr=lr)
+    sch = torch.optim.lr_scheduler.CosineAnnealingLR(opt, T_max=steps)
+    lossfn = nn.BCEWithLogitsLoss()
+    for e in range(steps):
+        opt.zero_grad()
+        loss = lossfn(net(Xd), Yd)
+        loss.backward(); opt.step(); sch.step()
+        if e % report == 0 or e == steps - 1:
+            ok, tot = verify(net, X, Ybits); net.to(DEV)
+            print(f"    [{tag}] epoch {e:5d}  loss {loss.item():.2e}  verified {ok}/{tot}")
+            if ok == tot:
+                print(f"    [{tag}] -> N/N"); break
+    return net.to("cpu")
+
+
+def run8(net, v: int) -> int:
+    return int_of((net(pm(bits_of(v, 8)).unsqueeze(0))[0] > 0).int().tolist())

storage/diskimage.py

@@ -0,0 +1,141 @@
+"""
+Whole-drive imaging into a self-healing .pt.
+
+Scans a block device (Windows: r"\\\\.\\C:" or r"\\\\.\\PhysicalDrive0"; may need
+Administrator) or an image file, and turns it into a self-healing .pt archive:
+content-defined chunks, Reed-Solomon shards, per-shard SHA-256 (Step 1-2 +
+archive.py). It self-heals bit-rot in the archive and reconstructs the image
+bit-exact.
+
+Streaming by segment keeps memory bounded, so it scales past RAM:
+  * small input (<= one segment)  -> a single self-contained  <out>.pt
+  * large drive                    -> <out>.pt manifest + <out>.pt.part0, part1, ...
+
+HONEST LIMITS (read these):
+  * RS adds redundancy: the .pt is ~ (k+m)/k x the imaged data (e.g. 1.5x). This
+    does NOT shrink a drive -- you cannot beat entropy.
+  * Imaging an "entire HDD" means writing >= 1.5x its used size to disk and takes
+    real time; use --part-mb and expect a big output.
+  * Raw whole-disk reads need Administrator and the disk should be idle/unmounted
+    (image a partition, or a disk you are not writing to).
+  * RS heals the ARCHIVE against future bit-rot. Sectors unreadable at scan time
+    are recorded as bad, not invented.
+"""
+from __future__ import annotations
+import os
+import torch
+from . import archive, scan
+
+DEFAULT_PART = 64 * 1024 * 1024        # 64 MB segments
+
+
+def _read_segment(f, part_bytes: int, sector: int):
+    """Read up to part_bytes in sector-sized reads. -> (data, bad_sectors, eof)."""
+    data = bytearray()
+    bad = []
+    idx = 0
+    while len(data) < part_bytes:
+        try:
+            blk = f.read(sector)
+        except OSError:
+            bad.append(idx)
+            data += bytes(sector)
+            idx += 1
+            try:
+                f.seek(sector, os.SEEK_CUR)
+            except OSError:
+                return bytes(data), bad, True
+            continue
+        if not blk:
+            return bytes(data), bad, True
+        data += blk
+        idx += 1
+        if len(blk) < sector:
+            return bytes(data), bad, True
+    return bytes(data), bad, False
+
+
+def create(device: str, out_pt: str, k: int = 4, m: int = 2,
+           chunk: int = archive.CHUNK, part_mb: int = 64,
+           sector: int = scan.SECTOR, label: str | None = None) -> dict:
+    label = label or os.path.basename(device.rstrip("\\/")) or "disk"
+    part_bytes = part_mb * 1024 * 1024
+    with open(device, "rb", buffering=0) as f:
+        seg, bad, eof = _read_segment(f, part_bytes, sector)
+        if eof:                                    # whole device fit in one segment
+            arc = archive.build(seg, label=label, k=k, m=m, bad_sectors=bad)
+            arc["meta"]["device"] = device
+            torch.save(arc, out_pt)
+            return {"mode": "single", "bytes": len(seg), "bad_sectors": len(bad),
+                    "parts": 1, "out": out_pt}
+        parts, n, total_bad = [], 0, 0
+        while True:
+            p = f"{out_pt}.part{n}"
+            arc = archive.build(seg, label=f"{label}.part{n}", k=k, m=m, bad_sectors=bad)
+            torch.save(arc, p)
+            parts.append({"path": os.path.basename(p), "size": len(seg)})
+            total_bad += len(bad)
+            n += 1
+            if eof:
+                break
+            seg, bad, eof = _read_segment(f, part_bytes, sector)
+            if not seg and eof:
+                break
+    manifest = {"format": "neural-storage/diskimage-1", "device": device, "label": label,
+                "k": k, "m": m, "chunk": chunk, "parts": parts,
+                "total": sum(p["size"] for p in parts), "bad_sectors": total_bad}
+    torch.save(manifest, out_pt)
+    return {"mode": "multi", "bytes": manifest["total"], "bad_sectors": total_bad,
+            "parts": len(parts), "out": out_pt}
+
+
+def _is_manifest(obj) -> bool:
+    return isinstance(obj, dict) and obj.get("format", "").startswith("neural-storage/diskimage")
+
+
+def verify(out_pt: str) -> dict:
+    obj = archive.load(out_pt)
+    if not _is_manifest(obj):
+        return archive.verify(obj)
+    base = os.path.dirname(out_pt)
+    agg = {"chunks": 0, "corrupted_shards": 0, "repairable_chunks": 0, "lost_chunks": 0}
+    for p in obj["parts"]:
+        h = archive.verify(archive.load(os.path.join(base, p["path"])))
+        for kk in agg:
+            agg[kk] += h[kk]
+    agg["healthy"] = agg["corrupted_shards"] == 0 and agg["lost_chunks"] == 0
+    return agg
+
+
+def heal(out_pt: str) -> int:
+    obj = archive.load(out_pt)
+    if not _is_manifest(obj):
+        n = archive.heal(obj)
+        torch.save(obj, out_pt)
+        return n
+    base = os.path.dirname(out_pt)
+    total = 0
+    for p in obj["parts"]:
+        pth = os.path.join(base, p["path"])
+        arc = archive.load(pth)
+        n = archive.heal(arc)
+        if n:
+            torch.save(arc, pth)
+        total += n
+    return total
+
+
+def restore(out_pt: str, out_file: str) -> int:
+    obj = archive.load(out_pt)
+    with open(out_file, "wb") as w:
+        if not _is_manifest(obj):
+            data = archive.restore(obj)
+            w.write(data)
+            return len(data)
+        base = os.path.dirname(out_pt)
+        total = 0
+        for p in obj["parts"]:
+            data = archive.restore(archive.load(os.path.join(base, p["path"])))
+            w.write(data)
+            total += len(data)
+        return total

storage/gf256.py

@@ -0,0 +1,80 @@
+"""
+Step 1 keystone: verified GF(2^8) multiply -- the arithmetic core of Reed-Solomon
+erasure coding (Step 2).
+
+A direct 16-bit -> 8-bit neural multiply is hard to drive to N/N. So we decompose
+it the way real RS implementations do -- via log/exp tables:
+
+    a * b = EXP[ (LOG[a] + LOG[b]) mod 255 ]      (and 0 if a==0 or b==0)
+
+LOG and EXP are 256-entry lookups -> trivially N/N verifiable neural units. We
+train both, compose them into a full multiply, and verify the COMPOSED multiply
+against the golden field over ALL 65,536 (a,b) pairs.
+
+Field: GF(2^8), reduction polynomial 0x11D, generator 2 (standard for RS).
+"""
+from __future__ import annotations
+import torch
+from .common import mlp, train, verify, pm, bits_of, int_of, run8
+
+POLY = 0x11D
+
+
+def build_tables():
+    exp = [0] * 512
+    log = [0] * 256
+    x = 1
+    for i in range(255):
+        exp[i] = x
+        log[x] = i
+        x <<= 1
+        if x & 0x100:
+            x ^= POLY
+    for i in range(255, 512):
+        exp[i] = exp[i - 255]
+    return exp, log
+
+
+EXP, LOG = build_tables()
+
+
+def gf_mul(a: int, b: int) -> int:
+    if a == 0 or b == 0:
+        return 0
+    return EXP[LOG[a] + LOG[b]]
+
+
+# ---- neural LOG (inputs 1..255) and EXP (indices 0..254) ----
+def log_domain():
+    xs = [pm(bits_of(a, 8)) for a in range(1, 256)]
+    ys = [[float(b) for b in bits_of(LOG[a], 8)] for a in range(1, 256)]
+    return torch.stack(xs), torch.tensor(ys)
+
+
+def exp_domain():
+    xs = [pm(bits_of(i, 8)) for i in range(255)]
+    ys = [[float(b) for b in bits_of(EXP[i], 8)] for i in range(255)]
+    return torch.stack(xs), torch.tensor(ys)
+
+
+def gf_mul_neural(a: int, b: int, net_log, net_exp) -> int:
+    if a == 0 or b == 0:
+        return 0
+    s = (run8(net_log, a) + run8(net_log, b)) % 255
+    return run8(net_exp, s)
+
+
+def train_units():
+    print("  training LOG (8->8):")
+    net_log = train(mlp(8, 8, 256, 2), *log_domain(), steps=8000, tag="log")
+    print("  training EXP (8->8):")
+    net_exp = train(mlp(8, 8, 256, 2), *exp_domain(), steps=8000, tag="exp")
+    return net_log, net_exp
+
+
+def verify_mul(net_log, net_exp) -> tuple[int, int]:
+    ok = 0
+    for a in range(256):
+        for b in range(256):
+            ok += (gf_mul_neural(a, b, net_log, net_exp) == gf_mul(a, b))
+    return ok, 256 * 256

storage/rs.py

@@ -0,0 +1,104 @@
+"""
+Step 2: Reed-Solomon erasure coding over the verified GF(256) field.
+
+Split data into k systematic shards + m parity shards (n = k+m). ANY k of the n
+shards reconstruct the original -- so up to m shards can be lost or corrupted and
+the data still survives. This is the honest sense of "small pieces that together
+rebuild the whole": redundancy, not shrinkage (n shards total n/k x the data).
+
+All coefficients come from a Cauchy matrix over GF(256) (every square submatrix
+invertible => any-k-of-n recovery). Every multiply is the GF(256) multiply that
+the neural LOG/EXP units were verified bit-exact against (65536/65536), so the
+fast table path here is the proven-equivalent of the neural coding core.
+"""
+from __future__ import annotations
+from .gf256 import EXP, LOG, gf_mul
+
+MUL = [[gf_mul(a, b) for b in range(256)] for a in range(256)]   # full product table
+
+
+def gf_inv(a: int) -> int:
+    return EXP[255 - LOG[a]] if a else 0
+
+
+def cauchy(k: int, m: int):
+    return [[gf_inv((k + i) ^ j) for j in range(k)] for i in range(m)]
+
+
+def _gen_row(r: int, k: int, C):
+    if r < k:
+        row = [0] * k
+        row[r] = 1
+        return row
+    return C[r - k]
+
+
+def encode(data: bytes, k: int, m: int):
+    """-> (shards[n], shard_len, orig_len). shards[:k] are systematic data."""
+    orig = len(data)
+    L = (orig + k - 1) // k
+    d = data + bytes(L * k - orig)
+    shards = [bytearray(d[j * L:(j + 1) * L]) for j in range(k)]
+    C = cauchy(k, m)
+    for i in range(m):
+        P = bytearray(L)
+        Ci = C[i]
+        for j in range(k):
+            MC = MUL[Ci[j]]
+            Dj = shards[j]
+            for b in range(L):
+                P[b] ^= MC[Dj[b]]
+        shards.append(P)
+    return shards, L, orig
+
+
+def _invert(M, k):
+    A = [list(M[i]) + [1 if j == i else 0 for j in range(k)] for i in range(k)]
+    for col in range(k):
+        piv = col
+        while A[piv][col] == 0:
+            piv += 1
+        A[col], A[piv] = A[piv], A[col]
+        inv = gf_inv(A[col][col])
+        A[col] = [MUL[inv][x] for x in A[col]]
+        for r in range(k):
+            if r != col and A[r][col]:
+                f = A[r][col]
+                A[r] = [A[r][x] ^ MUL[f][A[col][x]] for x in range(2 * k)]
+    return [A[i][k:] for i in range(k)]
+
+
+def decode(present: dict, k: int, m: int, L: int, orig: int) -> bytes:
+    """present: {shard_index: bytes}; needs >= k entries."""
+    idxs = sorted(present.keys())[:k]
+    C = cauchy(k, m)
+    Minv = _invert([_gen_row(r, k, C) for r in idxs], k)
+    data_shards = [bytearray(L) for _ in range(k)]
+    for j in range(k):
+        row = Minv[j]
+        Dj = data_shards[j]
+        for t, r in enumerate(idxs):
+            c = row[t]
+            if not c:
+                continue
+            MC = MUL[c]
+            src = present[r]
+            for b in range(L):
+                Dj[b] ^= MC[src[b]]
+    return b"".join(bytes(s) for s in data_shards)[:orig]
+
+
+def neural_parity_equiv(net_log, net_exp, k=4, m=2, L=16) -> bool:
+    """Compute one parity shard with the NEURAL gf_mul and confirm it matches the
+    golden coding -- i.e. the verified units drive the erasure code bit-exactly."""
+    import os
+    from .gf256 import gf_mul_neural
+    data = os.urandom(L * k)
+    shards, _, _ = encode(data, k, m)
+    C = cauchy(k, m)
+    Dsh = [data[j * L:(j + 1) * L] for j in range(k)]
+    P = bytearray(L)
+    for j in range(k):
+        for b in range(L):
+            P[b] ^= gf_mul_neural(C[0][j], Dsh[j][b], net_log, net_exp)
+    return bytes(P) == bytes(shards[k])         # neural parity == golden parity

storage/scan.py

@@ -0,0 +1,57 @@
+"""
+Disc scanner.
+
+Reads a source sector by sector into bytes. The source can be a raw optical
+drive (Windows: r"\\\\.\\D:", may need admin) or an .iso / image file (for
+testing). Unreadable sectors (scratches / disc rot) are recorded as BAD and
+zero-filled -- ddrescue-style -- rather than aborting the scan.
+
+Important: bad sectors are data you never read. RS self-healing (archive.py)
+protects the STORED copy against future corruption; it cannot invent sectors the
+drive could not read. To recover those, re-scan the disc and merge good sectors
+across attempts (`merge_scans`).
+"""
+from __future__ import annotations
+
+SECTOR = 2048                     # CD-ROM Mode-1 user data sector size
+
+
+def scan_image(path: str, sector: int = SECTOR):
+    """-> (data: bytes, bad_sectors: list[int]). Robust to unreadable sectors."""
+    data = bytearray()
+    bad = []
+    with open(path, "rb", buffering=0) as f:
+        idx = 0
+        while True:
+            try:
+                blk = f.read(sector)
+            except OSError:
+                bad.append(idx)
+                data += bytes(sector)
+                try:
+                    f.seek((idx + 1) * sector)
+                except OSError:
+                    break
+                idx += 1
+                continue
+            if not blk:
+                break
+            data += blk                                # keep true length (last
+            idx += 1                                   # sector may be partial)
+    return bytes(data), bad
+
+
+def merge_scans(scans, sector: int = SECTOR):
+    """Combine multiple (data, bad) scans of the same failing disc: a sector is
+    good if ANY scan read it. Returns (merged_data, still_bad_sectors)."""
+    n = max(len(d) // sector for d, _ in scans)
+    merged = bytearray(n * sector)
+    known = [False] * n
+    for data, bad in scans:
+        badset = set(bad)
+        for s in range(len(data) // sector):
+            if s not in badset and not known[s]:
+                merged[s * sector:(s + 1) * sector] = data[s * sector:(s + 1) * sector]
+                known[s] = True
+    still_bad = [s for s in range(n) if not known[s]]
+    return bytes(merged), still_bad

storage/vault.py

@@ -0,0 +1,152 @@
+"""
+Step 3 backing store: the Vault.
+
+Combines everything so far into an on-disk repository:
+  * files are content-defined-chunked (Step 1 CDC)
+  * unique chunks are stored once, addressed by SHA-256 (dedup, bit-verified)
+  * each unique chunk is Reed-Solomon split into n = k+m shard files (Step 2),
+    so any k shards rebuild it -- up to m shard files can be lost/corrupted
+
+On-disk layout:
+  <vault>/chunks/<sha256>/shard_0 .. shard_{n-1}, meta.json
+  <vault>/files/<relpath>.json   (manifest: name, size, [chunk hashes])
+
+Reads reconstruct on demand and RS-heal transparently; `heal()` regenerates any
+missing shard files from the survivors. Nothing here beats entropy -- dedup
+removes redundancy, RS adds redundancy for resilience.
+"""
+from __future__ import annotations
+import os
+import json
+import hashlib
+from .chunkstore import chunk_stream
+from . import rs
+
+
+class Vault:
+    def __init__(self, path: str, k: int = 4, m: int = 2):
+        self.path = path
+        self.k, self.m = k, m
+        self.cdir = os.path.join(path, "chunks")
+        self.fdir = os.path.join(path, "files")
+        os.makedirs(self.cdir, exist_ok=True)
+        os.makedirs(self.fdir, exist_ok=True)
+
+    # ---- chunks ----
+    def _cd(self, h: str) -> str:
+        return os.path.join(self.cdir, h)
+
+    def _store_chunk(self, ch: bytes) -> str:
+        h = hashlib.sha256(ch).hexdigest()
+        d = self._cd(h)
+        if os.path.exists(os.path.join(d, "meta.json")):
+            return h                                    # dedup: already stored
+        os.makedirs(d, exist_ok=True)
+        shards, L, orig = rs.encode(ch, self.k, self.m)
+        for i, s in enumerate(shards):
+            with open(os.path.join(d, f"shard_{i}"), "wb") as f:
+                f.write(bytes(s))
+        json.dump({"L": L, "orig": orig, "k": self.k, "m": self.m},
+                  open(os.path.join(d, "meta.json"), "w"))
+        return h
+
+    def _load_chunk(self, h: str) -> bytes:
+        d = self._cd(h)
+        meta = json.load(open(os.path.join(d, "meta.json")))
+        n = meta["k"] + meta["m"]
+        present = {}
+        for i in range(n):
+            p = os.path.join(d, f"shard_{i}")
+            if os.path.exists(p):
+                b = open(p, "rb").read()
+                if len(b) == meta["L"]:
+                    present[i] = b
+        if len(present) < meta["k"]:
+            raise IOError(f"chunk {h[:8]}: {len(present)} shards, need {meta['k']}")
+        ch = rs.decode(present, meta["k"], meta["m"], meta["L"], meta["orig"])
+        if hashlib.sha256(ch).hexdigest() != h:
+            raise IOError(f"chunk {h[:8]}: hash mismatch after decode")
+        return ch
+
+    # ---- files ----
+    def store_file(self, relpath: str, data: bytes) -> None:
+        chunks = [self._store_chunk(data[o:o + l]) for o, l in chunk_stream(data)]
+        mfp = os.path.join(self.fdir, relpath + ".json")
+        os.makedirs(os.path.dirname(mfp) or ".", exist_ok=True)
+        json.dump({"name": relpath, "size": len(data), "chunks": chunks}, open(mfp, "w"))
+
+    def store_folder(self, src: str) -> None:
+        for root, _, names in os.walk(src):
+            for nm in names:
+                fp = os.path.join(root, nm)
+                rel = os.path.relpath(fp, src).replace("\\", "/")
+                with open(fp, "rb") as f:
+                    self.store_file(rel, f.read())
+
+    def list_files(self):
+        out = []
+        for root, _, names in os.walk(self.fdir):
+            for nm in names:
+                if nm.endswith(".json"):
+                    mf = json.load(open(os.path.join(root, nm)))
+                    out.append((mf["name"], mf["size"]))
+        return sorted(out)
+
+    def read_file(self, relpath: str) -> bytes:
+        mf = json.load(open(os.path.join(self.fdir, relpath + ".json")))
+        data = b"".join(self._load_chunk(h) for h in mf["chunks"])
+        return data[:mf["size"]]
+
+    def export(self, dst: str) -> int:
+        n = 0
+        for rel, _ in self.list_files():
+            data = self.read_file(rel)
+            op = os.path.join(dst, rel)
+            os.makedirs(os.path.dirname(op) or ".", exist_ok=True)
+            with open(op, "wb") as f:
+                f.write(data)
+            n += 1
+        return n
+
+    # ---- resilience ----
+    def heal(self) -> int:
+        healed = 0
+        for h in os.listdir(self.cdir):
+            d = self._cd(h)
+            mp = os.path.join(d, "meta.json")
+            if not os.path.isfile(mp):
+                continue
+            meta = json.load(open(mp))
+            n = meta["k"] + meta["m"]
+            missing = [i for i in range(n) if not os.path.exists(os.path.join(d, f"shard_{i}"))]
+            if missing:
+                ch = self._load_chunk(h)                # decode from survivors
+                shards, _, _ = rs.encode(ch, meta["k"], meta["m"])
+                for i in missing:
+                    with open(os.path.join(d, f"shard_{i}"), "wb") as f:
+                        f.write(bytes(shards[i]))
+                healed += len(missing)
+        return healed
+
+    def integrity_ok(self) -> bool:
+        try:
+            for rel, _ in self.list_files():
+                self.read_file(rel)
+            return True
+        except IOError:
+            return False
+
+    def stats(self) -> dict:
+        logical = sum(sz for _, sz in self.list_files())
+        stored = 0
+        chunks = 0
+        for h in os.listdir(self.cdir):
+            d = self._cd(h)
+            if not os.path.isfile(os.path.join(d, "meta.json")):
+                continue
+            chunks += 1
+            for f in os.listdir(d):
+                if f.startswith("shard_"):
+                    stored += os.path.getsize(os.path.join(d, f))
+        return {"logical": logical, "stored": stored, "unique_chunks": chunks,
+                "overhead": stored / max(1, logical)}

storage/vaultfs.py

@@ -0,0 +1,129 @@
+"""
+Read interface over a Vault, plus an optional WinFsp mount.
+
+`VaultFS` is a dependency-free read model: list files, stat, and read byte ranges
+that are reconstructed (and RS-healed) on demand from the Vault. This is what the
+CLI `export` uses and what the mount serves.
+
+`mount()` exposes the Vault as a real drive letter via WinFsp (Windows FUSE). It
+is guarded: if `winfspy` / WinFsp is not installed, it raises with instructions
+instead of failing obscurely. The dependency-free path (VaultFS + CLI export)
+works everywhere; the live drive lights up once WinFsp is present.
+"""
+from __future__ import annotations
+from collections import OrderedDict
+from .vault import Vault
+
+
+class VaultFS:
+    def __init__(self, vault: Vault, cache_files: int = 8):
+        self.vault = vault
+        self._cache: OrderedDict[str, bytes] = OrderedDict()
+        self._cache_n = cache_files
+        self._index = {rel: sz for rel, sz in vault.list_files()}
+
+    def listdir(self):
+        return sorted(self._index.items())            # [(relpath, size)]
+
+    def size(self, relpath: str) -> int:
+        return self._index[relpath]
+
+    def _data(self, relpath: str) -> bytes:
+        if relpath in self._cache:
+            self._cache.move_to_end(relpath)
+            return self._cache[relpath]
+        data = self.vault.read_file(relpath)           # reconstruct + RS-heal
+        self._cache[relpath] = data
+        if len(self._cache) > self._cache_n:
+            self._cache.popitem(last=False)
+        return data
+
+    def read(self, relpath: str, offset: int = 0, length: int | None = None) -> bytes:
+        data = self._data(relpath)
+        if length is None:
+            return data[offset:]
+        return data[offset:offset + length]
+
+
+def winfsp_available() -> bool:
+    try:
+        import winfspy  # noqa: F401
+        return True
+    except Exception:
+        return False
+
+
+def mount(vault: Vault, mountpoint: str, label: str = "NeuralVault"):
+    """Mount the vault as a read-only drive via WinFsp. Requires WinFsp + winfspy.
+    Experimental: the dependency-free `export` path is the tested one."""
+    if not winfsp_available():
+        raise RuntimeError(
+            "WinFsp / winfspy not found. Install WinFsp (https://winfsp.dev) and "
+            "`pip install winfspy`, or use the dependency-free CLI: "
+            "`python cli.py export <vault> <folder>`."
+        )
+    # Imported lazily so the module loads without winfspy present.
+    from winfspy import (FileSystem, BaseFileSystemOperations, NTStatusObjectNameNotFound,
+                         FILE_ATTRIBUTE)
+    import time
+
+    fs_view = VaultFS(vault)
+    now = int(time.time() * 1e7) + 116444736000000000  # FILETIME epoch
+
+    class Ops(BaseFileSystemOperations):
+        def _info(self, is_dir, size=0):
+            return {
+                "file_attributes": FILE_ATTRIBUTE.FILE_ATTRIBUTE_DIRECTORY if is_dir
+                else FILE_ATTRIBUTE.FILE_ATTRIBUTE_NORMAL,
+                "allocation_size": size, "file_size": size,
+                "creation_time": now, "last_access_time": now,
+                "last_write_time": now, "change_time": now, "index_number": 0,
+            }
+
+        def _norm(self, p):
+            return p.replace("\\", "/").lstrip("/")
+
+        def get_volume_info(self):
+            return {"total_size": 0, "free_size": 0, "volume_label": label}
+
+        def get_security_by_name(self, file_name):
+            rel = self._norm(file_name)
+            is_dir = rel == "" or any(f.startswith(rel + "/") for f in fs_view._index)
+            if not is_dir and rel not in fs_view._index:
+                raise NTStatusObjectNameNotFound()
+            return (self._info(is_dir, 0 if is_dir else fs_view.size(rel))["file_attributes"], None, 0)
+
+        def open(self, file_name, create_options, granted_access):
+            rel = self._norm(file_name)
+            is_dir = rel == "" or any(f.startswith(rel + "/") for f in fs_view._index)
+            if not is_dir and rel not in fs_view._index:
+                raise NTStatusObjectNameNotFound()
+            return {"rel": rel, "is_dir": is_dir}
+
+        def get_file_info(self, fh):
+            return self._info(fh["is_dir"], 0 if fh["is_dir"] else fs_view.size(fh["rel"]))
+
+        def read(self, fh, offset, length):
+            return fs_view.read(fh["rel"], offset, length)
+
+        def read_directory(self, fh, marker):
+            prefix = "" if fh["rel"] == "" else fh["rel"] + "/"
+            names = set()
+            for f in fs_view._index:
+                if f.startswith(prefix):
+                    rest = f[len(prefix):]
+                    names.add(rest.split("/")[0])
+            out = []
+            for nm in sorted(names):
+                full = prefix + nm
+                is_dir = full not in fs_view._index
+                out.append({"file_name": nm,
+                            "file_info": self._info(is_dir, 0 if is_dir else fs_view.size(full))})
+            return out
+
+        def close(self, fh):
+            pass
+
+    fs = FileSystem(mountpoint, Ops(), volume_label=label)
+    fs.start()
+    return fs