From c65f65450acdb85bfcf41debacf46c34f8377037 Mon Sep 17 00:00:00 2001 From: Segran Date: Mon, 19 Oct 2026 17:04:04 +0200 Subject: [PATCH 1/5] fix(history): expose accepted household subtotal after gaps Signed-off-by: Segran --- .changeset/home-use-subtotal.md | 5 + go/internal/api/api.go | 12 +- go/internal/api/api_household_subtotal.go | 16 ++ .../api/api_household_subtotal_test.go | 103 +++++++++++ go/internal/state/household_subtotal.go | 62 +++++++ go/internal/state/household_subtotal_test.go | 170 ++++++++++++++++++ web/energy-today-subtotal.test.mjs | 47 +++++ web/energy-today.js | 18 +- 8 files changed, 429 insertions(+), 4 deletions(-) create mode 100644 .changeset/home-use-subtotal.md create mode 100644 go/internal/api/api_household_subtotal.go create mode 100644 go/internal/api/api_household_subtotal_test.go create mode 100644 go/internal/state/household_subtotal.go create mode 100644 go/internal/state/household_subtotal_test.go create mode 100644 web/energy-today-subtotal.test.mjs diff --git a/.changeset/home-use-subtotal.md b/.changeset/home-use-subtotal.md new file mode 100644 index 000000000..8fa55d0f2 --- /dev/null +++ b/.changeset/home-use-subtotal.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Keep an estimated Home use subtotal visible after missing source readings or a recording gap. Publish its accepted interval coverage separately from the unknown full total and measured energy. Preserve original history and exclude gaps rather than filling them. diff --git a/go/internal/api/api.go b/go/internal/api/api.go index d9e2a9b4b..015e864b5 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -1205,6 +1205,9 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { // instead of "no data yet". Mirror the old `len(pts) > 1` guard. if d.Intervals > 0 { energyToday = energyTodayFields(d) + if err := s.addHouseholdSubtotal(r.Context(), midnight.UnixMilli(), now.UnixMilli(), energyToday); err != nil { + slog.Warn("household energy subtotal unavailable", "err", err) + } } } else { slog.Warn("failed to integrate today's energy", "err", err) @@ -2603,7 +2606,7 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { http.Error(w, "history coverage unavailable", http.StatusInternalServerError) return } - out = append(out, map[string]any{ + dayEnergy := map[string]any{ "coverage_ms": coverage, "period_ms": periodEnd - dayStart.UnixMilli(), "history_quality": de.Missing.Quality(), @@ -2614,7 +2617,12 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { "bat_charged_wh": de.Missing.Value(state.MissingBattery, de.BatChargedWh), "bat_discharged_wh": de.Missing.Value(state.MissingBattery, de.BatDischargedWh), "load_wh": de.Missing.Value(state.MissingLoad, de.LoadWh), - }) + } + if err := s.addHouseholdSubtotal(r.Context(), dayStart.UnixMilli(), periodEnd, dayEnergy); err != nil { + http.Error(w, "household energy subtotal unavailable", http.StatusInternalServerError) + return + } + out = append(out, dayEnergy) } writeJSON(w, 200, map[string]any{"days": out, "tz": localzone.Name()}) } diff --git a/go/internal/api/api_household_subtotal.go b/go/internal/api/api_household_subtotal.go new file mode 100644 index 000000000..2681a66a5 --- /dev/null +++ b/go/internal/api/api_household_subtotal.go @@ -0,0 +1,16 @@ +package api + +import "context" + +// This separate field must never populate measured.load_wh, the complete +// load_wh total, forecast training or a self-powered percentage. +func (s *Server) addHouseholdSubtotal(ctx context.Context, from, to int64, fields map[string]any) error { + if fields == nil || fields["load_wh"] != nil { + return nil + } + subtotal, err := s.deps.State.HouseholdEnergySubtotal(ctx, from, to) + if err == nil && subtotal != nil { + fields["load_subtotal"] = subtotal + } + return err +} diff --git a/go/internal/api/api_household_subtotal_test.go b/go/internal/api/api_household_subtotal_test.go new file mode 100644 index 000000000..3ad90c5e2 --- /dev/null +++ b/go/internal/api/api_household_subtotal_test.go @@ -0,0 +1,103 @@ +package api + +import ( + "context" + "encoding/json" + "net/http/httptest" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/control" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +func TestHouseholdSubtotalStatusAndDailyKeepIncompleteTotalSeparate(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) + } + now := time.Now() + midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) + if now.Sub(midnight) < 16*time.Minute { + t.Skip("need fifteen completed minutes today") + } + for _, day := range []time.Time{midnight.AddDate(0, 0, -1), midnight} { + for i := int64(0); i <= 90; i++ { + at := day.Add(time.Duration(i) * 10 * time.Second).UnixMilli() + v := &state.HistorySources{AtMS: at, Topology: "site", GridW: 1000} + if i == 42 { + v.Missing = state.MissingPV + } + p := &state.HistoryPoint{TsMs: at, GridW: 1000, LoadW: 1000, JSON: `{}`, Sources: v} + if err := st.EnqueueTelemetryTick(p, nil, nil); err != nil { + t.Fatal(err) + } + if i%10 == 0 { + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + } + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + srv := New(&Deps{State: st, Tel: telemetry.NewStore(), Ctrl: &control.State{}, CtrlMu: &sync.Mutex{}, + CapMu: &sync.RWMutex{}, CfgMu: &sync.RWMutex{}, Cfg: &config.Config{}}) + check := func(fields map[string]any) { + t.Helper() + if fields["load_wh"] != nil || fields["history_quality"] != "incomplete" { + t.Fatal("partial data claimed as a complete day", fields) + } + sub, ok := fields["load_subtotal"].(map[string]any) + if !ok || sub["quality"] != "estimated" || sub["coverage_ms"] != float64(600000) { + t.Fatal("accepted subtotal or its coverage missing", fields) + } + if wh, ok := sub["wh"].(float64); !ok || wh < 166.66 || wh > 166.67 { + t.Fatal(sub) + } + } + rr := httptest.NewRecorder() + srv.handleStatus(rr, httptest.NewRequest("GET", "/api/status", nil)) + var status struct { + Energy struct { + Today map[string]any `json:"today"` + } `json:"energy"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &status); err != nil { + t.Fatal(err) + } + check(status.Energy.Today) + if status.Energy.Today["measured"].(map[string]any)["load_wh"] != nil { + t.Fatal("estimated subtotal leaked into measured energy") + } + // Repeat the read: yesterday takes the persistent/in-memory cache path, + // while its independently queried subtotal must retain the same coverage. + for i := 0; i < 2; i++ { + rr = httptest.NewRecorder() + srv.handleEnergyDaily(rr, httptest.NewRequest("GET", "/api/energy/daily?days=2", nil)) + if rr.Code != 200 { + t.Fatal(rr.Code, rr.Body.String()) + } + var daily struct { + Days []map[string]any `json:"days"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &daily); err != nil { + t.Fatal(err) + } + if len(daily.Days) != 2 { + t.Fatal(daily) + } + for _, fields := range daily.Days { + check(fields) + } + } +} diff --git a/go/internal/state/household_subtotal.go b/go/internal/state/household_subtotal.go new file mode 100644 index 000000000..436fca92a --- /dev/null +++ b/go/internal/state/household_subtotal.go @@ -0,0 +1,62 @@ +package state + +import "context" + +// HouseholdSubtotal describes accepted, complete energy intervals only. It is +// not a complete range total, an independent meter, or a gap-filled estimate. +type HouseholdSubtotal struct { + Wh float64 `json:"wh"` + CoverageMS int64 `json:"coverage_ms"` + PeriodMS int64 `json:"period_ms"` + // Minimum interval width; older retained evidence can be coarser. + BucketMS int64 `json:"minimum_bucket_ms"` + Quality string `json:"quality"` + Method string `json:"method"` +} + +// HouseholdEnergySubtotal reads the same independent source energy and legacy +// site intervals as DailyEnergy, without changing their quality or caches. +// Balance signed energy before deciding whether an interval is usable; never +// clip negative samples, interpolate missing sources or bridge recording gaps. +func (s *Store) HouseholdEnergySubtotal(ctx context.Context, from, to int64) (*HouseholdSubtotal, error) { + if to <= from || s.history == nil { + return nil, nil + } + const width int64 = 5 * 60000 + buckets := map[int64]*sourceLoadBucket{} + err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { + // Retained coarse intervals remain indivisible; never split an older + // observed interval into invented five-minute measurements. + w := max(width, r.width) + start := bucketStart(r.start, w) + b := buckets[start] + if b == nil { + b = &sourceLoadBucket{start: start, width: w, first: r.first, last: r.last} + buckets[start] = b + } + if b.width != w || r.first < start || r.last > start+w || start < from || start+w > to { + b.missing |= MissingLoad + } + b.first = min(b.first, r.first) + b.last = max(b.last, r.last) + b.covered += r.covered + b.wh += r.load + b.missing |= r.missing + return nil + }) + if err != nil { + return nil, err + } + out := HouseholdSubtotal{PeriodMS: to - from, BucketMS: width, Quality: "estimated", Method: "accepted_interval_balance_v1"} + for _, b := range buckets { + if b.missing&MissingPower != 0 || b.covered != b.width || b.first != b.start || b.last != b.start+b.width || b.wh < 0 || !finiteHistoryValue(b.wh) { + continue + } + out.Wh += b.wh + out.CoverageMS += b.covered + } + if out.CoverageMS == 0 { + return nil, nil + } + return &out, nil +} diff --git a/go/internal/state/household_subtotal_test.go b/go/internal/state/household_subtotal_test.go new file mode 100644 index 000000000..da7057363 --- /dev/null +++ b/go/internal/state/household_subtotal_test.go @@ -0,0 +1,170 @@ +package state + +import ( + "context" + "fmt" + "reflect" + "testing" + "time" +) + +func TestHouseholdSubtotalKeepsAcceptedIntervalsAcrossSourceAndWriterGaps(t *testing.T) { + for _, kind := range []string{"PV missing", "EV missing", "writer gap", "topology"} { + t.Run(kind, func(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i <= 90; i++ { + if kind == "writer gap" && i == 42 { + continue + } + v := HistorySources{GridW: 1200, EVW: 200} + if kind == "topology" && i >= 42 { + v.Topology = "replacement meter" + } + if i == 42 && kind == "PV missing" { + v.Missing = MissingPV + } + if i == 42 && kind == "EV missing" { + v.Missing = MissingEV + } + sourceTick(t, s, base+i*10000, v) + } + before, err := s.DailyEnergy(base, base+900000) + if err != nil || before.Missing&MissingLoad == 0 { + t.Fatal("fixture must have an incomplete full total", before, err) + } + check := func(s *Store) { + t.Helper() + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+900000) + if err != nil || got == nil { + t.Fatal("accepted intervals hidden", got, err) + } + near(t, "accepted Wh", got.Wh, 1000.0/6) + if got.CoverageMS != 600000 || got.PeriodMS != 900000 || got.Quality != "estimated" { + t.Fatal("gap became full coverage or measured energy", got) + } + after, err := s.DailyEnergy(base, base+900000) + if err != nil || !reflect.DeepEqual(before, after) { + t.Fatal("subtotal rewrote full total or original quality", before, after, err) + } + } + check(s) + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + check(reopened) + }) + } +} + +func TestHouseholdSubtotalCombinesLegacyAndSourceWithoutDoubleCounting(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i <= 30; i++ { + at := base + i*10000 + p := &HistoryPoint{TsMs: at, GridW: 1000, LoadW: 1000, JSON: `{}`} + if i == 30 { + p.Sources = &HistorySources{AtMS: at, Topology: "grid+charger", GridW: 3000, EVW: 2000} + } + dashboardTick(t, s, fmt.Sprint(at), p) + } + sourceSeries(t, s, base+300000, []float64{1000, 1000, 1000, 1000, 1000}) + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+600000) + if err != nil || got == nil || got.CoverageMS != 600000 { + t.Fatal(got, err) + } + near(t, "mixed subtotal", got.Wh, 1000.0/6) +} + +func TestHouseholdSubtotalRetainsCoarseEvidenceWithoutSplittingItsBounds(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + minutes := make([]float64, 15) + for i := range minutes { + minutes[i] = 1000 + } + sourceSeries(t, s, base, minutes) + if err := s.rollSiteEnergy(context.Background(), base+900000); err != nil { + t.Fatal(err) + } + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+900000) + if err != nil || got == nil || got.CoverageMS != 900000 { + t.Fatal(got, err) + } + near(t, "retained subtotal", got.Wh, 250) + if got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+600000); err != nil || got != nil { + t.Fatal("coarse interval was split into invented observations", got, err) + } +} + +func TestHouseholdSubtotalPreservesSignedBalanceAndDistinguishesZeroFromUnknown(t *testing.T) { + for _, tc := range []struct { + name string + minutes []float64 + wantWh float64 + wantCoverage int64 + }{ + {"signed balance", []float64{-1000, 3000, 1000, 1000, 1000}, 1000.0 / 12, 300000}, + {"zero", []float64{0, 0, 0, 0, 0}, 0, 300000}, + {"negative interval", []float64{-1000, -1000, -1000, -1000, -1000}, 0, 0}, + {"partial tail", []float64{1000, 1000, 1000, 1000}, 0, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + sourceSeries(t, s, base, tc.minutes) + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+int64(len(tc.minutes))*60000) + if err != nil { + t.Fatal(err) + } + if tc.wantCoverage == 0 { + if got != nil { + t.Fatal("unknown became a known zero", got) + } + return + } + if got == nil || got.CoverageMS != tc.wantCoverage { + t.Fatal(got) + } + near(t, "signed subtotal", got.Wh, tc.wantWh) + }) + } +} + +func TestHouseholdSubtotalNeverBridgesRestartOrClaimsAbsentEvidence(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + if got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+900000); err != nil || got != nil { + t.Fatal("empty history became zero", got, err) + } + sourceSeries(t, s, base, []float64{1000, 1000, 1000, 1000, 1000}) + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + if err := reopened.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + sourceSeries(t, reopened, base+600000, []float64{1000, 1000, 1000, 1000, 1000}) + got, err := reopened.HouseholdEnergySubtotal(context.Background(), base, base+900000) + if err != nil || got == nil || got.CoverageMS != 600000 { + t.Fatal(got, err) + } + near(t, "restart subtotal", got.Wh, 1000.0/6) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := reopened.HouseholdEnergySubtotal(ctx, base, base+900000); err == nil { + t.Fatal("cancelled diagnostic read succeeded") + } +} diff --git a/web/energy-today-subtotal.test.mjs b/web/energy-today-subtotal.test.mjs new file mode 100644 index 000000000..f26891f29 --- /dev/null +++ b/web/energy-today-subtotal.test.mjs @@ -0,0 +1,47 @@ +import assert from "node:assert/strict"; +import {test} from "node:test"; +import {energyTodayReading, formatEnergyToday, updateEnergyToday} from "./energy-today.js"; + +const subtotal = {wh: 26603, coverage_ms: 48900000, period_ms: 50400000, + minimum_bucket_ms: 300000, quality: "estimated", method: "accepted_interval_balance_v1"}; +const incomplete = {load_wh: null, history_quality: "incomplete", + measured: {load_wh: null}, load_subtotal: subtotal}; + +test("a source-integrated partial day displays its own estimated subtotal, never a complete total", () => { + assert.equal(formatEnergyToday(incomplete, "load_wh"), "≈ 26.6 kWh (partial)"); + assert.equal(incomplete.load_wh, null); + assert.equal(incomplete.measured.load_wh, null); + assert.equal(formatEnergyToday({...incomplete, load_wh: 30000, history_quality: "estimated"}, "load_wh"), "≈ 30.0 kWh"); + assert.equal(formatEnergyToday({...incomplete, load_wh: 0}, "load_wh"), "0.00 kWh"); + assert.equal(formatEnergyToday({...incomplete, load_subtotal: {...subtotal, wh: 0}}, "load_wh"), "≈ 0.00 kWh (partial)"); + assert.equal(formatEnergyToday(incomplete, "pv_wh"), "Unknown"); +}); + +test("missing or malformed interval evidence cannot manufacture a partial zero", () => { + for (const invalid of [undefined, null, {...subtotal, wh: NaN}, {...subtotal, wh: -1}, + {...subtotal, coverage_ms: 0}, {...subtotal, coverage_ms: subtotal.period_ms + 1}, + {...subtotal, period_ms: "50400000"}, {...subtotal, quality: "measured"}, + {...subtotal, method: "interpolated"}]) { + assert.equal(energyTodayReading({...incomplete, load_subtotal: invalid}, "load_wh").wh, null); + } +}); + +test("cards explain accepted coverage and clear the partial state after recovery or missing data", () => { + const elements = new Map(); + const group = {append(el) { elements.set(el.id, el); }}; + for (const id of ["e-load", "e-pv"]) elements.set(id, {closest() { return group; }}); + const root = {getElementById(id) { return elements.get(id); }, createElement() { return {}; }}; + updateEnergyToday(incomplete, root); + assert.equal(elements.get("e-load").textContent, "≈ 26.6 kWh (partial)"); + assert.match(elements.get("e-load").title, /gaps are excluded/); + assert.match(elements.get("energy-today-quality").textContent, /815 of 840 min/); + const complete = {load_wh: 30000, pv_wh: 10000, import_wh: 10000, export_wh: 0, + bat_charged_wh: 0, bat_discharged_wh: 0}; + updateEnergyToday(complete, root); + assert.equal(elements.get("e-load").textContent, "30.0 kWh"); + assert.equal(elements.get("e-load").title, ""); + assert.equal(elements.get("energy-today-quality").hidden, true); + updateEnergyToday(null, root); + assert.equal(elements.get("e-load").textContent, "Unknown"); + assert.equal(elements.get("energy-today-quality").textContent, "Today's energy is not available yet."); +}); diff --git a/web/energy-today.js b/web/energy-today.js index 4b37e2ed0..88302c8f8 100644 --- a/web/energy-today.js +++ b/web/energy-today.js @@ -20,6 +20,14 @@ export function energyTodayReading(today, key) { ? {wh: total, partial: false, estimated: true} : {wh: total, partial: false}; const measured = today?.measured?.[key]; if (knownWh(measured)) return {wh: measured, partial: true}; + const subtotal = today?.load_subtotal; + if (key === "load_wh" && knownWh(subtotal?.wh) && subtotal.quality === "estimated" && + subtotal.method === "accepted_interval_balance_v1" && + Number.isFinite(subtotal.coverage_ms) && Number.isFinite(subtotal.period_ms) && + subtotal.coverage_ms > 0 && subtotal.coverage_ms <= subtotal.period_ms) { + return {wh: subtotal.wh, partial: true, estimated: true, + coverageMs: subtotal.coverage_ms, periodMs: subtotal.period_ms}; + } return {wh: null, partial: false}; } @@ -28,7 +36,7 @@ export function formatEnergyToday(today, key, withUnit = true) { if (wh === null) return "Unknown"; const kwh = wh / 1000; const digits = kwh >= 100 ? 0 : kwh >= 10 ? 1 : 2; - return `${estimated ? "≈ " : partial ? "≥ " : ""}${kwh.toFixed(digits)}${withUnit ? " kWh" : ""}`; + return `${estimated ? "≈ " : partial ? "≥ " : ""}${kwh.toFixed(digits)}${withUnit ? " kWh" : ""}${estimated && partial ? " (partial)" : ""}`; } export function updateEnergyToday(today, root = document) { @@ -37,7 +45,8 @@ export function updateEnergyToday(today, root = document) { if (!el) continue; const {partial, estimated} = energyTodayReading(today, key); el.textContent = formatEnergyToday(today, key); - el.title = estimated ? "Estimated from source energy during brief balance failures." + el.title = estimated && partial ? "Estimated subtotal from accepted energy intervals only; gaps are excluded." + : estimated ? "Estimated from source energy during brief balance failures." : partial ? "Recorded energy only; missing readings are excluded." : ""; } const group = root.getElementById("e-pv")?.closest(".energy-group"); @@ -57,4 +66,9 @@ export function updateEnergyToday(today, root = document) { ? "Gaps in today's readings. ≥ shows recorded energy only." : "Today's energy is not available yet."; if (estimated) note.textContent = (incomplete ? note.textContent + " " : "") + "Home use includes estimates for brief balance failures."; + const home = energyTodayReading(today, "load_wh"); + if (home.estimated && home.partial) { + note.hidden = false; + note.textContent += ` Home use: estimated subtotal for ${Math.floor(home.coverageMs / 60000)} of ${Math.ceil(home.periodMs / 60000)} min; gaps excluded.`; + } } From 6f71db11534ce2115fdc522c78a45f37b9fc1e9f Mon Sep 17 00:00:00 2001 From: Segran Date: Mon, 19 Oct 2026 17:55:38 +0200 Subject: [PATCH 2/5] fix(history): reconstruct missing historical household energy on audited copies Signed-off-by: Segran --- .changeset/home-use-subtotal.md | 2 + docs/history-repair.md | 110 ++++ go/cmd/ftw-history-repair/main.go | 92 ++++ go/cmd/ftw-history-repair/main_test.go | 94 ++++ go/internal/api/api.go | 6 + go/internal/api/api_energy_today.go | 6 + go/internal/api/api_history_repair_test.go | 19 + go/internal/state/energy_ledger.go | 13 + .../state/history_dashboard_retention.go | 6 +- go/internal/state/history_repair.go | 470 ++++++++++++++++++ go/internal/state/history_repair_read.go | 233 +++++++++ go/internal/state/history_repair_test.go | 350 +++++++++++++ go/internal/state/household_subtotal.go | 6 +- go/internal/state/store.go | 21 +- web/components/energy-flow-readings.js | 2 +- web/energy-today.js | 8 +- web/history-repair-quality.test.mjs | 12 + 17 files changed, 1442 insertions(+), 8 deletions(-) create mode 100644 docs/history-repair.md create mode 100644 go/cmd/ftw-history-repair/main.go create mode 100644 go/cmd/ftw-history-repair/main_test.go create mode 100644 go/internal/api/api_history_repair_test.go create mode 100644 go/internal/state/history_repair.go create mode 100644 go/internal/state/history_repair_read.go create mode 100644 go/internal/state/history_repair_test.go create mode 100644 web/history-repair-quality.test.mjs diff --git a/.changeset/home-use-subtotal.md b/.changeset/home-use-subtotal.md index 8fa55d0f2..bee13acec 100644 --- a/.changeset/home-use-subtotal.md +++ b/.changeset/home-use-subtotal.md @@ -3,3 +3,5 @@ --- Keep an estimated Home use subtotal visible after missing source readings or a recording gap. Publish its accepted interval coverage separately from the unknown full total and measured energy. Preserve original history and exclude gaps rather than filling them. + +Add an offline historical repair tool that snapshots the original databases, prepares a hashed plan from selected retained AC power series, and applies audited estimates only to a working copy. Core uses these corrections in household daily totals and history without changing source measurements or their quality. Unsupported gaps remain unknown. diff --git a/docs/history-repair.md b/docs/history-repair.md new file mode 100644 index 000000000..f5286a7e3 --- /dev/null +++ b/docs/history-repair.md @@ -0,0 +1,110 @@ +# Repair historical household energy + +`ftw-history-repair` reconstructs missing household energy from retained, +explicitly selected power series. It writes durable, audited corrections to +`history_load_estimates` on a **separate working copy**, not the source +databases. Core then reads the corrections in daily household totals, chart +aggregates and the household energy ledger. Original samples, other energy +flows, quality masks, forecasts and control remain unchanged. + +These are estimates, not exact recovery of lost measurements. A correction +does not make a missing PV measurement known. The API marks household energy +with `load_quality: estimated`, independently of other missing flows. It keeps +derived energy out of `measured.load_wh` and self-powered percentages. A live +day corrected this way reports `load_until_ms` at its last completed minute. + +## Prepare a copy and inspect its plan + +Build the tool from the same patched source as Core: + +```sh +cd go +go build -o ../bin/ftw-history-repair ./cmd/ftw-history-repair +``` + +For a deployment, stop Core using the installation's actual service manager +before taking the snapshot, and keep it stopped until the corrected history +is installed or the operation is abandoned. A copy prepared while Core keeps +collecting data can be used for analysis, but must not later replace the live +history: that would discard newer measurements. Keep a verified full FTW +backup and the original release for deployment rollback. This tool additionally +preserves checked copies of the affected state/history databases. + +Select a completed range, at most 48 hours, with explicit time zones and +minute boundaries. Select every electrical flow or explicitly declare it +absent. Use site-side **AC** battery power; do not combine cabinet DC energy +with AC grid/PV readings. Check the running box's metric catalogue and physical +topology before using these example driver/metric names: + +```sh +bin/ftw-history-repair prepare \ + --state /var/lib/ftw/state.db \ + --workdir ./repair-2026-10-19 \ + --since 2026-10-19T00:00:00+02:00 \ + --until 2026-10-19T14:00:00+02:00 \ + --grid pixii:meter_w --pv sma:pv_w \ + --battery pixii:battery_ac_w --ev easee-cloud:ev_w --v2x none \ + --max-gap 15s --reason 'Reconstruct missing Home use intervals, issue 1750' +``` + +The work directory must be new. `backup/` contains state/history snapshots; +`corrected/` starts as a separate copy. `plan.json` records their hashes, +selected sources, period, reason, per-minute Wh, interpolation evidence, +unsupported minutes and the exact plan SHA-256. Preparing does not apply +corrections. The originals are opened read-only; both copies are checked +through Core's database integrity/generation checks. + +The reconstruction uses retained power means at minute resolution or finer. +It integrates between their observed time centres over the same minute for +every flow, using `grid - PV - battery - EV - V2X`. It does not extrapolate +edges, split coarse hourly evidence, clip signed residuals, or replace an +existing estimate. It refuses duplicate source mappings and non-watt inputs. + +The default maximum point gap is 15 seconds. Unsupported minutes remain +unknown. If the owner explicitly accepts bounded interpolation, prepare a +**new** plan with a larger bound, at most 180 seconds, and inspect the changed +minutes and `max_interpolated_ms_per_source` values. This is an estimate of +allocation, not proof of constant power or an instrument's delay. Signed +minute residuals are retained; each displayed aggregate must still have +complete supported coverage and a nonnegative resulting balance. + +## Apply the reviewed plan to the copy + +After reviewing the actual plan and its source selections, supply its exact +hash: + +```sh +bin/ftw-history-repair apply --workdir ./repair-2026-10-19 \ + --sha256 REPLACE_WITH_THE_REVIEWED_PLAN_SHA256 +``` + +The operation verifies the backup hashes and unchanged working inputs, +inserts all correction minutes and clears derived daily caches in one SQLite +transaction. A conflict rolls the transaction back. Retrying the same applied +plan does not add energy again. `apply-result.json` reports before/after +household Wh and quality for the **selected range**, not an assumed full day. +If another gap remains unsupported, the complete total stays unknown and the +accepted subtotal remains available. + +The correction rows carry the plan hash, backup hash, timestamp, reason and +source mapping in their method metadata. They survive portable backup/restore +and Core restart. Minute energy needed to interpret a corrected quarter is +retained rather than merged away by energy rollup. Original measurement +coverage remains separate from reconstructed household coverage. + +## Validate and deploy separately + +Use the patched Core to read `corrected/state.db` and `corrected/history.db` +in an isolated test setup without equipment control. Check the selected +period's daily total, ledger provenance, estimate labels and remaining gaps. +The tool never starts a second Core and has no install or restart action. + +For the owner's stopped installation, an operator must verify the generation, +release layout, file ownership, backup and absence of newer source data before +installing the corrected history and patched Core together. The same source +state configuration must still bind that history generation. Do not replace +the owner's configuration with a stale test copy. To abandon or undo the +repair before deployment, discard the working copy; the source is unchanged. +For deployment rollback, restore the verified original history snapshot and +matching original release while Core is stopped. Never copy SQLite files +over a running Core or leave mismatched WAL/SHM files beside a restored DB. diff --git a/go/cmd/ftw-history-repair/main.go b/go/cmd/ftw-history-repair/main.go new file mode 100644 index 000000000..a07cd20e5 --- /dev/null +++ b/go/cmd/ftw-history-repair/main.go @@ -0,0 +1,92 @@ +// ftw-history-repair prepares and applies historical household corrections on +// a separate database copy. It never writes the owner's source databases. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "strings" + "time" + + "github.com/srcfl/ftw/go/internal/state" +) + +func run(ctx context.Context, args []string, out io.Writer) error { + if len(args) == 0 { + return errors.New("use prepare --state PATH --workdir NEW_DIR --since RFC3339 --until RFC3339 --grid DRIVER:METRIC --pv DRIVER:METRIC|none --battery DRIVER:AC_METRIC|none --ev DRIVER:METRIC|none --v2x DRIVER:METRIC|none --reason TEXT [--max-gap 15s], then apply --workdir DIR --sha256 REVIEWED_PLAN_HASH") + } + f := flag.NewFlagSet(args[0], flag.ContinueOnError) + workdir := f.String("workdir", "", "new private working directory for prepare; same directory for apply") + if args[0] == "apply" { + sha := f.String("sha256", "", "SHA-256 of the exact reviewed plan") + if err := f.Parse(args[1:]); err != nil { + return err + } + if *workdir == "" || f.NArg() != 0 { + return errors.New("workdir is required; no positional arguments") + } + result, err := state.ApplyHistoryRepair(ctx, *workdir, *sha) + if err != nil { + return err + } + return json.NewEncoder(out).Encode(result) + } + if args[0] != "prepare" { + return errors.New("unknown repair operation") + } + source := f.String("state", "", "source state.db; opened read-only") + since := f.String("since", "", "start RFC3339 with timezone, on a minute boundary") + until := f.String("until", "", "end RFC3339 with timezone, on a completed minute boundary") + reason := f.String("reason", "", "reason for this historical estimate") + gap := f.Duration("max-gap", 15*time.Second, "largest source-point gap to interpolate; up to 180s, explicitly reviewed") + flows := map[string]*string{} + for _, name := range []string{"grid", "pv", "battery", "ev", "v2x"} { + flows[name] = f.String(name, "", "explicit driver:power_metric or none; battery must be site-side AC") + } + if err := f.Parse(args[1:]); err != nil { + return err + } + if *source == "" || *workdir == "" || f.NArg() != 0 { + return errors.New("state and workdir required; no positional arguments") + } + start, err := time.Parse(time.RFC3339, *since) + if err != nil { + return err + } + end, err := time.Parse(time.RFC3339, *until) + if err != nil { + return err + } + r := state.HistoryRepairRequest{SinceMS: start.UnixMilli(), UntilMS: end.UnixMilli(), MaxGapMS: gap.Milliseconds(), Reason: *reason} + for name, value := range flows { + v := state.RepairPowerSource{Flow: name} + if *value == "none" { + v.Absent = true + } else { + parts := strings.SplitN(*value, ":", 2) + if len(parts) != 2 { + return fmt.Errorf("%s needs driver:power_metric or explicit none", name) + } + v.Driver, v.Metric = parts[0], parts[1] + } + r.Sources = append(r.Sources, v) + } + state.SortRepairSources(r.Sources) + plan, err := state.PrepareHistoryRepair(ctx, *source, *workdir, r) + if err != nil { + return err + } + return json.NewEncoder(out).Encode(plan) +} + +func main() { + if err := run(context.Background(), os.Args[1:], os.Stdout); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} diff --git a/go/cmd/ftw-history-repair/main_test.go b/go/cmd/ftw-history-repair/main_test.go new file mode 100644 index 000000000..0b0b297af --- /dev/null +++ b/go/cmd/ftw-history-repair/main_test.go @@ -0,0 +1,94 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "github.com/srcfl/ftw/go/internal/state" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestRepairCommandRequiresExplicitFlowsAndReviewedPlan(t *testing.T) { + for _, args := range [][]string{nil, {"write"}, {"apply", "--workdir", "/not-a-repair"}, {"prepare", "--state", "state.db", "--workdir", "new", "--since", "2026-10-19T00:00:00+02:00", "--until", "2026-10-19T14:00:00+02:00", "--grid", "meter:meter_w"}} { + var out bytes.Buffer + if err := run(context.Background(), args, &out); err == nil { + t.Fatal("incomplete repair command succeeded", args) + } + } + var out bytes.Buffer + err := run(context.Background(), []string{"prepare", "--state", "state.db", "--workdir", "new", "--since", "2026-10-19T00:00:00", "--until", "2026-10-19T14:00:00+02:00"}, &out) + if err == nil || !strings.Contains(err.Error(), "cannot parse") { + t.Fatal("timezone-less date accepted", err) + } +} + +func TestRepairCommandPrepareThenApplyChangesHistoricalCopy(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + path := filepath.Join(dir, "state.db") + s, err := state.Open(path) + if err != nil { + t.Fatal(err) + } + if err := s.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + base := time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour) + for i := int64(0); i <= 90; i++ { + at := base.Add(time.Duration(i) * 10 * time.Second).UnixMilli() + v := &state.HistorySources{AtMS: at, Topology: "meter", GridW: 1000} + if i == 42 { + v.Missing = state.MissingPV + } + p := &state.HistoryPoint{TsMs: at, GridW: 1000, LoadW: 1000, JSON: `{}`, Sources: v} + if err := s.EnqueueTelemetryTick(p, nil, nil); err != nil { + t.Fatal(err) + } + if i%10 == 0 { + if err := s.FlushHistory(ctx); err != nil { + t.Fatal(err) + } + } + } + if err := s.FlushHistory(ctx); err != nil { + t.Fatal(err) + } + var samples []state.Sample + for i := int64(-6); i <= 186; i++ { + samples = append(samples, state.Sample{Driver: "meter", Metric: "meter_w", TsMs: base.Add(time.Duration(i) * 5 * time.Second).UnixMilli(), Value: 1000, Unit: "W"}) + } + if err := s.RecordSamples(samples); err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + work := filepath.Join(dir, "repair") + var out bytes.Buffer + args := []string{"prepare", "--state", path, "--workdir", work, "--since", base.Format(time.RFC3339), "--until", base.Add(15 * time.Minute).Format(time.RFC3339), "--grid", "meter:meter_w", "--pv", "none", "--battery", "none", "--ev", "none", "--v2x", "none", "--reason", "CLI repair regression"} + if err := run(ctx, args, &out); err != nil { + t.Fatal(err) + } + var plan state.HistoryRepairPlan + if err := json.Unmarshal(out.Bytes(), &plan); err != nil { + t.Fatal(err) + } + out.Reset() + if err := run(ctx, []string{"apply", "--workdir", work, "--sha256", plan.SHA256}, &out); err != nil { + t.Fatal(err) + } + var result state.HistoryRepairResult + if err := json.Unmarshal(out.Bytes(), &result); err != nil { + t.Fatal(err) + } + if result.BeforeWh != nil || result.AfterWh == nil || !result.Estimated { + t.Fatal("CLI did not repair history", result) + } + if wh := result.AfterWh.(float64); wh < 249.99 || wh > 250.01 { + t.Fatal("wrong corrected total", wh) + } + t.Logf("REPAIR_RESULT_JSON=%s", strings.TrimSpace(out.String())) +} diff --git a/go/internal/api/api.go b/go/internal/api/api.go index 015e864b5..19593f404 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -2618,6 +2618,12 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { "bat_discharged_wh": de.Missing.Value(state.MissingBattery, de.BatDischargedWh), "load_wh": de.Missing.Value(state.MissingLoad, de.LoadWh), } + if de.Missing&state.EstimatedLoad != 0 { + dayEnergy["load_quality"] = "estimated" + } + if de.LoadUntilMS > 0 { + dayEnergy["load_until_ms"] = de.LoadUntilMS + } if err := s.addHouseholdSubtotal(r.Context(), dayStart.UnixMilli(), periodEnd, dayEnergy); err != nil { http.Error(w, "household energy subtotal unavailable", http.StatusInternalServerError) return diff --git a/go/internal/api/api_energy_today.go b/go/internal/api/api_energy_today.go index de8e5cc58..34625440e 100644 --- a/go/internal/api/api_energy_today.go +++ b/go/internal/api/api_energy_today.go @@ -7,6 +7,12 @@ func energyTodayFields(d state.DayEnergy) map[string]any { return nil } out := map[string]any{"history_quality": d.Missing.Quality()} + if d.Missing&state.EstimatedLoad != 0 { + out["load_quality"] = "estimated" + } + if d.LoadUntilMS > 0 { + out["load_until_ms"] = d.LoadUntilMS + } measured := make(map[string]any) for key, field := range map[string]struct { flag state.HistoryMissing diff --git a/go/internal/api/api_history_repair_test.go b/go/internal/api/api_history_repair_test.go new file mode 100644 index 000000000..311418c1d --- /dev/null +++ b/go/internal/api/api_history_repair_test.go @@ -0,0 +1,19 @@ +package api + +import ( + "github.com/srcfl/ftw/go/internal/state" + "testing" +) + +func TestHistoricalRepairEnergyFieldsPreserveOtherUnknownFlows(t *testing.T) { + got := energyTodayFields(state.DayEnergy{Missing: state.MissingPV | state.EstimatedLoad, Intervals: 10, LoadWh: 500, LoadUntilMS: 1800000}) + if got["load_wh"] != float64(500) || got["load_quality"] != "estimated" || got["load_until_ms"] != int64(1800000) { + t.Fatal(got) + } + if got["pv_wh"] != nil || got["history_quality"] != "incomplete" { + t.Fatal("repair relabelled missing PV as known", got) + } + if got["measured"].(map[string]any)["load_wh"] != nil { + t.Fatal("repair became a measurement", got) + } +} diff --git a/go/internal/state/energy_ledger.go b/go/internal/state/energy_ledger.go index 4b4f9c5cf..490f1203b 100644 --- a/go/internal/state/energy_ledger.go +++ b/go/internal/state/energy_ledger.go @@ -501,6 +501,19 @@ func (s *Store) LoadEnergyHistoryContext(ctx context.Context, q EnergyHistoryQue if err != nil { return nil, false, err } + out, err = s.applyHistoricalLoadLedger(ctx, q, out) + if err != nil { + return nil, false, err + } + rank := 0 + for i, p := range out { + if i == 0 || p.BucketStartMS != out[i-1].BucketStartMS { + rank++ + } + if rank > q.Limit { + return out[:i], true, nil + } + } return out, truncated || sourceTruncated, nil } diff --git a/go/internal/state/history_dashboard_retention.go b/go/internal/state/history_dashboard_retention.go index 2cf1f7f59..7e535435e 100644 --- a/go/internal/state/history_dashboard_retention.go +++ b/go/internal/state/history_dashboard_retention.go @@ -194,7 +194,11 @@ func (s *Store) rollSiteEnergyStorage(ctx context.Context, cutoff int64, table, cutoff = bucketStart(cutoff, width) for { var first sql.NullInt64 - if err := s.history.QueryRowContext(ctx, `SELECT MIN(start_ms) FROM `+table+` WHERE resolution_ms=? AND start_ms=` + table + `.start_ms-(` + table + `.start_ms%900000) AND r.start_ms<` + table + `.start_ms-(` + table + `.start_ms%900000)+900000)` + if err := s.history.QueryRowContext(ctx, `SELECT MIN(start_ms) FROM `+table+` WHERE resolution_ms=? AND start_ms 48*3600000 || r.UntilMS > time.Now().Add(-time.Minute).UnixMilli() { + return errors.New("choose completed minute boundaries, at most 48 hours, ending at least one minute ago") + } + if r.MaxGapMS < 1000 || r.MaxGapMS > 180000 || strings.TrimSpace(r.Reason) == "" { + return errors.New("reason and max-gap between 1 and 180 seconds are required") + } + seen := map[string]bool{} + electrical := map[string]bool{} + for _, v := range r.Sources { + if v.Flow != "grid" && v.Flow != "pv" && v.Flow != "battery" && v.Flow != "ev" && v.Flow != "v2x" { + return errors.New("unknown electrical flow") + } + if seen[v.Flow] { + return errors.New("duplicate electrical flow") + } + seen[v.Flow] = true + if v.Absent { + if v.Flow == "grid" || v.Driver != "" || v.Metric != "" { + return errors.New("grid cannot be absent; absent flows cannot name a source") + } + continue + } + if v.Driver == "" || !strings.HasSuffix(v.Metric, "_w") { + return errors.New("each present flow needs an explicit driver and power metric ending in _w") + } + key := v.Driver + ":" + v.Metric + if electrical[key] { + return errors.New("one electrical source cannot be counted twice") + } + electrical[key] = true + } + if len(seen) != 5 { + return errors.New("select grid, pv, battery, ev and v2x; explicitly use none for absent flows") + } + return nil +} + +type repairPowerPoint struct { + at int64 + watts float64 +} + +// Aggregates are retained means, not reconstructed raw polls. Linear +// allocation and any bridge are estimates recorded in the correction plan. +func repairIntegral(points []repairPowerPoint, from, to, maxGap int64) (float64, int64, bool) { + var wh float64 + var covered, interpolated int64 + first := max(1, sort.Search(len(points), func(i int) bool { return points[i].at >= from })) + for i := first; i < len(points); i++ { + a, b := points[i-1], points[i] + if a.at >= to { + break + } + lo, hi := max(from, a.at), min(to, b.at) + if hi <= lo { + continue + } + dt := b.at - a.at + if dt <= 0 || dt > maxGap { + return 0, 0, false + } + left := a.watts + (b.watts-a.watts)*float64(lo-a.at)/float64(dt) + right := a.watts + (b.watts-a.watts)*float64(hi-a.at)/float64(dt) + wh += (left + right) / 2 * float64(hi-lo) / 3600000 + covered += hi - lo + if dt > maxLoadEstimateTickMS { + interpolated += hi - lo + } + } + return wh, interpolated, covered == to-from && finiteHistoryValue(wh) +} + +func (s *Store) planHistoryRepair(ctx context.Context, r HistoryRepairRequest) ([]repairMinute, map[int64]string, error) { + if err := r.validate(); err != nil { + return nil, nil, err + } + original := map[int64]*sourceLoadBucket{} + err := s.walkSiteEnergy(ctx, r.SinceMS, r.UntilMS, func(v siteEnergyRow) error { + if v.width != ArchiveResolutionMS { + return errors.New("repair needs retained minute energy; coarse historical intervals are not split") + } + b := original[v.start] + if b == nil { + b = &sourceLoadBucket{} + original[v.start] = b + } + b.covered += v.covered + b.wh += v.load + b.missing |= v.missing + return nil + }) + if err != nil { + return nil, nil, err + } + series := map[string][]repairPowerPoint{} + for _, v := range r.Sources { + if v.Absent { + continue + } + var unit string + if err := s.history.QueryRowContext(ctx, `SELECT COALESCE(unit,'') FROM ts_metrics WHERE name=?`, v.Metric).Scan(&unit); err != nil { + return nil, nil, fmt.Errorf("source %s:%s: %w", v.Driver, v.Metric, err) + } + if unit != "" && unit != "W" { + return nil, nil, errors.New("repair source is not in watts") + } + p, err := s.LoadSeriesBucketsOrRawContext(ctx, v.Driver, v.Metric, r.SinceMS-r.MaxGapMS-60000, r.UntilMS+r.MaxGapMS+60000, 0) + if err != nil { + return nil, nil, err + } + for _, q := range p { + if q.ResolutionMS > 60000 || !finiteHistoryValue(q.V) { + return nil, nil, errors.New("repair requires finite power evidence at minute resolution or finer") + } + at := q.TsMs + if q.FirstMS > 0 { + at = q.FirstMS + (q.TsMs-q.FirstMS)/2 + } + series[v.Flow] = append(series[v.Flow], repairPowerPoint{at, q.V}) + } + for i := 1; i < len(series[v.Flow]); i++ { + if series[v.Flow][i].at <= series[v.Flow][i-1].at { + return nil, nil, errors.New("conflicting source observation times") + } + } + } + var minutes []repairMinute + skipped := map[int64]string{} + for at := r.SinceMS; at < r.UntilMS; at += 60000 { + b := original[at] + if b != nil && b.covered == 60000 && b.missing&MissingPower == 0 && b.wh >= 0 { + continue + } + var existing int + if err := s.history.QueryRowContext(ctx, `SELECT COUNT(*) FROM history_load_estimates WHERE start_ms=?`, at).Scan(&existing); err != nil { + return nil, nil, err + } + if existing != 0 { + skipped[at] = "existing estimate preserved" + continue + } + p := repairMinute{StartMS: at} + valid := true + for _, v := range r.Sources { + if v.Absent { + continue + } + wh, bridge, ok := repairIntegral(series[v.Flow], at, at+60000, r.MaxGapMS) + if !ok { + skipped[at] = "insufficient bounded power evidence: " + v.Flow + valid = false + break + } + if v.Flow == "grid" { + p.Wh += wh + } else { + p.Wh -= wh + } + p.InterpolatedMS = max(p.InterpolatedMS, bridge) + } + if valid { + if !plausibleEnergy(absRepairWh(p.Wh), 60000) { + skipped[at] = "implausible reconstructed energy" + continue + } + minutes = append(minutes, p) + } + } + return minutes, skipped, nil +} + +func absRepairWh(v float64) float64 { + if v < 0 { + return -v + } + return v +} + +func repairFileHash(path string) (string, error) { + info, err := os.Lstat(path) + if err != nil { + return "", err + } + if !info.Mode().IsRegular() { + return "", errors.New("repair files must be regular files, not links") + } + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + h := sha256.New() + if _, err = io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +func repairCopyFile(from, to string) error { + a, err := os.Open(from) + if err != nil { + return err + } + defer a.Close() + b, err := os.OpenFile(to, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600) + if err != nil { + return err + } + _, copyErr := io.Copy(b, a) + syncErr := b.Sync() + return errors.Join(copyErr, syncErr, b.Close()) +} + +func writeRepairJSON(path string, v any) error { + b, err := json.MarshalIndent(v, "", " ") + if err != nil { + return err + } + return os.WriteFile(path, append(b, '\n'), 0600) +} + +// PrepareHistoryRepair never opens the owner's databases for writing. It +// snapshots state and history, then operates exclusively on the new copies. +func PrepareHistoryRepair(ctx context.Context, statePath, workdir string, r HistoryRepairRequest) (HistoryRepairPlan, error) { + var plan HistoryRepairPlan + if err := r.validate(); err != nil { + return plan, err + } + if err := os.Mkdir(workdir, 0700); err != nil { + return plan, err + } + for _, dir := range []string{"backup", "corrected"} { + if err := os.Mkdir(filepath.Join(workdir, dir), 0700); err != nil { + return plan, err + } + } + s, err := OpenBackupSource(statePath) + if err != nil { + return plan, err + } + if s.history == nil { + s.Close() + return plan, errors.New("repair requires primary SQLite history") + } + if err := EnsureDiskSpace(workdir, 3*s.BackupSourceBytes()+backupScratchHeadroom); err != nil { + s.Close() + return plan, err + } + _, err = s.db.ExecContext(ctx, `VACUUM INTO ?`, filepath.Join(workdir, "backup", "state.db")) + if err == nil { + _, err = s.history.ExecContext(ctx, `VACUUM INTO ?`, filepath.Join(workdir, "backup", HistoryFilename)) + } + closeErr := s.Close() + if err != nil { + return plan, err + } + if closeErr != nil { + return plan, closeErr + } + for _, name := range []string{"state.db", HistoryFilename} { + if err := repairCopyFile(filepath.Join(workdir, "backup", name), filepath.Join(workdir, "corrected", name)); err != nil { + return plan, err + } + } + plan.BackupStateSHA256, err = repairFileHash(filepath.Join(workdir, "backup", "state.db")) + if err != nil { + return plan, err + } + plan.BackupHistorySHA256, err = repairFileHash(filepath.Join(workdir, "backup", HistoryFilename)) + if err != nil { + return plan, err + } + c, err := Open(filepath.Join(workdir, "corrected", "state.db")) + if err != nil { + return plan, err + } + var integrity string + err = c.history.QueryRowContext(ctx, `PRAGMA quick_check`).Scan(&integrity) + if err == nil && integrity != "ok" { + err = errors.New("history integrity check failed") + } + if err == nil { + plan.Minutes, plan.Skipped, err = c.planHistoryRepair(ctx, r) + } + closeErr = c.Close() + if err != nil { + return plan, err + } + if closeErr != nil { + return plan, closeErr + } + plan.WorkingHistorySHA256, err = repairFileHash(filepath.Join(workdir, "corrected", HistoryFilename)) + if err != nil { + return plan, err + } + plan.WorkingStateSHA256, err = repairFileHash(filepath.Join(workdir, "corrected", "state.db")) + if err != nil { + return plan, err + } + plan.Request = r + plan.CreatedAt = time.Now().UTC().Format(time.RFC3339Nano) + plan.SHA256 = plan.digest() + return plan, writeRepairJSON(filepath.Join(workdir, "plan.json"), plan) +} + +type HistoryRepairResult struct { + PlanSHA256 string `json:"plan_sha256"` + AppliedMinutes int `json:"applied_minutes"` + BeforeWh any `json:"before_load_wh"` + AfterWh any `json:"after_load_wh"` + AfterQuality string `json:"after_quality"` + Estimated bool `json:"load_estimated"` +} + +// ApplyHistoryRepair writes the approved additive estimates to the working +// copy only. A matching retry is idempotent; stale/tampered plans fail closed. +func ApplyHistoryRepair(ctx context.Context, workdir, approvedSHA string) (HistoryRepairResult, error) { + var result HistoryRepairResult + var p HistoryRepairPlan + b, err := os.ReadFile(filepath.Join(workdir, "plan.json")) + if err != nil { + return result, err + } + if err = json.Unmarshal(b, &p); err != nil { + return result, err + } + if len(approvedSHA) != 64 || p.SHA256 != approvedSHA || p.digest() != approvedSHA { + return result, errors.New("exact reviewed plan SHA-256 is required") + } + if err = p.Request.validate(); err != nil { + return result, err + } + for _, v := range []struct{ name, want string }{{"state.db", p.BackupStateSHA256}, {HistoryFilename, p.BackupHistorySHA256}} { + got, err := repairFileHash(filepath.Join(workdir, "backup", v.name)) + if err != nil || got != v.want { + return result, errors.New("verified backup changed or is unavailable") + } + } + working := filepath.Join(workdir, "corrected", HistoryFilename) + stateHash, err := repairFileHash(filepath.Join(workdir, "corrected", "state.db")) + if err != nil || stateHash != p.WorkingStateSHA256 { + return result, errors.New("working state changed after the plan") + } + hash, err := repairFileHash(working) + if err != nil { + return result, err + } + s, err := Open(filepath.Join(workdir, "corrected", "state.db")) + if err != nil { + return result, err + } + defer s.Close() + methodJSON, _ := json.Marshal(map[string]any{"plan_sha256": p.SHA256, "backup_history_sha256": p.BackupHistorySHA256, "request": p.Request, "created_at": p.CreatedAt}) + method := historicalRepairMethod + string(methodJSON) + if len(p.Minutes) == 0 { + return result, errors.New("plan has no supported corrections") + } + allApplied := true + for _, v := range p.Minutes { + var wh float64 + var saved string + err := s.history.QueryRowContext(ctx, `SELECT load_wh,method FROM history_load_estimates WHERE start_ms=? AND resolution_ms=60000`, v.StartMS).Scan(&wh, &saved) + if err != nil || saved != method || wh != v.Wh { + allApplied = false + break + } + } + if hash != p.WorkingHistorySHA256 && !allApplied { + return result, errors.New("working history changed after the plan; prepare a new copy") + } + beforeSource, err := OpenBackupSource(filepath.Join(workdir, "backup", "state.db")) + if err != nil { + return result, err + } + before, err := beforeSource.DailyEnergy(p.Request.SinceMS, p.Request.UntilMS) + beforeSource.Close() + if err != nil { + return result, err + } + if !allApplied { + tx, err := s.history.BeginTx(ctx, nil) + if err != nil { + return result, err + } + defer tx.Rollback() + for _, v := range p.Minutes { + if v.StartMS < p.Request.SinceMS || v.StartMS+60000 > p.Request.UntilMS || v.StartMS%60000 != 0 || !finiteHistoryValue(v.Wh) { + return result, errors.New("invalid correction minute") + } + if _, err = tx.ExecContext(ctx, `INSERT INTO history_load_estimates(start_ms,resolution_ms,covered_ms,load_wh,method) VALUES(?,60000,60000,?,?)`, v.StartMS, v.Wh, method); err != nil { + return result, err + } + } + // All these tables are derived caches. Original observations and their + // quality are untouched; a new Core process recomputes corrected days. + for _, table := range []string{"energy_daily", "energy_daily_partial", "history_daily_quality"} { + if _, err = tx.ExecContext(ctx, `DELETE FROM `+table); err != nil { + return result, err + } + } + if err = tx.Commit(); err != nil { + return result, err + } + } + after, err := s.DailyEnergy(p.Request.SinceMS, p.Request.UntilMS) + if err != nil { + return result, err + } + result = HistoryRepairResult{p.SHA256, len(p.Minutes), before.Missing.Value(MissingLoad, before.LoadWh), after.Missing.Value(MissingLoad, after.LoadWh), after.Missing.Quality(), after.Missing&EstimatedLoad != 0} + return result, writeRepairJSON(filepath.Join(workdir, "apply-result.json"), result) +} + +// Keep plan order deterministic for callers that construct requests in maps. +func SortRepairSources(sources []RepairPowerSource) { + sort.Slice(sources, func(i, j int) bool { return sources[i].Flow < sources[j].Flow }) +} diff --git a/go/internal/state/history_repair_read.go b/go/internal/state/history_repair_read.go new file mode 100644 index 000000000..84966c04d --- /dev/null +++ b/go/internal/state/history_repair_read.go @@ -0,0 +1,233 @@ +package state + +import ( + "context" + "encoding/json" + "errors" + "sort" + "strings" +) + +const historicalRepairMethod = "historical_power_repair_v1:" + +type repairMinute struct { + StartMS int64 `json:"start_ms"` + Wh float64 `json:"wh"` + InterpolatedMS int64 `json:"max_interpolated_ms_per_source"` +} + +func (s *Store) historicalLoadRepairs(ctx context.Context, from, to int64) (map[int64]repairMinute, error) { + out := map[int64]repairMinute{} + if s.history == nil { + return out, nil + } + rows, err := s.history.QueryContext(ctx, `SELECT start_ms,resolution_ms,covered_ms,load_wh,method FROM history_load_estimates WHERE start_ms>=? AND start_ms to { + continue + } + if err := visit(siteEnergyRow{start: start, width: ArchiveResolutionMS, first: start, last: start + ArchiveResolutionMS, + covered: ArchiveResolutionMS, load: p.Wh, missing: EstimatedLoad}); err != nil { + return err + } + } + return nil +} + +func (s *Store) repairedLoadBuckets(ctx context.Context, from, to, width int64) ([]sourceLoadBucket, bool, error) { + repairs, err := s.historicalLoadRepairs(ctx, from, to) + if err != nil || len(repairs) == 0 { + return nil, false, err + } + byStart := map[int64]*sourceLoadBucket{} + err = s.walkHouseholdLoad(ctx, from, to, repairs, func(r siteEnergyRow) error { + w := max(width, r.width) + start := from + (r.start-from)/w*w + b := byStart[start] + if b == nil { + b = &sourceLoadBucket{start: start, width: w, first: r.first, last: r.last} + byStart[start] = b + } + if b.width != w || r.first < start || r.last > start+w { + b.missing |= MissingLoad + } + b.first = min(b.first, r.first) + b.last = max(b.last, r.last) + b.covered += r.covered + b.n += r.n + b.wh += r.load + b.missing |= r.missing + return nil + }) + if err != nil { + return nil, true, err + } + out := make([]sourceLoadBucket, 0, len(byStart)) + for _, b := range byStart { + if b.covered != b.width || b.first != b.start || b.last != b.start+b.width || b.wh < 0 || !finiteHistoryValue(b.wh) { + b.missing |= MissingLoad + } + out = append(out, *b) + } + sort.Slice(out, func(i, j int) bool { return out[i].start < out[j].start }) + return out, true, nil +} + +func (s *Store) applyHistoricalDailyLoad(ctx context.Context, from, to int64, d *DayEnergy) error { + // A historical correction cannot invent the currently unfinished minute. + end := bucketStart(to, ArchiveResolutionMS) + if end <= from { + return nil + } + b, have, err := s.repairedLoadBuckets(ctx, from, end, end-from) + if err != nil || !have { + return err + } + if len(b) == 1 && b[0].missing&(MissingLoad|BalanceMismatch) == 0 && b[0].covered == end-from { + d.LoadWh = b[0].wh + d.Missing &^= MissingLoad | BalanceMismatch + d.Missing |= EstimatedLoad + d.LoadUntilMS = end + } + return nil +} + +func (s *Store) applyHistoricalLoadChart(ctx context.Context, points []HistoryPoint, from, to, width int64) ([]HistoryPoint, error) { + if width < ArchiveResolutionMS { + return points, nil + } // Do not invent raw poll values. + b, have, err := s.repairedLoadBuckets(ctx, from, to, width) + if err != nil || !have { + return points, err + } + byStart := map[int64]sourceLoadBucket{} + seen := map[int64]bool{} + repairs, err := s.historicalLoadRepairs(ctx, from, to) + if err != nil { + return points, err + } + changed := map[int64]bool{} + for start := range repairs { + changed[from+(start-from)/width*width] = true + } + for _, p := range b { + byStart[p.start] = p + } + for i := range points { + p := &points[i] + start := from + (p.TsMs-from)/width*width + if !changed[start] { + continue + } + v, ok := byStart[start] + if !ok || v.missing&(MissingLoad|BalanceMismatch) != 0 { + continue + } + p.LoadW = v.wh * 3600000 / float64(v.width) + seen[start] = true + p.Missing &^= MissingLoad | BalanceMismatch + p.Missing |= EstimatedLoad + var detail map[string]any + if json.Unmarshal([]byte(p.JSON), &detail) == nil && detail != nil { + detail["history_load_method"] = "historical_power_repair_v1" + detail["load_quality"] = "estimated" + if data, err := json.Marshal(detail); err == nil { + p.JSON = string(data) + } + } + } + for _, v := range b { + if !changed[v.start] || seen[v.start] || v.missing&(MissingLoad|BalanceMismatch) != 0 { + continue + } + points = append(points, HistoryPoint{TsMs: v.start + v.width - 1, FirstMS: v.start, ResolutionMS: v.width, LoadW: v.wh * 3600000 / float64(v.width), + Missing: MissingGrid | MissingPV | MissingBattery | MissingEV | MissingV2X | MissingSoC | EstimatedLoad, + JSON: `{"history_load_method":"historical_power_repair_v1","load_quality":"estimated","forecast_measurement_quality":"aggregate_not_for_training"}`}) + } + sort.SliceStable(points, func(i, j int) bool { return points[i].TsMs < points[j].TsMs }) + return points, nil +} + +func (s *Store) applyHistoricalLoadLedger(ctx context.Context, q EnergyHistoryQuery, points []EnergyLedgerPoint) ([]EnergyLedgerPoint, error) { + if q.AssetID != "" && q.AssetID != ObservedConsumerAssetID { + return points, nil + } + b, have, err := s.repairedLoadBuckets(ctx, q.SinceMS, q.UntilMS, q.BucketMS) + if err != nil || !have { + return points, err + } + repairs, err := s.historicalLoadRepairs(ctx, q.SinceMS, q.UntilMS) + if err != nil { + return nil, err + } + changed := map[int64]bool{} + for start := range repairs { + changed[q.SinceMS+(start-q.SinceMS)/q.BucketMS*q.BucketMS] = true + } + out := points[:0] + for _, p := range points { + if p.Flow != FlowConsumerUse || !changed[p.BucketStartMS] { + out = append(out, p) + } + } + for _, p := range b { + if !changed[p.start] { + continue + } + if p.missing&(MissingLoad|BalanceMismatch) != 0 { + continue + } + asset := "system" + if q.AssetID != "" { + asset = q.AssetID + } + out = append(out, EnergyLedgerPoint{SchemaVersion: EnergyLedgerSchemaVersion, AssetID: asset, Flow: FlowConsumerUse, + BucketStartMS: p.start, BucketLenMS: p.width, EnergyWh: p.wh, Source: "historical_repair", Quality: "estimated", Provenance: "historical_power_repair_v1", SampleCount: p.n}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].BucketStartMS < out[j].BucketStartMS }) + return out, nil +} diff --git a/go/internal/state/history_repair_test.go b/go/internal/state/history_repair_test.go new file mode 100644 index 000000000..64b4d9c2c --- /dev/null +++ b/go/internal/state/history_repair_test.go @@ -0,0 +1,350 @@ +package state + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "reflect" + "testing" + "time" +) + +func historicalRepairFixture(t *testing.T, powerGap bool) (string, HistoryRepairRequest) { + t.Helper() + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour).UnixMilli() + for i := int64(0); i <= 90; i++ { + v := HistorySources{GridW: 2000, PVW: -500, BatteryW: 100, EVW: 400} + if i == 42 { + v.Missing = MissingPV + } + if i == 66 { + continue + } // retained scalar series can repair a writer gap too. + sourceTick(t, s, base+i*10000, v) + } + var samples []Sample + for at := base - 30000; at <= base+930000; at += 5000 { + for metric, w := range map[string]float64{"meter_w": 2000, "pv_w": -500, "battery_ac_w": 100, "ev_w": 400} { + if powerGap && metric == "pv_w" && at >= base+360000 && at <= base+480000 { + continue + } + samples = append(samples, Sample{Driver: "equipment", Metric: metric, TsMs: at, Value: w, Unit: "W"}) + } + } + if err := s.RecordSamples(samples); err != nil { + t.Fatal(err) + } + path := s.mainDBPath + if err := s.SaveDailyEnergy("stale-cache", DayEnergy{Missing: MissingLoad, Intervals: 1}); err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + r := HistoryRepairRequest{SinceMS: base, UntilMS: base + 900000, MaxGapMS: 15000, Reason: "fixture: reconstruct retained AC source energy", + Sources: []RepairPowerSource{{Flow: "grid", Driver: "equipment", Metric: "meter_w"}, {Flow: "pv", Driver: "equipment", Metric: "pv_w"}, {Flow: "battery", Driver: "equipment", Metric: "battery_ac_w"}, {Flow: "ev", Driver: "equipment", Metric: "ev_w"}, {Flow: "v2x", Absent: true}}} + return path, r +} + +func TestHistoryRepairCopyRecalculatesHistoricalTotalsAndPreservesOriginal(t *testing.T) { + ctx := context.Background() + source, r := historicalRepairFixture(t, false) + originalState, _ := repairFileHash(source) + originalHistory, _ := repairFileHash(filepath.Join(filepath.Dir(source), HistoryFilename)) + work := filepath.Join(t.TempDir(), "repair") + plan, err := PrepareHistoryRepair(ctx, source, work, r) + if err != nil { + t.Fatal(err) + } + if len(plan.Minutes) == 0 || len(plan.Skipped) != 0 { + t.Fatal("fixture has reconstructable gaps", plan) + } + if _, err := ApplyHistoryRepair(ctx, work, "unreviewed"); err == nil { + t.Fatal("unreviewed plan applied") + } + result, err := ApplyHistoryRepair(ctx, work, plan.SHA256) + if err != nil { + t.Fatal(err) + } + if result.BeforeWh != nil || result.AfterWh == nil || !result.Estimated { + t.Fatal("historical total not corrected", result) + } + near(t, "reconstructed period Wh", result.AfterWh.(float64), 500) + again, err := ApplyHistoryRepair(ctx, work, plan.SHA256) + if err != nil || !reflect.DeepEqual(result, again) { + t.Fatal("retry double counted", again, err) + } + if h, _ := repairFileHash(source); h != originalState { + t.Fatal("source state changed") + } + if h, _ := repairFileHash(filepath.Join(filepath.Dir(source), HistoryFilename)); h != originalHistory { + t.Fatal("source history changed") + } + s, err := Open(filepath.Join(work, "corrected", "state.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + var quality int + if err := s.history.QueryRow(`SELECT COUNT(*) FROM history_quality WHERE missing&?!=0`, MissingPV).Scan(&quality); err != nil || quality == 0 { + t.Fatal("original rejected evidence erased", quality, err) + } + var cached int + if err := s.history.QueryRow(`SELECT COUNT(*) FROM energy_daily_partial WHERE day='stale-cache'`).Scan(&cached); err != nil || cached != 0 { + t.Fatal("stale cache retained", cached, err) + } + d, err := s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&MissingLoad != 0 || d.Missing&EstimatedLoad == 0 { + t.Fatal("restart lost repair", d, err) + } + if d.Missing&MissingPV == 0 { + t.Fatal("load correction falsely repaired the PV measurement") + } + if err := s.SaveDailyEnergy("corrected-cache", d); err != nil { + t.Fatal(err) + } + c, hit, err := s.LoadDailyEnergy("corrected-cache") + if err != nil || !hit || c.Missing&EstimatedLoad == 0 { + t.Fatal(c, hit, err) + } + near(t, "persisted corrected Wh", c.LoadWh, 500) + points, truncated, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: r.SinceMS, UntilMS: r.UntilMS, BucketMS: 300000, Limit: 10}) + if err != nil || truncated { + t.Fatal(err) + } + var wh float64 + var n int + var estimated int + for _, p := range points { + if p.Flow == FlowConsumerUse { + wh += p.EnergyWh + n++ + if p.Quality == "estimated" { + estimated++ + } + } + } + if n != 3 { + t.Fatal("historical ledger not corrected", points) + } + if estimated == 0 { + t.Fatal("repair provenance missing from ledger") + } + near(t, "ledger Wh", wh, 500) + limited, truncated, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: r.SinceMS, UntilMS: r.UntilMS, BucketMS: 300000, Limit: 1}) + if err != nil || !truncated { + t.Fatal("repair bypassed limit", err) + } + for _, p := range limited { + if p.BucketStartMS != r.SinceMS { + t.Fatal("extra buckets exceeded limit", p) + } + } + chart, err := s.LoadHistory(r.SinceMS, r.UntilMS, 3) + if err != nil { + t.Fatal(err) + } + for _, p := range chart { + if p.Missing&MissingLoad != 0 { + t.Fatal("chart kept a repaired load gap", p) + } + } + if len(chart) == 0 { + t.Fatal("corrected chart is empty") + } + // Retention must keep the originals needed to interpret the additive + // minute corrections, while ordinary uncorrected quarters may roll up. + if err := s.rollSiteEnergy(ctx, r.UntilMS); err != nil { + t.Fatal(err) + } + d, err = s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&MissingLoad != 0 { + t.Fatal("retention lost repair", d, err) + } + near(t, "retained corrected Wh", d.LoadWh, 500) + backup := filepath.Join(t.TempDir(), "restored.db") + if err := s.copyStateForBackup(ctx, backup); err != nil { + t.Fatal(err) + } + if err := s.exportHistoryToSQLite(ctx, backup); err != nil { + t.Fatal(err) + } + restored, err := Open(backup) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + d, err = restored.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&MissingLoad != 0 || d.Missing&EstimatedLoad == 0 { + t.Fatal("portable backup lost repair", d, err) + } + near(t, "restored corrected Wh", d.LoadWh, 500) +} + +func TestHistoryRepairAddsDerivedChartIntervalWithoutInventingOtherMeasurements(t *testing.T) { + source, r := historicalRepairFixture(t, false) + work := filepath.Join(t.TempDir(), "repair") + p, err := PrepareHistoryRepair(context.Background(), source, work, r) + if err != nil { + t.Fatal(err) + } + if _, err := ApplyHistoryRepair(context.Background(), work, p.SHA256); err != nil { + t.Fatal(err) + } + s, err := Open(filepath.Join(work, "corrected", "state.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + start := p.Minutes[0].StartMS + for _, table := range []string{"history_dashboard", "history_dashboard_partial"} { + if _, err := s.history.Exec(`DELETE FROM `+table+` WHERE last_ms>=? AND last_ms= 0 { + t.Fatal("negative residual clipped", negative, ok) + } + if _, _, ok := repairIntegral(p, -1, 180000, 60000); ok { + t.Fatal("missing edge was extrapolated") + } + if _, _, ok := repairIntegral(p, 0, 180000, 15000); ok { + t.Fatal("long unapproved gaps filled") + } +} + +func TestHistoryRepairRefusesInsufficientCopySpace(t *testing.T) { + source, r := historicalRepairFixture(t, false) + probe := backupDiskAvail + backupDiskAvail = func(string) (int64, error) { return 0, nil } + defer func() { backupDiskAvail = probe }() + if _, err := PrepareHistoryRepair(context.Background(), source, filepath.Join(t.TempDir(), "repair"), r); err == nil { + t.Fatal("repair started without room for backup and working copy") + } +} diff --git a/go/internal/state/household_subtotal.go b/go/internal/state/household_subtotal.go index 436fca92a..8d587959c 100644 --- a/go/internal/state/household_subtotal.go +++ b/go/internal/state/household_subtotal.go @@ -23,8 +23,12 @@ func (s *Store) HouseholdEnergySubtotal(ctx context.Context, from, to int64) (*H return nil, nil } const width int64 = 5 * 60000 + repairs, err := s.historicalLoadRepairs(ctx, from, to) + if err != nil { + return nil, err + } buckets := map[int64]*sourceLoadBucket{} - err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { + err = s.walkHouseholdLoad(ctx, from, to, repairs, func(r siteEnergyRow) error { // Retained coarse intervals remain indivisible; never split an older // observed interval into invented five-minute measurements. w := max(width, r.width) diff --git a/go/internal/state/store.go b/go/internal/state/store.go index 0ab26cafb..6e1733d50 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -1505,6 +1505,11 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, return nil, err } sourceMode = sourceMode && untilMs >= cut + if !sourceMode { + if err := s.history.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM history_load_estimates WHERE start_ms>=? AND start_ms 0 { @@ -1615,6 +1620,10 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, return nil, err } } + all, err = s.applyHistoricalLoadChart(ctx, all, queryFrom, untilMs, bucketMs) + if err != nil { + return nil, err + } return all, nil } @@ -1633,6 +1642,9 @@ type DayEnergy struct { BatChargedWh float64 BatDischargedWh float64 LoadWh float64 + // Historical repairs use completed minute evidence, not an extrapolated + // current tail. Status exposes this bound independently of live power. + LoadUntilMS int64 // Intervals is the number of integration intervals that contributed to // the totals — i.e. the count of rows that had a predecessor (prev_ts IS // NOT NULL). Zero means there was at most one history row in the range, so @@ -1678,7 +1690,11 @@ func (s *Store) DailyEnergy(sinceMs, untilMs int64) (DayEnergy, error) { return DayEnergy{}, err } if !ok || untilMs <= from { - return s.legacyDailyEnergy(sinceMs, untilMs) + d, err := s.legacyDailyEnergy(sinceMs, untilMs) + if err == nil { + err = s.applyHistoricalDailyLoad(ctx, sinceMs, untilMs, &d) + } + return d, err } out, err := s.aggregateDayEnergy(ctx, max(from, sinceMs), untilMs) if err != nil { @@ -1691,6 +1707,9 @@ func (s *Store) DailyEnergy(sinceMs, untilMs int64) (DayEnergy, error) { } out = addDayEnergy(old, out) } + if err := s.applyHistoricalDailyLoad(ctx, sinceMs, untilMs, &out); err != nil { + return out, err + } return out, nil } diff --git a/web/components/energy-flow-readings.js b/web/components/energy-flow-readings.js index d32ad6092..dfb82f8df 100644 --- a/web/components/energy-flow-readings.js +++ b/web/components/energy-flow-readings.js @@ -252,7 +252,7 @@ export function flowReadingsFromStatus(status, opts) { // quiet meter blanks % SELF-POWERED NOW (unknown instantaneous load) // but must not hide a day that already happened. let selfPoweredPctToday = null; - if (loadWh > 1 && importWh !== null && today.history_quality !== "estimated") { + if (loadWh > 1 && importWh !== null && today.history_quality !== "estimated" && today.load_quality !== "estimated") { selfPoweredPctToday = Math.max(0, Math.min(100, (1 - importWh / loadWh) * 100)); } diff --git a/web/energy-today.js b/web/energy-today.js index 88302c8f8..71c8b3f37 100644 --- a/web/energy-today.js +++ b/web/energy-today.js @@ -16,7 +16,7 @@ function knownWh(value) { export function energyTodayReading(today, key) { const total = today?.[key]; - if (knownWh(total)) return key === "load_wh" && today?.history_quality === "estimated" + if (knownWh(total)) return key === "load_wh" && (today?.history_quality === "estimated" || today?.load_quality === "estimated") ? {wh: total, partial: false, estimated: true} : {wh: total, partial: false}; const measured = today?.measured?.[key]; if (knownWh(measured)) return {wh: measured, partial: true}; @@ -46,7 +46,7 @@ export function updateEnergyToday(today, root = document) { const {partial, estimated} = energyTodayReading(today, key); el.textContent = formatEnergyToday(today, key); el.title = estimated && partial ? "Estimated subtotal from accepted energy intervals only; gaps are excluded." - : estimated ? "Estimated from source energy during brief balance failures." + : estimated ? "Estimated from source energy; may include audited historical corrections." : partial ? "Recorded energy only; missing readings are excluded." : ""; } const group = root.getElementById("e-pv")?.closest(".energy-group"); @@ -60,12 +60,12 @@ export function updateEnergyToday(today, root = document) { } const keys = new Set(Object.values(ENERGY_FIELDS)); const incomplete = [...keys].some(key => !knownWh(today?.[key])); - const estimated = today?.history_quality === "estimated"; + const estimated = today?.history_quality === "estimated" || today?.load_quality === "estimated"; note.hidden = !incomplete && !estimated; note.textContent = today ? "Gaps in today's readings. ≥ shows recorded energy only." : "Today's energy is not available yet."; - if (estimated) note.textContent = (incomplete ? note.textContent + " " : "") + "Home use includes estimates for brief balance failures."; + if (estimated) note.textContent = (incomplete ? note.textContent + " " : "") + "Home use includes estimates from source energy."; const home = energyTodayReading(today, "load_wh"); if (home.estimated && home.partial) { note.hidden = false; diff --git a/web/history-repair-quality.test.mjs b/web/history-repair-quality.test.mjs new file mode 100644 index 000000000..d6f918c23 --- /dev/null +++ b/web/history-repair-quality.test.mjs @@ -0,0 +1,12 @@ +import assert from "node:assert/strict"; +import {test} from "node:test"; +import {formatEnergyToday} from "./energy-today.js"; +import {flowReadingsFromStatus} from "./components/energy-flow-readings.js"; + +test("corrected household energy stays estimated even when another original flow is missing",()=>{ + const today={load_wh:28000,load_quality:"estimated",history_quality:"incomplete",pv_wh:null,import_wh:12000,measured:{load_wh:null}}; + assert.equal(formatEnergyToday(today,"load_wh"),"≈ 28.0 kWh"); + assert.equal(formatEnergyToday(today,"pv_wh"),"Unknown"); + assert.equal(flowReadingsFromStatus({load_w:1200,energy:{today}}).selfPoweredPctToday,null); + assert.equal(today.measured.load_wh,null); +}); From 6141506ddfe8da8a33e1011edd5a02e376cbbbaf Mon Sep 17 00:00:00 2001 From: Segran Date: Mon, 19 Oct 2026 23:00:35 +0200 Subject: [PATCH 3/5] fix(update): retain native updates for local repair builds Signed-off-by: Segran --- .changeset/home-use-subtotal.md | 2 ++ go/cmd/ftw/native_update.go | 27 ++++++++++++++++++++++++--- go/cmd/ftw/native_update_test.go | 31 +++++++++++++++++++++++++++++++ 3 files changed, 57 insertions(+), 3 deletions(-) diff --git a/.changeset/home-use-subtotal.md b/.changeset/home-use-subtotal.md index bee13acec..d4ad35448 100644 --- a/.changeset/home-use-subtotal.md +++ b/.changeset/home-use-subtotal.md @@ -5,3 +5,5 @@ Keep an estimated Home use subtotal visible after missing source readings or a recording gap. Publish its accepted interval coverage separately from the unknown full total and measured energy. Preserve original history and exclude gaps rather than filling them. Add an offline historical repair tool that snapshots the original databases, prepares a hashed plan from selected retained AC power series, and applies audited estimates only to a working copy. Core uses these corrections in household daily totals and history without changing source measurements or their quality. Unsupported gaps remain unknown. + +Keep the native launcher and update API available when a local patch identifies the installed release with SemVer build metadata. Match the current slot and clean-exit marker using the base release while keeping published trial identity checks exact. diff --git a/go/cmd/ftw/native_update.go b/go/cmd/ftw/native_update.go index 211e74bd5..333e310f0 100644 --- a/go/cmd/ftw/native_update.go +++ b/go/cmd/ftw/native_update.go @@ -6,6 +6,7 @@ import ( "log/slog" "os" "path/filepath" + "strings" "time" "github.com/srcfl/ftw/go/internal/nativeupdate" @@ -14,6 +15,26 @@ import ( const nativeTrialTimeout = 6 * time.Hour +// Build metadata identifies local patches without changing the installed +// release's SemVer identity. Published trial binaries still require an exact tag. +func nativeSlotVersion(version string) string { + base, metadata, found := strings.Cut(version, "+") + if !found || !nativeupdate.ValidTag(base) { + return version + } + for _, id := range strings.Split(metadata, ".") { + if id == "" { + return version + } + for _, c := range id { + if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '-') { + return version + } + } + } + return base +} + type nativeTrial struct { manager nativeupdate.Manager tag string @@ -36,7 +57,7 @@ func beginNativeTrial(root, tag, version string) (*nativeTrial, error) { return nil, err } if tag == "" { - if state.Current != version { + if state.Current != nativeSlotVersion(version) { return nil, fmt.Errorf("native current slot %s does not match binary %s", state.Current, version) } return nil, nil @@ -101,7 +122,7 @@ func reconcileNativeFallback(root, version string, checker *selfupdate.Checker) return } state, err := (nativeupdate.Manager{Root: root}).Read() - if err != nil || state.Current != version || state.LastFailed == "" || status.Target != state.LastFailed { + if err != nil || state.Current != nativeSlotVersion(version) || state.LastFailed == "" || status.Target != state.LastFailed { return } switch status.State { @@ -128,7 +149,7 @@ func markNativeCleanExit(root string) { if root == "" { return } - if err := (nativeupdate.Manager{Root: root}).MarkCleanExit(Version); err != nil { + if err := (nativeupdate.Manager{Root: root}).MarkCleanExit(nativeSlotVersion(Version)); err != nil { slog.Warn("native clean-exit mark could not be saved", "err", err) } } diff --git a/go/cmd/ftw/native_update_test.go b/go/cmd/ftw/native_update_test.go index 05e921b21..f121a5af6 100644 --- a/go/cmd/ftw/native_update_test.go +++ b/go/cmd/ftw/native_update_test.go @@ -80,6 +80,37 @@ func TestNativeTrialCommitsOnlyAfterReadiness(t *testing.T) { } } +func TestNativeCurrentBuildMetadataKeepsReleaseUpdates(t *testing.T) { + root := t.TempDir() + current := "v0.154.0-beta.1" + nativeFixtureRelease(t, root, current) + if err := (nativeupdate.Manager{Root: root}).Init(current); err != nil { + t.Fatal(err) + } + for _, version := range []string{current, current + "+history-repair.6f71db1"} { + trial, err := beginNativeTrial(root, "", version) + if err != nil || trial != nil { + t.Fatalf("installed release build %q rejected: %v", version, err) + } + } + previousVersion := Version + Version = current + "+history-repair.6f71db1" + t.Cleanup(func() { Version = previousVersion }) + markNativeCleanExit(root) + mark, err := os.ReadFile(filepath.Join(root, ".clean-exit")) + if err != nil || strings.TrimSpace(string(mark)) != current { + t.Fatalf("clean exit did not identify installed release: %q %v", mark, err) + } + for _, version := range []string{"v0.153.0-beta.1+history-repair", current + "+", current + "+bad..metadata", current + "+bad/metadata"} { + if _, err := beginNativeTrial(root, "", version); err == nil { + t.Fatalf("wrong release or malformed metadata accepted: %q", version) + } + } + if _, err := beginNativeTrial(root, current, current+"+history-repair"); err == nil { + t.Fatal("published trial identity must remain exact") + } +} + func TestNativeFallbackReportsFailedTrial(t *testing.T) { root := t.TempDir() current, next := "v0.131.0-beta.1", "v0.131.0-beta.2" From 074a4c6bafa5c36bc75e18c31ea329e775088052 Mon Sep 17 00:00:00 2001 From: Segran Date: Mon, 19 Oct 2026 23:21:32 +0200 Subject: [PATCH 4/5] fix(history): repair home and solar gaps from audited endpoint means Signed-off-by: Segran --- .changeset/home-use-subtotal.md | 2 + docs/history-endpoint-repair.md | 74 ++++ go/cmd/ftw-history-endpoints/main.go | 95 +++++ go/internal/api/api.go | 9 + go/internal/api/api_energy_today.go | 12 + go/internal/api/api_history_repair_test.go | 10 + .../state/history_dashboard_retention.go | 1 + go/internal/state/history_endpoint_read.go | 213 ++++++++++ go/internal/state/history_endpoint_repair.go | 376 ++++++++++++++++++ .../state/history_endpoint_repair_test.go | 263 ++++++++++++ go/internal/state/history_repair_read.go | 22 +- go/internal/state/store.go | 7 + web/energy-today.js | 3 +- web/history-repair-quality.test.mjs | 6 + 14 files changed, 1088 insertions(+), 5 deletions(-) create mode 100644 docs/history-endpoint-repair.md create mode 100644 go/cmd/ftw-history-endpoints/main.go create mode 100644 go/internal/state/history_endpoint_read.go create mode 100644 go/internal/state/history_endpoint_repair.go create mode 100644 go/internal/state/history_endpoint_repair_test.go diff --git a/.changeset/home-use-subtotal.md b/.changeset/home-use-subtotal.md index d4ad35448..7a31c378b 100644 --- a/.changeset/home-use-subtotal.md +++ b/.changeset/home-use-subtotal.md @@ -7,3 +7,5 @@ Keep an estimated Home use subtotal visible after missing source readings or a r Add an offline historical repair tool that snapshots the original databases, prepares a hashed plan from selected retained AC power series, and applies audited estimates only to a working copy. Core uses these corrections in household daily totals and history without changing source measurements or their quality. Unsupported gaps remain unknown. Keep the native launcher and update API available when a local patch identifies the installed release with SemVer build metadata. Match the current slot and clean-exit marker using the base release while keeping published trial identity checks exact. + +Support audited endpoint-mean repairs for both historical Home use and Solar. Prepare while Core collects data, review the exact hash, and publish additive correction metadata atomically without replacing a live database. Show per-flow estimate labels and completed-through timestamps, preserving source observations and update support. diff --git a/docs/history-endpoint-repair.md b/docs/history-endpoint-repair.md new file mode 100644 index 000000000..649a361fd --- /dev/null +++ b/docs/history-endpoint-repair.md @@ -0,0 +1,74 @@ +# Repair Home use and Solar from bounding power + +`ftw-history-endpoints` estimates only incomplete retained minutes. It uses +`Wh = (before_W + after_W) / 2 × duration_hours`, keeping the same mean through +a contiguous gap. Home use endpoints are the closest valid retained minute +source-energy means before and after the gap. Solar endpoints are valid +retained ten-second PV means, expressed as positive generation. Two observed +night-time zeros give 0 Wh. Missing endpoints never imply zero. + +The endpoint distance is bounded explicitly (default 15 minutes, maximum 30). +This is historical estimation, not recovered measurement. Changes between +the two readings are unknown. A corrected minute replaces that flow's whole +incomplete minute in the read view; its originally recorded part is not added +a second time. Complete minutes and other flows keep their original evidence. + +Build Core and the helper from the same source. Native local Core builds may +retain valid SemVer build metadata while using the installed base-release +slot; keep the normal launcher so updates remain available. Prepare the build +and transfer it before a short stopped-Core binary swap. Do not stop Core +while copying or reviewing the database. Never start a second controlling Core. + +Take an independent, integrity-checked SQLite online backup of `history.db` +while the live Core continues collecting. Use SQLite's backup API with a read +snapshot, not a filesystem copy of an active WAL database. Keep matching state +configuration and the prior binary for rollback. Leave disk headroom for the +backup and WAL growth. Then prepare a completed minute range, up to 48 hours: + +```sh +ftw-history-endpoints prepare --state /var/lib/ftw/state.db \ + --plan ./endpoint-plan.json --since 2026-10-19T00:00:00+02:00 \ + --until 2026-10-19T23:15:00+02:00 --max-gap 20m \ + --reason 'Owner requested endpoint mean, issue 1750' +``` + +Inspect `minutes` (flow, whole-minute Wh, endpoint times and powers), `skipped`, +source generation and evidence digest. The boundary timestamps are not raw +polls: their retained resolution is part of the method. Choose the smallest +bound that covers the reviewed gaps. An unsupported edge remains unknown. +Apply the actual reviewed hash and verified independent backup: + +```sh +ftw-history-endpoints apply --state /var/lib/ftw/state.db \ + --plan ./endpoint-plan.json --sha256 ACTUAL_REVIEWED_SHA256 \ + --backup /private/backup/history-before.db +``` + +Apply is safe alongside the live collector. It briefly reserves the SQLite +writer, rechecks historical evidence, inserts one audited metadata record and +invalidates derived daily caches in one transaction. It neither replaces a +live database nor changes raw energy/power/quality rows, device configuration, +control or forecasts. New unrelated measurements survive. An identical retry +is idempotent; tampered, stale or conflicting plans fail without partial writes. + +Plans are stored in the existing `energy_ledger_meta` namespace +`endpoint-repair-v1:`, with backup hash and application time. This +metadata already participates in portable backup/restore. Original minute +energy in affected quarters is protected from rollup so replacements retain +their boundaries. No new database schema or original quality flags are needed. +Home corrections take read precedence over earlier source-power repairs; +the earlier audit remains present. + +Daily totals, system ledger buckets and aggregate charts use the corrections. +Solar has `pv_quality: estimated`; inferred Solar is excluded from the +`measured` fallback. Home use remains `load_quality: estimated`. Current totals +end at the latest committed complete minute, with independent `load_until_ms` +and `pv_until_ms`; the unfinished writer tail is not extrapolated. Aggregate +charts carry estimated/not-for-training markers. Raw samples remain unchanged. + +Verify both totals, flow quality, ledger provenance, writer progress and native +update availability after application. If a client has cached a closed-day +aggregate, reload it. To undo a plan, stop Core briefly or take a SQLite writer +transaction, remove only its exact `energy_ledger_meta` key and invalidate the +three derived day-cache tables. Keep later measurements; do not overwrite them +with an older whole-database backup. diff --git a/go/cmd/ftw-history-endpoints/main.go b/go/cmd/ftw-history-endpoints/main.go new file mode 100644 index 000000000..cba47530c --- /dev/null +++ b/go/cmd/ftw-history-endpoints/main.go @@ -0,0 +1,95 @@ +// ftw-history-endpoints repairs completed historical minutes from bounding +// power means. Preparing and applying leave equipment control to the live Core. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "github.com/srcfl/ftw/go/internal/state" + "os" + "time" +) + +func run(ctx context.Context, args []string) error { + if len(args) == 0 { + return errors.New("use prepare or apply") + } + f := flag.NewFlagSet(args[0], flag.ContinueOnError) + source := f.String("state", "", "existing state.db") + planFile := f.String("plan", "", "reviewable plan JSON") + switch args[0] { + case "prepare": + since := f.String("since", "", "RFC3339 start") + until := f.String("until", "", "RFC3339 completed end") + gap := f.Duration("max-gap", 15*time.Minute, "maximum distance between valid endpoints (1–30m)") + reason := f.String("reason", "", "audit reason") + if err := f.Parse(args[1:]); err != nil { + return err + } + if *source == "" || *planFile == "" || f.NArg() != 0 { + return errors.New("state and new plan path required") + } + from, err := time.Parse(time.RFC3339, *since) + if err != nil { + return err + } + to, err := time.Parse(time.RFC3339, *until) + if err != nil { + return err + } + p, err := state.PrepareEndpointRepair(ctx, *source, state.EndpointRepairRequest{SinceMS: from.UnixMilli(), UntilMS: to.UnixMilli(), MaxGapMS: gap.Milliseconds(), Reason: *reason}) + if err != nil { + return err + } + b, err := json.MarshalIndent(p, "", " ") + if err != nil { + return err + } + file, err := os.OpenFile(*planFile, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return err + } + _, err = file.Write(append(b, '\n')) + closeErr := file.Close() + if err != nil { + return err + } + if closeErr != nil { + return closeErr + } + return json.NewEncoder(os.Stdout).Encode(map[string]any{"plan_sha256": p.SHA256, "correction_minutes": len(p.Minutes), "skipped": p.Skipped}) + case "apply": + sha := f.String("sha256", "", "exact reviewed plan digest") + backup := f.String("backup", "", "independent verified history.db backup") + if err := f.Parse(args[1:]); err != nil { + return err + } + if *source == "" || *planFile == "" || *backup == "" || f.NArg() != 0 { + return errors.New("state, plan and backup required") + } + b, err := os.ReadFile(*planFile) + if err != nil { + return err + } + var p state.EndpointRepairPlan + if err = json.Unmarshal(b, &p); err != nil { + return err + } + n, err := state.ApplyEndpointRepair(ctx, *source, *backup, p, *sha) + if err != nil { + return err + } + return json.NewEncoder(os.Stdout).Encode(map[string]any{"plan_sha256": p.SHA256, "applied_minutes": n, "method": "endpoint_mean_v1"}) + default: + return errors.New("unknown operation") + } +} +func main() { + if err := run(context.Background(), os.Args[1:]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} diff --git a/go/internal/api/api.go b/go/internal/api/api.go index 19593f404..5537628a0 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -2621,6 +2621,15 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { if de.Missing&state.EstimatedLoad != 0 { dayEnergy["load_quality"] = "estimated" } + if de.Missing&state.EndpointEstimatedPV != 0 { + dayEnergy["pv_quality"] = "estimated" + if dayEnergy["history_quality"] == "complete" { + dayEnergy["history_quality"] = "estimated" + } + } + if de.PVUntilMS > 0 { + dayEnergy["pv_until_ms"] = de.PVUntilMS + } if de.LoadUntilMS > 0 { dayEnergy["load_until_ms"] = de.LoadUntilMS } diff --git a/go/internal/api/api_energy_today.go b/go/internal/api/api_energy_today.go index 34625440e..0149b5a66 100644 --- a/go/internal/api/api_energy_today.go +++ b/go/internal/api/api_energy_today.go @@ -13,6 +13,15 @@ func energyTodayFields(d state.DayEnergy) map[string]any { if d.LoadUntilMS > 0 { out["load_until_ms"] = d.LoadUntilMS } + if d.Missing&state.EndpointEstimatedPV != 0 { + out["pv_quality"] = "estimated" + if out["history_quality"] == "complete" { + out["history_quality"] = "estimated" + } + } + if d.PVUntilMS > 0 { + out["pv_until_ms"] = d.PVUntilMS + } measured := make(map[string]any) for key, field := range map[string]struct { flag state.HistoryMissing @@ -34,6 +43,9 @@ func energyTodayFields(d state.DayEnergy) map[string]any { // in the measured-only fallback used by clients after a data gap. measured["load_wh"] = nil } + if d.Missing&state.EndpointEstimatedPV != 0 { + measured["pv_wh"] = nil + } out["measured"] = measured return out } diff --git a/go/internal/api/api_history_repair_test.go b/go/internal/api/api_history_repair_test.go index 311418c1d..f76cd5467 100644 --- a/go/internal/api/api_history_repair_test.go +++ b/go/internal/api/api_history_repair_test.go @@ -17,3 +17,13 @@ func TestHistoricalRepairEnergyFieldsPreserveOtherUnknownFlows(t *testing.T) { t.Fatal("repair became a measurement", got) } } + +func TestEndpointSolarEnergyFieldsAreEstimatedNotMeasured(t *testing.T) { + got := energyTodayFields(state.DayEnergy{Missing: state.EndpointEstimatedPV, Intervals: 10, PVWh: 1234, PVUntilMS: 1800000}) + if got["pv_wh"] != float64(1234) || got["pv_quality"] != "estimated" || got["pv_until_ms"] != int64(1800000) || got["history_quality"] != "estimated" { + t.Fatal(got) + } + if got["measured"].(map[string]any)["pv_wh"] != nil { + t.Fatal("estimated solar labelled measured", got) + } +} diff --git a/go/internal/state/history_dashboard_retention.go b/go/internal/state/history_dashboard_retention.go index 7e535435e..07de225e5 100644 --- a/go/internal/state/history_dashboard_retention.go +++ b/go/internal/state/history_dashboard_retention.go @@ -198,6 +198,7 @@ func (s *Store) rollSiteEnergyStorage(ctx context.Context, cutoff int64, table, // Keep that quarter's originals rather than merging away the boundary // needed to subtract/replace only the corrected household contribution. protected := ` AND NOT EXISTS (SELECT 1 FROM history_load_estimates r WHERE r.method GLOB 'historical_power_repair_v1:*' AND r.start_ms>=` + table + `.start_ms-(` + table + `.start_ms%900000) AND r.start_ms<` + table + `.start_ms-(` + table + `.start_ms%900000)+900000)` + protected += ` AND NOT EXISTS (SELECT 1 FROM energy_ledger_meta a, json_each(CASE WHEN a.key GLOB 'endpoint-repair-v1:*' THEN a.value ELSE '{}' END,'$.plan.minutes') m WHERE a.key GLOB 'endpoint-repair-v1:*' AND json_extract(m.value,'$.start_ms')>=` + table + `.start_ms-(` + table + `.start_ms%900000) AND json_extract(m.value,'$.start_ms')<` + table + `.start_ms-(` + table + `.start_ms%900000)+900000)` if err := s.history.QueryRowContext(ctx, `SELECT MIN(start_ms) FROM `+table+` WHERE resolution_ms=? AND start_ms=? AND start_ms 0 { + end = min(end, bucketStart(last, 60000)) + } + return end, nil +} +func (s *Store) endpointPVBuckets(ctx context.Context, from, to, width int64) ([]sourceLoadBucket, map[int64]EndpointRepairMinute, error) { + all, err := s.endpointRepairs(ctx, from, to) + if err != nil { + return nil, nil, err + } + repairs := all["pv"] + if len(repairs) == 0 { + return nil, repairs, nil + } + by := map[int64]*sourceLoadBucket{} + add := func(r siteEnergyRow) error { + w := max(width, r.width) + start := from + (r.start-from)/w*w + b := by[start] + if b == nil { + b = &sourceLoadBucket{start: start, width: w, first: r.first, last: r.last} + by[start] = b + } + if b.width != w || r.first < start || r.last > start+w { + b.missing |= MissingPV + } + b.first = min(b.first, r.first) + b.last = max(b.last, r.last) + b.covered += r.covered + b.n += r.n + b.wh += r.pv + b.missing |= r.missing & MissingPV + return nil + } + err = s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { + if _, ok := repairs[r.start]; ok && r.width == 60000 { + return nil + } + return add(r) + }) + if err != nil { + return nil, nil, err + } + for at, p := range repairs { + _ = add(siteEnergyRow{start: at, width: 60000, first: at, last: at + 60000, covered: 60000, pv: p.Wh}) + } + var out []sourceLoadBucket + for _, b := range by { + if b.covered != b.width || b.first != b.start || b.last != b.start+b.width || b.wh < 0 || !finiteHistoryValue(b.wh) { + b.missing |= MissingPV + } + out = append(out, *b) + } + sort.Slice(out, func(i, j int) bool { return out[i].start < out[j].start }) + return out, repairs, nil +} +func (s *Store) applyEndpointDailyPV(ctx context.Context, from, to int64, d *DayEnergy) error { + end, err := s.endpointReadUntil(ctx, from, to) + if err != nil || end <= from { + return err + } + b, repairs, err := s.endpointPVBuckets(ctx, from, end, end-from) + if err != nil || len(repairs) == 0 { + return err + } + if len(b) == 1 && b[0].missing&MissingPV == 0 && b[0].covered == end-from { + d.PVWh = b[0].wh + d.Missing &^= MissingPV + d.Missing |= EndpointEstimatedPV + d.PVUntilMS = end + } + return nil +} +func (s *Store) applyHistoricalLoadChart(ctx context.Context, points []HistoryPoint, from, to, width int64) ([]HistoryPoint, error) { + out, err := s.applyHistoricalLoadChartOnly(ctx, points, from, to, width) + if err != nil || width < 60000 { + return out, err + } + all, err := s.endpointRepairs(ctx, from, to) + if err != nil { + return nil, err + } + changedLoad := map[int64]bool{} + for at := range all["load"] { + changedLoad[from+(at-from)/width*width] = true + } + for i := range out { + p := &out[i] + at := from + (p.TsMs-from)/width*width + if !changedLoad[at] || p.Missing&MissingLoad != 0 { + continue + } + var detail map[string]any + _ = json.Unmarshal([]byte(p.JSON), &detail) + if detail == nil { + detail = map[string]any{} + } + detail["history_load_method"] = "endpoint_mean_v1" + detail["load_quality"] = "estimated" + detail["forecast_measurement_quality"] = "aggregate_not_for_training" + raw, _ := json.Marshal(detail) + p.JSON = string(raw) + } + buckets, repairs, err := s.endpointPVBuckets(ctx, from, to, width) + if err != nil || len(repairs) == 0 { + return out, err + } + changed := map[int64]bool{} + for at := range repairs { + changed[from+(at-from)/width*width] = true + } + by := map[int64]sourceLoadBucket{} + for _, b := range buckets { + by[b.start] = b + } + seen := map[int64]bool{} + for i := range out { + p := &out[i] + at := from + (p.TsMs-from)/width*width + b, ok := by[at] + if !ok || !changed[at] || b.missing&MissingPV != 0 { + continue + } + p.PVW = -b.wh * 3600000 / float64(b.width) + p.Missing &^= MissingPV + p.Missing |= EstimatedLoad + seen[at] = true + var detail map[string]any + _ = json.Unmarshal([]byte(p.JSON), &detail) + if detail == nil { + detail = map[string]any{} + } + detail["pv_quality"] = "estimated" + detail["history_pv_method"] = "endpoint_mean_v1" + detail["forecast_measurement_quality"] = "aggregate_not_for_training" + raw, _ := json.Marshal(detail) + p.JSON = string(raw) + } + for _, b := range buckets { + if !changed[b.start] || seen[b.start] || b.missing&MissingPV != 0 { + continue + } + out = append(out, HistoryPoint{TsMs: b.start + b.width - 1, FirstMS: b.start, ResolutionMS: b.width, PVW: -b.wh * 3600000 / float64(b.width), Missing: MissingGrid | MissingBattery | MissingEV | MissingV2X | MissingSoC | MissingLoad | EstimatedLoad, JSON: `{"pv_quality":"estimated","history_pv_method":"endpoint_mean_v1","forecast_measurement_quality":"aggregate_not_for_training"}`}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].TsMs < out[j].TsMs }) + return out, nil +} +func (s *Store) applyHistoricalLoadLedger(ctx context.Context, q EnergyHistoryQuery, points []EnergyLedgerPoint) ([]EnergyLedgerPoint, error) { + out, err := s.applyHistoricalLoadLedgerOnly(ctx, q, points) + if err != nil { + return nil, err + } + all, err := s.endpointRepairs(ctx, q.SinceMS, q.UntilMS) + if err != nil { + return nil, err + } + changedLoad := map[int64]bool{} + for at := range all["load"] { + changedLoad[q.SinceMS+(at-q.SinceMS)/q.BucketMS*q.BucketMS] = true + } + for i := range out { + if out[i].Flow == FlowConsumerUse && changedLoad[out[i].BucketStartMS] { + out[i].Provenance = "endpoint_mean_v1" + } + } + if q.AssetID != "" { + return out, nil + } + buckets, repairs, err := s.endpointPVBuckets(ctx, q.SinceMS, q.UntilMS, q.BucketMS) + if err != nil || len(repairs) == 0 { + return out, err + } + changed := map[int64]bool{} + for at := range repairs { + changed[q.SinceMS+(at-q.SinceMS)/q.BucketMS*q.BucketMS] = true + } + filtered := out[:0] + for _, p := range out { + if p.Flow != FlowPVGeneration || !changed[p.BucketStartMS] { + filtered = append(filtered, p) + } + } + out = filtered + for _, b := range buckets { + if changed[b.start] && b.missing&MissingPV == 0 { + out = append(out, EnergyLedgerPoint{SchemaVersion: EnergyLedgerSchemaVersion, AssetID: "system", Flow: FlowPVGeneration, BucketStartMS: b.start, BucketLenMS: b.width, EnergyWh: b.wh, Source: "historical_repair", Quality: "estimated", Provenance: "endpoint_mean_v1", SampleCount: b.n}) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].BucketStartMS < out[j].BucketStartMS }) + return out, nil +} diff --git a/go/internal/state/history_endpoint_repair.go b/go/internal/state/history_endpoint_repair.go new file mode 100644 index 000000000..cba154a52 --- /dev/null +++ b/go/internal/state/history_endpoint_repair.go @@ -0,0 +1,376 @@ +package state + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "time" +) + +const endpointRepairPrefix = "endpoint-repair-v1:" + +type EndpointRepairRequest struct { + SinceMS int64 `json:"since_ms"` + UntilMS int64 `json:"until_ms"` + MaxGapMS int64 `json:"max_gap_ms"` + Reason string `json:"reason"` +} +type EndpointRepairMinute struct { + Flow string `json:"flow"` + StartMS int64 `json:"start_ms"` + Wh float64 `json:"wh"` + BeforeMS int64 `json:"before_ms"` + AfterMS int64 `json:"after_ms"` + BeforeW float64 `json:"before_w"` + AfterW float64 `json:"after_w"` +} +type EndpointRepairPlan struct { + SHA256 string `json:"sha256"` + EvidenceSHA256 string `json:"evidence_sha256"` + Generation string `json:"generation"` + Request EndpointRepairRequest `json:"request"` + Minutes []EndpointRepairMinute `json:"minutes"` + Skipped map[string]string `json:"skipped"` +} +type endpointApplied struct { + Plan EndpointRepairPlan `json:"plan"` + BackupSHA256 string `json:"backup_sha256"` + AppliedAt string `json:"applied_at"` +} + +func endpointDigest(v any) string { + b, _ := json.Marshal(v) + h := sha256.Sum256(b) + return hex.EncodeToString(h[:]) +} +func (p EndpointRepairPlan) digest() string { p.SHA256 = ""; return endpointDigest(p) } +func (r EndpointRepairRequest) validate() error { + if r.SinceMS%60000 != 0 || r.UntilMS%60000 != 0 || r.UntilMS <= r.SinceMS || r.UntilMS-r.SinceMS > 48*3600000 || r.UntilMS > time.Now().Add(-time.Minute).UnixMilli() || r.MaxGapMS < 60000 || r.MaxGapMS > 30*60000 || strings.TrimSpace(r.Reason) == "" { + return errors.New("choose a completed minute range up to 48h, max-gap 1–30m and a reason") + } + return nil +} + +type endpointOriginal struct { + Start int64 + Covered int64 + First int64 + Last int64 + Load float64 + PV float64 + Missing HistoryMissing +} + +func (s *Store) endpointOriginals(ctx context.Context, from, to int64) ([]endpointOriginal, error) { + by := map[int64]*endpointOriginal{} + err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { + if r.width != 60000 { + return errors.New("endpoint repair requires retained minute energy") + } + b := by[r.start] + if b == nil { + b = &endpointOriginal{Start: r.start, First: r.first} + by[r.start] = b + } + b.Covered += r.covered + b.First = min(b.First, r.first) + b.Last = max(b.Last, r.last) + b.Load += r.load + b.PV += r.pv + b.Missing |= r.missing + return nil + }) + if err != nil { + return nil, err + } + out := make([]endpointOriginal, 0, (to-from)/60000) + for at := from; at < to; at += 60000 { + if b := by[at]; b != nil { + out = append(out, *b) + } else { + out = append(out, endpointOriginal{Start: at, Missing: MissingPower}) + } + } + return out, nil +} +func endpointOriginalValid(b endpointOriginal, flow string) bool { + if b.Covered != 60000 || b.First != b.Start || b.Last != b.Start+60000 { + return false + } + if flow == "pv" { + return b.Missing&MissingPV == 0 && finiteHistoryValue(b.PV) && b.PV >= 0 + } + return b.Missing&(MissingPower|BalanceMismatch) == 0 && finiteHistoryValue(b.Load) && b.Load >= 0 +} + +type endpointPower struct { + At int64 + W float64 +} + +func (s *Store) endpointPowers(ctx context.Context, from, to int64, flow string) ([]endpointPower, error) { + if flow == "load" { + originals, err := s.endpointOriginals(ctx, bucketStart(from, 60000), bucketStart(to, 60000)+60000) + if err != nil { + return nil, err + } + var out []endpointPower + for _, b := range originals { + if endpointOriginalValid(b, "load") { + out = append(out, endpointPower{At: b.Start + 30000, W: b.Load * 60}) + } + } + return out, nil + } + column, mask := "load_w", MissingPower|BalanceMismatch + if flow == "pv" { + column = "-pv_w" + mask = MissingPV + } + // Retained ten-second means are explicit endpoints, not invented raw polls. + rows, err := s.history.QueryContext(ctx, `SELECT e.last_ms,e.`+strings.TrimPrefix(column, "-")+`,COALESCE(q.missing,0) FROM ( + SELECT *, 'dashboard' kind FROM history_dashboard UNION ALL SELECT *, 'partial_dashboard' kind FROM history_dashboard_partial + ) e LEFT JOIN history_quality q ON q.kind=e.kind AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms + WHERE e.resolution_ms=10000 AND e.last_ms>=? AND e.first_ms<=? ORDER BY e.last_ms`, from, to) + if err != nil { + return nil, err + } + defer rows.Close() + var out []endpointPower + for rows.Next() { + var p endpointPower + var missing HistoryMissing + if err := rows.Scan(&p.At, &p.W, &missing); err != nil { + return nil, err + } + if flow == "pv" { + p.W = -p.W + } + if missing&mask == 0 && finiteHistoryValue(p.W) && p.W >= 0 { + out = append(out, p) + } + } + return out, rows.Err() +} +func (s *Store) planEndpointRepair(ctx context.Context, r EndpointRepairRequest) (EndpointRepairPlan, error) { + p := EndpointRepairPlan{Request: r, Skipped: map[string]string{}} + if err := r.validate(); err != nil { + return p, err + } + if err := s.history.QueryRowContext(ctx, `SELECT name FROM history_migrations WHERE name LIKE 'generation:%'`).Scan(&p.Generation); err != nil { + return p, err + } + original, err := s.endpointOriginals(ctx, r.SinceMS, r.UntilMS) + if err != nil { + return p, err + } + p.EvidenceSHA256 = endpointDigest(original) + for _, flow := range []string{"load", "pv"} { + powers, err := s.endpointPowers(ctx, r.SinceMS-r.MaxGapMS, r.UntilMS+r.MaxGapMS, flow) + if err != nil { + return p, err + } + for i := 0; i < len(original); { + if endpointOriginalValid(original[i], flow) { + i++ + continue + } + start := i + for i < len(original) && !endpointOriginalValid(original[i], flow) { + i++ + } + from, to := original[start].Start, original[i-1].Start+60000 + before := sort.Search(len(powers), func(j int) bool { return powers[j].At > from }) - 1 + after := sort.Search(len(powers), func(j int) bool { return powers[j].At >= to }) + key := fmt.Sprintf("%s:%d:%d", flow, from, to) + if before < 0 || after == len(powers) { + p.Skipped[key] = "missing valid bounding power" + continue + } + a, b := powers[before], powers[after] + if b.At <= a.At || b.At-a.At > r.MaxGapMS { + p.Skipped[key] = "bounding observations exceed max-gap" + continue + } + wh := (a.W + b.W) / 2 / 60 + if !finiteHistoryValue(wh) || wh < 0 || !plausibleEnergy(wh, 60000) { + p.Skipped[key] = "implausible endpoint estimate" + continue + } + for at := from; at < to; at += 60000 { + p.Minutes = append(p.Minutes, EndpointRepairMinute{flow, at, wh, a.At, b.At, a.W, b.W}) + } + } + } + p.SHA256 = p.digest() + return p, nil +} + +// PrepareEndpointRepair reads the live box without migrations or a long copy. +// Only completed evidence is considered. It never starts equipment control. +func PrepareEndpointRepair(ctx context.Context, statePath string, r EndpointRepairRequest) (EndpointRepairPlan, error) { + s, err := OpenBackupSource(statePath) + if err != nil { + return EndpointRepairPlan{}, err + } + defer s.Close() + if s.history == nil { + return EndpointRepairPlan{}, errors.New("primary SQLite history required") + } + return s.planEndpointRepair(ctx, r) +} +func (s *Store) endpointRepairs(ctx context.Context, from, to int64) (map[string]map[int64]EndpointRepairMinute, error) { + out := map[string]map[int64]EndpointRepairMinute{"load": {}, "pv": {}} + if s.history == nil { + return out, nil + } + rows, err := s.history.QueryContext(ctx, `SELECT value FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*' ORDER BY key`) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var raw string + var a endpointApplied + if err := rows.Scan(&raw); err != nil { + return nil, err + } + if err := json.Unmarshal([]byte(raw), &a); err != nil { + return nil, err + } + if a.Plan.SHA256 != a.Plan.digest() || len(a.BackupSHA256) != 64 { + return nil, errors.New("invalid endpoint repair audit") + } + for _, m := range a.Plan.Minutes { + if m.StartMS < from || m.StartMS+60000 > to { + continue + } + if out[m.Flow] == nil || m.StartMS%60000 != 0 || !finiteHistoryValue(m.Wh) || m.Wh < 0 { + return nil, errors.New("invalid endpoint correction") + } + if old, ok := out[m.Flow][m.StartMS]; ok && old != m { + return nil, errors.New("conflicting endpoint corrections") + } + out[m.Flow][m.StartMS] = m + } + } + return out, rows.Err() +} + +// ApplyEndpointRepair writes only audited metadata and derived caches, in one +// short SQLite transaction. It never replaces a live database with a snapshot. +// The independently verified backup and exact plan hash are mandatory. +func ApplyEndpointRepair(ctx context.Context, statePath, backupPath string, p EndpointRepairPlan, sha string) (int, error) { + if len(sha) != 64 || p.SHA256 != sha || p.digest() != sha { + return 0, errors.New("exact reviewed plan hash required") + } + if err := p.Request.validate(); err != nil { + return 0, err + } + if len(p.Minutes) == 0 { + return 0, errors.New("no supported corrections") + } + source := historyDatabasePath(statePath) + src, err := filepath.EvalSymlinks(source) + if err != nil { + return 0, err + } + bak, err := filepath.EvalSymlinks(backupPath) + if err != nil { + return 0, err + } + srcInfo, err := os.Stat(src) + if err != nil { + return 0, err + } + bakInfo, err := os.Stat(bak) + if err != nil { + return 0, err + } + if os.SameFile(srcInfo, bakInfo) { + return 0, errors.New("backup must be independent") + } + backupSHA, err := repairFileHash(bak) + if err != nil { + return 0, err + } + backup, err := sql.Open("sqlite", ReadOnlyDatabaseURI(bak)) + if err != nil { + return 0, err + } + var generation string + err = backup.QueryRowContext(ctx, `SELECT name FROM history_migrations WHERE name=?`, p.Generation).Scan(&generation) + backup.Close() + if err != nil { + return 0, errors.New("backup generation differs") + } + s, err := OpenBackupSource(statePath) + if err != nil { + return 0, err + } + defer s.Close() + db, err := sql.Open("sqlite", source+"?_pragma=busy_timeout(5000)&_pragma=synchronous(FULL)") + if err != nil { + return 0, err + } + defer db.Close() + conn, err := db.Conn(ctx) + if err != nil { + return 0, err + } + defer conn.Close() + if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil { + return 0, err + } + defer conn.ExecContext(context.Background(), "ROLLBACK") + // Holding the write reservation keeps Core from changing the evidence between + // review verification and publication, while WAL readers remain available. + current, err := s.planEndpointRepair(ctx, p.Request) + if err != nil { + return 0, err + } + if current.SHA256 != sha { + return 0, errors.New("historical evidence changed; prepare a new plan") + } + existing, err := s.endpointRepairs(ctx, p.Request.SinceMS, p.Request.UntilMS) + if err != nil { + return 0, err + } + for _, m := range p.Minutes { + if old, ok := existing[m.Flow][m.StartMS]; ok && old != m { + return 0, errors.New("overlapping plan conflicts") + } + } + var saved string + err = conn.QueryRowContext(ctx, `SELECT value FROM energy_ledger_meta WHERE key=?`, endpointRepairPrefix+sha).Scan(&saved) + if err == nil { + return len(p.Minutes), nil + } + if !errors.Is(err, sql.ErrNoRows) { + return 0, err + } + raw, err := json.Marshal(endpointApplied{p, backupSHA, time.Now().UTC().Format(time.RFC3339Nano)}) + if err != nil { + return 0, err + } + if _, err = conn.ExecContext(ctx, `INSERT INTO energy_ledger_meta(key,value) VALUES(?,?)`, endpointRepairPrefix+sha, string(raw)); err != nil { + return 0, err + } + for _, table := range []string{"energy_daily", "energy_daily_partial", "history_daily_quality"} { + if _, err = conn.ExecContext(ctx, `DELETE FROM `+table); err != nil { + return 0, err + } + } + if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil { + return 0, err + } + return len(p.Minutes), nil +} diff --git a/go/internal/state/history_endpoint_repair_test.go b/go/internal/state/history_endpoint_repair_test.go new file mode 100644 index 000000000..8c7bd7348 --- /dev/null +++ b/go/internal/state/history_endpoint_repair_test.go @@ -0,0 +1,263 @@ +package state + +import ( + "context" + "encoding/json" + "fmt" + "path/filepath" + "testing" + "time" +) + +func endpointFixture(t *testing.T, night bool) (*Store, EndpointRepairRequest, string) { + t.Helper() + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour).UnixMilli() + for at := base - 60000; at <= base+31*60000; at += 5000 { + if at >= base+10*60000 && at < base+22*60000 { + continue + } + load, pv := 2000.0, 1000.0 + if at >= base+22*60000 { + load, pv = 4000, 3000 + } + if night { + pv = 0 + } + v := HistorySources{AtMS: at, Topology: "grid+pv", GridW: load - pv, PVW: -pv} + p := HistoryPoint{TsMs: at, GridW: v.GridW, PVW: v.PVW, LoadW: load, JSON: `{"ev_w":0,"v2x_w":0}`, Sources: &v} + dashboardTick(t, s, fmt.Sprint(at), &p) + } + r := EndpointRepairRequest{base, base + 30*60000, 15 * 60000, "owner-authorized endpoint mean"} + backup := filepath.Join(t.TempDir(), "history-before.db") + if _, err := s.history.Exec(`VACUUM INTO ?`, backup); err != nil { + t.Fatal(err) + } + return s, r, backup +} +func TestEndpointRepairMeansBothFlowsWhileLiveAndPreservesOriginals(t *testing.T) { + ctx := context.Background() + s, r, backup := endpointFixture(t, false) + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if len(p.Minutes) == 0 || len(p.Skipped) != 0 { + t.Fatal(p) + } + original, err := s.endpointOriginals(ctx, r.SinceMS, r.UntilMS) + if err != nil { + t.Fatal(err) + } + beforeDigest := endpointDigest(original) + // Seed the independent measured PV ledger; the repair must preserve buckets + // outside its affected windows and replace whole affected system buckets. + pvLedger := map[int64]float64{} + for _, b := range original { + pvLedger[bucketStart(b.Start, 300000)] += b.PV + } + for at, wh := range pvLedger { + insertLedgerEntryTest(t, s, "pv-device", FlowPVGeneration, at, 300000, wh, "hardware_counter", "measured", "counter", 1) + } + for _, m := range p.Minutes { + want := 50.0 + if m.Flow == "pv" { + want = 2000.0 / 60 + } + near(t, "endpoint formula", m.Wh, want) + if m.AfterMS-m.BeforeMS > r.MaxGapMS { + t.Fatal("bound ignored", m) + } + } + // Simulate new receipts arriving after review: they must survive publication. + next := r.UntilMS + 120000 + sourceTick(t, s, next, HistorySources{GridW: 3210}) + var through string + if err := s.history.QueryRow(`SELECT value FROM history_source_cursor WHERE key='through'`).Scan(&through); err != nil { + t.Fatal(err) + } + n, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256) + if err != nil || n != len(p.Minutes) { + t.Fatal(n, err) + } + again, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256) + if err != nil || again != n { + t.Fatal("retry", again, err) + } + afterOriginal, _ := s.endpointOriginals(ctx, r.SinceMS, r.UntilMS) + if endpointDigest(afterOriginal) != beforeDigest { + t.Fatal("original energy/quality changed") + } + var afterThrough string + _ = s.history.QueryRow(`SELECT value FROM history_source_cursor WHERE key='through'`).Scan(&afterThrough) + if through != afterThrough { + t.Fatal("new data lost") + } + var plans int + _ = s.history.QueryRow(`SELECT count(*) FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*'`).Scan(&plans) + if plans != 1 { + t.Fatal("double counted", plans) + } + d, err := s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&(MissingLoad|MissingPV) != 0 || d.Missing&EndpointEstimatedPV == 0 || d.Missing&EstimatedLoad == 0 { + t.Fatal("daily", d, err) + } + expected := map[string]float64{"load": 0, "pv": 0} + corrections := map[string]map[int64]float64{"load": {}, "pv": {}} + for _, m := range p.Minutes { + corrections[m.Flow][m.StartMS] = m.Wh + } + for _, b := range original { + for _, flow := range []string{"load", "pv"} { + if wh, ok := corrections[flow][b.Start]; ok { + expected[flow] += wh + } else if flow == "load" { + expected[flow] += b.Load + } else { + expected[flow] += b.PV + } + } + } + near(t, "home total", d.LoadWh, expected["load"]) + near(t, "solar total", d.PVWh, expected["pv"]) + ledger, truncated, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: r.SinceMS, UntilMS: r.UntilMS, BucketMS: 300000, Limit: 20}) + if err != nil || truncated { + t.Fatal(err) + } + sums := map[EnergyFlow]float64{} + for _, v := range ledger { + if v.Flow == FlowConsumerUse || v.Flow == FlowPVGeneration { + sums[v.Flow] += v.EnergyWh + if v.Source == "historical_repair" && v.Provenance != "endpoint_mean_v1" { + t.Fatal(v) + } + } + } + near(t, "ledger home", sums[FlowConsumerUse], d.LoadWh) + near(t, "ledger solar", sums[FlowPVGeneration], d.PVWh) + points, err := s.LoadHistory(r.SinceMS, r.UntilMS, 30) + if err != nil { + t.Fatal(err) + } + var marked bool + for _, v := range points { + var detail map[string]any + _ = json.Unmarshal([]byte(v.JSON), &detail) + if detail["pv_quality"] == "estimated" { + marked = true + if v.Missing&EstimatedLoad == 0 { + t.Fatal("training mark missing") + } + } + } + if !marked { + t.Fatal("chart correction missing") + } + if err := s.SaveDailyEnergy("endpoint-cache", d); err != nil { + t.Fatal(err) + } + cached, hit, err := s.LoadDailyEnergy("endpoint-cache") + if err != nil || !hit || cached.Missing&EndpointEstimatedPV == 0 { + t.Fatal("solar estimate marker lost in cache", cached, hit, err) + } + // Retention must preserve the minute boundaries used to replace each flow. + if err := s.rollSiteEnergy(ctx, r.UntilMS+900000); err != nil { + t.Fatal(err) + } + retained, err := s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil { + t.Fatal(err) + } + near(t, "retained solar", retained.PVWh, d.PVWh) + near(t, "retained home", retained.LoadWh, d.LoadWh) + exported := filepath.Join(t.TempDir(), "restored.db") + if err := s.copyStateForBackup(ctx, exported); err != nil { + t.Fatal(err) + } + if err := s.exportHistoryToSQLite(ctx, exported); err != nil { + t.Fatal(err) + } + restored, err := Open(exported) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + got, err := restored.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || got.Missing&(MissingLoad|MissingPV) != 0 { + t.Fatal(got, err) + } + near(t, "restored home", got.LoadWh, d.LoadWh) + near(t, "restored solar", got.PVWh, d.PVWh) +} +func TestEndpointRepairNightZeroAndBounds(t *testing.T) { + s, r, backup := endpointFixture(t, true) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + for _, m := range p.Minutes { + if m.Flow == "pv" && m.Wh != 0 { + t.Fatal("night must use observed zero endpoints", m) + } + } + bad := p + bad.Minutes = append([]EndpointRepairMinute(nil), p.Minutes...) + bad.Minutes[0].Wh++ + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, bad, p.SHA256); err == nil { + t.Fatal("tampering accepted") + } + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, s.historyPath, p, p.SHA256); err == nil { + t.Fatal("source accepted as backup") + } + r.MaxGapMS = 5 * 60000 + p, err = PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil || len(p.Minutes) != 0 || len(p.Skipped) != 2 { + t.Fatal("long gap was filled", p, err) + } +} +func TestEndpointRepairRejectsChangedEvidence(t *testing.T) { + s, r, backup := endpointFixture(t, false) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if _, err = s.history.Exec(`UPDATE history_source_energy SET import_wh=import_wh+1 WHERE start_ms=?`, r.SinceMS); err != nil { + t.Fatal(err) + } + if _, err = ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256); err == nil { + t.Fatal("stale evidence accepted") + } + var n int + _ = s.history.QueryRow(`SELECT count(*) FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*'`).Scan(&n) + if n != 0 { + t.Fatal("partial application") + } +} + +func TestEndpointRepairRollsBackWhenCacheInvalidationFails(t *testing.T) { + s, r, backup := endpointFixture(t, false) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if err := s.SaveDailyEnergy("guard", DayEnergy{Intervals: 1}); err != nil { + t.Fatal(err) + } + if _, err := s.history.Exec(`CREATE TRIGGER endpoint_cache_guard BEFORE DELETE ON energy_daily_partial BEGIN SELECT RAISE(ABORT,'cache locked for test'); END`); err != nil { + t.Fatal(err) + } + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256); err == nil { + t.Fatal("expected transaction failure") + } + var n int + _ = s.history.QueryRow(`SELECT count(*) FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*'`).Scan(&n) + if n != 0 { + t.Fatal("audit partially committed") + } + if _, hit, err := s.LoadDailyEnergy("guard"); err != nil || !hit { + t.Fatal("cache partially deleted", hit, err) + } +} diff --git a/go/internal/state/history_repair_read.go b/go/internal/state/history_repair_read.go index 84966c04d..85917c111 100644 --- a/go/internal/state/history_repair_read.go +++ b/go/internal/state/history_repair_read.go @@ -41,7 +41,18 @@ func (s *Store) historicalLoadRepairs(ctx context.Context, from, to int64) (map[ } out[p.StartMS] = p } - return out, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + rows.Close() + endpoints, err := s.endpointRepairs(ctx, from, to) + if err != nil { + return nil, err + } + for at, m := range endpoints["load"] { + out[at] = repairMinute{StartMS: at, Wh: m.Wh, InterpolatedMS: m.AfterMS - m.BeforeMS} + } + return out, nil } // walkHouseholdLoad replaces only derived household energy in the read view. @@ -119,7 +130,10 @@ func (s *Store) repairedLoadBuckets(ctx context.Context, from, to, width int64) func (s *Store) applyHistoricalDailyLoad(ctx context.Context, from, to int64, d *DayEnergy) error { // A historical correction cannot invent the currently unfinished minute. - end := bucketStart(to, ArchiveResolutionMS) + end, err := s.endpointReadUntil(ctx, from, to) + if err != nil { + return err + } if end <= from { return nil } @@ -136,7 +150,7 @@ func (s *Store) applyHistoricalDailyLoad(ctx context.Context, from, to int64, d return nil } -func (s *Store) applyHistoricalLoadChart(ctx context.Context, points []HistoryPoint, from, to, width int64) ([]HistoryPoint, error) { +func (s *Store) applyHistoricalLoadChartOnly(ctx context.Context, points []HistoryPoint, from, to, width int64) ([]HistoryPoint, error) { if width < ArchiveResolutionMS { return points, nil } // Do not invent raw poll values. @@ -192,7 +206,7 @@ func (s *Store) applyHistoricalLoadChart(ctx context.Context, points []HistoryPo return points, nil } -func (s *Store) applyHistoricalLoadLedger(ctx context.Context, q EnergyHistoryQuery, points []EnergyLedgerPoint) ([]EnergyLedgerPoint, error) { +func (s *Store) applyHistoricalLoadLedgerOnly(ctx context.Context, q EnergyHistoryQuery, points []EnergyLedgerPoint) ([]EnergyLedgerPoint, error) { if q.AssetID != "" && q.AssetID != ObservedConsumerAssetID { return points, nil } diff --git a/go/internal/state/store.go b/go/internal/state/store.go index 6e1733d50..0dd3f9a8d 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -1645,6 +1645,7 @@ type DayEnergy struct { // Historical repairs use completed minute evidence, not an extrapolated // current tail. Status exposes this bound independently of live power. LoadUntilMS int64 + PVUntilMS int64 // Intervals is the number of integration intervals that contributed to // the totals — i.e. the count of rows that had a predecessor (prev_ts IS // NOT NULL). Zero means there was at most one history row in the range, so @@ -1693,6 +1694,9 @@ func (s *Store) DailyEnergy(sinceMs, untilMs int64) (DayEnergy, error) { d, err := s.legacyDailyEnergy(sinceMs, untilMs) if err == nil { err = s.applyHistoricalDailyLoad(ctx, sinceMs, untilMs, &d) + if err == nil { + err = s.applyEndpointDailyPV(ctx, sinceMs, untilMs, &d) + } } return d, err } @@ -1707,6 +1711,9 @@ func (s *Store) DailyEnergy(sinceMs, untilMs int64) (DayEnergy, error) { } out = addDayEnergy(old, out) } + if err := s.applyEndpointDailyPV(ctx, sinceMs, untilMs, &out); err != nil { + return out, err + } if err := s.applyHistoricalDailyLoad(ctx, sinceMs, untilMs, &out); err != nil { return out, err } diff --git a/web/energy-today.js b/web/energy-today.js index 71c8b3f37..1de2322b5 100644 --- a/web/energy-today.js +++ b/web/energy-today.js @@ -16,7 +16,7 @@ function knownWh(value) { export function energyTodayReading(today, key) { const total = today?.[key]; - if (knownWh(total)) return key === "load_wh" && (today?.history_quality === "estimated" || today?.load_quality === "estimated") + if (knownWh(total)) return ((key === "load_wh" && (today?.history_quality === "estimated" || today?.load_quality === "estimated")) || (key === "pv_wh" && today?.pv_quality === "estimated")) ? {wh: total, partial: false, estimated: true} : {wh: total, partial: false}; const measured = today?.measured?.[key]; if (knownWh(measured)) return {wh: measured, partial: true}; @@ -66,6 +66,7 @@ export function updateEnergyToday(today, root = document) { ? "Gaps in today's readings. ≥ shows recorded energy only." : "Today's energy is not available yet."; if (estimated) note.textContent = (incomplete ? note.textContent + " " : "") + "Home use includes estimates from source energy."; + if (today?.pv_quality === "estimated") {note.hidden = false; note.textContent += " Solar includes estimated energy across historical gaps.";} const home = energyTodayReading(today, "load_wh"); if (home.estimated && home.partial) { note.hidden = false; diff --git a/web/history-repair-quality.test.mjs b/web/history-repair-quality.test.mjs index d6f918c23..be41ec5c6 100644 --- a/web/history-repair-quality.test.mjs +++ b/web/history-repair-quality.test.mjs @@ -10,3 +10,9 @@ test("corrected household energy stays estimated even when another original flow assert.equal(flowReadingsFromStatus({load_w:1200,energy:{today}}).selfPoweredPctToday,null); assert.equal(today.measured.load_wh,null); }); + +test("endpoint solar and home totals both show estimate markers",()=>{ + const today={load_wh:41000,load_quality:"estimated",pv_wh:37000,pv_quality:"estimated",history_quality:"incomplete",measured:{load_wh:null,pv_wh:null}}; + assert.equal(formatEnergyToday(today,"load_wh"),"≈ 41.0 kWh"); + assert.equal(formatEnergyToday(today,"pv_wh"),"≈ 37.0 kWh"); +}); From 6554c6707256ea720a3fffe1ae5a7ea2f15d255f Mon Sep 17 00:00:00 2001 From: Segran Date: Mon, 19 Oct 2026 23:22:26 +0200 Subject: [PATCH 5/5] test(history): verify repaired live totals exclude unfinished writer tail Signed-off-by: Segran --- go/internal/state/history_endpoint_read.go | 2 +- .../state/history_endpoint_repair_test.go | 25 ++++++++++++++++++- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/go/internal/state/history_endpoint_read.go b/go/internal/state/history_endpoint_read.go index 42a261369..b59c51c0d 100644 --- a/go/internal/state/history_endpoint_read.go +++ b/go/internal/state/history_endpoint_read.go @@ -12,7 +12,7 @@ const EndpointEstimatedPV HistoryMissing = 1 << 9 func (s *Store) endpointReadUntil(ctx context.Context, from, to int64) (int64, error) { end := bucketStart(to, 60000) - if to < time.Now().Add(-2*time.Minute).UnixMilli() { + if s.history == nil || to < time.Now().Add(-2*time.Minute).UnixMilli() { return end, nil } var last int64 diff --git a/go/internal/state/history_endpoint_repair_test.go b/go/internal/state/history_endpoint_repair_test.go index 8c7bd7348..122e0054c 100644 --- a/go/internal/state/history_endpoint_repair_test.go +++ b/go/internal/state/history_endpoint_repair_test.go @@ -9,10 +9,13 @@ import ( "time" ) -func endpointFixture(t *testing.T, night bool) (*Store, EndpointRepairRequest, string) { +func endpointFixture(t *testing.T, night bool, at ...int64) (*Store, EndpointRepairRequest, string) { t.Helper() s := sourceStore(t) base := time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour).UnixMilli() + if len(at) > 0 { + base = at[0] + } for at := base - 60000; at <= base+31*60000; at += 5000 { if at >= base+10*60000 && at < base+22*60000 { continue @@ -261,3 +264,23 @@ func TestEndpointRepairRollsBackWhenCacheInvalidationFails(t *testing.T) { t.Fatal("cache partially deleted", hit, err) } } + +func TestEndpointRepairLiveTotalsStopAtCommittedMinute(t *testing.T) { + base := time.Now().Truncate(time.Minute).Add(-40 * time.Minute).UnixMilli() + s, r, backup := endpointFixture(t, false, base) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256); err != nil { + t.Fatal(err) + } + d, err := s.DailyEnergy(base, time.Now().UnixMilli()) + if err != nil || d.Missing&(MissingLoad|MissingPV) != 0 { + t.Fatal("writer tail voided repaired totals", d, err) + } + if d.LoadUntilMS != base+31*60000 || d.PVUntilMS != d.LoadUntilMS { + t.Fatal("extrapolated unfinished tail", d) + } +}