Pre-push review: T-1341 Off-Peak From Readings

diff --git a/internal/derivedstats/integrate_offpeak.go b/internal/derivedstats/integrate_offpeak.gonew file mode 100644index 0000000..8ba8133--- /dev/null+++ b/internal/derivedstats/integrate_offpeak.go@@ -0,0 +1,223 @@+package derivedstats++// OffpeakDeltas is the five integrated kWh values for an off-peak window+// plus provenance counts the writer persists per AC 5.4. All five kWh fields+// are non-negative by construction (per-sample clamping per Decision 8).+type OffpeakDeltas struct {+	GridImportKwh       float64 // ∫ max(pgrid, 0)+	GridExportKwh       float64 // ∫ max(-pgrid, 0)+	BatteryChargeKwh    float64 // ∫ max(-pbat, 0)+	BatteryDischargeKwh float64 // ∫ max(pbat, 0)+	SolarKwh            float64 // ∫ max(ppv, 0)++	// SampleCount counts readings with Timestamp in [startUnix, endUnix).+	// Synthesised edge points do not count.+	SampleCount int+	// SkippedPairs counts adjacent in-window reading pairs separated by more+	// than maxPairGapSeconds. Identical across the five selectors because the+	// gap rule looks only at timestamps.+	SkippedPairs int+}++// IntegrateOffpeakDeltas returns the five integrated kWh values over the+// half-open interval [startUnix, endUnix) plus provenance counts (AC 5.4).+// Returns (zero, false) when fewer than two usable points (in-window or+// bracketing within 60 s of the boundary) exist — caller treats the row as+// "missing record" per AC 1.6.+//+// Per-sample clamping (Decision 8): signed power is split per-sample before+// integration, so import and export over the same window can both be+// non-zero on a day where the sign of pgrid flipped.+//+// 60-second pair-gap rule (AC 1.3 / Decision 4): any consecutive reading pair+// more than 60 s apart contributes zero energy to the integral. Edge+// synthesis is skipped when the bracketing pair fails the same rule.+//+// Precondition: readings must be sorted by Timestamp ascending. DynamoDB+// queries on the sort key satisfy this in production.+func IntegrateOffpeakDeltas(readings []Reading, startUnix, endUnix int64) (OffpeakDeltas, bool) {+	if startUnix >= endUnix || len(readings) == 0 {+		return OffpeakDeltas{}, false+	}++	iL, iR := bracketIndices(readings, startUnix, endUnix)++	// Single pass over the in-window readings: tally sample count, skipped+	// pairs, and the bracket-synthesis edge flags. Both edge flags follow the+	// same "pair within maxPairGapSeconds" rule that integrate() applies, so+	// the usability decision and the integration agree by construction.+	var samples, skipped int+	prevIdx := -1+	for i := iL + 1; i < iR; i++ {+		t := readings[i].Timestamp+		if t < startUnix || t >= endUnix {+			continue+		}+		samples+++		if prevIdx >= 0 {+			gap := readings[i].Timestamp - readings[prevIdx].Timestamp+			if gap > maxPairGapSeconds {+				skipped+++			}+		}+		prevIdx = i+	}++	leftEdge := 0+	if iL >= 0 && iL+1 < len(readings) {+		next := readings[iL+1]+		if next.Timestamp > startUnix &&+			next.Timestamp-readings[iL].Timestamp <= maxPairGapSeconds {+			leftEdge = 1+		}+	}+	rightEdge := 0+	if iR > 0 && iR < len(readings) {+		prev := readings[iR-1]+		next := readings[iR]+		if next.Timestamp-prev.Timestamp <= maxPairGapSeconds {+			rightEdge = 1+		}+	}++	// Usability gate (AC 1.6): integrate() needs at least two construction+	// points. samples counts only strict in-window readings; the two edge+	// flags add synthesised boundary points when the bracketing pair is+	// within the gap rule.+	if samples+leftEdge+rightEdge < 2 {+		return OffpeakDeltas{}, false+	}++	out := OffpeakDeltas{+		GridImportKwh:       integrate(readings, iL, iR, startUnix, endUnix, func(r Reading) float64 { return max(r.Pgrid, 0) }),+		GridExportKwh:       integrate(readings, iL, iR, startUnix, endUnix, func(r Reading) float64 { return max(-r.Pgrid, 0) }),+		BatteryChargeKwh:    integrate(readings, iL, iR, startUnix, endUnix, func(r Reading) float64 { return max(-r.Pbat, 0) }),+		BatteryDischargeKwh: integrate(readings, iL, iR, startUnix, endUnix, func(r Reading) float64 { return max(r.Pbat, 0) }),+		SolarKwh:            integrate(readings, iL, iR, startUnix, endUnix, func(r Reading) float64 { return max(r.Ppv, 0) }),+		SampleCount:         samples,+		SkippedPairs:        skipped,+	}+	return out, true+}++// bracketIndices returns:+//   - iL: largest index with readings[iL].Timestamp < startUnix, or -1 when none.+//   - iR: smallest index > iL with readings[iR].Timestamp >= endUnix, or len(readings).+//+// Matches the bracket discovery used in integratePpv / integratePload.+func bracketIndices(readings []Reading, startUnix, endUnix int64) (iL, iR int) {+	iL = -1+	for i, r := range readings {+		if r.Timestamp < startUnix {+			iL = i+		} else {+			break+		}+	}+	iR = len(readings)+	for i := iL + 1; i < len(readings); i++ {+		if readings[i].Timestamp >= endUnix {+			iR = i+			break+		}+	}+	return iL, iR+}++// integrate is the shared trapezoidal integration over [startUnix, endUnix)+// per Decision 9. Control flow mirrors integratePpv / integratePload (see+// integrate.go); two intentional differences:+//+//  1. Sample-count tallying is removed. IntegrateOffpeakDeltas computes+//     SampleCount and SkippedPairs in a single pass before calling this+//     helper five times — duplicating that work per call would be five+//     extra linear passes for no extra information.+//+//  2. The clamping (e.g. max(r.Pgrid, 0)) is the caller's responsibility via+//     the selector. The helper's job is the algorithm; the per-channel sign+//     policy lives at the call site.+//+// Returns kWh (watt-seconds / 3,600,000). When fewer than two construction+// points exist, returns 0 — IntegrateOffpeakDeltas gates this case with its+// own usability check before calling integrate() five times.+//+// Bracket indices (iL, iR) are supplied by the caller so the five selector+// invocations share one bracketIndices scan instead of redoing it each time.+//+// Precondition: readings must be sorted by Timestamp ascending. The bracket+// searches use first-match early-break and produce silently-wrong results on+// unsorted input.+func integrate(readings []Reading, iL, iR int, startUnix, endUnix int64,+	selector func(Reading) float64,+) float64 {+	if startUnix >= endUnix || len(readings) == 0 {+		return 0+	}++	type pt struct {+		ts int64+		p  float64+	}+	pts := make([]pt, 0, (iR-iL-1)+2)++	// Left edge synthesis.+	if iL >= 0 && iL+1 < len(readings) {+		next := readings[iL+1]+		if next.Timestamp > startUnix {+			gap := next.Timestamp - readings[iL].Timestamp+			if gap <= maxPairGapSeconds {+				prev := readings[iL]+				p0 := selector(prev)+				p1 := selector(next)+				frac := float64(startUnix-prev.Timestamp) / float64(next.Timestamp-prev.Timestamp)+				pts = append(pts, pt{+					ts: startUnix,+					p:  p0 + (p1-p0)*frac,+				})+			}+		}+		// next.Timestamp == startUnix → skip; the interior loop picks up the reading.+	}++	// Interior readings.+	for i := iL + 1; i < iR; i++ {+		r := readings[i]+		if r.Timestamp < startUnix || r.Timestamp >= endUnix {+			continue+		}+		pts = append(pts, pt{ts: r.Timestamp, p: selector(r)})+	}++	// Right edge synthesis. iR is the first index with Timestamp >= endUnix,+	// so readings[iR-1].Timestamp < endUnix is guaranteed when iR > 0.+	if iR > 0 && iR < len(readings) {+		prev := readings[iR-1]+		next := readings[iR]+		gap := next.Timestamp - prev.Timestamp+		if gap <= maxPairGapSeconds {+			p0 := selector(prev)+			p1 := selector(next)+			frac := float64(endUnix-prev.Timestamp) / float64(next.Timestamp-prev.Timestamp)+			pts = append(pts, pt{+				ts: endUnix,+				p:  p0 + (p1-p0)*frac,+			})+		}+	}++	if len(pts) < 2 {+		return 0+	}++	var wattSeconds float64+	for i := 1; i < len(pts); i++ {+		a := pts[i-1]+		b := pts[i]+		dt := b.ts - a.ts+		if dt <= 0 || dt > maxPairGapSeconds {+			continue+		}+		wattSeconds += (a.p + b.p) / 2 * float64(dt)+	}+	return wattSeconds / 3_600_000+}

diff --git a/internal/derivedstats/regression_test.go b/internal/derivedstats/regression_test.gonew file mode 100644index 0000000..88f867a--- /dev/null+++ b/internal/derivedstats/regression_test.go@@ -0,0 +1,110 @@+package derivedstats++import (+	"testing"+	"time"++	"github.com/stretchr/testify/assert"+	"github.com/stretchr/testify/require"+)++// TestRegression_20260518_PeakDropsBelow025Kwh is the bug-incident assertion+// for T-1341 (AC 2.1). On 2026-05-18 the snapshot-diff method recorded+// 1.74 kWh peak / 18.95 kWh off-peak against the meter's 0.17 / 20.75 kWh.+// After re-integrating readings over [11:00, 14:00) Sydney, the peak+// (= eInput − gridUsageKwh) should drop to ≤ 0.25 kWh — within sensor+// calibration of the meter's 0.17 reading.+//+// Fixture provenance: see testdata/offpeak_2026_05_18.json. The actual+// 10-second readings from the incident were not captured before the 30-day+// TTL pruned them, so the fixture is synthesised to reproduce the bug+// shape — a heavy 6.9 kW grid-charging plateau through Phase A and a soft+// rampdown to 5.4 kW over the final 10 minutes before the inverter+// physically stopped at 14:00:00. Production queries flux-readings with+// BETWEEN [windowStart, windowEnd] (inclusive), so the sample at exactly+// 14:00:00 is what the integrator actually receives at the right boundary;+// the integration's right-edge synthesis (AC 1.5) clips that pair to the+// boundary.+func TestRegression_20260518_PeakDropsBelow025Kwh(t *testing.T) {+	loc, err := time.LoadLocation("Australia/Sydney")+	require.NoError(t, err)++	readings, windowStart, windowEnd := readings20260518(loc)+	got, ok := IntegrateOffpeakDeltas(readings, windowStart.Unix(), windowEnd.Unix())+	require.True(t, ok, "fixture must produce a usable integration")+	t.Logf("2026-05-18 fixture: gridUsageKwh=%.4f sampleCount=%d skippedPairs=%d",+		got.GridImportKwh, got.SampleCount, got.SkippedPairs)++	// AC 2.1: peak (= eInput − gridUsageKwh) ≤ 0.25 kWh.+	// eInput is the AlphaESS daily total — 20.69 on 2026-05-18 per the+	// poller's getOneDateEnergyBySn snapshot at 14:00.+	const eInputKwh = 20.69+	peak := eInputKwh - got.GridImportKwh+	assert.LessOrEqual(t, peak, 0.25,+		"AC 2.1: peak (= eInput − gridUsageKwh) must be ≤ 0.25 kWh after re-integration; got %.3f (gridUsageKwh = %.3f)",+		peak, got.GridImportKwh)+	// AC 2.2: gridUsageKwh ≤ eInput is the structural sanity bound — off-peak+	// grid import is a subset of total grid import for the day.+	assert.LessOrEqual(t, got.GridImportKwh, eInputKwh,+		"AC 2.2: gridUsageKwh (%.3f) must be ≤ eInput (%.3f)", got.GridImportKwh, eInputKwh)+	// Sanity: the bug shape — readings-integration should differ substantially+	// from the snapshot-diff value (1.74 kWh peak under the old method). If+	// this assertion ever fails, the fixture has been weakened to the point+	// where it no longer reproduces the original incident.+	assert.Greater(t, got.GridImportKwh, 19.0,+		"fixture should reproduce a heavy-charge day (>19 kWh integrated); got %.3f", got.GridImportKwh)+}++// readings20260518 builds the synthetic 2026-05-18 reading series described+// in testdata/offpeak_2026_05_18.json. Two phases (constant plateau then+// linear rampdown) plus a final reading at exactly windowEnd modelling the+// inverter-stop event. Production queries flux-readings with BETWEEN+// [windowStart, windowEnd] inclusive, so a sample at exactly windowEnd is+// what the integrator actually receives from the table at the right edge.+func readings20260518(loc *time.Location) ([]Reading, time.Time, time.Time) {+	day := time.Date(2026, 5, 18, 0, 0, 0, 0, loc)+	windowStart := day.Add(11 * time.Hour)               // 11:00:00 Sydney+	phaseATail := day.Add(13*time.Hour + 50*time.Minute) // 13:50:00+	windowEnd := day.Add(14 * time.Hour)                 // 14:00:00 Sydney (exclusive interior)++	const (+		phaseAPgrid      = 6900.0+		phaseAPbat       = -6600.0+		phaseBStartPgrid = 6900.0+		phaseBEndPgrid   = 5400.0+		phaseBStartPbat  = -6600.0+		phaseBEndPbat    = -5100.0+		intervalSeconds  = 10+	)++	out := make([]Reading, 0, 1083)++	// Phase A: 11:00:00 → 13:50:00 (inclusive end), 10s cadence, constant.+	for ts := windowStart; !ts.After(phaseATail); ts = ts.Add(intervalSeconds * time.Second) {+		out = append(out, Reading{+			Timestamp: ts.Unix(),+			Pgrid:     phaseAPgrid,+			Pbat:      phaseAPbat,+		})+	}+	// Phase B: 13:50:10 → 13:59:50, linear ramp on pgrid + pbat.+	phaseBDuration := windowEnd.Sub(phaseATail) // 10 minutes+	for ts := phaseATail.Add(intervalSeconds * time.Second); ts.Before(windowEnd); ts = ts.Add(intervalSeconds * time.Second) {+		frac := float64(ts.Sub(phaseATail)) / float64(phaseBDuration)+		out = append(out, Reading{+			Timestamp: ts.Unix(),+			Pgrid:     phaseBStartPgrid + (phaseBEndPgrid-phaseBStartPgrid)*frac,+			Pbat:      phaseBStartPbat + (phaseBEndPbat-phaseBStartPbat)*frac,+		})+	}+	// Right boundary reading at exactly windowEnd: the inverter stopped at+	// 14:00:00. Production's BETWEEN [windowStart, windowEnd] query returns+	// this sample, and the integrator's right-edge synthesis (AC 1.5) clips+	// the final pair to the boundary.+	out = append(out, Reading{+		Timestamp: windowEnd.Unix(),+		Pgrid:     0,+		Pbat:      0,+	})+	return out, windowStart, windowEnd+}

diff --git a/internal/derivedstats/testdata/offpeak_2026_05_18.json b/internal/derivedstats/testdata/offpeak_2026_05_18.jsonnew file mode 100644index 0000000..df97bc9--- /dev/null+++ b/internal/derivedstats/testdata/offpeak_2026_05_18.json@@ -0,0 +1,27 @@+{+  "description": "2026-05-18 regression fixture for T-1341. The day Flux's snapshot-diff reported 1.74 kWh peak / 18.95 kWh off-peak while the meter recorded 0.17 / 20.75 kWh. The 1.57 kWh discrepancy was the last 15-30 minutes of grid-charging that AlphaESS's intraday eInput lagged. After integrating readings over [11:00, 14:00) Sydney, gridUsageKwh should land near the meter's 20.75 kWh and peak (= eInput - gridUsageKwh) drops to <=0.25 kWh.",+  "date": "2026-05-18",+  "windowStartSydney": "2026-05-18T11:00:00+10:00",+  "windowEndSydney":   "2026-05-18T14:00:00+10:00",+  "eInputDailyKwh": 20.69,+  "meterPeakKwh": 0.17,+  "meterOffpeakKwh": 20.75,+  "syntheticReadings": {+    "kind": "constant-then-rampdown-with-boundary-bracket",+    "phaseAStart": "2026-05-18T11:00:00+10:00",+    "phaseAEnd":   "2026-05-18T13:50:00+10:00",+    "phaseAPgridWatts": 6900,+    "phaseAPbatWatts":  -6600,+    "phaseBStart": "2026-05-18T13:50:00+10:00",+    "phaseBEnd":   "2026-05-18T14:00:00+10:00",+    "phaseBPgridStartWatts": 6900,+    "phaseBPgridEndWatts":   5400,+    "phaseBPbatStartWatts":  -6600,+    "phaseBPbatEndWatts":    -5100,+    "intervalSeconds": 10,+    "rightBoundaryAt": "2026-05-18T14:00:00+10:00",+    "rightBoundaryPgridWatts": 0,+    "rightBoundaryPbatWatts": 0,+    "provenance": "Synthesised to reproduce the bug shape — actual reading values from the incident were not captured before TTL pruning. Phase A holds a realistic 6.9 kW grid-charging plateau (heavy combined inverter + EV draw). Phase B ramps down over the last 10 minutes to model the soft cutoff before the inverter physically stops at 14:00:00. Production queries flux-readings with BETWEEN [windowStart, windowEnd] (inclusive), so a sample at exactly windowEnd is what the integrator actually receives at the right boundary; the integration's right-edge synthesis (AC 1.5) clips that pair to the boundary."+  }+}

