Read flushed conversation content with the key when uploading (0.2.13)

CHANGELOG.md

@@ -11,7 +11,21 @@ RFC; a release that bumps one says so here.
 
 ## [Unreleased]
 
-## [0.2.12] — 2026-09-09
+## [0.2.13] — 2026-09-09
+
+- The uploader reads flushed content back with the database key. Under
+  `messages` (and `full`) an event whose content the daemon's periodic
+  flush had already moved into an encrypted blob was uploaded as bare
+  metadata: the sync path opened the database without a key provider and
+  the blob reader yielded nothing, silently. Events uploaded within a few
+  seconds of capture were never affected; anything held back by an
+  outage, a paused daemon or a large backlog was. Now a blob that cannot
+  be read holds the upload with a clear error (restore the key, or switch
+  the profile to `semantic`) instead of sending stripped events. Rows
+  already uploaded without their text stay that way — the server does not
+  backfill content for a duplicate event id.
+- Under `messages` only the kinds that can carry something said open their
+  blobs; the inference recomputation never opens one.
 
 - Export the conversation over OTel by default: `attempt hook install`
   sets `OTEL_LOG_USER_PROMPTS=1` and `OTEL_LOG_ASSISTANT_RESPONSES=1` for

Cargo.toml

@@ -16,7 +16,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.2.12"
+version = "0.2.13"
 edition = "2024"
 rust-version = "1.94"
 license = "Apache-2.0"

PROGRESS.md

@@ -2,6 +2,31 @@
 
 Execution log for `TODO.md`. Newest session first. Read this before working.
 
+## 2026-09-09 — flushed conversation was uploaded without its text (0.2.13)
+
+Found while verifying the release on the owner's tenant. During the OOM
+outage the daemon's periodic flush moved the 05:59–06:04 UTC events into a
+segment (content into encrypted blobs) before their first successful upload
+at 06:14:50. Those rows reached the server as bare metadata: the local
+`attempt query` still shows their text and `content_ref`, the server shows
+`content_json` null for exactly that window, and every row uploaded within
+seconds of capture on either side has its text.
+
+Cause: `sync::open_read_only` opened the database with no key provider, so
+`events_after` built a `BlobReader` without keys and `resolve()` recorded
+`NoKey` and returned `None` — silently. The daemon itself opens with
+`keys::provider_for_db`; only the upload path did not. Fix: the upload
+open carries the key provider; a reader note (missing key, unreadable
+blob) fails the upload with the cursor kept, so a conversation never leaves
+as metadata by accident; under `messages` only said-kinds open their
+blobs, and the inference recompute opens none. Regression test flushes a
+conversation into blobs under a key file, removes the key (upload held,
+nothing on the server, cursor 0), restores it (text arrives). The rows
+already uploaded without text cannot be repaired from the client — the
+server deduplicates by event id and does not merge content — so the owner's
+5 messages from that window stay metadata on the server; the local
+database has them.
+
 ## 2026-09-09 — the conversation leaves the device by default
 
 The owner's decision: every VibeMon install must collect and upload both

crates/attemptdb-capture/src/sync.rs

@@ -734,12 +734,16 @@ struct Ack {
     stripped_content: usize,
 }
 
