refactor(sdk): drop async on ingest verbs that never yield (#419)

* refactor(sdk): drop async on ingest verbs that never yield (#334)

Every ingest verb was `async fn` but the bodies were sync filesystem
walks and rusqlite writes — no `.await` actually reached the runtime.
Drop the `async` annotation so the type matches the semantics; callers
that need to run these from an async context can wrap them in
`tokio::task::spawn_blocking` (the sdk-node binding now does).

Also delete `run_ingest_tick` (a one-line wrapper over `ingest().await`)
and the dead `in_flight.lock().await` belt-and-braces in
`WatchController::stop` — the task handle await already covers any
in-flight tick.

CLI presenters that built a tokio runtime just to `block_on` ingest now
call the sync verb directly; only `burn ingest --watch` keeps its
runtime (the watch loop still spawns async tasks for the FS-event
driver).

https://claude.ai/code/session_01RAt76YyVgMYdij9zeqPUFB

* fix(sdk): restore in_flight barrier in WatchController::stop

Codex review on #419 flagged that dropping the trailing
`in_flight.lock().await` regresses the "await any in-flight tick"
contract: the public `tick()` path doesn't check the `stopped` flag, so
a concurrent caller can grab `in_flight` and run an ingest after the
loop task has already exited. Awaiting the spawned task handle alone
covers the runner-driven path but not that one — callers that tear down
state right after `stop().await` could race an active write.

Put the trailing `in_flight.lock().await` back and document why it
isn't redundant.

https://claude.ai/code/session_01RAt76YyVgMYdij9zeqPUFB

* docs(changelog): trim Unreleased ingest entry to impact-first bullet

CodeRabbit review feedback on #419: per the project's CLAUDE.md
guidelines, changelog entries should be concise and impact-first.

https://claude.ai/code/session_01RAt76YyVgMYdij9zeqPUFB

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: willwashburn

CHANGELOG.md

@@ -6,6 +6,8 @@ Cross-package release notes for relayburn. Package changelogs contain package-le
 
 ### Changed
 
+- `relayburn-sdk`: ingest verbs are now synchronous; async callers should
+  run them via `tokio::task::spawn_blocking`.
 - `relayburn-sdk`: lower per-record allocations in reader hashing, tool-result
   sizing, relationship dedup, and project resolution. Cuts overhead during
   large session imports and concurrent `resolve_project` calls.
@@ -15,6 +17,10 @@ Cross-package release notes for relayburn. Package changelogs contain package-le
   JSON lines lacking a trailing newline still surface in the single-shot
   output.
 
+### Removed
+
+- `relayburn-sdk`: removed `run_ingest_tick`.
+
 ## [2.8.3] - 2026-05-11
 
 ### Changed

crates/relayburn-cli/src/commands/hotspots.rs

@@ -228,12 +228,9 @@ fn run_inner(globals: &GlobalArgs, args: HotspotsArgs) -> anyhow::Result<i32> {
     progress.set_task("opening ledger");
     let mut handle = Ledger::open(opts)?;
 
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()?;
     progress.set_task("refreshing ledger");
     let raw_opts = progress.ingest_options(ledger_home.clone());
-    rt.block_on(ingest_all(handle.raw_mut(), &raw_opts))?;
+    ingest_all(handle.raw_mut(), &raw_opts)?;
     drop(handle);
 
     let session_filter = match args.session.as_deref() {

crates/relayburn-cli/src/commands/ingest.rs

@@ -74,8 +74,7 @@ pub fn run(globals: &GlobalArgs, args: IngestArgs) -> i32 {
 }
 
 /// One-shot scan: open the ledger, run a single `ingest_all`, log the
-/// summary, exit. Drives a current-thread tokio runtime so the otherwise
-/// sync presenter can drive the async SDK verb.
+/// summary, exit.
 ///
 /// Summary line is emitted on **stdout** (matching TS `runIngestOnce`
 /// at `packages/cli/src/commands/ingest.ts:121-126`) so callers can
@@ -92,20 +91,9 @@ fn run_once(globals: &GlobalArgs, quiet: bool) -> i32 {
             return report_error(&err, globals);
         }
     };
-    progress.set_task("starting runtime");
-    let rt = match tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-    {
-        Ok(rt) => rt,
-        Err(err) => {
-            progress.finish_and_clear();
-            return report_error(&err, globals);
-        }
-    };
     progress.set_task("scanning sessions");
     let opts = progress.ingest_options(globals.ledger_path.clone());
-    let result = rt.block_on(ingest_all(handle.raw_mut(), &opts));
+    let result = ingest_all(handle.raw_mut(), &opts);
     progress.finish_and_clear();
     match result {
         Ok(report) => {
@@ -203,7 +191,7 @@ fn run_watch(globals: &GlobalArgs, args: &IngestArgs) -> i32 {
                 } else {
                     progress.ingest_options(ledger_home)
                 };
-                let result = ingest_all(guard.raw_mut(), &opts).await;
+                let result = ingest_all(guard.raw_mut(), &opts);
                 progress.set_task(watch_message);
                 result
             })
@@ -322,30 +310,14 @@ fn run_hook(globals: &GlobalArgs, hook: &str, quiet: bool) -> i32 {
             return 0;
         }
     };
-    if let Some(progress) = &progress {
-        progress.set_task("starting runtime");
-    }
-    let rt = match tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-    {
-        Ok(rt) => rt,
-        Err(err) => {
-            if let Some(progress) = &progress {
-                progress.finish_and_clear();
-            }
-            eprintln!("[burn] ingest: {err}");
-            return 0;
-        }
-    };
     if let Some(progress) = &progress {
         progress.set_task("scanning sessions");
     }
     let opts = match &progress {
         Some(progress) => progress.ingest_options(globals.ledger_path.clone()),
         None => TaskProgress::quiet_ingest_options(globals.ledger_path.clone()),
     };
-    let result = rt.block_on(ingest_all(handle.raw_mut(), &opts));
+    let result = ingest_all(handle.raw_mut(), &opts);
     if let Some(progress) = &progress {
         progress.finish_and_clear();
     }

crates/relayburn-cli/src/commands/state.rs

@@ -488,10 +488,7 @@ fn run_reset(globals: &GlobalArgs, args: crate::cli::StateResetArgs) -> i32 {
     print_reset_report(globals, &summary, /*executed=*/ true, ingest_report.as_ref())
 }
 
-/// Drive a single `ingest_all` sweep on the open handle. Mirrors the
-/// `run_ingest` helper in `commands/summary.rs`: the SDK verb is async,
-/// so we spin a current-thread tokio runtime to drive it from this
-/// otherwise-sync presenter.
+/// Drive a single `ingest_all` sweep on the open handle.
 ///
 /// `ledger_home` propagates the global `--ledger-path` override into
 /// `RawIngestOptions::ledger_home` so sidecar ingest state (config and
@@ -504,12 +501,9 @@ fn run_reset_reingest(
     ledger_home: Option<std::path::PathBuf>,
     progress: &TaskProgress,
 ) -> anyhow::Result<relayburn_sdk::IngestReport> {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()?;
     progress.set_task("re-ingesting sessions");
     let opts = progress.ingest_options(ledger_home);
-    rt.block_on(ingest_all(handle.raw_mut(), &opts))
+    ingest_all(handle.raw_mut(), &opts)
 }
 
 fn print_reset_report(

crates/relayburn-cli/src/commands/summary.rs

@@ -325,27 +325,18 @@ fn parse_tag_filters(tags: &[String]) -> anyhow::Result<BTreeMap<String, String>
     Ok(out)
 }
 
-/// Run an ingest sweep on the open handle. Builds a current-thread tokio
-/// runtime so the otherwise-sync presenter can drive the async verb.
+/// Run an ingest sweep on the open handle.
 fn run_ingest(
     handle: &mut LedgerHandle,
     progress: &TaskProgress,
     ledger_home: Option<std::path::PathBuf>,
 ) -> anyhow::Result<relayburn_sdk::IngestReport> {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .map_err(|err| {
-            progress.finish_and_clear();
-            err
-        })?;
     progress.set_task("refreshing ledger");
     let opts = progress.ingest_options(ledger_home);
-    rt.block_on(ingest_all(handle.raw_mut(), &opts))
-        .map_err(|err| {
-            progress.finish_and_clear();
-            err
-        })
+    ingest_all(handle.raw_mut(), &opts).map_err(|err| {
+        progress.finish_and_clear();
+        err
+    })
 }
 
 const COVERAGE_FIELDS: [CoverageField; 5] = [

crates/relayburn-cli/src/harnesses/claude.rs

@@ -146,7 +146,7 @@ impl HarnessAdapter for ClaudeAdapter {
         let mut handle = Ledger::open(LedgerOpenOptions::default())?;
         let cwd_str = ctx.cwd.to_string_lossy().into_owned();
         let opts = RawIngestOptions::default();
-        ingest_claude_session(handle.raw_mut(), &cwd_str, session_id, &opts).await
+        ingest_claude_session(handle.raw_mut(), &cwd_str, session_id, &opts)
     }
 }
 

crates/relayburn-cli/src/harnesses/codex.rs

@@ -10,13 +10,12 @@
 //! * `ingest_sessions` — defers to [`relayburn_sdk::ingest_codex_sessions`],
 //!   the codex-only ingest pass. The factory opens a fresh ledger handle
 //!   per call (mirrors the TS lock-then-write-then-close shape; SQLite WAL
-//!   keeps the per-tick open cheap).
+//!   keeps the per-tick open cheap). The SDK verb is sync, so we pass it
+//!   directly as a fn pointer to [`pending_stamp::session_store_adapter`].
 
-use std::future::Future;
 use std::path::PathBuf;
-use std::pin::Pin;
 
-use relayburn_sdk::{ingest_codex_sessions, IngestReport, RawIngestOptions, RawLedger};
+use relayburn_sdk::ingest_codex_sessions;
 
 use super::pending_stamp;
 use super::HarnessAdapter;
@@ -29,20 +28,11 @@ fn codex_sessions_dir() -> PathBuf {
     home_dir().join(".codex").join("sessions")
 }
 
-/// Box-pin the SDK's `async fn ingest_codex_sessions` into a fn pointer
-/// the [`pending_stamp::SessionIngestor`] type alias accepts.
-fn codex_ingest<'a>(
-    ledger: &'a mut RawLedger,
-    opts: &'a RawIngestOptions,
-) -> Pin<Box<dyn Future<Output = anyhow::Result<IngestReport>> + Send + 'a>> {
-    Box::pin(ingest_codex_sessions(ledger, opts))
-}
-
 /// Hand out a `&'static dyn HarnessAdapter` for codex. The registry calls
 /// this once at lazy-init time. See
 /// [`pending_stamp::session_store_adapter`] for the leak semantics.
 pub fn adapter() -> &'static dyn HarnessAdapter {
