#!/usr/bin/env python3 # SPDX-License-Identifier: MIT OR Apache-2.0 """bankml · view — the read-only page for anyone on the LAN who wants to watch the testing. Why not Gradio: the Gradio installed here (3.37) has path-traversal bugs that let a client read files from the host (e.g. CVE-2023-51449, fixed in 4.11). That is acceptable on loopback for the operator, not on a network. This server is the Python standard library with a fixed set of GET routes, nothing else: / the page (HTML + CSS + JS inline; the JS renders every value with textContent) /api/state JSON: the live testing log, its stage, the machine, the release records' names, CI, bankml serve's verification, Savante's office and ledger check /api/result?name= one release record — only a name that `testing/results/` actually lists /savante.png the canon's named image (read-only) No chat, no .history, no command execution, no file paths from the client. Refreshes every 2 s. python3 sAGI/view.py --host 0.0.0.0 --port 7874 # the LAN """ from __future__ import annotations import argparse import json import sys import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path sys.dont_write_bytecode = True sys.path.insert(0, str(Path(__file__).resolve().parent)) import savante as S # noqa: E402 (stdlib-only at import; gradio is imported only by its build()) CANON = S.Canon(S.CANON) KNOBS = Path(__file__).resolve().parent / "voice" / "knobs" / "savante_knobs.js" import re # noqa: E402 import speak # noqa: E402 _LISTEN = {"key": None, "t": 0.0, "data": None} def listen() -> dict: """listen_uncached(), recomputed when the voice manifest, an export or TECHNICAL.md changes (or every 30 s).""" import time as _t def mt(p): try: return p.stat().st_mtime_ns except OSError: return 0 key = (mt(speak.VOICE_DIR / "savante" / "manifest.json"), mt(speak.REPO / "docs" / "TECHNICAL.md"), tuple(mt(speak.EXPORT_DIR / f"{n}.json") for n in speak.EXPORTS)) if key != _LISTEN["key"] or _t.time() - _LISTEN["t"] > 30 or _LISTEN["data"] is None: _LISTEN.update(key=key, t=_t.time(), data=listen_uncached()) return _LISTEN["data"] def listen_uncached() -> dict: """Savante's introduction and voice examples, as far as they are rendered (keys only; audio via /audio/).""" chs = speak.intro_chapters(S.CANON, CANON.persona, CANON.card) ex = [x for x in CANON.persona.get("voice_examples") or [] if isinstance(x, str)] out = [] rd = [("The reading — " + t, x) for t, x in speak.reading_chapters()] for title, sents in chs + rd + [("Savante speaks — her voice examples", ex)]: items = speak.cached(sents) if all(items): out.append({"title": title, "items": [{"text": i["text"], "key": i["key"], "seconds": i["seconds"]} for i in items]}) else: out.append({"title": title, "rendering": True, "count": len(sents)}) exports = [] for name, (lead, cs) in speak.export_sets(S.CANON, CANON.persona, CANON.card).items(): st = speak.export_state(name, ([("Voice examples", lead)] if lead else []) + cs) exports.append({"name": name, "what": speak.EXPORTS[name], "ready": st["ready"], "total": st["total"], "current": st["current"], "mb": round(st["file"].stat().st_size / 1e6, 1) if st["current"] else None}) return {"voice": speak.savante_voice(), "chapters": out, "exports": exports} def export_file(name: str): """Only a named export (speak.EXPORTS) that is complete and current.""" if name not in speak.EXPORTS: return None cur = any(x["name"] == name and x["current"] for x in listen()["exports"]) # the same (cached) judgement /api/state shows return speak.EXPORT_DIR / f"{name}.opus" if cur else None def audio_file(key: str): """Only a 24-hex key the voice manifest lists; nothing else is reachable.""" if not re.fullmatch(r"[0-9a-f]{24}", key or ""): return None try: man = json.loads((speak.VOICE_DIR / "savante" / "manifest.json").read_text(encoding="utf-8")) except (OSError, ValueError): return None f = speak.VOICE_DIR / "savante" / f"{key}.ogg" return f if key in man and f.is_file() else None _STATE = {"t": 0.0, "body": None} def state_json() -> bytes: """/api/state, computed at most every 2 s however many viewers poll (serve_status alone can wait 3 s).""" import time as _t if _STATE["body"] is None or _t.time() - _STATE["t"] > 2: _STATE.update(t=_t.time(), body=json.dumps(state()).encode()) return _STATE["body"] def state() -> dict: p, card = CANON.persona, CANON.card serve = S.serve_status() return { "live": {**S.live_state(), "tail": S.live_tail(80)}, "machine": S.machine_state(), "results": S.results_list(), "ci": S.ci_runs(), "serve": {"ok": "error" not in serve, "error": serve.get("error"), "model": Path(serve.get("model", "")).name, "sha256": (serve.get("verified") or {}).get("model_sha256"), "bankml": (serve.get("verified") or {}).get("bankml"), "engine": serve.get("engine")}, "private": {"history": S.commitment(S.HISTORY), "memory": S.commitment(S.MEMORY)}, "listen": listen(), "savante": {"name": p.get("name"), "kind": p.get("kind"), "mantra": p.get("mantra"), "oath": p.get("oath"), "status": (card.get("savante") or {}).get("status"), "type": card.get("type"), "doctrine_root": (CANON.ledger.get("doctrine_root") or {}).get("value"), "ledger": [{"entry": n, "path": r, "ok": ok, "detail": d} for n, r, ok, d in CANON.rows]}, } PAGE = r""" bankML · view
Savante

