Report truncation when a health stream hits its record ceiling (#613)

docs/design/health-connect-expansion.md

@@ -45,9 +45,14 @@ temperatures, weight, height, body fat, bone mass, lean body mass, VO2 max) and
 bucket over a wide range (max 1000 buckets). Aggregation streams records page by page and folds
 them as they arrive, so a wide range is never materialised.
 
-A single aggregation folds at most `MAX_RECORDS` (50,000) samples. A range denser than that is
-reported with `truncated: true` and a `note` naming how far the buckets actually reach — the
-buckets then cover only the earliest part of the range, and nothing claims otherwise.
+A single aggregation runs under two ceilings that count different things: it folds at most
+`MAX_RECORDS` (50,000) samples, and the paged read beneath it walks at most `STREAM_MAX_RECORDS`
+(51,000) records. Records normally yield at least one sample each, so the sample cap binds first;
+where they yield fewer — a `HeartRate` record carrying few or no samples — the record ceiling binds
+instead and the fold never reaches its cap. Whichever ceiling stopped it, the answer is reported
+with `truncated: true` and a `note` naming how far the buckets actually reach — the buckets then
+cover only the earliest part of the range, and nothing claims otherwise. A range that simply ran
+out is not truncated, however small the answer.
 
 Health Connect's own `aggregateGroupByDuration` is deliberately not used: as of
 `connect-client:1.1.0` it has no `AggregateMetric` for BloodGlucose, HRV, SpO2, respiratory rate, or

integrations/src/main/java/com/rousecontext/integrations/health/HealthConnectMcpServer.kt

