Move admin HTTP routes to /v1/stream/ prefix, add /healthz

Admin endpoints now served under /v1/stream/{status,topics,groups}
following the /<version>/<service>/<path> convention. Root /
redirects to /v1/stream/. Added /healthz for K8s HTTP probes.

LLM.md

@@ -83,11 +83,24 @@ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
 ```
 
+## Admin HTTP API (port 9093)
+
+All management endpoints under `/v1/stream/`:
+```
+GET /v1/stream/status  — service status (JSON)
+GET /v1/stream/topics  — list topics with partition details
+GET /v1/stream/groups  — list consumer group offsets
+GET /healthz           — health check (K8s probes)
+```
+
+Root `/` redirects to `/v1/stream/`.
+
 ## Deployment (hanzo-k8s, do-sfo3)
 ```
 Namespace: hanzo
 PubSub:    pubsub.hanzo.svc:4222   (nats:2.10-alpine, 1 replica, 20Gi PVC)
 Stream:    stream.hanzo.svc:9092   (ghcr.io/hanzoai/stream:latest, 2 replicas)
+Admin:     :9093 (HTTP, /healthz for K8s probes)
 ```
 Dockerfile builds linux/amd64 via `GOARCH=amd64`. CI pushes to GHCR on every main push.
 

protocol/admin.go

@@ -46,17 +46,23 @@ type GroupOffset struct {
 	Offset    int64  `json:"offset"`
 }
 
-// StartAdmin starts the admin HTTP server on the configured port
+// StartAdmin starts the admin HTTP server on the configured port.
+// All management endpoints are served under /v1/stream/.
+// /healthz is served at root for K8s probes.
 func (b *Broker) StartAdmin() {
 	if b.Config.AdminPort == 0 {
 		return
 	}
 
+	const prefix = "/v1/stream"
+
 	mux := http.NewServeMux()
-	mux.HandleFunc("/status", b.handleStatus)
-	mux.HandleFunc("/topics", b.handleTopics)
-	mux.HandleFunc("/groups", b.handleGroups)
-	mux.HandleFunc("/", b.handleIndex)
+	mux.HandleFunc(prefix+"/status", b.handleStatus)
+	mux.HandleFunc(prefix+"/topics", b.handleTopics)
+	mux.HandleFunc(prefix+"/groups", b.handleGroups)
+	mux.HandleFunc("/healthz", b.handleHealthz)
+	mux.HandleFunc(prefix+"/", b.handleIndex)
+	mux.HandleFunc("/", b.handleRoot)
 
 	addr := fmt.Sprintf(":%d", b.Config.AdminPort)
 	log.Info("Admin HTTP server listening on %s", addr)
@@ -67,16 +73,31 @@ func (b *Broker) StartAdmin() {
 	}()
 }
 
-func (b *Broker) handleIndex(w http.ResponseWriter, r *http.Request) {
+func (b *Broker) handleRoot(w http.ResponseWriter, r *http.Request) {
 	if r.URL.Path != "/" {
 		http.NotFound(w, r)
 		return
 	}
+	http.Redirect(w, r, "/v1/stream/", http.StatusMovedPermanently)
+}
+
+func (b *Broker) handleIndex(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "text/plain")
 	fmt.Fprintf(w, "Hanzo Stream Admin\n\n")
-	fmt.Fprintf(w, "GET /status  — service health\n")
-	fmt.Fprintf(w, "GET /topics  — list topics with partition details\n")
-	fmt.Fprintf(w, "GET /groups  — list consumer group offsets\n")
+	fmt.Fprintf(w, "GET /v1/stream/status  — service status\n")
+	fmt.Fprintf(w, "GET /v1/stream/topics  — list topics with partition details\n")
+	fmt.Fprintf(w, "GET /v1/stream/groups  — list consumer group offsets\n")
+	fmt.Fprintf(w, "GET /healthz           — health check\n")
+}
+
+func (b *Broker) handleHealthz(w http.ResponseWriter, r *http.Request) {
+	if b.PubSub != nil && b.PubSub.NC != nil && b.PubSub.NC.IsConnected() {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, "ok\n")
+		return
+	}
+	w.WriteHeader(http.StatusServiceUnavailable)
+	fmt.Fprintf(w, "pubsub disconnected\n")
 }
 
 func (b *Broker) handleStatus(w http.ResponseWriter, r *http.Request) {