fix(kebab-store-sqlite): purge stale assets row on workspace_path orphan + smoke

P7-3 통합 테스트가 노출한 storage 레이어 버그 fix.
`assets.workspace_path` 의 UNIQUE 제약과 `upsert_asset_row` 의
`ON CONFLICT(asset_id)` 만 처리하던 gap 사이 — byte 가 변경된 자산
re-ingest 시 새 asset_id 가 같은 workspace_path 에서 secondary UNIQUE
충돌. md / image / pdf 모두 영향.

Fix:
- 새 helper `purge_orphan_at_workspace_path` 가 같은 `workspace_path`
  의 *다른* `asset_id` 를 발견하면 documents → assets 순서로 sweep.
  documents 의 ON DELETE RESTRICT 회피 + CASCADE 로 blocks / chunks /
  embedding_records 정리. copied 모드면 storage_path 의 byte 파일도
  best-effort 삭제.
- `put_asset_with_bytes` 의 두 분기 (copy / reference) + `DocumentStore
  ::put_asset` 모두 호출.
- 회귀 테스트 `put_asset_with_bytes_sweeps_workspace_path_orphan` (이전
  의 "UPSERT 실패시 orphan 청소" 테스트가 더 이상 doable 하지 않으므로
  대체).
- `re_ingest_edited_pdf_produces_new_doc_id` integration `#[ignore]` 해제 →
  9 통합 테스트 모두 default 로 통과.

Vector store orphan 은 별도 P+ task — LanceDB 가 SQLite cascade 와 무관하게
운영되므로 stale chunk_id vector 가 디스크에 남음. 검색에는 영향 없음 (search 가
SQLite join 통해 surface).

Smoke 검증 (release binary, markdown 2 + image 1 + PDF 2):
- doctor pass
- 첫 ingest: 5 new
- list docs: 5 docs all media types
- search lexical "pdf-page-v1 chunker" → whitepaper.pdf hit
- search hybrid → cross-media 결과
- inspect doc PDF: parser_version=pdf-text-v1, blocks 가 SourceSpan::Page
- 동일 byte re-ingest: 5 updated, 0 errors (P1 idempotency)
- byte 수정 후 re-ingest: 1 new (해당 PDF) + 4 updated, 0 errors (storage fix)
- corrupt PDF 추가: errors+=1 + IngestItem.error 메시지 정확, 다른 자산 영향 0
- 정리 후 다시 ingest: errors=0
- RAG ask: PDF 인용 + `citations[].citation` 에 `kind: "page"` + `page: <N>` +
  `path: <pdf_path>` 정확히 노출

운영 fixture 보조:
- `crates/kebab-parse-pdf/examples/gen_smoke_pdf.rs` — `cargo run --release
  --example gen_smoke_pdf -p kebab-parse-pdf -- <out.pdf> <text-pages>` 로
  reportlab/qpdf 없이 in-tree PDF 생성.
- `crates/kebab-parse-image/examples/gen_smoke_png.rs` — 동일 방식의 PNG
  fixture 생성.
- SMOKE.md 가 두 example 사용법 + 갱신된 HOTFIXES 동작 (byte 수정 시
  errors+=1 → new+=1) 반영.

HOTFIXES `2026-05-02 P7-3` entry 가 \"deferred\" → \"fixed in same PR\" 로
업데이트, vector store orphan caveat 만 남음.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/kebab-store-sqlite/tests/asset_writer.rs

@@ -100,22 +100,23 @@ fn reference_mode_does_not_write_file_but_records_path() {
 }
 
 #[test]
