fix(history): derive household Load from interval source energy

.changeset/history-source-load.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Calculate recorded household Load from source energy over the same interval in Core. Different meter and charger reporting rates no longer depend on a valid instantaneous Load sample. The app, web history and daily totals use the same source balance. Keep signed source energy through aggregation, preserve real gaps, and leave live display and control unchanged.

docs/roadmap.md

@@ -58,6 +58,13 @@ using an estimate in planning. A model prediction must never verify itself or
 raise a safety limit. These are acceptance requirements, not shipped support.
 
 Necessary safety, security, recovery and support fixes continue throughout.
+
+Household Load history must balance source energy over the same interval in
+Core. Live updates remain immediate. Different reporting rates must not
+require a valid instantaneous Load sample before energy can be recorded.
+Verify the app, web, daily totals, tariff boundaries, restart and missing
+source cases against the same stored evidence.
+
 Correct misleading value labels when their scope is known; that need not wait
 for the new counterfactual model. Reading and analysis access for agents can
 also support the evidence work before agents receive control authority.

docs/site-convention.md

@@ -46,9 +46,19 @@ this one thing were the only thing running.
 Balance equation (all signed):
 
 ```
-grid_w = load_w + bat_w + pv_w
+grid_w = load_w + bat_w + pv_w + ev_w + v2x_w
 ```
 
+Live household Load uses the current fresh site-side powers. Historical Load
+uses source energy over a common interval, with EV charging excluded. Core
+holds each qualified snapshot forward until the next safe snapshot, at most
+15 seconds later, and stores signed source energy by UTC minute. It balances
+the sums when reading a chart interval, day or tariff period. A negative
+minute may contribute to a valid longer interval; it is never clipped first.
+Chart means need complete interval coverage. Missing sources, writer gaps,
+process stops and topology changes cannot become measured zero. This is an
+estimate from received measurements, not a correction of device clocks.
+
 In self-consumption mode, the controller drives the site meter toward the
 grid target (normally 0 W): it charges from live surplus and may discharge
 to cover local load. It must not intentionally export via the battery; export

go/cmd/ftw/energy_history_test.go

@@ -186,8 +186,14 @@ func TestPersistTelemetryTickUsesPersistenceFreshness(t *testing.T) {
 			if (len(history) == 1) != tc.wantSaved {
 				t.Fatalf("history = %+v, want saved=%v", history, tc.wantSaved)
 			}
-			if tc.wantSaved && (history[0].TsMs != tickMS || history[0].LoadW != 1200) {
-				t.Fatalf("tick timestamp or household power changed: %+v", history[0])
+			if tc.wantSaved && (history[0].TsMs != tickMS || history[0].GridW != 1200 || history[0].Missing&state.MissingLoad == 0) {
+				t.Fatal("fresh grid receipt lost or instantaneous Load entered history")
+			}
+			if tc.wantSaved {
+				p, _ := buildHistoryPoint(tel, ctrl, tickMS, time.Minute)
+				if p.Sources.AtMS < sampleAt.UnixMilli() || p.Sources.Missing != 0 {
+					t.Fatal("fresh source snapshot predates receipt or became unknown")
+				}
 			}
 			assets, err := st.EnergyAssets()
 			if err != nil {
@@ -199,8 +205,8 @@ func TestPersistTelemetryTickUsesPersistenceFreshness(t *testing.T) {
 					consumerSaved = asset.LastSeenMS == sampleAt.UnixMilli()
 				}
 			}
-			if consumerSaved != tc.wantSaved {
-				t.Fatalf("consumer ledger assets = %+v, want saved=%v", assets, tc.wantSaved)
+			if consumerSaved {
+				t.Fatal("instantaneous consumer observation entered interval history")
 			}
 		})
 	}
