fix(history): keep working measurements when a device is offline

.changeset/history-offline-devices.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Keep history for working devices when another device is offline. Show missing power, state of charge and energy as unknown, preserve gaps through retention and backup, and flag incomplete measurements or an invalid household balance. Separate additive schema-7 tables keep partial rows out of older readers and preserve them across a native rollback.

docs/architecture.md

@@ -58,6 +58,17 @@ a reading after the observed connection and no measured charging. An ambiguous
 match cannot finish the goal. A charger declining current is reported as a refusal, never
 as an invented battery level or proof that the target was reached.
 
+### History measurement quality
+
+An offline flow leaves working devices recording. Missing inputs make their
+values and the derived household balance unknown. Separate additive schema-7
+tables preserve partial chart and energy rows, including missing-value evidence,
+through retention, downsampling and full backup. A recovered sample cannot fill
+an earlier gap. Older Cores see only complete rows in the original tables;
+their integration cursor cannot bridge a partial interval after rollback.
+Partial rows remain stored and become visible again on the newer Core.
+
+
 ### Control state and disk writes
 
 Core loads EV restart records before starting control. During operation, a

go/cmd/ftw/energy_history_test.go

@@ -82,16 +82,16 @@ func TestBuildHistoryPointExcludesUnavailableTelemetry(t *testing.T) {
 
 	point, available := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "site-meter"},
 		now.UnixMilli(), time.Minute)
-	if available {
-		t.Fatalf("offline site meter produced history: %+v", point)
+	if !available || point.Missing&state.MissingGrid == 0 {
+		t.Fatalf("offline site meter became known or lost live devices: %+v", point)
 	}
 
 	tel.Update("site-meter", telemetry.DerMeter, 800, nil, nil)
 	tel.RecordDriverSuccess("site-meter")
 	point, available = buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "site-meter"},
 		now.UnixMilli(), time.Minute)
-	if available {
-		t.Fatalf("missing PV/battery treated as zero in household history: %+v", point)
+	if !available || point.Missing&(state.MissingPV|state.MissingBattery|state.MissingLoad) == 0 {
+		t.Fatalf("missing PV/battery became known or lost live meter: %+v", point)
 	}
 	// The aggregate becomes known only after every significant flow recovers.
 	tel.Update("stale-pv", telemetry.DerPV, -800, nil, nil)
@@ -140,8 +140,8 @@ func TestBuildHistoryPointRequiresFreshEVAndV2X(t *testing.T) {
 
 	point, available := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "site-meter"},
 		now.UnixMilli(), time.Minute)
-	if available {
-		t.Fatalf("aged EV/V2X became zero household demand: %+v", point)
+	if !available || point.Missing&(state.MissingEV|state.MissingV2X|state.MissingLoad) == 0 {
+		t.Fatalf("aged EV/V2X became known or lost live meter: %+v", point)
 	}
 
 }
@@ -162,6 +162,9 @@ func TestPersistTelemetryTickUsesPersistenceFreshness(t *testing.T) {
 				t.Fatal(err)
 			}
 			t.Cleanup(func() { _ = st.Close() })
+			if err := st.EnableHistoryAggregation(); err != nil {
+				t.Fatal(err)
+			}
 			tickMS := time.Now().Add(-5 * time.Second).UnixMilli()
 			tel := telemetry.NewStore()
 			tel.EnsureDriverHealth("meter")
