diff --git a/docs/API_V3.md b/docs/API_V3.md
index d7e5110..6f4b57b 100644
--- a/docs/API_V3.md
+++ b/docs/API_V3.md
@@ -174,6 +174,39 @@ The two uploads look symmetric and are not. Three traps, in the order they bite:
 Everything else is forgiving: resends are free, so retry on any transport error rather than trying to
 reason about what got through.
 
+### Our outbox — the local `pending_uploads` table
+
+Upload state is **not** stored on `measurements`. It lives in a separate outbox table holding only
+keys — `pending_uploads(stream_id, time)`, composite FK → `measurements` `ON DELETE CASCADE`. Each
+measurement is stored once, in `measurements`; an outbox row is just "this one hasn't reached the
+server yet". Written by `MeasurementRepository.insertRecorded`, read via `UploadOutbox.pending`,
+cleared via `UploadOutbox.acknowledge` (a `DELETE` — the data row is untouched; `measurements` stays
+insert-only, per `DOMAIN_KNOWLEDGE.md` rule #4). The traps above shape how it may be used:
+
+1. **Only data recorded on this phone is enqueued.** Fixed sessions are uploaded by the AirBeam itself
+   (with `session_token`); downloaded, followed or Explore-saved sessions belong to the server (often
+   another user). Those paths call `insertAll`, which never touches the outbox — so the safe default is
+   "not uploaded". Only the recording path calls `insertRecorded`, which stores **and** enqueues in one
+   transaction, enqueuing only rows actually newly stored (an `OR IGNORE`-skipped duplicate was
+   enqueued when first recorded).
+2. **Acknowledged means "accepted in a `200` batch", not "stored".** See trap 1. So it is only truthful
+   if the uploader **never sends a frame the server will silently drop**:
+   - **Split against `finished_at` before sending** (trap 3). Frames after the session's server-side
+     `finished_at` are never stored — don't send them, don't acknowledge them. They also must not sit in
+     the outbox forever. **Open decision:** bound `pending` by `finished_at`, or add a `state` column to
+     `pending_uploads` (e.g. "unuploadable"). Either way the change stays in the outbox; the
+     measurement itself is never deleted.
+3. **Sync the clock before the first batch** (trap 2, empty body → `X-Server-Time`). On mobile one
+   out-of-range timestamp fails the **whole batch** with `400 invalid_epoch`, and a naive "retry the
+   oldest pending rows" loop then resends the same poisoned batch forever. The uploader must detect
+   `invalid_epoch` and handle the bad frames rather than retry unchanged.
+4. **Track per row `(stream_id, time)`, never a time cursor.** Mini V2 reconnect-sync backfills
+   *older* timestamps after newer ones were already uploaded; a per-session "uploaded up to T" cursor
+   would treat them as done and lose them.
+5. **Retry is otherwise blind and safe.** On transport errors, timeouts and `5xx`, leave the outbox
+   rows in place — they go out again next pass. Resends are deduped server-side on
+   `(stream, timestamp)`. State lives in SQLite, so it survives process death.
+
 ### Error codes
 
 All v3 errors share `{ error_code, message, fields? }` (`fields` only when the request shape is wrong).
diff --git a/docs/DOMAIN_KNOWLEDGE.md b/docs/DOMAIN_KNOWLEDGE.md
index 6b1bc2a..18e8ce0 100644
--- a/docs/DOMAIN_KNOWLEDGE.md
+++ b/docs/DOMAIN_KNOWLEDGE.md
@@ -112,13 +112,13 @@ Enums:
 2. **Configure** — user fills details → `StartRecordingEvent`.
 3. **Start** — orchestrator sets status, inserts session in DB, posts `ConfigureSession`;
    for fixed, first creates the backend session (below). Mobile also schedules averaging +
-   starts the measurement observer.
+   starts the measurement observer. *(Averaging step dropped in the rewrite — §7.)*
 4. **Record** — connector writes config to device; measurements flow back to DB + UI.
 5. **Disconnect** — unexpected BLE drop → `SensorDisconnectedUnexpectedlyEvent`; session →
    `DISCONNECTED`; a 2s-interval reconnect loop starts (no hard cap).
 6. **Resume** — reconnect success → `RECORDING`. User can also disconnect deliberately
    (standalone mode) keeping the session.
-7. **Finish** — `StopRecordingEvent` → final averaging, delete active-session rows, status
+7. **Finish** — `StopRecordingEvent` → final averaging *(dropped in the rewrite — §7)*, delete active-session rows, status
    `FINISHED`, backend sync. All services/connectors self-tear-down.
 
 Orchestration classes (all **CHANGE-LIKELY** at the EventBus edges, logic **KEEP**):
@@ -190,7 +190,8 @@ transports (BLE, WiFi-SoftAP, file I/O) are **PLATFORM**; HTTP is **CHANGE-LIKEL
    download, one-shot followed-session download.
 
 **Dedup differs:** V1 mobile filters `time > lastMeasurementTime`; V2 relies on the DB
-unique index `(session_id, stream_id, time)` + `IGNORE`. Preserve the index.
+unique index `(session_id, stream_id, time)` + `IGNORE`. Preserve the dedup — in the rewrite
+the key is `PRIMARY KEY (stream_id, time)` (see rule #5).
 
 ---
 
@@ -219,7 +220,7 @@ graph/map.
 ### Graph — **MPAndroidChart `v3.1.0`** (PLATFORM rendering)
 Mini card chart + full-screen detail graph with pinch-zoom/pan, colored **target zones**
 (threshold bands), note markers, midnight limit lines. Data shaping (`GraphDataGenerator`,
-`ChartData`, averaging, date formatting, note hit-testing) is **KEEP**; the `LineChart`/
+`ChartData`, averaging *(read-side only — §7)*, date formatting, note hit-testing) is **KEEP**; the `LineChart`/
 `CombinedChart` rendering + gestures are **PLATFORM**.
 
 ### Map — **Google Maps** + **Places** (PLATFORM rendering)
@@ -328,9 +329,36 @@ v37_38 (`tracked_locations`). Missing 29_30 & 32_33 hops rely on destructive fal
 Sessions, Measurements, MeasurementStreams, ActiveSessionMeasurements, Thresholds, Notes.
 `ExpandedCardsRepository` is SharedPreferences-backed (not DB) → multiplatform-settings.
 
-### Averaging — **KEEP** (algorithm), **PLATFORM** (scheduling/DAO)
+### Averaging — ⛔ **DROPPED as storage mutation; display-only in the rewrite**
+
+> **Decision (2026-10-15).** The rewrite **never averages, down-samples, or deletes stored
+> measurements.** Every measurement received from the device (live or sync) or downloaded is
+> kept at native resolution, exactly as recorded. Averaging exists **only on the read side**:
+> the UI (graph, map, heatmap, stats) buckets raw rows into windows when it renders.
+>
+> Why: the legacy in-place averaging was lossy and irreversible, needed scheduling during
+> recording, and caused real data loss (the sparse-interval bug that wiped whole sessions —
+> V2 guide, "Sparse-interval averaging gotcha"). Storage is cheap (1s × 5 streams × 24h ≈
+> 432k rows/day — trivial for SQLite); lost data is not.
+>
+> Consequences for the rewrite:
+> - No `AveragingService`, no averaging scheduler, no `averaging_frequency` column — the
+>   recording engine just saves.
+> - The **native-interval gate (old rule #4) is moot** — there's nothing destructive to guard.
+>   `measurement_interval` is still stored on the session: display bucketing uses it to avoid
+>   choosing a window finer than the native rate, and to tell gaps from sparse sampling.
+> - Read-side averaging must be **cheap on large streams** — bucket in SQL
+>   (`GROUP BY time / window`) or over a time-bounded query, not by loading a 10h session
+>   into memory per frame.
+> - **Uploads send raw measurements.** The backend gets full resolution (the legacy app could
+>   upload already-averaged rows for long sessions). Expect more v3 upload batches
+>   (mobile cap: 3000 frames/batch).
+>
+> The window values below remain a reasonable *display* default; the mutation described
+> after them is legacy behavior, kept here for reference only.
+
 `AveragingWindow`: `ZERO(1s)`, `FIRST(5s)`, `SECOND(60s)`. `TimeThreshold`: FIRST=2h, SECOND=9h.
-Rule: session > 2h → 5s averages; > 9h → 60s averages. Down-samples in place (mutates
+*Legacy:* session > 2h → 5s averages; > 9h → 60s averages. Down-samples in place (mutates
 `measurements` rows, sets `averaging_frequency`, deletes the rest). **Native-interval gate:**
 if `session.measurementInterval >= window.value`, averaging is skipped (prevents wiping
 sparse V2 data — the reason the v37 column exists). See V2 guide §5D.
@@ -401,8 +429,13 @@ Mini V2 mobile. Level banding lives in `Measurement.getLevel`.
    binary → epoch. Getting this wrong offsets every graph. (V2 guide §8a.)
 2. **HTTPS force-upgrade + port 80→443** in base-URL construction.
 3. **V2 sync UUID validation** — discard sync chunks whose device session UUID ≠ app session.
-4. **Averaging native-interval gate** — skip when `measurementInterval >= window`.
-5. **Measurements unique index** `(session_id, stream_id, time)` — the V2 dedup strategy.
+4. **Raw measurements are never averaged, down-sampled, or deleted in storage** — averaging is
+   display-only (§7). *Supersedes the legacy "averaging native-interval gate" rule, which only
+   existed to limit damage from in-place averaging.*
+5. **Measurements dedup on `(stream, time)`** — the V2 dedup strategy, and the backend's own v3
+   rule. Legacy index was `(session_id, stream_id, time)`; the rewrite uses
+   `PRIMARY KEY (stream_id, time)` + `INSERT OR IGNORE` (`session_id` was redundant). There is
+   no separate active-session buffer table — one `measurements` table, one write path.
 6. **Backend URL/port preserved on logout.**
 7. **Two measurement "drain" flags for V2 finish-while-syncing** (V2 guide §9).
 8. **V2→V1 fallback disconnect-path guards** — keep the service/EventBus alive across the
diff --git a/docs/KMP_ARCHITECTURE.md b/docs/KMP_ARCHITECTURE.md
index c29e283..7654c6d 100644
--- a/docs/KMP_ARCHITECTURE.md
+++ b/docs/KMP_ARCHITECTURE.md
@@ -41,7 +41,7 @@
 :core            kotlinx-datetime, coroutines setup, Result types, logging, expect utils
 :domain          pure models + enums + state machines + business rules   (NO platform deps)
                  └ Session/Measurement/Stream/Threshold/Note, Level banding,
-                   averaging algorithm, recording state machine, threshold color levels
+                   display averaging (read-side bucketing), recording state machine, threshold color levels
 :data:db         SQLDelight schema (.sq) + queries + column adapters      (KEEP + PLATFORM driver)
 :data:network    Ktor client + kotlinx.serialization DTOs + API service   (KEEP + PLATFORM engine)
 :data:settings   multiplatform-settings wrapper (auth token, prefs)       (PLATFORM secure storage)
@@ -154,7 +154,6 @@ class RecordingEngine(
     private val driver: SensorDriver,
     private val sessions: SessionRepository,
     private val measurements: MeasurementRepository,
-    private val averaging: AveragingService,
     private val scope: CoroutineScope,
 ) {
     val state: StateFlow<RecordingState>            // NEW→RECORDING→DISCONNECTED→FINISHED
@@ -168,8 +167,10 @@ class RecordingEngine(
   flow through one `saveMeasurements()` path with the dedup rule (`INSERT … IGNORE` on the
   unique index). No more V1-vs-V2 fork.
 - **State machine** is `Session.Status` transitions, pure `commonMain`.
-- **Averaging** (`AveragingService` + windows) ports as-is — pure algorithm; inject a
-  dispatcher instead of `Dispatchers.Default`; keep the native-interval gate.
+- **No averaging in the engine.** The legacy `AveragingService` (in-place down-sampling of
+  stored rows) is **not ported**. Stored measurements stay at native resolution forever;
+  averaging is a read-side, display-only concern (graph/map/stats bucket raw rows when
+  rendering). See `DOMAIN_KNOWLEDGE.md` §7 and load-bearing rule #4.
 - **Background execution** is the one platform escape hatch: recording must survive
   backgrounding. `expect` a `BackgroundSessionHost` — Android = foreground service
   (`FOREGROUND_SERVICE_TYPE_CONNECTED_DEVICE`), iOS = Core Bluetooth background mode +
@@ -183,9 +184,17 @@ class RecordingEngine(
 ## 5. Data Layer
 
 ### Persistence — SQLDelight (`:data:db`)
-- Port the Room schema (v38) to `.sq` files. Preserve: the unique measurements index
-  (`session_id, stream_id, time`), `measurement_interval`, `session_token` + fixed indices,
-  the active-measurements rolling window (>998-row guard → `transaction {}`).
+- **Raw measurements are immutable** — never averaged/down-sampled/deleted by the app
+  (only by session deletion cascade). No `averaging_frequency` column.
+- Port the Room schema (v38) to `.sq` files — as a redesign, not a copy. Preserve the
+  measurements dedup key, now `PRIMARY KEY (stream_id, time)` on `measurements` (the legacy
+  `session_id` in that index was redundant — a stream belongs to one session — and v3 dedups
+  on `(stream, timestamp)` too), plus `measurement_interval` and `session_token`.
+- **No `active_sessions_measurements` table.** The legacy rolling live buffer (10h of rows,
+  >998-row guard) existed because Room couldn't cheaply tail a growing table. The live view is
+  now a windowed query on `measurements` (`ORDER BY time DESC LIMIT ?`, an index walk on the
+  PK), so there is one table and one write path. Batch inserts still go in one
+  `transaction {}`.
 - Column adapters for `Date`→`Instant`, tags list, `Session.Status/Type`, `DeviceType`.
 - `expect` the driver: `AndroidSqliteDriver` / `NativeSqliteDriver`.
 - DAOs → SQLDelight queries returning `Flow` (`.asFlow().mapToList()`). The
@@ -218,7 +227,7 @@ plaintext today. Preserve backend URL/port on logout.
 ### Repositories (`:data:repo`)
 Interfaces in `:domain`, impls here. Extract interfaces for all current repos (only
 `MeasurementsRepository` is one today). Business rules (`getLevel`, threshold find-or-create,
-rolling-window sizing, status transitions) move into `:domain`.
+status transitions) move into `:domain`.
 
 ---
 
@@ -386,7 +395,7 @@ These change the skeleton — resolve early:
 
 Vertical slices, each provable end-to-end, lowest-risk first:
 
-1. **Foundation:** `:core` + `:domain` (models, enums, state machine, averaging, thresholds)
+1. **Foundation:** `:core` + `:domain` (models, enums, state machine, display averaging, thresholds)
    with unit tests. No platform code — proves the shared logic compiles KMP.
 2. **Persistence:** `:data:db` (SQLDelight schema + queries) on both platforms.
 3. **Networking:** `:data:network` (Ktor + DTOs) + **timestamp round-trip tests** + auth/login.
diff --git a/docs/ble_mobile_app_guide.md b/docs/ble_mobile_app_guide.md
index efad1ea..5eaa9fb 100644
--- a/docs/ble_mobile_app_guide.md
+++ b/docs/ble_mobile_app_guide.md
@@ -276,6 +276,10 @@ Defaults `DEFAULT_MOBILE_INTERVAL_SECONDS = 1` / `DEFAULT_FIXED_INTERVAL_SECONDS
 `AirBeamMiniV2Configurator`'s companion object and apply only when the param is null
 (legacy/non-V2 paths).
 
+> **Legacy only.** The KMP rewrite does not average stored measurements at all — averaging
+> is display-only (`DOMAIN_KNOWLEDGE.md` §7), so this bug class cannot occur. Kept as the
+> motivating example for that decision.
+
 **Sparse-interval averaging gotcha (`AveragingService`).** Averaging assumes the native
 sample rate is finer-grained than the averaging window. When that is violated (a
 5-second-native session in the 5s FIRST window, or any session with native ≥ window),
diff --git a/shared/build.gradle.kts b/shared/build.gradle.kts
index 2eb447c..68bb3cc 100644
--- a/shared/build.gradle.kts
+++ b/shared/build.gradle.kts
@@ -85,6 +85,9 @@ kotlin {
       implementation(libs.turbine)
       implementation(libs.multiplatform.settings.test)
     }
+    getByName("androidHostTest").dependencies {
+      implementation(libs.sqldelight.sqlite.driver)
+    }
   }
 }
 
diff --git a/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/MeasurementRepositoryTest.kt b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/MeasurementRepositoryTest.kt
new file mode 100644
index 0000000..89ed219
--- /dev/null
+++ b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/MeasurementRepositoryTest.kt
@@ -0,0 +1,136 @@
+package pl.llp.aircasting.data.local
+
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.test.runTest
+import pl.llp.aircasting.domain.GeoLocation
+import pl.llp.aircasting.domain.session.Measurement
+import pl.llp.aircasting.domain.session.Session
+import pl.llp.aircasting.domain.session.SessionStatus
+import pl.llp.aircasting.domain.session.SessionType
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.Instant
+
+class MeasurementRepositoryTest {
+
+  private val db = testDatabase()
+  private val sessions = LocalSessionRepository(db.sessionQueries)
+  private val streams = LocalMeasurementStreamRepository(db.measurementStreamQueries)
+  private val measurements = LocalMeasurementRepository(db.measurementQueries, db.pendingUploadQueries)
+
+  private val uuid = "22222222-2222-2222-2222-222222222222"
+  private val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000)
+
+  private suspend fun seedSession() = sessions.insert(
+    Session(
+      uuid = uuid,
+      type = SessionType.MOBILE,
+      status = SessionStatus.RECORDING,
+      name = "Test",
+      tags = emptyList(),
+      startTime = t0,
+      timeZone = "UTC",
+      measurementInterval = 1.seconds,
+    )
+  )
+
+  @Test
+  fun `findOrCreate returns the same stream id for the same sensor`() = runTest {
+    seedSession()
+    val first = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    val second = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    assertEquals(first, second)
+    assertEquals(1, streams.findBySession(uuid).size)
+  }
+
+  @Test
+  fun `insertAll ignores a measurement that already exists for stream and time`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+
+    assertEquals(1, measurements.insertAll(listOf(Measurement(stream, t0, value = 12.0))))
+    assertEquals(0, measurements.insertAll(listOf(Measurement(stream, t0, value = 99.0))))
+
+    val stored = measurements.observeRecent(stream, limit = 10).first()
+    assertEquals(1, stored.size)
+    assertEquals(12.0, stored.single().value)
+  }
+
+  @Test
+  fun `insertAll counts only newly stored rows in a partially overlapping batch`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    measurements.insertAll(listOf(Measurement(stream, t0, 1.0)))
+
+    val stored = measurements.insertAll(
+      listOf(
+        Measurement(stream, t0, 1.0),
+        Measurement(stream, t0 + 1.seconds, 2.0),
+        Measurement(stream, t0 + 2.seconds, 3.0),
+      )
+    )
+
+    assertEquals(2, stored)
+  }
+
+  @Test
+  fun `two streams may hold the same timestamp`() = runTest {
+    seedSession()
+    val pm = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    val rh = streams.findOrCreate(uuid, "AirBeam-RH", "%")
+
+    measurements.insertAll(
+      listOf(Measurement(pm, t0, 12.0), Measurement(rh, t0, 45.0))
+    )
+
+    assertEquals(1, measurements.observeRecent(pm, 10).first().size)
+    assertEquals(1, measurements.observeRecent(rh, 10).first().size)
+  }
+
+  @Test
+  fun `a measurement round-trips its location`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    val here = GeoLocation(50.06, 19.94)
+
+    measurements.insertAll(listOf(Measurement(stream, t0, 12.0, location = here)))
+
+    assertEquals(here, measurements.observeRecent(stream, 10).first().single().location)
+  }
+
+  @Test
+  fun `lastTime returns the newest timestamp for the stream`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    measurements.insertAll(
+      listOf(
+        Measurement(stream, t0, 1.0),
+        Measurement(stream, t0 + 2.seconds, 2.0),
+        Measurement(stream, t0 + 1.seconds, 3.0),
+      )
+    )
+    assertEquals(t0 + 2.seconds, measurements.lastTime(stream))
+  }
+
+  @Test
+  fun `lastTime is null for an empty stream`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    assertNull(measurements.lastTime(stream))
+  }
+
+  @Test
+  fun `deleting a session removes its streams and measurements`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    measurements.insertAll(listOf(Measurement(stream, t0, 12.0)))
+
+    sessions.delete(uuid)
+
+    assertTrue(streams.findBySession(uuid).isEmpty())
+    assertTrue(measurements.observeRecent(stream, 10).first().isEmpty())
+  }
+}
\ No newline at end of file
diff --git a/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/SessionRepositoryTest.kt b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/SessionRepositoryTest.kt
new file mode 100644
index 0000000..67d417f
--- /dev/null
+++ b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/SessionRepositoryTest.kt
@@ -0,0 +1,81 @@
+package pl.llp.aircasting.data.local
+
+import app.cash.turbine.test
+import kotlinx.coroutines.test.runTest
+import pl.llp.aircasting.domain.GeoLocation
+import pl.llp.aircasting.domain.session.Session
+import pl.llp.aircasting.domain.session.SessionStatus
+import pl.llp.aircasting.domain.session.SessionType
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.Instant
+
+class SessionRepositoryTest {
+
+  private val db = testDatabase()
+  private val repo = LocalSessionRepository(db.sessionQueries)
+
+  private val session = Session(
+    uuid = "11111111-1111-1111-1111-111111111111",
+    type = SessionType.MOBILE,
+    status = SessionStatus.RECORDING,
+    name = "Morning walk",
+    tags = listOf("bike", "commute"),
+    startTime = Instant.fromEpochMilliseconds(1_700_000_000_000),
+    deviceId = "AA:BB:CC:DD:EE:FF",
+    deviceModel = "AirBeam3",
+    location = GeoLocation(50.06, 19.94),
+    timeZone = "Europe/Warsaw",
+    measurementInterval = 1.seconds,
+  )
+
+  @Test
+  fun `insert then find round-trips every field`() = runTest {
+    repo.insert(session)
+    assertEquals(session, repo.find(session.uuid))
+  }
+
+  @Test
+  fun `find returns null for an unknown uuid`() = runTest {
+    assertNull(repo.find("nope"))
+  }
+
+  @Test
+  fun `a session without a location round-trips as null`() = runTest {
+    val locationless = session.copy(uuid = "no-loc", location = null)
+    repo.insert(locationless)
+    assertNull(repo.find("no-loc")?.location)
+  }
+
+  @Test
+  fun `observeAll re-emits when a session is inserted`() = runTest {
+    repo.observeAll().test {
+      assertEquals(emptyList(), awaitItem())
+      repo.insert(session)
+      assertEquals(listOf(session), awaitItem())
+    }
+  }
+
+  @Test
+  fun `finish sets status and end time together`() = runTest {
+    val end = session.startTime + 60.seconds
+    repo.insert(session)
+    repo.finish(session.uuid, end)
+
+    val stored = repo.find(session.uuid)!!
+    assertEquals(SessionStatus.FINISHED, stored.status)
+    assertEquals(end, stored.endTime)
+  }
+
+  @Test
+  fun `updateStatus leaves end time untouched`() = runTest {
+    repo.insert(session)
+    repo.updateStatus(session.uuid, SessionStatus.DISCONNECTED)
+
+    val stored = repo.find(session.uuid)!!
+    assertEquals(SessionStatus.DISCONNECTED, stored.status)
+    assertNull(stored.endTime)
+  }
+}
\ No newline at end of file
diff --git a/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/TestDatabase.kt b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/TestDatabase.kt
new file mode 100644
index 0000000..e82e776
--- /dev/null
+++ b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/TestDatabase.kt
@@ -0,0 +1,11 @@
+package pl.llp.aircasting.data.local
+
+import app.cash.sqldelight.driver.jdbc.sqlite.JdbcSqliteDriver
+import pl.llp.aircasting.data.local.db.AppDatabase
+
+internal fun testDatabase(): AppDatabase {
+  val driver = JdbcSqliteDriver(JdbcSqliteDriver.IN_MEMORY)
+  AppDatabase.Schema.create(driver)
+  driver.execute(null, "PRAGMA foreign_keys=ON;", 0)
+  return createDatabase { driver }
+}
\ No newline at end of file
diff --git a/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/UploadOutboxTest.kt b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/UploadOutboxTest.kt
new file mode 100644
index 0000000..5b16a85
--- /dev/null
+++ b/shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/UploadOutboxTest.kt
@@ -0,0 +1,157 @@
+package pl.llp.aircasting.data.local
+
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.test.runTest
+import pl.llp.aircasting.domain.session.Measurement
+import pl.llp.aircasting.domain.session.Session
+import pl.llp.aircasting.domain.session.SessionStatus
+import pl.llp.aircasting.domain.session.SessionType
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFails
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.Instant
+
+class UploadOutboxTest {
+
+  private val db = testDatabase()
+  private val sessions = LocalSessionRepository(db.sessionQueries)
+  private val streams = LocalMeasurementStreamRepository(db.measurementStreamQueries)
+  private val measurements = LocalMeasurementRepository(db.measurementQueries, db.pendingUploadQueries)
+  private val outbox = LocalUploadOutbox(db.pendingUploadQueries)
+
+  private val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000)
+
+  private suspend fun seedStream(uuid: String = "33333333-3333-3333-3333-333333333333"): Long {
+    sessions.insert(
+      Session(
+        uuid = uuid,
+        type = SessionType.MOBILE,
+        status = SessionStatus.RECORDING,
+        name = "Test",
+        tags = emptyList(),
+        startTime = t0,
+        timeZone = "UTC",
+        measurementInterval = 1.seconds,
+      )
+    )
+    return streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+  }
+
+  /** The recording path: one call, the row is both stored (for the UI) and queued (for upload). */
+  @Test
+  fun `insertRecorded stores the measurement and enqueues it`() = runTest {
+    val stream = seedStream()
+    val m = Measurement(stream, t0, 12.0)
+
+    assertEquals(1, measurements.insertRecorded(listOf(m)))
+
+    assertEquals(listOf(m), measurements.observeRecent(stream, 10).first())
+    assertEquals(listOf(m), outbox.pending("33333333-3333-3333-3333-333333333333", 10))
+  }
+
+  /**
+   * The safety property the whole design exists for: downloaded / explored / fixed data
+   * goes through insertAll and must never become upload work.
+   */
+  @Test
+  fun `insertAll stores but never enqueues`() = runTest {
+    val stream = seedStream()
+
+    measurements.insertAll(listOf(Measurement(stream, t0, 12.0)))
+
+    assertTrue(outbox.pending("33333333-3333-3333-3333-333333333333", 10).isEmpty())
+  }
+
+  /**
+   * A duplicate skipped by OR IGNORE was enqueued when first recorded. Re-enqueuing it
+   * after an ack would resend a measurement the server already has.
+   */
+  @Test
+  fun `insertRecorded does not re-enqueue a duplicate that was already acknowledged`() = runTest {
+    val stream = seedStream()
+    val m = Measurement(stream, t0, 12.0)
+    measurements.insertRecorded(listOf(m))
+    outbox.acknowledge(listOf(m))
+
+    assertEquals(0, measurements.insertRecorded(listOf(m)))
+    assertTrue(outbox.pending("33333333-3333-3333-3333-333333333333", 10).isEmpty())
+  }
+
+  /** Oldest first, regardless of arrival order — backfill from a BLE sync arrives late. */
+  @Test
+  fun `pending yields oldest first and respects the limit`() = runTest {
+    val stream = seedStream()
+    measurements.insertRecorded(
+      listOf(
+        Measurement(stream, t0 + 2.seconds, 3.0),
+        Measurement(stream, t0, 1.0),
+        Measurement(stream, t0 + 1.seconds, 2.0),
+      )
+    )
+
+    val pending = outbox.pending("33333333-3333-3333-3333-333333333333", limit = 2)
+
+    assertEquals(listOf(t0, t0 + 1.seconds), pending.map { it.time })
+  }
+
+  /** Each upload is per session uuid (`POST /mobile_sessions/{uuid}/measurements`). */
+  @Test
+  fun `pending is scoped to one session`() = runTest {
+    val a = seedStream("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
+    val b = seedStream("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")
+    measurements.insertRecorded(listOf(Measurement(a, t0, 1.0), Measurement(b, t0, 2.0)))
+
+    val pending = outbox.pending("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", 10)
+
+    assertEquals(listOf(a), pending.map { it.streamId })
+  }
+
+  /** Ack is a DELETE on the outbox only. The data row is never touched. */
+  @Test
+  fun `acknowledge empties the outbox but keeps the measurement`() = runTest {
+    val stream = seedStream()
+    val m = Measurement(stream, t0, 12.0)
+    measurements.insertRecorded(listOf(m))
+
+    outbox.acknowledge(listOf(m))
+
+    assertTrue(outbox.pending("33333333-3333-3333-3333-333333333333", 10).isEmpty())
+    assertEquals(listOf(m), measurements.observeRecent(stream, 10).first())
+  }
+
+  /**
+   * Atomicity: the second row points at a stream that doesn't exist, so its FK fails.
+   * The whole transaction must roll back — including the first row's measurement AND its
+   * outbox entry. No half-stored batch, no orphan outbox row.
+   */
+  @Test
+  fun `a failing batch stores and enqueues nothing`() = runTest {
+    val stream = seedStream()
+
+    assertFails {
+      measurements.insertRecorded(
+        listOf(Measurement(stream, t0, 1.0), Measurement(streamId = 999, time = t0, value = 2.0))
+      )
+    }
+
+    assertTrue(measurements.observeRecent(stream, 10).first().isEmpty())
+    assertTrue(outbox.pending("33333333-3333-3333-3333-333333333333", 10).isEmpty())
+  }
+
+  /**
+   * Proves the composite FK's ON DELETE CASCADE. Without it, deleting the session would
+   * cascade into `measurements`, hit an outbox row still referencing one, and *throw* —
+   * foreign keys are enforced. So "delete succeeds" is itself the assertion.
+   */
+  @Test
+  fun `deleting a session with pending uploads succeeds and clears them`() = runTest {
+    val stream = seedStream()
+    measurements.insertRecorded(listOf(Measurement(stream, t0, 12.0)))
+
+    sessions.delete("33333333-3333-3333-3333-333333333333")
+
+    assertTrue(outbox.pending("33333333-3333-3333-3333-333333333333", 10).isEmpty())
+  }
+}
diff --git a/shared/src/androidMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.android.kt b/shared/src/androidMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.android.kt
index 6ff7b4d..e3adc9b 100644
--- a/shared/src/androidMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.android.kt
+++ b/shared/src/androidMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.android.kt
@@ -1,16 +1,23 @@
 package pl.llp.aircasting.data.local
 
 import android.content.Context
+import androidx.sqlite.db.SupportSQLiteDatabase
 import app.cash.sqldelight.db.SqlDriver
 import app.cash.sqldelight.driver.android.AndroidSqliteDriver
 import pl.llp.aircasting.data.local.db.AppDatabase
 
-class AndroidDatabaseDriverFactory(private val context: Context): DatabaseDriverFactory {
+class AndroidDatabaseDriverFactory(private val context: Context) : DatabaseDriverFactory {
   override fun createDriver(): SqlDriver {
     return AndroidSqliteDriver(
       schema = AppDatabase.Schema,
       context = context,
-      name = "aircasting.db"
+      name = "aircasting.db",
+      // SQLite disables foreign keys per connection; without this ON DELETE CASCADE is a no-op.
+      callback = object : AndroidSqliteDriver.Callback(AppDatabase.Schema) {
+        override fun onOpen(db: SupportSQLiteDatabase) {
+          db.setForeignKeyConstraintsEnabled(true)
+        }
+      },
     )
   }
 }
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/ColumnAdapters.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/ColumnAdapters.kt
new file mode 100644
index 0000000..3f4cefa
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/ColumnAdapters.kt
@@ -0,0 +1,16 @@
+package pl.llp.aircasting.data.local
+
+import app.cash.sqldelight.ColumnAdapter
+import kotlin.time.Instant
+
+internal val instantAdapter = object : ColumnAdapter<Instant, Long> {
+  override fun decode(databaseValue: Long): Instant = Instant.fromEpochMilliseconds(databaseValue)
+  override fun encode(value: Instant): Long = value.toEpochMilliseconds()
+}
+
+internal val stringListAdapter = object : ColumnAdapter<List<String>, String> {
+  override fun decode(databaseValue: String): List<String> =
+    if (databaseValue.isEmpty()) emptyList() else databaseValue.split(",")
+
+  override fun encode(value: List<String>): String = value.joinToString(",")
+}
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/DatabaseFactory.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/DatabaseFactory.kt
index 6d2831e..2fd1480 100644
--- a/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/DatabaseFactory.kt
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/DatabaseFactory.kt
@@ -1,5 +1,25 @@
 package pl.llp.aircasting.data.local
 
+import app.cash.sqldelight.EnumColumnAdapter
 import pl.llp.aircasting.data.local.db.AppDatabase
+import pl.llp.aircasting.data.local.db.Measurements
+import pl.llp.aircasting.data.local.db.Pending_uploads
+import pl.llp.aircasting.data.local.db.Sessions
 
-fun createDatabase(driverFactory: DatabaseDriverFactory) = AppDatabase(driverFactory.createDriver())
\ No newline at end of file
+
+fun createDatabase(driverFactory: DatabaseDriverFactory) = AppDatabase(
+  driver = driverFactory.createDriver(),
+  sessionsAdapter = Sessions.Adapter(
+    typeAdapter = EnumColumnAdapter(),
+    statusAdapter = EnumColumnAdapter(),
+    tagsAdapter = stringListAdapter,
+    start_timeAdapter = instantAdapter,
+    end_timeAdapter = instantAdapter,
+  ),
+  measurementsAdapter = Measurements.Adapter(
+    timeAdapter = instantAdapter,
+  ),
+  pending_uploadsAdapter = Pending_uploads.Adapter(
+    timeAdapter = instantAdapter,
+  ),
+)
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalMeasurementRepository.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalMeasurementRepository.kt
new file mode 100644
index 0000000..3208082
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalMeasurementRepository.kt
@@ -0,0 +1,64 @@
+package pl.llp.aircasting.data.local
+
+import app.cash.sqldelight.coroutines.asFlow
+import app.cash.sqldelight.coroutines.mapToList
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.IO
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.withContext
+import pl.llp.aircasting.data.local.db.MeasurementQueries
+import pl.llp.aircasting.data.local.db.Measurements
+import pl.llp.aircasting.data.local.db.PendingUploadQueries
+import pl.llp.aircasting.domain.GeoLocation
+import pl.llp.aircasting.domain.session.Measurement
+import pl.llp.aircasting.domain.session.MeasurementRepository
+import kotlin.time.Instant
+
+class LocalMeasurementRepository(
+  private val queries: MeasurementQueries,
+  private val outboxQueries: PendingUploadQueries,
+  private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
+) : MeasurementRepository {
+
+  override suspend fun insertAll(measurements: List<Measurement>): Int =
+    insert(measurements, enqueue = false)
+
+  override suspend fun insertRecorded(measurements: List<Measurement>): Int =
+    insert(measurements, enqueue = true)
+
+  private suspend fun insert(measurements: List<Measurement>, enqueue: Boolean): Int =
+    withContext(ioDispatcher) {
+      queries.transactionWithResult {
+        measurements.count { m ->
+          val stored = queries.insertMeasurement(
+            stream_id = m.streamId,
+            time = m.time,
+            value_ = m.value,
+            latitude = m.location?.latitude,
+            longitude = m.location?.longitude,
+          ).value == 1L
+          if (stored && enqueue) outboxQueries.enqueue(m.streamId, m.time)
+          stored
+        }
+      }
+    }
+
+  override fun observeRecent(streamId: Long, limit: Int): Flow<List<Measurement>> =
+    queries.selectRecentByStream(streamId, limit.toLong())
+      .asFlow()
+      .mapToList(ioDispatcher)
+      .map { rows -> rows.map { it.toDomain() } }
+
+  override suspend fun lastTime(streamId: Long): Instant? = withContext(ioDispatcher) {
+    queries.selectLastTimeByStream(streamId).executeAsOneOrNull()
+  }
+}
+
+internal fun Measurements.toDomain() = Measurement(
+  streamId = stream_id,
+  time = time,
+  value = value_,
+  location = latitude?.let { lat -> longitude?.let { lon -> GeoLocation(lat, lon) } },
+)
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalMeasurementStreamRepository.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalMeasurementStreamRepository.kt
new file mode 100644
index 0000000..32bdf38
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalMeasurementStreamRepository.kt
@@ -0,0 +1,59 @@
+package pl.llp.aircasting.data.local
+
+import app.cash.sqldelight.coroutines.asFlow
+import app.cash.sqldelight.coroutines.mapToList
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.IO
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.withContext
+import pl.llp.aircasting.data.local.db.MeasurementStreamQueries
+import pl.llp.aircasting.data.local.db.Measurement_streams
+import pl.llp.aircasting.domain.session.MeasurementStream
+import pl.llp.aircasting.domain.session.MeasurementStreamRepository
+
+class LocalMeasurementStreamRepository(
+  private val queries: MeasurementStreamQueries,
+  private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
+) : MeasurementStreamRepository {
+
+  override suspend fun findOrCreate(
+    sessionUuid: String,
+    sensorName: String,
+    unitSymbol: String,
+    sensorTypeId: Int?,
+  ): Long = withContext(ioDispatcher) {
+    queries.transactionWithResult {
+      queries.insertStreamIfAbsent(
+        session_uuid = sessionUuid,
+        sensor_name = sensorName,
+        unit_symbol = unitSymbol,
+        sensor_type_id = sensorTypeId?.toLong(),
+      )
+      queries.selectStream(sessionUuid, sensorName).executeAsOne().id
+    }
+  }
+
+  override suspend fun findBySession(sessionUuid: String): List<MeasurementStream> =
+    withContext(ioDispatcher) {
+      queries.selectStreamsBySession(sessionUuid).executeAsList().map { it.toDomain() }
+    }
+
+  override fun observeBySession(sessionUuid: String): Flow<List<MeasurementStream>> =
+    queries.selectStreamsBySession(sessionUuid)
+      .asFlow()
+      .mapToList(ioDispatcher)
+      .map { rows -> rows.map { it.toDomain() } }
+
+  override suspend fun setSensorTypeId(streamId: Long, sensorTypeId: Int): Unit =
+    withContext(ioDispatcher) { queries.updateSensorTypeId(sensorTypeId.toLong(), streamId) }
+}
+
+private fun Measurement_streams.toDomain() = MeasurementStream(
+  id = id,
+  sessionUuid = session_uuid,
+  sensorName = sensor_name,
+  unitSymbol = unit_symbol,
+  sensorTypeId = sensor_type_id?.toInt(),
+)
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalSessionRepository.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalSessionRepository.kt
new file mode 100644
index 0000000..16ad6e4
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalSessionRepository.kt
@@ -0,0 +1,88 @@
+package pl.llp.aircasting.data.local
+
+import app.cash.sqldelight.coroutines.asFlow
+import app.cash.sqldelight.coroutines.mapToList
+import app.cash.sqldelight.coroutines.mapToOneOrNull
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.IO
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.withContext
+import pl.llp.aircasting.data.local.db.SessionQueries
+import pl.llp.aircasting.data.local.db.Sessions
+import pl.llp.aircasting.domain.GeoLocation
+import pl.llp.aircasting.domain.session.Session
+import pl.llp.aircasting.domain.session.SessionRepository
+import pl.llp.aircasting.domain.session.SessionStatus
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Instant
+
+class LocalSessionRepository(
+  private val queries: SessionQueries,
+  private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
+) : SessionRepository {
+
+  override suspend fun insert(session: Session): Unit = withContext(ioDispatcher) {
+    queries.insertSession(
+      uuid = session.uuid,
+      type = session.type,
+      status = session.status,
+      name = session.name,
+      tags = session.tags,
+      device_id = session.deviceId,
+      device_model = session.deviceModel,
+      start_time = session.startTime,
+      end_time = session.endTime,
+      latitude = session.location?.latitude,
+      longitude = session.location?.longitude,
+      is_indoor = session.isIndoor,
+      contribute = session.contribute,
+      time_zone = session.timeZone,
+      measurement_interval_millis = session.measurementInterval.inWholeMilliseconds,
+      session_token = session.sessionToken,
+    )
+  }
+
+  override suspend fun find(uuid: String): Session? = withContext(ioDispatcher) {
+    queries.selectSessionByUuid(uuid).executeAsOneOrNull()?.toDomain()
+  }
+  override fun observe(uuid: String): Flow<Session?> =
+    queries.selectSessionByUuid(uuid)
+      .asFlow()
+      .mapToOneOrNull(ioDispatcher)
+      .map { it?.toDomain() }
+
+  override fun observeAll(): Flow<List<Session>> =
+    queries.selectAllSessions()
+      .asFlow()
+      .mapToList(ioDispatcher)
+      .map { rows -> rows.map { it.toDomain() } }
+
+  override suspend fun updateStatus(uuid: String, status: SessionStatus): Unit =
+    withContext(ioDispatcher) { queries.updateSessionStatus(status, uuid) }
+
+  override suspend fun finish(uuid: String, endTime: Instant): Unit = withContext(ioDispatcher) {
+    queries.finishSession(status = SessionStatus.FINISHED, endTime = endTime, uuid = uuid)
+  }
+  override suspend fun delete(uuid: String): Unit =
+    withContext(ioDispatcher) { queries.deleteSession(uuid) }
+}
+
+private fun Sessions.toDomain() = Session(
+  uuid = uuid,
+  type = type,
+  status = status,
+  name = name,
+  tags = tags,
+  startTime = start_time,
+  timeZone = time_zone,
+  measurementInterval = measurement_interval_millis.milliseconds,
+  endTime = end_time,
+  deviceId = device_id,
+  deviceModel = device_model,
+  location = latitude?.let { lat -> longitude?.let { lon -> GeoLocation(lat, lon) } },
+  isIndoor = is_indoor,
+  contribute = contribute,
+  sessionToken = session_token,
+)
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalUploadOutbox.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalUploadOutbox.kt
new file mode 100644
index 0000000..d7fa921
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/data/local/LocalUploadOutbox.kt
@@ -0,0 +1,31 @@
+package pl.llp.aircasting.data.local
+
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.IO
+import kotlinx.coroutines.withContext
+import pl.llp.aircasting.data.local.db.PendingUploadQueries
+import pl.llp.aircasting.domain.session.Measurement
+import pl.llp.aircasting.domain.session.UploadOutbox
+
+class LocalUploadOutbox(
+  private val queries: PendingUploadQueries,
+  private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
+) : UploadOutbox {
+
+  override suspend fun pending(sessionUuid: String, limit: Int): List<Measurement> =
+    withContext(ioDispatcher) {
+      queries.selectPending(sessionUuid, limit.toLong()).executeAsList().map { it.toDomain() }
+    }
+
+  /**
+   * Deletes by natural key, exactly the rows that were sent — correct even when a BLE sync
+   * backfilled older timestamps after newer ones went out.
+   */
+  override suspend fun acknowledge(measurements: List<Measurement>): Unit =
+    withContext(ioDispatcher) {
+      queries.transaction {
+        measurements.forEach { queries.acknowledge(it.streamId, it.time) }
+      }
+    }
+}
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/di/DatabaseModule.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/di/DatabaseModule.kt
index 1dc4716..7a61f98 100644
--- a/shared/src/commonMain/kotlin/pl/llp/aircasting/di/DatabaseModule.kt
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/di/DatabaseModule.kt
@@ -1,13 +1,31 @@
 package pl.llp.aircasting.di
 
 import org.koin.dsl.module
+import pl.llp.aircasting.data.local.LocalMeasurementRepository
+import pl.llp.aircasting.data.local.LocalMeasurementStreamRepository
+import pl.llp.aircasting.data.local.LocalSessionRepository
+import pl.llp.aircasting.data.local.LocalUploadOutbox
 import pl.llp.aircasting.data.local.createDatabase
 import pl.llp.aircasting.data.local.db.AppDatabase
 import pl.llp.aircasting.domain.LocalThresholdsRepository
 import pl.llp.aircasting.domain.ThresholdsRepository
+import pl.llp.aircasting.domain.session.MeasurementRepository
+import pl.llp.aircasting.domain.session.MeasurementStreamRepository
+import pl.llp.aircasting.domain.session.SessionRepository
+import pl.llp.aircasting.domain.session.UploadOutbox
 
 val databaseModule = module {
   single { createDatabase(get()) }
+  single { get<AppDatabase>().pendingUploadQueries }
   single { get<AppDatabase>().sensorThresholdQueries }
+  single { get<AppDatabase>().sessionQueries }
+  single { get<AppDatabase>().measurementStreamQueries }
+  single { get<AppDatabase>().measurementQueries }
+
   single<ThresholdsRepository> { LocalThresholdsRepository(get()) }
+  single<SessionRepository> { LocalSessionRepository(get()) }
+  single<MeasurementStreamRepository> { LocalMeasurementStreamRepository(get()) }
+  single<MeasurementRepository> { LocalMeasurementRepository(get(), get()) }
+  single<UploadOutbox> { LocalUploadOutbox(get()) }
+
 }
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/Measurement.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/Measurement.kt
new file mode 100644
index 0000000..b05928b
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/Measurement.kt
@@ -0,0 +1,12 @@
+package pl.llp.aircasting.domain.session
+
+import pl.llp.aircasting.domain.GeoLocation
+import kotlin.time.Instant
+
+data class Measurement(
+  val streamId: Long,
+  val time: Instant,
+  val value: Double,
+  /** Mobile only — a fixed session's position lives on the Session. */
+  val location: GeoLocation? = null,
+)
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/MeasurementStream.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/MeasurementStream.kt
new file mode 100644
index 0000000..48b51b1
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/MeasurementStream.kt
@@ -0,0 +1,10 @@
+package pl.llp.aircasting.domain.session
+
+data class MeasurementStream(
+  val id: Long,
+  val sessionUuid: String,
+  val sensorName: String,
+  val unitSymbol: String,
+  /** Assigned by the backend at session create; goes into each binary upload frame. */
+  val sensorTypeId: Int? = null,
+)
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/Session.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/Session.kt
new file mode 100644
index 0000000..d92cd91
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/Session.kt
@@ -0,0 +1,31 @@
+package pl.llp.aircasting.domain.session
+
+import pl.llp.aircasting.domain.GeoLocation
+import kotlin.time.Duration
+import kotlin.time.Instant
+
+enum class SessionType { MOBILE, FIXED }
+
+enum class SessionStatus { NEW, RECORDING, DISCONNECTED, FINISHED }
+data class Session(
+  val uuid: String,
+  val type: SessionType,
+  val status: SessionStatus,
+  val name: String,
+  val tags: List<String>,
+  val startTime: Instant,
+  /** IANA zone id. v3 create requires it, and it's mandatory for indoor sessions. */
+  val timeZone: String,
+  // TODO: We won't have averaging
+  /** Rule #4: averaging is skipped when this is >= the averaging window. */
+  val measurementInterval: Duration,
+  val endTime: Instant? = null,
+  val deviceId: String? = null,
+  val deviceModel: String? = null,
+  /** Per-session fix for fixed sessions; mobile carries location per measurement instead. */
+  val location: GeoLocation? = null,
+  val isIndoor: Boolean = false,
+  val contribute: Boolean = true,
+  /** v3 fixed only: the Bearer token the AirBeam itself uses to upload. Not the user token. */
+  val sessionToken: String? = null,
+)
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/SessionRepositories.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/SessionRepositories.kt
new file mode 100644
index 0000000..b6c3c6b
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/SessionRepositories.kt
@@ -0,0 +1,46 @@
+package pl.llp.aircasting.domain.session
+
+import kotlinx.coroutines.flow.Flow
+import kotlin.time.Instant
+
+interface SessionRepository {
+  suspend fun insert(session: Session)
+  suspend fun find(uuid: String): Session?
+  fun observe(uuid: String): Flow<Session?>
+  fun observeAll(): Flow<List<Session>>
+  suspend fun updateStatus(uuid: String, status: SessionStatus)
+  suspend fun finish(uuid: String, endTime: Instant)
+  suspend fun delete(uuid: String)
+}
+
+interface MeasurementStreamRepository {
+  suspend fun findOrCreate(
+    sessionUuid: String,
+    sensorName: String,
+    unitSymbol: String,
+    sensorTypeId: Int? = null,
+  ): Long
+
+  suspend fun findBySession(sessionUuid: String): List<MeasurementStream>
+  fun observeBySession(sessionUuid: String): Flow<List<MeasurementStream>>
+  suspend fun setSensorTypeId(streamId: Long, sensorTypeId: Int)
+}
+
+interface MeasurementRepository {
+  /**
+   * Data that originated elsewhere — downloaded, followed, explored, fixed. Stored only;
+   * never becomes upload work.
+   * @return how many rows were newly stored; the rest already existed for (stream, time).
+   */
+  suspend fun insertAll(measurements: List<Measurement>): Int
+  /**
+   * Data recorded on this phone. Stores AND enqueues for upload, in one transaction.
+   * Only newly stored rows are enqueued — an OR IGNORE-skipped duplicate was enqueued
+   * when it was first recorded.
+   * @return how many rows were newly stored.
+   */
+  suspend fun insertRecorded(measurements: List<Measurement>): Int
+  /** Newest-first window. Replaces the old `active_sessions_measurements` table. */
+  fun observeRecent(streamId: Long, limit: Int): Flow<List<Measurement>>
+  suspend fun lastTime(streamId: Long): Instant?
+}
\ No newline at end of file
diff --git a/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/UploadOutbox.kt b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/UploadOutbox.kt
new file mode 100644
index 0000000..c48cb52
--- /dev/null
+++ b/shared/src/commonMain/kotlin/pl/llp/aircasting/domain/session/UploadOutbox.kt
@@ -0,0 +1,12 @@
+package pl.llp.aircasting.domain.session
+interface UploadOutbox {
+  /** Oldest first, for one session — matching `POST /mobile_sessions/{uuid}/measurements`. */
+  suspend fun pending(sessionUuid: String, limit: Int): List<Measurement>
+
+  /**
+   * The server accepted these in a `200` batch. Removes them from the outbox; the stored
+   * measurements are untouched. A `200` is not proof of storage — the uploader must split
+   * out frames past `finished_at` before sending.
+   */
+  suspend fun acknowledge(measurements: List<Measurement>)
+}
\ No newline at end of file
diff --git a/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/Measurement.sq b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/Measurement.sq
new file mode 100644
index 0000000..29b9294
--- /dev/null
+++ b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/Measurement.sq
@@ -0,0 +1,28 @@
+import kotlin.time.Instant;
+
+CREATE TABLE measurements (
+    stream_id INTEGER NOT NULL REFERENCES measurement_streams(id) ON DELETE CASCADE,
+    time INTEGER AS Instant NOT NULL,
+    value REAL NOT NULL,
+    latitude REAL,
+    longitude REAL,
+    PRIMARY KEY (stream_id, time)
+);
+
+insertMeasurement:
+INSERT OR IGNORE INTO measurements(stream_id, time, value, latitude, longitude)
+VALUES (?, ?, ?, ?, ?);
+
+selectRecentByStream:
+SELECT *
+FROM measurements
+WHERE stream_id = :streamId
+ORDER BY time DESC
+LIMIT :limit;
+
+selectLastTimeByStream:
+SELECT time
+FROM measurements
+WHERE stream_id = :streamId
+ORDER BY time DESC
+LIMIT 1;
\ No newline at end of file
diff --git a/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/MeasurementStream.sq b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/MeasurementStream.sq
new file mode 100644
index 0000000..cd7413b
--- /dev/null
+++ b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/MeasurementStream.sq
@@ -0,0 +1,29 @@
+CREATE TABLE measurement_streams (
+    id INTEGER NOT NULL PRIMARY KEY,
+    session_uuid TEXT NOT NULL REFERENCES sessions(uuid) ON DELETE CASCADE,
+    sensor_name TEXT NOT NULL,
+    unit_symbol TEXT NOT NULL,
+    sensor_type_id INTEGER
+);
+CREATE UNIQUE INDEX measurement_streams_session_sensor
+ON measurement_streams(session_uuid, sensor_name);
+
+insertStreamIfAbsent:
+INSERT OR IGNORE INTO measurement_streams(session_uuid, sensor_name, unit_symbol, sensor_type_id)
+VALUES (?, ?, ?, ?);
+
+selectStream:
+SELECT *
+FROM measurement_streams
+WHERE session_uuid = :sessionUuid AND sensor_name = :sensorName;
+
+selectStreamsBySession:
+SELECT *
+FROM measurement_streams
+WHERE session_uuid = :sessionUuid
+ORDER BY id;
+
+updateSensorTypeId:
+UPDATE measurement_streams
+SET sensor_type_id = :sensorTypeId
+WHERE id = :id;
\ No newline at end of file
diff --git a/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/PendingUpload.sq b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/PendingUpload.sq
new file mode 100644
index 0000000..aa529ee
--- /dev/null
+++ b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/PendingUpload.sq
@@ -0,0 +1,26 @@
+import kotlin.time.Instant;
+CREATE TABLE pending_uploads (
+    stream_id INTEGER NOT NULL,
+    time INTEGER AS Instant NOT NULL,
+    PRIMARY KEY (stream_id, time),
+    FOREIGN KEY (stream_id, time) REFERENCES measurements(stream_id, time) ON DELETE CASCADE
+);
+
+enqueue:
+INSERT OR IGNORE INTO pending_uploads(stream_id, time)
+VALUES (?, ?);
+
+selectPending:
+SELECT measurements.*
+FROM pending_uploads
+JOIN measurements
+ON measurements.stream_id = pending_uploads.stream_id
+ AND measurements.time = pending_uploads.time
+JOIN measurement_streams ON measurement_streams.id = pending_uploads.stream_id
+WHERE measurement_streams.session_uuid = :sessionUuid
+ORDER BY pending_uploads.time
+LIMIT :limit;
+
+acknowledge:
+DELETE FROM pending_uploads
+WHERE stream_id = :streamId AND time = :time;
\ No newline at end of file
diff --git a/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/Session.sq b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/Session.sq
new file mode 100644
index 0000000..ad3ce73
--- /dev/null
+++ b/shared/src/commonMain/sqldelight/pl/llp/aircasting/data/local/db/Session.sq
@@ -0,0 +1,56 @@
+import kotlin.Boolean;
+import kotlin.String;
+import kotlin.collections.List;
+import kotlin.time.Instant;
+import pl.llp.aircasting.domain.session.SessionStatus;
+import pl.llp.aircasting.domain.session.SessionType;
+
+CREATE TABLE sessions (
+    uuid TEXT NOT NULL PRIMARY KEY,
+    type TEXT AS SessionType NOT NULL,
+    status TEXT AS SessionStatus NOT NULL,
+    name TEXT NOT NULL,
+    tags TEXT AS List<String> NOT NULL,
+    device_id TEXT,
+    device_model TEXT,
+    start_time INTEGER AS Instant NOT NULL,
+    end_time INTEGER AS Instant,
+    latitude REAL,
+    longitude REAL,
+    is_indoor INTEGER AS Boolean NOT NULL DEFAULT 0,
+    contribute INTEGER AS Boolean NOT NULL DEFAULT 1,
+    time_zone TEXT NOT NULL,
+    measurement_interval_millis INTEGER NOT NULL,
+    session_token TEXT
+);
+
+insertSession:
+INSERT INTO sessions(
+    uuid, type, status, name, tags, device_id, device_model,
+    start_time, end_time, latitude, longitude,
+    is_indoor, contribute, time_zone, measurement_interval_millis, session_token
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
+
+selectSessionByUuid:
+SELECT *
+FROM sessions
+WHERE uuid = :uuid;
+
+selectAllSessions:
+SELECT *
+FROM sessions
+ORDER BY start_time DESC;
+
+updateSessionStatus:
+UPDATE sessions
+SET status = :status
+WHERE uuid = :uuid;
+
+finishSession:
+UPDATE sessions
+SET status = :status, end_time = :endTime
+WHERE uuid = :uuid;
+
+deleteSession:
+DELETE FROM sessions
+WHERE uuid = :uuid;
\ No newline at end of file
diff --git a/shared/src/iosMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.ios.kt b/shared/src/iosMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.ios.kt
index 3fc03b8..7c5e0d4 100644
--- a/shared/src/iosMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.ios.kt
+++ b/shared/src/iosMain/kotlin/pl/llp/aircasting/data/local/DatabaseDriverFactory.ios.kt
@@ -4,11 +4,16 @@ import app.cash.sqldelight.db.SqlDriver
 import app.cash.sqldelight.driver.native.NativeSqliteDriver
 import pl.llp.aircasting.data.local.db.AppDatabase
 
-class IosDatabaseDriverFactory: DatabaseDriverFactory {
+class IosDatabaseDriverFactory : DatabaseDriverFactory {
   override fun createDriver(): SqlDriver {
     return NativeSqliteDriver(
       schema = AppDatabase.Schema,
-      name = "aircasting.db" //todo: set as a const
+      name = "aircasting.db", //todo: set as a const
+      onConfiguration = { config ->
+        config.copy(
+          extendedConfig = config.extendedConfig.copy(foreignKeyConstraints = true),
+        )
+      },
     )
   }
 }
\ No newline at end of file