-/// Open the database read-only: coexists with a running daemon.
+/// Open the database read-only: coexists with a running daemon. The key
+/// provider comes along: once the daemon's periodic flush has moved an
+/// event's content into an encrypted blob, a content profile can only
+/// read the conversation back with the key.
 fn open_read_only(locator: &Locator) -> Result<Database> {
     Database::open(
         &locator.db_dir,
         OpenOptions {
             read_only: true,
+            keys: crate::keys::provider_for_db(locator, &locator.db_dir),
             ..Default::default()
         },
     )
@@ -773,7 +777,7 @@ pub fn upload_once_with(
     let newest_seq = db.stats().last_source_seq;
     let after = state.last_acked_source_seq;
     let mut pending: Vec<Event> = if newest_seq > after {
-        events_after(&db, cfg, after)?
+        events_after(&db, cfg, after, cfg.sends_any_content())?
     } else {
         Vec::new()
     };
@@ -792,7 +796,9 @@ pub fn upload_once_with(
                 .as_deref()
                 .is_some_and(|e| e.starts_with("inferences:")));
     let allowed: Vec<Event> = if recompute_inferences {
-        let mut all = events_after(&db, cfg, 0)?;
+        // Inferences are computed from metadata; the whole history is
+        // re-read here, so no blob is opened for it.
+        let mut all = events_after(&db, cfg, 0, false)?;
         all.retain(|e| cfg.allows(e));
         all.sort_by_key(|e| e.source_seq);
         all
@@ -841,15 +847,40 @@ pub fn upload_once_with(
 
 /// Events past `after` in `source_seq` order, decoded from the segments
 /// whose range reaches past it plus the WAL. Content is resolved only when
-/// the profile sends it: the encrypted blobs are one file each, and a
-/// metadata upload never opens them.
-fn events_after(db: &Database, cfg: &PeerConfig, after: u64) -> Result<Vec<Event>> {
-    let reader = cfg.sends_any_content().then(|| {
+/// `with_content` asks for it (the profile sends it): the encrypted blobs
+/// are one file each, and a metadata upload never opens them. Under
+/// `messages` only the kinds that can carry something said open theirs.
+///
+/// A blob that cannot be read — no key, unreadable file — is an error, not
+/// a silently empty event: the caller keeps the cursor and retries, so a
+/// conversation never leaves the device as bare metadata by accident.
+fn events_after(
+    db: &Database,
+    cfg: &PeerConfig,
+    after: u64,
+    with_content: bool,
+) -> Result<Vec<Event>> {
+    let reader = with_content.then(|| {
         attemptdb_storage::blobs::BlobReader::new(
             db.blob_store(),
             db.key_provider().map(|k| k.as_ref()),
         )
     });
+    let all_kinds = |_: EventKind| true;
+    let said_kinds = |k: EventKind| {
+        matches!(
+            k,
+            EventKind::PromptSubmitted
+                | EventKind::AgentMessage
+                | EventKind::TurnStopped
+                | EventKind::Unknown
+        )
+    };
+    let wants_content: &dyn Fn(EventKind) -> bool = if cfg.send_content {
+        &all_kinds
+    } else {
+        &said_kinds
+    };
     let mut out = Vec::new();
     for seg in &db.manifest().segments {
         if seg.max_source_seq <= after {
@@ -860,10 +891,26 @@ fn events_after(db: &Database, cfg: &PeerConfig, after: u64) -> Result<Vec<Event
             .with_context(|| format!("reading segment {}", seg.file))?
         {
             out.extend(
-                attemptdb_storage::segment::batch_to_events_with(&b, reader.as_ref())
-                    .with_context(|| format!("decoding segment {}", seg.file))?
-                    .into_iter()
-                    .filter(|e| e.source_seq > after),
+                attemptdb_storage::segment::batch_to_events_where(
+                    &b,
+                    reader.as_ref(),
+                    wants_content,
+                )
+                .with_context(|| format!("decoding segment {}", seg.file))?
+                .into_iter()
+                .filter(|e| e.source_seq > after),
+            );
+        }
+    }
+    if let Some(reader) = &reader {
+        let notes = reader.notes();
+        if !notes.is_empty() {
+            bail!(
+                "content could not be read for the `{}` profile ({}); the upload is held so \
+                 nothing leaves without its text — restore the key, or `attempt sync profile \
+                 semantic` to send metadata only",
+                cfg.profile(),
+                notes.join("; ")
             );
         }
     }

crates/attemptdb-capture/tests/sync.rs

@@ -5,13 +5,14 @@
 #![cfg(unix)]
 
 use attemptdb_capture::ingest;
+use attemptdb_capture::keys::{self, InitOptions, KeyStoreOptions};
 use attemptdb_capture::locator::Locator;
 use attemptdb_capture::sync::{PeerConfig, SyncState, upload_once};
 use attemptdb_core::event::{EventContent, Provider};
 use attemptdb_core::{CaptureMode, DeviceId, Event, EventKind, ProjectRef};
 use attemptdb_server::auth::digest_hex;
 use attemptdb_server::{Server, ServerConfig};
-use attemptdb_storage::{Database, OpenOptions, ScanFilter};
+use attemptdb_storage::{Database, Identity, OpenOptions, ScanFilter};
 use serde_json::json;
 use std::path::{Path, PathBuf};
 
@@ -746,3 +747,70 @@ async fn two_peers_with_different_profiles_keep_independent_cursors() {
     meta.stop().await;
     sem.stop().await;
 }
+
+/// The daemon flushes the memtable into a segment every few minutes; from
+/// then on an event's content lives in an encrypted blob. An upload that
+/// comes later — after an outage, say — must read it back with the key,
+/// and must hold the batch rather than send bare metadata when it cannot.
+#[tokio::test]
+async fn messages_profile_reads_the_conversation_out_of_flushed_segments_or_holds_the_upload() {
+    let tmp = tempfile::tempdir().unwrap();
+    let (locator, device) = local_db(tmp.path());
+    let db_id = Identity::load(&locator.db_dir).unwrap().db_id;
+    keys::init(
+        &locator,
+        db_id,
+        &InitOptions {
+            key_file: true,
+            passphrase_env: None,
+            store: Some(KeyStoreOptions::offline()),
+        },
+    )
+    .unwrap();
+    {
+        let mut db = ingest::open_writer(&locator, false).unwrap();
+        assert!(db.key_provider().is_some(), "the writer holds the key file");
+        assert_eq!(db.ingest(conversation(device)).unwrap().duplicates, 0);
+        db.flush().unwrap().expect("a segment was written");
+        assert!(
+            db.blob_stats().unwrap().count > 0,
+            "content moved into encrypted blobs at the flush"
+        );
+    }
+    let server = start_server_with(tmp.path(), device, 4, CaptureMode::LocalSemantic).await;
+    let c = peer(&server.url, SyncProfile::Messages);
+
+    // Without the key the conversation cannot be read: the upload is held,
+    // nothing reaches the server, and the cursor does not move.
+    let key_file = keys::default_key_file(&locator, db_id);
+    let saved = attemptdb_storage::blobs::read_key_file(&key_file).unwrap();
+    std::fs::remove_file(&key_file).unwrap();
+    let err = upload(&locator, &c).await.unwrap_err().to_string();
+    assert!(err.contains("content could not be read"), "{err}");
+    assert!(
+        !server.data_dir.join("tenants").join("t1").exists(),
+        "nothing reached the server"
+    );
+    let state_path = SyncState::path(&locator.paths.data_dir, &locator.db_dir, "default");
+    assert_eq!(
+        SyncState::load(&state_path)
+            .map(|s| s.last_acked_source_seq)
+            .unwrap_or(0),
+        0
+    );
+
+    // With the key back, the flushed conversation arrives intact.
+    attemptdb_storage::blobs::write_key_file(&key_file, &saved).unwrap();
+    let r = upload(&locator, &c).await.unwrap();
+    assert_eq!(r.accepted, 4);
+    let stored = server.tenant_events();
+    assert_eq!(stored.len(), 4);
+    let text = serde_json::to_string(&stored).unwrap();
+    assert!(text.contains("make the retries idempotent"), "{text}");
+    assert!(
+        text.contains("I will read the webhook handler first."),
+        "{text}"
+    );
+    assert!(!text.contains("CANARY"), "{text}");
+    server.stop().await;
+}