Function run_source_runtime 

pub async fn run_source_runtime<D, S>(
    driver: &mut D,
    sink: &S,
    pipeline: &Pipeline,
    apply_opts: &ApplyOpts,
    runtime_opts: &SourceRuntimeOpts,
) -> Result<RuntimeExit>
where
    D: SourceDriver,
    S: SurrealSink,

Incremental loop over a SourceDriver.

Shared apply order per batch: transform → ordered write → watermark (note_sunk_events → advance_watermark → policy → optional persist_checkpoint, sink-safe only). Between cycles: between_events → hooks (ad-hoc receives AdhocApply). Stops on is_finished (after drain), driver stop_reason, or SourceRuntimeOpts cancel/deadline.