diff --git a/internal/api/compute.go b/internal/api/compute.goindex 342a161..a760487 100644--- a/internal/api/compute.go+++ b/internal/api/compute.go@@ -2,6 +2,7 @@ package api  import ( 	"math"+	"sort" 	"time"  	"github.com/ArjenSchwarz/flux/internal/derivedstats"@@ -18,41 +19,23 @@ var sydneyTZ = func() *time.Location { 	return loc }() -// offpeakDeltas resolves the energy deltas for an off-peak record.+// offpeakDeltas resolves the energy deltas for a complete off-peak record.+// Pending records return (_, false); callers needing today's in-window value+// live-integrate via liveOffpeakDeltas. //-// A complete record carries final deltas from the poller. A pending record-// requires a current snapshot (the running totals for the same day as op)-// to project against the start snapshot; without one the deltas are-// unknown. Returns ok=false when the data is not usable.-func offpeakDeltas(op dynamo.OffpeakItem, current *TodayEnergy) (deltas offpeakDeltaValues, ok bool) {-	switch op.Status {-	case dynamo.OffpeakStatusComplete:-		return offpeakDeltaValues{-			GridImport:       op.GridUsageKwh,-			Solar:            op.SolarKwh,-			BatteryCharge:    op.BatteryChargeKwh,-			BatteryDischarge: op.BatteryDischargeKwh,-			GridExport:       op.GridExportKwh,-		}, true-	case dynamo.OffpeakStatusPending:-		if current == nil {-			return offpeakDeltaValues{}, false-		}-		// Energy counters are monotonically non-decreasing, so deltas-		// should never be negative. They can briefly appear negative if-		// the running snapshot lags the start snapshot (poller writes the-		// start record, then a later reconciliation reduces the running-		// total). Clamp to zero so the dashboard never shows nonsense-		// like "-0.1 kWh imported".-		return offpeakDeltaValues{-			GridImport:       max(0, current.EInput-op.StartEInput),-			Solar:            max(0, current.Epv-op.StartEpv),-			BatteryCharge:    max(0, current.ECharge-op.StartECharge),-			BatteryDischarge: max(0, current.EDischarge-op.StartEDischarge),-			GridExport:       max(0, current.EOutput-op.StartEOutput),-		}, true-	}-	return offpeakDeltaValues{}, false+// AC 5.3: this function does not read op.StartE* / op.EndE* — those fields+// exist on the row for operator diagnostics only.+func offpeakDeltas(op dynamo.OffpeakItem) (offpeakDeltaValues, bool) {+	if op.Status != dynamo.OffpeakStatusComplete {+		return offpeakDeltaValues{}, false+	}+	return offpeakDeltaValues{+		GridImport:       op.GridUsageKwh,+		Solar:            op.SolarKwh,+		BatteryCharge:    op.BatteryChargeKwh,+		BatteryDischarge: op.BatteryDischargeKwh,+		GridExport:       op.GridExportKwh,+	}, true }  // offpeakDeltaValues holds the five energy deltas derived from an off-peak record.@@ -64,6 +47,63 @@ type offpeakDeltaValues struct { 	GridExport       float64 } +// liveOffpeakDeltas integrates readings over [offpeakStart, min(now, offpeakEnd))+// for today's date in Sydney local time and returns the five energy deltas.+//+// offpeakStart and offpeakEnd are the raw "HH:MM" config values. Returns+// (_, false) when the window is unparseable, when now is at or before the+// window start (AC 4.3 — pre-window behaviour), or when the readings slice+// does not contain enough usable samples to integrate (AC 1.6).+//+// Pure function: no state and no clock except the explicit now parameter. This+// is the determinism contract that backs AC 4.4's monotonicity guarantee.+func liveOffpeakDeltas(readings []dynamo.ReadingItem, now time.Time,+	offpeakStart, offpeakEnd string,+) (offpeakDeltaValues, bool) {+	startMin, endMin, parsed := derivedstats.ParseOffpeakWindow(offpeakStart, offpeakEnd)+	if !parsed {+		return offpeakDeltaValues{}, false+	}+	local := now.In(sydneyTZ)+	dayStart := time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, sydneyTZ)+	opStart := dayStart.Add(time.Duration(startMin) * time.Minute)+	opEnd := dayStart.Add(time.Duration(endMin) * time.Minute)++	if !local.After(opStart) {+		return offpeakDeltaValues{}, false+	}+	endTime := opEnd+	if local.Before(opEnd) {+		endTime = local+	}++	// Trim the readings slice to just the bracketed window before the+	// derivedstats conversion. A /status request hands in ~8640 readings of+	// which only ~1080 sit inside the off-peak window; converting the rest+	// to []derivedstats.Reading is wasted work. The window slice keeps one+	// reading before opStart (for left-edge synthesis) and one reading at or+	// after endTime (for right-edge synthesis), so bracket integration in+	// IntegrateOffpeakDeltas observes the same boundary samples it would on+	// the full slice.+	windowed := sliceWindow(readings, opStart.Unix(), endTime.Unix())++	deltas, ok := derivedstats.IntegrateOffpeakDeltas(+		toDerivedReadings(windowed),+		opStart.Unix(),+		endTime.Unix(),+	)+	if !ok {+		return offpeakDeltaValues{}, false+	}+	return offpeakDeltaValues{+		GridImport:       deltas.GridImportKwh,+		Solar:            deltas.SolarKwh,+		BatteryCharge:    deltas.BatteryChargeKwh,+		BatteryDischarge: deltas.BatteryDischargeKwh,+		GridExport:       deltas.GridExportKwh,+	}, true+}+ // computeCutoffTime estimates when the battery will reach the cutoff percentage // using linear extrapolation. Returns nil if the battery is not discharging or // SOC is already at/below cutoff.@@ -264,11 +304,11 @@ func computeTodayEnergy(readings []dynamo.ReadingItem, midnightUnix int64) *Toda 	}  	return &TodayEnergy{-		Epv:        roundEnergy(epvWh / 1000),-		EInput:     roundEnergy(eInputWh / 1000),-		EOutput:    roundEnergy(eOutputWh / 1000),-		ECharge:    roundEnergy(eChargeWh / 1000),-		EDischarge: roundEnergy(eDischargeWh / 1000),+		Epv:        derivedstats.RoundEnergy(epvWh / 1000),+		EInput:     derivedstats.RoundEnergy(eInputWh / 1000),+		EOutput:    derivedstats.RoundEnergy(eOutputWh / 1000),+		ECharge:    derivedstats.RoundEnergy(eChargeWh / 1000),+		EDischarge: derivedstats.RoundEnergy(eDischargeWh / 1000), 	} } @@ -324,16 +364,48 @@ func startOfDaySydney(now time.Time) time.Time { 	return time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, sydneyTZ) } -// roundEnergy rounds a kWh value to 2 decimal places.-func roundEnergy(v float64) float64 {-	return math.Round(v*100) / 100-}- // roundPower rounds a watts or SOC value to 1 decimal place. func roundPower(v float64) float64 { 	return math.Round(v*10) / 10 } +// sliceWindow returns the contiguous sub-slice of readings spanning the+// off-peak window, plus one bracketing reading on each side when present.+// Readings must be sorted by Timestamp ascending (the DynamoDB sort-key+// guarantee for /status's today query). The returned slice aliases the input+// — callers must not mutate it.+//+// The conservative bracket extension (one reading before startUnix, one at or+// after endUnix) preserves derivedstats.IntegrateOffpeakDeltas's left and+// right edge-synthesis behaviour: it needs a neighbour outside the window to+// interpolate the boundary points.+func sliceWindow(readings []dynamo.ReadingItem, startUnix, endUnix int64) []dynamo.ReadingItem {+	if len(readings) == 0 {+		return readings+	}+	// First index with Timestamp >= startUnix.+	leftIn := sort.Search(len(readings), func(i int) bool {+		return readings[i].Timestamp >= startUnix+	})+	// First index with Timestamp > endUnix — anything strictly greater can+	// only contribute as a right-edge bracket of [start, end), so we keep+	// exactly one (the search returns the cut point; one slot beyond is+	// outside the half-open window and any further reading is irrelevant).+	rightOut := sort.Search(len(readings), func(i int) bool {+		return readings[i].Timestamp > endUnix+	})++	// Extend one reading to the left for left-edge bracket synthesis.+	if leftIn > 0 {+		leftIn--+	}+	// Extend one reading to the right for right-edge bracket synthesis.+	if rightOut < len(readings) {+		rightOut+++	}+	return readings[leftIn:rightOut]+}+ // toDerivedReadings converts a slice of dynamo.ReadingItem to the leaf // []derivedstats.Reading. Per Decision 9 this conversion is duplicated at // each call site (api, poller) to keep the derivedstats package free of

diff --git a/internal/api/status.go b/internal/api/status.goindex 6166b31..f3ff85e 100644--- a/internal/api/status.go+++ b/internal/api/status.go@@ -187,22 +187,19 @@ func (h *Handler) handleStatus(ctx context.Context, _ events.LambdaFunctionURLRe 	var storedEnergy *TodayEnergy 	if deItem != nil { 		storedEnergy = &TodayEnergy{-			Epv:        roundEnergy(deItem.Epv),-			EInput:     roundEnergy(deItem.EInput),-			EOutput:    roundEnergy(deItem.EOutput),-			ECharge:    roundEnergy(deItem.ECharge),-			EDischarge: roundEnergy(deItem.EDischarge),+			Epv:        derivedstats.RoundEnergy(deItem.Epv),+			EInput:     derivedstats.RoundEnergy(deItem.EInput),+			EOutput:    derivedstats.RoundEnergy(deItem.EOutput),+			ECharge:    derivedstats.RoundEnergy(deItem.ECharge),+			EDischarge: derivedstats.RoundEnergy(deItem.EDischarge), 		} 	} 	resp.TodayEnergy = reconcileEnergy(computedEnergy, storedEnergy)  	// Off-peak data — always includes window times, plus deltas when-	// complete or projectable from today's reconciled totals (live--	// integrated, with the AlphaESS counter as ground truth via-	// reconcileEnergy). Diffing reconciled-vs-window-start ensures the-	// dashboard never lags the underlying counter and never reports a-	// number lower than the snapshot baseline.-	resp.Offpeak = buildOffpeak(opItem, resp.TodayEnergy, h.offpeakStart, h.offpeakEnd)+	// complete (from the finalised row) or pending today (live-integrated+	// from the readings already in memory for live compute).+	resp.Offpeak = buildOffpeak(opItem, allReadings, now, h.offpeakStart, h.offpeakEnd) 	resp.Note = noteText  	return jsonResponse(resp)@@ -222,32 +219,45 @@ func filterReadings(readings []dynamo.ReadingItem, from, to int64) []dynamo.Read // buildOffpeak constructs the OffpeakData response. // // Window times are always included. Deltas come from one of two sources:-//   - Complete record: the poller has captured both start and end snapshots-//     and computed final deltas.-//   - In-progress: a pending record exists (start snapshot only) and a-//     current snapshot is supplied, so deltas can be computed against the-//     running totals. Battery delta percent is unknown mid-window because-//     we lack a current SOC-as-of-now in this slice.+//   - Complete record: the poller has finalised the five integration-sourced+//     deltas, served directly from the row.+//   - Pending record on today, with now inside the window: live-integrate+//     readings over [offpeak-start, min(now, offpeak-end)). Battery delta+//     percent is unknown mid-window because we lack a fixed end SOC. //-// Returns deltas as nil when neither source is usable.-func buildOffpeak(item *dynamo.OffpeakItem, today *TodayEnergy, windowStart, windowEnd string) *OffpeakData {+// Returns deltas as nil when neither source is usable (no row, pending row+// before the window opens, or sparse readings).+func buildOffpeak(item *dynamo.OffpeakItem, readings []dynamo.ReadingItem, now time.Time,+	offpeakStart, offpeakEnd string,+) *OffpeakData { 	od := &OffpeakData{-		WindowStart: windowStart,-		WindowEnd:   windowEnd,+		WindowStart: offpeakStart,+		WindowEnd:   offpeakEnd, 	} 	if item == nil { 		return od 	}-	deltas, ok := offpeakDeltas(*item, today)++	var (+		deltas offpeakDeltaValues+		ok     bool+	)+	switch item.Status {+	case dynamo.OffpeakStatusComplete:+		deltas, ok = offpeakDeltas(*item)+	case dynamo.OffpeakStatusPending:+		deltas, ok = liveOffpeakDeltas(readings, now, offpeakStart, offpeakEnd)+	} 	if !ok { 		return od 	}+ 	od.Status = item.Status-	od.GridUsageKwh = floatPtr(roundEnergy(deltas.GridImport))-	od.SolarKwh = floatPtr(roundEnergy(deltas.Solar))-	od.BatteryChargeKwh = floatPtr(roundEnergy(deltas.BatteryCharge))-	od.BatteryDischargeKwh = floatPtr(roundEnergy(deltas.BatteryDischarge))-	od.GridExportKwh = floatPtr(roundEnergy(deltas.GridExport))+	od.GridUsageKwh = floatPtr(derivedstats.RoundEnergy(deltas.GridImport))+	od.SolarKwh = floatPtr(derivedstats.RoundEnergy(deltas.Solar))+	od.BatteryChargeKwh = floatPtr(derivedstats.RoundEnergy(deltas.BatteryCharge))+	od.BatteryDischargeKwh = floatPtr(derivedstats.RoundEnergy(deltas.BatteryDischarge))+	od.GridExportKwh = floatPtr(derivedstats.RoundEnergy(deltas.GridExport)) 	if item.Status == dynamo.OffpeakStatusComplete { 		od.BatteryDeltaPercent = floatPtr(roundPower(item.BatteryDeltaPercent)) 	}

diff --git a/internal/api/history.go b/internal/api/history.goindex 343529f..70d1fcd 100644--- a/internal/api/history.go+++ b/internal/api/history.go@@ -129,11 +129,11 @@ func (h *Handler) handleHistory(ctx context.Context, req events.LambdaFunctionUR 	result := make([]DayEnergy, len(items)) 	for i, item := range items { 		stored := &TodayEnergy{-			Epv:        roundEnergy(item.Epv),-			EInput:     roundEnergy(item.EInput),-			EOutput:    roundEnergy(item.EOutput),-			ECharge:    roundEnergy(item.ECharge),-			EDischarge: roundEnergy(item.EDischarge),+			Epv:        derivedstats.RoundEnergy(item.Epv),+			EInput:     derivedstats.RoundEnergy(item.EInput),+			EOutput:    derivedstats.RoundEnergy(item.EOutput),+			ECharge:    derivedstats.RoundEnergy(item.ECharge),+			EDischarge: derivedstats.RoundEnergy(item.EDischarge), 		} 		isItemToday := item.Date == today 		energy := stored@@ -149,7 +149,7 @@ func (h *Handler) handleHistory(ctx context.Context, req events.LambdaFunctionUR 			EDischarge: energy.EDischarge, 		} 		if op, ok := offpeakByDate[item.Date]; ok {-			imp, exp, hasSplit := offpeakSplit(op, energy, isItemToday)+			imp, exp, hasSplit := offpeakSplit(op, todayReadings, now, isItemToday, h.offpeakStart, h.offpeakEnd) 			if hasSplit { 				day.OffpeakGridImportKwh = floatPtr(imp) 				day.OffpeakGridExportKwh = floatPtr(exp)@@ -197,18 +197,28 @@ func (h *Handler) handleHistory(ctx context.Context, req events.LambdaFunctionUR  // offpeakSplit returns the off-peak grid import and export for a single day. //-// Complete records carry final deltas computed at window close. A pending-// record on today's date can be projected forward against the running daily-// energy totals; pending records on past dates indicate a poller failure and-// are reported as missing rather than zero. Returns hasSplit=false when the-// data is not usable.-func offpeakSplit(op dynamo.OffpeakItem, energy *TodayEnergy, isToday bool) (imp, exp float64, hasSplit bool) {-	if op.Status == dynamo.OffpeakStatusPending && !isToday {+// Complete records pass through the finalised deltas. A pending record on+// today's date live-integrates from the readings slice over+// [offpeak-start, min(now, offpeak-end)). Pending records on past dates+// indicate a poller failure and are reported as missing rather than zero.+// Returns hasSplit=false when the data is not usable (sparse readings or+// pre-window now).+func offpeakSplit(op dynamo.OffpeakItem, readings []dynamo.ReadingItem, now time.Time,+	isToday bool, offpeakStart, offpeakEnd string,+) (imp, exp float64, hasSplit bool) {+	if op.Status == dynamo.OffpeakStatusComplete {+		deltas, ok := offpeakDeltas(op)+		if !ok {+			return 0, 0, false+		}+		return derivedstats.RoundEnergy(deltas.GridImport), derivedstats.RoundEnergy(deltas.GridExport), true+	}+	if op.Status != dynamo.OffpeakStatusPending || !isToday { 		return 0, 0, false 	}-	deltas, ok := offpeakDeltas(op, energy)+	deltas, ok := liveOffpeakDeltas(readings, now, offpeakStart, offpeakEnd) 	if !ok { 		return 0, 0, false 	}-	return roundEnergy(deltas.GridImport), roundEnergy(deltas.GridExport), true+	return derivedstats.RoundEnergy(deltas.GridImport), derivedstats.RoundEnergy(deltas.GridExport), true }

diff --git a/internal/api/offpeak_live_perf_test.go b/internal/api/offpeak_live_perf_test.gonew file mode 100644index 0000000..6fac548--- /dev/null+++ b/internal/api/offpeak_live_perf_test.go@@ -0,0 +1,113 @@+package api++import (+	"context"+	"sort"+	"testing"+	"time"++	"github.com/ArjenSchwarz/flux/internal/dynamo"+	"github.com/stretchr/testify/assert"+	"github.com/stretchr/testify/require"+)++// TestHandleStatus_LiveOffpeak_P95Under500ms covers AC 8.2: a /status+// response on a day with live-integrated off-peak (pending row + readings+// integration over [offpeak-start, min(now, offpeak-end))) must complete+// within 500 ms p95 warm at the production memory configuration.+//+// The test drives a full /status request through the same handler that+// runs in Lambda — including JSON marshal, mux routing, and the live+// integration over a synthetic 1080-reading off-peak window. p95 is+// measured across many warm invocations against the same in-memory+// mockReader (no I/O latency to confound the measurement) and logged in+// the test output so the spec can reference the recorded number.+//+// 500 ms is generous on developer hardware (typical: <5 ms p95). The+// budget exists primarily to catch regressions that change algorithmic+// complexity or add per-request I/O.+func TestHandleStatus_LiveOffpeak_P95Under500ms(t *testing.T) {+	if testing.Short() {+		t.Skip("skipping perf test in -short mode")+	}+	loc := sydneyTZ+	now := time.Date(2026, 4, 15, 13, 30, 0, 0, loc) // 30 min before window-end+	opStart := time.Date(2026, 4, 15, 11, 0, 0, 0, loc)++	// A full off-peak window's worth of 10-second readings spanning the+	// query horizon (nowUnix-86400 → nowUnix). Pgrid varies sample-to-sample+	// so the integration does real per-sample work, matching the production+	// pattern.+	var readings []dynamo.ReadingItem+	dayStart := time.Date(2026, 4, 15, 0, 0, 0, 0, loc)+	const intervalSeconds = 10+	for ts := dayStart.Unix(); ts <= now.Unix(); ts += intervalSeconds {+		readings = append(readings, dynamo.ReadingItem{+			Timestamp: ts,+			Ppv:       float64(500 + int(ts)%2000),+			Pgrid:     float64(-1500 + int(ts)%5500),+			Pbat:      float64(-2500 + int(ts)%4500),+			Pload:     float64(500 + int(ts)%4000),+			Soc:       float64(20 + int(ts)%70),+		})+	}+	require.GreaterOrEqual(t, len(readings), 1080,+		"fixture must include a full off-peak window's worth of readings")++	mr := &mockReader{+		queryReadingsFn: func(_ context.Context, _ string, _, _ int64) ([]dynamo.ReadingItem, error) {+			return readings, nil+		},+		getOffpeakFn: func(_ context.Context, _, _ string) (*dynamo.OffpeakItem, error) {+			return &dynamo.OffpeakItem{Status: dynamo.OffpeakStatusPending}, nil+		},+	}+	h := NewHandler(mr, nil, testSerial, testToken, "11:00", "14:00")+	h.nowFunc = func() time.Time { return now }++	// Warm-up — first call pays one-off costs (sync.Once init, allocator+	// warm-up) we don't want to attribute to steady-state p95.+	const warmup = 10+	for range warmup {+		resp, err := h.Handle(context.Background(), statusRequest())+		require.NoError(t, err)+		require.Equal(t, 200, resp.StatusCode)+	}++	// Warm measurement: integrate the live deltas across many requests.+	// 200 samples → p95 lands on the 190th sorted timing.+	const samples = 200+	timings := make([]time.Duration, samples)+	for i := range samples {+		started := time.Now()+		resp, err := h.Handle(context.Background(), statusRequest())+		timings[i] = time.Since(started)+		require.NoError(t, err)+		require.Equal(t, 200, resp.StatusCode)+	}++	// Confirm the live integration actually executed — sanity-checks the+	// fixture so a future refactor that bypasses liveOffpeakDeltas doesn't+	// silently make the timing meaningless.+	{+		resp, err := h.Handle(context.Background(), statusRequest())+		require.NoError(t, err)+		sr := parseStatusResponse(t, resp)+		require.NotNil(t, sr.Offpeak)+		require.Equal(t, dynamo.OffpeakStatusPending, sr.Offpeak.Status)+		require.NotNil(t, sr.Offpeak.GridUsageKwh,+			"live path must populate gridUsageKwh from the readings integration")+	}++	sort.Slice(timings, func(i, j int) bool { return timings[i] < timings[j] })+	p50 := timings[samples/2]+	p95 := timings[(samples*95)/100]+	p99 := timings[(samples*99)/100]+	max := timings[samples-1]+	opStartStr := opStart.Format("15:04:05")+	t.Logf("/status live-offpeak (opStart=%s, now=%s, readings=%d, n=%d): p50=%s p95=%s p99=%s max=%s",+		opStartStr, now.Format("15:04:05"), len(readings), samples, p50, p95, p99, max)++	assert.Less(t, p95, 500*time.Millisecond,+		"AC 8.2: /status p95 with live off-peak integration must be < 500 ms; got %s", p95)+}

diff --git a/internal/poller/offpeak.go b/internal/poller/offpeak.goindex 76ec405..729c848 100644--- a/internal/poller/offpeak.go+++ b/internal/poller/offpeak.go@@ -2,6 +2,7 @@ package poller  import ( 	"context"+	"errors" 	"fmt" 	"log/slog" 	"sync"@@ -9,12 +10,24 @@ import (  	"github.com/ArjenSchwarz/flux/internal/alphaess" 	"github.com/ArjenSchwarz/flux/internal/config"+	"github.com/ArjenSchwarz/flux/internal/derivedstats" 	"github.com/ArjenSchwarz/flux/internal/dynamo" )  const ( 	snapshotRetries   = 3 	defaultRetryDelay = 10 * time.Second++	// defaultEndWaitBudget is the maximum time handleEnd waits after+	// offpeak-end for a reading with timestamp >= offpeak-end to land+	// (specs/offpeak-from-readings AC 3.1). The 10 s live-poll cadence means+	// the first at-or-after-boundary reading typically arrives within 5–15 s.+	defaultEndWaitBudget = 30 * time.Second++	// defaultEndWaitPollInterval is the cadence at which handleEnd probes+	// the readings table while waiting for an at-or-after-boundary reading.+	// Two seconds keeps the probe lightweight against the live-poll cadence.+	defaultEndWaitPollInterval = 2 * time.Second )  // windowPosition represents the poller's position relative to the off-peak window.@@ -35,11 +48,17 @@ type OffpeakScheduler struct { 	// In-memory state for current day's off-peak calculation. 	startSnapshot *alphaess.EnergyData 	socStart      float64-	hasStart      bool  	// retryDelay between snapshot attempts (overridable for tests). 	retryDelay time.Duration +	// endWaitBudget / endWaitPollInterval control the boundary wait in+	// handleEnd (specs/offpeak-from-readings AC 3.1). Zero means "use+	// default" (defaultEndWaitBudget / defaultEndWaitPollInterval). Tests+	// override these to keep wall time short.+	endWaitBudget       time.Duration+	endWaitPollInterval time.Duration+ 	// now returns the current time. Injectable for deterministic testing. 	now func() time.Time }@@ -68,7 +87,8 @@ func (o *OffpeakScheduler) Run(loopCtx, drainCtx context.Context, wg *sync.WaitG  	switch pos { 	case positionBefore:-		// Wait for start time, then handle start and end.+		// Wait for start time, then handle start and end. handleStart+		// populates in-memory state, so handleEnd doesn't need the pending row. 		if !o.waitUntil(loopCtx, wallClockTime(now, o.cfg.Location, o.cfg.OffpeakStart)) { 			return 		}@@ -80,24 +100,29 @@ func (o *OffpeakScheduler) Run(loopCtx, drainCtx context.Context, wg *sync.WaitG 		if !o.waitUntil(loopCtx, wallClockTime(now, o.cfg.Location, o.cfg.OffpeakEnd)) { 			return 		}-		o.handleEndOrCleanup(drainCtx, date)+		o.handleEndOrCleanup(drainCtx, date, nil)  	case positionDuring:-		// Try to recover from existing pending record.-		if err := o.recoverMidWindow(drainCtx, date); err != nil {-			slog.Error("offpeak mid-window recovery failed", "date", date, "error", err)-		}-		if o.hasStart {+		// T-1341: recovery surfaces the pending row to the caller. In-memory+		// state isn't rebuilt — handleEnd reads readings directly and uses+		// the pending row's StartE* fields for diagnostic snapshot population.+		pending, _ := o.recoverMidWindow(drainCtx, date)+		if pending != nil { 			if !o.waitUntil(loopCtx, wallClockTime(now, o.cfg.Location, o.cfg.OffpeakEnd)) { 				return 			}-			o.handleEndOrCleanup(drainCtx, date)+			o.handleEndOrCleanup(drainCtx, date, pending) 		} else { 			slog.Info("offpeak: no pending record found, skipping today", "date", date) 		}  	case positionAfter:-		slog.Info("offpeak: past window, skipping today", "date", date)+		// T-1341 AC 3.4: restart between offpeak-end and 24:00. If a pending+		// row exists, finalise it now (skip boundary wait — the boundary is+		// in the past). Otherwise log+skip.+		if err := o.recoverAfterWindow(drainCtx, date); err != nil {+			slog.Error("offpeak post-window recovery failed", "date", date, "error", err)+		} 	}  nextDay:@@ -122,7 +147,7 @@ nextDay: 			return 		} -		o.handleEndOrCleanup(drainCtx, date)+		o.handleEndOrCleanup(drainCtx, date, nil) 	} } @@ -135,7 +160,6 @@ func (o *OffpeakScheduler) handleStart(ctx context.Context, date string) error {  	o.startSnapshot = energy 	o.socStart = soc-	o.hasStart = true  	item := dynamo.OffpeakItem{ 		SysSn: o.cfg.Serial, Date: date, Status: dynamo.OffpeakStatusPending,@@ -151,22 +175,174 @@ func (o *OffpeakScheduler) handleStart(ctx context.Context, date string) error { 	return nil } -// handleEnd captures the end snapshot, computes deltas, and writes the complete record.-func (o *OffpeakScheduler) handleEnd(ctx context.Context, date string) error {+// handleEnd finalises the day's off-peak row by integrating the readings+// table over the SSM window (specs/offpeak-from-readings T-1341).+//+// When pending is non-nil (post-restart recovery path) its StartE* fields+// populate the diagnostic start snapshot; otherwise handleEnd uses the+// in-memory state captured by handleStart in this process. One of the two+// must be set — both nil indicates a programming error and returns immediately.+//+// The boundary-wait step is skipped automatically when offpeak-end is already+// in the past (positionAfter recovery): probing for an at-or-after-boundary+// reading would only burn the budget on a moot wait.+//+// Flow (matches design.md "Window-end finalisation state machine"):+//  1. Capture the AlphaESS end snapshot (Decision 2 — diagnostic only).+//  2. Wait up to endWaitBudget for a reading at-or-after offpeak-end+//     (AC 3.1), unless the boundary is already in the past.+//  3. Strongly-consistent query of readings over [offpeak-start, offpeak-end).+//  4. Integrate the five deltas via derivedstats.IntegrateOffpeakDeltas.+//  5. Conditional write with WriteOffpeakIfPendingOrAbsent — fails only when+//     a concurrent writer (backfill CLI) reached `complete` first; in that+//     case we log+skip and accept the other writer's value (AC 3.5).+func (o *OffpeakScheduler) handleEnd(ctx context.Context, date string, pending *dynamo.OffpeakItem) error { 	energy, soc, err := o.captureSnapshot(ctx, date) 	if err != nil { 		return fmt.Errorf("capture end snapshot: %w", err) 	} -	item := computeOffpeakDeltas(o.cfg.Serial, date, o.startSnapshot, energy, o.socStart, soc)-	if err := o.store.WriteOffpeak(ctx, item); err != nil {+	// Resolve the diagnostic start snapshot. The integration over readings+	// is the source of truth for the five deltas (Decision 2); the start+	// snapshot is retained only for drift logging and operator forensics.+	startSnap := o.startSnapshot+	startSoc := o.socStart+	if startSnap == nil {+		if pending == nil {+			// Unreachable in production: both Run() and recoverAfterWindow+			// only call handleEnd with either in-memory state (handleStart+			// ran in this process) or a non-nil pending row.+			return fmt.Errorf("handleEnd: no start snapshot available for %s (no in-memory state, no pending row)", date)+		}+		startSnap = &alphaess.EnergyData{+			Epv: pending.StartEpv, EInput: pending.StartEInput, EOutput: pending.StartEOutput,+			ECharge: pending.StartECharge, EDischarge: pending.StartEDischarge,+			EGridCharge: pending.StartEGridCharge,+		}+		startSoc = pending.SocStart+	}++	// Resolve boundary window from cfg in Sydney local time. The wall-clock+	// helper exists already; we re-use it so DST handling stays uniform.+	day, _ := time.ParseInLocation(dateLayout, date, o.cfg.Location)+	windowStart := wallClockTime(day, o.cfg.Location, o.cfg.OffpeakStart)+	windowEnd := wallClockTime(day, o.cfg.Location, o.cfg.OffpeakEnd)++	// Skip the boundary-wait when offpeak-end is already in the past: the+	// at-or-after-boundary reading either already exists or never will, and+	// the wait would just burn its budget on a moot probe. This subsumes the+	// former skipBoundaryWait parameter — the positionAfter recovery path+	// always satisfies this condition (Run() only enters it when now is past+	// windowEnd) and the normal positionBefore/positionDuring paths fire+	// handleEnd within a second of windowEnd so the wait runs.+	if !o.now().After(windowEnd) {+		budget := o.endWaitBudget+		if budget == 0 {+			budget = defaultEndWaitBudget+		}+		pollInterval := o.endWaitPollInterval+		if pollInterval == 0 {+			pollInterval = defaultEndWaitPollInterval+		}+		found, waitErr := o.waitForReadingAtOrAfter(ctx, windowEnd, budget, pollInterval)+		if waitErr != nil {+			// Wait surfaced a store error — log and continue; the query+			// below will either succeed and integrate what it sees, or+			// fail and we propagate that error. This matches "best-effort+			// within the 5-min deadline" (AC 3.2).+			slog.Warn("offpeak boundary wait failed; proceeding with integration",+				"date", date, "error", waitErr)+		} else if !found {+			slog.Warn("offpeak boundary wait timed out; integrating with available readings",+				"date", date, "budget", budget)+		}+	}++	readings, err := o.store.QueryReadingsConsistent(ctx, o.cfg.Serial,+		windowStart.Unix(), windowEnd.Unix())+	if err != nil {+		return fmt.Errorf("query readings for offpeak integration: %w", err)+	}++	deltas := integrateReadings(readings, windowStart, windowEnd)+	item := buildOffpeakRow(o.cfg.Serial, date, startSnap, energy,+		startSoc, soc, deltas, o.now().UTC())++	dynamo.LogOffpeakDrift(date, item)++	if err := o.store.WriteOffpeakIfPendingOrAbsent(ctx, item); err != nil {+		if errors.Is(err, dynamo.ErrOffpeakConditionFailed) {+			slog.Warn("offpeak conditional write rejected (peer writer finalised first); skipping",+				"date", date)+			return nil+		} 		return fmt.Errorf("write complete offpeak: %w", err) 	} -	slog.Info("offpeak end captured", "date", date, "socStart", o.socStart, "socEnd", soc)+	slog.Info("offpeak end captured",+		"date", date, "socStart", startSoc, "socEnd", soc,+		"gridUsageKwh", item.GridUsageKwh, "solarKwh", item.SolarKwh,+		"sampleCount", item.IntegrationSampleCount,+		"skippedPairs", item.IntegrationSkippedPairs) 	return nil } +// integrateReadings converts a []dynamo.ReadingItem to []derivedstats.Reading+// and runs the off-peak integration over [windowStart, windowEnd). Empty or+// sparse readings produce zero-valued OffpeakDeltas with SampleCount == 0+// (the (_, false) case from IntegrateOffpeakDeltas) — see handleEnd which+// persists that as a `complete` row per AC 1.6 + AC 3.2.+func integrateReadings(readings []dynamo.ReadingItem, windowStart, windowEnd time.Time) derivedstats.OffpeakDeltas {+	pts := make([]derivedstats.Reading, len(readings))+	for i, r := range readings {+		pts[i] = derivedstats.Reading{+			Timestamp: r.Timestamp,+			Ppv:       r.Ppv,+			Pload:     r.Pload,+			Soc:       r.Soc,+			Pbat:      r.Pbat,+			Pgrid:     r.Pgrid,+		}+	}+	d, _ := derivedstats.IntegrateOffpeakDeltas(pts, windowStart.Unix(), windowEnd.Unix())+	return d+}++// buildOffpeakRow composes the OffpeakItem persisted at window-end. The five+// kWh deltas come from the readings integration (T-1341 AC 5.2); the+// startE*/endE* snapshot fields are retained as diagnostics only (Decision 2).+// Rounded to two decimal places (AC 7.7) so the poller and the backfill CLI+// produce byte-equal values for the same readings.+func buildOffpeakRow(+	serial, date string,+	start, end *alphaess.EnergyData,+	socStart, socEnd float64,+	deltas derivedstats.OffpeakDeltas,+	integratedAt time.Time,+) dynamo.OffpeakItem {+	return dynamo.OffpeakItem{+		SysSn: serial, Date: date, Status: dynamo.OffpeakStatusComplete,+		StartEpv:        start.Epv,+		StartEInput:     start.EInput,+		StartEOutput:    start.EOutput,+		StartECharge:    start.ECharge,+		StartEDischarge: start.EDischarge, StartEGridCharge: start.EGridCharge,+		SocStart: socStart,+		EndEpv:   end.Epv, EndEInput: end.EInput, EndEOutput: end.EOutput,+		EndECharge: end.ECharge, EndEDischarge: end.EDischarge, EndEGridCharge: end.EGridCharge,+		SocEnd:                  socEnd,+		GridUsageKwh:            derivedstats.RoundEnergy(deltas.GridImportKwh),+		SolarKwh:                derivedstats.RoundEnergy(deltas.SolarKwh),+		BatteryChargeKwh:        derivedstats.RoundEnergy(deltas.BatteryChargeKwh),+		BatteryDischargeKwh:     derivedstats.RoundEnergy(deltas.BatteryDischargeKwh),+		GridExportKwh:           derivedstats.RoundEnergy(deltas.GridExportKwh),+		BatteryDeltaPercent:     socEnd - socStart,+		IntegrationSampleCount:  deltas.SampleCount,+		IntegrationSkippedPairs: deltas.SkippedPairs,+		IntegratedAt:            integratedAt.Format(time.RFC3339),+	}+}+ // captureSnapshot calls GetOneDateEnergy + GetLastPowerData in parallel with retry. // Both API calls are independent and run concurrently within each attempt. func (o *OffpeakScheduler) captureSnapshot(ctx context.Context, date string) (*alphaess.EnergyData, float64, error) {@@ -215,32 +391,62 @@ func (o *OffpeakScheduler) captureSnapshot(ctx context.Context, date string) (*a 	return nil, 0, fmt.Errorf("off-peak snapshot failed after %d attempts: %w", snapshotRetries, lastErr) } -// recoverMidWindow checks for an existing pending record and recovers state.-func (o *OffpeakScheduler) recoverMidWindow(ctx context.Context, date string) error {+// recoverMidWindow surfaces the pending row to the caller so handleEnd can+// use its StartE* fields for diagnostic snapshot population. Post T-1341 the+// method no longer rebuilds in-memory snapshot/SoC state — handleEnd+// integrates the readings table directly (Decision 2).+//+// Returns the pending row when one exists with status="pending"; nil when+// the row is absent, already complete, or the store query fails (logged at+// warn, not propagated — same defensive policy as before).+func (o *OffpeakScheduler) recoverMidWindow(ctx context.Context, date string) (*dynamo.OffpeakItem, error) { 	item, err := o.store.GetOffpeak(ctx, o.cfg.Serial, date) 	if err != nil { 		slog.Warn("offpeak mid-window recovery: store query failed", "date", date, "error", err)-		return nil // Log and skip, don't propagate.+		return nil, nil 	}- 	if item == nil || item.Status != dynamo.OffpeakStatusPending {-		return nil+		return nil, nil 	}+	slog.Info("offpeak: pending record confirmed for mid-window recovery", "date", date)+	return item, nil+} -	o.startSnapshot = &alphaess.EnergyData{-		Epv: item.StartEpv, EInput: item.StartEInput, EOutput: item.StartEOutput,-		ECharge: item.StartECharge, EDischarge: item.StartEDischarge, EGridCharge: item.StartEGridCharge,+// recoverAfterWindow handles the positionAfter restart path (T-1341 AC 3.4):+// when the poller starts up between offpeak-end and 24:00 with a pending row,+// run the integration path immediately so the day is finalised without+// waiting another 24 hours for the next start tick. When no row exists or+// the row is already complete, log+skip — no work to do.+//+// handleEnd internally skips the boundary-wait when offpeak-end is already+// in the past (which it is by definition on this path), so there's no need+// for an explicit override.+func (o *OffpeakScheduler) recoverAfterWindow(ctx context.Context, date string) error {+	item, err := o.store.GetOffpeak(ctx, o.cfg.Serial, date)+	if err != nil {+		slog.Warn("offpeak post-window recovery: store query failed", "date", date, "error", err)+		return nil+	}+	if item == nil {+		slog.Info("offpeak: past window with no pending row, skipping today", "date", date)+		return nil+	}+	if item.Status == dynamo.OffpeakStatusComplete {+		slog.Info("offpeak: past window with already-complete row, nothing to recover", "date", date)+		return nil+	}+	if err := o.handleEnd(ctx, date, item); err != nil {+		slog.Warn("offpeak post-window recovery: handleEnd failed", "date", date, "error", err) 	}-	o.socStart = item.SocStart-	o.hasStart = true--	slog.Info("offpeak: recovered pending record", "date", date, "socStart", o.socStart) 	return nil } -// handleEndOrCleanup attempts the end snapshot; on failure, deletes the pending record.-func (o *OffpeakScheduler) handleEndOrCleanup(ctx context.Context, date string) {-	if err := o.handleEnd(ctx, date); err != nil {+// handleEndOrCleanup attempts the end snapshot; on failure, deletes the+// pending record. The caller passes the pending row (from recoverMidWindow)+// when handleStart did not run in this process; otherwise nil and handleEnd+// uses in-memory state.+func (o *OffpeakScheduler) handleEndOrCleanup(ctx context.Context, date string, pending *dynamo.OffpeakItem) {+	if err := o.handleEnd(ctx, date, pending); err != nil { 		slog.Warn("offpeak end failed, deleting pending record", "date", date, "error", err) 		if delErr := o.store.DeleteOffpeak(ctx, o.cfg.Serial, date); delErr != nil { 			slog.Error("delete pending offpeak failed", "date", date, "error", delErr)@@ -252,7 +458,6 @@ func (o *OffpeakScheduler) handleEndOrCleanup(ctx context.Context, date string) func (o *OffpeakScheduler) resetState() { 	o.startSnapshot = nil 	o.socStart = 0-	o.hasStart = false }  // waitUntil blocks until the target time or context cancellation.@@ -270,6 +475,66 @@ func (o *OffpeakScheduler) waitUntil(ctx context.Context, target time.Time) bool 	} } +// waitForReadingAtOrAfter polls the readings table every pollInterval until+// either a reading with Timestamp >= target.Unix() exists or the budget+// expires (specs/offpeak-from-readings AC 3.1). Uses the strongly-consistent+// query so the observed reading is guaranteed to be visible to the+// subsequent integration query in handleEnd.+//+// Returns (true, nil) when a qualifying reading lands, (false, nil) on+// timeout, (false, err) on store error. Context cancellation returns+// (false, ctx.Err()).+//+// The helper uses real wall time (time.Now / time.After) intentionally — it+// polls live I/O, not orchestration state. The scheduler's injectable o.now+// governs "what date is it / when should I wake up to do the next thing";+// this helper's clock governs "have I waited long enough on the DB." Tests+// that freeze o.now to skip waitUntil still need the wait-helper deadline+// to advance, so the two clocks are kept distinct on purpose.+//+// The probe's upper bound slides forward each iteration to current real time+// so late-arriving readings beyond the originally-expected window are still+// caught. The lower bound is fixed at the target — anything older is+// irrelevant to the at-or-after check.+func (o *OffpeakScheduler) waitForReadingAtOrAfter(+	ctx context.Context,+	target time.Time,+	budget time.Duration,+	pollInterval time.Duration,+) (bool, error) {+	deadline := time.Now().Add(budget)+	targetUnix := target.Unix()+	for {+		from := targetUnix - 1+		to := time.Now().Unix() + 1+		if to < from {+			to = from+		}+		readings, err := o.store.QueryReadingsConsistent(ctx, o.cfg.Serial, from, to)+		if err != nil {+			return false, fmt.Errorf("query readings for boundary wait: %w", err)+		}+		for _, r := range readings {+			if r.Timestamp >= targetUnix {+				return true, nil+			}+		}+		remaining := time.Until(deadline)+		if remaining <= 0 {+			return false, nil+		}+		sleep := pollInterval+		if remaining < sleep {+			sleep = remaining+		}+		select {+		case <-ctx.Done():+			return false, ctx.Err()+		case <-time.After(sleep):+		}+	}+}+ // timePosition returns the current time's position relative to the off-peak window. func timePosition(now time.Time, start, end time.Duration) windowPosition { 	midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())@@ -292,22 +557,3 @@ func wallClockTime(day time.Time, loc *time.Location, d time.Duration) time.Time 	m := int(d.Minutes()) % 60 	return time.Date(local.Year(), local.Month(), local.Day(), h, m, 0, 0, loc) }--// computeOffpeakDeltas computes all energy deltas between start and end snapshots.-func computeOffpeakDeltas(serial, date string, start, end *alphaess.EnergyData, socStart, socEnd float64) dynamo.OffpeakItem {-	return dynamo.OffpeakItem{-		SysSn: serial, Date: date, Status: dynamo.OffpeakStatusComplete,-		StartEpv: start.Epv, StartEInput: start.EInput, StartEOutput: start.EOutput,-		StartECharge: start.ECharge, StartEDischarge: start.EDischarge, StartEGridCharge: start.EGridCharge,-		SocStart: socStart,-		EndEpv:   end.Epv, EndEInput: end.EInput, EndEOutput: end.EOutput,-		EndECharge: end.ECharge, EndEDischarge: end.EDischarge, EndEGridCharge: end.EGridCharge,-		SocEnd:              socEnd,-		GridUsageKwh:        end.EInput - start.EInput,-		SolarKwh:            end.Epv - start.Epv,-		BatteryChargeKwh:    end.ECharge - start.ECharge,-		BatteryDischargeKwh: end.EDischarge - start.EDischarge,-		GridExportKwh:       end.EOutput - start.EOutput,-		BatteryDeltaPercent: socEnd - socStart,-	}-}

diff --git a/internal/dynamo/dynamostore.go b/internal/dynamo/dynamostore.goindex 6a015ba..c323d2e 100644--- a/internal/dynamo/dynamostore.go+++ b/internal/dynamo/dynamostore.go@@ -2,6 +2,7 @@ package dynamo  import ( 	"context"+	"errors" 	"fmt" 	"log/slog" 	"strconv"@@ -142,16 +143,68 @@ func (s *DynamoStore) GetDailyEnergy(ctx context.Context, sysSn, date string) (*  // QueryReadings paginates the flux-readings table for the given serial and // timestamp range. Used by the daily-derived-stats summarisation pass.+// Reads are eventually-consistent — the poller's off-peak window-end pass+// uses QueryReadingsConsistent instead to avoid missing the at-or-after-+// boundary reading it has just confirmed. func (s *DynamoStore) QueryReadings(ctx context.Context, serial string, from, to int64) ([]ReadingItem, error) {-	return queryAll[ReadingItem](ctx, s.client, s.tables.Readings, "readings",-		"sysSn = :serial AND #ts BETWEEN :from AND :to",-		map[string]string{"#ts": "timestamp"},-		map[string]types.AttributeValue{+	return queryReadings(ctx, s.client, s.tables.Readings, serial, from, to, false)+}++// QueryReadingsConsistent is the strongly-consistent variant used by the+// off-peak window-end finalisation (AC 3.5). The poller has just observed+// that a reading at or after offpeak-end exists; a strongly-consistent read+// guarantees the integration query includes that reading.+func (s *DynamoStore) QueryReadingsConsistent(ctx context.Context, serial string, from, to int64) ([]ReadingItem, error) {+	return queryReadings(ctx, s.client, s.tables.Readings, serial, from, to, true)+}++// queryReadings is the shared body used by QueryReadings and+// QueryReadingsConsistent. The ScanIndexForward and pagination behaviour+// stay identical to queryAll's defaults; only ConsistentRead toggles.+func queryReadings(ctx context.Context, client interface {+	Query(ctx context.Context, params *dynamodb.QueryInput, optFns ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error)+}, table, serial string, from, to int64, consistent bool,+) ([]ReadingItem, error) {+	forward := true+	keyCond := "sysSn = :serial AND #ts BETWEEN :from AND :to"+	input := &dynamodb.QueryInput{+		TableName:                &table,+		KeyConditionExpression:   &keyCond,+		ExpressionAttributeNames: map[string]string{"#ts": "timestamp"},+		ExpressionAttributeValues: map[string]types.AttributeValue{ 			":serial": &types.AttributeValueMemberS{Value: serial}, 			":from":   &types.AttributeValueMemberN{Value: fmt.Sprintf("%d", from)}, 			":to":     &types.AttributeValueMemberN{Value: fmt.Sprintf("%d", to)}, 		},-	)+		ScanIndexForward: &forward,+	}+	if consistent {+		c := true+		input.ConsistentRead = &c+	}++	var items []ReadingItem+	for {+		out, err := client.Query(ctx, input)+		if err != nil {+			return nil, fmt.Errorf("query readings (table=%s): %w", table, err)+		}+		page := make([]ReadingItem, len(out.Items))+		for i, av := range out.Items {+			if err := attributevalue.UnmarshalMap(av, &page[i]); err != nil {+				return nil, fmt.Errorf("unmarshal readings (table=%s): %w", table, err)+			}+		}+		items = append(items, page...)+		if out.LastEvaluatedKey == nil {+			break+		}+		input.ExclusiveStartKey = out.LastEvaluatedKey+	}+	if items == nil {+		items = []ReadingItem{}+	}+	return items, nil }  // formatFloat formats a float64 for a DynamoDB N attribute value.@@ -222,6 +275,64 @@ func (s *DynamoStore) WriteOffpeak(ctx context.Context, item OffpeakItem) error 	return s.putItem(ctx, s.tables.Offpeak, item, fmt.Sprintf("offpeak (sysSn=%s, date=%s)", item.SysSn, item.Date)) } +// WriteOffpeakIfPendingOrAbsent writes the row only when no row exists yet+// OR the existing row has status="pending". A row with status="complete"+// causes the put to fail with ErrOffpeakConditionFailed — the poller's+// callers log+skip per design.md "Concurrent writer guard". The pending-row+// write from handleStart continues to use unconditional WriteOffpeak.+func (s *DynamoStore) WriteOffpeakIfPendingOrAbsent(ctx context.Context, item OffpeakItem) error {+	return s.writeOffpeakConditional(ctx, item,+		"attribute_not_exists(#status) OR #status = :pending",+		map[string]types.AttributeValue{+			":pending": &types.AttributeValueMemberS{Value: OffpeakStatusPending},+		},+	)+}++// WriteOffpeakIfComplete writes the row only when the existing row has+// status="complete". A missing row OR status="pending" causes the put to+// fail with ErrOffpeakConditionFailed — protects the backfill CLI against+// overwriting a row mid-poll.+func (s *DynamoStore) WriteOffpeakIfComplete(ctx context.Context, item OffpeakItem) error {+	return s.writeOffpeakConditional(ctx, item,+		"#status = :complete",+		map[string]types.AttributeValue{+			":complete": &types.AttributeValueMemberS{Value: OffpeakStatusComplete},+		},+	)+}++// writeOffpeakConditional marshals + PutItem with the given condition,+// mapping ConditionalCheckFailedException to ErrOffpeakConditionFailed. The+// #status alias decouples the expression text from any future rename of the+// `status` attribute and avoids the DynamoDB reserved-word check.+func (s *DynamoStore) writeOffpeakConditional(+	ctx context.Context,+	item OffpeakItem,+	condition string,+	values map[string]types.AttributeValue,+) error {+	av, err := attributevalue.MarshalMap(item)+	if err != nil {+		return fmt.Errorf("marshal offpeak (sysSn=%s, date=%s): %w", item.SysSn, item.Date, err)+	}+	_, err = s.client.PutItem(ctx, &dynamodb.PutItemInput{+		TableName:                 &s.tables.Offpeak,+		Item:                      av,+		ConditionExpression:       &condition,+		ExpressionAttributeNames:  map[string]string{"#status": "status"},+		ExpressionAttributeValues: values,+	})+	if err != nil {+		var ccf *types.ConditionalCheckFailedException+		if errors.As(err, &ccf) {+			return fmt.Errorf("offpeak (sysSn=%s, date=%s): %w", item.SysSn, item.Date, ErrOffpeakConditionFailed)+		}+		return fmt.Errorf("put conditional offpeak (table=%s, sysSn=%s, date=%s): %w", s.tables.Offpeak, item.SysSn, item.Date, err)+	}+	return nil+}+ func (s *DynamoStore) DeleteOffpeak(ctx context.Context, serial, date string) error { 	_, err := s.client.DeleteItem(ctx, &dynamodb.DeleteItemInput{ 		TableName: &s.tables.Offpeak,

diff --git a/internal/dynamo/offpeak_drift.go b/internal/dynamo/offpeak_drift.gonew file mode 100644index 0000000..6e973bb--- /dev/null+++ b/internal/dynamo/offpeak_drift.go@@ -0,0 +1,66 @@+package dynamo++import (+	"log/slog"+	"math"+)++// LogOffpeakDrift emits a structured INFO log entry comparing the snapshot-+// diff value (endE* − startE*) with the readings-integrated value for each+// of the five deltas, plus the absolute difference (specs/offpeak-from-readings+// AC 6.1 / 6.2).+//+// Lives in the dynamo package so both the poller (handleEnd) and the backfill+// CLI can call it without introducing a poller → CLI import; OffpeakItem is+// already defined here.+//+// The emission goes through slog.Info — production wires up slog.JSONHandler+// (cmd/poller/logging.go), CLIs use slog.TextHandler. Both formats are parseable+// by CloudWatch Logs Insights via the `fields date, driftGrid, …` syntax.+// No metric/alarm is emitted; alerting is deferred per Decision 6.+//+// When the row was finalised via positionAfter recovery the end snapshot is+// missing (handleStart ran but handleEnd never recorded the AlphaESS end+// values before the restart), so EndE* are all zero. In that case the+// snapshot-diff numbers would collapse to -startE* which is misleading, so a+// dedicated log line is emitted with the integrated values only.+func LogOffpeakDrift(date string, item OffpeakItem) {+	if item.EndEInput == 0 && item.EndEpv == 0 && item.EndECharge == 0 &&+		item.EndEDischarge == 0 && item.EndEOutput == 0 {+		slog.Info("offpeak_drift_no_snapshot_pair",+			"date", date,+			"reason", "positionAfterRecovery",+			"integratedGrid", item.GridUsageKwh,+			"integratedSolar", item.SolarKwh,+			"integratedCharge", item.BatteryChargeKwh,+			"integratedDischarge", item.BatteryDischargeKwh,+			"integratedExport", item.GridExportKwh,+		)+		return+	}++	snapGrid := item.EndEInput - item.StartEInput+	snapSolar := item.EndEpv - item.StartEpv+	snapCharge := item.EndECharge - item.StartECharge+	snapDischarge := item.EndEDischarge - item.StartEDischarge+	snapExport := item.EndEOutput - item.StartEOutput++	slog.Info("offpeak drift",+		"date", date,+		"snapshotGrid", snapGrid,+		"integratedGrid", item.GridUsageKwh,+		"driftGrid", math.Abs(item.GridUsageKwh-snapGrid),+		"snapshotSolar", snapSolar,+		"integratedSolar", item.SolarKwh,+		"driftSolar", math.Abs(item.SolarKwh-snapSolar),+		"snapshotCharge", snapCharge,+		"integratedCharge", item.BatteryChargeKwh,+		"driftCharge", math.Abs(item.BatteryChargeKwh-snapCharge),+		"snapshotDischarge", snapDischarge,+		"integratedDischarge", item.BatteryDischargeKwh,+		"driftDischarge", math.Abs(item.BatteryDischargeKwh-snapDischarge),+		"snapshotExport", snapExport,+		"integratedExport", item.GridExportKwh,+		"driftExport", math.Abs(item.GridExportKwh-snapExport),+	)+}

diff --git a/internal/dynamo/models.go b/internal/dynamo/models.goindex d63b674..7b29fd2 100644--- a/internal/dynamo/models.go+++ b/internal/dynamo/models.go@@ -148,6 +148,15 @@ type OffpeakItem struct { 	BatteryDischargeKwh float64 `dynamodbav:"batteryDischargeKwh"` 	GridExportKwh       float64 `dynamodbav:"gridExportKwh"` 	BatteryDeltaPercent float64 `dynamodbav:"batteryDeltaPercent"`++	// Integration provenance (AC 5.4). Populated when the five deltas are+	// computed via derivedstats.IntegrateOffpeakDeltas (poller window-end+	// and backfill CLI). `omitempty` keeps pre-feature rows clean of zero-+	// valued fields and lets the marshalling round-trip past rows safely.+	// No API consumer reads these fields; they exist for operator diagnostics.+	IntegrationSampleCount  int    `dynamodbav:"integrationSampleCount,omitempty"`+	IntegrationSkippedPairs int    `dynamodbav:"integrationSkippedPairs,omitempty"`+	IntegratedAt            string `dynamodbav:"integratedAt,omitempty"` // RFC3339 UTC }  // NewReadingItem transforms AlphaESS power data into a DynamoDB reading item.

diff --git a/cmd/backfill-offpeak/main.go b/cmd/backfill-offpeak/main.gonew file mode 100644index 0000000..4c79547--- /dev/null+++ b/cmd/backfill-offpeak/main.go@@ -0,0 +1,390 @@+// Package main is the standalone backfill CLI for the off-peak from readings+// feature (T-1341).+//+// It walks flux-offpeak rows in a date range and recomputes the five energy+// deltas (gridUsageKwh, solarKwh, batteryChargeKwh, batteryDischargeKwh,+// gridExportKwh) by integrating the corresponding power channels from+// flux-readings over the SSM off-peak window. Writes use+// WriteOffpeakIfComplete so a row mid-poll (pending or absent) is never+// overwritten (AC 7.8). Today's row is always skipped — the poller is the+// single authoritative writer for today (AC 7.2).+//+// Usage (with operator AWS credentials):+//+//	go run ./cmd/backfill-offpeak \+//	    --serial=AB1234 \+//	    --from=2026-04-19 --to=2026-05-18 \+//	    --offpeak-start=11:00 --offpeak-end=14:00 \+//	    --table-offpeak=flux-offpeak \+//	    --table-readings=flux-readings \+//	    [--dry-run]+//+// Defaults: from = today - 30d, to = yesterday (the practical readings TTL+// window). Set --dry-run to log intended writes and print summaries without+// invoking PutItem.+package main++import (+	"context"+	"errors"+	"flag"+	"fmt"+	"log/slog"+	"math"+	"os"+	"strconv"+	"time"++	_ "time/tzdata"++	awsconfig "github.com/aws/aws-sdk-go-v2/config"+	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"+	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"++	"github.com/ArjenSchwarz/flux/internal/derivedstats"+	"github.com/ArjenSchwarz/flux/internal/dynamo"+)++// dynamoAPI is the subset of the DynamoDB client this CLI uses. It mirrors+// dynamo.DynamoAPI so the CLI can construct a real *dynamo.DynamoStore from+// the same client and route conditional writes through the canonical+// condition-expression source (see writeOffpeakIfComplete in dynamostore.go).+type dynamoAPI interface {+	Query(ctx context.Context, params *dynamodb.QueryInput, optFns ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error)+	PutItem(ctx context.Context, params *dynamodb.PutItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error)+	DeleteItem(ctx context.Context, params *dynamodb.DeleteItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.DeleteItemOutput, error)+	GetItem(ctx context.Context, params *dynamodb.GetItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.GetItemOutput, error)+	UpdateItem(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error)+	BatchWriteItem(ctx context.Context, params *dynamodb.BatchWriteItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.BatchWriteItemOutput, error)+}++type backfillOpts struct {+	serial        string+	tableOffpeak  string+	tableReadings string+	from          string+	to            string+	offpeakStart  string+	offpeakEnd    string+	location      *time.Location+	dryRun        bool+	now           func() time.Time+}++type backfillResult struct {+	RowsScanned         int+	RowsSkipped         int // today's row skipped per AC 7.2+	RowsWritten         int+	RowsDryRun          int+	RowsSparseSkipped   int // <2 usable readings in window per AC 7.4+	RowsConditionFailed int // WriteOffpeakIfComplete condition rejected+	Summary             []string+}++func main() {+	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, nil)))++	loc, err := time.LoadLocation("Australia/Sydney")+	if err != nil {+		slog.Error("load Sydney timezone", "error", err)+		os.Exit(1)+	}+	today := time.Now().In(loc)+	defaultTo := today.AddDate(0, 0, -1).Format("2006-01-02")+	defaultFrom := today.AddDate(0, 0, -30).Format("2006-01-02")++	opts := backfillOpts{location: loc, now: time.Now}+	flag.StringVar(&opts.serial, "serial", os.Getenv("SYSTEM_SERIAL"), "AlphaESS system serial number (or env SYSTEM_SERIAL)")+	flag.StringVar(&opts.tableOffpeak, "table-offpeak", os.Getenv("TABLE_OFFPEAK"), "flux-offpeak table name (or env TABLE_OFFPEAK)")+	flag.StringVar(&opts.tableReadings, "table-readings", os.Getenv("TABLE_READINGS"), "flux-readings table name (or env TABLE_READINGS)")+	flag.StringVar(&opts.from, "from", defaultFrom, "start date inclusive (YYYY-MM-DD, Sydney TZ)")+	flag.StringVar(&opts.to, "to", defaultTo, "end date inclusive (YYYY-MM-DD, Sydney TZ)")+	flag.StringVar(&opts.offpeakStart, "offpeak-start", os.Getenv("OFFPEAK_START"), "off-peak window start HH:MM (or env OFFPEAK_START)")+	flag.StringVar(&opts.offpeakEnd, "offpeak-end", os.Getenv("OFFPEAK_END"), "off-peak window end HH:MM (or env OFFPEAK_END)")+	flag.BoolVar(&opts.dryRun, "dry-run", false, "log intended writes without invoking PutItem")+	flag.Parse()++	if err := validateOpts(opts); err != nil {+		slog.Error("invalid options", "error", err)+		os.Exit(2)+	}++	ctx := context.Background()+	awsCfg, err := awsconfig.LoadDefaultConfig(ctx)+	if err != nil {+		slog.Error("load AWS config", "error", err)+		os.Exit(1)+	}+	client := dynamodb.NewFromConfig(awsCfg)++	res, err := runBackfill(ctx, client, opts)+	if err != nil {+		slog.Error("backfill failed", "error", err)+		os.Exit(1)+	}+	for _, line := range res.Summary {+		fmt.Println(line)+	}+	slog.Info("backfill complete",+		"scanned", res.RowsScanned,+		"written", res.RowsWritten,+		"dryRun", res.RowsDryRun,+		"todaySkipped", res.RowsSkipped,+		"sparseSkipped", res.RowsSparseSkipped,+		"conditionFailed", res.RowsConditionFailed,+	)+}++func validateOpts(o backfillOpts) error {+	if o.serial == "" {+		return fmt.Errorf("--serial is required")+	}+	if o.tableOffpeak == "" {+		return fmt.Errorf("--table-offpeak is required")+	}+	if o.tableReadings == "" {+		return fmt.Errorf("--table-readings is required")+	}+	if o.offpeakStart == "" || o.offpeakEnd == "" {+		return fmt.Errorf("--offpeak-start and --offpeak-end are required")+	}+	from, err := time.ParseInLocation("2006-01-02", o.from, o.location)+	if err != nil {+		return fmt.Errorf("invalid --from %q: %w", o.from, err)+	}+	to, err := time.ParseInLocation("2006-01-02", o.to, o.location)+	if err != nil {+		return fmt.Errorf("invalid --to %q: %w", o.to, err)+	}+	if from.After(to) {+		return fmt.Errorf("--from %s is after --to %s", o.from, o.to)+	}+	if _, _, ok := derivedstats.ParseOffpeakWindow(o.offpeakStart, o.offpeakEnd); !ok {+		return fmt.Errorf("invalid --offpeak-start %q / --offpeak-end %q", o.offpeakStart, o.offpeakEnd)+	}+	return nil+}++// runBackfill is the testable core: it scans the offpeak table for the+// configured date range, recomputes the five deltas via+// derivedstats.IntegrateOffpeakDeltas, emits a drift log entry, and writes+// the patched row via the WriteOffpeakIfComplete conditional put. Today's+// row (per Sydney-local opts.now) is always skipped (AC 7.2). Days with+// fewer than two usable readings in the window emit a SKIPPED summary line+// and are left unchanged (AC 7.4).+func runBackfill(ctx context.Context, client dynamoAPI, opts backfillOpts) (*backfillResult, error) {+	res := &backfillResult{}+	rows, err := queryOffpeakRange(ctx, client, opts.tableOffpeak, opts.serial, opts.from, opts.to)+	if err != nil {+		return nil, fmt.Errorf("query offpeak (%s): %w", opts.tableOffpeak, err)+	}++	today := opts.now().In(opts.location).Format("2006-01-02")++	for _, row := range rows {+		res.RowsScanned++++		if row.Date == today {+			res.RowsSkipped+++			slog.Info("skip: today's row is the poller's responsibility", "date", row.Date)+			continue+		}++		day, err := time.ParseInLocation("2006-01-02", row.Date, opts.location)+		if err != nil {+			res.RowsSkipped+++			slog.Warn("skip: invalid row date", "date", row.Date, "error", err)+			continue+		}+		windowStart, windowEnd, ok := offpeakBoundaries(day, opts.location, opts.offpeakStart, opts.offpeakEnd)+		if !ok {+			res.RowsSkipped+++			slog.Warn("skip: invalid offpeak window", "date", row.Date)+			continue+		}++		readings, err := queryReadingsRange(ctx, client, opts.tableReadings, opts.serial,+			windowStart.Unix(), windowEnd.Unix())+		if err != nil {+			return res, fmt.Errorf("query readings (date=%s): %w", row.Date, err)+		}++		deltas, ok := derivedstats.IntegrateOffpeakDeltas(+			toDerivedReadings(readings), windowStart.Unix(), windowEnd.Unix())+		if !ok {+			res.RowsSparseSkipped+++			line := fmt.Sprintf("%s  SKIPPED (sparse readings; <2 usable samples in window)", row.Date)+			res.Summary = append(res.Summary, line)+			slog.Info("skip: sparse readings", "date", row.Date, "readings", len(readings))+			continue+		}++		patched := patchOffpeakRow(row, deltas, opts.now().UTC())+		summary := summaryLine(row.Date, row, patched)+		res.Summary = append(res.Summary, summary)++		dynamo.LogOffpeakDrift(row.Date, patched)++		if opts.dryRun {+			res.RowsDryRun+++			slog.Info("dry-run patch", "date", row.Date,+				"gridUsageKwh", patched.GridUsageKwh,+				"solarKwh", patched.SolarKwh,+				"sampleCount", patched.IntegrationSampleCount)+			continue+		}++		store := dynamo.NewDynamoStore(client, dynamo.TableNames{Offpeak: opts.tableOffpeak})+		if err := store.WriteOffpeakIfComplete(ctx, patched); err != nil {+			if errors.Is(err, dynamo.ErrOffpeakConditionFailed) {+				res.RowsConditionFailed+++				slog.Warn("conditional-write rejected (row state changed); skipping", "date", row.Date)+				continue+			}+			return res, fmt.Errorf("write offpeak (date=%s): %w", row.Date, err)+		}+		res.RowsWritten+++		slog.Info("patched offpeak deltas", "date", row.Date,+			"gridUsageKwh", patched.GridUsageKwh,+			"sampleCount", patched.IntegrationSampleCount)+	}++	return res, nil+}++// offpeakBoundaries returns the absolute Sydney-local times of the off-peak+// window on the given day. False indicates an unparseable window.+func offpeakBoundaries(day time.Time, loc *time.Location, start, end string) (time.Time, time.Time, bool) {+	startMin, endMin, ok := derivedstats.ParseOffpeakWindow(start, end)+	if !ok {+		return time.Time{}, time.Time{}, false+	}+	midnight := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, loc)+	return midnight.Add(time.Duration(startMin) * time.Minute),+		midnight.Add(time.Duration(endMin) * time.Minute), true+}++// patchOffpeakRow returns a copy of stored with the five integration-sourced+// deltas, the three provenance fields, and status=complete. Every other+// field — the diagnostic startE*/endE* snapshots, socStart/socEnd,+// batteryDeltaPercent — is preserved verbatim per Decision 2.+func patchOffpeakRow(stored dynamo.OffpeakItem, deltas derivedstats.OffpeakDeltas, integratedAt time.Time) dynamo.OffpeakItem {+	out := stored+	out.Status = dynamo.OffpeakStatusComplete+	out.GridUsageKwh = derivedstats.RoundEnergy(deltas.GridImportKwh)+	out.SolarKwh = derivedstats.RoundEnergy(deltas.SolarKwh)+	out.BatteryChargeKwh = derivedstats.RoundEnergy(deltas.BatteryChargeKwh)+	out.BatteryDischargeKwh = derivedstats.RoundEnergy(deltas.BatteryDischargeKwh)+	out.GridExportKwh = derivedstats.RoundEnergy(deltas.GridExportKwh)+	out.IntegrationSampleCount = deltas.SampleCount+	out.IntegrationSkippedPairs = deltas.SkippedPairs+	out.IntegratedAt = integratedAt.Format(time.RFC3339)+	return out+}++// summaryLine formats a per-day comparison of the prior row and the patched+// row so the operator can sanity-check the magnitude of each correction+// before it propagates to the clients (AC 7.5). Surfaces all five deltas+// (grid import, solar, battery charge, battery discharge, grid export) with+// the prior value, the new value, and the absolute difference so the+// operator can see at a glance which channels moved most.+func summaryLine(date string, prev, next dynamo.OffpeakItem) string {+	col := func(label string, p, n float64) string {+		return fmt.Sprintf("%s %.2f→%.2f |Δ|=%.2f", label, p, n, math.Abs(n-p))+	}+	return fmt.Sprintf(+		"%s  %s  %s  %s  %s  %s  samples=%d skipped=%d",+		date,+		col("grid", prev.GridUsageKwh, next.GridUsageKwh),+		col("solar", prev.SolarKwh, next.SolarKwh),+		col("chg", prev.BatteryChargeKwh, next.BatteryChargeKwh),+		col("dis", prev.BatteryDischargeKwh, next.BatteryDischargeKwh),+		col("exp", prev.GridExportKwh, next.GridExportKwh),+		next.IntegrationSampleCount,+		next.IntegrationSkippedPairs,+	)+}++// queryOffpeakRange paginates flux-offpeak for one serial and a closed date+// range. Mirrors dynamo.DynamoReader.QueryOffpeak but uses only the small+// dynamoAPI surface so the CLI's tests don't have to satisfy the full Reader.+func queryOffpeakRange(ctx context.Context, client dynamoAPI, table, serial, from, to string) ([]dynamo.OffpeakItem, error) {+	keyCondition := "sysSn = :serial AND #d BETWEEN :start AND :end"+	exprNames := map[string]string{"#d": "date"}+	exprValues := map[string]types.AttributeValue{+		":serial": &types.AttributeValueMemberS{Value: serial},+		":start":  &types.AttributeValueMemberS{Value: from},+		":end":    &types.AttributeValueMemberS{Value: to},+	}+	return paginate[dynamo.OffpeakItem](ctx, client, table, keyCondition, exprNames, exprValues)+}++// queryReadingsRange paginates flux-readings for one serial and a closed+// timestamp range.+func queryReadingsRange(ctx context.Context, client dynamoAPI, table, serial string, fromTS, toTS int64) ([]dynamo.ReadingItem, error) {+	keyCondition := "sysSn = :serial AND #ts BETWEEN :from AND :to"+	exprNames := map[string]string{"#ts": "timestamp"}+	exprValues := map[string]types.AttributeValue{+		":serial": &types.AttributeValueMemberS{Value: serial},+		":from":   &types.AttributeValueMemberN{Value: strconv.FormatInt(fromTS, 10)},+		":to":     &types.AttributeValueMemberN{Value: strconv.FormatInt(toTS, 10)},+	}+	return paginate[dynamo.ReadingItem](ctx, client, table, keyCondition, exprNames, exprValues)+}++func paginate[T any](+	ctx context.Context,+	client dynamoAPI,+	table, keyCondition string,+	exprNames map[string]string,+	exprValues map[string]types.AttributeValue,+) ([]T, error) {+	forward := true+	input := &dynamodb.QueryInput{+		TableName:                 &table,+		KeyConditionExpression:    &keyCondition,+		ExpressionAttributeValues: exprValues,+		ScanIndexForward:          &forward,+	}+	if len(exprNames) > 0 {+		input.ExpressionAttributeNames = exprNames+	}+	var out []T+	for {+		page, err := client.Query(ctx, input)+		if err != nil {+			return nil, err+		}+		decoded := make([]T, len(page.Items))+		for i, av := range page.Items {+			if err := attributevalue.UnmarshalMap(av, &decoded[i]); err != nil {+				return nil, fmt.Errorf("unmarshal %s row: %w", table, err)+			}+		}+		out = append(out, decoded...)+		if page.LastEvaluatedKey == nil {+			break+		}+		input.ExclusiveStartKey = page.LastEvaluatedKey+	}+	return out, nil+}++// toDerivedReadings mirrors api.toDerivedReadings / poller.summaryToDerivedReadings+// — kept duplicated for the same Decision 9 reason: derivedstats must not+// import dynamo, and this is the only consumer in cmd/backfill-offpeak.+func toDerivedReadings(in []dynamo.ReadingItem) []derivedstats.Reading {+	out := make([]derivedstats.Reading, len(in))+	for i, r := range in {+		out[i] = derivedstats.Reading{+			Timestamp: r.Timestamp,+			Ppv:       r.Ppv,+			Pload:     r.Pload,+			Soc:       r.Soc,+			Pbat:      r.Pbat,+			Pgrid:     r.Pgrid,+		}+	}+	return out+}

diff --git a/cmd/backfill-offpeak/main_test.go b/cmd/backfill-offpeak/main_test.gonew file mode 100644index 0000000..f21e316--- /dev/null+++ b/cmd/backfill-offpeak/main_test.go@@ -0,0 +1,551 @@+package main++import (+	"bytes"+	"context"+	"errors"+	"fmt"+	"log/slog"+	"math"+	"strconv"+	"testing"+	"time"++	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"+	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"+	"github.com/stretchr/testify/assert"+	"github.com/stretchr/testify/require"++	"github.com/ArjenSchwarz/flux/internal/derivedstats"+	"github.com/ArjenSchwarz/flux/internal/dynamo"+)++const (+	testSerial        = "TEST123"+	testOffpeakTable  = "flux-offpeak-test"+	testReadingsTable = "flux-readings-test"+)++// fakeDynamo is a lightweight in-memory stand-in for the DynamoDB client.+// It returns canned offpeak and readings rows from Query, and records every+// PutItem call so tests can assert on the persisted payload and condition+// expression.+type fakeDynamo struct {+	offpeakRows      map[string][]dynamo.OffpeakItem // keyed by "*"+	readingsByDate   map[string][]dynamo.ReadingItem // keyed by Sydney YYYY-MM-DD+	location         *time.Location+	queries          []*dynamodb.QueryInput+	puts             []*dynamodb.PutItemInput+	queryErrForTable map[string]error+	putErr           error+}++func (f *fakeDynamo) Query(_ context.Context, params *dynamodb.QueryInput, _ ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error) {+	f.queries = append(f.queries, params)+	if err, ok := f.queryErrForTable[*params.TableName]; ok && err != nil {+		return nil, err+	}+	switch *params.TableName {+	case testOffpeakTable:+		return marshalQueryItems(f.offpeakRows["*"])+	case testReadingsTable:+		date := readingsQueryDate(params, f.location)+		return marshalQueryItems(f.readingsByDate[date])+	}+	return &dynamodb.QueryOutput{}, nil+}++func (f *fakeDynamo) PutItem(_ context.Context, params *dynamodb.PutItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.PutItemOutput, error) {+	f.puts = append(f.puts, params)+	if f.putErr != nil {+		return nil, f.putErr+	}+	return &dynamodb.PutItemOutput{}, nil+}++// The CLI now constructs a *dynamo.DynamoStore (Fix 6 of the pre-push review)+// so the conditional-write expression lives in one place. The store's+// interface (dynamo.DynamoAPI) requires Delete/Get/Update/BatchWrite — the CLI+// itself never calls them, so the fake returns benign zero values.++func (f *fakeDynamo) DeleteItem(_ context.Context, _ *dynamodb.DeleteItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.DeleteItemOutput, error) {+	return &dynamodb.DeleteItemOutput{}, nil+}++func (f *fakeDynamo) GetItem(_ context.Context, _ *dynamodb.GetItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.GetItemOutput, error) {+	return &dynamodb.GetItemOutput{}, nil+}++func (f *fakeDynamo) UpdateItem(_ context.Context, _ *dynamodb.UpdateItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.UpdateItemOutput, error) {+	return &dynamodb.UpdateItemOutput{}, nil+}++func (f *fakeDynamo) BatchWriteItem(_ context.Context, _ *dynamodb.BatchWriteItemInput, _ ...func(*dynamodb.Options)) (*dynamodb.BatchWriteItemOutput, error) {+	return &dynamodb.BatchWriteItemOutput{}, nil+}++// readingsQueryDate inverts the :from timestamp on a readings Query back to+// the Sydney YYYY-MM-DD date the fixture is keyed on. The CLI queries+// readings with [windowStart, windowEnd-1] in Sydney TZ, so the from value+// lands inside the off-peak window of that date.+func readingsQueryDate(params *dynamodb.QueryInput, loc *time.Location) string {+	from := params.ExpressionAttributeValues[":from"].(*types.AttributeValueMemberN).Value+	ts, err := strconv.ParseInt(from, 10, 64)+	if err != nil {+		panic("readingsQueryDate: malformed :from value " + from + ": " + err.Error())+	}+	return time.Unix(ts, 0).In(loc).Format("2006-01-02")+}++func marshalQueryItems[T any](items []T) (*dynamodb.QueryOutput, error) {+	avs := make([]map[string]types.AttributeValue, 0, len(items))+	for i := range items {+		av, err := attributevalue.MarshalMap(items[i])+		if err != nil {+			return nil, err+		}+		avs = append(avs, av)+	}+	return &dynamodb.QueryOutput{Items: avs}, nil+}++// chargeReadings builds dense (10s) readings across the 11:00-14:00 Sydney+// window with a constant 4 kW grid import — exercises a heavy-charge day so+// the integration produces a non-trivial gridUsageKwh.+func chargeReadings(t *testing.T, date string, loc *time.Location) []dynamo.ReadingItem {+	t.Helper()+	day, err := time.ParseInLocation("2006-01-02", date, loc)+	require.NoError(t, err)+	start := day.Add(11 * time.Hour)+	end := day.Add(14 * time.Hour)+	out := make([]dynamo.ReadingItem, 0, 3*60*60/10+1)+	for ts := start.Unix(); ts <= end.Unix(); ts += 10 {+		out = append(out, dynamo.ReadingItem{+			SysSn:     testSerial,+			Timestamp: ts,+			Pgrid:     4000,+			Pbat:      -3800,+			Ppv:       0,+			Soc:       30,+		})+	}+	return out+}++// existingPendingRow returns a row in pending state — must NOT be overwritten+// by the CLI per AC 7.8.+func existingPendingRow(date string) dynamo.OffpeakItem {+	return dynamo.OffpeakItem{+		SysSn:        testSerial,+		Date:         date,+		Status:       dynamo.OffpeakStatusPending,+		StartEInput:  10.0,+		StartEpv:     0,+		StartECharge: 5.0,+	}+}++// existingCompleteRow returns a row in complete state with the lagged+// snapshot-diff values from the bug. The CLI overwrites this with+// readings-integrated values.+func existingCompleteRow(date string) dynamo.OffpeakItem {+	return dynamo.OffpeakItem{+		SysSn:               testSerial,+		Date:                date,+		Status:              dynamo.OffpeakStatusComplete,+		StartEInput:         10.0,+		EndEInput:           28.95,+		StartEpv:            0,+		EndEpv:              0,+		StartECharge:        5.0,+		EndECharge:          18.0,+		StartEDischarge:     2.0,+		EndEDischarge:       2.5,+		StartEOutput:        1.0,+		EndEOutput:          1.5,+		GridUsageKwh:        18.95, // lagged snapshot-diff+		SolarKwh:            0,+		BatteryChargeKwh:    13.0,+		BatteryDischargeKwh: 0.5,+		GridExportKwh:       0.5,+		BatteryDeltaPercent: 60,+	}+}++func sydney(t *testing.T) *time.Location {+	t.Helper()+	loc, err := time.LoadLocation("Australia/Sydney")+	require.NoError(t, err)+	return loc+}++func backfillOptsForTest(loc *time.Location, dates ...string) backfillOpts {+	from := dates[0]+	to := dates[0]+	if len(dates) > 1 {+		to = dates[len(dates)-1]+	}+	// Pin clock so the today-skip gate (AC 7.2) is stable. "Today" is+	// 2026-05-20 here so the historical dates 2026-05-18/19 are past.+	now := time.Date(2026, 5, 20, 12, 0, 0, 0, loc)+	return backfillOpts{+		serial:        testSerial,+		tableOffpeak:  testOffpeakTable,+		tableReadings: testReadingsTable,+		from:          from,+		to:            to,+		offpeakStart:  "11:00",+		offpeakEnd:    "14:00",+		location:      loc,+		now:           func() time.Time { return now },+	}+}++// captureSlog swaps slog.Default for a JSON-buffered logger so tests can+// assert log lines without coupling to the binary's output handler.+func captureSlog() (*bytes.Buffer, func()) {+	var buf bytes.Buffer+	logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))+	old := slog.Default()+	slog.SetDefault(logger)+	return &buf, func() { slog.SetDefault(old) }+}++func TestBackfill_DryRun_NoPutItemCalled(t *testing.T) {+	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	f := &fakeDynamo{+		location:       loc,+		offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {row}},+		readingsByDate: map[string][]dynamo.ReadingItem{date: chargeReadings(t, date, loc)},+	}+	opts := backfillOptsForTest(loc, date)+	opts.dryRun = true++	res, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)+	assert.Empty(t, f.puts, "dry-run must not invoke PutItem")+	assert.Equal(t, 1, res.RowsScanned)+	assert.Equal(t, 1, res.RowsDryRun)+	assert.Zero(t, res.RowsWritten)+	assert.NotEmpty(t, res.Summary, "dry-run should emit at least one summary line")+}++func TestBackfill_LiveRun_WritesRecomputedDeltas(t *testing.T) {+	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	readings := chargeReadings(t, date, loc)+	f := &fakeDynamo{+		location:       loc,+		offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {row}},+		readingsByDate: map[string][]dynamo.ReadingItem{date: readings},+	}+	opts := backfillOptsForTest(loc, date)++	res, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)+	require.Len(t, f.puts, 1, "live run must call PutItem exactly once")+	assert.Equal(t, 1, res.RowsWritten)++	put := f.puts[0]+	require.NotNil(t, put.ConditionExpression)+	assert.Contains(t, *put.ConditionExpression, "#status = :complete",+		"PutItem must use WriteOffpeakIfComplete's condition (AC 7.8)")++	persisted := decodeOffpeakItem(t, put.Item)+	assert.Equal(t, dynamo.OffpeakStatusComplete, persisted.Status)+	// 4000W × 10800s = 43_200_000Ws = 12.0 kWh exactly.+	assert.InDelta(t, 12.0, persisted.GridUsageKwh, 0.01)+	assert.InDelta(t, 11.4, persisted.BatteryChargeKwh, 0.01)+	// Half-open [start, end): the reading at exactly endUnix is not counted as+	// an interior sample. chargeReadings emits 1081 timestamps from start to+	// end inclusive; SampleCount reflects the 1080 inside the window.+	assert.Equal(t, 1080, persisted.IntegrationSampleCount)+	assert.Equal(t, 0, persisted.IntegrationSkippedPairs)+	assert.NotEmpty(t, persisted.IntegratedAt)+	// Diagnostic snapshot preserved verbatim from the existing row (Decision 2).+	assert.Equal(t, row.StartEInput, persisted.StartEInput)+	assert.Equal(t, row.EndEInput, persisted.EndEInput)+}++func TestBackfill_Idempotent_DeltaFieldsBitEqualAcrossRuns(t *testing.T) {+	// AC 7.3 + AC 7.7: re-running the CLI against the same readings produces+	// identical values for the five delta fields and the two count fields.+	// integratedAt MAY differ — Decision 10 — so we pin two distinct clocks+	// for the two runs and assert the deltas match exactly while integratedAt+	// reflects each run's time.+	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	readings := chargeReadings(t, date, loc)++	clock1 := time.Date(2026, 5, 20, 12, 0, 0, 0, loc)+	clock2 := clock1.Add(time.Hour)++	run := func(now time.Time) dynamo.OffpeakItem {+		f := &fakeDynamo{+			location:       loc,+			offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {row}},+			readingsByDate: map[string][]dynamo.ReadingItem{date: readings},+		}+		opts := backfillOptsForTest(loc, date)+		opts.now = func() time.Time { return now }+		_, err := runBackfill(context.Background(), f, opts)+		require.NoError(t, err)+		require.Len(t, f.puts, 1)+		return decodeOffpeakItem(t, f.puts[0].Item)+	}++	first := run(clock1)+	second := run(clock2)++	assert.Equal(t, first.GridUsageKwh, second.GridUsageKwh)+	assert.Equal(t, first.SolarKwh, second.SolarKwh)+	assert.Equal(t, first.BatteryChargeKwh, second.BatteryChargeKwh)+	assert.Equal(t, first.BatteryDischargeKwh, second.BatteryDischargeKwh)+	assert.Equal(t, first.GridExportKwh, second.GridExportKwh)+	assert.Equal(t, first.IntegrationSampleCount, second.IntegrationSampleCount)+	assert.Equal(t, first.IntegrationSkippedPairs, second.IntegrationSkippedPairs)+	// integratedAt is excluded from the idempotence guarantee per Decision 10.+	assert.NotEqual(t, first.IntegratedAt, second.IntegratedAt,+		"integratedAt is the time of integration; the two runs were pinned to different clocks")+}++func TestBackfill_RoundingConsistency_PersistedFieldsRoundedTwoDecimals(t *testing.T) {+	// AC 7.7: the persisted delta values are rounded to two decimal places so+	// the poller and the backfill CLI produce byte-equal values for the same+	// readings. A reading series with sub-millikWh precision would otherwise+	// drift across runs once the values are compared via marshalled rows.+	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	readings := chargeReadings(t, date, loc)+	f := &fakeDynamo{+		location:       loc,+		offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {row}},+		readingsByDate: map[string][]dynamo.ReadingItem{date: readings},+	}+	opts := backfillOptsForTest(loc, date)+	_, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)+	require.Len(t, f.puts, 1)+	persisted := decodeOffpeakItem(t, f.puts[0].Item)+	for _, v := range []float64{+		persisted.GridUsageKwh,+		persisted.SolarKwh,+		persisted.BatteryChargeKwh,+		persisted.BatteryDischargeKwh,+		persisted.GridExportKwh,+	} {+		assert.Equal(t, derivedstats.RoundEnergy(v), v,+			"persisted delta %v must already be rounded to 2 decimal places", v)+	}+}++func TestBackfill_DefaultTo_SkipsToday(t *testing.T) {+	// AC 7.2: --to defaults to yesterday; today's row must never be processed+	// even when the query result includes a row dated today (e.g. the poller+	// pre-staged a pending row at offpeak-start).+	loc := sydney(t)+	today := time.Date(2026, 5, 20, 12, 0, 0, 0, loc).Format("2006-01-02")+	yesterday := time.Date(2026, 5, 19, 0, 0, 0, 0, loc).Format("2006-01-02")+	yesterdayRow := existingCompleteRow(yesterday)+	todayRow := existingPendingRow(today)+	f := &fakeDynamo{+		location: loc,+		offpeakRows: map[string][]dynamo.OffpeakItem{"*": {+			yesterdayRow, todayRow,+		}},+		readingsByDate: map[string][]dynamo.ReadingItem{+			yesterday: chargeReadings(t, yesterday, loc),+			today:     chargeReadings(t, today, loc),+		},+	}+	opts := backfillOptsForTest(loc, yesterday)+	res, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)+	require.Len(t, f.puts, 1, "only yesterday's row must be written")+	persisted := decodeOffpeakItem(t, f.puts[0].Item)+	assert.Equal(t, yesterday, persisted.Date)+	assert.Equal(t, 1, res.RowsWritten)+	assert.Equal(t, 1, res.RowsSkipped, "today's row must be skipped")+}++func TestBackfill_TodayExplicitlyRequested_StillSkipped(t *testing.T) {+	// AC 7.2 also covers the case where the operator passes --to=today+	// explicitly. The CLI's runtime gate skips today regardless of the flag.+	loc := sydney(t)+	today := time.Date(2026, 5, 20, 12, 0, 0, 0, loc).Format("2006-01-02")+	todayRow := existingPendingRow(today)+	f := &fakeDynamo{+		location:       loc,+		offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {todayRow}},+		readingsByDate: map[string][]dynamo.ReadingItem{today: chargeReadings(t, today, loc)},+	}+	opts := backfillOptsForTest(loc, today)+	res, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)+	assert.Empty(t, f.puts, "today's row must not be written even when --to=today")+	assert.Equal(t, 1, res.RowsSkipped)+}++func TestBackfill_SparseReadings_SkippedNoWrite(t *testing.T) {+	// AC 7.4: a day with fewer than two usable readings in the window is+	// reported SKIPPED and the row is left unchanged.+	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	day, _ := time.ParseInLocation("2006-01-02", date, loc)+	sparseTs := day.Add(12 * time.Hour).Unix()+	f := &fakeDynamo{+		location:    loc,+		offpeakRows: map[string][]dynamo.OffpeakItem{"*": {row}},+		readingsByDate: map[string][]dynamo.ReadingItem{+			date: {{SysSn: testSerial, Timestamp: sparseTs, Pgrid: 1000}},+		},+	}+	opts := backfillOptsForTest(loc, date)+	res, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)+	assert.Empty(t, f.puts, "sparse-readings day must not be written")+	assert.Equal(t, 1, res.RowsSparseSkipped)+	require.Len(t, res.Summary, 1)+	assert.Contains(t, res.Summary[0], "SKIPPED",+		"sparse-readings days must emit a SKIPPED summary line")+}++func TestBackfill_ConditionalWriteFailure_RowReportedExitsZero(t *testing.T) {+	// A row that has transitioned to pending between Query and PutItem (or a+	// row absent at the moment of the put) trips the WriteOffpeakIfComplete+	// condition. The CLI logs and continues — it does NOT exit non-zero.+	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	f := &fakeDynamo{+		location:       loc,+		offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {row}},+		readingsByDate: map[string][]dynamo.ReadingItem{date: chargeReadings(t, date, loc)},+		putErr:         &types.ConditionalCheckFailedException{},+	}+	opts := backfillOptsForTest(loc, date)+	res, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err, "conditional-write failure must not propagate as a fatal error")+	assert.Equal(t, 1, res.RowsConditionFailed)+	assert.Zero(t, res.RowsWritten)+}++func TestBackfill_DriftLogEmittedPerRow(t *testing.T) {+	// AC 6.1: the writer emits a structured INFO log entry per row, with the+	// five drift values. The backfill CLI shares the LogOffpeakDrift function+	// with the poller (Decision 6).+	buf, restore := captureSlog()+	defer restore()++	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	f := &fakeDynamo{+		location:       loc,+		offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {row}},+		readingsByDate: map[string][]dynamo.ReadingItem{date: chargeReadings(t, date, loc)},+	}+	opts := backfillOptsForTest(loc, date)+	_, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)++	out := buf.String()+	for _, key := range []string{+		"driftGrid", "driftSolar", "driftCharge", "driftDischarge", "driftExport",+		"snapshotGrid", "integratedGrid",+	} {+		assert.True(t, bytes.Contains([]byte(out), []byte(key)),+			"drift log must include %q. got: %s", key, out)+	}+	assert.Contains(t, out, "offpeak drift")+}++func TestBackfill_QueryError_PropagatedAsFatal(t *testing.T) {+	loc := sydney(t)+	f := &fakeDynamo{+		location:         loc,+		queryErrForTable: map[string]error{testOffpeakTable: errors.New("throttled")},+	}+	opts := backfillOptsForTest(loc, "2026-05-18")+	_, err := runBackfill(context.Background(), f, opts)+	require.Error(t, err)+	assert.Empty(t, f.puts)+}++// TestBackfill_SummaryLine_ShowsAbsDifferencePerDelta covers AC 7.5: the+// per-day summary line includes a |Δ|=X.XX column for each of the five+// deltas, holding the absolute difference between the prior stored value+// and the newly-integrated value rounded to two decimal places.+func TestBackfill_SummaryLine_ShowsAbsDifferencePerDelta(t *testing.T) {+	loc := sydney(t)+	date := "2026-05-18"+	row := existingCompleteRow(date)+	readings := chargeReadings(t, date, loc)+	f := &fakeDynamo{+		location:       loc,+		offpeakRows:    map[string][]dynamo.OffpeakItem{"*": {row}},+		readingsByDate: map[string][]dynamo.ReadingItem{date: readings},+	}+	opts := backfillOptsForTest(loc, date)+	res, err := runBackfill(context.Background(), f, opts)+	require.NoError(t, err)+	require.Len(t, res.Summary, 1, "expect one summary line for one row")+	line := res.Summary[0]++	// Decode the persisted row to read the new values exactly as the writer+	// rounded them, then compute the expected |Δ| against the stored prior.+	require.Len(t, f.puts, 1)+	persisted := decodeOffpeakItem(t, f.puts[0].Item)++	for _, c := range []struct {+		label string+		prev  float64+		next  float64+	}{+		{"grid", row.GridUsageKwh, persisted.GridUsageKwh},+		{"solar", row.SolarKwh, persisted.SolarKwh},+		{"chg", row.BatteryChargeKwh, persisted.BatteryChargeKwh},+		{"dis", row.BatteryDischargeKwh, persisted.BatteryDischargeKwh},+		{"exp", row.GridExportKwh, persisted.GridExportKwh},+	} {+		want := math.Abs(c.next - c.prev)+		// Format matches summaryLine: "label prev→next |Δ|=X.XX".+		fragment := fmt.Sprintf("%s %.2f→%.2f |Δ|=%.2f", c.label, c.prev, c.next, want)+		assert.Contains(t, line, fragment,+			"summary line must contain %q. got: %s", fragment, line)+	}+}++func TestValidateOpts_RejectsReversedDateRange(t *testing.T) {+	loc := sydney(t)+	opts := backfillOptsForTest(loc, "2026-05-18")+	opts.from = "2026-04-20"+	opts.to = "2026-04-10"++	err := validateOpts(opts)+	require.Error(t, err)+	assert.Contains(t, err.Error(), "after")+}++func TestValidateOpts_RejectsMissingWindow(t *testing.T) {+	loc := sydney(t)+	opts := backfillOptsForTest(loc, "2026-05-18")+	opts.offpeakStart = ""++	err := validateOpts(opts)+	require.Error(t, err)+}++func decodeOffpeakItem(t *testing.T, av map[string]types.AttributeValue) dynamo.OffpeakItem {+	t.Helper()+	var got dynamo.OffpeakItem+	require.NoError(t, attributevalue.UnmarshalMap(av, &got))+	return got+}

diff --git a/specs/offpeak-from-readings/decision_log.md b/specs/offpeak-from-readings/decision_log.mdnew file mode 100644index 0000000..39fd524--- /dev/null+++ b/specs/offpeak-from-readings/decision_log.md@@ -0,0 +1,413 @@+# Decision Log: Off-Peak From Readings++## Decision 1: Backfill scope — recompute all retained days++**Date**: 2026-05-24+**Status**: accepted++### Context++The bug fix changes how `flux-offpeak` rows are computed going forward. Existing rows carry lagged values from the snapshot-diff method. The `flux-readings` table has a 30-day TTL, so any backfill window is bounded.++### Decision++A one-time CLI run will recompute every `flux-offpeak` row whose date is still covered by `flux-readings`. Forward-only fixes are insufficient because the History 7/14/30-day stats and the new Daily Costs feature would carry the old lagged values for ~30 days post-deploy.++### Rationale++The discrepancy is large enough to be visible in the iOS app (4× overstatement of peak imports). Leaving it visible for a month after the deploy degrades user trust in the figures. The readings table is the ground truth and is still available, so the backfill is cheap and high-value.++### Alternatives Considered++- **Forward-only, no backfill**: Simpler but the History stats stay wrong for ~30 days post-deploy. Rejected.+- **On-demand only (ship CLI, don't run it)**: Operator overhead for no benefit. Rejected.++### Consequences++**Positive:**+- History and costs surfaces reflect corrected values within minutes of the deploy.+- Single one-time operator action with a clear endpoint.++**Negative:**+- Days older than 30 days remain on the lagged values forever (no remedy is possible — readings have aged out).+- A race exists between the backfill and the poller for "today" — handled by [[design]] (CLI skips today by default).++---++## Decision 2: Keep snapshot fields on OffpeakItem as diagnostic-only++**Date**: 2026-05-24+**Status**: accepted++### Context++`OffpeakItem` carries `startEpv`, `startEInput`, `startEOutput`, `startECharge`, `startEDischarge`, `startEGridCharge`, plus the matching `endE*` fields. With the readings-integration change these are no longer the source of the five deltas. They could be removed, retained, or made conditional.++### Decision++Keep the `startE*` / `endE*` fields on every new `flux-offpeak` row. The poller continues capturing both snapshots at the window boundaries. The five delta fields (`gridUsageKwh` etc.) are sourced from readings integration, not from the snapshot diff. No API consumer reads the snapshot fields.++### Rationale++Retaining the snapshot values gives a free forensic comparison every day: any future drift between the snapshot-diff number and the readings-integrated number is immediately visible in the row, without needing extra instrumentation. The cost is six `float64` fields per row on a table with <1000 rows lifetime — negligible.++### Alternatives Considered++- **Remove the snapshot fields entirely**: Cleaner schema but loses diagnostic visibility. Rejected — the storage cost is trivial.+- **Stop capturing snapshots (skip the AlphaESS API calls)**: Saves two API calls per day but removes the diagnostic comparison and would need a schema migration. Rejected.++### Consequences++**Positive:**+- Forensic visibility into snapshot-vs-readings drift on every row.+- No schema migration; existing dynamo marshalling unchanged.+- Today's pre-window-start `/status` response still has a `startEInput` to display for diagnostics if needed.++**Negative:**+- Two AlphaESS API calls per day continue (no cost saving on that axis).+- Future readers of the schema must understand that `startE*` / `endE*` are diagnostic and `endE* − startE* ≠ gridUsageKwh` by design.++---++## Decision 3: Live split for today integrates readings up to `now`++**Date**: 2026-05-24+**Status**: accepted++### Context++With snapshot-diff, the existing `offpeakDeltas` function projected pending records by `current.EInput − op.StartEInput`. With readings-integration, that projection logic no longer applies. Today's split during the window needs a fresh approach.++### Decision++While `now` is inside the off-peak window, integrate `flux-readings` from `offpeak-start` up to `now`. After the window closes, serve the deltas from the day's `flux-offpeak` row (the finalised value written by the poller per Requirement 3). Before the window opens, omit the split (existing behaviour).++### Rationale++The Dashboard and Day Detail off-peak tile is meaningful during the window — it shows grid-charging progress in real time, which several existing tiles depend on (e.g. the off-peak savings on the costs card for today). Suppressing it for the entire 11:00–14:00 window would regress on existing behaviour. Integrating live keeps the same UX with corrected values.++### Alternatives Considered++- **Suppress until window closes**: Off-peak tile blank from midnight to 14:00 (~14h/day). Regresses existing UX. Rejected.+- **Keep snapshot-diff projection for today, readings-integration for past days**: Hybrid keeps the lag bug for the active window. Rejected.++### Consequences++**Positive:**+- Today's tile updates as the window progresses, like the existing implementation but accurate.+- Single computation primitive shared by the poller's window-end writer, the Lambda's live projection, and the backfill CLI.++**Negative:**+- Each Lambda request inside the window pulls ≤1080 readings and integrates. Bounded by AC 8.2 (≤500 ms p95).+- Today's `/status` and `/day` no longer return the same off-peak value across two requests 10 seconds apart (it grows). Documented behaviour, matches the existing dashboard auto-refresh model.++---++## Decision 4: Handle sparse readings with best-effort integration++**Date**: 2026-05-24+**Status**: accepted++### Context++Readings can have gaps — poller outage, AlphaESS API failure, ECS deployment, container restart. The integration needs a defined behaviour for gaps.++### Decision++Skip any consecutive reading pair more than 60 seconds apart (matching the existing `computeTodayEnergy` pattern). The integration uses whatever pairs remain. If zero readings exist in the window for the day, omit the off-peak split entirely (matching the existing missing-record behaviour). The backfill CLI additionally reports days that hit the sparse-reading skip path so the operator can decide whether to investigate.++### Rationale++Aligns with the existing `computeTodayEnergy` policy — the same threshold that's been in production for live-compute since the original poller spec. Operators already understand "60-second gap = skip pair". Falling back to snapshot-diff on gappy days would reintroduce the lag for exactly the days where the lag matters least (the poller was offline, so there are fewer kWh to misclassify), at the cost of making the codepath nondeterministic between two methods.++### Alternatives Considered++- **Treat any gap as "missing"**: Drops too many days when the readings are mostly intact. Rejected.+- **Fall back to snapshot-diff when gaps exist**: Hybrid reintroduces the lag bug. Rejected.++### Consequences++**Positive:**+- Single integration policy; same behaviour in the poller, Lambda, and backfill CLI.+- Days with brief poller outages still get an accurate split for the parts the poller saw.++**Negative:**+- A day with a 5-minute outage in the middle of the window slightly under-counts the affected delta during the outage. Acceptable — the result is a small undercount, not a misclassification between peak and off-peak.++---++## Decision 5: Correctness validation is the bug-incident + structural bound++**Date**: 2026-05-25+**Status**: accepted++### Context++A first draft of the requirements expressed the correctness target as "the off-peak grid-import delta SHALL agree with the meter to ~1%, as on 2026-05-18". Both the design-critic and peer-review-validator flagged this as anecdotal — n=1 isn't a tolerance, and the 1% on a 20 kWh day day reads very differently from 1% on a 0.5 kWh day.++### Decision++Replace the single tolerance AC with two complementary acceptance criteria: (a) the specific bug-incident on 2026-05-18 must drop to a recomputed peak ≤ 0.25 kWh (the meter's 0.17 + an 0.08 calibration headroom); (b) for every day with a row, `gridUsageKwh ≤ eInput` as a structural sanity bound.++### Rationale++The first AC ties the success criterion to the actual bug report — if a future regression breaks this case, the test fails. The second is a structural invariant that cannot be wrong by design; together they cover the specific incident plus an "obviously broken" detector for unknown future days. A multi-day validation set was considered but rejected — it blocks on the user manually collecting meter splits for 5+ days, which is friction with marginal benefit beyond the structural bound.++### Alternatives Considered++- **Multi-day validation set (5+ days)**: Stronger statistical guarantee. Rejected — blocks on data the user has to collect manually.+- **Keep the 1% tolerance**: Anecdotal and not robustly testable. Rejected.++### Consequences++**Positive:**+- AC tied directly to the bug report — explicit traceability from incident → fix → test.+- Structural bound is cheap to assert at write-time as a defence-in-depth check.++**Negative:**+- Doesn't catch a regression that overstates peak by a small amount on a day other than 2026-05-18. Mitigated by [[design]] including a drift-logging hook (Decision 6) and by the existing iOS UI showing the kWh figures the user will compare to their bill.++---++## Decision 6: Drift observability is logging-only, no alerting++**Date**: 2026-05-25+**Status**: accepted++### Context++The bug survived in production for a release because no metric or log compared the snapshot-diff number against an integrated baseline. With [[decision-2-keep-snapshot-fields-as-diagnostic]] retaining both `endE*−startE*` and the readings integration on every row, a comparison is essentially free.++### Decision++On every off-peak row write (poller window-end and backfill CLI), emit a structured log entry containing the snapshot-diff value, the readings-integrated value, and the absolute difference per delta. No automatic alerting in scope; the log is queryable in CloudWatch for any operator investigation.++### Rationale++Logging is cheap, immediately useful for ad-hoc diagnosis (`fields date, gridUsageKwh, snapshotGridUsage` in CloudWatch Insights), and lets a future spec add alerting on top without re-instrumenting. Alerting was considered but the threshold ("when is drift large enough to act?") is unclear — better to ship visibility first and decide on thresholds after seeing real distributions.++### Alternatives Considered++- **Log + CloudWatch alarm above N kWh drift**: More valuable end-to-end but the threshold is unjustified up front. Deferred.+- **No observability**: The bug just survived a release for this reason. Rejected.++### Consequences++**Positive:**+- Future drift between AlphaESS's intraday and final eInput becomes visible without a user reporting a bill mismatch.+- Sets up a follow-up observability ticket with concrete data to set thresholds against.++**Negative:**+- Logs alone require operator pull (no automatic notification). Accepted given the two-user system.++---++## Decision 7: Window-end finalisation defers until a reading ≥ offpeak-end++**Date**: 2026-05-25+**Status**: accepted++### Context++A first draft had the poller compute the window-end deltas at exactly `offpeak-end`. The peer-review-validator pointed out this recreates the same class of boundary bug at the readings layer: if the poller fires at 14:00:00 sharp, the last sample may be 13:59:55 and a 14:00:01 charge tail is missed by the integration.++### Decision++Defer the window-end computation until the poller observes at least one reading with timestamp ≥ `offpeak-end`, bounded by a maximum wait of 30 seconds. After the 30 s budget expires, finalise with whatever readings exist (continuing to satisfy the 5-minute persistence deadline of AC 3.2).++### Rationale++The 10-second poller cadence means a reading at-or-after the boundary normally arrives within 5–15 s. Waiting up to 30 s is a small budget that buys the assurance that the last sample inside the window is bounded by a sample at or past the boundary, so trapezoidal integration on the last pair is well-defined (no extrapolation beyond the closing edge). Boundary interpolation was the alternative but it adds complexity to the integration code and produces the same result given dense sampling.++### Alternatives Considered++- **Interpolate from bracketing samples at offpeak-end**: Mathematically cleaner; same numerical result with 10 s sampling. Rejected for code complexity.+- **Fire at offpeak-end + safety buffer (e.g. +30 s) without waiting on a reading**: Simplest. Drops the last ~5 s of charging on ~50% of days. Boundary loss of ~0.01 kWh/day is negligible in isolation but matches the *category* of bug we're fixing, so rejected on principle.++### Consequences++**Positive:**+- No boundary-loss class of bug at the new computation layer.+- Maximum 30 s of additional latency post-window vs the snapshot-diff path's immediate write.++**Negative:**+- The poller now has a wait-on-condition state machine for window-end. Slightly more complex than "tick at 14:00:00".++---++## Decision 8: Sign convention and per-sample clamping policy++**Date**: 2026-05-25+**Status**: accepted++### Context++`flux-readings` stores `pgrid`, `pbat`, `ppv` as signed instantaneous power. Deriving five non-negative kWh deltas from three signed channels requires an unambiguous policy: per-sample clamp before integrating, or integrate the signed value and clamp the total. The two produce different numbers when the channel changes sign within the window.++### Decision++Split the signed power into positive / negative components on each sample before integration: grid import = `max(pgrid, 0)`, grid export = `max(-pgrid, 0)`, battery discharge = `max(pbat, 0)`, battery charge = `max(-pbat, 0)`, solar = `max(ppv, 0)`. Integrate each non-negative-by-sample series independently.++### Rationale++Per-sample clamping matches what a meter or independent energy counter would see — physical energy flow in one direction at one moment doesn't "cancel" flow in the other direction at another moment. Integrating signed values then clamping would zero out a day where 5 kWh imported in the morning and 5 kWh exported in the afternoon, which is wrong. The per-sample policy is also what `computeTodayEnergy` already uses for the daily totals, so the off-peak deltas stay consistent with the daily figures by construction.++### Alternatives Considered++- **Integrate signed power, clamp the total**: Cancels import vs export over the window. Wrong. Rejected.+- **Integrate signed power, take absolute value at the end**: Slightly less wrong but still loses directional information. Rejected.++### Consequences++**Positive:**+- The five deltas behave like five independent meter counters, matching real-world energy accounting.+- Consistency with `computeTodayEnergy`'s existing convention.++**Negative:**+- Negligible additional CPU per sample (one branch and one negation). Not material at the 1080-readings scale.++---++## Decision 9: Single integration helper with selectors, existing siblings retained++**Date**: 2026-05-25+**Status**: accepted++### Context++`derivedstats` already carries two near-identical integration siblings — `integratePload` and `integratePpv` — kept separate by Decision 6 of `specs/solar-by-block/decision_log.md` because folding two functions behind a closure-based generic wasn't worth the indirection cost. This feature needs five power channels integrated identically: pgrid (positive), pgrid (negative), pbat (positive), pbat (negative), ppv (positive).++### Decision++Introduce one new helper `integrate(readings, startUnix, endUnix, selector func(Reading) float64) float64` in `internal/derivedstats/integrate_offpeak.go`. `IntegrateOffpeakDeltas` calls it five times with the five selectors. The two existing siblings (`integratePload`, `integratePpv`) stay as-is — refactoring them is out of scope for the bug fix.++### Rationale++Decision 6 of solar-by-block was made when N=2. At N=5 (or N=7 if the existing two were folded in), the cost-benefit shifts: the algorithmic body is ~100 lines and any future change has to be mirrored five places. A single helper with five selectors mirrors `database/sql` and other selector-pattern stdlib precedent — readable, testable, and removes the "mirror this change five times" tax. The existing two siblings stay because (a) they're used by an unrelated feature and refactoring them now expands the blast radius, (b) the new helper's API is the right shape, so a follow-up that folds the older two in is mechanical.++### Alternatives Considered++- **Five new siblings, consistent with solar-by-block Decision 6**: Matches existing precedent but quintuples the mirroring tax. Rejected.+- **Refactor everything to the generic now**: Touches solar-by-block's tests + integration test suite. Out of scope for a bug fix. Deferred.++### Consequences++**Positive:**+- Single source of truth for the integration algorithm across the new feature's five deltas.+- The new helper's signature is what a future "fold all integrations into one place" refactor would land on.++**Negative:**+- Two integration code paths in `derivedstats` (the new generic + the two existing siblings) until a future refactor consolidates. Documented.++### Impact++Add a one-line `// TODO(offpeak-from-readings): fold into integrate() in integrate_offpeak.go` breadcrumb at the top of `integratePload` and `integratePpv` so the consolidation work is discoverable from the existing sites, not just from this decision log.++---++## Decision 10: Idempotence scoped to delta + count fields, not `integratedAt`++**Date**: 2026-05-25+**Status**: accepted++### Context++A first draft of AC 7.3 asserted the backfill CLI is "idempotent at the row level: re-running produces a row equal byte-for-byte to the prior run." Both the design-critic and the peer-review-validator flagged this as a defect: the `integratedAt` provenance field (Decision 2 + AC 5.4) is by definition the timestamp of the integration run and changes per run.++### Decision++AC 7.3 idempotence applies to the five delta fields (`gridUsageKwh`, `solarKwh`, `batteryChargeKwh`, `batteryDischargeKwh`, `gridExportKwh`) and the two integration-count fields (`integrationSampleCount`, `integrationSkippedPairs`). `integratedAt` is explicitly excluded.++### Rationale++The point of idempotence is "running the CLI twice doesn't corrupt the row, and two runs converge on the same numbers." That holds for the seven fields that describe what the integration computed. `integratedAt` describes WHEN it was computed — a different question, and a more useful piece of diagnostic information if it actually reflects the latest run time. Forcing it to a deterministic value (e.g. max-reading timestamp) sacrifices that diagnostic value for a property nothing relies on.++### Alternatives Considered++- **Force `integratedAt` to a deterministic value** (e.g., max reading timestamp): Preserves the byte-equal claim but loses the "when did the integration last run" diagnostic. Rejected.+- **Drop `integratedAt` entirely**: Saves the field but loses diagnostic value. The new fields are `omitempty` so cost is one string per row. Rejected.++### Consequences++**Positive:**+- The CLI's `--dry-run` summary can compare current vs prior numbers field-by-field with a clear scope of what should match.+- `integratedAt` retains its diagnostic value.++**Negative:**+- A naïve "diff the row" check after a re-run reports a difference on `integratedAt`. Mitigated by the test asserting field-scoped idempotence, not whole-row equality.++---++## Decision 11: Separate usability gate from integration math++**Date**: 2026-05-25+**Status**: accepted++### Context++`IntegrateOffpeakDeltas` reports `(_, false)` when fewer than two usable points can be constructed for the window (AC 1.6). The same predicate is also what each `integrate(...)` selector call needs to return zero rather than emit a spurious integral. A first draft folded these together: run one `integrate` call, inspect its construction-point count, and use that to decide the gate. That conflates "the integral is zero because there were no usable points" with "the integral is zero because every sample's clamped value was zero" — two distinct conditions with the same numerical result.++### Decision++Compute the construction-point count once up front (samples in `[start, end)` plus the two bracket-edge flags) in `IntegrateOffpeakDeltas`, and use that as the explicit usability gate. The shared `integrate` helper assumes its callers have already passed the gate and just does the trapezoidal math.++### Rationale++A window with all-zero power readings is a legitimately-integrable case (the integral is exactly 0 kWh and `SampleCount > 0`). A window with one in-window reading and no usable bracket is unusable and must report `(_, false)`. The two cases must be distinguishable to the caller — the API layer treats them differently (the first renders "0 kWh charged off-peak today", the second omits the off-peak tile). Folding the gate into the integrator loses that distinction.++The single up-front pass also avoids five duplicate scans across the five selectors. The gate check (samples + edge flags) is identical for all five channels because the gap rule looks only at timestamps.++### Alternatives Considered++- **Probe one channel, infer gate from its construction-point count**: Cheaper at first glance but conflates the two zero conditions and adds a hidden dependency on channel ordering. Rejected.+- **Return a third state from `integrate` (usable/unusable/zero)**: Adds a new sum type for two clear cases. Rejected for complexity.++### Consequences++**Positive:**+- Adding a new metric to the integrator is a one-line addition (new selector) — the gate doesn't have to be updated.+- Sparse-reading days are reported consistently across the five deltas; no chance of one channel saying "usable" and another "not".++**Negative:**+- The gate logic duplicates a small subset of the bracket-edge decision that `integrate` already encodes. Mitigated by both consulting the same `maxPairGapSeconds` constant and `bracketIndices` helper.++### Impact++Anyone adding a sixth integrated metric (e.g. a new battery channel) adds a selector to `IntegrateOffpeakDeltas` and is done; the usability gate doesn't need an update.++---++## Decision 12: `waitForReadingAtOrAfter` uses real clock, not the injectable scheduler clock++**Date**: 2026-05-25+**Status**: accepted++### Context++The `OffpeakScheduler` already injects a `now func() time.Time` so deterministic tests can pin the scheduler's notion of "when does the next thing happen". The new `waitForReadingAtOrAfter` helper (Decision 7) also looks at time — it needs to know when its budget expires and when to wake up for the next probe. Reusing `o.now` for both would superficially look more consistent.++### Decision++`waitForReadingAtOrAfter` uses `time.Now()` and `time.After()` directly. The scheduler's injectable clock governs orchestration ("when should the daily loop fire"); the wait-helper's clock governs live I/O against DynamoDB ("have I polled long enough for a row to arrive"). The two are kept distinct.++### Rationale++Tests that freeze `o.now` still need the wait-helper to advance real time when they do hit it (e.g. `TestHandleEnd_BoundaryWaitTimeout_StillWritesRow` pins `o.now` before `windowEnd` so the wait runs, but the 30 ms budget must elapse in wall time for the test to finish). If both clocks were `o.now`, the test would need a mock that advances on each poll iteration — significantly more test scaffolding for no production benefit.++The wait-helper's clock is also genuinely live I/O behaviour: the test for it (`TestWaitForReadingAtOrAfter_BudgetExpires`) measures wall-clock elapsed time against the budget, which would be impossible to assert against a frozen `o.now`.++### Alternatives Considered++- **Pass an injectable `now` and `after` into the wait-helper**: Eliminates the wall-clock dependency at test time but doubles the surface area of the helper's signature. Rejected — the two clocks are conceptually different and the present design makes that explicit.+- **Use `o.now` throughout**: Would require tests to drive the clock forward manually inside the wait loop. Rejected as over-engineering for a single-purpose helper.++### Consequences++**Positive:**+- Tests with frozen `o.now` keep working as the boundary-wait path uses real wall time.+- Helper signature is small (four params) and its purpose is unambiguous.++**Negative:**+- Tests that exercise the wait must accept short real-time delays (10–50 ms budgets). Acceptable — the existing helper tests already use ≤50 ms budgets.++---

diff --git a/CHANGELOG.md b/CHANGELOG.mdindex 5b4c4aa..4533dad 100644--- a/CHANGELOG.md+++ b/CHANGELOG.md@@ -12,8 +12,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - **Battery Can't Empty Before Off-Peak indicator** (T-1327). The Dashboard hero now shows "Won't empty before HH:MM" when, even at a 5 kW sustained-discharge ceiling, the battery cannot reach the 5 % cutoff before the next off-peak window opens — a `pbat`-independent answer to "is there any way?" rather than "given current rate, when?". The check is server-computed: `/status` carries a new nullable `battery.cantEmptyBeforeOffpeak` (true or null, encoded like `estimatedCutoffTime`), evaluated inside the existing `liveFresh` branch and reusing `derivedstats.ParseOffpeakWindow` (so the no-midnight-spanning-window limit carries over). Swift `BatteryInfo` gains a defaulted optional so the five existing constructors compile unchanged; `DashboardHeroPanel` composes a new `secondaryText` subview alongside the existing status line (no `Mode` enum changes) with a literal VoiceOver label `"Battery won't empty before off-peak at <HH:MM>"`. iOS and macOS share the panel via FluxCore. - **Daily Costs** (T-1326). Day Detail now shows a four-row costs card — peak imports, solar feed-in income, net, off-peak savings — and the History range gets a four-tile Period Costs card with an "N of M days priced" caption when coverage is partial. Costs are computed in FluxCore from the existing `/day` and `/history` responses (`DayCosts.swift`, `PeriodCosts.swift`) so the API shape stays unchanged. Pricing periods themselves are managed in Settings → Pricing: add/edit/delete time-of-use bands with peak, feed-in, and off-peak savings rates to 4 decimal-place precision; per-error remediation banners (inverted dates, overlap, rate precision, rate out of range, second open-ended, concurrent open-ended write) and a one-tap "close the current open-ended period on the date you just picked" remediation when the editor surfaces an overlap. Backed by a new `flux-pricing` DynamoDB table with the sentinel-row + `TransactWriteItems` pattern that makes the "at most one open-ended pricing period" invariant race-safe across concurrent writers on multiple devices. Lambda gains `GET /pricing`, `POST /pricing`, `PUT /pricing/{id}`, `DELETE /pricing/{id}`, and `POST /pricing/replace-open-ended`, all behind the existing bearer-token middleware. +### Fixed++- **Off-peak window deltas no longer count grid-charging tail as peak** (T-1341). The five off-peak energy deltas (grid usage, grid export, solar, battery charge, battery discharge) are now computed by trapezoidal integration of the 10-second `flux-readings` series over the window, replacing the previous `endE* − startE*` snapshot diff. AlphaESS's intraday `eInput` lags physical reality by 15–30 minutes, so the 14:00 snapshot routinely missed the tail of grid-charging (~4× overcount on the History peak imports tile; 2026-05-18 reference: meter reported 0.17 / 20.75 kWh peak/off-peak, Flux reported 1.74 / 18.95). All three computation sites move together: the poller's window-end finalisation (`internal/poller/offpeak.go`) now waits for a reading at-or-after `offpeak-end` (30 s budget) before integrating, so a missing post-window reading defers rather than commits a partial window; the live `/status` and `/day` paths (`internal/api/compute.go`, `status.go`, `history.go`) integrate the in-memory today readings up to `min(now, offpeak-end)` for pending rows and return the stored row for completed days; past-day reads serve the persisted integration. A new `derivedstats.IntegrateOffpeakDeltas` reuses the per-sample clamping, 60 s gap rule, and boundary interpolation already shipped for `integratePload` / `integratePpv`, so the off-peak path matches the existing live-energy integrator. Three provenance fields on `OffpeakItem` (`integrationSampleCount`, `integrationSkippedPairs`, `integratedAt`) record what each row was built from; the original snapshot fields are retained for drift logging. Conditional writes (`WriteOffpeakIfPendingOrAbsent` and `WriteOffpeakIfComplete`) make the poller-vs-backfill race safe. A new one-shot `cmd/backfill-offpeak` CLI (mirroring `cmd/backfill-solar`) recomputes any retained `flux-offpeak` row within the readings TTL window; the CLI never touches today's row and emits a five-column per-day summary (`prev→new |Δ|=abs` for all five deltas plus sample / skipped counts). Drift logging on every write surfaces snapshot-vs-integration divergence in CloudWatch.+ ### Documentation +- **Off-Peak From Readings spec** (T-1341). The spec captures the 2026-05-18 reference incident, the four-phase task plan (Foundation → Poller → API Live Path → Backfill + regression), and the design decisions that drove the implementation — most notably the deferred window-end finalisation (Decision around `waitForReadingAtOrAfter`) that prevents the readings-layer from re-introducing the boundary-loss bug class. See `specs/offpeak-from-readings/`. - **iPad adaptive layout spec** (T-1150). Implementation spec for swapping the stretched iPhone V5 shell on iPad for a `NavigationSplitView(.balanced)` sidebar at regular size class, with compact widths (Slide Over / narrow Split View) falling back to the existing tab-bar shell. Covers Dashboard / History / Day Detail multi-column layouts above 700 / 1000 pt detail widths, midnight rollover on the Today entry via a new `DayDetailViewModel.setDate(_:)`, and a `userInterfaceIdiom == .pad && hSizeClass == .regular` gate so iPhone Plus/Max landscape stays on the iPhone shell. See `specs/ipad-adaptive-layout/`. - **Daily Costs spec** (T-1326). The spec captures the 18+ design decisions and the TDD task plan (Backend stream + FluxCore/UI stream) that drove the implementation. Decision 24 standardises the multi-row pricing endpoints on a `{"pricing": [...]}` JSON envelope. See `specs/daily-costs/`.