-fn put_asset_with_bytes_orphan_cleanup_on_upsert_failure() {
-    // Goal: prove that if the row UPSERT fails AFTER the bytes have been
-    // staged on disk, no `<aa>/<asset_id>` file is left behind.
+fn put_asset_with_bytes_sweeps_workspace_path_orphan() {
+    // HOTFIXES 2026-05-02 P7-3: the original behaviour erred on
+    // workspace_path UNIQUE conflict (`ON CONFLICT(asset_id)` only) so
+    // a re-ingest of an edited file was unrecoverable. The fix is
+    // `purge_orphan_at_workspace_path`, which sweeps the stale
+    // documents → assets chain before the new INSERT lands.
     //
-    // Lever: the `assets` table has a UNIQUE INDEX on `workspace_path`
-    // (V001), but the UPSERT is `ON CONFLICT(asset_id)`. So if some other
-    // row already owns this `workspace_path`, the INSERT half of the
-    // UPSERT trips a UNIQUE constraint that the ON CONFLICT clause does
-    // NOT handle — UPSERT errors. The new asset's bytes were already
-    // staged; we assert they are NOT visible at the final destination.
+    // This test exercises the no-documents flavour (raw asset row only)
+    // — the put_asset_with_bytes path. The documents-cascade flavour
+    // is exercised end-to-end in `kebab-app::tests::pdf_pipeline::
+    // re_ingest_edited_pdf_produces_new_doc_id`.
     let env = common::TestEnv::with_threshold(100);
     let store = SqliteStore::open(&env.config()).unwrap();
     store.run_migrations().unwrap();
 
-    // Pre-populate a row that owns `notes/foo.md` (the workspace_path our
-    // fixture asset will also claim) under a *different* asset_id.
+    // Pre-populate a row that owns `notes/foo.md` under a *different*
+    // asset_id, simulating a prior ingest of an earlier byte version.
     env.with_conn(|c| {
         c.execute(
             "INSERT INTO assets (
@@ -140,36 +141,36 @@ fn put_asset_with_bytes_orphan_cleanup_on_upsert_failure() {
     let cs = b3_full_hex(bytes);
     let asset = fixed_asset(bytes, bytes.len() as u64, &cs);
 
-    let err = store
+    store
         .put_asset_with_bytes(&asset, bytes)
-        .expect_err("UPSERT must fail on workspace_path UNIQUE violation");
-    let msg = format!("{err:#}");
-    assert!(
-        msg.to_lowercase().contains("unique") || msg.to_lowercase().contains("constraint"),
-        "expected UNIQUE constraint failure, got: {msg}"
-    );
+        .expect("orphan sweep + INSERT must succeed");
 
-    // Final destination must NOT exist (no orphan).
+    // Stale row gone, new row owns the workspace_path.
+    let stale_count: i64 = env.with_conn(|c| {
+        c.query_row(
+            "SELECT COUNT(*) FROM assets WHERE asset_id = ?",
+            rusqlite::params!["b".repeat(32)],
+            |row| row.get(0),
+        )
+    });
+    assert_eq!(stale_count, 0, "stale asset_id must be purged");
+    let new_count: i64 = env.with_conn(|c| {
+        c.query_row(
+            "SELECT COUNT(*) FROM assets WHERE asset_id = ?",
+            rusqlite::params![asset.asset_id.0],
+            |row| row.get(0),
+        )
+    });
+    assert_eq!(new_count, 1, "new asset_id must own the workspace_path slot");
+
+    // New asset's bytes published at the final destination.
     let aa = &asset.asset_id.0[..2];
     let dest = env.data_dir().join("assets").join(aa).join(&asset.asset_id.0);
     assert!(
-        !dest.exists(),
-        "asset bytes were left orphan at {} after UPSERT failure",
+        dest.exists(),
+        "new asset bytes must be visible at {}",
         dest.display()
     );
-    // No `*.tmp.*` either — temp file must be cleaned up too.
-    let shard_dir = env.data_dir().join("assets").join(aa);
-    if let Ok(entries) = std::fs::read_dir(&shard_dir) {
-        for entry in entries.flatten() {
-            let name = entry.file_name();
-            let s = name.to_string_lossy();
-            assert!(
-                !s.contains(".tmp."),
-                "temp file leaked at {}",
-                entry.path().display()
-            );
-        }
-    }
 }
 
 #[test]