feat(control): a journal of every change in the answer to "Are we in control?"

.changeset/control-journal.md

@@ -0,0 +1,13 @@
+---
+"ftw": minor
+---
+
+The box keeps a control journal. Every few seconds it records the answer to
+"Are we in control?" for each controlled device, each source, whether the site
+may dispatch and the operating mode; a change starts a new stretch, with the
+power Core asked for and the power delivered while it held. Each scored plan
+slot is kept too, with its planned and delivered energy. Read it over any
+window with GET /api/control/journal, and as an agent with events.get, so the
+owner, an assistant or support can learn afterwards why the app warned. A
+notification the database was too busy to record is now written once it is
+free.

docs/agents.md

@@ -384,7 +384,7 @@ The registry's `reads` table names each read and the scope it needs.
 | `hist.query` | As for a phone | `hist.chunk`, `hist.end` |
 | `health.get` | `{}` | `health {version, uptimeS, clock, drivers: {ok, degraded, faulted, offline}, meter: {srcState, ageMs}, dispatchBlockedBy}` |
 | `drivers.get` | `{}` | `drivers {drivers: [{src, kind, state, ageMs}]}` |
-| `events.get` | `{n, beforeMs, afterMs}`, `n` at most 50 | `events {events, more}`, each FTW Cloud's `event` body with agent ids |
+| `events.get` | `{n, beforeMs, afterMs, src, minSeverity}`, `n` at most 50 | `events {spans, notices, events, more}`: the control journal over the window, newest first (see below) |
 | `logs.get` | `{n, level, beforeMs, afterMs}`; `n` at most 200, `level` `WARN` by default | `logs {lines: [{ts, level, src, msg}], dropped, more}` |
 | `config.get` | `{section}` | `config {sections}` without a section; `config {section, doc}` with one |
 | `diag.run` | `{check, src}`; `check` is `clock`, `link`, `meter`, `planner` or `driver` | `diag {check, ok, facts}` |
@@ -407,8 +407,22 @@ The registry's `reads` table names each read and the scope it needs.
   source that blocks dispatch goes by its agent id. Planner solve times,
   fallbacks, refusals and plan against actual join it when the box keeps
   them; a field the box cannot fill is absent.
-- The box answers `events.get` and `diag.run` with `E_UNKNOWN_OP` until it
-  builds them. `cars.get` carries no name, maker, VIN, identifier or source,
+- `events.get` reads the box's control journal: each stretch in which a
+  device function, a source, the site's dispatch or the mode held one
+  verdict, each scored plan slot, and the notifications the box sent, over
+  `afterMs`..`beforeMs` (the last 24 hours by default). A span gives `kind`
+  (battery, ev, pv, v2x_charger, device, source, dispatch, mode, slot),
+  `src` by agent id, Core's `status`, `reason` and `severity` as in
+  `control_feedback`, `startMs` and `endMs` (0 while it holds), `samples`,
+  `requestedW` and `actualW` as min, max and mean, and `last`: the numbers
+  of the latest verdict by field name. Text leaves only as Core's own codes,
+  never names. `minSeverity: "warning"` keeps warnings and alarms; `src`
+  keeps one source. `events` stays empty; it is kept for older readers.
+  A source span is the driver's health as Core's watchdog sees it. A slot
+  span is the battery fleet's planned and delivered energy over one plan
+  slot, as average watts. A warning starts a span at once; a change to an
+  info verdict counts once it has held 15 s, from when it was first seen.
+- The box answers `diag.run` with `E_UNKNOWN_OP` until it builds it. `cars.get` carries no name, maker, VIN, identifier or source,
   and the redactor takes those out of `logs.get` and `config.get`; a box
   without the car store answers it `E_UNAVAILABLE`. `plan.get` names each
   charger and its car by agent id.

docs/roadmap.md

