From 12b260f35a4cbcf0ae26c1c096ccd0085c2564fc Mon Sep 17 00:00:00 2001 From: Story Crater Bot <19826264+Riotpiaole@users.noreply.github.com> Date: Sun, 16 Aug 2026 10:24:03 -0700 Subject: [PATCH] refactor(agent-pod): merge hub into pi container, drop kubectl-exec spawn step agent-console complained about the two-step workflow: kubectl exec a script to spawn an agent, separate terminal to watch it. Root cause was hub being a separate Go sidecar with no access to the pi binary or its config, so it could only watch, never trigger. hub.js now runs inside the pi container itself (same filesystem, same PATH) and exposes POST /run, which spawns `pi -p --mode json` directly and streams it over the same /console WebSocket everyone else is already watching. agent-console gained a `run ` subcommand that POSTs then watches -- no separate exec step, no separate terminal. Drops the golang:1.25-alpine sidecar container and agent-run-configmap.yaml entirely (both superseded). ConfigMaps regenerated via local-harness/hub/sync-to-homelab.sh instead of hand-copied, so hub.js/config.json/settings.json each have exactly one source of truth. --- k8s/apps/agent-pod/agent-run-configmap.yaml | 74 ----- k8s/apps/agent-pod/configmap.yaml | 8 +- k8s/apps/agent-pod/deployment.yaml | 41 +-- k8s/apps/agent-pod/hub-configmap.yaml | 318 +++++++------------- k8s/apps/agent-pod/kustomization.yaml | 1 - 5 files changed, 128 insertions(+), 314 deletions(-) delete mode 100644 k8s/apps/agent-pod/agent-run-configmap.yaml diff --git a/k8s/apps/agent-pod/agent-run-configmap.yaml b/k8s/apps/agent-pod/agent-run-configmap.yaml deleted file mode 100644 index 83277b8..0000000 --- a/k8s/apps/agent-pod/agent-run-configmap.yaml +++ /dev/null @@ -1,74 +0,0 @@ -apiVersion: v1 -kind: ConfigMap -metadata: - name: agent-run-script - namespace: agent-pod -data: - agent-run.js: | - #!/usr/bin/env node - // Spawns `pi -p --mode json` and relays its session-protocol events to - // agent-hub (localhost:9090, same pod) line by line as they're emitted, so - // agent-console sees the turn live instead of after the process exits. - // - // Usage: node agent-run.js [-- ...] - const { spawn } = require("node:child_process"); - const readline = require("node:readline"); - const crypto = require("node:crypto"); - - const HUB = process.env.HUB_ADDR || "http://localhost:9090"; - - async function post(path, body) { - await fetch(`${HUB}${path}`, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body, - }).catch((err) => console.error(`hub post ${path} failed:`, err.message)); - } - - // `rawLine` is already a full JSON object (one per pi --mode json stdout - // line) -- embed it as-is rather than JSON.stringify-ing it into a nested - // string, or the hub ends up storing an escaped string instead of an event. - function postEvent(id, rawLine) { - return post("/agent/event", `{"id":${JSON.stringify(id)},"event":${rawLine}}`); - } - - async function main() { - const args = process.argv.slice(2); - const agent = args.shift(); - if (!agent || args.length === 0) { - console.error("usage: agent-run.js [pi args...] "); - process.exit(1); - } - - const id = crypto.randomUUID(); - await post("/agent/start", JSON.stringify({ id, agent })); - - const child = spawn("pi", ["-p", "--mode", "json", ...args], { - stdio: ["ignore", "pipe", "pipe"], - }); - - // Track in-flight event posts so /agent/end can't race ahead of them -- - // fetch() isn't awaited per-line (would serialize on network latency), but - // 'close' must wait for all of them before reporting done. - const pending = []; - const rl = readline.createInterface({ input: child.stdout }); - rl.on("line", (line) => { - if (!line.trim()) return; - try { - JSON.parse(line); // pi emits one JSON object per line; validate before relay - pending.push(postEvent(id, line)); - } catch { - // non-JSON stdout noise, ignore - } - }); - - child.stderr.on("data", (chunk) => process.stderr.write(chunk)); - - child.on("close", async (code) => { - await Promise.all(pending); - await post("/agent/end", JSON.stringify({ id, status: code === 0 ? "done" : "error" })); - process.exit(code ?? 1); - }); - } - - main(); diff --git a/k8s/apps/agent-pod/configmap.yaml b/k8s/apps/agent-pod/configmap.yaml index 4cb4fa8..97847ba 100644 --- a/k8s/apps/agent-pod/configmap.yaml +++ b/k8s/apps/agent-pod/configmap.yaml @@ -1,8 +1,4 @@ apiVersion: v1 -kind: ConfigMap -metadata: - name: pi-config - namespace: agent-pod data: config.json: | { @@ -39,3 +35,7 @@ data: "defaultThinkingLevel": "medium", "theme": "light" } +kind: ConfigMap +metadata: + name: pi-config + namespace: agent-pod diff --git a/k8s/apps/agent-pod/deployment.yaml b/k8s/apps/agent-pod/deployment.yaml index 04ac38a..969d240 100644 --- a/k8s/apps/agent-pod/deployment.yaml +++ b/k8s/apps/agent-pod/deployment.yaml @@ -22,12 +22,19 @@ spec: hostnames: - "api.riotpiao.com" containers: + # hub.js runs in the same container as pi (not a sidecar) so it can + # spawn `pi -p --mode json` directly via child_process -- a separate + # container can't exec into another container's filesystem/PATH. + # It IS the container's long-running process now; no more `sleep + # infinity` placeholder. - name: pi image: node:22-slim command: - sh - -c - - "npm install -g @earendil-works/pi-coding-agent@0.84.2 && sleep infinity" + - "npm install -g @earendil-works/pi-coding-agent@0.84.2 && npm install --prefix /root ws && node /root/hub.js" + ports: + - containerPort: 9090 resources: requests: cpu: "2" @@ -45,34 +52,9 @@ spec: - name: pi-models mountPath: /root/.pi/agent/models.json subPath: models.json - - name: agent-run-script - mountPath: /root/agent-run.js - subPath: agent-run.js - # Sidecar: relays pi's --mode json session events to agent-console. - # Runs via `go run` on mounted source rather than a prebuilt image -- - # same "no registry yet" tradeoff as the pi container's npm install. - - name: hub - image: golang:1.25-alpine - workingDir: /hub - command: - - sh - - -c - - "go run ." - env: - - name: HUB_ADDR - value: ":9090" - ports: - - containerPort: 9090 - resources: - requests: - cpu: 100m - memory: 128Mi - limits: - cpu: 500m - memory: 256Mi - volumeMounts: - name: hub-src - mountPath: /hub + mountPath: /root/hub.js + subPath: hub.js volumes: - name: pi-config configMap: @@ -80,9 +62,6 @@ spec: - name: pi-models secret: secretName: pi-models - - name: agent-run-script - configMap: - name: agent-run-script - name: hub-src configMap: name: hub-src diff --git a/k8s/apps/agent-pod/hub-configmap.yaml b/k8s/apps/agent-pod/hub-configmap.yaml index b4c52f9..9943e4e 100644 --- a/k8s/apps/agent-pod/hub-configmap.yaml +++ b/k8s/apps/agent-pod/hub-configmap.yaml @@ -1,209 +1,119 @@ apiVersion: v1 +data: + hub.js: | + #!/usr/bin/env node + // agent-hub: lives inside the pi container (not a sidecar) so it can spawn + // `pi -p --mode json` directly. One persistent in-cluster service -- + // POST /run to trigger an agent, GET /console (WebSocket) to watch every + // concurrent run live as it happens, relaying pi's own session protocol + // verbatim (same event shape Claude Code sessions use). + const http = require("node:http"); + const crypto = require("node:crypto"); + const { spawn } = require("node:child_process"); + const readline = require("node:readline"); + const { WebSocketServer } = require("ws"); + + const PORT = process.env.HUB_PORT || 9090; + + const sessions = new Map(); // id -> {id, agent, status, events, startedAt, endedAt} + const viewers = new Set(); // WebSocket connections watching /console + + function broadcast(type, session) { + const msg = JSON.stringify({ type, session }); + for (const ws of viewers) { + if (ws.readyState === ws.OPEN) ws.send(msg); + } + } + + function startSession(agent) { + const id = crypto.randomUUID(); + const session = { id, agent, status: "running", events: [], startedAt: new Date().toISOString() }; + sessions.set(id, session); + broadcast("start", session); + return session; + } + + function addEvent(session, rawLine) { + session.events.push(JSON.parse(rawLine)); + broadcast("event", session); + } + + function endSession(session, status) { + session.status = status; + session.endedAt = new Date().toISOString(); + broadcast("end", session); + } + + // Spawns `pi -p --mode json ` and relays it live. Same shape as the + // old agent-run.js wrapper, just in-process instead of a separate exec. + function runAgent(agent, prompt, extraArgs = []) { + const session = startSession(agent); + const child = spawn("pi", ["-p", "--mode", "json", ...extraArgs, prompt], { + stdio: ["ignore", "pipe", "pipe"], + }); + const rl = readline.createInterface({ input: child.stdout }); + rl.on("line", (line) => { + if (!line.trim()) return; + try { + addEvent(session, line); + } catch { + // non-JSON stdout noise, ignore + } + }); + child.stderr.on("data", (chunk) => process.stderr.write(chunk)); + child.on("close", (code) => endSession(session, code === 0 ? "done" : "error")); + return session; + } + + const server = http.createServer((req, res) => { + const url = new URL(req.url, "http://localhost"); + + if (url.pathname === "/healthz") { + res.writeHead(200).end(); + return; + } + + if (url.pathname === "/sessions" && req.method === "GET") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify([...sessions.values()])); + return; + } + + if (url.pathname === "/run" && req.method === "POST") { + let body = ""; + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => { + try { + const { agent, prompt, provider, model } = JSON.parse(body); + if (!agent || !prompt) throw new Error("agent and prompt are required"); + const extraArgs = []; + if (provider) extraArgs.push("--provider", provider); + if (model) extraArgs.push("--model", model); + const session = runAgent(agent, prompt, extraArgs); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ id: session.id })); + } catch (err) { + res.writeHead(400, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: err.message })); + } + }); + return; + } + + res.writeHead(404).end(); + }); + + const wss = new WebSocketServer({ server, path: "/console" }); + wss.on("connection", (ws) => { + for (const session of sessions.values()) { + ws.send(JSON.stringify({ type: "snapshot", session })); + } + viewers.add(ws); + ws.on("close", () => viewers.delete(ws)); + }); + + server.listen(PORT, () => console.log(`agent-hub listening on :${PORT}`)); kind: ConfigMap metadata: name: hub-src namespace: agent-pod -data: - go.mod: | - module agent-hub - - go 1.25.0 - - require github.com/gorilla/websocket v1.5.3 - go.sum: | - github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= - github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= - main.go: | - // agent-hub sits inside agent-pod alongside pi. Spawned agent runs report - // their lifecycle over localhost HTTP, relaying pi's own `--mode json` - // session protocol (session/agent_start/turn_start/message_start/ - // message_update/message_end/turn_end/agent_end -- the same shape Claude - // Code sessions use) verbatim. agent-console (outside the pod, via kubectl - // port-forward) watches all concurrent agents live over one WebSocket and - // can render each session as a real transcript, not a log tail. - package main - - import ( - "encoding/json" - "log" - "net/http" - "os" - "sync" - "time" - - "github.com/gorilla/websocket" - ) - - type Session struct { - ID string `json:"id"` - Agent string `json:"agent"` - Status string `json:"status"` // running | done | error - Events []json.RawMessage `json:"events"` // raw pi --mode json events, in order - StartedAt time.Time `json:"startedAt"` - EndedAt time.Time `json:"endedAt,omitempty"` - } - - type Notification struct { - Type string `json:"type"` // snapshot | start | event | end - Session Session `json:"session"` - } - - type Hub struct { - mu sync.Mutex - sessions map[string]*Session - viewers map[*websocket.Conn]struct{} - } - - func newHub() *Hub { - return &Hub{ - sessions: make(map[string]*Session), - viewers: make(map[*websocket.Conn]struct{}), - } - } - - func (h *Hub) broadcast(n Notification) { - h.mu.Lock() - defer h.mu.Unlock() - for conn := range h.viewers { - if err := conn.WriteJSON(n); err != nil { - conn.Close() - delete(h.viewers, conn) - } - } - } - - func (h *Hub) start(id, agent string) { - h.mu.Lock() - s := &Session{ID: id, Agent: agent, Status: "running", StartedAt: time.Now()} - h.sessions[id] = s - h.mu.Unlock() - h.broadcast(Notification{Type: "start", Session: *s}) - } - - func (h *Hub) event(id string, raw json.RawMessage) { - h.mu.Lock() - s, ok := h.sessions[id] - if !ok { - h.mu.Unlock() - return - } - s.Events = append(s.Events, raw) - snapshot := *s - h.mu.Unlock() - h.broadcast(Notification{Type: "event", Session: snapshot}) - } - - func (h *Hub) end(id, status string) { - h.mu.Lock() - s, ok := h.sessions[id] - if !ok { - h.mu.Unlock() - return - } - s.Status = status - s.EndedAt = time.Now() - snapshot := *s - h.mu.Unlock() - h.broadcast(Notification{Type: "end", Session: snapshot}) - } - - func (h *Hub) snapshot() []Session { - h.mu.Lock() - defer h.mu.Unlock() - out := make([]Session, 0, len(h.sessions)) - for _, s := range h.sessions { - out = append(out, *s) - } - return out - } - - var upgrader = websocket.Upgrader{ - CheckOrigin: func(r *http.Request) bool { return true }, - } - - func main() { - addr := os.Getenv("HUB_ADDR") - if addr == "" { - addr = ":9090" - } - h := newHub() - - http.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) - - // Plain JSON snapshot for curl/debugging -- the WebSocket below is the - // live path agent-console actually uses. - http.HandleFunc("/sessions", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(h.snapshot()) - }) - - // Agent side: fire-and-forget POSTs, no persistent connection needed - // since each pi run is short-lived. - http.HandleFunc("/agent/start", func(w http.ResponseWriter, r *http.Request) { - var body struct{ ID, Agent string } - if err := json.NewDecoder(r.Body).Decode(&body); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - h.start(body.ID, body.Agent) - w.WriteHeader(http.StatusOK) - }) - - http.HandleFunc("/agent/event", func(w http.ResponseWriter, r *http.Request) { - var body struct { - ID string - Event json.RawMessage - } - if err := json.NewDecoder(r.Body).Decode(&body); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - h.event(body.ID, body.Event) - w.WriteHeader(http.StatusOK) - }) - - http.HandleFunc("/agent/end", func(w http.ResponseWriter, r *http.Request) { - var body struct{ ID, Status string } - if err := json.NewDecoder(r.Body).Decode(&body); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - h.end(body.ID, body.Status) - w.WriteHeader(http.StatusOK) - }) - - // Console side: connect, get a snapshot of everything running, then - // stream live start/event/end notifications as they happen. - http.HandleFunc("/console", func(w http.ResponseWriter, r *http.Request) { - conn, err := upgrader.Upgrade(w, r, nil) - if err != nil { - return - } - h.mu.Lock() - for _, s := range h.sessions { - conn.WriteJSON(Notification{Type: "snapshot", Session: *s}) - } - h.viewers[conn] = struct{}{} - h.mu.Unlock() - - // Drain reads to detect disconnect; console never sends anything. - go func() { - defer func() { - h.mu.Lock() - delete(h.viewers, conn) - h.mu.Unlock() - conn.Close() - }() - for { - if _, _, err := conn.ReadMessage(); err != nil { - return - } - } - }() - }) - - log.Printf("agent-hub listening on %s", addr) - log.Fatal(http.ListenAndServe(addr, nil)) - } diff --git a/k8s/apps/agent-pod/kustomization.yaml b/k8s/apps/agent-pod/kustomization.yaml index eb77cf7..564f982 100644 --- a/k8s/apps/agent-pod/kustomization.yaml +++ b/k8s/apps/agent-pod/kustomization.yaml @@ -5,6 +5,5 @@ resources: - deployment.yaml - configmap.yaml - hub-configmap.yaml - - agent-run-configmap.yaml - hub-service.yaml - console-ingress.yaml