@@ -247,6 +250,9 @@ func TestStaleMeterTickKeepsSamplesAndIndependentLedgerWithoutDispatch(t *testin
 	if err != nil {
 		t.Fatal(err)
 	}
+	if err := st.EnableHistoryAggregation(); err != nil {
+		t.Fatal(err)
+	}
 	t.Cleanup(func() { _ = st.Close() })
 	pvDeviceID, err := st.RegisterDevice(state.Device{
 		DriverName: "solar", Make: "PV Maker", Serial: "PV-42",
@@ -293,8 +299,8 @@ func TestStaleMeterTickKeepsSamplesAndIndependentLedgerWithoutDispatch(t *testin
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(history) != 0 {
-		t.Fatalf("stale meter wrote legacy history: %+v", history)
+	if len(history) == 0 || history[0].Missing&state.MissingGrid == 0 {
+		t.Fatalf("stale meter lost independent history or became known: %+v", history)
 	}
 	samples, err := st.LoadSeries("solar", "pv_w", 0, time.Now().Add(time.Second).UnixMilli(), 0)
 	if err != nil {

go/cmd/ftw/energy_identity_test.go

@@ -52,6 +52,9 @@ func TestStartupAliasCannotReplayCountersWhileRawHistoryKeepsWriting(t *testing.
 		t.Fatal(err)
 	}
 	t.Cleanup(func() { _ = st.Close() })
+	if err := st.EnableHistoryAggregation(); err != nil {
+		t.Fatal(err)
+	}
 	serial := state.Device{DriverName: "meter", Make: "inverter", Serial: "one", MAC: "aa:bb:cc:dd:ee:ff"}
 	if _, err := st.RegisterDevice(serial); err != nil {
 		t.Fatal(err)

go/cmd/ftw/history_options_test.go

@@ -54,20 +54,19 @@ func TestHistoryKeepsRunningWithOptionalPVAndUnconfirmedIdentity(t *testing.T) {
 		t.Fatalf("history stopped: point=%+v available=%v", point, ok)
 	}
 
-	// A known source that goes silent still stops history; only optional,
-	// never-reported PV is treated as absent.
+	// Missing flows remain unknown while independent live history continues.
 	tel.DriverHealthMut("pixii").SetOffline()
-	if _, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"},
-		time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); ok {
-		t.Fatal("offline battery treated as zero in history")
+	if p, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"},
+		time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); !ok || p.Missing&(state.MissingBattery|state.MissingLoad) == 0 {
+		t.Fatal("offline battery lost independent history or became known")
 	}
 	tel.Update("pixii", telemetry.DerBattery, 500, nil, nil)
 	tel.RecordDriverSuccess("pixii")
 	// Once the Zap does report PV, a stale PV reading stops history again.
 	tel.Update("sourceful-zap", telemetry.DerPV, -100, nil, nil)
 	tel.Get("sourceful-zap", telemetry.DerPV).UpdatedAt = time.Now().Add(-2 * time.Minute)
-	if _, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"},
-		time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); ok {
-		t.Fatal("stale reported PV treated as zero in history")
+	if p, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"},
+		time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); !ok || p.Missing&state.MissingPV == 0 {
+		t.Fatal("stale PV lost independent history or became known")
 	}
 }

go/cmd/ftw/history_partial_test.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"github.com/srcfl/ftw/go/internal/state"
+	"github.com/srcfl/ftw/go/internal/telemetry"
+	"testing"
+	"time"
+)
+
+func TestHistoryRetainsLiveMeterWhenBatteryOffline(t *testing.T) {
+	tel := telemetry.NewStore()
+	for _, name := range []string{"meter", "battery"} {
+		tel.EnsureDriverHealth(name)
+	}
+	tel.Update("meter", telemetry.DerMeter, 1200, nil, nil)
+	tel.RecordDriverSuccess("meter")
+	tel.DriverHealthMut("battery").SetOffline()
+	opts := telemetry.ForecastOptions{ExpectedFlows: []telemetry.ForecastFlow{{Driver: "battery", DerType: telemetry.DerBattery}}}
+	p, available := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "meter"}, time.Now().UnixMilli(), time.Minute, opts)
+	if !available || p.GridW != 1200 {
+		t.Fatalf("live meter lost when battery offline: available=%v point=%+v", available, p)
+	}
+}
+
+func TestHistoryConfiguredPVWithoutMeasurementIsUnknown(t *testing.T) {
+	tel := telemetry.NewStore()
+	tel.EnsureDriverHealth("meter")
+	tel.Update("meter", telemetry.DerMeter, 1200, nil, nil)
+	tel.RecordDriverSuccess("meter")
+	p, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "meter"}, time.Now().UnixMilli(), time.Minute, telemetry.ForecastOptions{HouseholdInvalidReason: "configured_pv_without_measurement_source"})
+	if !ok || p.Missing&state.MissingPV == 0 {
+		t.Fatalf("unmeasured configured PV reported as zero: %+v", p)
+	}
+}

go/cmd/ftw/main.go

