hub v0.73.0 — offsite_stale anchored on newborn tiers (never-ran = applied-only + consumed_at/escrow anchor; one state one owner)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NKSN3gSg4TKVBBqkwW2djR

hub/internal/store/store.go

@@ -967,6 +967,36 @@ func (s *Store) CountReportsOffsiteSince(customerID string, since time.Time) (to
 	return total, withOffsite, rows.Err()
 }
 
+// LatestEscrowTimeForCustomer returns the newest escrow-blob timestamp across the customer's
+// hosts (updated_at, falling back to created_at), or zero when no blob exists. Part-7 v0.73.0:
+// one leg of the never-ran offsite-staleness anchor — the escrow ceremony is the moment runs
+// become POSSIBLE, so the 48 h staleness threshold counts from it, not from the dawn of time.
+// Reduced in Go (not SQL MAX) because created_at formats vary (RFC3339 vs SQLite space form) and
+// lexicographic MAX would misorder them; parseSQLiteTime handles both.
+func (s *Store) LatestEscrowTimeForCustomer(customerID string) (time.Time, error) {
+	rows, err := s.db.Query(`
+		SELECT he.created_at, he.updated_at FROM host_escrow he
+		INNER JOIN hosts h ON h.host_id = he.host_id
+		WHERE h.customer_id = ?`, customerID)
+	if err != nil {
+		return time.Time{}, err
+	}
+	defer rows.Close()
+	var newest time.Time
+	for rows.Next() {
+		var createdAt, updatedAt string
+		if err := rows.Scan(&createdAt, &updatedAt); err != nil {
+			return time.Time{}, err
+		}
+		for _, raw := range []string{createdAt, updatedAt} {
+			if t := parseSQLiteTime(raw); t.After(newest) {
+				newest = t
+			}
+		}
+	}
+	return newest, rows.Err()
+}
+
 // LastEventAt returns the created_at of the most recent event of the given type for a customer
 // (zero time when none). Durable across hub restarts — used as the cooldown/rate-limit source for
 // hub-emitted detector events (R-70/R-71c: a repeating pattern must surface as repeating events on