fix(forecast): capture live observation time with telemetry

.changeset/current-forecast-observation.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Use the telemetry snapshot time for live forecast observations so delayed archive writes do not mistake fresh readings for future data.

go/cmd/ftw/forecast_observation_time_test.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"context"
+	"math"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/forecasting"
+	"github.com/srcfl/ftw/go/internal/state"
+	"github.com/srcfl/ftw/go/internal/telemetry"
+)
+
+func TestForecastObservationUsesCurrentTelemetryCutoff(t *testing.T) {
+	ctx := context.Background()
+	st, err := state.Open(filepath.Join(t.TempDir(), "state.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	if err = st.InitForecastArchive(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	tel := telemetry.NewStore()
+	site := trackerSite()
+	f := trackerFixture(time.Now())
+	f.store = st
+	f.tele = tel
+	f.site = func() forecastSite { return site }
+	var observedAt time.Time
+	f.curtailed = func(at time.Time) bool {
+		observedAt = at
+		return false
+	}
+	reading := func(at time.Time) telemetry.ForecastReading {
+		return telemetry.ForecastReading{
+			At: at, Earliest: at, Latest: at, PVEarliest: at, PVLatest: at,
+			Valid: true, PVValid: true, HouseholdW: 1200, PVW: -800,
+		}
+	}
+	feed := func(at time.Time) []forecasting.Observation {
+		return f.observationIntervals(reading(at), site, at)
+	}
+
+	// Prime the real accumulator from the start of the current quarter through
+	// the scheduled tick. A synchronous archive retry can delay observe until a
+	// newer telemetry sample has arrived.
+	scheduledAt := time.Now().UTC()
+	quarterStart := scheduledAt.Truncate(15 * time.Minute)
+	if out := feed(quarterStart); len(out) != 0 {
+		t.Fatalf("first sample completed an interval: %+v", out)
+	}
+	for at := quarterStart.Add(time.Minute); at.Before(scheduledAt); at = at.Add(time.Minute) {
+		if out := feed(at); len(out) != 0 {
+			t.Fatalf("partial quarter completed early: %+v", out)
+		}
+	}
+	if scheduledAt.After(quarterStart) {
+		if out := feed(scheduledAt); len(out) != 0 {
+			t.Fatalf("scheduled sample completed early: %+v", out)
+		}
+	}
+
+	tel.Update("site", telemetry.DerMeter, 400, nil, nil)
+	tel.Update("pv", telemetry.DerPV, -800, nil, nil)
+	tel.RecordDriverSuccess("site")
+	tel.RecordDriverSuccess("pv")
+	f.observe(ctx)
+	if observedAt.IsZero() {
+		t.Fatal("observation did not apply the curtailment cutoff")
+	}
+
+	var completed []forecasting.Observation
+	quarterEnd := quarterStart.Add(15 * time.Minute)
+	if observedAt.Before(quarterEnd) {
+		for at := observedAt.Add(time.Minute); at.Before(quarterEnd); at = at.Add(time.Minute) {
+			completed = append(completed, feed(at)...)
+		}
+		completed = append(completed, feed(quarterEnd)...)
+	}
+	persisted, err := st.LoadForecastObservations(ctx, quarterStart.UnixMilli(), quarterEnd.UnixMilli())
+	if err != nil {
+		t.Fatal(err)
+	}
+	completed = append(completed, persisted...)
+	if len(completed) != 1 || !completed[0].LoadKnown || completed[0].Quality != telemetry.ForecastIntervalQuality ||
+		!completed[0].PVKnown || math.Abs(completed[0].LoadW-1200) > 1e-6 || math.Abs(completed[0].PVW-800) > 1e-6 ||
+		completed[0].StartMS != quarterStart.UnixMilli() || completed[0].EndMS != quarterEnd.UnixMilli() {
+		t.Fatalf("delayed observation lost or changed the complete quarter: %+v", completed)
+	}
+}
+
+func TestForecastObservationStillRejectsLongGap(t *testing.T) {
+	f := &forecastTracker{}
+	site := trackerSite()
+	start := time.Date(2026, 1, 5, 12, 0, 0, 0, time.UTC)
+	reading := func(at time.Time) telemetry.ForecastReading {
+		return telemetry.ForecastReading{At: at, Earliest: at, Latest: at, Valid: true, HouseholdW: 1200}
+	}
+	if out := f.observationIntervals(reading(start), site, start); len(out) != 0 {
+		t.Fatalf("first sample completed an interval: %+v", out)
+	}
+	if out := f.observationIntervals(reading(start.Add(3*time.Minute)), site, start.Add(3*time.Minute)); len(out) != 0 {
+		t.Fatalf("long gap produced an interval: %+v", out)
+	}
+	var completed []forecasting.Observation
+	for at := start.Add(4 * time.Minute); !at.After(start.Add(15 * time.Minute)); at = at.Add(time.Minute) {
+		completed = append(completed, f.observationIntervals(reading(at), site, at)...)
+	}
+	if len(completed) != 0 {
+		t.Fatalf("quarter with a long gap was accepted: %+v", completed)
+	}
+}

go/cmd/ftw/forecast_occupancy_test.go

@@ -167,7 +167,7 @@ func TestForecastPendingIdentityCannotUseSavedModels(t *testing.T) {
 	// Pending startup must not touch live telemetry or the archive either.
 	f.tele = nil
 	f.store = nil
-	f.observe(context.Background(), at)
+	f.observe(context.Background())
 	if refreshes != 2 {
 		t.Fatalf("identity refreshes=%d want2", refreshes)
 	}

go/cmd/ftw/forecast_tracking.go

@@ -166,8 +166,8 @@ func (f *forecastTracker) run(ctx context.Context) {
 				continue
 			}
 			pending = nil
-		case now := <-tick.C:
-			f.observe(ctx, now)
+		case <-tick.C:
+			f.observe(ctx)
 		}
 	}
 }
@@ -220,7 +220,7 @@ func forecastArchiveRetryable(err error) bool {
 	return false
 }
 
-func (f *forecastTracker) observe(ctx context.Context, now time.Time) {
+func (f *forecastTracker) observe(ctx context.Context) {
 	if f.refreshIdentity != nil {
 		f.refreshIdentity()
 	}
@@ -239,7 +239,10 @@ func (f *forecastTracker) observe(ctx context.Context, now time.Time) {
 		return
 	}
 
-	r := f.tele.ForecastMeasurement(now, site.Meter, site.Options)
+	// A delayed tick only wakes the observer. Capture time under the telemetry
+	// lock so fresh polls are not compared with an old scheduled timestamp.
+	r := f.tele.ForecastMeasurementNow(site.Meter, site.Options)
+	now := r.At.UTC()
 	if f.configMu != nil {
 		f.configMu.RUnlock()
 	}