"""Local-day activity derivation from measured sample intervals. Computes durable local-day read/write summaries using recorded local midnight boundaries. Collection derives new measured intervals; migration and repair rebuild legacy summaries from surviving evidence. UTC hour/day aggregates remain separate inputs to endurance projections, and readers stay read-only (ADR 0010). Key contracts: - A compatible interval wholly inside a local day contributes its volume once - Midnight-spanning and incompatible intervals persist once as unallocated evidence - UTC hour/day aggregates are never modified or deleted - Migration cannot manufacture local precision from historical UTC data Retention policy (issue #93): - Raw three-minute samples are pruned after 14 days (spec ยง3.4, ST-5). - Hour observations and day aggregates are retained indefinitely. - Local-day summaries and unallocated boundaries are persisted at collection time and retained indefinitely. They survive raw-sample pruning. - After detail expires, aged local summaries remain queryable with their recorded timezone and UTC boundaries. The evidence-availability flag on each history entry indicates whether the underlying raw detail is still present or has been pruned. - A later timezone change does not rewrite historical day boundaries. Each summary retains the timezone and offsets recorded at collection. - No fabricated evidence replaces missing precision. Incomplete or unavailable dates are labelled as such in the history readout. """ import sqlite3 from dataclasses import dataclass from datetime import date, datetime, time, timedelta, timezone from itertools import pairwise from typing import Any from zoneinfo import ZoneInfo @dataclass(frozen=True) class LocalDaySummary: """One local day's aggregated read/write volumes.""" local_date: str # e.g. "2026-09-01" tz_name: str # e.g. "Asia/Kolkata" tz_offset: str # e.g. "+05:30" utc_start: str # ISO 8601 UTC: the local midnight that starts this day utc_end: str # ISO 8601 UTC: the local midnight that ends this day bytes_written: int | None bytes_read: int | None coverage: float # known / (known + unknown) in UTC hours sample_count: int complete: bool # day's UTC range fully covered by hour observations activity_seconds: int = 0 activity_intervals: int = 1 activity_incomplete: bool = False activity_precision: str = "measured" @dataclass(frozen=True) class CoarseLocalDayActivity: """Rebuilt byte evidence and its coverage limits for one local day.""" bytes_written: int bytes_read: int activity_seconds: int activity_intervals: int incomplete: bool segment_totals: dict[int, "LocalDaySegmentTotals"] @dataclass class LocalDaySegmentTotals: """Activity totals attributed to one controller segment.""" bytes_written: int = 0 bytes_read: int = 0 activity_seconds: int = 0 activity_intervals: int = 0 def add( self, *, bytes_written: int, bytes_read: int, activity_seconds: int, activity_intervals: int = 1, ) -> None: """Accumulate one measured interval or coarse hour.""" self.bytes_written += bytes_written self.bytes_read += bytes_read self.activity_seconds += activity_seconds self.activity_intervals += activity_intervals @dataclass(frozen=True) class ReconstructedLocalDayInterval: """One counter delta with source and timezone provenance.""" start: datetime end: datetime bytes_written: int bytes_read: int activity_seconds: int segment_id: int start_sample_id: int end_sample_id: int start_tz: str | None end_tz: str | None @dataclass(frozen=True) class LocalDayHistoryEntry: """A local-day summary with evidence-limit metadata for history readout. Includes the summary data plus flags that indicate whether the underlying raw three-minute detail is still available or has been pruned, and whether the summary was derived from surviving evidence or is a legacy UTC-only aggregate. """ local_date: str tz_name: str tz_offset: str utc_start: str utc_end: str bytes_written: int | None bytes_read: int | None coverage: float sample_count: int complete: bool detail_available: bool # True if raw samples for this day are within 14-day retention derived_from_surviving: bool # True if derived from hour observations, not raw samples shared_bytes_written: int = 0 shared_bytes_read: int = 0 unallocated_bytes_written: int = 0 unallocated_bytes_read: int = 0 shared_evidence_count: int = 0 unallocated_evidence_count: int = 0 activity_state: str = "unavailable" @classmethod def from_summary( cls, summary: dict, detail_available: bool, derived_from_surviving: bool = True, ) -> "LocalDayHistoryEntry": """Create a history entry from a stored summary dict.""" return cls( local_date=summary["local_date"], tz_name=summary["tz_name"], tz_offset=summary["tz_offset"], utc_start=summary["utc_start"], utc_end=summary["utc_end"], bytes_written=summary["bytes_written"], bytes_read=summary["bytes_read"], coverage=summary["coverage"], sample_count=summary["sample_count"], complete=summary["complete"], detail_available=detail_available, derived_from_surviving=derived_from_surviving, shared_bytes_written=summary.get("shared_bytes_written", 0), shared_bytes_read=summary.get("shared_bytes_read", 0), unallocated_bytes_written=summary.get("unallocated_bytes_written", 0), unallocated_bytes_read=summary.get("unallocated_bytes_read", 0), shared_evidence_count=summary.get("shared_evidence_count", 0), unallocated_evidence_count=summary.get("unallocated_evidence_count", 0), activity_state=summary.get("activity_state", "unavailable"), ) def local_day_boundaries(local_date: str, tz_name: str) -> tuple[datetime, datetime, str]: """Return recorded UTC boundaries and midnight offset for one local date.""" day = date.fromisoformat(local_date) zone = ZoneInfo(tz_name) start_local = datetime.combine(day, time.min, tzinfo=zone) end_local = datetime.combine(day + timedelta(days=1), time.min, tzinfo=zone) start_utc = start_local.astimezone(timezone.utc) end_utc = end_local.astimezone(timezone.utc) offset = start_local.utcoffset() if offset is None: raise ValueError("Local midnight has no UTC offset") offset_seconds = int(offset.total_seconds()) sign = "+" if offset_seconds >= 0 else "-" offset_seconds = abs(offset_seconds) offset_text = "%s%02d:%02d" % ( sign, offset_seconds // 3600, (offset_seconds % 3600) // 60 ) return start_utc, end_utc, offset_text def derive_local_day_summary( conn: sqlite3.Connection, tz_name: str, clock_now: datetime, ) -> LocalDaySummary | None: """Build local-day coverage while preserving measured interval totals. Hour rows support the established usage-habit coverage contract. They cannot allocate a read/write delta to a half-hour local boundary, so local activity totals come only from measured sample intervals. """ local_dt = clock_now.astimezone(ZoneInfo(tz_name)) local_date = local_dt.date().isoformat() utc_start, utc_end, tz_offset_str = local_day_boundaries(local_date, tz_name) utc_start_iso = utc_start.isoformat() utc_end_iso = utc_end.isoformat() rows = conn.execute( "SELECT hour, sample_count, active_seconds, idle_seconds, " " powered_off_seconds, unknown_seconds " "FROM hour_observations WHERE hour >= ? AND hour < ? ORDER BY hour", (utc_start_iso, utc_end_iso), ).fetchall() # A UTC hour can straddle local midnight. Count its existence for the # coverage summary, but never use its bytes as local-day activity. midnight_hour = utc_start.replace(minute=0, second=0, microsecond=0) midnight_hour_iso = midnight_hour.strftime("%Y-%m-%dT%H:00:00+00:00") previous_hour = conn.execute( "SELECT hour, sample_count FROM hour_observations WHERE hour = ?", (midnight_hour_iso,), ).fetchone() total_samples = 0 known_seconds = 0 hour_count = 0 for row in rows: total_samples += row[1] or 0 known_seconds += (row[2] or 0) + (row[3] or 0) + (row[4] or 0) hour_count += 1 if previous_hour is not None and all(row[0] != previous_hour[0] for row in rows): total_samples += previous_hour[1] or 0 hour_count += 1 if hour_count == 0: return None day_seconds = int((utc_end - utc_start).total_seconds()) coverage = known_seconds / day_seconds if day_seconds > 0 else 0.0 complete = known_seconds >= day_seconds and hour_count > 0 existing = conn.execute( "SELECT bytes_written, bytes_read, activity_seconds, activity_intervals, " " activity_incomplete, activity_precision " "FROM local_days WHERE local_date = ? AND tz_name = ?", (local_date, tz_name), ).fetchone() if existing is None: activity = (0, 0, 0, 0, False, "unavailable") else: activity = existing return LocalDaySummary( local_date=local_date, tz_name=tz_name, tz_offset=tz_offset_str, utc_start=utc_start_iso, utc_end=utc_end_iso, bytes_written=activity[0] or 0, bytes_read=activity[1] or 0, coverage=coverage, sample_count=total_samples, complete=complete, activity_seconds=activity[2] or 0, activity_intervals=activity[3] or 0, activity_incomplete=bool(activity[4]), activity_precision=activity[5], ) def persist_local_day( conn: sqlite3.Connection, summary: LocalDaySummary, ) -> bool: """Upsert a local-day summary. Returns True if newly created.""" existing = conn.execute( "SELECT id FROM local_days WHERE local_date = ? AND tz_name = ?", (summary.local_date, summary.tz_name), ).fetchone() if existing is None: conn.execute( "INSERT INTO local_days " "(local_date, tz_name, tz_offset, utc_start, utc_end, " " bytes_written, bytes_read, coverage, sample_count, complete, " " activity_seconds, activity_intervals, activity_incomplete, activity_precision) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (summary.local_date, summary.tz_name, summary.tz_offset, summary.utc_start, summary.utc_end, summary.bytes_written, summary.bytes_read, summary.coverage, summary.sample_count, summary.complete, summary.activity_seconds, summary.activity_intervals, summary.activity_incomplete, summary.activity_precision), ) created = True else: conn.execute( "UPDATE local_days " "SET tz_offset = ?, utc_start = ?, utc_end = ?, " " bytes_written = ?, bytes_read = ?, " " coverage = ?, sample_count = ?, complete = ?, " " activity_seconds = ?, activity_intervals = ?, " " activity_incomplete = ?, activity_precision = ? " "WHERE id = ?", (summary.tz_offset, summary.utc_start, summary.utc_end, summary.bytes_written, summary.bytes_read, summary.coverage, summary.sample_count, summary.complete, summary.activity_seconds, summary.activity_intervals, summary.activity_incomplete, summary.activity_precision, existing[0]), ) created = False # Caller manages transactions (collector commits once after all derivation) return created def _utc_datetime(value: str | datetime) -> datetime: parsed = value if isinstance(value, datetime) else datetime.fromisoformat(value) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _same_monitoring_period( conn: sqlite3.Connection, start: datetime, end: datetime, ) -> bool: """Return whether one recorded monitoring period contains the interval.""" for started_at, ended_at in conn.execute( "SELECT started_at, ended_at FROM monitoring_periods" ): period_start = _utc_datetime(started_at) period_end = _utc_datetime(ended_at) if ended_at else None if period_start <= start and (period_end is None or end <= period_end): return True return False def repair_legacy_local_day_evidence(conn: sqlite3.Connection) -> int: """Rebuild untrusted local-day totals from surviving evidence. Sample intervals take precedence over UTC-hour summaries. An interval is usable only when its counters, controller segment, monitoring period, and recorded local-day boundaries all agree. Old byte totals remain stored but hidden when no such evidence survives. """ tables = { row[0] for row in conn.execute( "SELECT name FROM sqlite_master WHERE type='table'" ) } if "local_days" not in tables: return 0 recorded_days = conn.execute( "SELECT id, local_date, tz_name, utc_start, utc_end, " " activity_incomplete, activity_precision " "FROM local_days ORDER BY utc_start, id" ).fetchall() local_days = [ row[:6] for row in recorded_days if row[6] in ("legacy", "unavailable") ] if not local_days: return 0 required_sample_columns = {"id", "ts", "bytes_written", "bytes_read", "segment_id"} periods = [] segment_boundaries = [] segment_windows = {} if {"monitoring_periods", "controller_segments"}.issubset(tables): periods = [ (_utc_datetime(start), _utc_datetime(end) if end else None) for start, end in conn.execute( "SELECT started_at, ended_at FROM monitoring_periods" ) ] segment_boundaries = [ (segment_id, _utc_datetime(opened_at)) for segment_id, opened_at in conn.execute( "SELECT id, opened_at FROM controller_segments ORDER BY opened_at, id" ) ] segment_windows = { segment_id: ( opened_at, segment_boundaries[index + 1][1] if index + 1 < len(segment_boundaries) else None, ) for index, (segment_id, opened_at) in enumerate(segment_boundaries) } sample_columns = ( {row[1] for row in conn.execute("PRAGMA table_info(samples)")} if "samples" in tables else set() ) intervals: list[ReconstructedLocalDayInterval] = [] samples = [] if required_sample_columns.issubset(sample_columns): sample_select = ( "SELECT id, ts, bytes_written, bytes_read, segment_id, " + ("local_tz " if "local_tz" in sample_columns else "NULL AS local_tz ") + "FROM samples ORDER BY ts, id" ) samples = [ { "id": row[0], "ts": _utc_datetime(row[1]), "bytes_written": row[2], "bytes_read": row[3], "segment_id": row[4], "local_tz": row[5], } for row in conn.execute(sample_select) ] for previous, current in pairwise(samples): start = previous["ts"] end = current["ts"] segment_id = current["segment_id"] segment_window = segment_windows.get(segment_id) if ( end <= start or segment_id is None or previous["segment_id"] != segment_id or segment_window is None or segment_window[0] > start or (segment_window[1] is not None and end >= segment_window[1]) or bool(previous["local_tz"]) != bool(current["local_tz"]) or previous["bytes_written"] is None or current["bytes_written"] is None or previous["bytes_read"] is None or current["bytes_read"] is None ): continue bytes_written = current["bytes_written"] - previous["bytes_written"] bytes_read = current["bytes_read"] - previous["bytes_read"] if bytes_written < 0 or bytes_read < 0: continue if not any( period_start <= start and (period_end is None or end <= period_end) for period_start, period_end in periods ): continue intervals.append(ReconstructedLocalDayInterval( start=start, end=end, bytes_written=bytes_written, bytes_read=bytes_read, activity_seconds=int((end - start).total_seconds()), segment_id=segment_id, start_sample_id=previous["id"], end_sample_id=current["id"], start_tz=previous["local_tz"], end_tz=current["local_tz"], )) _record_reconstructed_shared_intervals(conn, intervals, recorded_days) day_bounds = [ { "id": row[0], "local_date": row[1], "tz_name": row[2], "start": _utc_datetime(row[3]), "end": _utc_datetime(row[4]), "precision": row[6], } for row in recorded_days ] rebuilding_ids = {row[0] for row in local_days} matching_by_day: dict[int, list[ReconstructedLocalDayInterval]] = { local_day_id: [] for local_day_id in rebuilding_ids } for interval in intervals: candidates = [ day for day in day_bounds if day["start"] <= interval.start and interval.end <= day["end"] and all( zone is None or zone == day["tz_name"] for zone in (interval.start_tz, interval.end_tz) ) ] if len(candidates) == 1 and candidates[0]["id"] in rebuilding_ids: matching_by_day[candidates[0]["id"]].append(interval) rebuilt = 0 for row in local_days: ( local_day_id, local_date, tz_name, utc_start, utc_end, prior_incomplete, ) = row start = _utc_datetime(utc_start) end = _utc_datetime(utc_end) matching = matching_by_day[local_day_id] if not matching: has_samples = any( start <= sample["ts"] < end for sample in samples ) coarse = ( None if has_samples else _coarse_local_day_activity( conn, local_day_id, start, end, periods, segment_boundaries, ) ) if coarse is not None: incomplete = bool(prior_incomplete) or coarse.incomplete conn.execute( "UPDATE local_days SET bytes_written = ?, bytes_read = ?, " "activity_seconds = ?, activity_intervals = ?, " "activity_incomplete = ?, activity_precision = 'coarse', " "last_sample_id = NULL WHERE id = ?", (coarse.bytes_written, coarse.bytes_read, coarse.activity_seconds, coarse.activity_intervals, incomplete, local_day_id), ) _replace_local_day_segment_totals( conn, local_day_id, coarse.segment_totals, ) rebuilt += 1 continue conn.execute( "UPDATE local_days SET activity_precision = 'unavailable', " "activity_incomplete = CASE WHEN EXISTS (" " SELECT 1 FROM local_day_unallocated_evidence " " WHERE (start_local_date = ? AND start_tz_name = ?) " " OR (end_local_date = ? AND end_tz_name = ?)" ") THEN 1 ELSE 0 END WHERE id = ?", (local_date, tz_name, local_date, tz_name, local_day_id), ) continue totals: dict[int, LocalDaySegmentTotals] = {} bytes_written = 0 bytes_read = 0 activity_seconds = 0 for interval in matching: segment_total = totals.setdefault( interval.segment_id, LocalDaySegmentTotals(), ) segment_total.add( bytes_written=interval.bytes_written, bytes_read=interval.bytes_read, activity_seconds=interval.activity_seconds, ) bytes_written += interval.bytes_written bytes_read += interval.bytes_read activity_seconds += interval.activity_seconds day_seconds = int((end - start).total_seconds()) incomplete = bool(prior_incomplete) or activity_seconds < day_seconds last_sample_id = max(interval.end_sample_id for interval in matching) _replace_local_day_segment_totals(conn, local_day_id, totals) conn.execute( "UPDATE local_days SET bytes_written = ?, bytes_read = ?, " "activity_seconds = ?, activity_intervals = ?, " "activity_incomplete = ?, activity_precision = 'measured', " "last_sample_id = ? WHERE id = ?", (bytes_written, bytes_read, activity_seconds, len(matching), incomplete, last_sample_id, local_day_id), ) rebuilt += 1 return rebuilt def _replace_local_day_segment_totals( conn: sqlite3.Connection, local_day_id: int, totals: dict[int, LocalDaySegmentTotals], ) -> None: """Replace segment provenance rows for one rebuilt local day.""" conn.execute( "DELETE FROM local_day_segment_totals WHERE local_day_id = ?", (local_day_id,), ) conn.executemany( "INSERT INTO local_day_segment_totals " "(local_day_id, segment_id, bytes_written, bytes_read, " " activity_seconds, activity_intervals) " "VALUES (?, ?, ?, ?, ?, ?)", ( ( local_day_id, segment_id, values.bytes_written, values.bytes_read, values.activity_seconds, values.activity_intervals, ) for segment_id, values in totals.items() ), ) def _record_reconstructed_shared_intervals( conn: sqlite3.Connection, intervals: list[ReconstructedLocalDayInterval], recorded_days: list[tuple], ) -> None: """Retain compatible counter deltas that cross stored local boundaries.""" day_rows = [ { "id": row[0], "local_date": row[1], "tz_name": row[2], "utc_start": _utc_datetime(row[3]), "utc_end": _utc_datetime(row[4]), "precision": row[6], } for row in recorded_days ] def find_day(instant: datetime, zone: str | None, *, end: bool): candidates = [ row for row in day_rows if (row["utc_start"] < instant <= row["utc_end"] if end else row["utc_start"] <= instant < row["utc_end"]) and (zone is None or row["tz_name"] == zone) ] return candidates[0] if len(candidates) == 1 else None for interval in intervals: start_day = find_day(interval.start, interval.start_tz, end=False) end_day = find_day(interval.end, interval.end_tz, end=True) if start_day is None or end_day is None or start_day["id"] == end_day["id"]: continue if start_day["precision"] not in ("legacy", "unavailable") and ( end_day["precision"] not in ("legacy", "unavailable") ): continue if start_day["tz_name"] == end_day["tz_name"]: zone_days = sorted( (row for row in day_rows if row["tz_name"] == start_day["tz_name"]), key=lambda row: row["utc_start"], ) start_index = next( (i for i, row in enumerate(zone_days) if row["id"] == start_day["id"]), None, ) end_index = next( (i for i, row in enumerate(zone_days) if row["id"] == end_day["id"]), None, ) if start_index is None or end_index is None or end_index <= start_index: continue chain = zone_days[start_index:end_index + 1] if any( left["utc_end"] != right["utc_start"] for left, right in pairwise(chain) ): continue reason = "local_midnight" else: reason = "timezone_change" conn.execute( "INSERT OR IGNORE INTO local_day_unallocated_evidence " "(start_sample_id, end_sample_id, start_local_date, end_local_date, " " start_tz_name, end_tz_name, started_at, ended_at, bytes_written, " " bytes_read, reason, segment_id) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( interval.start_sample_id, interval.end_sample_id, start_day["local_date"], end_day["local_date"], start_day["tz_name"], end_day["tz_name"], interval.start.isoformat(), interval.end.isoformat(), interval.bytes_written, interval.bytes_read, reason, interval.segment_id, ), ) conn.execute( "UPDATE local_days SET activity_incomplete = 1 " "WHERE id IN (?, ?)", (start_day["id"], end_day["id"]), ) def _coarse_local_day_activity( conn: sqlite3.Connection, local_day_id: int, day_start: datetime, day_end: datetime, periods: list[tuple[datetime, datetime | None]], segments: list[tuple[int, datetime]], ) -> CoarseLocalDayActivity | None: """Return UTC hours unique to these recorded local-day bounds.""" if not periods or not segments or conn.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='hour_observations'" ).fetchone() is None: return None hour = timedelta(hours=1) full_hours: dict[datetime, tuple] = {} boundary_hour_found = False other_day_bounds = [ (_utc_datetime(row[0]), _utc_datetime(row[1])) for row in conn.execute( "SELECT utc_start, utc_end FROM local_days WHERE id != ?", (local_day_id,), ) ] first_hour = day_start.replace(minute=0, second=0, microsecond=0) for row in conn.execute( "SELECT hour, bytes_written_delta, bytes_read_delta, sample_count, " " active_seconds, idle_seconds, powered_off_seconds " "FROM hour_observations WHERE hour >= ? AND hour < ? ORDER BY hour", (first_hour.isoformat(), day_end.isoformat()), ): hour_start = _utc_datetime(row[0]) hour_end = hour_start + hour if hour_start >= day_end or hour_end <= day_start: continue if day_start <= hour_start and hour_end <= day_end: if any( other_start < hour_end and hour_start < other_end for other_start, other_end in other_day_bounds ): # Another historical local-day row claims this whole or # partial UTC hour too. Keep the byte delta unattributed. boundary_hour_found = True continue full_hours[hour_start] = row else: boundary_hour_found = True expected_hours = [] expected_start = day_start.replace(minute=0, second=0, microsecond=0) if expected_start < day_start: expected_start += hour while expected_start + hour <= day_end: expected_hours.append(expected_start) expected_start += hour def hour_context(hour_start: datetime) -> int | None: hour_end = hour_start + hour period_matches = any( period_start <= hour_start and (period_end is None or hour_end <= period_end) for period_start, period_end in periods ) segment_matches = [] for index, (segment_id, opened_at) in enumerate(segments): next_segment = segments[index + 1][1] if index + 1 < len(segments) else None if opened_at <= hour_start and (next_segment is None or hour_end <= next_segment): segment_matches.append(segment_id) if not period_matches or len(segment_matches) != 1: return None return segment_matches[0] totals: dict[int, LocalDaySegmentTotals] = {} bytes_written = 0 bytes_read = 0 activity_seconds = 0 activity_intervals = 0 context_complete = True for hour_start, row in full_hours.items(): segment_id = hour_context(hour_start) if segment_id is None: context_complete = False continue hour_seconds = (row[4] or 0) + (row[5] or 0) + (row[6] or 0) activity_seconds += hour_seconds if (row[3] or 0) <= 0 and (row[1] or 0) == 0 and (row[2] or 0) == 0: continue segment_total = totals.setdefault(segment_id, LocalDaySegmentTotals()) segment_total.add( bytes_written=row[1] or 0, bytes_read=row[2] or 0, activity_seconds=hour_seconds, ) bytes_written += row[1] or 0 bytes_read += row[2] or 0 activity_intervals += 1 if not activity_intervals: return None all_hours_present = all( hour_start in full_hours for hour_start in expected_hours ) has_unallocated_bytes = False if conn.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='day_aggregates'" ).fetchone() is not None: last_utc_day = (day_end - timedelta(microseconds=1)).date() has_unallocated_bytes = conn.execute( "SELECT 1 FROM day_aggregates WHERE day >= ? AND day <= ? " "AND (COALESCE(unattributed_bytes_written, 0) > 0 " " OR COALESCE(unattributed_bytes_read, 0) > 0) LIMIT 1", (day_start.date().isoformat(), last_utc_day.isoformat()), ).fetchone() is not None complete = ( all_hours_present and activity_seconds >= int((day_end - day_start).total_seconds()) and not boundary_hour_found and not has_unallocated_bytes and context_complete ) return CoarseLocalDayActivity( bytes_written=bytes_written, bytes_read=bytes_read, activity_seconds=activity_seconds, activity_intervals=activity_intervals, incomplete=not complete, segment_totals=totals, ) def _upsert_activity_day( conn: sqlite3.Connection, local_date: str, tz_name: str, *, bytes_written: int = 0, bytes_read: int = 0, seconds: int = 0, interval_count: int = 0, incomplete: bool = False, last_sample_id: int | None = None, ) -> None: start, end, offset = local_day_boundaries(local_date, tz_name) conn.execute( "INSERT INTO local_days " "(local_date, tz_name, tz_offset, utc_start, utc_end, bytes_written, " " bytes_read, activity_seconds, activity_intervals, activity_incomplete, " " activity_precision, last_sample_id) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'measured', ?) " "ON CONFLICT(local_date, tz_name) DO UPDATE SET " "bytes_written = CASE WHEN local_days.activity_precision IN ('legacy', 'unavailable') " " AND excluded.activity_intervals > 0 " " THEN excluded.bytes_written ELSE local_days.bytes_written + excluded.bytes_written END, " "bytes_read = CASE WHEN local_days.activity_precision IN ('legacy', 'unavailable') " " AND excluded.activity_intervals > 0 " " THEN excluded.bytes_read ELSE local_days.bytes_read + excluded.bytes_read END, " "activity_seconds = CASE WHEN local_days.activity_precision IN ('legacy', 'unavailable') " " AND excluded.activity_intervals > 0 " " THEN excluded.activity_seconds ELSE local_days.activity_seconds + excluded.activity_seconds END, " "activity_intervals = CASE WHEN local_days.activity_precision IN ('legacy', 'unavailable') " " AND excluded.activity_intervals > 0 " " THEN excluded.activity_intervals ELSE local_days.activity_intervals + excluded.activity_intervals END, " "activity_incomplete = MAX(local_days.activity_incomplete, excluded.activity_incomplete), " "activity_precision = CASE WHEN excluded.activity_intervals > 0 " " THEN 'measured' ELSE local_days.activity_precision END, " "last_sample_id = COALESCE(excluded.last_sample_id, local_days.last_sample_id)", ( local_date, tz_name, offset, start.isoformat(), end.isoformat(), bytes_written, bytes_read, seconds, interval_count, incomplete, last_sample_id, ), ) def record_local_activity_interval( conn: sqlite3.Connection, previous: dict[str, Any], current: dict[str, Any], *, start_sample_id: int, end_sample_id: int, segment_id: int | None, ) -> str | None: """Persist one measured interval as known local activity or once-only evidence. Counter differences are allocated only when both sample timezones match, both samples share one monitoring period, and the interval stays inside one recorded local date. Every other valid difference remains one unallocated row. """ start = _utc_datetime(previous["ts"]) end = _utc_datetime(current["ts"]) if end <= start: return None previous_tz = previous.get("local_tz") current_tz = current.get("local_tz") if not current_tz: return None from_zone = ZoneInfo(previous_tz) if previous_tz else None to_zone = ZoneInfo(current_tz) start_date = start.astimezone(from_zone).date().isoformat() if from_zone else None end_date = end.astimezone(to_zone).date().isoformat() if start_date is None: start_date = end_date written = current.get("bytes_written") prior_written = previous.get("bytes_written") read = current.get("bytes_read") prior_read = previous.get("bytes_read") if ( written is None or prior_written is None or read is None or prior_read is None ): return None bytes_written = written - prior_written bytes_read = read - prior_read if bytes_written < 0 or bytes_read < 0: for local_date, zone_name in ( (start_date, previous_tz), (end_date, current_tz), ): if zone_name is not None: _upsert_activity_day( conn, local_date, zone_name, incomplete=True, ) return "counter_discontinuity" elif previous_tz is None: reason = "legacy_timezone_unknown" elif previous_tz != current_tz: reason = "timezone_change" elif segment_id is None: reason = "segment_unknown" elif not _same_monitoring_period(conn, start, end): reason = "monitoring_period" elif start_date != end_date: reason = "local_midnight" else: already_accounted = conn.execute( "SELECT last_sample_id, activity_precision FROM local_days " "WHERE local_date = ? AND tz_name = ?", (end_date, current_tz), ).fetchone() if ( already_accounted is not None and already_accounted[0] is not None and already_accounted[0] >= end_sample_id ): return "known" replacing_coarse = ( already_accounted is not None and already_accounted[1] == "coarse" ) if replacing_coarse: local_day_id = conn.execute( "SELECT id FROM local_days WHERE local_date = ? AND tz_name = ?", (end_date, current_tz), ).fetchone()[0] conn.execute( "DELETE FROM local_day_segment_totals WHERE local_day_id = ?", (local_day_id,), ) conn.execute( "UPDATE local_days SET bytes_written = 0, bytes_read = 0, " "activity_seconds = 0, activity_intervals = 0, " "activity_precision = 'unavailable', activity_incomplete = 1, " "last_sample_id = NULL WHERE id = ?", (local_day_id,), ) seconds = int((end - start).total_seconds()) _upsert_activity_day( conn, end_date, current_tz, bytes_written=bytes_written, bytes_read=bytes_read, seconds=seconds, interval_count=1, incomplete=replacing_coarse, last_sample_id=end_sample_id, ) local_day_id = conn.execute( "SELECT id FROM local_days WHERE local_date = ? AND tz_name = ?", (end_date, current_tz), ).fetchone()[0] conn.execute( "INSERT INTO local_day_segment_totals " "(local_day_id, segment_id, bytes_written, bytes_read, " " activity_seconds, activity_intervals) " "VALUES (?, ?, ?, ?, ?, 1) " "ON CONFLICT(local_day_id, segment_id) DO UPDATE SET " "bytes_written = local_day_segment_totals.bytes_written + excluded.bytes_written, " "bytes_read = local_day_segment_totals.bytes_read + excluded.bytes_read, " "activity_seconds = local_day_segment_totals.activity_seconds + excluded.activity_seconds, " "activity_intervals = local_day_segment_totals.activity_intervals + 1", (local_day_id, segment_id, bytes_written, bytes_read, seconds), ) return "known" conn.execute( "INSERT OR IGNORE INTO local_day_unallocated_evidence " "(start_sample_id, end_sample_id, start_local_date, end_local_date, " " start_tz_name, end_tz_name, started_at, ended_at, bytes_written, " " bytes_read, reason, segment_id) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( start_sample_id, end_sample_id, start_date, end_date, previous_tz, current_tz, start.isoformat(), end.isoformat(), bytes_written, bytes_read, reason, segment_id, ), ) affected_days: set[tuple[str, str]] = set() if previous_tz is not None: affected_days.add((start_date, previous_tz)) affected_days.add((end_date, current_tz)) if previous_tz is not None and previous_tz == current_tz and start_date < end_date: day = date.fromisoformat(start_date) + timedelta(days=1) last = date.fromisoformat(end_date) while day < last: affected_days.add((day.isoformat(), current_tz)) day += timedelta(days=1) for local_date, zone_name in affected_days: _upsert_activity_day( conn, local_date, zone_name, incomplete=True, ) return reason def mark_local_activity_gap( conn: sqlite3.Connection, current_ts: str, current_tz: str, previous: dict | None = None, ) -> None: """Mark dates around a controller boundary as missing local activity.""" current = _utc_datetime(current_ts) current_date = current.astimezone(ZoneInfo(current_tz)).date().isoformat() _upsert_activity_day(conn, current_date, current_tz, incomplete=True) if previous is None or not previous.get("local_tz"): return previous_at = _utc_datetime(previous["ts"]) previous_tz = previous["local_tz"] previous_date = previous_at.astimezone(ZoneInfo(previous_tz)).date().isoformat() _upsert_activity_day(conn, previous_date, previous_tz, incomplete=True) def query_local_day_summary( conn: sqlite3.Connection, local_date: str, tz_name: str | None = None, clock_now: datetime | None = None, ) -> dict | None: """Query a stored local-day summary by date. Returns a dict with the summary fields, or None if no entry exists. """ tz_filter = " AND tz_name = ?" if tz_name else "" params = (local_date, tz_name) if tz_name else (local_date,) row = conn.execute( "SELECT local_date, tz_name, tz_offset, utc_start, utc_end, " " bytes_written, bytes_read, coverage, sample_count, complete, " " activity_seconds, activity_intervals, activity_incomplete, " " activity_precision " "FROM local_days WHERE local_date = ?" + tz_filter + " ORDER BY id DESC LIMIT 1", params, ).fetchone() if row is None: return None ( recorded_date, recorded_tz, tz_offset, utc_start, utc_end, bytes_written, bytes_read, coverage, sample_count, complete, activity_seconds, activity_intervals, activity_incomplete, activity_precision, ) = row evidence = conn.execute( "SELECT COUNT(*), " " COALESCE(SUM(CASE WHEN reason = 'local_midnight' THEN 1 ELSE 0 END), 0), " " COALESCE(SUM(CASE WHEN reason != 'local_midnight' THEN 1 ELSE 0 END), 0), " " COALESCE(SUM(CASE WHEN reason = 'local_midnight' " " THEN bytes_written ELSE 0 END), 0), " " COALESCE(SUM(CASE WHEN reason = 'local_midnight' " " THEN bytes_read ELSE 0 END), 0), " " COALESCE(SUM(CASE WHEN reason != 'local_midnight' " " THEN bytes_written ELSE 0 END), 0), " " COALESCE(SUM(CASE WHEN reason != 'local_midnight' " " THEN bytes_read ELSE 0 END), 0) " "FROM local_day_unallocated_evidence " "WHERE (start_local_date = ? AND start_tz_name = ?) " " OR (end_local_date = ? AND end_tz_name = ?) " " OR (start_tz_name = ? AND end_tz_name = ? " " AND start_local_date < ? AND end_local_date > ?)", ( recorded_date, recorded_tz, recorded_date, recorded_tz, recorded_tz, recorded_tz, recorded_date, recorded_date, ), ).fetchone() ( evidence_count, shared_evidence_count, unallocated_evidence_count, shared_bytes_written, shared_bytes_read, unallocated_bytes_written, unallocated_bytes_read, ) = evidence local_day_id = conn.execute( "SELECT id FROM local_days WHERE local_date = ? AND tz_name = ?", (recorded_date, recorded_tz), ).fetchone()[0] segments = [ { "segment_id": segment_id, "bytes_written": segment_written, "bytes_read": segment_read, "activity_seconds": segment_seconds, "activity_intervals": segment_intervals, } for ( segment_id, segment_written, segment_read, segment_seconds, segment_intervals, ) in conn.execute( "SELECT segment_id, bytes_written, bytes_read, activity_seconds, " " activity_intervals " "FROM local_day_segment_totals WHERE local_day_id = ? " "ORDER BY segment_id", (local_day_id,), ) ] interval_count = activity_intervals or 0 has_unallocated = evidence_count > 0 trusted_precision = activity_precision in ("measured", "coarse") has_known = trusted_precision and interval_count > 0 if not has_known and not has_unallocated: activity_state = "incomplete" if bool(activity_incomplete) else "unavailable" else: now = clock_now or datetime.now(timezone.utc) is_current_day = ( now.astimezone(ZoneInfo(recorded_tz)).date().isoformat() == recorded_date ) day_seconds = int( (_utc_datetime(utc_end) - _utc_datetime(utc_start)).total_seconds() ) interval_seconds = activity_seconds or 0 full_activity_day = interval_seconds >= day_seconds or ( interval_count == 0 and bool(complete) ) incomplete = ( bool(activity_incomplete) or has_unallocated or not full_activity_day ) if is_current_day: activity_state = "so_far" elif incomplete: activity_state = "incomplete" elif has_known and (bytes_written or 0) == 0 and (bytes_read or 0) == 0: activity_state = "zero" else: activity_state = "complete" exact = trusted_precision and has_known return { "local_date": recorded_date, "tz_name": recorded_tz, "tz_offset": tz_offset, "utc_start": utc_start, "utc_end": utc_end, "bytes_written": bytes_written if exact else None, "bytes_read": bytes_read if exact else None, "shared_evidence_count": shared_evidence_count, "unallocated_evidence_count": unallocated_evidence_count, "shared_bytes_written": shared_bytes_written, "shared_bytes_read": shared_bytes_read, "unallocated_bytes_written": unallocated_bytes_written, "unallocated_bytes_read": unallocated_bytes_read, "coverage": coverage, "sample_count": sample_count, "complete": bool(complete), "activity_seconds": activity_seconds or 0, "activity_intervals": interval_count, "activity_incomplete": bool(activity_incomplete), "activity_precision": activity_precision, "activity_state": activity_state, "segments": segments, } def query_current_local_day( conn: sqlite3.Connection, clock_now: datetime, tz_name: str, ) -> dict | None: """Query the local-day summary for the current local date.""" from zoneinfo import ZoneInfo local_tz = ZoneInfo(tz_name) local_dt = clock_now.astimezone(local_tz) local_date = local_dt.strftime("%Y-%m-%d") return query_local_day_summary(conn, local_date, tz_name, clock_now) def _is_detail_available( conn: sqlite3.Connection, utc_start: str, utc_end: str, now: datetime, retention_days: int = 14, ) -> bool: """Check if raw samples covering the local-day range are still retained. Returns True if at least one sample within [utc_start, utc_end] is younger than retention_days. This is a conservative check; the actual pruning boundary depends on boundary-anchor logic. """ cutoff = now - timedelta(days=retention_days) cutoff_iso = cutoff.isoformat() # If any sample in the range is newer than cutoff, detail is available row = conn.execute( "SELECT 1 FROM samples WHERE ts >= ? AND ts < ? AND ts >= ? LIMIT 1", (utc_start, utc_end, cutoff_iso), ).fetchone() return row is not None def query_local_day_history( conn: sqlite3.Connection, start_date: str, end_date: str, now: datetime, ) -> list[LocalDayHistoryEntry]: """Query local-day summaries for a date range with evidence-limit metadata. Returns history entries sorted by local_date, each annotated with whether the underlying raw detail is still available (within the 14-day retention window) or has been pruned. Timezone information and UTC boundaries are always present, even after detail expires. Args: conn: Connection to the observation store. start_date: Inclusive start date (e.g. "2026-09-01"). end_date: Inclusive end date (e.g. "2026-09-14"). now: Current UTC time for retention boundary check. Returns: List of LocalDayHistoryEntry sorted by local_date. """ rows = conn.execute( "SELECT local_date, tz_name, tz_offset, utc_start, utc_end, " " bytes_written, bytes_read, coverage, sample_count, complete, " " activity_seconds, activity_intervals, activity_incomplete, " " activity_precision " "FROM local_days " "WHERE local_date >= ? AND local_date <= ? " "ORDER BY local_date, id", (start_date, end_date), ).fetchall() entries = [] for row in rows: summary = query_local_day_summary(conn, row[0], row[1], now) if summary is None: continue detail_available = _is_detail_available(conn, row[3], row[4], now) entries.append( LocalDayHistoryEntry.from_summary( summary, detail_available=detail_available, derived_from_surviving=True, ) ) return entries