Method run_stream

crates/libsy/src/core/algorithm.rs:650–705

Process a request to completion, returning a stream of [`Step`]s. The consumer must fulfill every [`Step::CallLlm`] before the algorithm can continue. The bounded step channel applies backpressure when the consumer is not polling. A successful run ends with [`Step::ReturnToAgent`]; a failure is emitted as an `Err` item. Dropping the stream aborts the spawned algorithm task. Every invocation owns

(
        self: Arc<Self>,
        ctx: Context,
        request: Request,
        observer: Option<RunObserver>,
    )

    /// Every invocation owns a separate [`Driver`]. `observer`, when present, receives
    /// each completed model call and, after a successful routed run, its routing overhead.
    fn run_stream(
        self: Arc<Self>,
        ctx: Context,
        request: Request,
        observer: Option<RunObserver>,
    ) -> StepStream {
        // Stamp the algorithm's telemetry label into the request context; the
        // context rides on every driver call, so its telemetry is attributed.
        let mut ctx = ctx;
        ctx.values.insert(
            observability::ALGORITHM_KEY.to_string(),
            self.name().to_string(),
        );
        let driver = Driver::with_observer(observer);
        let task_driver = driver.clone();
        let task_ctx = ctx.clone();
        let stream = task_driver.stream();
        // One `libsy.run` span covers the whole algorithm task; the driver's
        // `libsy.llm_call` spans and decision logs nest inside it via `tracing`'s
        // contextual parenting.
        let span = observability::run_span(self.name(), &request);
        let observed_driver = task_driver.clone();
        let handle = tokio::spawn(
            async move {
                observability::observe_run(
                    task_ctx.clone(),
                    observed_driver,
                    self.create_run_task(task_ctx, task_driver, request),
                )
                .await
            }
            .instrument(span),
        );
        // Dropping the stream aborts the algorithm task when its consumer goes away.
        let abort_guard = AbortOnDrop(handle.abort_handle());

        let finish_driver = driver.clone();
        let finish_ctx = ctx;
        let tail: StepStream = Box::pin(
            futures::stream::once(async move {
                let result = match handle.await {
                    Ok(response) => response,
                    Err(source) => Err(LibsyError::AlgorithmTask { source }),
                };
                finish_driver.finish(finish_ctx, result).await
            })
            .filter_map(|finish_result| async move { finish_result.err().map(Err) }),
        );

        let stream: StepStream = Box::pin(stream);
        Box::pin(futures::stream::select(stream, tail).map(move |step| {
            // link abort guard to stream
            let _keep_alive = &abort_guard;
            step
        }))
    }

    /// Process a request to completion, returning the final [`Response`] and the trace of