fix: integrate verified live observation clock correction

Source: joint-planner PR springfall2008#5498, commit 5653cc0. Shadow-only; no control activation.

Author: arturonaredo

apps/predbat/joint_site_native.py

@@ -103,6 +103,11 @@ def _parse_soc_raw(state, unit, capacity_kwh):
     return value
 
 
+def _observation_clock():
+    """Read wall time, never replace an entity's actual observation timestamp."""
+    return datetime.now(timezone.utc)
+
+
 def _options(base):
     """Read a dated expected session and entity-driven evidence without treating them as permission to switch."""
     settings = base.args.get("joint_site")
@@ -175,6 +180,10 @@ def _options(base):
     soc_payload = _read_entity_state(base, soc_entity) if soc_entity else None
     connection_payload = _read_entity_state(base, connection_entity) if connection_entity else None
     permission_payload = _read_entity_state(base, permission_entity) if permission_entity else None
+    # Predbat's forecast reference is floored to five minutes. A real observation
+    # later in that block is NOT future-dated. Read the observation clock after
+    # the cached states so a concurrent HA update is compared with current time.
+    observed_now = _aware(_observation_clock(), "observation_clock")
     manual_target_active = bool(getattr(base, "car_charging_manual_soc", None) and base.car_charging_manual_soc and base.car_charging_manual_soc[0])
     manual_target_kwh = None
     if manual_target_active:
@@ -192,7 +201,7 @@ def _options(base):
         # No live entity configured: dated session is the only source.
         observed = _aware(session["soc_observed_at"], "soc_observed_at") if isinstance(session["soc_observed_at"], datetime) else _parse_entity_timestamp(session["soc_observed_at"], "soc_observed_at")
         max_age = _number(settings.get("soc_max_age_minutes", 60), "soc_max_age_minutes", 1, 1440)
-        age = (now - observed).total_seconds() / 60
+        age = (observed_now - observed).total_seconds() / 60
         if age < 0 or age > max_age:
             raise ValueError("EV SoC observation is stale or future-dated")
         ev_soc = _number(session["soc_kwh"], "soc_kwh", 0, base.car_charging_battery_size[0])
@@ -211,7 +220,7 @@ def _options(base):
                 entity_blockers.append("soc_entity_out_of_bounds")
             else:
                 soc_max_age = _number(settings.get("soc_max_age_minutes", 60), "soc_max_age_minutes", 1, 1440)
-                soc_age_minutes = (now - soc_payload["last_updated"]).total_seconds() / 60
+                soc_age_minutes = (observed_now - soc_payload["last_updated"]).total_seconds() / 60
                 if soc_age_minutes < 0 or soc_age_minutes > soc_max_age:
                     entity_blockers.append("soc_entity_stale_or_future")
                 else:
@@ -239,7 +248,7 @@ def _options(base):
         raise ValueError("relay_entity state must be 'on' or 'off'")
     relay_last_changed = relay_payload["last_changed"] if relay_payload else None
     if relay_last_changed is not None:
-        elapsed_seconds = (now - relay_last_changed).total_seconds()
+        elapsed_seconds = (observed_now - relay_last_changed).total_seconds()
         if elapsed_seconds < 0:
             entity_blockers.append("relay_timestamp_future")
             elapsed_minutes_floor = 0
@@ -271,7 +280,10 @@ def _options(base):
         "arrival": start_offset,
         "control_model": control_model,
         "deadline": end_offset,
-        "departure_passed": departure <= now,
+        "departure_passed": departure <= observed_now,
+        "planning_reference_at": now.isoformat(),
+        "observations_checked_at": observed_now.isoformat(),
+        "planning_alignment_seconds": (observed_now - now).total_seconds(),
         "ev_soc": ev_soc,
         "ev_soc_source": ev_soc_source,
         "load_cap": _number(settings.get("load_port_limit_w"), "load_port_limit_w", 1) / 1000,
@@ -570,6 +582,9 @@ def evaluate_joint_site(base, publish=True):
             "counterfactual_controls": options["control_model"] != "native",
             "activation_requirements": execution_requirements,
             "manual_domestic_floor_slots": len(base.all_active_keep),
+            "planning_reference_at": options["planning_reference_at"],
+            "observations_checked_at": options["observations_checked_at"],
+            "planning_alignment_seconds": options["planning_alignment_seconds"],
         }
         blockers = _collect_blockers(model_base, options)
         # Always surface the entity observations even when topology blockers

apps/predbat/tests/test_joint_site_native.py

@@ -687,6 +687,22 @@ def test_closed_session_does_not_roll_to_tomorrow(base):
     assert _controls(base) == before and report["ready_to_activate"] is False
 
 
+def test_observations_after_rounded_plan_time_are_not_future(base):
+    """Reproduce a mid-slot live startup without accepting truly future data."""
+    _setup(base)
+    actual_now = base.now_utc + timedelta(minutes=2)
+    _set_entity(base, "sensor.test_ev_soc", "53", last_updated=base.now_utc + timedelta(seconds=90))
+    _set_entity(base, "binary_sensor.test_relay", "off", last_changed=actual_now - timedelta(seconds=59), last_updated=actual_now)
+    base.args["joint_site"].update(soc_entity="sensor.test_ev_soc", soc_entity_unit="%", relay_entity="binary_sensor.test_relay")
+    with patch("joint_site_native._observation_clock", return_value=actual_now):
+        options = _options(base)
+        assert not options["entity_blockers"], options
+        assert options["ev_soc"] == 13.25 and options["initial_off_minutes"] == 0
+        assert options["planning_alignment_seconds"] == 120
+        _set_entity(base, "sensor.test_ev_soc", "53", last_updated=actual_now + timedelta(seconds=1))
+        assert "soc_entity_stale_or_future" in _options(base)["entity_blockers"]
+
+
 def run_joint_site_native_tests(base):
     """Run native integration cases without a live HA connection or inverter writes."""
     for test in (
@@ -724,7 +740,9 @@ def run_joint_site_native_tests(base):
         test_real_ha_timestamp_cache_and_elapsed_cooldown,
         test_deye_counterfactual_preserves_native_capabilities_and_manual_floor,
         test_closed_session_does_not_roll_to_tomorrow,
+        test_observations_after_rounded_plan_time_are_not_future,
     ):
-        test(base)
+        with patch("joint_site_native._observation_clock", side_effect=lambda: base.now_utc):
+            test(base)
         print("PASS:", test.__name__)
     return False

docs/joint-site-planning.md

@@ -91,6 +91,12 @@ HA's `last_updated` separately from `last_changed` avoids equating an unchanged
 value with an old observation. Cached HA freshness still does not prove physical
 device freshness or connection, and no hardware refresh or service is requested.
 
+Freshness and elapsed relay-OFF time use a wall clock read after the cached
+observations, not Predbat's rounded five-minute forecast reference. Otherwise a
+valid observation inside the current block would appear future-dated at startup.
+Reports expose both references and their alignment difference. Entity timestamps
+are never replaced with the time at which the cache was read.
+
 The initial integration supports one metered EV and one inverter, full-rate charge
 and charge-freeze controls, no iBoost, and no manual mode/ceiling overrides. Domestic
 manual minimum targets are retained as time-local SOC constraints and checked in