"""The WebUI: a CRT terminal, the diagnostics, the visualisations, the assistant. Served by ``http.server`` and nothing else — no framework, no bundler, no build step. The page is one string; the data comes from JSON endpoints that call the same objects the CLI does, so the browser cannot show something the machine does not actually do. Endpoints: GET / the page (terminal + panels) GET /api/state boot screen, terminal cells, kernel status POST /api/exec {"line": "run"} -> console output POST /api/chat {"text": "..."} -> assistant reply + executed command GET /api/autonomy idle/readahead state: graded accuracy, lookahead, cache POST /api/autonomy {"enabled": true} -> toggle the autonomous pass GET /api/os the webOS frame, windows, memory map POST /api/os/key {"key": "F5"} -> one named key to the window manager POST /api/os/text {"text": "stat"} -> typing into the focused window POST /api/os/click {"x": 12, "y": 4} -> focus the window at that cell POST /api/os/launch {"app": "ide"} -> open an app POST /api/boot re-run POST + boot sector GET /api/diag every layer's measured report GET /api/viz/ one visualisation as ASCII GET /api/pipeline the stages, with their real counters It binds ``127.0.0.1`` by default and holds a single kernel in memory: this is a local instrument panel, not a multi-tenant service. There is no authentication because there is no remote exposure. """ from __future__ import annotations import json import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from .assistant import INTENTS, Assistant from .autonomy import AutonomousMode from .boot import BootSequence from .hypervec import DEFAULT_DIM from .os import Desktop from .semantic import load as load_semantic from .terminal import Kernel from .x86 import Boot PAGE = """ HDCSTORAGE

HDCSTORAGE — CRTC mono, 80x25 


    

      
      
    

    
connecting

  

  

    

      
      
      
      
      
    

    

  




"""