@@ -288,14 +294,14 @@ func TestStaleMeterTickKeepsSamplesAndIndependentLedgerWithoutDispatch(t *testin
 	tel.Update("solar", telemetry.DerPV, -400, nil, json.RawMessage(`{"generation_wh":110}`))
 	tel.Get("solar", telemetry.DerPV).UpdatedAt = now
 	tel.RecordDriverSuccess("solar")
-	if _, err := persistTelemetryTick(st, tel, ctrl, now.Add(time.Second).UnixMilli(), time.Minute, testEnergyIdentity(st)); err != nil {
+	if _, err := persistTelemetryTick(st, tel, ctrl, time.Now().UnixMilli(), time.Minute, testEnergyIdentity(st)); err != nil {
 		t.Fatal(err)
 	}
 
 	if err := st.FlushHistory(context.Background()); err != nil {
 		t.Fatal(err)
 	}
-	history, err := st.LoadHistory(base.UnixMilli(), now.Add(time.Second).UnixMilli(), 0)
+	history, err := st.LoadHistory(base.UnixMilli(), time.Now().Add(time.Second).UnixMilli(), 0)
 	if err != nil {
 		t.Fatal(err)
 	}

go/cmd/ftw/forecast_measurements.go

@@ -31,6 +31,9 @@ func historyMeasurementOptions(cfg *config.Config, catalog []drivers.CatalogEntr
 		}
 	}
 	opts.ExpectedFlows = flows
+	if cfg != nil {
+		opts.HistoryRevision = cfg.Revision
+	}
 	return opts
 }
 

go/cmd/ftw/history_disabled_battery_test.go

