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