class App:
    """Kernel + assistant + boot state, shared by every request."""

    def __init__(self, d: int = DEFAULT_DIM):
        self.d = d
        # Encoding is capped at 2000 bits regardless of the machine's d: the
        # assistant projects ~330 words at startup, and at d=10000 that alone
        # took 21 s before the first response. 2000 keeps it under a second and
        # is the codec's documented floor.
        self.enc_d = min(d, 2000)
        self.lock = threading.Lock()
        self.encoder = load_semantic(self.enc_d)
        self.kernel = Kernel(d=d)
        self.assistant = Assistant(self.kernel, d=self.enc_d, encoder=self.encoder)
        # idle + learning + readahead over the same assistant; off until asked
        self.autonomy = AutonomousMode(self.assistant, d=self.enc_d, enabled=False)
        # the webOS shares this kernel's storage, so the desktop's filesystem and
        # memory map describe the machine the terminal is running on
        self.desktop = Desktop(d=d, kernel=self.kernel, autonomy=self.autonomy)
        # POST -> shader splash -> AUTOEXEC, run once; the AUTOEXEC lines are the
        # ones that launch the windows, so the desktop handed over is the result
        self.boot_sequence = BootSequence(self.desktop, d=d)
        self.boot_frames = self.boot_sequence.run()
        self.boot = Boot()
        self.boot_log: list[str] = []
        self._diag_cache: dict[str, str] = {}
        self._cold_boot()

    def _cold_boot(self) -> None:
        self.boot = Boot()
        self.boot.post()
        result = self.boot.boot()
        self.boot_log = [line for line in result["screen"] if line.strip()]
        self.kernel.boot(self.boot)
        self.boot_result = result

    # ---------------------------------------------------------------- views
    def state(self) -> dict:
        return {
            "boot": self.boot_log,
            "screen": self.kernel.term.screen(),
            "status": "kernel running | d=%d | cells=%d | pc=%d" % (
                self.d, len(self.kernel.storage), self.kernel.machine.pc),
            "booted": self.kernel.booted,
        }

    def exec(self, line: str) -> dict:
        with self.lock:
            output = self.kernel.execute(line)
        return {"output": output, "screen": self.kernel.term.screen()}

    def chat(self, text: str) -> dict:
        with self.lock:
            # through the autonomy layer: every turn is graded against the
            # prediction made for it, and a readahead hit is served from cache
            record = self.autonomy.chat(text)
        return dict(record, screen=self.kernel.term.screen())

    def chat_history(self) -> list[dict]:
        return self.assistant.history[-12:]

    def autonomy_state(self) -> dict:
        return {"stats": self.autonomy.stats(), "report": self.autonomy.report()}

    # ------------------------------------------------------------ the webOS
    def os_state(self) -> dict:
        with self.lock:
            return self.desktop.state()

    def os_key(self, key: str) -> dict:
        with self.lock:
            self.desktop.key(str(key))
            return self.desktop.state()

    def os_text(self, text: str) -> dict:
        with self.lock:
            self.desktop.text(str(text))
            return self.desktop.state()

    def os_click(self, x: int, y: int) -> dict:
        with self.lock:
            self.desktop.click(int(x), int(y))
            return self.desktop.state()

    def os_launch(self, app: str) -> dict:
        with self.lock:
            self.desktop.launch(str(app))
            return self.desktop.state()

    def boot_state(self) -> dict:
        return self.boot_frames

    def set_autonomy(self, enabled: bool) -> dict:
        with self.lock:
            self.autonomy.enabled = bool(enabled)
            if enabled:
                self.autonomy.idle(2)      # two ticks: learn, then predict
        return self.autonomy_state()

    def pipeline(self) -> dict:
        kernel = self.kernel
        stages = [
            ("1 BIOS / POST", "16-bit real mode, IVT installed, text buffer at B800",
             "stage=%s, %d instructions" % (self.boot_result["stage"],
                                            self.boot_result["instructions"])),
            ("2 Boot sector", "hand-assembled 512 bytes; int 10h teletype, int 21h handoff",
             "signature 55 AA verified, DS=07C0"),
            ("3 Kernel", "HDC interpreter took over at the handoff",
             "cells=%d pc=%d" % (len(kernel.storage), kernel.machine.pc)),
            ("4 Assembler", "text -> hypervector instruction image, verified by decoding back",
             "program=%s" % ("loaded" if kernel.program else "none")),
            ("5 Machine", "fetch from storage, decode by cleanup, execute",
             "steps=%d status=%s" % (kernel.machine.steps, kernel.machine.status)),
            ("6 Tools", "MCP tool records resolved by similarity",
             "%d registered" % len(kernel.tools.list_tools())),
            ("7 Shader", "coded-image operators: dim, edges, normals, depth",
             "fabric d=%d" % self.d),
            ("8 State", "predictive transitions, beam planner, surprise",
             "planner horizon 3"),
            ("9 Ring", "historic ledger, PAST->PRESENT->FUTURE, energy budget",
             "ledger capacity 16"),
            ("10 Text / semantic", "TextRing encoder, optional static embeddings",
             "encoder=%s" % ("semantic (potion-base-8M)" if self.encoder else "character n-gram")),
            ("11 Assistant", "prototype classification over encoded utterances",
             "%d intents, %d trained" % (len(INTENTS), self.assistant.trained)),
        ]
        return {"stages": [{"name": n, "detail": d_, "metric": m} for n, d_, m in stages]}

    DIAG = ("help", "isa", "stat", "net", "shader", "memory modes",
            "state", "ring", "text", "semantic", "topology", "hub", "autonomy")

    def diagnostics(self, name: str) -> dict:
        """One diagnostics section, cached.

        Computing all eleven at once took 9 s at d=10000 and the browser just
        sat there.  One section at a time keeps every click responsive, and
        the cache makes a repeat click instant.
        """
        if name in self._diag_cache:
            return {"name": name, "body": self._diag_cache[name], "cached": True}
        body = self._compute_diagnostic(name)
        self._diag_cache[name] = body
        return {"name": name, "body": body, "cached": False}

    def _compute_diagnostic(self, name: str) -> str:
        from .isa import opcode_table_text
        from .memory_modes import demo_report as memory_report
        from .net import selfcheck
        from .ring import report as ring_report
        from .semantic import report as semantic_report
        from .shader import demo_report as shader_report
        from .state import report as state_report
        from .terminal import HELP
        from .text import report as text_report
        from .topology import report as topology_report

        if name == "hub":
            # offline: the cache from `python -m hdc hub` is used, never the
            # network — a diagnostics click must not wait on the Hub
            from .hub import report as hub_report

            return hub_report(offline=True)
        if name == "autonomy":
            from .autonomy import report as autonomy_report

            return autonomy_report(d=self.enc_d)

        if name == "help":
            return HELP
        if name == "isa":
            return opcode_table_text()
        if name == "stat":
            return self.kernel.execute("stat")
        if name == "net":
            return "\n".join("%-22s %s" % (k, v) for k, v in selfcheck().items())
        # the remaining reports run at the codec floor: they are measurements
        # of the algorithms, not of this machine's dimension
        if name == "shader":
            return shader_report(d=self.enc_d)
        if name == "memory modes":
            return memory_report(d=self.enc_d)
        if name == "state":
            return state_report(d=self.enc_d)
        if name == "ring":
            return ring_report(d=self.enc_d)
        if name == "text":
            return text_report(d=self.enc_d)
        if name == "semantic":
            return semantic_report(d=self.enc_d)
        if name == "topology":
            return topology_report(d=self.enc_d)
        raise KeyError(name)

    # ------------------------------------------------------------ visuals
    VIZ = ("shader2d", "shader3d", "shader4d", "fractal", "crystal",
           "manifold", "semantic", "ring", "state")

    def visualisation(self, name: str) -> dict:
        from .shader import ShaderFabric, step_image
        from .shader3d import Volume, cube_volume
        from .shader4d import sweep_volume

        d = self.d
        fabric = ShaderFabric(d=d)
        if name == "shader2d":
            image = step_image(fabric=fabric)
            body = "\n".join([
                "source:", image.ascii(),
                "", "edges (exact |a-b|):", image.edges().ascii(),
                "", "dim(1) — exact halving:", image.dim(1).ascii(),
                "", "dilate(2) then bleed(2):", image.dilate(2).bleed(2).ascii(),
            ])
            return {"title": "2D shader operators", "body": body}
        if name == "shader3d":
            volume = cube_volume(5, 180, fabric)
            return {"title": "3D volume: shaded+projected, curvature",
                    "body": "shaded projection:\n" + volume.ascii_render()
                            + "\n\nslice z=2:\n" + volume.ascii_slice(2)
                            + "\n\nedges:\n" + volume.edges().ascii_slice(2)}
        if name == "shader4d":
            vol = sweep_volume(step=6, size=5, value=200, fabric=fabric)
            entropy = vol.entropy_along_u(8)
            return {"title": "4D entropic stack",
                    "body": "slice u=0:\n" + vol.ascii_u(0)
                            + "\n\nslice u=3:\n" + vol.ascii_u(3)
                            + "\n\nentropy along u (z=2):\n" + entropy.ascii_slice(2)
                            + "\n\ndrift %.4f" % vol.drift()}
        if name == "fractal":
            from .fractal import fbm, fern, sierpinski

            return {"title": "fractal: Sierpinski, fBm, fern",
                    "body": "sierpinski(5):\n" + sierpinski(5, fabric=fabric).ascii()
                            + "\n\nfBm 32x32:\n" + fbm(32, 32, 4, 8, 7, fabric).ascii()
                            + "\n\nfern(3000):\n" + fern(3000, 34, 36, fabric).ascii()}
        if name == "crystal":
            from .crystal import grain_boundaries, lattice_shade, report

            volume = cube_volume(5, 180, fabric)
            return {"title": "crystal: grains + lattice banding",
                    "body": report(volume) + "\n\nlattice banding z=2:\n"
                            + lattice_shade(volume, spacing=2).ascii_slice(2)}
        if name == "manifold":
            from .manifold import curvature, geodesic_field, report

            rows = [[[180 if (x in (0, 4) or y in (0, 4) or z in (0, 4)) else 0
                      for x in range(5)] for y in range(5)] for z in range(5)]
            volume = Volume.from_levels(rows, fabric)
            return {"title": "manifold: geodesic + curvature",
                    "body": report(volume, seed=(0, 0, 0))
                            + "\n\ngeodesic from (0,0,0) at z=0:\n"
                            + geodesic_field(volume, (0, 0, 0)).ascii_slice(0)}
        if name == "semantic":
            encoder = self.encoder
            if encoder is None:
                return {"title": "semantic", "body": "model not downloaded — see hdc/semantic.py"}
            words = ["car", "automobile", "doctor", "physician", "banana", "tectonics"]
            header = "%-11s" % "" + " ".join("%-10s" % w[:10] for w in words)
            rows = [header]
            for a in words:
                cells = ["%+.3f" % encoder.similarity(a, b) for b in words]
                rows.append("%-11s" % a[:11] + " ".join("%-10s" % c for c in cells))
            return {"title": "semantic similarity (model cosine)",
                    "body": "\n".join(rows)}
        if name == "ring":
            from .ring import report

            return {"title": "ring controller", "body": report(d=d)}
        if name == "state":
            from .state import report

            return {"title": "predictive state", "body": report(d=d)}
        raise KeyError(name)