-    pending_stamp::session_store_adapter("codex", codex_sessions_dir, codex_ingest)
+    pending_stamp::session_store_adapter("codex", codex_sessions_dir, ingest_codex_sessions)
 }
 
 #[cfg(test)]

crates/relayburn-cli/src/harnesses/opencode.rs

@@ -13,13 +13,12 @@
 //!   [`relayburn_sdk::ingest_opencode_sessions`], the opencode-only ingest
 //!   pass. The factory opens a fresh ledger handle per call (mirrors the
 //!   TS lock-then-write-then-close shape; SQLite WAL keeps the per-tick
-//!   open cheap).
+//!   open cheap). The SDK verb is sync, so we pass it directly as a fn
+//!   pointer to [`pending_stamp::session_store_adapter`].
 
-use std::future::Future;
 use std::path::PathBuf;
-use std::pin::Pin;
 
-use relayburn_sdk::{ingest_opencode_sessions, IngestReport, RawIngestOptions, RawLedger};
+use relayburn_sdk::ingest_opencode_sessions;
 
 use super::pending_stamp;
 use super::HarnessAdapter;
@@ -36,20 +35,11 @@ fn opencode_sessions_dir() -> PathBuf {
         .join("session")
 }
 
-/// Box-pin the SDK's `async fn ingest_opencode_sessions` into a fn
-/// pointer the [`pending_stamp::SessionIngestor`] type alias accepts.
-fn opencode_ingest<'a>(
-    ledger: &'a mut RawLedger,
-    opts: &'a RawIngestOptions,
-) -> Pin<Box<dyn Future<Output = anyhow::Result<IngestReport>> + Send + 'a>> {
-    Box::pin(ingest_opencode_sessions(ledger, opts))
-}
-
 /// Hand out a `&'static dyn HarnessAdapter` for opencode. The registry
 /// calls this once at lazy-init time. See
 /// [`pending_stamp::session_store_adapter`] for the leak semantics.
 pub fn adapter() -> &'static dyn HarnessAdapter {
