Merge pull request 'refactor(app): #231 derivation cache 배칭 + 계측 — 가설은 재현 안 됨' (#237) from refactor/derivation-cache-batching into main

README.md

@@ -81,7 +81,7 @@ Markdown · PDF · 이미지(OCR + caption) · 소스코드(Rust/Python/TS/JS/Go
 | 명령 | 동작 |
 |------|------|
 | `kebab init` | XDG 경로에 데이터 디렉토리 + config.toml 생성 |
-| `kebab ingest [<path>]` | 워크스페이스 스캔 후 새/변경 문서 색인 (idempotent · incremental, `--force-reingest` 로 강제 재처리). 미지원 확장자는 자동 skip. 진행바는 현재 **파일명** · 느린 **phase(ocr/caption/embed)+모델명** · **경과초**`(Ns)` · 문서별 청크 수 · phase별 소요시간(parse/chunk/ocr/caption/embed/store)을 표시하고, 종료 시 **최장 소요 파일 top-5** 를 요약한다 (`--json` 은 `asset_phase`/`asset_chunked`/`asset_timings` 이벤트로, 사람용 요약은 미출력) |
+| `kebab ingest [<path>]` | 워크스페이스 스캔 후 새/변경 문서 색인 (idempotent · incremental, `--force-reingest` 로 강제 재처리). 미지원 확장자는 자동 skip. 진행바는 현재 **파일명** · 느린 **phase(ocr/caption/embed)+모델명** · **경과초**`(Ns)` · 문서별 청크 수 · phase별 소요시간(parse/chunk/ocr/caption/embed/store)과 임베딩 캐시 적중(`cache 히트/전체 소요`)을 표시하고, 종료 시 **최장 소요 파일 top-5** 를 요약한다 (`--json` 은 `asset_phase`/`asset_chunked`/`asset_timings` 이벤트로, 사람용 요약은 미출력) |
 | `kebab ingest-file <path>` | 단일 파일 ingest (workspace 외부 가능 — `_external/` 로 deterministic copy) |
 | `kebab ingest-stdin --title <T>` | stdin 의 markdown 본문 ingest |
 | `kebab search --mode {lexical,vector,hybrid} "<query>" [flags]` | 검색 (default hybrid = RRF fusion, citation 포함). 출처 필터 `--source <id>` (`[[workspace.sources]]` id) · `--source-type {markdown,note,paper,reference,inbox}` (둘 다 repeatable/comma-sep, OR). 그 외 필터/budget flag 는 `--help` |

crates/kebab-app/src/ingest.rs

@@ -1010,42 +1010,86 @@ fn unsupported_media_warning(path: &str) -> String {
     }
 }
 
+/// What one asset's trip through the derivation cache cost and produced.
+///
+/// Carried as a struct rather than four out-params: they are read and
+/// reported together (the `asset_timings` event wants all four), and a
+/// caller that updates three of them and forgets the fourth would report
+/// a cache that looks free.
+#[derive(Default)]
+pub(crate) struct CacheStats {
+    /// Chunks whose embedding was already cached.
+    pub hit: usize,
+    /// Chunks whose embedding had to be computed.
+    pub miss: usize,
+    /// Keys that hit, for the batched `last_used_at` bump the caller runs
+    /// after the vector upsert.
+    pub touch_keys: Vec<String>,
+    /// Cache-key hashing + lookup + hit/miss split + insert + touch, in
+    /// **microseconds**. Not the embedder call the misses trigger.
+    ///
+    /// The blake3 key hashing is inside the timer on purpose: it hashes
+    /// every chunk's full text and is pure CPU, so a label that said only
+    /// "lookup" would misattribute it on a miss-heavy run.
+    ///
+    /// Microseconds because a per-asset millisecond truncation loses the
+    /// measurement entirely: a document's cache work is routinely
+    /// sub-millisecond, so accumulating `as_millis()` per span reported
+    /// zero for most assets and made the total a lower bound rather than
+    /// a figure. The wire event converts to ms at emit.
+    pub us: u64,
+}
+
 /// Embed `texts` with the derivation cache (design 2026-05-31 §3.4).
 ///
-/// 1) 각 text 의 embedding cache_key 계산 → 히트/미스 분리.
+/// 1) 각 text 의 embedding cache_key 계산 → **한 번의 배치 조회**로 히트/미스 분리.
 /// 2) 미스 text 만 `emb.embed`(축소 배치) 호출.
-/// 3) 미스 결과를 `Vec<f32>` little-endian 으로 캐시 put.
+/// 3) 미스 결과를 `Vec<f32>` little-endian 으로 **한 트랜잭션에** 캐시 put.
 /// 4) 히트(bytes→Vec<f32>) + 미스 벡터를 **원래 순서대로** 합쳐 반환.
 ///
 /// 손상된 payload(길이 misalign)는 미스로 강등 → 재계산(정확성 우선, §3.5).
 /// 히트 키는 `touch_keys` 에 누적(호출측이 배치로 last_used_at 갱신).
+///
+/// Issue #231: 이 함수는 청크마다 SQLite 를 왕복했다. 그러면 캐시 히트가
+/// 미스보다 SQLite 작업을 덜 하지 않는다 — 미스가 추가로 내는 비용인 임베딩은
+/// GPU 에서 배치로, 전역 뮤텍스를 잡지 않고 돌기 때문이다. 결과적으로 캐시가
+/// "병렬 GPU 배치 1회" 를 "직렬 SQLite 왕복 N회" 로 바꿔치기하는 형태였다.
+/// 조회와 삽입을 배치 단위로 접어 그 비대칭을 없앤다.
 fn embed_with_cache(
     emb: &dyn Embedder,
     sqlite: &kebab_store_sqlite::SqliteStore,
     texts: &[&str],
     version_key: &str,
-    hit: &mut usize,
-    miss: &mut usize,
-    touch_keys: &mut Vec<String>,
+    stats: &mut CacheStats,
 ) -> anyhow::Result<Vec<Vec<f32>>> {
+    let t_cache = std::time::Instant::now();
+    let keys: Vec<String> = texts
+        .iter()
+        .map(|text| kebab_core::derivation_cache_key("embedding", text, version_key))
+        .collect();
+    let cached = sqlite.derivation_cache_get_many(&keys)?;
+    stats.us += u64::try_from(t_cache.elapsed().as_micros()).unwrap_or(u64::MAX);
+
     let mut out: Vec<Option<Vec<f32>>> = Vec::with_capacity(texts.len());
     let mut miss_indices: Vec<usize> = Vec::new();
     let mut miss_inputs: Vec<EmbeddingInput<'_>> = Vec::new();
-    let mut keys: Vec<String> = Vec::with_capacity(texts.len());
 
+    // The hit/miss split, including decoding a hit's payload back into
+    // `Vec<f32>`, is part of what the cache costs — a metric that stopped
+    // at the SQL boundary would flatter the cache.
+    let t_decode = std::time::Instant::now();
     for (i, text) in texts.iter().enumerate() {
-        let key = kebab_core::derivation_cache_key("embedding", text, version_key);
         // 히트 = 캐시에 있고 payload 가 정상 디코드되는 경우. 손상 payload 는
         // 미스로 강등(재계산, 정확성 우선 §3.5).
-        let cached = sqlite
-            .derivation_cache_get(&key)?
-            .and_then(|p| crate::derivation_payload::decode_embedding(&p));
-        if let Some(v) = cached {
-            *hit += 1;
-            touch_keys.push(key.clone());
+        let decoded = cached
+            .get(&keys[i])
+            .and_then(|p| crate::derivation_payload::decode_embedding(p));
+        if let Some(v) = decoded {
+            stats.hit += 1;
+            stats.touch_keys.push(keys[i].clone());
             out.push(Some(v));
         } else {
-            *miss += 1;
+            stats.miss += 1;
             miss_indices.push(i);
             miss_inputs.push(EmbeddingInput {
                 text,
@@ -1053,19 +1097,23 @@ fn embed_with_cache(
             });
             out.push(None);
         }
-        keys.push(key);
     }
+    stats.us += u64::try_from(t_decode.elapsed().as_micros()).unwrap_or(u64::MAX);
 
     if !miss_inputs.is_empty() {
         let miss_vectors = emb.embed(&miss_inputs)?;
+        let mut puts: Vec<(String, String, Vec<u8>)> = Vec::with_capacity(miss_indices.len());
         for (slot, v) in miss_indices.iter().zip(miss_vectors) {
-            sqlite.derivation_cache_put(
-                &keys[*slot],
-                "embedding",
-                &crate::derivation_payload::encode_embedding(&v),
-            )?;
+            puts.push((
+                keys[*slot].clone(),
+                "embedding".to_string(),
+                crate::derivation_payload::encode_embedding(&v),
+            ));
             out[*slot] = Some(v);
         }
+        let t_put = std::time::Instant::now();
+        sqlite.derivation_cache_put_many(&puts)?;
+        stats.us += u64::try_from(t_put.elapsed().as_micros()).unwrap_or(u64::MAX);
     }
 
     Ok(out
@@ -1369,8 +1417,7 @@ fn ingest_one_asset(
     // (purge + upsert), so per-phase timings attribute the bottleneck
     // correctly (review fix). Runs before any new upsert, as before.
     purge_vector_orphans_for_workspace_path(app, asset, vector_store)?;
-    let mut emb_cache_hit = 0_usize;
-    let mut emb_cache_miss = 0_usize;
+    let mut emb_cache = CacheStats::default();
     if let (Some(emb), Some(vec_store)) = (embedder, vector_store) {
         if !chunks.is_empty() {
             let model_id = emb.model_id();
@@ -1383,9 +1430,8 @@ fn ingest_one_asset(
             // (Document=`passage:`, Query=`query:`)를 붙여 같은 text 라도 벡터가
             // 달라지므로, 미래에 query 임베딩이 같은 캐시를 타도 충돌하지 않도록
             // 방어적으로 분리(현재 ingest 는 Document 고정이라 live 버그 없음).
-            let emb_version_key =
-                format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
-            let mut emb_touch_keys: Vec<String> = Vec::new();
+            let emb_version_key = format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
+
             // 본문 청크 text 로 캐시 조회 → 미스만 embed → 원래 순서로 합침.
             let body_texts: Vec<&str> = chunks.iter().map(|c| c.text.as_str()).collect();
             let vectors = embed_with_cache(
@@ -1393,9 +1439,7 @@ fn ingest_one_asset(
                 &app.sqlite,
                 &body_texts,
                 &emb_version_key,
-                &mut emb_cache_hit,
-                &mut emb_cache_miss,
-                &mut emb_touch_keys,
+                &mut emb_cache,
             )
             .context("Embedder::embed (document chunks)")?;
             let records: Vec<VectorRecord> = chunks
@@ -1420,7 +1464,11 @@ fn ingest_one_asset(
                 .collect();
             vec_store.upsert(&records).context("VectorStore::upsert")?;
             // 히트한 embedding 키들의 last_used_at 갱신(LRU 보존, §3.5).
-            app.sqlite.derivation_cache_touch(&emb_touch_keys)?;
+            {
+                let t_touch = std::time::Instant::now();
+                app.sqlite.derivation_cache_touch(&emb_cache.touch_keys)?;
+                emb_cache.us += u64::try_from(t_touch.elapsed().as_micros()).unwrap_or(u64::MAX);
+            }
         }
     }
 
@@ -1440,16 +1488,22 @@ fn ingest_one_asset(
             store_ms,
             ocr_ms: 0,
             caption_ms: 0,
+            cache_hit: u32::try_from(emb_cache.hit).unwrap_or(u32::MAX),
+            cache_miss: u32::try_from(emb_cache.miss).unwrap_or(u32::MAX),
+            cache_ms: emb_cache.us.div_ceil(1_000),
         },
     );
 
     // 검증용 hit/miss 카운트 노출(§3.4 / §6): warm 재색인이 embed 0회임을
     // 로그로 확인. tracing target 은 stderr 로 흐른다.
-    if emb_cache_hit + emb_cache_miss > 0 {
+    if emb_cache.hit + emb_cache.miss > 0 {
         tracing::info!(
             target: "kebab-app",
             doc = %canonical.doc_id.0,
-            "derivation cache: embedding hit={emb_cache_hit} miss={emb_cache_miss}"
+            "derivation cache: embedding hit={} miss={} in {}us",
+            emb_cache.hit,
+            emb_cache.miss,
+            emb_cache.us
         );
     }
 
@@ -1778,8 +1832,7 @@ fn ingest_one_image_asset(
         },
     );
     let t_embed = std::time::Instant::now();
-    let mut emb_cache_hit = 0_usize;
-    let mut emb_cache_miss = 0_usize;
+    let mut emb_cache = CacheStats::default();
     if let (Some(emb), Some(vec_store)) = (embedder, vector_store)
         && !chunks.is_empty()
     {
@@ -1789,18 +1842,15 @@ fn ingest_one_image_asset(
         // derivation cache(§3.4): same version_key formula + same code path as
         // the markdown handler (ingest.rs:1374). Media-agnostic — identical
         // chunk text shares one entry across media.
-        let emb_version_key =
-            format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
+        let emb_version_key = format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
         let body_texts: Vec<&str> = chunks.iter().map(|c| c.text.as_str()).collect();
-        let mut emb_touch_keys: Vec<String> = Vec::new();
+
         let vectors = embed_with_cache(
             &**emb,
             &app.sqlite,
             &body_texts,
             &emb_version_key,
-            &mut emb_cache_hit,
-            &mut emb_cache_miss,
-            &mut emb_touch_keys,
+            &mut emb_cache,
         )
         .context("Embedder::embed (image chunks)")?;
         let records: Vec<VectorRecord> = chunks
@@ -1826,7 +1876,11 @@ fn ingest_one_image_asset(
         vec_store
             .upsert(&records)
             .context("VectorStore::upsert (image)")?;
-        app.sqlite.derivation_cache_touch(&emb_touch_keys)?;
+        {
+            let t_touch = std::time::Instant::now();
+            app.sqlite.derivation_cache_touch(&emb_cache.touch_keys)?;
+            emb_cache.us += u64::try_from(t_touch.elapsed().as_micros()).unwrap_or(u64::MAX);
+        }
     }
     let embed_ms = u64::try_from(t_embed.elapsed().as_millis()).unwrap_or(u64::MAX);
 
@@ -1845,16 +1899,22 @@ fn ingest_one_image_asset(
             store_ms,
             ocr_ms,
             caption_ms,
+            cache_hit: u32::try_from(emb_cache.hit).unwrap_or(u32::MAX),
+            cache_miss: u32::try_from(emb_cache.miss).unwrap_or(u32::MAX),
+            cache_ms: emb_cache.us.div_ceil(1_000),
         },
     );
 
     // 검증용 hit/miss 카운트 노출(§3.4 / §6): warm 재색인이 embed 0회임을
     // 로그로 확인. tracing target 은 stderr 로 흐른다.
-    if emb_cache_hit + emb_cache_miss > 0 {
+    if emb_cache.hit + emb_cache.miss > 0 {
         tracing::info!(
             target: "kebab-app",
             doc = %canonical.doc_id.0,
-            "derivation cache: embedding hit={emb_cache_hit} miss={emb_cache_miss}"
+            "derivation cache: embedding hit={} miss={} in {}us",
+            emb_cache.hit,
+            emb_cache.miss,
+            emb_cache.us
         );
     }
 
@@ -2669,26 +2729,22 @@ fn ingest_one_pdf_asset(
         },
     );
     let t_embed = std::time::Instant::now();
-    let mut emb_cache_hit = 0_usize;
-    let mut emb_cache_miss = 0_usize;
+    let mut emb_cache = CacheStats::default();
     if let (Some(emb), Some(vec_store)) = (embedder, vector_store)
         && !chunks.is_empty()
     {
         let model_id = emb.model_id();
         let model_version = emb.model_version();
         let dimensions = emb.dimensions();
-        let emb_version_key =
-            format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
+        let emb_version_key = format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
         let body_texts: Vec<&str> = chunks.iter().map(|c| c.text.as_str()).collect();
-        let mut emb_touch_keys: Vec<String> = Vec::new();
+
         let vectors = embed_with_cache(
             &**emb,
             &app.sqlite,
             &body_texts,
             &emb_version_key,
-            &mut emb_cache_hit,
-            &mut emb_cache_miss,
-            &mut emb_touch_keys,
+            &mut emb_cache,
         )
         .context("Embedder::embed (pdf chunks)")?;
         let records: Vec<VectorRecord> = chunks
@@ -2714,7 +2770,11 @@ fn ingest_one_pdf_asset(
         vec_store
             .upsert(&records)
             .context("VectorStore::upsert (pdf)")?;
-        app.sqlite.derivation_cache_touch(&emb_touch_keys)?;
+        {
+            let t_touch = std::time::Instant::now();
+            app.sqlite.derivation_cache_touch(&emb_cache.touch_keys)?;
+            emb_cache.us += u64::try_from(t_touch.elapsed().as_micros()).unwrap_or(u64::MAX);
+        }
     }
     let embed_ms = u64::try_from(t_embed.elapsed().as_millis()).unwrap_or(u64::MAX);
 
@@ -2733,16 +2793,22 @@ fn ingest_one_pdf_asset(
             store_ms,
             ocr_ms: pdf_ocr_ms_total.unwrap_or(0),
             caption_ms: 0,
+            cache_hit: u32::try_from(emb_cache.hit).unwrap_or(u32::MAX),
+            cache_miss: u32::try_from(emb_cache.miss).unwrap_or(u32::MAX),
+            cache_ms: emb_cache.us.div_ceil(1_000),
         },
     );
 
     // 검증용 hit/miss 카운트 노출(§3.4 / §6): warm 재색인이 embed 0회임을
     // 로그로 확인. tracing target 은 stderr 로 흐른다.
-    if emb_cache_hit + emb_cache_miss > 0 {
+    if emb_cache.hit + emb_cache.miss > 0 {
         tracing::info!(
             target: "kebab-app",
             doc = %canonical.doc_id.0,
-            "derivation cache: embedding hit={emb_cache_hit} miss={emb_cache_miss}"
+            "derivation cache: embedding hit={} miss={} in {}us",
+            emb_cache.hit,
+            emb_cache.miss,
+            emb_cache.us
         );
     }
 
@@ -3012,26 +3078,22 @@ fn ingest_one_code_asset(
     purge_vector_orphans_for_workspace_path(app, asset, vector_store)?;
     store_document_records(app, asset, &bytes, &canonical, &chunks, " (code)")?;
 
-    let mut emb_cache_hit = 0_usize;
-    let mut emb_cache_miss = 0_usize;
+    let mut emb_cache = CacheStats::default();
     if let (Some(emb), Some(vec_store)) = (embedder, vector_store)
         && !chunks.is_empty()
     {
         let model_id = emb.model_id();
         let model_version = emb.model_version();
         let dimensions = emb.dimensions();
-        let emb_version_key =
-            format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
+        let emb_version_key = format!("doc|{}|{}|{}", model_id.0, model_version.0, dimensions);
         let body_texts: Vec<&str> = chunks.iter().map(|c| c.text.as_str()).collect();
-        let mut emb_touch_keys: Vec<String> = Vec::new();
+
         let vectors = embed_with_cache(
             &**emb,
             &app.sqlite,
             &body_texts,
             &emb_version_key,
-            &mut emb_cache_hit,
-            &mut emb_cache_miss,
-            &mut emb_touch_keys,
+            &mut emb_cache,
         )
         .context("Embedder::embed (code chunks)")?;
         let records: Vec<VectorRecord> = chunks
@@ -3057,16 +3119,23 @@ fn ingest_one_code_asset(
         vec_store
             .upsert(&records)
             .context("VectorStore::upsert (code)")?;
-        app.sqlite.derivation_cache_touch(&emb_touch_keys)?;
+        {
+            let t_touch = std::time::Instant::now();
+            app.sqlite.derivation_cache_touch(&emb_cache.touch_keys)?;
+            emb_cache.us += u64::try_from(t_touch.elapsed().as_micros()).unwrap_or(u64::MAX);
+        }
     }
 
     // 검증용 hit/miss 카운트 노출(§3.4 / §6): warm 재색인이 embed 0회임을
     // 로그로 확인. tracing target 은 stderr 로 흐른다.
-    if emb_cache_hit + emb_cache_miss > 0 {
+    if emb_cache.hit + emb_cache.miss > 0 {
         tracing::info!(
             target: "kebab-app",
             doc = %canonical.doc_id.0,
-            "derivation cache: embedding hit={emb_cache_hit} miss={emb_cache_miss}"
+            "derivation cache: embedding hit={} miss={} in {}us",
+            emb_cache.hit,
+            emb_cache.miss,
+            emb_cache.us
         );
     }
 

crates/kebab-app/src/ingest_progress.rs

@@ -139,6 +139,40 @@ pub enum IngestEvent {
         ocr_ms: u64,
         #[serde(default)]
         caption_ms: u64,
+        /// v0.32.1 (additive, issue #231): derivation-cache outcome for
+        /// this asset's **embedding** chunks, and what the cache path
+        /// cost. Without these the only way to tell whether the cache is
+        /// paying for itself was a `tracing::info!` on stderr, which is
+        /// gone the moment the run ends — so "is the cache a win on my
+        /// corpus" had no answer a user could look up.
+        ///
+        /// Scope: embeddings only. The OCR and caption derivations share
+        /// the same table but go through the single-key API and are not
+        /// counted here, so an image-heavy corpus under-reports what the
+        /// cache actually did.
+        #[serde(default)]
+        cache_hit: u32,
+        #[serde(default)]
+        cache_miss: u32,
+        /// Cache-key hashing, lookup, payload decode, insert, and the
+        /// `last_used_at` touch that every hit triggers — everything the
+        /// cache costs except the embedder call the misses go on to make.
+        ///
+        /// Rounded **up** to the millisecond. An asset's cache work is
+        /// routinely sub-millisecond, so truncating would report zero for
+        /// most assets and make any sum a systematic undercount; the cost
+        /// of rounding up is at most 1 ms per asset in the other
+        /// direction, which does not hide a cache that is expensive.
+        ///
+        /// The touch is counted deliberately: issue #231's sharpest claim
+        /// is that a read-only cache hit generates write traffic, and a
+        /// metric that left it out could not test that claim.
+        ///
+        /// **Included in `embed_ms`, not additional to it.** The embed
+        /// timer spans the whole vector phase, cache work included, so
+        /// summing the phase fields double-counts this one.
+        #[serde(default)]
+        cache_ms: u64,
     },
     /// v0.32.1 (additive): the post-scan sweep for documents whose source
     /// file is gone is starting, with `total` stored paths to examine
@@ -315,6 +349,9 @@ mod tests {
             store_ms: 20,
             ocr_ms: 1_200,
             caption_ms: 3_400,
+            cache_hit: 9,
+            cache_miss: 4,
+            cache_ms: 55,
         };
         let v = serde_json::to_value(&ev).unwrap();
         assert_eq!(
@@ -329,6 +366,12 @@ mod tests {
             ("embed_ms", 800),
             ("store_ms", 20),
             ("ocr_ms", 1_200),
+            // issue #231: the derivation-cache counters ride the same
+            // event, so an agent reading `asset_timings` can answer
+            // "did the cache pay for itself" without a second source.
+            ("cache_hit", 9),
+            ("cache_miss", 4),
+            ("cache_ms", 55),
             ("caption_ms", 3_400),
         ] {
             assert_eq!(

crates/kebab-app/tests/ingest_log_smoke.rs

@@ -207,6 +207,20 @@ fn ingest_log_records_the_deleted_file_sweep() {
     )
     .expect("first ingest should succeed");
 
+    // Remember the first run's log so the second can be identified by
+    // elimination. Sorting by filename does not work: the run id is
+    // `<second-resolution timestamp>-<random hex>`, so two runs inside the
+    // same second sort by the random half.
+    let logs_of = |dir: &std::path::Path| -> std::collections::BTreeSet<PathBuf> {
+        std::fs::read_dir(dir)
+            .unwrap()
+            .filter_map(Result::ok)
+            .map(|e| e.path())
+            .filter(|p| p.extension().is_some_and(|x| x == "ndjson"))
+            .collect()
+    };
+    let before = logs_of(&log_dir);
+
     std::fs::remove_file(&doomed).unwrap();
     let report = ingest_with_config(
         minimal_config(&workspace, &log_dir),
@@ -216,15 +230,11 @@ fn ingest_log_records_the_deleted_file_sweep() {
     .expect("second ingest should succeed");
     assert_eq!(report.purged_deleted_files, 1);
 
-    // The second run's log is the later one; both runs write into log_dir.
-    let mut logs: Vec<PathBuf> = std::fs::read_dir(&log_dir)
-        .unwrap()
-        .filter_map(Result::ok)
-        .map(|e| e.path())
-        .filter(|p| p.extension().is_some_and(|x| x == "ndjson"))
-        .collect();
-    logs.sort();
-    let body = std::fs::read_to_string(logs.last().expect("a log per run")).unwrap();
+    let after = logs_of(&log_dir);
+    let mut fresh = after.difference(&before);
+    let log = fresh.next().expect("the second run wrote a log");
+    assert!(fresh.next().is_none(), "and exactly one");
+    let body = std::fs::read_to_string(log).unwrap();
 
     let events: Vec<Value> = body
         .lines()

crates/kebab-cli/src/progress.rs

@@ -295,6 +295,9 @@ impl ProgressDisplay {
                 store_ms,
                 ocr_ms,
                 caption_ms,
+                cache_hit,
+                cache_miss,
+                cache_ms,
                 ..
             } => {
                 // v0.26.1: accumulate (path, total_ms) for the slowest summary.
@@ -321,6 +324,17 @@ impl ProgressDisplay {
                         parts.push(format!("caption {}", fmt_ms(*caption_ms)));
                     }
                     parts.push(format!("embed {}", fmt_ms(*embed_ms)));
+                    // Only when the cache was consulted at all — a run
+                    // with no embedder configured leaves these at zero and
+                    // the line stays as short as it was.
+                    if *cache_hit + *cache_miss > 0 {
+                        parts.push(format!(
+                            "cache {}/{} {}",
+                            cache_hit,
+                            cache_hit + cache_miss,
+                            fmt_ms(*cache_ms)
+                        ));
+                    }
                     parts.push(format!("store {}", fmt_ms(*store_ms)));
                     let _ = writeln!(err, "  ⏱ {}", parts.join(" · "));
                 }

crates/kebab-store-sqlite/src/derivation_cache.rs

@@ -23,18 +23,96 @@ impl SqliteStore {
     /// per batch (cheaper than a write per `get`).
     pub fn derivation_cache_get(&self, cache_key: &str) -> Result<Option<Vec<u8>>> {
         let conn = self.lock_conn();
+        // `prepare_cached`, not `query_row`: the latter prepares (and so
+        // re-parses the SQL) on every call. Issue #231.
         let payload: Option<Vec<u8>> = conn
-            .query_row(
-                "SELECT payload FROM derivation_cache WHERE cache_key = ?",
-                params![cache_key],
-                |row| row.get::<_, Vec<u8>>(0),
-            )
+            .prepare_cached("SELECT payload FROM derivation_cache WHERE cache_key = ?")
+            .map_err(StoreError::from)?
+            .query_row(params![cache_key], |row| row.get::<_, Vec<u8>>(0))
             .optional()
             .map_err(StoreError::from)
             .context("derivation_cache_get")?;
         Ok(payload)
     }
 
+    /// Look up many keys at once, returning only the ones present.
+    ///
+    /// One statement per `SQLITE_MAX_VARIABLE_NUMBER`-sized batch instead
+    /// of one round trip per key (issue #231). The per-key form made a
+    /// cache *hit* cost as many `lock_conn()` acquisitions as a miss —
+    /// and a miss at least does its expensive half (the embedding) on the
+    /// GPU, batched, outside the lock. Duplicate keys in `keys` are fine;
+    /// the result is keyed by cache_key so they collapse.
+    pub fn derivation_cache_get_many(
+        &self,
+        keys: &[String],
+    ) -> Result<std::collections::HashMap<String, Vec<u8>>> {
+        let mut found = std::collections::HashMap::with_capacity(keys.len());
+        if keys.is_empty() {
+            return Ok(found);
+        }
+        let conn = self.lock_conn();
+        // Conservative against the 999-parameter default: the bundled
+        // build allows far more, but the cost of a few extra statements
+        // is nothing next to the per-key round trips this replaces.
+        for batch in keys.chunks(900) {
+            let placeholders = std::iter::repeat_n("?", batch.len())
+                .collect::<Vec<_>>()
+                .join(",");
+            let mut stmt = conn
+                .prepare_cached(&format!(
+                    "SELECT cache_key, payload FROM derivation_cache WHERE cache_key IN ({placeholders})"
+                ))
+                .map_err(StoreError::from)
+                .context("derivation_cache_get_many: prepare")?;
+            let rows = stmt
+                .query_map(rusqlite::params_from_iter(batch.iter()), |row| {
+                    Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
+                })
+                .map_err(StoreError::from)
+                .context("derivation_cache_get_many: query")?;
+            for row in rows {
+                let (k, v) = row.map_err(StoreError::from)?;
+                found.insert(k, v);
+            }
+        }
+        Ok(found)
+    }
+
+    /// Insert many payloads in one transaction.
+    ///
+    /// The single-key [`Self::derivation_cache_put`] runs outside any
+    /// explicit transaction, so each call is its own implicit commit — a
+    /// WAL frame write and wal-index update per embedded chunk (issue
+    /// #231). A batch of misses is one logical unit of work and commits
+    /// as one.
+    pub fn derivation_cache_put_many(&self, entries: &[(String, String, Vec<u8>)]) -> Result<()> {
+        if entries.is_empty() {
+            return Ok(());
+        }
+        let now = OffsetDateTime::now_utc()
+            .format(&Rfc3339)
+            .context("format derivation_cache.created_at")?;
+        let mut conn = self.lock_conn();
+        let tx = conn.transaction().map_err(StoreError::from)?;
+        {
+            let mut stmt = tx
+                .prepare_cached(
+                    "INSERT OR REPLACE INTO derivation_cache
+                        (cache_key, kind, payload, created_at, last_used_at)
+                     VALUES (?, ?, ?, ?, ?)",
+                )
+                .map_err(StoreError::from)?;
+            for (key, kind, payload) in entries {
+                stmt.execute(params![key, kind, payload, now, now])
+                    .map_err(StoreError::from)
+                    .context("derivation_cache_put_many")?;
+            }
+        }
+        tx.commit().map_err(StoreError::from)?;
+        Ok(())
+    }
+
     /// Insert (or overwrite) a cached derivation payload.
     ///
     /// `INSERT OR REPLACE` so a re-computation of the same key (e.g. after a
@@ -47,12 +125,13 @@ impl SqliteStore {
             .format(&Rfc3339)
             .context("format derivation_cache.created_at")?;
         let conn = self.lock_conn();
-        conn.execute(
+        conn.prepare_cached(
             "INSERT OR REPLACE INTO derivation_cache
                 (cache_key, kind, payload, created_at, last_used_at)
              VALUES (?, ?, ?, ?, ?)",
-            params![cache_key, kind, payload, now, now],
         )
+        .map_err(StoreError::from)?
+        .execute(params![cache_key, kind, payload, now, now])
         .map_err(StoreError::from)
         .context("derivation_cache_put")?;
         Ok(())
@@ -73,7 +152,7 @@ impl SqliteStore {
         let tx = conn.transaction().map_err(StoreError::from)?;
         {
             let mut stmt = tx
-                .prepare("UPDATE derivation_cache SET last_used_at = ? WHERE cache_key = ?")
+                .prepare_cached("UPDATE derivation_cache SET last_used_at = ? WHERE cache_key = ?")
                 .map_err(StoreError::from)?;
             for key in keys {
                 stmt.execute(params![now, key])
@@ -150,6 +229,91 @@ mod tests {
         );
     }
 
+    #[test]
+    fn get_many_returns_only_present_keys() {
+        let (_d, store) = open_store();
+        store.derivation_cache_put("a", "embedding", b"A").unwrap();
+        store.derivation_cache_put("c", "embedding", b"C").unwrap();
+
+        let found = store
+            .derivation_cache_get_many(&[
+                "a".to_string(),
+                "b".to_string(),
+                "c".to_string(),
+                // A repeat: the embed path derives keys from chunk text,
+                // and a document with two identical chunks produces the
+                // same key twice. The map must collapse it, not trip.
+                "a".to_string(),
+            ])
+            .unwrap();
+        assert_eq!(found.len(), 2, "absent keys are simply not in the map");
+        assert_eq!(found.get("a").map(Vec::as_slice), Some(b"A".as_slice()));
+        assert_eq!(found.get("c").map(Vec::as_slice), Some(b"C".as_slice()));
+        assert!(!found.contains_key("b"));
+    }
+
+    #[test]
+    fn get_many_is_empty_for_no_keys() {
+        let (_d, store) = open_store();
+        assert!(store.derivation_cache_get_many(&[]).unwrap().is_empty());
+    }
+
+    /// The batch splits at 900 parameters. A run that embeds more chunks
+    /// than that in one call must still see every one of them — an
+    /// off-by-one in the chunking would silently turn hits into misses,
+    /// which costs nothing but correctness-invisible re-embedding.
+    #[test]
+    fn get_many_spans_more_keys_than_one_statement_holds() {
+        let (_d, store) = open_store();
+        let keys: Vec<String> = (0..2_000).map(|i| format!("k{i:05}")).collect();
+        for k in &keys {
+            store
+                .derivation_cache_put(k, "embedding", k.as_bytes())
+                .unwrap();
+        }
+        let found = store.derivation_cache_get_many(&keys).unwrap();
+        assert_eq!(found.len(), keys.len());
+        for k in &keys {
+            assert_eq!(
+                found.get(k).map(Vec::as_slice),
+                Some(k.as_bytes()),
+                "key {k} must survive the batch split"
+            );
+        }
+    }
+
+    #[test]
+    fn put_many_writes_all_and_replaces() {
+        let (_d, store) = open_store();
+        store
+            .derivation_cache_put("dup", "embedding", b"old")
+            .unwrap();
+        store
+            .derivation_cache_put_many(&[
+                ("x".to_string(), "embedding".to_string(), b"X".to_vec()),
+                ("y".to_string(), "embedding".to_string(), b"Y".to_vec()),
+                ("dup".to_string(), "embedding".to_string(), b"new".to_vec()),
+            ])
+            .unwrap();
+        let found = store
+            .derivation_cache_get_many(&["x".to_string(), "y".to_string(), "dup".to_string()])
+            .unwrap();
+        assert_eq!(found.get("x").map(Vec::as_slice), Some(b"X".as_slice()));
+        assert_eq!(found.get("y").map(Vec::as_slice), Some(b"Y".as_slice()));
+        assert_eq!(
+            found.get("dup").map(Vec::as_slice),
+            Some(b"new".as_slice()),
+            "the batch form keeps INSERT OR REPLACE semantics"
+        );
+    }
+
+    #[test]
+    fn put_many_with_no_entries_is_noop() {
+        let (_d, store) = open_store();
+        store.derivation_cache_put_many(&[]).unwrap();
+        assert!(store.derivation_cache_get_many(&[]).unwrap().is_empty());
+    }
+
     #[test]
     fn touch_missing_keys_is_noop() {
         let (_d, store) = open_store();

docs/DOGFOOD.md

@@ -278,6 +278,7 @@ echo "# stdin content" | "$RELEASE_BIN" ingest-stdin --title "from stdin" --conf
 - sweep 이 끝난 뒤 진행바가 asset 분모·라벨로 돌아오는가 (TTY). sweep 이 같은 바를 빌려 쓰므로 복구가 빠지면 색인 구간 내내 `sweep [..] 4213/21` 로 그려진다.
 - ndjson 로그에 `purge` 줄과 `sweep_summary` 가 남는가 (이슈 #228 이전에는 이 구간이 0바이트였다). purge 실패 시 `purge_failed`.
 - sweep 중 Ctrl-C 한 번에 실제로 멈추는가. `sweep_completed.checked` 가 예고한 `total` 이 아니라 실제 검사한 수로 나와야 한다.
+- `asset_timings` 의 `cache_hit` / `cache_miss` / `cache_ms` (v0.32.1, issue #231). `--force-reingest` 로 warm 재색인하면 `cache_miss == 0` 이어야 한다(그냥 재색인하면 변경 없는 문서가 통째로 skip 되어 `asset_timings` 자체가 안 나온다). `cache_ms` 는 **`embed_ms` 에 포함된** 값이라 phase 를 합산할 때 이중 계상하지 말 것. 임베딩 kind 만 세므로 이미지·PDF 위주 코퍼스에서는 캐시가 한 일을 과소 표현한다. code 자산은 `asset_timings` 자체를 emit 하지 않는다.
 
 ---
 

docs/wire-schema/v1/ingest_progress.schema.json

@@ -221,6 +221,21 @@
       "type": "integer",
       "minimum": 0,
       "description": "sweep_completed: documents removed because their source file is gone."
+    },
+    "cache_hit": {
+      "type": "integer",
+      "minimum": 0,
+      "description": "asset_timings: chunks whose embedding came from the derivation cache. Embeddings only — the OCR and caption derivations share the table but are not counted. Emitted on the markdown / image / PDF paths; the code path does not emit asset_timings at all."
+    },
+    "cache_miss": {
+      "type": "integer",
+      "minimum": 0,
+      "description": "asset_timings: chunks whose embedding had to be computed. Same scope as cache_hit."
+    },
+    "cache_ms": {
+      "type": "integer",
+      "minimum": 0,
+      "description": "asset_timings: what the derivation-cache path cost — cache-key hashing, lookup, payload decode, insert, and the last_used_at touch — excluding the embedder call the misses trigger. Rounded UP to the millisecond, because an asset's cache work is routinely sub-millisecond and truncating would make any sum a systematic undercount. INCLUDED IN embed_ms rather than additional to it: the embed timer spans the whole vector phase, so summing the phase fields double-counts this one."
     }
   }
 }

tasks/HOTFIXES.md

@@ -14,6 +14,74 @@ historical contract that was implemented; this file accumulates the
 deltas so phase 5+ readers can find the live behavior without diffing
 git history.
 
+## 2026-08-16 — #231 derivation_cache: 이슈 가설이 재현되지 않음 + 계측 노출
+
+### 이슈가 요청한 실측을 채웠다
+
+#231 은 "캐시가 히트하는데도 캐시를 우회하고 전량 재임베딩하는 편이 더 빠르다" 고 보고하면서, 본문에 **"⚠️ 정량 실측 보완 필요 … 수치 미기록"** 이라고 표를 비워 뒀다. 그 표를 채운 것이 이 항목이다.
+
+측정 환경을 먼저 정확히 적는다 — 리뷰에서 두 표의 모수가 어긋난다는 지적을 받아 바로잡은 것이다.
+
+- 코퍼스: 나무위키에서 뽑은 **1,157 파일**. 그중 365개는 확장자가 없어 skip 되고 **792개(`.md`)가 색인된다**. (run 로그의 `skipped 730` 은 아래 이중 스캔 때문에 365 × 2 다.)
+- **측정 설정의 artifact**: 이 시험용 config 는 `[[workspace.sources]]` 두 개(`wiki` / `jira`)가 **같은 root 를 가리킨다**. dogfood config 를 `sed` 로 고쳐 쓰면서 두 source 의 root 가 같아졌다. 그래서 walker 가 파일을 두 번 스캔하고, run 하나가 **자산 처리 1,584건**을 낸다. A/B 두 run 이 완전히 같은 설정을 쓰므로 비교 자체는 유효하지만, 수치를 읽을 때 알고 있어야 한다.
+- 저장된 결과: 문서 792건 / chunk 16,379개 / 캐시 항목 14,599개. (asset 은 763건 — 내용이 같은 파일이 blake3 로 합쳐진다.)
+- run 당 임베딩 캐시 조회 32,758회 (자산 처리 1,584건 × 평균 21청크).
+- 임베더: ollama + snowflake-arctic-embed2 (1024-dim). `--force-reingest` 로 문서 skip 을 우회해 전 문서 재처리.
+- 캐시 우회는 `derivation_cache` 테이블을 비워 전량 미스로 만들었다.
+
+| 조건 | 총 소요 |
+|---|---|
+| 캐시 히트 경로 | **139.1초** |
+| 캐시 우회(전량 재임베딩) | **1179.6초** |
+
+**캐시 히트가 8.5배 빠르다.** 이슈의 가설은 이 환경에서 재현되지 않는다.
+
+배칭 수정 전후(A-before / A-after)는 139.1초 / 139.3초로 측정 오차 안이다.
+
+새로 넣은 `asset_timings` 계측으로 내역을 뜯어보면 더 분명하다. 아래는 계측 정밀도를 고친 뒤의 별도 run(141.3초, 위 A/B 와 같은 스냅샷에서 시작, 자산 처리 1,584건 / 히트 32,758 / 미스 0):
+
+| 구간 | 소요 | 비중 |
+|---|---|---|
+| `embed_ms` — 벡터 phase 전체 | 75.2초 | 53% |
+| `store_ms` — SQLite 문서·청크 기록 | 57.0초 | 40% |
+| `chunk_ms` | 4.4초 | 3% |
+| `parse_ms` | 0.1초 | 0% |
+| ↳ 그중 **캐시 경로**(키 해싱+조회+디코드+삽입+touch) | **0.6~2.2초** | **0.4~1.6%** |
+
+두 가지를 정확히 적어 둔다. 둘 다 리뷰에서 지적받아 고친 것이다.
+
+`embed_ms` 는 "Lance upsert" 가 아니다. `t_embed` 스팬은 orphan purge(`purge_vector_orphans_for_workspace_path`) + 캐시 경로 + 임베더 호출 + 레코드 구성 + Lance upsert + touch 를 전부 감싼다 — 코드 주석 자신이 "purge + upsert" 라고 적고 있다.
+
+캐시 경로는 **`embed_ms` 안에 포함된 부분집합**이지 별도 가산 항목이 아니다. 그리고 0.6초는 **하한**이었다. `cache_ms` 는 자산당 한 번 ms 로 절삭되는데 자산 하나의 캐시 작업이 대개 1 ms 미만이라 **1,584건 중 1,422건(90%)이 0 으로 찍혔다**. 절삭 상한이 자산당 1 ms 이므로 참값은 0.6~2.2초 구간이다. 초안은 이걸 0.6초 점추정으로 적었고 그 숫자를 근거로 제안 3·6 을 기각했다 — 구간으로 고쳐 적는다. 결론(캐시 경로가 run 의 몇 %)은 구간 어느 쪽에서도 같다.
+
+내부 누적을 마이크로초로 바꿔 절삭 지점을 자산당 3곳에서 1곳(emit)으로 줄였고, 그 1곳은 **올림**으로 바꿔 계통적 하한을 없앴다. 올림 후 같은 run 을 다시 재니 0 으로 찍히는 자산이 하나도 없고 합이 **2.1초** 였다 — 내림이 하한 0.6초, 올림이 상한 2.1초이므로 참값은 그 사이이고, 앞서 산술로 낸 0.6~2.2초 구간이 실측으로 확인된 셈이다. 어느 쪽이든 run 141초의 1.5% 이하다. 히트 payload 를 `Vec<f32>` 로 되돌리는 디코드 비용과 blake3 키 해싱도 캐시 경로에 계상했다 — SQL 경계에서 멈추는 지표는 캐시를 실제보다 싸 보이게 한다.
+
+### 왜 원 보고가 틀렸다고 단정하지 않는가
+
+보고 이후 #229(chunks_fts 삭제가 FTS5 전체 스캔)와 #230(Lance fragment 누적)이 머지됐다. **#231 본문 스스로 #229 를 "같은 Mutex<Connection> 을 공유하므로 상호 증폭" 이라고 적었다.** 원 보고 환경에서는 캐시 조회 52,000회가 전부 그 뮤텍스를 잡았다 놓는데, 같은 뮤텍스 위에서 chunk 삭제가 FTS5 전체 스캔을 돌리고 있었다. #229 를 고치면 이 이슈의 체감이 함께 줄어든다는 것도 이슈가 예측한 대로다.
+
+그러니 이 실측은 "보고가 틀렸다" 가 아니라 **"#229 가 머지된 뒤에는 재현되지 않는다"** 로 읽는 것이 맞다. 이 머신이 61 GB RAM 이라 DB 가 통째로 페이지 캐시에 올라간다는 점도 함께 적어 둔다.
+
+### 그래도 반영한 것
+
+제안 1·2·4 는 실측과 무관하게 왕복 수가 줄어들 뿐 잃는 것이 없어서 넣었다. 다만 **A/B 벽시계 차이는 139.1초 → 139.3초로 측정 오차 안** 이다. 정직하게 말해 이 코퍼스에서는 체감이 없다.
+
+- `derivation_cache_get_many` — `WHERE cache_key IN (…)` 로 배치 조회. 문서 하나가 평균 21 chunk 이므로 왕복이 21회에서 1회가 된다.
+- `derivation_cache_put_many` — 미스 벡터를 한 트랜잭션에. 기존 단건 `put` 은 명시 트랜잭션 밖이라 행마다 암묵 커밋이었다.
+- `prepare_cached` — get/put/touch 셋 다. `query_row` 는 호출마다 SQL 을 다시 파싱한다.
+
+제안 5(계측 노출)가 이번의 실질적 산출물이다. `asset_timings` 에 `cache_hit` / `cache_miss` / `cache_ms` 를 additive 로 실었다. 두 가지 한계를 문서·스키마에 명시했다 — **임베딩 kind 만** 센다(같은 테이블을 쓰는 OCR·caption 파생은 단건 API 라 안 잡힌다), 그리고 **code 자산은 `asset_timings` 자체를 emit 하지 않는다**(이 PR 이전부터의 공백이고, 채우려면 code 경로에 parse/chunk/store 타이머를 새로 깔아야 해서 #231 범위 밖이다). 이전에는 hit/miss 가 `tracing::info!` 로 stderr 에만 나가서 run 이 끝나면 사라졌고, "내 코퍼스에서 캐시가 이득인가" 를 사용자가 확인할 방법이 없었다. `cache_ms` 에는 **touch 도 포함**한다 — 이슈의 가장 날카로운 지적이 "읽기 전용이어야 할 히트 경로가 쓰기를 만든다" 인데, touch 를 빼고 재는 지표로는 그 주장을 검증할 수 없다.
+
+### 반영하지 않은 것
+
+- 제안 3(touch 를 히트 경로에서 분리). 실측상 캐시 경로 전체가 0.6~2.2초라 touch 만 떼어낼 이유가 없고, 이슈가 권한 (c)안(`created_at` 기반 TTL)은 LRU 를 age 기반 축출로 바꾸는 의미 변경이다. 근거 없이 할 변경이 아니다.
+- 제안 6(캐시 우회 스위치). 이슈 스스로 "1~4 로 해결되면 불필요 — 플래그부터 만들지 말 것" 이라고 적었다.
+- 원인 6(4 KB BLOB overflow page). `page_size` 변경은 기존 DB 에서 VACUUM 을 요구하는데, kebab 은 VACUUM 을 실행하지 않는다(#229 항목 참조). 1% 남짓을 줄이자고 낼 비용이 아니다.
+
+### 곁다리: 내가 만든 flaky test
+
+`ingest_log_records_the_deleted_file_sweep`(#228 에서 추가)이 두 run 의 로그 중 뒤엣것을 파일명 정렬로 골랐다. run id 가 `<초 단위 타임스탬프>-<난수 hex>` 라 같은 초에 끝난 두 run 은 난수 쪽으로 정렬된다. 이번 전체 테스트에서 우연히 터져 잡았다. 첫 run 의 로그 집합을 먼저 기록해 두고 차집합으로 고르도록 고쳤다.
+
 ## 2026-08-16 — #228 sweep 구간이 진행바·로그에 표시되지 않음
 
 ### 무엇이 문제였나