def make_handler(app: App):
    class Handler(BaseHTTPRequestHandler):
        server_version = "HDCSTORAGE/1.0"

        def log_message(self, fmt, *args):   # quiet: this is a local panel
            pass

        def _send(self, payload: bytes, content_type: str, code: int = 200) -> None:
            self.send_response(code)
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(len(payload)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(payload)

        def _json(self, data, code: int = 200) -> None:
            self._send(json.dumps(data).encode("utf-8"), "application/json", code)

        def _body(self) -> dict:
            length = int(self.headers.get("Content-Length") or 0)
            if not length:
                return {}
            try:
                return json.loads(self.rfile.read(length).decode("utf-8"))
            except json.JSONDecodeError:
                return {}

        def do_GET(self) -> None:                       # noqa: N802
            path = self.path.split("?")[0]
            if path in ("/", "/index.html"):
                self._send(PAGE.encode("utf-8"), "text/html; charset=utf-8")
            elif path == "/api/state":
                self._json(app.state())
            elif path == "/api/chat/history":
                self._json(app.chat_history())
            elif path == "/api/autonomy":
                self._json(app.autonomy_state())
            elif path == "/api/os":
                self._json(app.os_state())
            elif path == "/api/boot":
                self._json(app.boot_state())
            elif path == "/api/diag":
                self._json({"sections": list(App.DIAG)})
            elif path.startswith("/api/diag/"):
                try:
                    self._json(app.diagnostics(path.rsplit("/", 1)[1]))
                except KeyError:
                    self._json({"error": "unknown section"}, 404)
            elif path == "/api/pipeline":
                self._json(app.pipeline())
            elif path == "/api/viz":
                self._json({"list": list(App.VIZ)})
            elif path.startswith("/api/viz/"):
                try:
                    self._json(app.visualisation(path.rsplit("/", 1)[1]))
                except KeyError:
                    self._json({"error": "unknown visualisation"}, 404)
            else:
                self._json({"error": "not found"}, 404)

        def do_POST(self) -> None:                      # noqa: N802
            path = self.path.split("?")[0]
            body = self._body()
            if path == "/api/exec":
                self._json(app.exec(str(body.get("line", ""))))
            elif path == "/api/chat":
                self._json(app.chat(str(body.get("text", ""))))
            elif path == "/api/autonomy":
                self._json(app.set_autonomy(bool(body.get("enabled"))))
            elif path == "/api/os/key":
                self._json(app.os_key(str(body.get("key", ""))))
            elif path == "/api/os/text":
                self._json(app.os_text(str(body.get("text", ""))))
            elif path == "/api/os/click":
                self._json(app.os_click(int(body.get("x", 0)), int(body.get("y", 0))))
            elif path == "/api/os/launch":
                self._json(app.os_launch(str(body.get("app", "terminal"))))
            elif path == "/api/boot":
                with app.lock:
                    app._cold_boot()
                self._json(app.state())
            else:
                self._json({"error": "not found"}, 404)

    return Handler


def main(argv: list[str] | None = None) -> int:
    """``python -m hdc web [--port 8765] [--d 2000] [--open]``."""
    import sys
    import webbrowser

    argv = list(sys.argv[1:] if argv is None else argv)
    port = 8765
    host = "127.0.0.1"
    # 2000 is the codec floor and keeps every endpoint sub-second; the CLI's
    # 10 000-bit default made assembling a demo take seconds in the UI
    d = 2000
    open_browser = False
    i = 0
    while i < len(argv):
        if argv[i] == "--port" and i + 1 < len(argv):
            port = int(argv[i + 1]); i += 2
        elif argv[i] == "--d" and i + 1 < len(argv):
            d = int(argv[i + 1]); i += 2
        elif argv[i] == "--host" and i + 1 < len(argv):
            host = argv[i + 1]; i += 2
        elif argv[i] == "--open":
            open_browser = True; i += 1
        else:
            i += 1
    app = App(d=d)
    server = ThreadingHTTPServer((host, port), make_handler(app))
    url = "http://%s:%d/" % (host, port)
    print("HDCSTORAGE WebUI on %s  (d=%d, encoder=%s)"
          % (url, d, "semantic" if app.encoder else "n-gram"))
    print("  ctrl-c to stop")
    if open_browser:
        webbrowser.open(url)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nstopped")
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":  # pragma: no cover - manual run
    raise SystemExit(main())