bankML · view — watching the testing live

Read-only. A speed counts only if every oracle passed on the same code. Refreshes every 2 s ·

Testing — live

…
step: 

loading…

This laptop

bankml serve

Release records



CI — github.com/cryptoAGI/bankml
runresultwhen


Listen to Savante


New here? Savante reads herself to you — who she is, her oath, what she believes, how she
works, why she exists — verbatim from her canon, in her voice ().








Private data — commitments only


.history and .memory stay on this laptop. What is shown is their commitment: anyone given one
exchange and its inclusion proof can check it against this root, without seeing the rest.


Savante — office and iNFT ledger


 


ledger entrycheck




bankML · verified low-bit inference · cryptoAGI · this page serves no chat and no history


"""


class H(BaseHTTPRequestHandler):
    server_version = "bankml-view"
    timeout = 20  # a LAN client that sends nothing (slowloris) releases its thread after 20 s

    def send(self, code, ctype, body: bytes, cache="no-store", disp=None):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        if disp:
            self.send_header("Content-Disposition", disp)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", cache)
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header("Content-Security-Policy", "default-src 'none'; img-src 'self'; style-src 'unsafe-inline'; script-src 'self' 'unsafe-inline'; connect-src 'self'; media-src 'self'")
        self.end_headers()
        self.wfile.write(body)

    def send_file(self, f, ctype, disp=None):
        """A large file in 1 MiB chunks, not read into memory whole."""
        size = f.stat().st_size
        self.send_response(200)
        for k, v in (("Content-Type", ctype), ("Content-Length", str(size)), ("Cache-Control", "no-cache"), ("X-Content-Type-Options", "nosniff"),
                     ("Referrer-Policy", "no-referrer"), ("Content-Security-Policy", "default-src 'none'")):
            self.send_header(k, v)
        if disp:
            self.send_header("Content-Disposition", disp)
        self.end_headers()
        with open(f, "rb") as fh:
            while b := fh.read(1 << 20):
                self.wfile.write(b)

    def do_GET(self):
        try:
            return self.route()
        except (BrokenPipeError, ConnectionResetError):
            pass
        except Exception as e:  # noqa: BLE001 — one bad file must not drop every viewer's connection
            try:
                self.send(500, "text/plain", f"error: {type(e).__name__}".encode())
            except OSError:
                pass

    def route(self):
        u = urllib.parse.urlsplit(self.path)
        if u.path == "/":
            return self.send(200, "text/html; charset=utf-8", PAGE.encode())
        if u.path == "/favicon.ico":  # the DeltaVerse $ (deltaverse.pythai.net, MIT)
            return self.send(200, "image/x-icon", (Path(__file__).with_name("favicon.ico")).read_bytes())
        if u.path == "/api/state":
            return self.send(200, "application/json", state_json())
        if u.path == "/api/result":
            name = (urllib.parse.parse_qs(u.query).get("name") or [""])[0]
            if name in S.results_list():  # only names the directory listing produced
                return self.send(200, "text/plain; charset=utf-8", S.results_read(name).encode())
            return self.send(404, "text/plain", b"no such record")
        if u.path == "/knobs.js":  # the DreamKnob voice controls, the same bundle the card uses
            try:
                return self.send(200, "text/javascript; charset=utf-8", KNOBS.read_bytes(), cache="max-age=3600")
            except OSError:
                return self.send(404, "text/plain", b"no knobs")
        if u.path.startswith("/audio/") and u.path.endswith(".ogg"):
            f = audio_file(u.path[len("/audio/"):-len(".ogg")])
            return self.send(200, "audio/ogg", f.read_bytes(), cache="max-age=86400") if f else self.send(404, "text/plain", b"no such clip")
        if u.path.startswith("/export/") and u.path.endswith(".opus"):  # the complete introduction / reading, one file each
            f = export_file(u.path[len("/export/"):-len(".opus")])
            if not f:
                return self.send(404, "text/plain", b"no such export (it exists only when complete)")
            return self.send_file(f, "audio/ogg", disp=f'attachment; filename="{f.name}"')
        if u.path == "/savante.png":
            rel = (CANON.ledger.get("image_candidate") or {}).get("path") or "gfx/Savante3.png"
            b = CANON.file(rel)
            return self.send(200, "image/png", b, cache="max-age=3600") if b else self.send(404, "text/plain", b"no image")
        return self.send(404, "text/plain", b"not found")

    def do_POST(self):
        self.send(405, "text/plain", b"read-only")

    def log_message(self, *a):
        pass


def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--host", default="0.0.0.0")
    ap.add_argument("--port", type=int, default=7874)
    a = ap.parse_args()
    print(f"bankml view on http://{a.host}:{a.port}/ (read-only) · canon ledger {sum(r[2] for r in CANON.rows)}/{len(CANON.rows)}", flush=True)
    ThreadingHTTPServer((a.host, a.port), H).serve_forever()


if __name__ == "__main__":
    main()