feat: Reader.Reboot and Reader.Shutdown in mqtt-rpcd (TRA-1316)

backend/internal/readerrpc/config.go

@@ -106,7 +106,7 @@ type Capabilities struct {
 	Daemon *DaemonInfo `json:"daemon,omitempty"`
 }
 
-// Status is the response to Reader.GetStatus and the status-topic payload.
+// Status is the response to Reader.GetStatus. The status topic carries StatusDoc.
 type Status struct {
 	Online        bool   `json:"online"`
 	Reading       bool   `json:"reading"`

backend/internal/readerrpc/rpc.go

@@ -22,14 +22,34 @@ const (
 	MethodGetOperProfile  = "Reader.GetOperProfile"
 	MethodSetOperProfile  = "Reader.SetOperProfile"
 	MethodGetStatus       = "Reader.GetStatus"
+	MethodGpoSet          = "Gpo.Set"
+
+	// MethodReboot and MethodShutdown cleanly reboot or power off the reader
+	// host (sync, then systemd). The daemon replies first and acts after a short
+	// delay, announcing the planned outage on the retained status first.
+	// Neither takes required params; PowerParams is optional.
+	MethodReboot   = "Reader.Reboot"
+	MethodShutdown = "Reader.Shutdown"
 
 	// Reserved for future use.
 	MethodScanStart = "Scan.Start"
 	MethodScanStop  = "Scan.Stop"
-	MethodGpoSet    = "Gpo.Set"
-	MethodReboot    = "Reader.Reboot"
 )
 
+// PowerParams are the optional params of Reader.Reboot and Reader.Shutdown.
+type PowerParams struct {
+	// RequestedBy names the person or system asking, for the reader's log. The
+	// request's Src is logged as well.
+	RequestedBy string `json:"requested_by,omitempty"`
+}
+
+// PowerResult acknowledges Reader.Reboot or Reader.Shutdown. The action has not
+// happened yet: it starts InMs after the reply.
+type PowerResult struct {
+	Action string `json:"action"` // StatusReasonReboot or StatusReasonShutdown
+	InMs   int    `json:"in_ms"`
+}
+
 // Standard JSON-RPC 2.0 error codes.
 const (
 	CodeParseError     = -32700

mqtt-rpc/README.md

@@ -55,10 +55,11 @@ never evicts the reader's own reads connection.
 
 **Capability-flagged management** (declared per adapter in `supports[]` / `unsupported[]`):
 - `Gpo.Set(params={port, on, pulse_ms?})` → `{port, on, pulsed}` — supported by the CS463 adapter. See "Alarm outputs".
-- `Scan.Start`, `Scan.Stop`, `Reader.Reboot` — reserved; `unsupported` on the CS463.
+- `Scan.Start`, `Scan.Stop` — reserved; `unsupported` on the CS463.
 
 **Daemon** (acts on mqtt-rpcd itself; never touches the vendor `/API`):
 - `Daemon.Update(params={version, url, sha256, signature})` → `{accepted, from, to}` once verified; the install follows the reply. Refusals are error `-33002`. See "Remote update".
+- `Reader.Reboot`, `Reader.Shutdown(params={requested_by?})` → `{action: "reboot"|"shutdown", in_ms}`; the clean reboot or power-off follows the reply. Listed in `supports[]` when the daemon runs on a host it can reboot. See "Reboot and shutdown".
 
 Rules: neutral config (no vendor terms like "operation profile" leak into the
 wire); a new tunable is a field, not a method; `applied` is explicit; the contract
@@ -262,6 +263,32 @@ reconcile re-arms it.
 retry, but takes no recovery action. Use it on a bench that has to hold a wedge
 for diagnosis.
 
+If systemd refuses the reboot, the reader is still up: the daemon republishes
+the status online and tries again after the next window.
+
+### Reboot and shutdown (`Reader.Reboot`, `Reader.Shutdown`)
+A clean reboot or power-off an operator asks for over the broker, for a reader
+with no SSH. A reboot is the known fix for the vendor-app wedge; a shutdown
+before planned power work avoids the unclean power cut that causes it.
+
+```sh
+MQTT_URL=… mqtt-rpcctl reboot   -base trakrf.id/cs463-212 -yes
+MQTT_URL=… mqtt-rpcctl shutdown -base trakrf.id/cs463-212 -yes   # back only after an on-site power-cycle
+```
+
+1. The daemon logs the request (`requested_by`, the reply topic, `asked_at`) and
+   replies `{action, in_ms: 2000}`.
+2. 2 s later it publishes the retained status offline with `reason: "reboot"`
+   or `"shutdown"`, then runs `sync` and `systemctl reboot` or
+   `systemctl poweroff`. Never a power reset. systemd then stops the daemon,
+   whose stop drives every output off and keeps the reason on the status.
+3. A second request while one is under way, or while an automatic recovery
+   reboot is announced, is refused (`-32603`, "already going down").
+4. If systemd refuses, the status goes back online and a retry is accepted.
+
+After a reboot the daemon takes 2–7 min to reach the broker again (boot-time
+entropy for TLS).
+
 ### Status document (retained `<base>/status`)
 ```jsonc
 {"online": true, "at": "2026-11-08T17:40:00Z", "version": "v1.5.0-330-g…",
@@ -271,7 +298,8 @@ for diagnosis.
             "auto_reboots": 1, "auto_reboots_exhausted": false,
             "golden_config": "pending",           // "applied" | "pending" | "off"
             "cloudserver_fault": "cleanSession is false; golden is true"}}  // absent when clean
-// before an automatic reboot, and after it via Stop:
+// before a planned outage, and after it via Stop. reason is "auto-recovery"
+// (the daemon's own), or "reboot" / "shutdown" (Reader.Reboot / Reader.Shutdown):
 {"online": false, "reason": "auto-recovery", "version": "…", "health": {…}}
 // MQTT Last Will (the daemon died without Stop):
 {"online": false}

mqtt-rpc/cmd/mqtt-rpcctl/main.go

@@ -7,9 +7,15 @@
 //	mqtt-rpcctl call   -base TOPIC -method M [-params JSON] [-ca FILE]
 //	mqtt-rpcctl update -base TOPIC -manifest FILE|URL [-ca FILE]
 //	mqtt-rpcctl status -base TOPIC [-watch] [-ca FILE]
+//	mqtt-rpcctl reboot   -base TOPIC -yes [-by WHO] [-ca FILE]
+//	mqtt-rpcctl shutdown -base TOPIC -yes [-by WHO] [-ca FILE]
+//
+// reboot and shutdown are clean (sync, then systemd), and both take the reader
+// offline: a shutdown needs someone on site to power-cycle it back. -by goes to
+// the reader's log and defaults to $USER@hostname.
 //
 // sign reads the key from -key or, if absent, from $MQTT_RPCD_SIGNING_KEY.
-// call/update/status read the broker URL (with credentials) from $MQTT_URL so they never
+// call/update/status/reboot/shutdown read the broker URL (with credentials) from $MQTT_URL so they never
 // appear on a command line. -base is the reader's base topic, e.g.
 // trakrf.id/cs463-212.
 package main
@@ -54,6 +60,8 @@ func main() {
 		err = cmdUpdate(os.Args[2:])
 	case "status":
 		err = cmdStatus(os.Args[2:])
+	case "reboot", "shutdown":
+		err = cmdPower(os.Args[1], os.Args[2:])
 	default:
 		usage()
 	}
@@ -64,7 +72,7 @@ func main() {
 }
 
 func usage() {
-	fmt.Fprintln(os.Stderr, "usage: mqtt-rpcctl keygen|sign|call|update|status [flags]  (see -h on each)")
+	fmt.Fprintln(os.Stderr, "usage: mqtt-rpcctl keygen|sign|call|update|status|reboot|shutdown [flags]  (see -h on each)")
 	os.Exit(2)
 }
 

mqtt-rpc/cmd/mqtt-rpcctl/power.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+
+	"github.com/trakrf/platform/mqtt-rpc/internal/readerrpc"
+)
+
+// cmdPower handles `reboot` and `shutdown`: a clean reboot or power-off of the
+// reader host. Both take the reader offline, so both need -yes.
+func cmdPower(action string, args []string) error {
+	fs, base, ca := rpcFlags(action)
+	yes := fs.Bool("yes", false, "confirm: this takes the reader offline")
+	by := fs.String("by", defaultRequestedBy(), "who is asking, for the reader's log")
+	_ = fs.Parse(args)
+	if *base == "" {
+		return errors.New("-base is required")
+	}
+	method, params, err := powerRequest(action, *yes, *by)
+	if err != nil {
+		return err
+	}
+	return rpc(*base, *ca, method, params)
+}
+
+// powerRequest builds the Reader.Reboot or Reader.Shutdown request, refusing
+// without confirmation.
+func powerRequest(action string, yes bool, by string) (string, json.RawMessage, error) {
+	var method, cost string
+	switch action {
+	case "reboot":
+		method = readerrpc.MethodReboot
+		cost = "the reader stops reading and is off the broker for several minutes (2-7 after boot)"
+	case "shutdown":
+		method = readerrpc.MethodShutdown
+		cost = "the reader powers off and comes back only when someone on site power-cycles it"
+	default:
+		return "", nil, fmt.Errorf("unknown power action %q", action)
+	}
+	if !yes {
+		return "", nil, fmt.Errorf("%s: %s; pass -yes to go ahead", action, cost)
+	}
+	params, err := json.Marshal(readerrpc.PowerParams{RequestedBy: by})
+	return method, params, err
+}
+
+// defaultRequestedBy is $USER@hostname, or whichever half is known.
+func defaultRequestedBy() string {
+	host, _ := os.Hostname()
+	return os.Getenv("USER") + "@" + host
+}

mqtt-rpc/cmd/mqtt-rpcctl/power_test.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/trakrf/platform/mqtt-rpc/internal/readerrpc"
+)
+
+func TestPowerRequest(t *testing.T) {
+	cases := []struct {
+		name, action string
+		yes          bool
+		wantMethod   string
+		wantErr      string // substring; empty means no error
+	}{
+		{"reboot needs -yes", "reboot", false, "", "-yes"},
+		{"shutdown needs -yes and says what it costs", "shutdown", false, "", "power-cycle"},
+		{"reboot", "reboot", true, readerrpc.MethodReboot, ""},
+		{"shutdown", "shutdown", true, readerrpc.MethodShutdown, ""},
+		{"unknown action", "halt", true, "", "unknown"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			method, params, err := powerRequest(c.action, c.yes, "mike@bench")
+			if c.wantErr != "" {
+				if err == nil || !strings.Contains(err.Error(), c.wantErr) {
+					t.Fatalf("err = %v, want one containing %q", err, c.wantErr)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if method != c.wantMethod {
+				t.Errorf("method = %q, want %q", method, c.wantMethod)
+			}
+			var p readerrpc.PowerParams
+			if err := json.Unmarshal(params, &p); err != nil {
+				t.Fatal(err)
+			}
+			if p.RequestedBy != "mike@bench" {
+				t.Errorf("requested_by = %q, want mike@bench", p.RequestedBy)
+			}
+		})
+	}
+}
+
+func TestDefaultRequestedBy(t *testing.T) {
+	t.Setenv("USER", "mike")
+	if got := defaultRequestedBy(); !strings.HasPrefix(got, "mike@") || got == "mike@" {
+		t.Errorf("defaultRequestedBy = %q, want mike@<hostname>", got)
+	}
+}

mqtt-rpc/cmd/mqtt-rpcctl/status_test.go

@@ -14,6 +14,10 @@ func TestSummarizeStatus(t *testing.T) {
 			`online v1 api=healthy golden=applied cloudserver_fault="cleanSession is false; golden is true"`},
 		{"planned offline", `{"online":false,"reason":"auto-recovery","version":"v1"}`,
 			"offline (auto-recovery) v1"},
+		{"operator reboot", `{"online":false,"reason":"reboot","version":"v1"}`,
+			"offline (reboot) v1"},
+		{"operator shutdown", `{"online":false,"reason":"shutdown","version":"v1"}`,
+			"offline (shutdown) v1"},
 		{"last will", `{"online":false}`, "offline"},
 		{"older daemon without health", `{"online":true,"version":"v0"}`, "online v0"},
 		{"not json", `garbage`, "unparseable: garbage"},

mqtt-rpc/cmd/mqtt-rpcd/main.go

@@ -100,8 +100,12 @@ func main() {
 		}
 	}
 
+	// One host for the operator's Reader.Reboot/Shutdown and the monitor's
+	// automatic recovery.
+	host := readerd.NewSystemdHost(&logger)
 	d := readerd.New(cfg, adapter, &logger)
 	d.SetUpdater(upd)
+	d.SetHost(host)
 	if err := d.Start(); err != nil {
 		logger.Fatal().Err(err).Msg("start daemon")
 	}
@@ -111,12 +115,13 @@ func main() {
 	policy.AutoRecover = cfg.AutoRecover
 	policy.WedgeAfter = cfg.WedgeAfter
 	monitor := readerd.NewMonitor(policy, readerd.MonitorDeps{
-		Probe:     adapter.ProbeHealth,
-		Reconcile: reconcile,
-		Host:      readerd.NewSystemdHost(&logger),
-		Store:     readerd.RebootStore{Path: readerd.DefaultRebootStorePath},
-		Publish:   d.SetHealth,
-		GoingDown: d.GoingDown,
+		Probe:          adapter.ProbeHealth,
+		Reconcile:      reconcile,
+		Host:           host,
+		Store:          readerd.RebootStore{Path: readerd.DefaultRebootStorePath},
+		Publish:        d.SetHealth,
+		GoingDown:      d.GoingDown,
+		AbortGoingDown: d.AbortGoingDown,
 		CloudServer: func() string {
 			return d.CloudServerFault(cfg.CloudServer, time.Now())
 		},

mqtt-rpc/internal/readerd/cs463/adapter.go

@@ -234,10 +234,11 @@ func (a *Adapter) GetCapabilities(ctx context.Context) (readerrpc.Capabilities,
 			readerrpc.MethodGetStatus,
 			readerrpc.MethodGpoSet,
 		},
+		// Reader.Reboot and Reader.Shutdown run on the host, not the reader API;
+		// the daemon advertises them (readerd.Daemon.SetHost).
 		Unsupported: []string{
 			readerrpc.MethodScanStart,
 			readerrpc.MethodScanStop,
-			readerrpc.MethodReboot,
 		},
 	}, nil
 }

mqtt-rpc/internal/readerd/cs463/adapter_test.go

@@ -214,10 +214,11 @@ func TestGetCapabilities(t *testing.T) {
 	if !reflect.DeepEqual(caps.Supports, wantSupports) {
 		t.Errorf("supports = %v, want %v", caps.Supports, wantSupports)
 	}
+	// Reader.Reboot and Reader.Shutdown are the daemon's, not the adapter's: it
+	// advertises them itself when it has a host to run them on.
 	wantUnsupported := []string{
 		readerrpc.MethodScanStart,
 		readerrpc.MethodScanStop,
-		readerrpc.MethodReboot,
 	}
 	if !reflect.DeepEqual(caps.Unsupported, wantUnsupported) {
 		t.Errorf("unsupported = %v, want %v", caps.Unsupported, wantUnsupported)

mqtt-rpc/internal/readerd/daemon.go

@@ -52,9 +52,16 @@ type Daemon struct {
 	// publishRetained is the retained-status seam, set by Start (status.go).
 	publishRetained func(topic string, payload []byte) error
 
-	statusMu sync.Mutex
-	health   *readerrpc.Health // latest from the health monitor
-	reason   string            // set by GoingDown before a planned reboot
+	// host runs Reader.Reboot and Reader.Shutdown (SetHost); nil disables them.
+	// after schedules the delayed power action; a seam for tests.
+	host       Host
+	powerDelay time.Duration
+	after      func(time.Duration, func())
+
+	statusMu     sync.Mutex
+	health       *readerrpc.Health // latest from the health monitor
+	reason       string            // set by GoingDown before a planned reboot or power-off
+	powerPending string            // a Reader.Reboot/Shutdown accepted and not yet acted on
 
 	anchor anchorCheck // the last broker handshake, judged as CSL would
 }
@@ -83,6 +90,9 @@ func New(cfg Config, adapter Adapter, log *zerolog.Logger) *Daemon {
 
 		started:     time.Now(),
 		commitAfter: DefaultCommitAfter,
+
+		powerDelay: DefaultPowerDelay,
+		after:      func(wait time.Duration, f func()) { time.AfterFunc(wait, f) },
 	}
 }
 
@@ -311,11 +321,17 @@ func (d *Daemon) dispatch(ctx context.Context, req readerrpc.Request) readerrpc.
 		if err != nil {
 			return readerrpc.NewError(req, readerrpc.CodeInternal, err.Error())
 		}
-		return d.result(req, d.withDaemonInfo(caps))
+		return d.result(req, d.withPowerMethods(d.withDaemonInfo(caps)))
 
 	case readerrpc.MethodDaemonUpdate:
 		return d.daemonUpdate(req)
 
+	case readerrpc.MethodReboot:
+		return d.power(req, readerrpc.StatusReasonReboot)
+
+	case readerrpc.MethodShutdown:
+		return d.power(req, readerrpc.StatusReasonShutdown)
+
 	case readerrpc.MethodGetOperProfile:
 		var p readerrpc.OperProfileParams
 		if len(req.Params) > 0 {

mqtt-rpc/internal/readerd/health.go

@@ -70,6 +70,9 @@ type MonitorDeps struct {
 	Publish func(readerrpc.Health)
 	// GoingDown is called just before an automatic reboot, with the reason.
 	GoingDown func(reason string)
+	// AbortGoingDown takes GoingDown back when the reboot was refused, so a
+	// reader that stayed up is not announced as offline.
+	AbortGoingDown func()
 	// CloudServer returns the golden CloudServer fault ("" when clean), re-run
 	// every tick so an operator's fix, or a firmware update reverting one,
 	// shows without a daemon restart. nil checks nothing.
@@ -298,6 +301,9 @@ func (m *Monitor) reboot(now time.Time, h *readerrpc.Health) {
 	}
 	if err := m.deps.Host.Reboot(); err != nil {
 		m.log.Error().Err(err).Msg("auto-recovery: reboot failed; retrying after the next window")
+		if m.deps.AbortGoingDown != nil {
+			m.deps.AbortGoingDown()
+		}
 	}
 }