@@ -85,7 +85,7 @@ or reopen closed issues.
 |---|---|---|
 | Simple setup and first-day value | Discover mixed equipment. Confirm the main fuse and site meter. Read battery capacity where possible and ask for kWh when needed. Power settings and solar kWp are optional where safe device information and learning allow. Provide useful initial load and PV forecasts. | A fresh install reaches useful automatic operation without panel drawings or expert settings. Missing data and wrong start estimates have tested behaviour. Record device identity, known limits and uncertainty. Validate on named hardware combinations as well as simulators. |
 | A clear commissioning result | Check commands and measured response within known limits. Distinguish working telemetry from working control. Exclude failed control from both the plan and dispatch. | Show request, device response, measured effect and timing. Cover delayed response, refusal, disconnect, stale site data and recovery. A short commissioning test does not claim full hardware qualification. |
-| Verified control: “Are we in control?” | Keep the fast local feel. Make request, acceptance, command, response, physical effect and freshness visible in the normal Flow experience. | Browser review on desktop and mobile, timing measurements on a target box, and traces with different sampling rates. A successful call, echoed setpoint or old reading never appears as a verified physical result. Test accepted, measured and confirmed evidence per device, and the status each one shows. Confirm one device while another is offline; keep unavailable background flows in the residual. Show overview status, alarm on lost measured proof after the response wait, and clear it on measured recovery. Include direction, tolerance, response delay, simultaneous load/solar changes, shared measurement sources, overridden setpoints, device limits and recovery. A reduced request remains visibly limited even when the device follows the reduced command. Examine existing Live and Flow views before deciding their final layout. |
+| Verified control: “Are we in control?” | Keep the fast local feel. Make request, acceptance, command, response, physical effect and freshness visible in the normal Flow experience. Keep each change of the answer on the box, with the power asked for and delivered, so the owner, an agent or support can learn afterwards why the app warned. | Browser review on desktop and mobile, timing measurements on a target box, and traces with different sampling rates. A successful call, echoed setpoint or old reading never appears as a verified physical result. Test accepted, measured and confirmed evidence per device, and the status each one shows. Confirm one device while another is offline; keep unavailable background flows in the residual. Show overview status, alarm on lost measured proof after the response wait, and clear it on measured recovery. Include direction, tolerance, response delay, simultaneous load/solar changes, shared measurement sources, overridden setpoints, device limits and recovery. A reduced request remains visibly limited even when the device follows the reduced command. Examine existing Live and Flow views before deciding their final layout. |
 | Good automatic planning | Use site physics and charge/discharge efficiency. Default wear cost is zero; users may opt in. Keep hard SoC limits separate from forecast-based caution and explicit backup needs. | Cold-start and learned forecasts, stale inputs, multiple assets and unavailable optimizer paths have tests. Backtests use held-out periods and report uncertainty. Defaults and persisted settings agree across UI, Core and worker. Site runs establish practical benefit. |
 | Reliable daily charging | Persistent weekday target and deadline. Offline-car estimates. Direct SoC slider after connection, no extra save, prompt replanning, one-action Charge now and notifications when action is needed. | Test from app intent to charger/car response and delivered energy, including missed-goal risk, absent vehicle cloud, unknown SoC, reconnect and restart. Test notifications with the app closed. Confirm physical charging separately from simulation. |
 | Useful analysis and fair savings | Keep enough provenance to explain plans and outcomes. Main savings target compares with ordinary self-consumption on the same installation. | Actual cost reconciles with measured import/export and prices. Specify EV behaviour, initial and final stored-energy accounting, efficiency and coverage. Show missing and negative results. Label the current no-PV/no-battery comparison as total site value until replacement is verified. |

go/cmd/ftw/app_link.go

