luulu
ad648b4ea6
Part one of the two-part plan, closing out the userspace substrate before the kernel work: - PLAN-kernel-cubelinux.md: the decision (the kernel owns the store; coordinates are resolved in the kernel; the block driver is persistence plumbing; POSIX is an edge view) plus the two-part plan and the one design fork left for the write-authority phase. - README.md: what CUBE-OS is, the crate map, quickstart, the release gates, and the "claims not yet true" list (kernel not coordinate-native yet; installed kernel is CUBED's). - Superseded banners on the pre-split docs (PLAN-os-layer.md, INTEGRATION-cube-agent.md, PLAN-debugging.md, cube_harness.py) and the fixed stale rows (DESIGN-coordinate-layer.md, a cube-header comment). - cube_ctl.py: the `spawn` subcommand and CubeSpawner, which shelled the deleted cube-spawn crate, now refuse with a clear message instead of pretending. Release checklist green: 157 tests, 0 clippy warnings, image verifies (snapshot 8,399,506 B; live 16,537,516 B), cube-core cross-builds for thumbv7em-none-eabihf, durability harness PASS, and the name-tier path end to end. Tagged cubelinux-0.1.0.
578 lines
20 KiB
Python
578 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
cube_ctl.py — Python control plane for CUBELinux cubes.
|
|
|
|
Provides:
|
|
- DaemonClient: speaks the newline-delimited JSON protocol to a `cubed`
|
|
Unix socket (connect/read/write/delete/link/follow/header/adjacency/stats).
|
|
- CubeSpawner: thin wrapper around `cube-spawn` via a subprocess or a planned
|
|
pyo3 bridge; for now it shells `cubecli`/cargo-run helpers.
|
|
- CubeHarness: use-case runners (MachineCodeTracker, FeatureBallBench,
|
|
SessionReplayBench, RawBlockFuseBench).
|
|
|
|
This is Plan A's control plane. It reuses the protocol already documented in
|
|
`cube-daemon/src/lib.rs` — one JSON object per newline, {"cmd": "...", ...}
|
|
requests, {"ok": true, ...} / {"ok": false, "error": "..."} responses,
|
|
binary values base64-encoded, coordinates as "<64-hex-space>:<x>,<y>,<z>".
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Optional
|
|
|
|
|
|
def b64_encode(data: bytes) -> str:
|
|
return base64.b64encode(data).decode("ascii")
|
|
|
|
|
|
def b64_decode(s: str) -> bytes:
|
|
return base64.b64decode(s)
|
|
|
|
|
|
def coord_to_str(space_hex: str, x: int, y: int, z: int) -> str:
|
|
"""Format a coordinate as the daemon expects: "<64-hex-space>:<x>,<y>,<z>"."""
|
|
return f"{space_hex}:{x},{y},{z}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DaemonClient — JSON-over-Unix-socket client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class DaemonError(Exception):
|
|
"""Raised when the daemon returns {"ok": false, ...}."""
|
|
def __init__(self, cmd: str, error: str, response: dict):
|
|
self.cmd = cmd
|
|
self.error = error
|
|
self.response = response
|
|
super().__init__(f"{cmd}: {error}")
|
|
|
|
|
|
class DaemonClient:
|
|
"""Connects to a `cubed` Unix socket and speaks the JSON protocol.
|
|
|
|
Usage:
|
|
c = DaemonClient(socket_path="/run/cube/cube.sock")
|
|
c.write(namespace="MEM", name="test", value=b"hello")
|
|
data = c.read(namespace="MEM", name="test")
|
|
c.delete(namespace="MEM", name="test")
|
|
"""
|
|
|
|
def __init__(self, socket_path: str | Path):
|
|
self.socket_path = str(socket_path)
|
|
self._sock: Optional[socket.socket] = None
|
|
self._lock = threading.Lock()
|
|
|
|
def _ensure_connected(self) -> socket.socket:
|
|
if self._sock is None or self._sock.fileno() == -1:
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.settimeout(10.0)
|
|
sock.connect(self.socket_path)
|
|
self._sock = sock
|
|
return self._sock
|
|
|
|
def _send_request(self, cmd: str, **kwargs) -> dict:
|
|
"""Send one JSON request line and return the parsed response dict."""
|
|
payload = {"cmd": cmd}
|
|
payload.update(kwargs)
|
|
line = json.dumps(payload, ensure_ascii=False) + "\n"
|
|
with self._lock:
|
|
sock = self._ensure_connected()
|
|
sock.sendall(line.encode("utf-8"))
|
|
# Read one line back
|
|
buf = b""
|
|
while not buf.endswith(b"\n"):
|
|
chunk = sock.recv(4096)
|
|
if not chunk:
|
|
raise OSError("daemon socket closed while reading response")
|
|
buf += chunk
|
|
resp = json.loads(buf.decode("utf-8"))
|
|
if not resp.get("ok"):
|
|
raise DaemonError(cmd, resp.get("error", "unknown error"), resp)
|
|
return resp
|
|
|
|
def ping(self) -> dict:
|
|
return self._send_request("ping")
|
|
|
|
def write(
|
|
self,
|
|
namespace: str,
|
|
name: str,
|
|
value: bytes | str,
|
|
kind: Optional[str] = None,
|
|
visibility: Optional[str] = None,
|
|
) -> dict:
|
|
if isinstance(value, str):
|
|
value = value.encode("utf-8")
|
|
return self._send_request(
|
|
"write",
|
|
namespace=namespace,
|
|
name=name,
|
|
value=b64_encode(value),
|
|
kind=kind,
|
|
visibility=visibility,
|
|
)
|
|
|
|
def read(self, namespace: str, name: str) -> dict:
|
|
resp = self._send_request("read", namespace=namespace, name=name)
|
|
result: dict = {}
|
|
if resp.get("value") is not None:
|
|
result["value"] = b64_decode(resp["value"])
|
|
if resp.get("coord") is not None:
|
|
result["coord"] = resp["coord"]
|
|
return result
|
|
|
|
def delete(self, namespace: str, name: str) -> dict:
|
|
return self._send_request("delete", namespace=namespace, name=name)
|
|
|
|
def list(self, namespace: str) -> dict:
|
|
resp = self._send_request("list", namespace=namespace)
|
|
entries = []
|
|
for entry in resp.get("entries", []):
|
|
e = {"coord": entry.get("coord")}
|
|
if entry.get("value") is not None:
|
|
e["value"] = b64_decode(entry["value"])
|
|
entries.append(e)
|
|
return {"entries": entries}
|
|
|
|
def link(self, from_ns: str, from_name: str, to_ns: str, to_name: str) -> dict:
|
|
return self._send_request(
|
|
"link",
|
|
from_ns=from_ns,
|
|
from_name=from_name,
|
|
to_ns=to_ns,
|
|
to_name=to_name,
|
|
)
|
|
|
|
def follow(self, namespace: str, name: str, index: int = 0) -> dict:
|
|
resp = self._send_request("follow", namespace=namespace, name=name, index=index)
|
|
result: dict = {}
|
|
if resp.get("target") is not None:
|
|
result["target"] = resp["target"]
|
|
if resp.get("value") is not None:
|
|
result["value"] = b64_decode(resp["value"])
|
|
return result
|
|
|
|
def header(self, namespace: str, name: str) -> dict:
|
|
return self._send_request("header", namespace=namespace, name=name)
|
|
|
|
def adjacency(self, namespace: str, name: str) -> dict:
|
|
resp = self._send_request("adjacency", namespace=namespace, name=name)
|
|
return {
|
|
"outgoing": resp.get("outgoing", []),
|
|
"incoming": resp.get("incoming", []),
|
|
}
|
|
|
|
def stats(self) -> dict:
|
|
return self._send_request("stats")
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
if self._sock is not None:
|
|
try:
|
|
self._sock.close()
|
|
except Exception:
|
|
pass
|
|
self._sock = None
|
|
|
|
def __enter__(self) -> "DaemonClient":
|
|
return self
|
|
|
|
def __exit__(self, *args: Any) -> None:
|
|
self.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CubeSpawner — thin wrapper to invoke the Rust registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class CubeSpawner:
|
|
"""REMOVED from CUBE OS: `cube-spawn` and the VM/spawn tier moved to CUBED.
|
|
|
|
Kept only so the reference is explicit; the `spawn` command refuses with
|
|
this same message.
|
|
|
|
Spins up named cubes via the `cube-spawn` crate.
|
|
|
|
Currently shells `cargo run -p cube-spawn -- <args>` as a subprocess.
|
|
Long-term: replace with a pyo3-based direct call once cube-spawn exposes
|
|
a C-ABI or Python-binding entry point. Agents already drive everything
|
|
through Python (`cube_bridge.py`), so a Python-facing spawn API is the
|
|
right layer.
|
|
"""
|
|
|
|
def __init__(self, workspace: str = "/home/CUBE-OS"):
|
|
self.workspace = workspace
|
|
self.cargo = os.environ.get("CARGO_BIN", str(Path.home() / ".cargo" / "bin" / "cargo"))
|
|
|
|
def _run(
|
|
self,
|
|
args: list[str],
|
|
*,
|
|
check: bool = True,
|
|
timeout: float = 120.0,
|
|
) -> subprocess.CompletedProcess:
|
|
env = os.environ.copy()
|
|
env["PATH"] = os.pathsep.join([
|
|
str(Path.home() / ".cargo" / "bin"),
|
|
env.get("PATH", ""),
|
|
])
|
|
return subprocess.run(
|
|
[self.cargo, "run", "-p", "cube-spawn", "--"] + args,
|
|
cwd=self.workspace,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=check,
|
|
)
|
|
|
|
def spawn(
|
|
self,
|
|
name: str,
|
|
curve: str = "morton",
|
|
backend: str = "file",
|
|
store_path: Optional[str] = None,
|
|
with_daemon: bool = False,
|
|
socket_path: Optional[str] = None,
|
|
) -> dict:
|
|
"""Spawn a named cube. Returns the daemon/socket info if with_daemon.
|
|
|
|
Args:
|
|
name: human-readable cube name (must be unique in registry).
|
|
curve: one of "morton", "hilbert", "rowmajor".
|
|
backend: one of "file", "memory", "rawblock".
|
|
store_path: path for file/rawblock backends (auto-generated if None).
|
|
with_daemon: if True, launch a `cubed` daemon for this cube.
|
|
socket_path: daemon socket path (auto-generated if None and with_daemon).
|
|
"""
|
|
args = ["spawn", "--name", name, "--curve", curve, "--backend", backend]
|
|
if store_path:
|
|
args += ["--store", store_path]
|
|
if with_daemon:
|
|
args.append("--daemon")
|
|
if socket_path:
|
|
args += ["--socket", socket_path]
|
|
result = self._run(args)
|
|
out = result.stdout.strip()
|
|
try:
|
|
return json.loads(out) if out else {"status": "ok"}
|
|
except json.JSONDecodeError:
|
|
return {"status": "ok", "output": out}
|
|
|
|
def list(self) -> list[str]:
|
|
"""List all live cube names in the registry."""
|
|
result = self._run(["list"])
|
|
try:
|
|
return json.loads(result.stdout.strip()) if result.stdout.strip() else []
|
|
except json.JSONDecodeError:
|
|
return []
|
|
|
|
def get(self, name: str) -> dict:
|
|
"""Get info about a named cube."""
|
|
result = self._run(["get", "--name", name])
|
|
try:
|
|
return json.loads(result.stdout.strip())
|
|
except json.JSONDecodeError:
|
|
return {"error": result.stderr.strip() or result.stdout.strip()}
|
|
|
|
def destroy(self, name: str) -> dict:
|
|
"""Remove a cube from the registry."""
|
|
result = self._run(["destroy", "--name", name])
|
|
try:
|
|
return json.loads(result.stdout.strip()) if result.stdout.strip() else {"status": "ok"}
|
|
except json.JSONDecodeError:
|
|
return {"status": "ok", "output": result.stdout.strip()}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CubeHarness — use-case runners
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class MachineCodeTracker:
|
|
"""Write cubecode bytecode payloads into a cube and run them.
|
|
|
|
cubecode is the bytecode VM (276ns entry / 142ns leaf per the whitepaper).
|
|
This harness makes it scriptable: you compose a bytecode program, write it
|
|
into a dedicated SES/MEM cube, and invoke the daemon to run it.
|
|
"""
|
|
|
|
def __init__(self, client: DaemonClient, namespace: str = "MEM"):
|
|
self.client = client
|
|
self.namespace = namespace
|
|
|
|
def write_payload(self, name: str, bytecode: bytes, meta: Optional[dict] = None) -> dict:
|
|
"""Write a bytecode payload into the cube."""
|
|
payload = {"bytecode": b64_encode(bytecode)}
|
|
if meta:
|
|
payload["meta"] = meta
|
|
return self.client.write(
|
|
namespace=self.namespace,
|
|
name=name,
|
|
value=json.dumps(payload).encode("utf-8"),
|
|
kind="bytecode",
|
|
)
|
|
|
|
def read_payload(self, name: str) -> dict:
|
|
"""Read back a bytecode payload and decode it.
|
|
|
|
Handles two storage formats:
|
|
- Envelope: {"bytecode": <base64>} (what write_payload writes)
|
|
- Flat: {"op": ..., ...} (direct JSON storage)
|
|
"""
|
|
result = self.client.read(namespace=self.namespace, name=name)
|
|
raw = result.get("value")
|
|
if raw is None:
|
|
return {"error": "not found"}
|
|
raw_bytes = raw if isinstance(raw, bytes) else raw.encode("utf-8")
|
|
try:
|
|
outer = json.loads(raw_bytes.decode("utf-8"))
|
|
# Unwrap the {"bytecode": <base64>} envelope that write_payload writes
|
|
if "bytecode" in outer and isinstance(outer["bytecode"], str):
|
|
inner_bytes = b64_decode(outer["bytecode"])
|
|
inner = json.loads(inner_bytes.decode("utf-8"))
|
|
return inner
|
|
return outer
|
|
except (json.JSONDecodeError, KeyError, UnicodeDecodeError):
|
|
return {"raw": raw_bytes}
|
|
|
|
def run_payload(self, name: str) -> dict:
|
|
"""Invoke the daemon to run a bytecode payload.
|
|
|
|
Note: actual execution is via the daemon's `run` command (not yet
|
|
implemented in the JSON protocol). This is the hook once that exists.
|
|
"""
|
|
# Placeholder — the daemon protocol currently has no `run` command.
|
|
# Once cube-daemon adds `Run { namespace, name }` → `{ok, output}`,
|
|
# this call becomes real.
|
|
return {"status": "pending", "note": "daemon Run command not yet in protocol"}
|
|
|
|
|
|
class FeatureBallBench:
|
|
"""Benchmark feature-ball recall on a Hilbert vs Morton cube.
|
|
|
|
Writes quantized feature vectors into a cube and runs nearest-neighbour
|
|
queries. Measures bytes/seeks per the whitepaper's unproven-claim #1.
|
|
"""
|
|
|
|
def __init__(self, client: DaemonClient, namespace: str = "PRJ"):
|
|
self.client = client
|
|
self.namespace = namespace
|
|
self.vectors: list[dict] = []
|
|
|
|
def write_vector(self, fid: int, vector: list[float], curve_label: str = "morton") -> dict:
|
|
"""Write a quantized feature vector as a coordinate-named record."""
|
|
data = {
|
|
"fid": fid,
|
|
"vector": vector,
|
|
"curve": curve_label,
|
|
"timestamp": time.time(),
|
|
}
|
|
name = f"vec-{fid:08d}"
|
|
return self.client.write(
|
|
namespace=self.namespace,
|
|
name=name,
|
|
value=json.dumps(data).encode("utf-8"),
|
|
kind="feature-vector",
|
|
)
|
|
|
|
def nearest_neighbour(self, query: list[float], k: int = 1) -> list[dict]:
|
|
"""Run a nearest-neighbour query (placeholder — real impl needs
|
|
the adjacency index / prefix scan from the daemon)."""
|
|
# The real implementation drives `list` on the namespace and computes
|
|
# cosine/Euclidean distance in Python against the stored vectors.
|
|
entries = self.client.list(namespace=self.namespace).get("entries", [])
|
|
results = []
|
|
for entry in entries:
|
|
val = entry.get("value", b"")
|
|
try:
|
|
vec = json.loads(val.decode("utf-8"))
|
|
dist = self._cosine_dist(query, vec["vector"])
|
|
results.append({"fid": vec["fid"], "dist": dist})
|
|
except (json.JSONDecodeError, KeyError):
|
|
continue
|
|
results.sort(key=lambda r: r["dist"])
|
|
return results[:k]
|
|
|
|
@staticmethod
|
|
def _cosine_dist(a: list[float], b: list[float]) -> float:
|
|
dot = sum(x * y for x, y in zip(a, b))
|
|
na = sum(x * x for x in a) ** 0.5
|
|
nb = sum(y * y for y in b) ** 0.5
|
|
if na == 0 or nb == 0:
|
|
return float("inf")
|
|
return 1.0 - dot / (na * nb)
|
|
|
|
|
|
class SessionReplayBench:
|
|
"""Append+replay at control-loop scale (whitepaper unproven claim #3)."""
|
|
|
|
def __init__(self, client: DaemonClient, namespace: str = "SES"):
|
|
self.client = client
|
|
self.namespace = namespace
|
|
self.events: list[dict] = []
|
|
|
|
def append_event(self, event: dict) -> dict:
|
|
"""Append a session event record."""
|
|
event["ts"] = time.time()
|
|
name = f"evt-{event.get('tick', 0):08d}"
|
|
return self.client.write(
|
|
namespace=self.namespace,
|
|
name=name,
|
|
value=json.dumps(event).encode("utf-8"),
|
|
kind="session-event",
|
|
)
|
|
|
|
def replay(self, from_tick: int = 0, to_tick: int = 999999) -> list[dict]:
|
|
"""Replay events in tick order."""
|
|
entries = self.client.list(namespace=self.namespace).get("entries", [])
|
|
events = []
|
|
for entry in entries:
|
|
val = entry.get("value", b"")
|
|
try:
|
|
evt = json.loads(val.decode("utf-8"))
|
|
tick = evt.get("tick", 0)
|
|
if from_tick <= tick <= to_tick:
|
|
events.append(evt)
|
|
except (json.JSONDecodeError, KeyError):
|
|
continue
|
|
events.sort(key=lambda e: e.get("tick", 0))
|
|
return events
|
|
|
|
|
|
class RawBlockFuseBench:
|
|
"""Mount a RawBlock cube via FUSE, write/read files.
|
|
|
|
Requires M2(a) FUSE proxy completion first. Placeholder for now.
|
|
"""
|
|
|
|
def __init__(self, store_path: str, mount_point: str = "/tmp/cube-fuse-mount"):
|
|
self.store_path = store_path
|
|
self.mount_point = mount_point
|
|
self._mounted = False
|
|
|
|
def mount(self) -> bool:
|
|
"""Mount the FUSE proxy (placeholder — needs M2a)."""
|
|
# Once cube-fuse-proxy is complete, this calls the FUSE binary.
|
|
self._mounted = False
|
|
return False
|
|
|
|
def unmount(self) -> bool:
|
|
"""Unmount."""
|
|
self._mounted = False
|
|
return True
|
|
|
|
def write_file(self, path: str, data: bytes) -> bool:
|
|
"""Write a file through FUSE (placeholder)."""
|
|
if not self._mounted:
|
|
return False
|
|
full = os.path.join(self.mount_point, path)
|
|
os.makedirs(os.path.dirname(full), exist_ok=True)
|
|
Path(full).write_bytes(data)
|
|
return True
|
|
|
|
def read_file(self, path: str) -> Optional[bytes]:
|
|
"""Read a file back."""
|
|
if not self._mounted:
|
|
return None
|
|
full = os.path.join(self.mount_point, path)
|
|
if Path(full).exists():
|
|
return Path(full).read_bytes()
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Convenience: connect to the default daemon
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DEFAULT_SOCKET = (
|
|
os.environ.get("CUBE_SOCKET") or "/run/user/1000/cubelinux/cubed-names.sock"
|
|
)
|
|
|
|
|
|
def connect(socket_path: Optional[str] = None) -> DaemonClient:
|
|
return DaemonClient(socket_path=socket_path or DEFAULT_SOCKET)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI entry point (for ad-hoc testing)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main() -> None:
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="CUBELinux cube control plane")
|
|
sub = parser.add_subparsers(dest="command")
|
|
|
|
# ping
|
|
p = sub.add_parser("ping", help="health check the daemon")
|
|
p.add_argument("--socket", default=DEFAULT_SOCKET)
|
|
|
|
# write
|
|
p = sub.add_parser("write", help="write a value")
|
|
p.add_argument("namespace")
|
|
p.add_argument("name")
|
|
p.add_argument("value", help="raw bytes (string)")
|
|
p.add_argument("--socket", default=DEFAULT_SOCKET)
|
|
|
|
# read
|
|
p = sub.add_parser("read", help="read a value")
|
|
p.add_argument("namespace")
|
|
p.add_argument("name")
|
|
p.add_argument("--socket", default=DEFAULT_SOCKET)
|
|
|
|
# list
|
|
p = sub.add_parser("list", help="list entries in a namespace")
|
|
p.add_argument("namespace")
|
|
p.add_argument("--socket", default=DEFAULT_SOCKET)
|
|
|
|
# spawn (via cube-spawn crate) — REMOVED from CUBE OS; see CUBED.
|
|
p = sub.add_parser("spawn", help="spawn a named cube (removed: lives in CUBED)")
|
|
p.add_argument("name")
|
|
p.add_argument("--curve", default="morton")
|
|
p.add_argument("--backend", default="file")
|
|
p.add_argument("--store", default=None)
|
|
p.add_argument("--daemon", action="store_true")
|
|
p.add_argument("--socket-path", default=None)
|
|
|
|
args = parser.parse_args()
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
if args.command == "ping":
|
|
with connect(args.socket) as c:
|
|
print(json.dumps(c.ping(), indent=2))
|
|
elif args.command == "write":
|
|
with connect(args.socket) as c:
|
|
c.write(namespace=args.namespace, name=args.name, value=args.value)
|
|
print(f"wrote {args.namespace}/{args.name}")
|
|
elif args.command == "read":
|
|
with connect(args.socket) as c:
|
|
result = c.read(namespace=args.namespace, name=args.name)
|
|
value = result.get("value", b"")
|
|
try:
|
|
print(value.decode("utf-8"))
|
|
except UnicodeDecodeError:
|
|
print(value.hex())
|
|
elif args.command == "list":
|
|
with connect(args.socket) as c:
|
|
result = c.list(namespace=args.namespace)
|
|
print(json.dumps(result.get("entries", []), indent=2))
|
|
elif args.command == "spawn":
|
|
print("spawn removed from CUBE OS: the VM/spawn tier lives in CUBED", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|