@@ -2505,6 +2505,7 @@ func main() {
 		SiteDispatchBlocked: func() string {
 			return siteDispatchNow(tel, cfg, cfgMu, ctrl, ctrlMu, time.Now()).Reason
 		},
+		HistoryMeasurementSources: func() telemetry.ForecastOptions { return forecastSettings.Snapshot().HistoryOptions },
 		SiteMeasurementSources: func() telemetry.ForecastOptions {
 			return forecastSettings.Snapshot().Options
 		},
@@ -3894,7 +3895,7 @@ func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersis
 	energyObservations := buildEnergyObservations(st, tel, ctrl, hp, identity)
 	filtered := energyObservations[:0]
 	for _, observation := range energyObservations {
-		if !historyAvailable && observation.AssetKind == state.AssetObservedConsumer {
+		if (!historyAvailable || hp.Missing&state.MissingLoad != 0) && observation.AssetKind == state.AssetObservedConsumer {
 			continue
 		}
 		if observation.AssetKind != state.AssetObservedConsumer {
@@ -3921,102 +3922,81 @@ func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersis
 }
 
 func buildHistoryPoint(tel *telemetry.Store, ctrl tickPersistControl, nowMs int64, historyMaxAge time.Duration, options ...telemetry.ForecastOptions) (state.HistoryPoint, bool) {
-	unavailable := state.HistoryPoint{TsMs: nowMs}
-	if tel == nil || ctrl.SiteMeterDriver == "" {
-		return unavailable, false
+	if tel == nil {
+		return state.HistoryPoint{TsMs: nowMs}, false
 	}
 	opts := telemetry.ForecastOptions{MaxAge: historyMaxAge}
 	if len(options) > 0 {
 		opts = options[0]
 		opts.MaxAge = historyMaxAge
 	}
-	// Polls can update telemetry while dispatch runs. Keep the tick's history
-	// timestamp, but judge reading freshness when this snapshot is captured.
+	opts.UseWatchdogTimeout = true
 	balance := tel.ForecastMeasurementNow(ctrl.SiteMeterDriver, opts)
-	if !balance.Valid {
-		return unavailable, false
-	}
-	now := balance.At
-	gridW, pvW, batW := balance.GridW, balance.PVW, balance.BatteryW
-	evW, v2xW, loadW := balance.EVW, balance.V2XW, balance.HouseholdW
-	readingUsable := func(driver string, updatedAt time.Time) bool {
-		h := tel.DriverHealth(driver)
-		if !h.TelemetryLive() || updatedAt.After(now) {
-			return false
+	p := state.HistoryPoint{TsMs: nowMs, GridW: balance.GridW, PVW: balance.PVW, BatW: balance.BatteryW, LoadW: balance.HouseholdW}
+	for kind, flag := range map[telemetry.DerType]state.HistoryMissing{telemetry.DerMeter: state.MissingGrid, telemetry.DerPV: state.MissingPV, telemetry.DerBattery: state.MissingBattery, telemetry.DerEV: state.MissingEV, telemetry.DerV2X: state.MissingV2X} {
+		if !balance.Known[kind] {
+			p.Missing |= flag
 		}
-		maxAge := historyMaxAge
-		if h.WatchdogTimeoutOverride > 0 {
-			maxAge = h.WatchdogTimeoutOverride
-		}
-		return maxAge <= 0 || now.Sub(updatedAt) <= maxAge
 	}
-	var avgSoC, sumSoC float64
-	var socCount int
-	for _, r := range tel.ReadingsByType(telemetry.DerBattery) {
-		if readingUsable(r.Driver, r.UpdatedAt) && r.SoC != nil {
-			sumSoC += *r.SoC
-			socCount++
-		}
+	if opts.PVInvalidReason != "" {
+		p.Missing |= state.MissingPV
 	}
-	if socCount > 0 {
-		avgSoC = sumSoC / float64(socCount)
-	}
-
-	// Per-driver detail packed into the JSON column. The schema is
-	// schema-less by design — UI code reads what it understands and
-	// ignores the rest, so drivers can add fields without a migration.
-	perDriver := make(map[string]map[string]float64)
-	for name := range tel.AllHealth() {
-		row := map[string]float64{}
-		if r := tel.Get(name, telemetry.DerBattery); r != nil &&
-			readingUsable(name, r.UpdatedAt) {
-			row["bat_w"] = r.SmoothedW
-			if r.SoC != nil {
-				row["soc"] = *r.SoC
-			}
+	if !balance.Valid {
+		p.Missing |= state.MissingLoad
+	}
+	if balance.Reason == "invalid_balance" || strings.HasPrefix(balance.Reason, "duplicate_") || strings.HasPrefix(balance.Reason, "invalid_sign:") || balance.Reason == "time_skew" {
+		p.Missing |= state.BalanceMismatch
+	}
+	perDriver := map[string]map[string]any{}
+	var socSum float64
+	var batteries, socCount int
+	var hasPower bool
+	for _, source := range balance.Sources {
+		hasPower = hasPower || source.PowerW != nil
+		name := source.Flow.Driver
+		if name == "" {
+			continue
 		}
-		if r := tel.Get(name, telemetry.DerPV); r != nil &&
-			readingUsable(name, r.UpdatedAt) {
-			row["pv_w"] = r.SmoothedW
-		}
-		if r := tel.Get(name, telemetry.DerMeter); r != nil &&
-			readingUsable(name, r.UpdatedAt) {
-			row["meter_w"] = r.SmoothedW
-		}
-		// EV charge power: required for the live chart's EV series
-		// (web/app.js reads `d.ev_w` per driver from /api/history).
-		// Without it the chart's EV trace is always zero — the in-memory
-		// /api/status DOES carry ev_w, but history rows never did until
-		// this row was added.
-		if r := tel.Get(name, telemetry.DerEV); r != nil &&
-			readingUsable(name, r.UpdatedAt) {
-			row["ev_w"] = r.SmoothedW
-		}
-		if r := tel.Get(name, telemetry.DerV2X); r != nil &&
-			readingUsable(name, r.UpdatedAt) {
-			row["v2x_w"] = r.SmoothedW
-			if r.SoC != nil {
-				row["v2x_vehicle_soc"] = *r.SoC
+		row := perDriver[name]
+		if row == nil {
+			row = map[string]any{}
+			perDriver[name] = row
+		}
+		field := map[telemetry.DerType]string{telemetry.DerMeter: "meter_w", telemetry.DerPV: "pv_w", telemetry.DerBattery: "bat_w", telemetry.DerEV: "ev_w", telemetry.DerV2X: "v2x_w"}[source.Flow.DerType]
+		if field != "" {
+			row[field] = source.PowerW
+		}
+		if source.Flow.DerType == telemetry.DerBattery {
+			batteries++
+			row["soc"] = source.SoC
+			if source.SoC != nil {
+				socSum += *source.SoC
+				socCount++
 			}
 		}
-		perDriver[name] = row
+		if source.Flow.DerType == telemetry.DerV2X {
+			row["v2x_vehicle_soc"] = source.SoC
+		}
 	}
-	targets := make(map[string]float64)
+	if socCount > 0 && socCount == batteries {
+		p.BatSoC = socSum / float64(socCount)
+	} else {
+		p.Missing |= state.MissingSoC
+	}
+	// Numeric storage remains finite. The durable mask, never these placeholders,
+	// decides whether a chart or energy total is known.
+	p.ClearMissingValues()
+	targets := map[string]float64{}
 	for _, t := range ctrl.LastTargets {
 		targets[t.Driver] = t.TargetW
 	}
-	jsonBlob, _ := json.Marshal(map[string]any{
-		"drivers":                      perDriver,
-		"targets":                      targets,
-		"ev_w":                         evW,
-		"v2x_w":                        v2xW,
-		"load_house_w":                 loadW,
-		"forecast_measurement_quality": "complete_instantaneous_balance_v1",
-	})
-	return state.HistoryPoint{
-		TsMs: nowMs, GridW: gridW, PVW: pvW, BatW: batW, LoadW: loadW, BatSoC: avgSoC,
-		JSON: string(jsonBlob),
-	}, true
+	quality := "incomplete_measurement"
+	if balance.Valid {
+		quality = "complete_instantaneous_balance_v1"
+	}
+	blob, _ := json.Marshal(map[string]any{"drivers": perDriver, "targets": targets, "ev_w": p.Missing.Value(state.MissingEV, balance.EVW), "v2x_w": p.Missing.Value(state.MissingV2X, balance.V2XW), "load_house_w": p.LoadW, "forecast_measurement_quality": quality, "history_reason": balance.Reason})
+	p.JSON = string(blob)
+	return p, hasPower
 }
 
 func restoreLatestMPCDiagnostic(st *state.Store, svc *mpc.Service, now time.Time) {