@@ -37,20 +37,22 @@ func TestHistoryDisabledBatteryKeepsP1Load(t *testing.T) {
 		tel.Update("solar", telemetry.DerPV, -2000, nil, nil)
 		tel.RecordDriverSuccess("solar")
 		opts := historyMeasurementOptions(cfg, catalog)
-		at := time.Now().Add(-2 * time.Second).UnixMilli()
-		for i := int64(0); i < 2; i++ {
+		at := time.Now().Add(-10 * time.Minute).Truncate(5 * time.Minute).UnixMilli()
+		for i := int64(0); i <= 30; i++ {
 			for _, kind := range []telemetry.DerType{telemetry.DerMeter, telemetry.DerPV} {
 				name := map[telemetry.DerType]string{telemetry.DerMeter: "p1", telemetry.DerPV: "solar"}[kind]
-				tel.Get(name, kind).UpdatedAt = time.UnixMilli(at + i*1000)
+				tel.Get(name, kind).UpdatedAt = time.Now()
 			}
-			if _, err := persistTelemetryTick(st, tel, tickPersistControl{SiteMeterDriver: "p1"}, at+i*1000, time.Minute, nil, opts); err != nil {
+			p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "p1"}, at+i*10000, time.Minute, opts)
+			p.Sources.AtMS = at + i*10000
+			if err := st.RecordTickWithEnergy(p, nil, nil); err != nil {
 				t.Fatal(err)
 			}
 		}
 		if err := st.FlushHistory(context.Background()); err != nil {
 			t.Fatal(err)
 		}
-		points, err := st.LoadHistory(at, at+1000, 0)
+		points, err := st.LoadHistory(at, at+300000, 5)
 		if err != nil {
 			t.Fatal(err)
 		}

go/cmd/ftw/history_source_load_test.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"context"
+	"math"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/state"
+	"github.com/srcfl/ftw/go/internal/telemetry"
+)
+
+func TestHistoryLoadIndependentReportingCadence(t *testing.T) {
+	st, err := state.Open(filepath.Join(t.TempDir(), "state.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	if err := st.EnableHistoryAggregation(); err != nil {
+		t.Fatal(err)
+	}
+	tel := telemetry.NewStore()
+	tel.Update("grid", telemetry.DerMeter, 4000, nil, nil)
+	tel.RecordDriverSuccess("grid")
+	tel.Update("charger", telemetry.DerEV, 3000, nil, nil)
+	tel.RecordDriverSuccess("charger")
+	// Both powers are fresh. The charger reports more slowly than the meter.
+	tel.Get("charger", telemetry.DerEV).UpdatedAt = time.Now().Add(-45 * time.Second)
+	base := time.Now().Add(-10 * time.Minute).Truncate(5 * time.Minute).UnixMilli()
+	for i := int64(0); i <= 30; i++ {
+		p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "grid"}, base+i*10000, 90*time.Second)
+		if p.Sources.Missing != 0 {
+			t.Fatal("fresh charger rejected by historical source qualification")
+		}
+		p.Sources.AtMS = base + i*10000
+		if err := st.RecordTickWithEnergy(p, nil, nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := st.FlushHistory(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	d, err := st.DailyEnergy(base, base+300000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if d.Missing&state.MissingLoad != 0 || math.Abs(d.LoadWh-1000.0/12) > 0.001 {
+		t.Fatal("fresh asynchronous sources lost historical household energy")
+	}
+	native, err := (&appHistory{st: st}).Window(context.Background(), []string{"load_w"}, 300000, base, base+300000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(native["load_w"]) != 1 || math.Abs(native["load_w"][0].W-1000) > 0.001 {
+		t.Fatal("native history did not receive the interval household balance")
+	}
+}
+
+func TestHistorySourceQualificationDoesNotInventMissingFlows(t *testing.T) {
+	for _, tc := range []struct {
+		reason string
+		watts  float64
+		known  bool
+		want   state.HistoryMissing
+	}{
+		{"missing_config", 0, true, state.MissingLoad},
+		{"invalid_balance", 1000, true, 0},
+		{"nonfinite_sum", math.Inf(1), false, state.MissingLoad | state.MissingEV},
+	} {
+		known := map[telemetry.DerType]bool{telemetry.DerMeter: true, telemetry.DerPV: true, telemetry.DerBattery: true, telemetry.DerEV: tc.known, telemetry.DerV2X: true}
+		p := historySources(telemetry.ForecastReading{At: time.Now(), Known: known, Reason: tc.reason, EVW: tc.watts})
+		if p.Missing != tc.want || math.IsNaN(p.EVW) || math.IsInf(p.EVW, 0) {
+			t.Fatal("history source qualification lost missing evidence or rejected unrelated telemetry")
+		}
+	}
+}

go/cmd/ftw/main.go

@@ -14,6 +14,7 @@ import (
 	"io"
 	"io/fs"
 	"log/slog"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
@@ -22,6 +23,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -3910,6 +3912,20 @@ func snapshotTickPersistControl(ctrl *control.State, ctrlMu *sync.Mutex) tickPer
 
 func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersistControl, nowMs int64, historyMaxAge time.Duration, identity energyIdentityLookup, options ...telemetry.ForecastOptions) (int, error) {
 	hp, historyAvailable := buildHistoryPoint(tel, ctrl, nowMs, historyMaxAge, options...)
+	if hp.Sources != nil {
+		// Live Load is not an energy observation. The source interval writer
+		// owns historical Load; the chart placeholder stays explicitly unknown.
+		hp.LoadW = 0
+		hp.Missing |= state.MissingLoad
+		var detail map[string]any
+		if json.Unmarshal([]byte(hp.JSON), &detail) == nil {
+			detail["load_house_w"] = nil
+			detail["forecast_measurement_quality"] = "source_interval_pending"
+			if raw, err := json.Marshal(detail); err == nil {
+				hp.JSON = string(raw)
+			}
+		}
+	}
 	samples := tel.FlushSamples()
 	stSamples := make([]state.Sample, len(samples))
 	for i, sm := range samples {
@@ -3919,9 +3935,19 @@ func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersis
 		}
 	}
 	energyObservations := buildEnergyObservations(st, tel, ctrl, hp, identity)
+	if hp.Sources != nil {
+		var devices []string
+		for _, o := range energyObservations {
+			if o.DeviceID != "" {
+				devices = append(devices, string(o.AssetKind)+":"+o.DeviceID)
+			}
+		}
+		sort.Strings(devices)
+		hp.Sources.Topology += "\ndevices:" + strings.Join(devices, "\n")
+	}
 	filtered := energyObservations[:0]
 	for _, observation := range energyObservations {
-		if (!historyAvailable || hp.Missing&state.MissingLoad != 0) && observation.AssetKind == state.AssetObservedConsumer {
+		if (!historyAvailable || hp.Sources != nil || hp.Missing&state.MissingLoad != 0) && observation.AssetKind == state.AssetObservedConsumer {
 			continue
 		}
 		if observation.AssetKind != state.AssetObservedConsumer {
@@ -4022,9 +4048,44 @@ func buildHistoryPoint(tel *telemetry.Store, ctrl tickPersistControl, nowMs int6
 	}
 	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)
+	liveSources := tel.LiveMeasurementNow(ctrl.SiteMeterDriver, opts)
+	p.Sources = historySources(liveSources)
+	p.Sources.Topology += "\nsettings:" + strconv.FormatInt(opts.HistoryRevision, 10)
+	for _, source := range liveSources.Sources {
+		hasPower = hasPower || source.PowerW != nil
+	}
 	return p, hasPower
 }
 
+func historySources(reading telemetry.ForecastReading) *state.HistorySources {
+	p := &state.HistorySources{AtMS: reading.At.UnixMilli(), GridW: reading.GridW, PVW: reading.PVW, BatteryW: reading.BatteryW, EVW: reading.EVW, V2XW: reading.V2XW}
+	var flows []string
+	for _, source := range reading.Sources {
+		flows = append(flows, source.Flow.Driver+":"+source.Flow.DerType.String()+":"+source.Flow.FlowID)
+	}
+	sort.Strings(flows)
+	p.Topology = strings.Join(flows, "\n")
+	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 !reading.Known[kind] {
+			p.Missing |= flag
+		}
+	}
+	// A negative instantaneous residual does not invalidate its independent
+	// sources. Missing configuration and duplicate electrical flows do.
+	if reading.Reason != "" && reading.Reason != "invalid_balance" {
+		p.Missing |= state.MissingLoad
+	}
+	for flag, w := range map[state.HistoryMissing]*float64{state.MissingGrid: &p.GridW, state.MissingPV: &p.PVW, state.MissingBattery: &p.BatteryW, state.MissingEV: &p.EVW, state.MissingV2X: &p.V2XW} {
+		if math.IsNaN(*w) || math.IsInf(*w, 0) {
+			p.Missing |= flag
+		}
+		if p.Missing&flag != 0 {
+			*w = 0
+		}
+	}
+	return p
+}
+
 func restoreLatestMPCDiagnostic(st *state.Store, svc *mpc.Service, now time.Time) {
 	if st == nil || svc == nil {
 		return

go/internal/api/api.go

@@ -2504,6 +2504,11 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) {
 	now := time.Now()
 	loc := now.Location()
 	todayMidnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc)
+	sourceThrough, sourceMode, err := s.deps.State.HistorySourceProgress(r.Context())
+	if err != nil {
+		http.Error(w, "history load failed", http.StatusInternalServerError)
+		return
+	}
 
 	// Walk day-by-day: past days hit the immutable-day cache, today always
 	// recomputes (in-progress). Each cache miss / today fire is one SQL
@@ -2513,9 +2518,13 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) {
 	for i := days - 1; i >= 0; i-- {
 		dayStart := todayMidnight.AddDate(0, 0, -i)
 		dayKey := dayStart.Format("2006-01-02")
+		cacheKey := dayKey
+		if sourceMode {
+			cacheKey += ":source_interval_balance_v1"
+		}
 		// A bounded balance episode can close just after midnight. Let its
 		// final intervals arrive before treating the previous day as immutable.
-		cacheReady := dailyEnergyCacheReady(dayStart, now)
+		cacheReady := dailyEnergyCacheReady(dayStart, now) && (!sourceMode || sourceThrough >= dayStart.AddDate(0, 0, 1).UnixMilli())
 
 		var de state.DayEnergy
 		if !cacheReady {
@@ -2534,14 +2543,14 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) {
 			// per-day DailyEnergy SQL pass; with the table populated
 			// the same query reduces to N PK lookups.
 			s.dailyCacheMu.Lock()
-			cached, ok := s.dailyCache[dayKey]
+			cached, ok := s.dailyCache[cacheKey]
 			s.dailyCacheMu.Unlock()
 			if ok {
 				de = cached
 			} else if persisted, present, err := s.deps.State.LoadDailyEnergy(dayKey); err == nil && present {
 				de = persisted
 				s.dailyCacheMu.Lock()
-				s.dailyCache[dayKey] = de
+				s.dailyCache[cacheKey] = de
 				s.dailyCacheMu.Unlock()
 			} else {
 				dayEnd := dayStart.AddDate(0, 0, 1)
@@ -2553,14 +2562,14 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) {
 				}
 				de = d
 				s.dailyCacheMu.Lock()
-				s.dailyCache[dayKey] = de
+				s.dailyCache[cacheKey] = de
 				s.dailyCacheMu.Unlock()
 				// Persist for next restart. Closed days only —
 				// today is excluded via the isToday branch above.
 				// Best-effort: a write failure is logged but not
 				// surfaced to the operator since the in-memory
 				// cache still serves this request.
-				if err := s.deps.State.SaveDailyEnergy(dayKey, de); err != nil {
+				if err := s.deps.State.SaveDailyEnergyForSourceMode(dayKey, de, sourceMode); err != nil {
 					slog.Warn("handleEnergyDaily: persist daily aggregate failed", "err", err, "day", dayKey)
 				}
 			}