mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 10:59:59 +00:00
kevinveenbirkenbachandClaude Opus 5
f320d64836
A bond lives on the consumer's side of a service relation, so seeing how two roles couple means opening both roles' meta/services.yml and comparing, and retuning one means editing a scalar buried among dozens of unrelated keys. Add `infinito meta roles applications bond` (make bond / i8bond): it resolves every services.<key>.bond to a role-to-role edge through the service registry and serves the result as a matrix, each cell carrying both directions because the two need not agree. Brightness is the bond, black for none and white for 1, so the coupling of the whole repository reads at a glance. Clicking a cell writes the new value back to the declaring role. The write is line surgery rather than a YAML round-trip: these files carry `# nocheck:` directives and hand-tuned ordering that re-dumping would flatten, so only the bond scalar moves and a trailing comment on that line survives. The browser sends a role pair, never a path, and the server resolves the service key from its own edge map, so a pair it does not know is refused before any write. `--out FILE` exports a read-only page and `--format json` the raw edges, for callers that want to post-process rather than edit. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
"""Serve the bond matrix on localhost and write every cell edit back to disk.
|
|
|
|
The edge map is built once at startup and kept in step with the writes this
|
|
server performs, so a role file edited by hand elsewhere needs a restart.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from .edit import EditError, parse_bond, set_bond
|
|
from .matrix import participants
|
|
from .render import render, shade
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
MAX_BODY = 4096
|
|
|
|
|
|
class _Handler(BaseHTTPRequestHandler):
|
|
server_version = "bond-matrix"
|
|
roles_dir: Path
|
|
edges: dict[tuple[str, str], dict[str, Any]]
|
|
|
|
def log_message(self, fmt: str, *args: Any) -> None:
|
|
return
|
|
|
|
def _send(self, code: int, body: bytes, content_type: str) -> None:
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _json(self, code: int, payload: dict[str, Any]) -> None:
|
|
self._send(code, json.dumps(payload).encode(), "application/json")
|
|
|
|
def do_GET(self) -> None:
|
|
if self.path.split("?")[0] not in ("/", "/index.html"):
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
return
|
|
page = render(self.edges, participants(self.edges), editable=True)
|
|
self._send(200, page.encode(), "text/html; charset=utf-8")
|
|
|
|
def do_POST(self) -> None:
|
|
if self.path.split("?")[0] != "/bond":
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
return
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
if length > MAX_BODY:
|
|
self._json(413, {"ok": False, "error": "body too large"})
|
|
return
|
|
try:
|
|
payload = json.loads(self.rfile.read(length) or b"{}")
|
|
key = (str(payload["consumer"]), str(payload["provider"]))
|
|
raw = payload["value"]
|
|
except (ValueError, KeyError, TypeError):
|
|
self._json(400, {"ok": False, "error": "malformed request"})
|
|
return
|
|
|
|
edge = self.edges.get(key)
|
|
if edge is None:
|
|
self._json(404, {"ok": False, "error": f"no bond {key[0]} -> {key[1]}"})
|
|
return
|
|
|
|
try:
|
|
bond = parse_bond(raw)
|
|
text = set_bond(self.roles_dir, key[0], edge["service_key"], bond)
|
|
except EditError as exc:
|
|
self._json(400, {"ok": False, "error": str(exc)})
|
|
return
|
|
|
|
if bond is None:
|
|
del self.edges[key]
|
|
background, foreground = "#000", "#fff"
|
|
else:
|
|
edge["bond"] = bond
|
|
background, foreground = shade(bond)
|
|
written = f"{key[0]}/{edge['service_key']}.bond = {text or 'removed'}"
|
|
self._json(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"text": text,
|
|
"bg": background,
|
|
"fg": foreground,
|
|
"written": written,
|
|
},
|
|
)
|
|
|
|
|
|
def serve(
|
|
roles_dir: Path,
|
|
edges: dict[tuple[str, str], dict[str, Any]],
|
|
port: int,
|
|
) -> tuple[ThreadingHTTPServer, str]:
|
|
"""Return a server bound to loopback and the URL it listens on.
|
|
|
|
Args:
|
|
roles_dir: directory the writes land in.
|
|
edges: the collected edge map, mutated in place as edits arrive.
|
|
port: the port to bind, or 0 to let the OS pick one.
|
|
"""
|
|
handler = type(
|
|
"_BoundHandler", (_Handler,), {"roles_dir": roles_dir, "edges": edges}
|
|
)
|
|
httpd = ThreadingHTTPServer(("127.0.0.1", port), handler)
|
|
return httpd, f"http://127.0.0.1:{httpd.server_address[1]}/"
|