fleet-manager: fix CA install, add register-host, surface agent status

Four fixes from the on-host agent bring-up:

1. Bake the real fleet CA into the image. New `fleet_ca` build var (default =
   the committed placeholder) file-provisions the trust anchor to
   /etc/dot1x/radius-ca.pem — used by wpa_supplicant (RADIUS) AND the fleet-agent
   (verify the manager). CI materializes a FLEET_CA secret to a runner-temp file
   and passes -var fleet_ca; unset => placeholder (lab). No CA material committed.

2. server-pki.sh installs the agent CLIENT CA as root+intermediate (not root
   only). Host TPM certs are leaf-only in NV and signed by the step-ca
   intermediate, so a root-only client CA makes the manager advertise only the
   root in its CertificateRequest -> Go clients send no cert ("tls: certificate
   required") and it couldn't verify a leaf-only cert anyway. bootstrap.sh now
   exports intermediate_ca.crt so FLEET_CA_INTERMEDIATE resolves.

3. New `fleet-manager register-host <serial> [-name] [-hostnqn]` subcommand:
   seeds/updates an inventory record for a host enrolled out of band (bench
   broker, migrated fleet) through the real store, so its agent stream is
   accepted. The manual counterpart to the authorise-approve flow.

4. The agent surfaces the authoritative recv-side gRPC status instead of the
   bare io.EOF gRPC returns on the send side when the server ends the stream —
   so "host is not enrolled" (PermissionDenied) shows up instead of "EOF".

files/agent/main.go

@@ -15,6 +15,7 @@ package main
 
 import (
 	"context"
+	"io"
 	"log"
 	"math/rand"
 	"os"
@@ -117,18 +118,16 @@ func runSession(ctx context.Context, client fleetpb.FleetAgentClient, cfg config
 		return err
 	}
 
-	// Hello first — what we're running, so the manager reconciles version drift.
-	facts := fetchFacts(sctx, cfg.hostapiURL)
-	if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Hello{Hello: buildHello(facts, readKioskURL(cfg.kioskEnv))}}); err != nil {
-		return err
-	}
-	log.Printf("connected to %s — streaming health every %s", cfg.manager, cfg.interval)
-
-	// Receive loop: v1 LOGS desired-state/commands, does not act on them.
+	// Receive loop. Its error is the AUTHORITATIVE session status — when the manager
+	// ends the stream (e.g. PermissionDenied "host is not enrolled"), gRPC only
+	// returns a bare io.EOF on the SEND side, so a raw Send error is uninformative.
+	// We surface the recv-side status instead. v1 LOGS desired-state/commands.
+	recvErr := make(chan error, 1)
 	go func() {
 		for {
 			down, err := stream.Recv()
 			if err != nil {
+				recvErr <- err
 				cancel()
 				return
 			}
@@ -136,24 +135,53 @@ func runSession(ctx context.Context, client fleetpb.FleetAgentClient, cfg config
 		}
 	}()
 
+	// Hello first — what we're running, so the manager reconciles version drift.
+	facts := fetchFacts(sctx, cfg.hostapiURL)
+	if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Hello{Hello: buildHello(facts, readKioskURL(cfg.kioskEnv))}}); err != nil {
+		return statusFrom(err, recvErr)
+	}
+	log.Printf("connected to %s — streaming health every %s", cfg.manager, cfg.interval)
+
 	// Send one snapshot immediately, then on the interval.
 	t := time.NewTicker(cfg.interval)
 	defer t.Stop()
 	if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Snapshot{Snapshot: p.snapshot(sctx)}}); err != nil {
-		return err
+		return statusFrom(err, recvErr)
 	}
 	for {
 		select {
+		case err := <-recvErr:
+			return err
 		case <-sctx.Done():
-			return sctx.Err()
+			// Prefer the recv-side status if that's what cancelled us.
+			select {
+			case err := <-recvErr:
+				return err
+			default:
+				return sctx.Err()
+			}
 		case <-t.C:
 			if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Snapshot{Snapshot: p.snapshot(sctx)}}); err != nil {
-				return err
+				return statusFrom(err, recvErr)
 			}
 		}
 	}
 }
 
+// statusFrom turns a Send-side io.EOF (which gRPC returns when the server has
+// closed the stream) into the authoritative recv-side status — the real reason
+// the manager ended the session — falling back to the original error.
+func statusFrom(err error, recvErr chan error) error {
+	if err == io.EOF {
+		select {
+		case re := <-recvErr:
+			return re
+		case <-time.After(2 * time.Second):
+		}
+	}
+	return err
+}
+
 // logDown records a manager->host message. TELEMETRY-ONLY v1: received, logged,
 // NOT applied. The reconcile actuators land in a follow-up (see docs/TODO.md #14).
 func logDown(down *fleetpb.ManagerDown) {