-    pending_stamp::session_store_adapter("opencode", opencode_sessions_dir, opencode_ingest)
+    pending_stamp::session_store_adapter("opencode", opencode_sessions_dir, ingest_opencode_sessions)
 }
 
 #[cfg(test)]

crates/relayburn-cli/src/harnesses/pending_stamp.rs

@@ -131,15 +131,13 @@ pub fn adapter_static(config: PendingStampAdapter) -> &'static dyn HarnessAdapte
     Box::leak(Box::new(PendingStampAdapterImpl::new(config)))
 }
 
-/// Async fn pointer for an SDK session ingestor (`ingest_codex_sessions`,
-/// `ingest_opencode_sessions`). The shape matches both per-harness ingest
-/// passes verbatim — they live in `relayburn_sdk` as `async fn`, and the
-/// per-call `Box::pin` adaptation happens at the call site so the helper
+/// Fn pointer for an SDK session ingestor (`ingest_codex_sessions`,
+/// `ingest_opencode_sessions`). Both verbs are sync in the SDK; the
+/// per-tick `Box::pin` adaptation that drops them into [`IngestSessionsFn`]
+/// happens at the call site in [`session_store_adapter`] so the helper
 /// stays a fn pointer (no per-tick closure allocation).
-pub type SessionIngestor = for<'a> fn(
-    &'a mut RawLedger,
-    &'a RawIngestOptions,
-) -> Pin<Box<dyn Future<Output = anyhow::Result<IngestReport>> + Send + 'a>>;
+pub type SessionIngestor =
+    fn(&mut RawLedger, &RawIngestOptions) -> anyhow::Result<IngestReport>;
 
 /// One-call factory for pending-stamp adapters whose only differences are
 /// the harness name, the session-root resolver, and which SDK ingest pass
@@ -174,7 +172,7 @@ pub fn session_store_adapter(
                 ledger_home,
                 ..RawIngestOptions::default()
             };
-            ingestor(handle.raw_mut(), &opts).await
+            ingestor(handle.raw_mut(), &opts)
         })
     });
     adapter_static(PendingStampAdapter::new(name, session_root, ingest_sessions))

crates/relayburn-sdk-node/src/lib.rs

@@ -1261,8 +1261,12 @@ pub async fn ingest(opts: Option<IngestOptions>) -> Result<IngestReport, NapiErr
         on_progress: None,
         on_warn: None,
     };
-    let report = sdk::ingest(raw)
+    // SDK ingest is sync (filesystem walks + rusqlite writes). Run it on
+    // tokio's blocking pool so the napi runtime stays responsive while the
+    // sweep is in flight.
+    let report = tokio::task::spawn_blocking(move || sdk::ingest(raw))
         .await
+        .map_err(|e| NapiError::from_reason(format!("ingest task panicked: {e}")))?
         .map_err(|e| NapiError::from_reason(format!("{e:#}")))?;
     Ok(report.into())
 }