diff --git a/CHANGELOG.md b/CHANGELOG.md index 101c71c..b9a1acc 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/Cargo.lock b/Cargo.lock index dca8bc7..6411350 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -513,7 +513,7 @@ checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" [[package]] name = "attempt-hook" -version = "0.2.12" +version = "0.2.13" dependencies = [ "attemptdb-capture", "winresource", @@ -521,7 +521,7 @@ dependencies = [ [[package]] name = "attemptdb" -version = "0.2.12" +version = "0.2.13" dependencies = [ "anyhow", "attemptdb-adapters", @@ -549,7 +549,7 @@ dependencies = [ [[package]] name = "attemptdb-adapters" -version = "0.2.12" +version = "0.2.13" dependencies = [ "attemptdb-core", "serde", @@ -560,7 +560,7 @@ dependencies = [ [[package]] name = "attemptdb-bench" -version = "0.2.12" +version = "0.2.13" dependencies = [ "anyhow", "arrow", @@ -579,7 +579,7 @@ dependencies = [ [[package]] name = "attemptdb-capture" -version = "0.2.12" +version = "0.2.13" dependencies = [ "anyhow", "argon2", @@ -609,7 +609,7 @@ dependencies = [ [[package]] name = "attemptdb-core" -version = "0.2.12" +version = "0.2.13" dependencies = [ "chrono", "crc32c", @@ -624,7 +624,7 @@ dependencies = [ [[package]] name = "attemptdb-mcp" -version = "0.2.12" +version = "0.2.13" dependencies = [ "anyhow", "attemptdb-capture", @@ -641,7 +641,7 @@ dependencies = [ [[package]] name = "attemptdb-project" -version = "0.2.12" +version = "0.2.13" dependencies = [ "attemptdb-core", "serde", @@ -652,7 +652,7 @@ dependencies = [ [[package]] name = "attemptdb-query" -version = "0.2.12" +version = "0.2.13" dependencies = [ "anyhow", "async-trait", @@ -672,7 +672,7 @@ dependencies = [ [[package]] name = "attemptdb-server" -version = "0.2.12" +version = "0.2.13" dependencies = [ "anyhow", "attemptdb-adapters", @@ -696,7 +696,7 @@ dependencies = [ [[package]] name = "attemptdb-storage" -version = "0.2.12" +version = "0.2.13" dependencies = [ "arrow", "attemptdb-core", @@ -718,7 +718,7 @@ dependencies = [ [[package]] name = "attemptdb-ui" -version = "0.2.12" +version = "0.2.13" dependencies = [ "anyhow", "attemptdb-capture", diff --git a/Cargo.toml b/Cargo.toml index 5cb92e9..d8f24ec 100644 --- a/Cargo.toml +++ b/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" diff --git a/PROGRESS.md b/PROGRESS.md index 04b7bac..9418973 100644 --- a/PROGRESS.md +++ b/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 diff --git a/crates/attemptdb-capture/src/sync.rs b/crates/attemptdb-capture/src/sync.rs index f97e739..3866f38 100644 --- a/crates/attemptdb-capture/src/sync.rs +++ b/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::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 = 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 = 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> { - 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> { + 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 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("; ") ); } } diff --git a/crates/attemptdb-capture/tests/sync.rs b/crates/attemptdb-capture/tests/sync.rs index 6df12d7..47c0715 100644 --- a/crates/attemptdb-capture/tests/sync.rs +++ b/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; +}