@@ -101,8 +101,8 @@ internal class QueryHealthDataTool(private val repository: HealthConnectReposito
             "only the range's earliest slice, so the answer is instead per-period " +
             "stats {start,count,min,max,avg} across the range, with the chosen " +
             "'bucket' width and a 'note' saying so; narrow the range or raise " +
-            "'limit' for raw records. A very dense range can exceed the sample " +
-            "ceiling aggregation folds, in which case 'truncated' is true and the " +
+            "'limit' for raw records. A range can exceed the ceilings aggregation " +
+            "works under, in which case 'truncated' is true and the " +
             "buckets cover only the earliest part of it. Record types that cannot be aggregated " +
             "(sessions, multi-value, cumulative) return records evenly spread " +
             "across the range with downsampled=true. " +
@@ -218,9 +218,10 @@ internal class QueryHealthDataTool(private val repository: HealthConnectReposito
     }
 
     /**
-     * Says what the buckets actually cover. Aggregation stops at a sample
-     * ceiling, so on a very dense range the buckets end early — claiming
-     * whole-range coverage there would be false.
+     * Says what the buckets actually cover. Aggregation stops at a ceiling — its
+     * own sample cap, or the read's record ceiling — so on a range that trips
+     * either one the buckets end early, and claiming whole-range coverage there
+     * would be false.
      */
     private fun coverageNote(query: QueryResult.Buckets, to: Instant): String =
         if (query.truncated) {
@@ -231,7 +232,8 @@ internal class QueryHealthDataTool(private val repository: HealthConnectReposito
 
     private fun truncationNote(buckets: List<Bucket>): String {
         val covered = buckets.lastOrNull()?.start
-        return "Aggregation stopped at its ceiling of $MAX_RECORDS samples, so these " +
+        return "Aggregation stopped at a ceiling (at most $MAX_RECORDS samples, from at " +
+            "most $STREAM_MAX_RECORDS records), so these " +
             "buckets cover only the earliest part of the range" +
             (if (covered != null) ", up to $covered" else "") +
             " — narrow 'since'/'until' to see the rest."

integrations/src/main/java/com/rousecontext/integrations/health/HealthConnectRepository.kt

@@ -24,8 +24,9 @@ interface HealthConnectRepository {
      * would return its *earliest slice*, which is not an answer about the range.
      * So the query is instead answered with [QueryResult.Buckets]: per-window
      * count/min/max/avg across the range, streamed and folded so the range is
-     * never materialised. A range holding more than [MAX_RECORDS] samples folds
-     * only the earliest [MAX_RECORDS] of them and says so via
+     * never materialised. A range that trips either aggregation ceiling — more
+     * than [MAX_RECORDS] samples to fold, or more than [STREAM_MAX_RECORDS]
+     * records to read through — folds only its earliest part and says so via
      * [QueryResult.Buckets.truncated]. Record types that cannot be bucketed
      * (sessions, multi-value, cumulative) fall back to [QueryResult.Records]
      * evenly spread across the range with `downsampled` set.
@@ -112,8 +113,11 @@ sealed class QueryResult {
      * @param buckets per-window count/min/max/avg, ordered by start time
      * @param width the window width the range was divided into
      * @param totalCount number of samples aggregated
-     * @param truncated true when aggregation stopped at [MAX_RECORDS] samples, so
-     *   [buckets] cover only the earliest part of the requested range
+     * @param truncated true when aggregation stopped at a ceiling — the range
+     *   held more than [MAX_RECORDS] samples to fold, or more than
+     *   [STREAM_MAX_RECORDS] records to read through — so [buckets] cover only
+     *   the earliest part of the requested range. A range that simply ended is
+     *   not truncated, however few samples it held.
      */
     data class Buckets(
         val buckets: List<Bucket>,
@@ -133,8 +137,11 @@ sealed class QueryResult {
 sealed class BucketResult {
 
     /**
-     * @param truncated true when aggregation stopped at [MAX_RECORDS] samples, so
-     *   [buckets] cover only the earliest part of the requested range
+     * @param truncated true when aggregation stopped at a ceiling — the range
+     *   held more than [MAX_RECORDS] samples to fold, or more than
+     *   [STREAM_MAX_RECORDS] records to read through — so [buckets] cover only
+     *   the earliest part of the requested range. A range that simply ended is
+     *   not truncated, however few samples it held.
      */
     data class Success(
         val buckets: List<Bucket>,
@@ -153,8 +160,9 @@ sealed class BucketResult {
  * It does not bound a streamed read directly: a stream carries records, and the
  * fold that consumes them counts samples. Records normally yield at least one
  * sample each, so the sample cap trips first and the stream ends within a page of
- * this many records — but a record that yields none would not trip it at all,
- * which is what [STREAM_MAX_RECORDS] exists to stop.
+ * this many records — but a record that yields fewer would not trip it at all,
+ * which is what [STREAM_MAX_RECORDS] exists to stop. Truncation is therefore
+ * reported when either ceiling binds, not only this one.
  */
 const val MAX_RECORDS: Int = 50_000
 
@@ -178,9 +186,14 @@ const val READ_PAGE_SIZE: Int = 1000
 /**
  * Hard ceiling on records a single streamed read will fetch.
  *
- * One page of slack over [MAX_RECORDS], so it never cuts short a fold that the
- * sample cap would have satisfied, and it still terminates a range of records
- * that yield no samples at all.
+ * One page of slack over [MAX_RECORDS], so it does not cut short a fold when
+ * records yield at least one sample each — the sample cap trips first, and at
+ * the exact one-sample boundary the slack page is precisely what carries the
+ * fold to its cap. Below that yield this ceiling *is* what stops the read, which
+ * is the point: it terminates a range of records that yield no samples at all.
+ * A read that stops here says so, ending its stream with
+ * [com.rousecontext.integrations.health.query.Streamed.CeilingReached], so the
+ * fold reports the shortfall rather than passing it off as whole-range coverage.
  */
 const val STREAM_MAX_RECORDS: Int = MAX_RECORDS + READ_PAGE_SIZE
 

integrations/src/main/java/com/rousecontext/integrations/health/RealHealthConnectRepository.kt

@@ -14,6 +14,7 @@ import com.rousecontext.integrations.health.query.NutritionQueries
 import com.rousecontext.integrations.health.query.RecordReader
 import com.rousecontext.integrations.health.query.ReproductiveQueries
 import com.rousecontext.integrations.health.query.SleepQueries
+import com.rousecontext.integrations.health.query.Streamed
 import com.rousecontext.integrations.health.query.VitalsQueries
 import com.rousecontext.integrations.health.query.bucketize
 import com.rousecontext.integrations.health.query.downsampleEvenly
@@ -201,7 +202,9 @@ internal typealias PageFetcher =
  * cap costs a small read instead of materialising the whole range. A stream is
  * normally bounded by its collector, which ends collection once it has folded
  * enough; it also carries its own [STREAM_MAX_RECORDS] ceiling so that records
- * yielding nothing to fold cannot walk the whole range.
+ * yielding nothing to fold cannot walk the whole range. Stopping there ends the
+ * stream with [Streamed.CeilingReached], so the collector knows the range held
+ * more than it saw.
  *
  * [fetchPage] is a seam so pagination can be unit-tested without a real client.
  */
@@ -218,30 +221,46 @@ internal class HealthConnectClientRecordReader(private val fetchPage: PageFetche
         maxRecords: Int
     ): List<T> {
         val accumulated = mutableListOf<T>()
-        pages(type, from, to, maxRecords).collect { accumulated += it }
+        // The caller's own cap stopping the read is not news to the caller, so
+        // the terminal note is of no interest here.
+        pages(type, from, to, maxRecords).collect { item ->
+            if (item is Streamed.Value) accumulated += item.value
+        }
         return accumulated.take(maxRecords)
     }
 
-    override fun <T : Record> stream(type: KClass<T>, from: Instant, to: Instant): Flow<T> =
-        pages(type, from, to, STREAM_MAX_RECORDS).transform { page -> page.forEach { emit(it) } }
+    override fun <T : Record> stream(
+        type: KClass<T>,
+        from: Instant,
+        to: Instant
+    ): Flow<Streamed<T>> = pages(type, from, to, STREAM_MAX_RECORDS).transform { item ->
+        when (item) {
+            is Streamed.Value -> item.value.forEach { emit(Streamed.Value(it)) }
+            Streamed.CeilingReached -> emit(Streamed.CeilingReached)
+        }
+    }
 
     /**
      * Pages of records, requesting no more per page than [maxRecords] still needs
      * and stopping once that many have been fetched. Cancelling collection stops
      * the read, so a collector that needs only the first few records pays for one
      * page.
+     *
+     * Ends with [Streamed.CeilingReached] when [maxRecords] is what stopped it
+     * while the source still had more to give, and without it when the range ran
+     * out — the distinction a fold needs to report its coverage honestly.
      */
     @Suppress("UNCHECKED_CAST")
     private fun <T : Record> pages(
         type: KClass<T>,
         from: Instant,
         to: Instant,
         maxRecords: Int
-    ): Flow<List<T>> = flow {
+    ): Flow<Streamed<List<T>>> = flow {
         val filter = TimeRangeFilter.between(from, to)
         var fetched = 0
         var pageToken: String? = null
-        do {
+        while (fetched < maxRecords) {
             val response = fetchPage(
                 ReadRecordsRequest(
                     recordType = type,
@@ -250,11 +269,15 @@ internal class HealthConnectClientRecordReader(private val fetchPage: PageFetche
                     pageToken = pageToken
                 )
             ) as ReadRecordsResponse<T>
-            emit(response.records)
+            emit(Streamed.Value(response.records))
             fetched += response.records.size
             pageToken = response.pageToken
             // An empty page ends the read: without this a token that never
             // advances would loop forever.
-        } while (pageToken != null && response.records.isNotEmpty() && fetched < maxRecords)
+            if (pageToken == null || response.records.isEmpty()) return@flow
+        }
+        // Left the loop with the source still handing out pages: this read
+        // stopped at its own ceiling, not at the end of the range.
+        emit(Streamed.CeilingReached)
     }
 }

integrations/src/main/java/com/rousecontext/integrations/health/query/BodyQueries.kt

@@ -8,7 +8,6 @@ import androidx.health.connect.client.records.Vo2MaxRecord
 import androidx.health.connect.client.records.WeightRecord
 import java.time.Instant
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.map
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.put
@@ -122,22 +121,25 @@ class BodyQueries(private val reader: RecordReader) : CategoryQueries {
         else -> throw IllegalArgumentException("Unsupported record type: $recordType")
     }
 
-    override fun bucketValues(recordType: String, from: Instant, to: Instant): Flow<TimedValue>? =
-        when (recordType) {
-            "Weight" -> reader.stream(WeightRecord::class, from, to)
-                .map { TimedValue(it.time, it.weight.inKilograms) }
-            "Height" -> reader.stream(HeightRecord::class, from, to)
-                .map { TimedValue(it.time, it.height.inMeters) }
-            "BodyFat" -> reader.stream(BodyFatRecord::class, from, to)
-                .map { TimedValue(it.time, it.percentage.value) }
-            "BoneMass" -> reader.stream(BoneMassRecord::class, from, to)
-                .map { TimedValue(it.time, it.mass.inKilograms) }
-            "LeanBodyMass" -> reader.stream(LeanBodyMassRecord::class, from, to)
-                .map { TimedValue(it.time, it.mass.inKilograms) }
-            "Vo2Max" -> reader.stream(Vo2MaxRecord::class, from, to)
-                .map { TimedValue(it.time, it.vo2MillilitersPerMinuteKilogram) }
-            else -> null
-        }
+    override fun bucketValues(
+        recordType: String,
+        from: Instant,
+        to: Instant
+    ): Flow<Streamed<TimedValue>>? = when (recordType) {
+        "Weight" -> reader.stream(WeightRecord::class, from, to)
+            .mapValues { TimedValue(it.time, it.weight.inKilograms) }
+        "Height" -> reader.stream(HeightRecord::class, from, to)
+            .mapValues { TimedValue(it.time, it.height.inMeters) }
+        "BodyFat" -> reader.stream(BodyFatRecord::class, from, to)
+            .mapValues { TimedValue(it.time, it.percentage.value) }
+        "BoneMass" -> reader.stream(BoneMassRecord::class, from, to)
+            .mapValues { TimedValue(it.time, it.mass.inKilograms) }
+        "LeanBodyMass" -> reader.stream(LeanBodyMassRecord::class, from, to)
+            .mapValues { TimedValue(it.time, it.mass.inKilograms) }
+        "Vo2Max" -> reader.stream(Vo2MaxRecord::class, from, to)
+            .mapValues { TimedValue(it.time, it.vo2MillilitersPerMinuteKilogram) }
+        else -> null
+    }
 
     override suspend fun summary(from: Instant, to: Instant, granted: Set<String>): JsonObject =
         buildJsonObject {

integrations/src/main/java/com/rousecontext/integrations/health/query/BucketAggregation.kt

@@ -39,9 +39,11 @@ data class Bucket(
 /**
  * [buckets] plus the number of samples they were folded from.
  *
- * @param truncated true when the fold stopped at its sample ceiling, so [buckets]
+ * @param truncated true when a ceiling stopped the aggregation short — either the
+ *   fold's own sample cap, or the upstream reader's record ceiling — so [buckets]
  *   cover only the earliest part of the requested range. Callers must not claim
- *   whole-range coverage when this is set.
+ *   whole-range coverage when this is set. It is not set merely because the
+ *   answer is small: a range that simply ran out is complete.
  */
 data class BucketedValues(val buckets: List<Bucket>, val totalCount: Int, val truncated: Boolean)
 
@@ -57,9 +59,16 @@ data class BucketedValues(val buckets: List<Bucket>, val totalCount: Int, val tr
  * At most [maxValues] are folded; hitting that ceiling stops collection (and so
  * the underlying read) and sets [BucketedValues.truncated], because the buckets
  * then describe only the earliest part of the range.
+ *
+ * [maxValues] counts *values*, so it cannot be the only guard: a producer whose
+ * records yield fewer than one value each stops at a ceiling of its own long
+ * before this cap trips. That producer says so with [Streamed.CeilingReached],
+ * which sets [BucketedValues.truncated] just the same — otherwise a fold that
+ * was cut short would report coverage it does not have simply because it was the
+ * cap that did not trip.
  */
 suspend fun bucketize(
-    values: Flow<TimedValue>,
+    values: Flow<Streamed<TimedValue>>,
     from: Instant,
     width: Duration,
     maxValues: Int
@@ -70,8 +79,18 @@ suspend fun bucketize(
     var total = 0
     var truncated = false
     // One past the ceiling: enough to tell "that was all of it" from "there is
-    // more", without folding the extra value.
-    values.take(maxValues + 1).collect { tv ->
+    // more", without folding the extra value. The terminal note, when there is
+    // one, arrives within that same slack.
+    values.take(maxValues + 1).collect { item ->
+        val tv = when (item) {
+            // Upstream stopped at a bound of its own, so the range holds more
+            // than these buckets describe — even though the cap never tripped.
+            Streamed.CeilingReached -> {
+                truncated = true
+                return@collect
+            }
+            is Streamed.Value -> item.value
+        }
         if (total == maxValues) {
             truncated = true
             return@collect

integrations/src/main/java/com/rousecontext/integrations/health/query/CategoryQueries.kt

@@ -48,6 +48,12 @@ interface CategoryQueries {
      * Streaming is what lets a window wider than any record cap be aggregated
      * without materialising it. The default returns `null`; categories that own
      * scalar types override this.
+     *
+     * The stream carries [RecordReader.stream]'s terminal
+     * [Streamed.CeilingReached] note through the mapping, so the fold can tell a
+     * range that ran out from one the reader stopped reading. Use [mapValues] /
+     * [flatMapValues] rather than the plain flow operators, which would drop it.
      */
-    fun bucketValues(recordType: String, from: Instant, to: Instant): Flow<TimedValue>? = null
+    fun bucketValues(recordType: String, from: Instant, to: Instant): Flow<Streamed<TimedValue>>? =
+        null
 }

integrations/src/main/java/com/rousecontext/integrations/health/query/RecordReader.kt

@@ -40,8 +40,14 @@ interface RecordReader {
      * `maxValues`. That cap counts samples, though, so a record yielding none
      * would never trip it; implementations MUST also stop at
      * [STREAM_MAX_RECORDS] records so such a range terminates.
+     *
+     * When that record ceiling is what stopped the read, implementations MUST
+     * end the stream with [Streamed.CeilingReached]. Without it the collector
+     * cannot tell a range that ran out from one the reader abandoned, and a fold
+     * whose own cap never tripped would report whole-range coverage it does not
+     * have. A stream that reached the end of the range ends without the note.
      */
-    fun <T : Record> stream(type: KClass<T>, from: Instant, to: Instant): Flow<T>
+    fun <T : Record> stream(type: KClass<T>, from: Instant, to: Instant): Flow<Streamed<T>>
 }
 
 /**