@@ -797,8 +797,10 @@ func startAppLink(
 	agents.access = &appuplink.AgentAccess{Enroll: enroll, Clock: agentClock, Push: agents.push,
 		Holds: endAgentHolds(loadpoints)}
 	agents.state = appproto.NewAgentState()
+	var journalSrc appproto.JournalSource
 	if st != nil {
 		agents.audit = newStateAudit(ctx, st)
+		journalSrc = stateJournal{st: st}
 	}
 	caps = append(caps, appproto.CapAccessAgents, appproto.CapAccessInvites)
 	// One view for every agent session, so a charger's plug-in count is
@@ -838,6 +840,7 @@ func startAppLink(
 				Agents:       agents.state,
 				Audit:        agents.audit,
 				Logs:         ringLogs{ring: agents.logs},
+				Journal:      journalSrc,
 				Redactor:     agents.redactor,
 				Views:        views,
 				AgentKey:     agents.key,

go/cmd/ftw/control_journal.go

@@ -0,0 +1,131 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/appproto"
+	"github.com/srcfl/ftw/go/internal/journal"
+	"github.com/srcfl/ftw/go/internal/state"
+	"github.com/srcfl/ftw/go/protocol"
+)
+
+const (
+	// controlJournalEvery is how often the journal samples the box's
+	// answer to "Are we in control?".
+	controlJournalEvery = 5 * time.Second
+	// controlJournalKeep is how far back the journal answers.
+	controlJournalKeep = 366 * 24 * time.Hour
+)
+
+// controlJournal runs the recorder on its own goroutine until shutdown.
+type controlJournal struct {
+	rec  *journal.Recorder
+	stop context.CancelFunc
+	done chan struct{}
+}
+
+// newControlJournal ends the spans a previous run left open, at the last
+// moment that run saw them hold, and returns the journal for this one.
+func newControlJournal(st *state.Store) *controlJournal {
+	if n, err := st.CloseOpenControlSpans(context.Background()); err != nil {
+		slog.Warn("control journal: could not close spans the last run left open", "err", err)
+	} else if n > 0 {
+		slog.Info("control journal: closed spans the last run left open", "spans", n)
+	}
+	return &controlJournal{rec: journal.New(st)}
+}
+
+// start samples, writes and prunes until ctx ends or shutdown is called.
+func (j *controlJournal) start(ctx context.Context, sample func(time.Time) []journal.Verdict, st *state.Store) {
+	ctx, j.stop = context.WithCancel(ctx)
+	j.done = make(chan struct{})
+	go func() {
+		defer close(j.done)
+		runControlJournal(ctx, j.rec, sample, st)
+	}()
+}
+
+// shutdown stops the sampling, then ends every open span now and writes
+// them, so a clean stop leaves no span running through the time the box
+// was down.
+func (j *controlJournal) shutdown() {
+	if j.stop != nil {
+		j.stop()
+		select {
+		case <-j.done:
+		case <-time.After(2 * controlJournalEvery):
+		}
+	}
+	now := time.Now()
+	j.rec.CloseAll(now)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	j.rec.Flush(ctx, now)
+}
+
+// runControlJournal samples, writes and prunes until ctx ends.
+func runControlJournal(ctx context.Context, rec *journal.Recorder, sample func(time.Time) []journal.Verdict, st *state.Store) {
+	t := time.NewTicker(controlJournalEvery)
+	defer t.Stop()
+	var prunedAt time.Time
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case now := <-t.C:
+			rec.Observe(now, sample(now))
+			flushCtx, cancel := context.WithTimeout(ctx, controlJournalEvery)
+			rec.Flush(flushCtx, now)
+			cancel()
+			if now.Sub(prunedAt) < 24*time.Hour {
+				continue
+			}
+			prunedAt = now
+			if n, err := st.PruneControlSpans(ctx, now.Add(-controlJournalKeep).UnixMilli()); err != nil {
+				slog.Warn("control journal: prune failed", "err", err)
+			} else if n > 0 {
+				slog.Info("control journal: pruned old spans", "spans", n)
+			}
+		}
+	}
+}
+
+// stateJournal serves events.get from the control journal and the
+// notification history.
+type stateJournal struct{ st *state.Store }
+
+func (j stateJournal) Journal(ctx context.Context, q appproto.JournalQuery) ([]appproto.JournalSpan, []appproto.JournalNotice, bool, error) {
+	spans, more, err := j.st.ControlSpans(ctx, state.ControlSpanQuery{
+		SinceMs: q.SinceMs, UntilMs: q.UntilMs, Driver: q.Driver, Severities: q.Severities, Limit: q.Limit,
+	})
+	if err != nil {
+		return nil, nil, false, err
+	}
+	notices, noticesMore, err := j.st.NotificationsBetween(ctx, q.SinceMs, q.UntilMs, q.Driver, q.Limit)
+	if err != nil {
+		return nil, nil, false, err
+	}
+	out := make([]appproto.JournalSpan, 0, len(spans))
+	for _, sp := range spans {
+		out = append(out, appproto.JournalSpan{
+			Driver: sp.Driver, Kind: sp.Kind, Status: sp.Status, Reason: sp.Reason, Severity: sp.Severity, Mode: sp.Mode,
+			StartMs: sp.StartMs, EndMs: sp.EndMs, Samples: sp.Samples,
+			Requested: powerRange(sp.Requested), Actual: powerRange(sp.Actual), Last: sp.Last,
+		})
+	}
+	var ns []appproto.JournalNotice
+	for _, n := range notices {
+		ns = append(ns, appproto.JournalNotice{TsMs: n.TsMs, Type: n.EventType, Driver: n.Driver, Status: n.Status, Priority: n.Priority})
+	}
+	return out, ns, more || noticesMore, nil
+}
+
+func powerRange(p state.PowerStats) *protocol.PowerRange {
+	mean, ok := p.Mean()
+	if !ok {
+		return nil
+	}
+	return &protocol.PowerRange{Min: p.Min, Max: p.Max, Mean: mean}
+}

