slice 10A: hub desired-state serving + signed-jobs queue (Down channel) (hub v0.9.0)

Serve operator intent to authenticated hosts: PUT /admin/hosts/{id}/desired-state (global key) bumps desired_generation; GET /hosts/{id}/desired-state + /jobs are per-host self-scoped; the host-report envelope now carries the real generation + has_signed_ops. New signed_jobs table + store methods. Desired-state stored/served opaquely (agent owns the schema). Cross-repo golden (envelope + desired-state) byte-identical with felhom-agent; doc 03 §4/§9 updated. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-10 19:03:14 +02:00
parent f9af3243b9
commit e54f882e70
8 changed files with 669 additions and 30 deletions
@@ -3,6 +3,7 @@ package api
 import (
 	"bytes"
 	"crypto/subtle"
+	"database/sql"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -128,6 +129,20 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	case r.Method == http.MethodPut && strings.HasPrefix(path, "/hosts/") && strings.HasSuffix(path, "/escrow"):
 		hostID := strings.TrimSuffix(strings.TrimPrefix(path, "/hosts/"), "/escrow")
 		h.handleHostEscrowPut(w, r, hostID)
+	// Desired-state serving (slice 10A) — per-host-key, self-scoped (a host reads only its own).
+	case r.Method == http.MethodGet && strings.HasPrefix(path, "/hosts/") && strings.HasSuffix(path, "/desired-state"):
+		hostID := strings.TrimSuffix(strings.TrimPrefix(path, "/hosts/"), "/desired-state")
+		h.handleGetDesiredState(w, r, hostID)
+	case r.Method == http.MethodGet && strings.HasPrefix(path, "/hosts/") && strings.HasSuffix(path, "/jobs"):
+		hostID := strings.TrimSuffix(strings.TrimPrefix(path, "/hosts/"), "/jobs")
+		h.handleGetJobs(w, r, hostID)
+	// Admin-set (slice 10A) — global/operator key only; bumps the generation.
+	case r.Method == http.MethodPut && strings.HasPrefix(path, "/admin/hosts/") && strings.HasSuffix(path, "/desired-state"):
+		hostID := strings.TrimSuffix(strings.TrimPrefix(path, "/admin/hosts/"), "/desired-state")
+		h.handleAdminSetDesiredState(w, r, hostID)
+	case r.Method == http.MethodPost && strings.HasPrefix(path, "/admin/hosts/") && strings.HasSuffix(path, "/jobs"):
+		hostID := strings.TrimSuffix(strings.TrimPrefix(path, "/admin/hosts/"), "/jobs")
+		h.handleAdminEnqueueJob(w, r, hostID)
 	case r.Method == http.MethodPost && path == "/event":
 		h.handleEvent(w, r)
 	case r.Method == http.MethodPost && path == "/notify":
@@ -500,12 +515,26 @@ func (h *Handler) handleHostReport(w http.ResponseWriter, r *http.Request) {
 	if cc, err := h.store.GetCustomerConfig(custID); err == nil && cc != nil && cc.Status == "blocked" {
 		blocked = true
 	}
+
+	// Control envelope (slice 10A): the cheap change-notification. desired_generation is the
+	// host's current generation (the agent re-fetches the full desired-state only when it
+	// advances past its cached one); has_signed_ops flags a non-empty signed-jobs queue (the
+	// agent fetches/executes them in 10B). Both degrade safely to their slice-4 defaults on a
+	// store error — a heartbeat must never fail on the control channel.
+	var desiredGen int64
+	if host, err := h.store.GetHost(hostID); err == nil && host != nil {
+		desiredGen = host.DesiredGeneration
+	}
+	hasSignedOps := false
+	if n, err := h.store.CountSignedJobs(hostID); err == nil && n > 0 {
+		hasSignedOps = true
+	}
 	resp := map[string]interface{}{
 		"status":                "ok",
 		"poll_interval_seconds": defaultHostPollSeconds,
 		"blocked":               blocked,
-		"desired_generation":    0,     // reserved (slice 4)
-		"has_signed_ops":        false, // reserved (slice 4)
+		"desired_generation":    desiredGen,
+		"has_signed_ops":        hasSignedOps,
 	}
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
@@ -633,6 +662,175 @@ func (h *Handler) handleHostEscrowPut(w http.ResponseWriter, r *http.Request, pa
 	w.Write([]byte(`{"status":"ok"}`))
 }

+// handleGetDesiredState serves a host its authoritative desired-state (slice 10A). Per-host key,
+// SELF-SCOPED: a host reads ONLY its own (the global operator key may read any). The agent fetches
+// this when the heartbeat envelope's desired_generation has advanced past its cached one. The
+// response carries the generation the state corresponds to, so the agent caches it atomically.
+func (h *Handler) handleGetDesiredState(w http.ResponseWriter, r *http.Request, pathHostID string) {
+	authHostID, _, isGlobal, ok := h.checkAuthHost(r)
+	if !ok {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return
+	}
+	if pathHostID == "" {
+		http.Error(w, "Missing host_id", http.StatusBadRequest)
+		return
+	}
+	if !isGlobal && authHostID != pathHostID {
+		http.Error(w, "Forbidden: host_id mismatch", http.StatusForbidden)
+		return
+	}
+	host, err := h.store.GetHost(pathHostID)
+	if err != nil {
+		h.logger.Printf("[ERROR] desired-state lookup for %s: %v", pathHostID, err)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	if host == nil {
+		http.Error(w, "Unknown host_id", http.StatusNotFound)
+		return
+	}
+	desired := host.DesiredJSON
+	if strings.TrimSpace(desired) == "" {
+		desired = "{}"
+	}
+	resp := map[string]interface{}{
+		"generation":    host.DesiredGeneration,
+		"desired_state": json.RawMessage(desired), // opaque to the hub — agent owns the schema
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(resp)
+}
+
+// handleGetJobs serves a host its pending signed-op blobs (slice 10A). Per-host key, SELF-SCOPED.
+// The blobs are OPAQUE (the hub never forged or opened them); the agent verifies + executes them
+// in 10B. 10A only serves the queue.
+func (h *Handler) handleGetJobs(w http.ResponseWriter, r *http.Request, pathHostID string) {
+	authHostID, _, isGlobal, ok := h.checkAuthHost(r)
+	if !ok {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return
+	}
+	if pathHostID == "" {
+		http.Error(w, "Missing host_id", http.StatusBadRequest)
+		return
+	}
+	if !isGlobal && authHostID != pathHostID {
+		http.Error(w, "Forbidden: host_id mismatch", http.StatusForbidden)
+		return
+	}
+	jobs, err := h.store.GetSignedJobs(pathHostID)
+	if err != nil {
+		h.logger.Printf("[ERROR] jobs lookup for %s: %v", pathHostID, err)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	out := make([]map[string]string, 0, len(jobs))
+	for _, j := range jobs {
+		out = append(out, map[string]string{
+			"job_id":     j.JobID,
+			"blob_b64":   base64.StdEncoding.EncodeToString(j.Blob),
+			"created_at": j.CreatedAt,
+		})
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(map[string]interface{}{"jobs": out})
+}
+
+// handleAdminSetDesiredState sets a host's desired-state (slice 10A). GLOBAL/operator key ONLY —
+// a per-host key cannot author its own intent. The body is the desired-state JSON (opaque to the
+// hub: it stores + serves bytes, never validates/interprets the schema — the agent/CLI owns it).
+// Writing BUMPS desired_generation so the next heartbeat signals the agent to re-fetch.
+func (h *Handler) handleAdminSetDesiredState(w http.ResponseWriter, r *http.Request, pathHostID string) {
+	_, _, isGlobal, ok := h.checkAuthHost(r)
+	if !ok || !isGlobal {
+		http.Error(w, "Forbidden: global key required", http.StatusForbidden)
+		return
+	}
+	if pathHostID == "" {
+		http.Error(w, "Missing host_id", http.StatusBadRequest)
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
+	if err != nil {
+		http.Error(w, "Bad request", http.StatusBadRequest)
+		return
+	}
+	// Validate it is well-formed JSON (the hub does not interpret the schema, but a malformed
+	// blob would break the agent's parse — reject it at the door).
+	if !json.Valid(body) {
+		http.Error(w, "Invalid payload: body must be JSON", http.StatusBadRequest)
+		return
+	}
+	gen, err := h.store.SetHostDesired(pathHostID, body)
+	if err == sql.ErrNoRows {
+		http.Error(w, "Unknown host_id", http.StatusNotFound)
+		return
+	}
+	if err != nil {
+		h.logger.Printf("[ERROR] set desired-state for %s: %v", pathHostID, err)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	h.logger.Printf("[INFO] admin-set desired-state for host %s (generation now %d, %d bytes)", pathHostID, gen, len(body))
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "generation": gen})
+}
+
+// handleAdminEnqueueJob appends an opaque signed-op blob to a host's queue (slice 10A). GLOBAL key
+// ONLY. The blob is pre-signed off-hub (the hub holds no signing key); the hub stores it verbatim.
+// This is the minimal operator path to seed the queue so HasSignedOps/serving are exercisable; the
+// rich operator/signing UX is later. Execution is 10B.
+func (h *Handler) handleAdminEnqueueJob(w http.ResponseWriter, r *http.Request, pathHostID string) {
+	_, _, isGlobal, ok := h.checkAuthHost(r)
+	if !ok || !isGlobal {
+		http.Error(w, "Forbidden: global key required", http.StatusForbidden)
+		return
+	}
+	if pathHostID == "" {
+		http.Error(w, "Missing host_id", http.StatusBadRequest)
+		return
+	}
+	host, err := h.store.GetHost(pathHostID)
+	if err != nil || host == nil {
+		http.Error(w, "Unknown host_id", http.StatusNotFound)
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
+	if err != nil {
+		http.Error(w, "Bad request", http.StatusBadRequest)
+		return
+	}
+	var req struct {
+		JobID   string `json:"job_id"`
+		BlobB64 string `json:"blob_b64"`
+	}
+	if err := json.Unmarshal(body, &req); err != nil || req.BlobB64 == "" {
+		http.Error(w, "Invalid payload: blob_b64 required", http.StatusBadRequest)
+		return
+	}
+	blob, err := base64.StdEncoding.DecodeString(req.BlobB64)
+	if err != nil || len(blob) == 0 {
+		http.Error(w, "Invalid payload: blob_b64 not valid base64", http.StatusBadRequest)
+		return
+	}
+	if req.JobID == "" {
+		req.JobID, _ = configgen.RandomHex(8)
+	}
+	if err := h.store.EnqueueSignedJob(pathHostID, req.JobID, blob); err != nil {
+		h.logger.Printf("[ERROR] enqueue job for %s: %v", pathHostID, err)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	h.logger.Printf("[INFO] enqueued signed-op job %s for host %s (%d bytes)", req.JobID, pathHostID, len(blob))
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "job_id": req.JobID})
+}
+
 // allowedEventTypes lists all valid event_type values the Hub accepts.
 var allowedEventTypes = map[string]bool{
 	// Controller-pushed events