Olly Docs

Appearance

Updated Aug 22, 2026

Wearable streams: the compaction pipeline ​

Schema deep-dive · living document · #23 in the reading sequence

Tablesstreams.episode[1], streams.rollup[2], streams.baseline[3] (fed from streams.event_raw[4])
Owner servicebalance (Python / FastAPI / asyncpg / Alembic; sole writer)
Locatornatural key - no locators. uuid PKs; profile_id is the tenancy scope. Uniqueness: rollup (profile_id, episode_type, day, metric), baseline (profile_id, metric, window_days), episode none (application-deduped, §4)
Last updated2026-08-28
Companionprevious: Wearable streams: the registries (#22) · next: Wearable streams: scores and findings (#24) · triage streams sidecar (POST /streams/digest) · member-app activity / timeline (movement history)

1. Scope and usage ​

These three tables are the raw-to-derived heart of the wearable-streams service. A phone pushes verbatim Health Connect / HealthKit records into streams.event_raw (the raw sample store, ~157k rows, covered on the registries page); everything past that is derived, and the derivation runs here:

event_raw (compacted_at IS NULL)  --run_compaction-->  episode
episode  --rollup_profile_day-->  rollup  --compute_baseline-->  baseline

episode is a compacted span of one activity or state - a night's sleep, a day's steps, a workout - carrying its readings as a JSONB observations array. rollup is the per-metric, per-day aggregate folded out of those episodes. baseline is the per-user, per-metric rolling mean+stddev the detectors compare each new day against. The order is a strict dependency chain: a baseline is only as current as the rollups under it, and a rollup is only as current as the episodes under it.

Live population as of 2026-08-21: 3 927 episodes (2 334 derived, 978 measured, 615 parsed; 198 soft-deleted), 5 004 rollups, 112 baselines. Upstream, event_raw holds 157 070 rows of which 1 865 are still pending (compacted_at IS NULL) - the backlog the compaction worker exists to drain.

The compaction worker is the only writer of episodes stamped (source_modality='wearable', origin='derived') from the push surface; the intent is marked with a /* d50 compaction */ SQL comment on every insert, and the docstring states the invariant plainly (V1 does not enforce it at runtime)[5]. A second, unrelated writer inserts parsed/measured episodes from the chat/document parser path (insert_episode)[6]; the pipeline below folds both kinds identically once they are episodes.

2. Boundaries and relationships ​

This is not…That concern lives inJoin
the raw samplestreams.event_raw - one HC/HK reading per row, kept verbatim; compaction reads it, never the reverse (registries page)event_raw.compacted_at gate; no FK back
the scorestreams.movement_score / shape_score - shaped 0-100 numbers (scores page)reads rollup + baseline
the finding / insightstreams.finding, insight, knowledge_gap - what a detector concluded from a baseline comparisonfinding_evidence.episode_id FK to episode
the read seriesread/rollups.py sums rollup rows per day for the API; the stored rollup is per (episode_type, day, metric), so a day split across episode_types is several rows the read path collapses[7]SUM(value) GROUP BY day
the member identitystreams.profile - the balance-local profile that maps to a PTY- party (registries page)profile_id FK, ON DELETE CASCADE

profile_id is the tenancy key on all three tables, a real in-schema FK to streams.profile(id) with ON DELETE CASCADE[1][2][3] - deleting a profile cascades the entire derived chain away. metric on rollup and baseline is a soft text reference with no FK to the streams.metric registry (§4): the same reading can appear under two spellings until the taxonomy settles (§9).

3. Structure ​

DDL: episode[1] · rollup[2] · baseline[3]. Tables are raw SQL inside Alembic migrations (there is no ORM); the Pydantic models in streams/models.py are API shapes, not table definitions.

streams.episode ​

FieldTypeReqNotes
iduuid✓PK, gen_random_uuid()
profile_iduuid✓FK streams.profile, ON DELETE CASCADE
domaintext✓movement / sleep / body / … (the pillar family)
episode_typetext✓FK to streams.episode_type(id) - the only registry-checked column here; drives the rollup fold rule (§3 below)
start_attimestamptz✓span start; per-day episodes use 00:00:00Z
end_attimestamptzspan end; per-day episodes use 23:59:59.999999Z
source_modalitytext✓wearable for compaction inserts
source_vendortextthe SDK/vendor (health-connect, oura, …)
origintext✓measured / derived / parsed / inferred; compaction always writes derived
confidencedouble✓default 1.0
attributiontext[]✓default {}; the originating app package(s)
observationsjsonb✓default []; the readings - {metric, value, unit, origin, confidence} each. Append-only per compaction run (§9)
superseded_by / supersedesuuidself-FKs for correction chains (parser path)
deleted_attimestamptzsoft delete; live heads exclude it

episode_live_heads_idx is a partial index on (profile_id, start_at) WHERE deleted_at IS NULL AND superseded_by IS NULL[8] - the shape every reader uses to see only the current, non-deleted episodes.

streams.rollup ​

FieldTypeReqNotes
iduuid✓PK
profile_iduuid✓FK streams.profile, cascade
episode_typetext✓which episode family produced it; no FK (soft)
daydate✓the UTC calendar day rolled up
metrictext✓the folded metric (steps, sleep_duration, …); no FK (soft)
valuedouble✓the folded number
session_countintegernon-NULL only when the episode_type folds by session (else NULL)
computed_attimestamptz✓now(), refreshed on every upsert

UNIQUE (profile_id, episode_type, day, metric)[2] is the idempotency key: the rollup writer upserts on this tuple, so re-running the fold for a day overwrites rather than duplicates.

streams.baseline ​

FieldTypeReqNotes
iduuid✓PK
profile_iduuid✓FK streams.profile, cascade
metrictext✓soft reference (no FK)
window_daysinteger✓the rolling window (7 / 28 / 30 / 60 / 90…)
valuedouble✓the window mean
stddevdoublepopulation stddev (NULL/0 for a single sample)
ninteger✓sample size - counts rollup rows, not distinct days (§4, §6)
computed_attimestamptz✓now(), refreshed on upsert

UNIQUE (profile_id, metric, window_days)[3] is the upsert key: one baseline row per user per metric per window.

Field-by-field: what and why ​

episode.observations (jsonb) is append-per-run. A per-day aggregator does not rewrite the day's episode - it looks up the existing derived episode for (profile_id, episode_type, day) and, if found, concatenates the new observations onto the array with observations || %s::jsonb; only when none exists does it insert[9]. So a day that receives ten hourly step pushes over ten sync cycles ends with ten steps observations on one episode row (the live example in §6 has exactly this shape). The rollup fold collapses them, so readers never see the redundancy - but the episode row itself is not deduplicated (§9).

episode.origin and source_modality are stamped constant by the worker. The compaction INSERT hard-codes 'wearable' and 'derived'[10]; domain is looked up from a small episode_type -> domain map, and episode_type itself is derived from the raw wire type (steps and friends become everyday_movement, vitals become recovery_vitals, body metrics become body_metrics, sleep and exercise sessions get dedicated handlers).

rollup.value is a two-stage fold. rollup_profile_day reads the day's live episodes, groups observations per (episode_type, metric), then folds: if the episode_type's rollup_rule is session and the metric is additive it sums; if passthrough it takes the first; otherwise it folds by the metric's type - count sums, continuous/ordinal take the mean, categorical takes the first[11]. session_count is set to the number of episodes only when the rule is session, otherwise NULL. The result is upserted on the UNIQUE tuple[12].

baseline.n counts rollup rows, not distinct days. compute_baseline selects every rollup.value for the metric where day > as_of - window_days AND day <= as_of, then takes mean, pstdev, and len(values)[13]. The SELECT does not group by day. So when the same metric appears under two episode_types on one day (e.g. steps under both everyday_movement and a legacy step_day), that day contributes two rollup rows, the mean is over rows rather than days, and n can exceed window_days (§6 shows n=30 for a 28-day window). The read path's query_rollup_series sums-by-day and so diverges from the baseline's row-wise view[7].

compacted_at is what makes a raw row compact-once. It lives on event_raw, not on these tables, but it governs them: the worker's batch fetch filters WHERE compacted_at IS NULL AND type = ANY(_HANDLED_TYPES)[14], and every processed or deliberately-skipped row is stamped compacted_at = now() in the batch's final UPDATE[15]. The partial index event_raw_pending_idx WHERE compacted_at IS NULL[4] keeps that scan cheap as the table grows.

4. Invariants ​

InvariantEnforced by
One rollup per (profile, episode_type, day, metric)DB UNIQUE[2]; upsert ON CONFLICT … DO UPDATE[12]
One baseline per (profile, metric, window_days)DB UNIQUE[3]; upsert[16]
Every derived row belongs to a real profileDB FK profile_id ON DELETE CASCADE (all three tables)[1]
Every episode has a registered episode_typeDB FK episode_type -> streams.episode_type(id)[1]
rollup.episode_type / rollup.metric / baseline.metric name real registry entriesNothing - plain text, no FK; a typo or a legacy spelling persists silently (dual-name coexistence, §9)
A raw row is compacted at most onceApplication: batch fetch filters compacted_at IS NULL[14] + same-commit stamp to now()[15]; DB partial index accelerates it but does not enforce it
One derived episode per (profile, episode_type, day)Application only: attach-or-insert via SELECT … LIMIT 1 then UPDATE/INSERT[9]; no DB uniqueness - two concurrent runs could double-insert (§9)
Duplicate raw records are dropped at pushDB UNIQUE (profile_id, source_platform, source_id) + ON CONFLICT DO NOTHING[17]
Same activity from two apps yields one episodeApplication: origin-priority dedup (_pick_winning_origin) keeps the best-ranked source, skip-stamps the losers[18]
An unhandled raw type never stalls the workerApplication, belt-and-suspenders: type-filter at fetch[14] and skip-stamp any row that slips through[19] (bug 4, #1511)
Row changes captured to CDCDebezium publication dbz_balance (all three tables, live \d)

5. Lifecycle ​

Not a status machine - a derivation pipeline. Every arrow is a function that reads the stage before it and upserts the stage after.

recompute_profile is the orchestrator that runs the whole pass for one profile: step 0 compacts pending raw first (so the rollup sees a just-arrived sync), step 1 rolls up every day in the window, step 2 computes baselines and runs detectors per shape[20]. Compaction-before-rollup was wired 2026-08-13; before that a real wearable sync stranded as pending event_raw and never reached rollup, because only the manual admin endpoint compacted[21]. Step 0 loops rather than compacting once: run_compaction processes one batch_size (500 rows) per call, so a sync or import larger than one batch used to leave the remainder pending until the next trigger. The loop is bounded at 200 passes and stops as soon as a pass compacts and skips nothing, so a row that can never be stamped ends it rather than spinning[32].

What triggers a pass:

  • On ingest (durable, debounced). A wearable push with at least one accepted record calls dispatch_ingest_compaction, which signal-with-starts the Temporal CompactionWorkflow when balance.compaction.via_temporal is on and Temporal is reachable, and otherwise falls back to the in-process schedule_on_ingest[24][33]. Either way the pass is coalesced ~15s after the last chunk (a real sync is many chunked pushes, same balance.engine.ingest.debounce_seconds flag), so the push response stays fast[25]. On the Temporal path the debounce is a durable workflow timer and the workflow id is per-profile (compaction-<profile_id>), so a multi-chunk sync collapses to one run and the pass survives a worker restart; the activity carries a 15-minute start-to-close timeout and a bounded retry policy[34]. The narration (LLM) pass is skipped here and left to nightly. A chat/document episode that the Gate auto-bands dispatches the same way[35].
  • Nightly. The in-process APScheduler still runs _daily_pass on cron 0 4 * * * (04:00 UTC), recomputing every active profile - active meaning a live episode in the last 90 days or pending raw waiting to be compacted[22][23]. This is the one pass Temporal has not taken over yet.
  • On demand. POST /ingest/admin/compaction/run compacts the caller's own pending raw synchronously (any authenticated profile; no admin RBAC in V1)[26].

The recompute / late-data path. When a backfilled push lands late, its raw rows arrive as new compacted_at IS NULL rows; the next trigger compacts them, and because the day's derived episode already exists the worker appends the new observations to it rather than making a second episode[9]. recompute_profile then re-rolls every day in the trailing window and re-derives the baselines by upsert, so late data inside the window flows through end to end[27]. Data that lands on a day outside the trailing window is the caveat in §9.

Why the on-ingest trigger moved to Temporal (2026-08-27). That pass used to live entirely in the API's own APScheduler, inside the uvicorn --reload worker. It stopped firing silently - no error, no log - and wearable syncs stranded as pending event_raw again: the same symptom the 2026-08-13 compaction-before-rollup fix had cured, from a different cause. A profile's episodes simply stopped advancing while raw kept arriving, and because /movement/history reads rollup and episode rather than raw, the member's week read empty. The Temporal worker runs as its own process (balance-worker), so an edit to the API can no longer un-arm compaction; a pass that fails is retried under a bounded policy and is visible as a workflow rather than lost[36].

6. Populated example: one steps chain, walked end to end ​

Profile aa1851ac-9370-41ab-9119-7fb8b7a5ad09, day 2026-08-19. This is a real, non-PII chain: hourly Health Connect step aggregates -> one movement episode -> one rollup -> the 28-day baseline. All values are live.

The raw rows (upstream context) ​

The FE pushes HC aggregate(1h) buckets, so the day arrives as 15 hourly steps rows, source_id = hc_steps_1h_<hour>, origin_app NULL (the phone's HC aggregate already deduped across apps). Sum of value_num across the 15 rows: 3 649.

type=steps  start_at=2026-08-19T00:43Z  value_num=0     origin_app=<null>
type=steps  start_at=2026-08-19T08:43Z  value_num=702   origin_app=<null>
type=steps  start_at=2026-08-19T09:43Z  value_num=1052  origin_app=<null>
…  (15 rows total, all compacted_at IS NOT NULL)

Because origin_app is NULL on every row, _pick_winning_origin finds no priority match, buckets all rows as "other", and keeps them all - the server-side origin-priority is a no-op here, exactly as intended when the FE aggregate already deduped[18].

The episode ​

json
{
  "id": "026665d8-99ea-473d-8f11-d88c0a82a45b",
  "profile_id": "aa1851ac-9370-41ab-9119-7fb8b7a5ad09",
  "domain": "movement",
  "episode_type": "everyday_movement",
  "start_at": "2026-08-19T00:00:00Z",
  "end_at": "2026-08-19T23:59:59.999999Z",
  "source_modality": "wearable",
  "source_vendor": "health-connect",
  "origin": "derived",
  "confidence": 1.0,
  "attribution": [],
  "observations": "[ 60 observations: 15×steps, plus distance_m / active_calories / floors_climbed ]"
}
KeyRead byWhat actually happens
observations (60 entries)rollup_profile_dayeach of the 15 raw steps rows became one steps observation, appended across sync cycles onto this single per-day episode[9]
episode_type: everyday_movementthe foldits rollup_rule is reduce, so the fold falls to the metric-type rule
steps is metric-type count_FOLD["count"] = "sum"the 15 step observations fold by sum[11]
origin: derived, source_modality: wearableanyone auditing provenancethe constant stamp the compaction INSERT hard-codes[10]

The rollup it feeds ​

The fold produces four rows for (profile, everyday_movement, 2026-08-19) - one per metric on the episode. The steps one:

rollup: (profile=aa1851ac…, everyday_movement, 2026-08-19, steps)
        value = 3649   session_count = NULL   computed_at = 2026-08-21 04:19Z
sibling rows same day: distance_m=0, active_calories=0, floors_climbed=0

session_count is NULL because the rule is reduce, not session[11]. The value equals the raw sum exactly (3 649), because a chain of pure count folds is a straight addition from raw to rollup.

The baseline for the metric ​

baseline: (profile=aa1851ac…, metric=steps, window_days=28)
          value = 9319.4   stddev = 7809.5   n = 30   computed_at = 2026-08-21 04:21Z

The 2026-08-19 rollup of 3 649 is one of the samples inside this window mean. Note n = 30 for a 28-day window: this profile has steps rollups under two episode_types in the window - 26 days of everyday_movement plus 4 legacy step_day rows - and compute_baseline counts rollup rows, not distinct days, so 26 distinct days yield n=30[13]. The mean is likewise over 30 rows. A detector comparing today's steps against this baseline is comparing against a row-weighted, not day-weighted, average (§9).

7. Who references these tables ​

WhereColumn / mechanismMeaning there
streams.finding_evidenceepisode_id FK, ON DELETE CASCADEwhich episodes a detector's finding rests on
streams.alertepisode_id FK, ON DELETE SET NULLthe episode that raised an alert
streams.habit_completion / plan_session_completionepisode_id FK, ON DELETE SET NULLa watch-recorded episode that fulfilled a habit / training session
detectors (trend / anomaly / recovery / sleep)read rollup series + baseline tuplethe comparison that produces findings and insight objects[28]
movement / pillar / visualise surfacesread/rollups.py sums rollup per day[7]the member-app activity/timeline series
pillar snapshotscompute_pillar_snapshot reads rollup only[29]cheap before/after numbers for chat animations, no full recompute
training-plan matcherfulfil_session_for_episode on every insert[10]a compacted run can fulfil a plan session (#1559)

All episode references are real in-service FKs (balance owns the whole streams schema); nothing crosses a service boundary here.

8. Design determinations ​

  1. Compaction is a batch worker keyed on compacted_at, not real-time. Raw is stored verbatim and turned into episodes by a worker that reads pending rows and stamps them once - D-50 (#1491 mapping rules, #1494 worker)[5].
  2. The stall is designed out twice. An unhandled wire type is filtered at the SQL fetch and skip-stamped if it slips through, so no row can loop the worker forever - bug 4, #1511[14][19].
  3. Cross-app dedup is FE-first, server-fallback. The phone's HC aggregate(1h) dedups before push; _ORIGIN_PRIORITY is the server-side net for old clients, non-aggregatable sessions, and late third-party writes - bug 2, #1512[18].
  4. Compact-before-rollup, wired 2026-08-13. recompute_profile compacts pending raw as step 0 so a just-arrived sync reaches rollup in the same pass; the earlier design only compacted via the manual admin route and stranded real syncs[21].
  5. Two-stage rollup (DR-4.10). The fold consults the episode_type's rollup_rule first, then the metric's type, so count metrics sum and continuous metrics average without per-metric branching in the worker[11].
  6. One episode per day, grown by append. Per-day aggregators attach to the day's existing derived episode rather than inserting per raw row, keeping a single row per day across many compaction runs[9].
  7. Baseline is a plain rolling window. Mean + population stddev over the window's rollup rows, upserted per (metric, window_days) - no decay, no outlier trim in V1[13].

9. Caveats and extensibility ​

Group and individual. These tables know nothing about schemes or policies - they are keyed on a balance-local profile_id that maps to one member's PTY- party (registries page). Wearable data is inherently individual; whether that member is a scheme employee or a direct-to-consumer policyholder is an authorisation concern resolved before any of this runs. No schema change is needed for either.

Where to extend. Each stage of the chain is generic over its key: an observation is (metric, value, unit), a rollup is one number per (profile, episode_type, day, metric), a baseline is one comparison per (profile, metric, window_days). New signals travel the chain as data.

When we need …What to addWhere
a new metric to be aggregated (a new wearable reading, a new parsed value)nothing on rollup - the table is generic per (profile, episode_type, day, metric) with the fold selected from the registry, so a new metric folds and upserts with no schema change. Give it a metric_type so _mt knows whether to sum or meanstreams.rollup DDL + UNIQUE[2]; the two-stage fold in engine/rollup.py[11]; the metric row itself on the registries page
a new raw wire type to reach an episodeone entry in _PERDAY_OBSERVATION_SPEC mapping the wire type to (metric, unit, origin), plus the type in _HANDLED_TYPES - it then flows to an episode observation with no new branchengine/compaction.py[30]
a new derived signal (a detector that needs "is today unusual for this member")a baseline row for the metric: compute_baseline already produces the per-metric rolling mean + stddev over any window_days, and the detectors read that tuplestreams.baseline DDL[3]; engine/baseline.py[13]; the per-shape compute that feeds detectors[28]
a new fold behaviour (sum instead of mean, first-value passthrough)change the episode type's rollup_rule or the metric's metric_type - registry data, not worker codeengine/rollup.py _FOLD[11]
late or backdated data to re-derivethe recompute path already exists and is idempotent by upsert - compaction appends the late observations onto the day's episode and recompute_profile re-rolls and re-baselines. Widen window_days (or run a full-history recompute) to reach days outside the trailing window - see the first defect belowrecompute_profile's rollup loop[27]; attach-or-insert[9]
a fresh rollup immediately after a pushthe synchronous sync-time recompaction on the unmerged feature branch (§5) removes the ~15s debounce between a push and a fresh rollupengine/scheduler.py schedule_on_ingest[25] is the pinned path it replaces

Known defects. Each is a real behaviour gap in the derivation, phrased as the fix and where it goes:

  • Late data outside the trailing window does not re-roll.recompute_profile rolls up only as_of - delta for delta in range(window_days) (default 30 days ending today)[27]. A backfilled push dated 60 days ago is compacted into an episode, but its day is not re-rolled by the on-ingest or nightly pass, so no rollup or baseline reflects it. Fix: derive the rollup day set from the episodes the compaction run touched, rather than from a fixed trailing range, in engine/recompute.py L31-L34; the manual full-history recompute is the only path that catches it today.
  • baseline.n counts rollup rows, not distinct days. compute_baseline selects every rollup.value in the window without grouping by day, so a metric split across two episode_types double-counts those days into the mean and stddev - live-proven in §6, where n=30 over 26 distinct days. The read path sums-by-day and so disagrees with the baseline's own view[13][7]. Fix: aggregate the window SELECT by day (SUM(value) … GROUP BY day) to match query_rollup_series, in engine/baseline.py L10-L31.
  • episode.observations grows without bound. Each compaction run appends its new observations to the day's episode, so a day with many hourly pushes carries one observation per raw sample (the §6 episode carries 60). The fold collapses them, so readers are unaffected, but the JSONB itself is not deduplicated. Fix: dedupe on (metric, source window) inside the attach-or-insert before the observations || %s::jsonb concat, in engine/compaction.py L557-L590[9].
  • One-derived-episode-per-day is an application rule with no DB backing. The attach-or-insert reads SELECT … LIMIT 1 then decides[9]; there is no unique constraint on (profile_id, episode_type, day), so two concurrent compaction runs could each insert. The scheduler's per-profile coalescing (max_instances=1) makes the race unlikely, not impossible. Fix: a partial unique index on (profile_id, episode_type, date(start_at)) WHERE origin = 'derived' AND deleted_at IS NULL in a new migration, and turn the insert into an upsert.
  • The episode insert commits before the raw stamp (an at-least-once window). _insert_episode commits mid-function so the plan matcher can read it on a fresh connection[10], while the compacted_at stamp commits at the end of the batch[15]. A crash between the two leaves the episode written and the raw rows still pending, so the next run re-appends the same observations onto that day: the append is idempotent by day, not exactly-once. Fix: move the plan-matcher call after the batch commit so episode insert and stamp share one transaction, in engine/compaction.py L601-L613.

Stated plainly (facts, not defects):

  • metric on rollup and baseline is a soft text reference. With no FK to streams.metric the same reading can persist under two spellings while a taxonomy settles, which is what lets a rename roll through the pipeline without a migration; the recompute code reads both resting_hr and resting_heart_rate and takes the richer series so one insight does not become two[31]. The live rollup census shows exactly that pair (resting_heart_rate 208 rows, resting_hr 10).
  • 1 865 raw rows sit pending (compacted_at IS NULL) as of 2026-08-21: a mix of unhandled types (skip-stamped on the next pass) and rows for profiles the scheduler has not recomputed since their last push.

References ​

Code links are pinned to commit 08e65216 on main (2026-08-28); the file is the anchor if lines drift. Pins are checked mechanically by docs/site/scripts/check-code-refs.py.

  1. balance/migrations/versions/0004_episode.py#L16 - streams.episode DDL + indexes; profile/episode_type FKs
  2. 0007_derived.py#L16 - streams.rollup DDL, UNIQUE (profile_id, episode_type, day, metric)
  3. 0007_derived.py#L30 - streams.baseline DDL, UNIQUE (profile_id, metric, window_days)
  4. 0013_wearable_push.py#L30 - streams.event_raw DDL, compacted_at, event_raw_pending_idx, source UNIQUE
  5. engine/compaction.py#L36 - the ONLY-writer invariant + idempotency docstring
  6. ingest/storage.py#L24 - insert_episode, the parser-path episode writer
  7. read/rollups.py#L14 - query_rollup_series sums rollup per day (diverges from baseline)
  8. 0004_episode.py#L42 - episode_live_heads_idx partial index
  9. compaction.py#L557 - _handle_per_day attach-or-insert (one episode per day, append via jsonb concat)
  10. compaction.py#L601 - _insert_episode INSERT: origin='derived', source_modality='wearable', mid-function commit + plan matcher
  11. engine/rollup.py#L25 - rollup_profile_day: two-stage fold + session_count
  12. rollup.py#L55 - rollup upsert ON CONFLICT … DO UPDATE
  13. engine/baseline.py#L10 - compute_baseline: window SELECT (no group-by-day), mean+pstdev, n = len(rows)
  14. compaction.py#L262 - batch fetch: WHERE compacted_at IS NULL AND type = ANY(_HANDLED_TYPES)
  15. compaction.py#L385 - the batch stamp UPDATE … SET compacted_at = now() + commit
  16. baseline.py#L24 - baseline upsert ON CONFLICT … DO UPDATE
  17. ingest_routes.py#L374 - push INSERT into event_raw, ON CONFLICT (profile_id, source_platform, source_id) DO NOTHING
  18. compaction.py#L194 - _pick_winning_origin: origin-priority dedup, "all-other keeps everything" fallback
  19. compaction.py#L301 - per-day grouping + skip-stamp of unmapped types (bug 4 defense-in-depth)
  20. engine/recompute.py#L20 - recompute_profile: step 0 compact, step 1 rollup, step 2 baselines
  21. recompute.py#L26 - compact-before-rollup wired 2026-08-13 (the stranded-sync fix)
  22. engine/scheduler.py#L36 - _daily_pass active-profile query (live episode OR pending raw)
  23. scheduler.py#L91 - nightly cron default 0 4 * * *
  24. ingest_routes.py#L422 - push triggers dispatch_ingest_compaction (Temporal, else the APScheduler fallback) only when a record was accepted
  25. scheduler.py#L126 - schedule_on_ingest: debounced, coalesced per-profile recompute
  26. ingest_routes.py#L446 - POST /admin/compaction/run on-demand runner
  27. recompute.py#L31 - rollup loop over the trailing window_days only (late-data caveat)
  28. recompute.py#L41 - per-shape baseline compute feeding the detectors
  29. engine/snapshot.py#L34 - compute_pillar_snapshot reads rollup only (no recompute)
  30. compaction.py#L85 - _PERDAY_OBSERVATION_SPEC wire-type -> (metric, unit, origin) map (add-a-type extensibility)
  31. recompute.py#L336 - dual-name read (resting_hr / resting_heart_rate), the soft-metric consequence
  32. recompute.py#L28 - step 0 drains the backlog: run_compaction looped (bounded 200) until a pass compacts and skips nothing
  33. temporal/trigger.py#L52 - dispatch_ingest_compaction: Temporal when the flag is on and the server reachable, else schedule_on_ingest
  34. temporal/workflows.py#L34 - CompactionWorkflow: durable debounce timer reset by each poke, then the activity under a 15m timeout + bounded retry
  35. ingest_routes.py#L31 - the second dispatch site: a Gate auto-band episode triggers the same pass
  36. docker-compose.yml#L1026 - balance-worker: the Temporal worker as its own process, decoupled from the API's --reload cycle

Live-schema facts (row counts by origin/type/metric, the worked steps chain, the n=30-over-26-days baseline, pending-raw census, FK and Debezium listings) come from PGPASSWORD=… psql -h 10.0.1.2 -U olly -d balance · \d streams.episode, \d streams.rollup, \d streams.baseline, and the streams.event_raw / streams.rollup / streams.baseline queries, 2026-08-21.

Olly Health Insurance Platform

© 2026 Olly