diff --git a/k8s/apps/agent-pod/agent-run-configmap.yaml b/k8s/apps/agent-pod/agent-run-configmap.yaml new file mode 100644 index 0000000..e1319c2 --- /dev/null +++ b/k8s/apps/agent-pod/agent-run-configmap.yaml @@ -0,0 +1,62 @@ +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: JSON.stringify(body), + }).catch((err) => console.error(`hub post ${path} failed:`, err.message)); + } + + 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", { id, agent }); + + const child = spawn("pi", ["-p", "--mode", "json", ...args], { + stdio: ["ignore", "pipe", "pipe"], + }); + + 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 + post("/agent/event", { id, event: line }); + } catch { + // non-JSON stdout noise, ignore + } + }); + + child.stderr.on("data", (chunk) => process.stderr.write(chunk)); + + child.on("close", async (code) => { + await post("/agent/end", { id, status: code === 0 ? "done" : "error" }); + process.exit(code ?? 1); + }); + } + + main(); diff --git a/k8s/apps/agent-pod/deployment.yaml b/k8s/apps/agent-pod/deployment.yaml index 9cb0f05..04ac38a 100644 --- a/k8s/apps/agent-pod/deployment.yaml +++ b/k8s/apps/agent-pod/deployment.yaml @@ -45,6 +45,34 @@ 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 volumes: - name: pi-config configMap: @@ -52,3 +80,9 @@ 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 new file mode 100644 index 0000000..b4c52f9 --- /dev/null +++ b/k8s/apps/agent-pod/hub-configmap.yaml @@ -0,0 +1,209 @@ +apiVersion: v1 +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/hub-service.yaml b/k8s/apps/agent-pod/hub-service.yaml new file mode 100644 index 0000000..c1527e3 --- /dev/null +++ b/k8s/apps/agent-pod/hub-service.yaml @@ -0,0 +1,11 @@ +apiVersion: v1 +kind: Service +metadata: + name: agent-hub + namespace: agent-pod +spec: + selector: + app: agent-pod + ports: + - port: 9090 + targetPort: 9090 diff --git a/k8s/apps/agent-pod/kustomization.yaml b/k8s/apps/agent-pod/kustomization.yaml index a24a038..c0dba55 100644 --- a/k8s/apps/agent-pod/kustomization.yaml +++ b/k8s/apps/agent-pod/kustomization.yaml @@ -4,3 +4,6 @@ namespace: agent-pod resources: - deployment.yaml - configmap.yaml + - hub-configmap.yaml + - agent-run-configmap.yaml + - hub-service.yaml