go/cmd/ftw/control_journal_test.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/appproto"
+	"github.com/srcfl/ftw/go/internal/journal"
+	"github.com/srcfl/ftw/go/internal/state"
+)
+
+func openJournalStore(t *testing.T) *state.Store {
+	t.Helper()
+	st, err := state.Open(filepath.Join(t.TempDir(), "state.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+	return st
+}
+
+// A stop ends the spans now and writes them; the next start finds none
+// left open.
+func TestStopControlJournalClosesAndWrites(t *testing.T) {
+	st := openJournalStore(t)
+	j := newControlJournal(st)
+	now := time.Now()
+	sample := func(time.Time) []journal.Verdict {
+		return []journal.Verdict{{Subject: "site:dispatch", Kind: "dispatch", Status: "free", Severity: "info"}}
+	}
+	j.rec.Observe(now, sample(now))
+	j.start(context.Background(), sample, st)
+	j.shutdown()
+	spans, _, err := st.ControlSpans(context.Background(), state.ControlSpanQuery{SinceMs: 0, UntilMs: now.Add(time.Minute).UnixMilli()})
+	if err != nil || len(spans) != 1 || spans[0].EndMs == 0 {
+		t.Fatalf("spans %+v err %v", spans, err)
+	}
+	if n, _ := st.CloseOpenControlSpans(context.Background()); n != 0 {
+		t.Fatalf("%d spans left open after a clean stop", n)
+	}
+}
+
+// events.get for one source reads that source's spans and notices only.
+func TestStateJournalKeepsOneSource(t *testing.T) {
+	st := openJournalStore(t)
+	ctx := context.Background()
+	spans := []*state.ControlSpan{
+		{Subject: "battery:sungrow", Driver: "sungrow", Kind: "battery", Status: "limited", Severity: "warning", StartMs: 100, UpdatedMs: 200},
+		{Subject: "ev:easee", Driver: "easee", Kind: "ev", Status: "following", Severity: "info", StartMs: 100, UpdatedMs: 200},
+	}
+	spans[0].Actual.Add(ptrW(1250))
+	if err := st.SaveControlSpans(ctx, spans); err != nil {
+		t.Fatal(err)
+	}
+	for _, d := range []string{"sungrow", "easee"} {
+		if err := st.RecordNotification(state.NotificationEntry{TsMs: 150, EventType: "control.not_following", Driver: d, Status: "sent"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	got, notices, more, err := stateJournal{st: st}.Journal(ctx, appproto.JournalQuery{SinceMs: 0, UntilMs: 1_000, Driver: "sungrow", Limit: 50})
+	if err != nil || more || len(got) != 1 || got[0].Kind != "battery" || got[0].Actual == nil || got[0].Actual.Mean != 1250 || got[0].Requested != nil {
+		t.Fatalf("spans %+v more %v err %v", got, more, err)
+	}
+	if len(notices) != 1 || notices[0].Driver != "sungrow" {
+		t.Fatalf("notices %+v", notices)
+	}
+}
+
+// A locked database delays a notification row; it does not lose it.
+func TestRecordWithRetryWritesAfterAFailure(t *testing.T) {
+	saved := notificationLogRetries
+	notificationLogRetries = []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond}
+	t.Cleanup(func() { notificationLogRetries = saved })
+	var tries atomic.Int32
+	done := make(chan struct{})
+	recordWithRetry("notification_log", func() error {
+		if tries.Add(1) < 3 {
+			return errors.New("database is locked (5) (SQLITE_BUSY)")
+		}
+		close(done)
+		return nil
+	})
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("not written after %d tries", tries.Load())
+	}
+}
+
+func ptrW(v float64) *float64 { return &v }

go/cmd/ftw/main.go

@@ -57,6 +57,7 @@ import (
 	"github.com/srcfl/ftw/go/internal/ftwcli"
 	"github.com/srcfl/ftw/go/internal/gatewayidentity"
 	"github.com/srcfl/ftw/go/internal/ha"
+	"github.com/srcfl/ftw/go/internal/journal"
 	"github.com/srcfl/ftw/go/internal/loadmodel"
 	"github.com/srcfl/ftw/go/internal/loadpoint"
 	"github.com/srcfl/ftw/go/internal/localzone"
@@ -618,6 +619,11 @@ func main() {
 	// Restore durable modes and targets. Forced Charge ends at restart;
 	// the planner makes a fresh plan instead of resuming that override.
 	restoreStoredMode(ctrl, st, cfg)
+	// The control journal keeps each change in the box's answer to "Are we
+	// in control?", and each scored plan slot, so a warning can be explained
+	// after it has passed (GET /api/control/journal, events.get).
+	ctrlJournal := newControlJournal(st)
+	ctrl.SlotScored = func(d control.SlotDelivery) { ctrlJournal.rec.Record(journal.SlotSpan(d)) }
 	storedTrust, _ := st.LoadConfig(config.StateKeyForecastTrust)
 	storedExport, _ := st.LoadConfig(config.StateKeyBatteryExport)
 	storedSafetyK, _ := st.LoadConfig(config.StateKeySafetyK)
@@ -2725,6 +2731,7 @@ func main() {
 	// The control.not_following rule reads the box's own answer to "Are we
 	// in control?", the one GET /api/status serves.
 	notifSvc.SetControlReader(srv.ControlStates)
+	ctrlJournal.start(ctx, srv.JournalVerdicts, st)
 	// FuseReader: on each HealthTick the fuse_over_limit rule reads
 	// the site meter's live per-phase currents from telemetry and
 	// compares against cfg.Fuse.MaxAmps. Closes over cfg + cfgMu so
@@ -2759,7 +2766,7 @@ func main() {
 		if !ok {
 			return
 		}
-		if err := st.RecordNotification(state.NotificationEntry{
+		entry := state.NotificationEntry{
 			TsMs:      ev.Time.UnixMilli(),
 			EventType: ev.EventType,
 			Driver:    ev.Driver,
@@ -2768,9 +2775,8 @@ func main() {
 			Priority:  ev.Priority,
 			Status:    ev.Status,
 			Error:     ev.Error,
-		}); err != nil {
-			slog.Warn("notification_log: record failed", "err", err)
 		}
+		recordWithRetry("notification_log", func() error { return st.RecordNotification(entry) })
 	})
 	// Web push runs beside whatever provider the config names. The provider
 	// itself was built before the uplink (the dead man's rows carry its
@@ -2933,6 +2939,7 @@ func main() {
 			slog.Info("shutting down")
 			cloudLink.Bye(cloudwire.ByeShutdown, cloudByeWait)
 			markNativeCleanExit(nativeRoot)
+			ctrlJournal.shutdown()
 			flushHistoryOnStop(st)
 			if err := st.RecordEvent("shutdown"); err != nil {
 				slog.Warn("failed to persist shutdown event", "err", err)
@@ -2948,6 +2955,7 @@ func main() {
 			// back "box out of reach" until the box is back.
 			cloudLink.Bye(cloudwire.ByeUpdate, cloudByeWait)
 			markNativeCleanExit(nativeRoot)
+			ctrlJournal.shutdown()
 			flushHistoryOnStop(st)
 			if err := st.RecordEvent("restart"); err != nil {
 				slog.Warn("failed to persist restart event", "err", err)