File size: 5,947 Bytes
587a933 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """
KanBot web view — live board over Postgres LISTEN/NOTIFY.
Local-first, dependency-light: a single asyncio server that
- serves the static branded SPA (index.html) over HTTP,
- LISTENs on 'kb_events' and pushes every transition to connected browsers
over a raw WebSocket (RFC 6455, hand-rolled — no node, no npm, no build step),
- answers a tiny JSON snapshot endpoint so a fresh page loads the full board.
Why not Next.js: this is a dashboard, not a marketing site. One file, runs on the
M4 with the same asyncpg the rest of the stack uses. Swap it for a Next app later
without touching the core — it's just another MCP-contract client.
Run: python -m web.server then open http://localhost:8800
"""
from __future__ import annotations
import asyncio
import base64
import hashlib
import json
import os
import struct
from pathlib import Path
import asyncpg
HOST = os.environ.get("KANBOT_WEB_HOST", "127.0.0.1")
PORT = int(os.environ.get("KANBOT_WEB_PORT", "8800"))
DSN = os.environ.get("DATABASE_URL", "postgresql://kanbot:kanbot@localhost:5432/kanbot")
WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
INDEX = (Path(__file__).parent / "index.html").read_text()
_clients: set[asyncio.StreamWriter] = set()
# --------------------------------------------------------------------------- #
# board snapshot (fresh-load) — grouped by column
# --------------------------------------------------------------------------- #
async def board_snapshot(pool: asyncpg.Pool) -> dict:
async with pool.acquire() as c:
rows = await c.fetch(
'SELECT id,title,"column",agent_role,assigned_worker,attempt,'
"requires_human,tags FROM cards ORDER BY created_at")
cols: dict[str, list] = {k: [] for k in
("backlog", "ready", "in-flight", "review", "blocked", "done")}
for r in rows:
cols[r["column"]].append({
"id": r["id"], "title": r["title"], "role": r["agent_role"],
"worker": r["assigned_worker"], "attempt": r["attempt"],
"requires_human": r["requires_human"], "tags": json.loads(r["tags"]),
})
return cols
# --------------------------------------------------------------------------- #
# minimal WebSocket frame encoder (server->client text frames only)
# --------------------------------------------------------------------------- #
def _ws_frame(text: str) -> bytes:
data = text.encode()
n = len(data)
header = bytearray([0x81]) # FIN + text opcode
if n < 126:
header.append(n)
elif n < 65536:
header.append(126)
header += struct.pack(">H", n)
else:
header.append(127)
header += struct.pack(">Q", n)
return bytes(header) + data
async def _broadcast(text: str) -> None:
frame = _ws_frame(text)
dead = []
for w in list(_clients):
try:
w.write(frame)
await w.drain()
except Exception:
dead.append(w)
for w in dead:
_clients.discard(w)
# --------------------------------------------------------------------------- #
# single connection handler: HTTP for pages + snapshot, Upgrade for WS
# --------------------------------------------------------------------------- #
async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
pool: asyncpg.Pool) -> None:
try:
req = await reader.readuntil(b"\r\n\r\n")
except (asyncio.IncompleteReadError, ConnectionResetError):
writer.close(); return
head = req.decode(errors="ignore")
line = head.split("\r\n", 1)[0]
try:
_, path, _ = line.split(" ", 2)
except ValueError:
writer.close(); return
# WebSocket upgrade
if "upgrade: websocket" in head.lower():
key = ""
for h in head.split("\r\n"):
if h.lower().startswith("sec-websocket-key:"):
key = h.split(":", 1)[1].strip()
accept = base64.b64encode(
hashlib.sha1((key + WS_MAGIC).encode()).digest()).decode()
writer.write(
b"HTTP/1.1 101 Switching Protocols\r\n"
b"Upgrade: websocket\r\nConnection: Upgrade\r\n"
b"Sec-WebSocket-Accept: " + accept.encode() + b"\r\n\r\n")
await writer.drain()
_clients.add(writer)
# push initial snapshot so a fresh tab is instantly correct
writer.write(_ws_frame(json.dumps(
{"type": "snapshot", "board": await board_snapshot(pool)})))
await writer.drain()
try:
while not reader.at_eof():
await asyncio.sleep(30) # we only push; ignore client frames
finally:
_clients.discard(writer)
return
# plain HTTP
if path == "/snapshot":
body = json.dumps(await board_snapshot(pool)).encode()
ctype = b"application/json"
else:
body = INDEX.encode()
ctype = b"text/html; charset=utf-8"
writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: " + ctype +
b"\r\nContent-Length: " + str(len(body)).encode() +
b"\r\nConnection: close\r\n\r\n" + body)
await writer.drain()
writer.close()
async def main() -> None:
pool = await asyncpg.create_pool(DSN, min_size=1, max_size=4)
listen_conn = await asyncpg.connect(DSN)
async def on_notify(_c, _pid, _chan, payload):
# re-snapshot on any transition (board is small; correctness > cleverness)
await _broadcast(json.dumps(
{"type": "event", "event": json.loads(payload),
"board": await board_snapshot(pool)}))
await listen_conn.add_listener("kb_events", on_notify)
server = await asyncio.start_server(
lambda r, w: handle(r, w, pool), HOST, PORT)
print(f"[web] KanBot board live at http://{HOST}:{PORT}")
async with server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())
|