feat: add native result streaming (#6242)

* feat: add stream output feature to SSE job updates Adds stream_output field to JobUpdate struct that extracts log lines starting with '[wm_stream]:' from job logs. Regular logs now exclude stream lines, which are captured separately for specialized handling. - Added stream_output: Option<String> field to JobUpdate struct - Created extract_stream_output_from_logs() function to filter stream lines - Modified get_job_update_data() to use stream extraction logic - SSE clients now receive both new_logs and stream_output in job updates Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com> * feat: rename stream_output to stream and handle newlines as \n - Renamed `stream_output` field to `stream` in JobUpdate struct - Updated extract_stream_output_from_logs to extract_stream_from_logs - Changed stream output to join with literal \n instead of actual newlines - Stream lines are properly excluded from regular new_logs 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Ruben Fiszel <rubenfiszel@users.noreply.github.com> * decision tree nits * push ee ref * push ee ref * fix: fix id renaming in apps * remove duplicate caching (#6285) * feat: migrate audit log ids to bigints (blocking migration for EE) * fix(mcp): add proper check for mcp routes (#6282) * add proper check for mcp routes * cleaner * apply to flow * fix add checks scopes --------- Co-authored-by: dieriba <dieriba.pro@gmail.com> * chore(main): release 1.514.0 (#6283) * chore(main): release 1.514.0 * Apply automatic changes --------- Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com> * fix: pin tokio to 1.46.1 and aws-sdks-ts * pin rustls to 0.23.29 + pin aws-sdk * chore(main): release 1.514.1 (#6288) * chore(main): release 1.514.1 * Apply automatic changes --------- Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com> * fix: improve docker logs collection in docker mode * support $res: string in form inputs of arrays * fix import nit * fix: fix DynSelect * nits * fix: resource-type-ts-parser (#6289) * fix: resource types as arg in typescript handle imported defined types * Update nix flake (#6291) * merge * Small UI fixes (#6294) * fix step history not refreshing with staticInputs * fix array of obj not showing up in json editor in test this step * datatable scales correctly in DisplayResult and scrolling is much more usable * avoid next button disapearing and changing layout / hurting ux * nits * fix bug when renaming module A to B then module C to A, C takes the schema of A * fix bug with comments in sql repl * fix aggrid theme randomly not loading * bindable script * better delete button in db manager * property select doesnt exist * fix all warnings * delete $flowStateStore[id] on delete * feat(cli): generate cursor rules on init (#6270) * create cursor rules on init * change gen * add missing resource-type command * add resource type command in guidance * add schema option * revert * nit * nit * add flow guidance * nit * chore(main): release 1.515.0 (#6292) * chore(main): release 1.515.0 * Apply automatic changes --------- Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com> * fix: improved logs for script * nits logs * chore(main): release 1.515.1 (#6295) * chore(main): release 1.515.1 * Apply automatic changes --------- Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com> * merge * even more indexer tracings * add more tracing logs * feat: prevent too large results (>500Mb) from OOMing database * nit naming * feat: add CA certificate update at startup via environment variable (#6280) * feat: add CA certificate update at startup via environment variable Add support for running 'update-ca-certificates' at binary startup when RUN_UPDATE_CA_CERTIFICATE_AT_START environment variable is set to "true". - Check for RUN_UPDATE_CA_CERTIFICATE_AT_START env var on startup - Execute update-ca-certificates command if env var is set to "true" - Log success/failure appropriately with tracing - Continue startup even if CA certificate update fails - Non-blocking implementation with proper error handling Fixes #6279 Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com> * refactor: extract CA certificate update logic into separate function Extract the CA certificate update logic from windmill_main() into a dedicated update_ca_certificates_if_requested() function for better code organization and maintainability. Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com> * improvements --------- Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com> Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com> Co-authored-by: Alexander Petric <alpetric@users.noreply.github.com> Co-authored-by: Alexander Petric <alex@windmill.dev> * fix: indexer collection of job logs before indexing (#6300) * Add flume as dependecy for indexer * Update ee-repo-ref * Remove flags from cargo.toml * Update ee-repo-ref * Update ee-repo-ref * fix rust sdk build error (#6305) Signed-off-by: pyranota <pyra@duck.com> * fix broken audit logs filter (#6304) * rename to from to * goto fix * default to false if field not present operator settings (#6301) * git sync UI improvements (#6303) * ui improvements round 1 * modal cleanup * init * UI refactor * UI cleanup + refactor * legacy cleanup * success model -> github actions, non-ee warnings * sqlx * npm check * ee warning everywhere * last comments * formatting * no hardcoded theme * claude review improvemenets * fix: no process relative imports for scripts with codebase * fix: sqs oidc authentication disconnect #6307 * handle metadata for new scripts happen after commit * handle_deployment_metadata in a task * nits * chore: add windmill-utils-internal package (#6299) * add utils package * naming * cleaning * add docs * remove log * use autogenerated types * remove old * fix * cleaning * add docs * chore(main): release 1.516.0 (#6298) * chore(main): release 1.516.0 * Apply automatic changes --------- Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com> * merge * indexer improvements * upgrade tantivy to 0.24.2 * use tantivy fork * nit warnings * fix oss build * improve indexer * chore: use windmill-utils-internal for cli (#6297) * add utils package * naming * cleaning * simplify assignPath * rename old files * same for locks * create on confirm * default true * use replaceinlinescripts from utils * use extractscriptfromflows * make it compile * cleaning * use argsigtojson * fix * fix missing await * cleaner * cleaning * cleaning * use in frontend * add docs * testing * remove log * use autogenerated types * remove old * fix * cleaning * adapt usage * draft * better build script * fix build * revert to default creation * add docs * remove and rename * make everything work * add await * only if not installed * add vs code setting * add to publish action * fix bc * safer use of sep * fix * do not rename on push * no publish on release * use published package on frontend * nit * Add dependencies to run sqlx prepare to nix flake (#6309) * feat(cli): wmill-lock.yaml v2 for easier git merge diffs * merge * merge * all * all * rm warnings * fix styling on aichatinput (#6312) * fix: use with_capacity back presusre for tantivy directory multipart writes (#6313) * use with capacity for tantivy directory multi part uploads * Update ee repo ref * Update ee-repo-ref * Update ee-repo-ref * chore(main): release 1.517.0 (#6310) * chore(main): release 1.517.0 * Apply automatic changes --------- Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com> * fix typo on cli build (#6314) * cleanup * feat(utils): add flow.yaml validation function (#6316) * add validateflow function * cleaner code * preprocess json * cleaning * create specific package * cleaning * add tests * fix: cleanup concurrency_counter automatically + remove orphans keys automatically * fix: add disabled support to resource picker in schema forms * fix: add wm_labels to tracing spans * all * merge * all * fix: delete empty git connection (#6318) * fix checks * bun handling * all * all? * all * all * update * all * update * check * fix history * all * all * all * Remove leftover debug tracing statements - Remove commented debug trace in jobs.rs for stream output - Remove commented debug trace in result_stream.rs for stream processing Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com> * fix test * all * handle iter * fix --------- Signed-off-by: pyranota <pyra@duck.com> Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com> Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com> Co-authored-by: Ruben Fiszel <ruben@windmill.dev> Co-authored-by: centdix <40307056+centdix@users.noreply.github.com> Co-authored-by: dieriba <dieriba.pro@gmail.com> Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com> Co-authored-by: wendrul <53628737+wendrul@users.noreply.github.com> Co-authored-by: Diego Imbert <70353967+diegoimbert@users.noreply.github.com> Co-authored-by: Alexander Petric <alpetric@users.noreply.github.com> Co-authored-by: Alexander Petric <alex@windmill.dev> Co-authored-by: pyranota <92104930+pyranota@users.noreply.github.com>
2026-10-25 00:03:52 +00:00 · 2025-08-06 22:40:42 +00:00
parent 138a2a6379
commit 161dbbc6d6
63 changed files with 979 additions and 515 deletions
@@ -1,18 +1,24 @@
-use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use uuid::Uuid;
-use std::str::FromStr;
-use regex::Regex;
-use serde_json::Value;
 use crate::auth::OptTokened;
 use crate::db::{ApiAuthed, DB};
-use crate::jobs::{cancel_suspended_job, resume_suspended_job, QueryApprover, QueryOrBody, ResumeUrls, get_resume_urls_internal};
-use axum::{extract::{Path, Query}, Extension};
-use windmill_common::error::Error;
+use crate::jobs::{
+    cancel_suspended_job, get_resume_urls_internal, resume_suspended_job, QueryApprover,
+    QueryOrBody, ResumeUrls,
+};
+use axum::{
+    extract::{Path, Query},
+    Extension,
+};
+use regex::Regex;
+use serde::{Deserialize, Serialize};
+use serde_json::value::RawValue;
+use serde_json::Value;
+use std::collections::HashMap;
+use std::str::FromStr;
+use uuid::Uuid;
 use windmill_common::cache;
+use windmill_common::error::Error;
 use windmill_common::jobs::JobKind;
 use windmill_common::scripts::ScriptHash;
-use serde_json::value::RawValue;

 #[derive(Debug, Deserialize, Serialize)]
 pub struct ResumeSchema {
@@ -234,7 +240,7 @@ pub async fn get_approval_form_details(
    .ok_or_else(|| Error::BadRequest("This workflow is no longer running and has either already timed out or been cancelled or completed.".to_string()))
    .map(|r| (r.job_kind, r.script_hash, r.raw_flow, r.parent_job, r.created_at, r.created_by, r.script_path, r.args))?;

-    let flow_data = match cache::job::fetch_flow(&db, job_kind, script_hash).await {
+    let flow_data = match cache::job::fetch_flow(&db, &job_kind, script_hash).await {
        Ok(data) => data,
        Err(_) => {
            if let Some(parent_job_id) = parent_job_id.as_ref() {
@@ -900,11 +900,10 @@ impl<'a> GetQuery<'a> {
    /// when pushed from an un-updated workers.
    /// This function is used to make the above change transparent for the API, as the returned jobs
    /// will have the raw values as if they were still in the tables.
-    async fn resolve_raw_values<T>(
+    async fn resolve_raw_values<T: JobCommon>(
        &self,
        db: &DB,
        id: Uuid,
-        kind: JobKind,
        hash: Option<ScriptHash>,
        job: &mut JobExtended<T>,
    ) {
@@ -917,18 +916,18 @@ impl<'a> GetQuery<'a> {
            // Try to fetch the flow from the cache, fallback to the preview flow.
            // NOTE: This could check for the job kinds instead of the `or_else` but it's not
            // necessary as `fetch_flow` return early if the job kind is not a preview one.
-            cache::job::fetch_flow(db, kind, hash)
+            cache::job::fetch_flow(db, job.job_kind(), hash)
                .or_else(|_| cache::job::fetch_preview_flow(db, &id, raw_flow))
                .await
                .ok()
                .inspect(|data| job.raw_flow = Some(sqlx::types::Json(data.raw_flow.clone())));
        }
-        if self.with_code {
+        if self.with_code && job.job_kind() == &JobKind::Preview {
            // Try to fetch the code from the cache, fallback to the preview code.
            // NOTE: This could check for the job kinds instead of the `or_else` but it's not
            // necessary as `fetch_script` return early if the job kind is not a preview one.
            let conn = Connection::from(db.clone());
-            cache::job::fetch_script(db.clone(), kind, hash)
+            cache::job::fetch_script(db.clone(), job.job_kind(), hash)
                .or_else(|_| cache::job::fetch_preview_script(&conn, &id, raw_lock, raw_code))
                .await
                .ok()
@@ -958,7 +957,7 @@ impl<'a> GetQuery<'a> {

        self.check_auth(job.as_ref().map(|job| job.created_by.as_str()))?;
        if let Some(job) = job.as_mut() {
-            self.resolve_raw_values(&db, job.id, job.job_kind, job.script_hash, job)
+            self.resolve_raw_values(&db, job.id, job.script_hash, job)
                .await;
        }
        if self.with_flow {
@@ -992,7 +991,7 @@ impl<'a> GetQuery<'a> {

        self.check_auth(cjob.as_ref().map(|job| job.created_by.as_str()))?;
        if let Some(job) = cjob.as_mut() {
-            self.resolve_raw_values(db, job.id, job.job_kind, job.script_hash, job)
+            self.resolve_raw_values(db, job.id, job.script_hash, job)
                .await;
        }

@@ -2642,7 +2641,7 @@ pub async fn get_resume_urls_internal(
 }

 #[derive(sqlx::FromRow, Debug, Serialize)]
-pub struct JobExtended<T> {
+pub struct JobExtended<T: JobCommon> {
    #[sqlx(flatten)]
    #[serde(flatten)]
    inner: T,
@@ -2665,7 +2664,23 @@ pub struct JobExtended<T> {
    pub aggregate_wait_time_ms: Option<i64>,
 }

-impl<T> JobExtended<T> {
+pub trait JobCommon {
+    fn job_kind(&self) -> &JobKind;
+}
+
+impl JobCommon for QueuedJob {
+    fn job_kind(&self) -> &JobKind {
+        &self.job_kind
+    }
+}
+
+impl JobCommon for CompletedJob {
+    fn job_kind(&self) -> &JobKind {
+        &self.job_kind
+    }
+}
+
+impl<T: JobCommon> JobExtended<T> {
    pub fn new(
        self_wait_time_ms: Option<i64>,
        aggregate_wait_time_ms: Option<i64>,
@@ -2683,7 +2698,7 @@ impl<T> JobExtended<T> {
    }
 }

-impl<T> Deref for JobExtended<T> {
+impl<T: JobCommon> Deref for JobExtended<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
@@ -2691,7 +2706,7 @@ impl<T> Deref for JobExtended<T> {
    }
 }

-impl<T> DerefMut for JobExtended<T> {
+impl<T: JobCommon> DerefMut for JobExtended<T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.inner
    }
@@ -4538,8 +4553,13 @@ pub async fn run_wait_result_script_by_path_internal(
    check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;

    let mut tx = user_db.clone().begin(&authed).await?;
-    let (job_payload, tag, delete_after_use, timeout, on_behalf_of) =
-        script_path_to_payload(script_path.to_path(), &mut *tx, &w_id, run_query.skip_preprocessor).await?;
+    let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = script_path_to_payload(
+        script_path.to_path(),
+        &mut *tx,
+        &w_id,
+        run_query.skip_preprocessor,
+    )
+    .await?;
    drop(tx);

    let tag = run_query.tag.clone().or(tag);
@@ -5680,22 +5700,38 @@ pub async fn run_job_by_hash_inner(
 pub struct JobUpdateQuery {
    pub running: Option<bool>,
    pub log_offset: Option<i32>,
+    pub stream_offset: Option<i32>,
    pub get_progress: Option<bool>,
+    pub no_logs: Option<bool>,
    pub only_result: Option<bool>,
    pub fast: Option<bool>,
 }

 #[derive(Serialize, Debug)]
 pub struct JobUpdate {
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub running: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub completed: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub new_logs: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub new_result_stream: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub log_offset: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stream_offset: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub mem_peak: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub progress: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub flow_status: Option<Box<serde_json::value::RawValue>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub workflow_as_code_status: Option<Box<serde_json::value::RawValue>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub job: Option<Job>,
+    #[serde(skip_serializing_if = "Option::is_none")]
    pub only_result: Option<Box<serde_json::value::RawValue>>,
 }

@@ -5714,6 +5750,7 @@ impl Hash for JobUpdate {
        self.log_offset.hash(state);
        self.mem_peak.hash(state);
        self.progress.hash(state);
+        self.stream_offset.hash(state);
        if !self.completed.unwrap_or(false) {
            self.flow_status.as_ref().map(|x| x.get().hash(state));
            self.workflow_as_code_status
@@ -5779,7 +5816,7 @@ async fn get_job_update(
    opt_tokened: OptTokened,
    Extension(db): Extension<DB>,
    Path((w_id, job_id)): Path<(String, Uuid)>,
-    Query(JobUpdateQuery { log_offset, get_progress, running, only_result, .. }): Query<
+    Query(JobUpdateQuery { log_offset, stream_offset, get_progress, running, only_result, no_logs, .. }): Query<
        JobUpdateQuery,
    >,
 ) -> JsonResult<JobUpdate> {
@@ -5791,11 +5828,13 @@ async fn get_job_update(
            &w_id,
            &job_id,
            log_offset,
+            stream_offset,
            get_progress,
            running,
            true,
            false,
            only_result,
+            no_logs,
        )
        .await?,
    ))
@@ -5806,7 +5845,7 @@ async fn get_job_update_sse(
    opt_tokened: OptTokened,
    Extension(db): Extension<DB>,
    Path((w_id, job_id)): Path<(String, Uuid)>,
-    Query(JobUpdateQuery { log_offset, get_progress, running, only_result, fast }): Query<
+    Query(JobUpdateQuery { log_offset, stream_offset, get_progress, running, no_logs, only_result, fast }): Query<
        JobUpdateQuery,
    >,
 ) -> Response {
@@ -5817,10 +5856,12 @@ async fn get_job_update_sse(
        w_id,
        job_id,
        log_offset,
+        stream_offset,
        get_progress,
        running,
        only_result,
        fast,
+        no_logs,
    )
    .map(|x| {
        format!(
@@ -5857,19 +5898,24 @@ fn get_job_update_sse_stream(
    w_id: String,
    job_id: Uuid,
    initial_log_offset: Option<i32>,
+    initial_stream_offset: Option<i32>,
    get_progress: Option<bool>,
    running: Option<bool>,
    only_result: Option<bool>,
    fast: Option<bool>,
+    no_logs: Option<bool>,
 ) -> impl futures::Stream<Item = JobUpdateSSEStream> {
    let (tx, rx) = tokio::sync::mpsc::channel(32);

    tokio::spawn(async move {
        let mut log_offset = initial_log_offset;
+        let mut stream_offset = initial_stream_offset;
        let mut last_update_hash: Option<String> = None;

        // Send initial update immediately
        let mut running = running;
+        let mut mem_peak = 0;
+
        match get_job_update_data(
            &opt_authed,
            &opt_tokened,
@@ -5877,22 +5923,38 @@ fn get_job_update_sse_stream(
            &w_id,
            &job_id,
            log_offset,
+            stream_offset,
            get_progress,
            running,
            true,
            true,
            only_result,
+            no_logs,
        )
        .await
        {
-            Ok(update) => {
+            Ok(mut update) => {
                last_update_hash = Some(update.hash_str());
                let completion_sent = update.completed.unwrap_or(false);
                if running.is_some() && update.running.is_some_and(|x| x) {
                    running = Some(true);
                }
+                if let Some(new_mem_peak) = update.mem_peak {
+                    mem_peak = new_mem_peak;
+                }
                if let Some(new_offset) = update.log_offset {
-                    log_offset = Some(new_offset);
+                    if new_offset != log_offset.unwrap_or(0) {
+                        log_offset = Some(new_offset);
+                    } else {
+                        update.log_offset = None;
+                    }
+                }
+                if let Some(new_stream_offset) = update.stream_offset {
+                    if new_stream_offset != stream_offset.unwrap_or(0) {
+                        stream_offset = Some(new_stream_offset);
+                    } else {
+                        update.stream_offset = None;
+                    }
                }
                if tx.send(JobUpdateSSEStream::Update(update)).await.is_err() {
                    tracing::warn!("Failed to send initial job update for job {job_id}");
@@ -5918,6 +5980,7 @@ fn get_job_update_sse_stream(
        let mut i = 0;
        let start = Instant::now();
        let mut last_ping = Instant::now();
+
        loop {
            i += 1;
            let ms_duration = if i > 100 || !fast.unwrap_or(false) {
@@ -5950,18 +6013,27 @@ fn get_job_update_sse_stream(
                &w_id,
                &job_id,
                log_offset,
+                stream_offset,
                get_progress,
                running,
                false,
                true,
                only_result,
+                no_logs,
            )
            .await
            {
-                Ok(update) => {
+                Ok(mut update) => {
                    if running.is_some() && update.running.is_some_and(|x| x) {
                        running = Some(true);
                    }
+                    if update.completed.is_some_and(|x| !x) {
+                        update.completed = None;
+                    }
+                    if update.new_logs.as_ref().is_some_and(|x| x.is_empty()) {
+                        update.new_logs = None;
+                    }
+
                    // if !only_result.unwrap_or(false) {
                    //     tracing::error!("update {:?}", update);
                    // }
@@ -5970,7 +6042,25 @@ fn get_job_update_sse_stream(
                    if last_update_hash.as_ref() != Some(&update_last_status) {
                        // Update log offset if available
                        if let Some(new_offset) = update.log_offset {
-                            log_offset = Some(new_offset);
+                            if new_offset != log_offset.unwrap_or(0) {
+                                log_offset = Some(new_offset);
+                            } else {
+                                update.log_offset = None;
+                            }
+                        }
+                        if let Some(new_stream_offset) = update.stream_offset {
+                            if new_stream_offset != stream_offset.unwrap_or(0) {
+                                stream_offset = Some(new_stream_offset);
+                            } else {
+                                update.stream_offset = None;
+                            }
+                        }
+                        if let Some(new_mem_peak) = update.mem_peak {
+                            if new_mem_peak != mem_peak {
+                                mem_peak = new_mem_peak;
+                            } else {
+                                update.mem_peak = None;
+                            }
                        }
                        let completed = update.completed.unwrap_or(false);
                        if tx.send(JobUpdateSSEStream::Update(update)).await.is_err() {
@@ -5983,7 +6073,8 @@ fn get_job_update_sse_stream(
                        last_update_hash = Some(update_last_status);
                    }
                }
-                Err(_) => {
+                Err(e) => {
+                    tracing::error!("Error getting job update: {:?}", e);
                    if tx.send(JobUpdateSSEStream::NotFound).await.is_err() {
                        tracing::warn!("Failed to send job not found for job {job_id}");
                    }
@@ -6003,11 +6094,13 @@ async fn get_job_update_data(
    w_id: &str,
    job_id: &Uuid,
    log_offset: Option<i32>,
+    stream_offset: Option<i32>,
    get_progress: Option<bool>,
    running: Option<bool>,
    log_view: bool,
    get_full_job_on_completion: bool,
    only_result: Option<bool>,
+    no_logs: Option<bool>,
 ) -> error::Result<JobUpdate> {
    let tags = if log_view {
        log_job_view(
@@ -6028,19 +6121,22 @@ async fn get_job_update_data(

    if only_result.unwrap_or(false) {
        let result = if let Some(tags) = tags {
-            let r = sqlx::query!(
-                "SELECT result as \"result: sqlx::types::Json<Box<RawValue>>\", v2_job.tag,
-                v2_job_queue.running as \"running: Option<bool>\"
+            let r = 
+                sqlx::query!(
+                    "SELECT result as \"result: sqlx::types::Json<Box<RawValue>>\", v2_job.tag,
+                v2_job_queue.running as \"running: Option<bool>\", SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\", CHAR_LENGTH(rs.stream) AS stream_offset
                FROM v2_job
                LEFT JOIN v2_job_queue USING (id)
                LEFT JOIN v2_job_completed USING (id)
+                LEFT JOIN job_result_stream rs ON rs.job_id = $2
                WHERE v2_job.id = $2 AND v2_job.workspace_id = $1",
-                w_id,
-                job_id,
-            )
-            .fetch_optional(db)
-            .await?
-            .ok_or_else(|| Error::NotFound(format!("Job not found: {}", job_id)))?;
+                    w_id,
+                    job_id,
+                    stream_offset.unwrap_or(0),
+                )
+                .fetch_optional(db)
+                .await?
+                .ok_or_else(|| Error::NotFound(format!("Job not found: {}", job_id)))?;

            if !tags.contains(&r.tag.as_str()) {
                return Err(Error::NotAuthorized(format!(
@@ -6050,27 +6146,44 @@ async fn get_job_update_data(
                )));
            }
            let running = r.running.as_ref().map(|x| *x);
-            (r.result.map(|x| x.0), running)
+            (r.result.map(|x| x.0), running, r.result_stream.flatten(), r.stream_offset)
        } else {
            if running.is_some_and(|x| !x) {
                let r = sqlx::query!(
-                    "SELECT result as \"result: sqlx::types::Json<Box<RawValue>>\", v2_job_queue.running as \"running: Option<bool>\" FROM v2_job_completed FULL OUTER JOIN v2_job_queue USING (id) WHERE (v2_job_queue.id = $1 AND v2_job_queue.workspace_id = $2) OR (v2_job_completed.id = $1 AND v2_job_completed.workspace_id = $2)",
+                    "SELECT 
+                        result as \"result: sqlx::types::Json<Box<RawValue>>\",
+                        v2_job_queue.running as \"running: Option<bool>\",
+                        SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",
+                        CHAR_LENGTH(rs.stream) + 1 AS stream_offset
+                    FROM v2_job_completed FULL OUTER JOIN v2_job_queue USING (id) 
+                    LEFT JOIN job_result_stream rs ON rs.job_id = $1
+                    WHERE (v2_job_queue.id = $1 AND v2_job_queue.workspace_id = $2) OR (v2_job_completed.id = $1 AND v2_job_completed.workspace_id = $2)",
                    job_id,
                    w_id,
+                    stream_offset.unwrap_or(0),
                ).fetch_optional(db).await?;
                if let Some(r) = r {
                    let running = r.running.as_ref().map(|x| *x);
-                    (r.result.map(|x| x.0), running)
+                    (r.result.map(|x| x.0), running, r.result_stream.flatten(), r.stream_offset)
                } else {
-                    (None, None)
+                    (None, None, None, None)
                }
            } else {
-                (sqlx::query_scalar!(
-                    "SELECT result as \"result: sqlx::types::Json<Box<RawValue>>\" FROM v2_job_completed WHERE id = $2 AND workspace_id = $1",
+                let q =  sqlx::query!(
+                    "SELECT result as \"result: sqlx::types::Json<Box<RawValue>>\", SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\", CHAR_LENGTH(rs.stream) + 1 AS stream_offset
+                 FROM v2_job_completed  FULL OUTER JOIN job_result_stream rs ON rs.job_id = v2_job_completed.id WHERE (v2_job_completed.id = $2 AND v2_job_completed.workspace_id = $1 OR rs.workspace_id = $1)",
                    w_id,
                    job_id,
-                ).fetch_optional(db).await?.flatten()
-                .map(|x| x.0), running)
+                    stream_offset.unwrap_or(0),
+                )
+                .fetch_optional(db)
+                .await?;
+                tracing::error!("q {:?}", q);
+                if let Some(r) = q {
+                    (r.result.map(|x| x.0), running, r.result_stream.flatten(), r.stream_offset)
+                } else {
+                    (None, None, None, None)
+                }
            }
        };
        Ok(JobUpdate {
@@ -6078,6 +6191,8 @@ async fn get_job_update_data(
            completed: if result.0.is_some() { Some(true) } else { None },
            log_offset: None,
            new_logs: None,
+            new_result_stream: result.2,
+            stream_offset: result.3,
            mem_peak: None,
            progress: None,
            job: None,
@@ -6093,20 +6208,24 @@ async fn get_job_update_data(
                    WHEN q.id IS NOT NULL THEN (CASE WHEN NOT $5 AND q.running THEN true ELSE null END)
                    ELSE false
                END AS running,
-                SUBSTR(logs, GREATEST($1 - log_offset, 0)) AS logs,
+                CASE WHEN $7::BOOLEAN THEN NULL ELSE SUBSTR(logs, GREATEST($1 - log_offset, 0)) END AS logs,
+                SUBSTR(rs.stream, $8) AS new_result_stream,
                COALESCE(r.memory_peak, c.memory_peak) AS mem_peak,
                COALESCE(c.flow_status, f.flow_status) AS \"flow_status: sqlx::types::Json<Box<RawValue>>\",
                COALESCE(c.workflow_as_code_status, f.workflow_as_code_status) AS \"workflow_as_code_status: sqlx::types::Json<Box<RawValue>>\",
-                job_logs.log_offset + CHAR_LENGTH(job_logs.logs) + 1 AS log_offset,
+                CASE WHEN $7::BOOLEAN THEN NULL ELSE job_logs.log_offset + CHAR_LENGTH(job_logs.logs) + 1 END AS log_offset,
+                CHAR_LENGTH(rs.stream) + 1 AS stream_offset,
                created_by AS \"created_by!\",
                CASE WHEN $4::BOOLEAN THEN (
                    SELECT scalar_int FROM job_stats WHERE job_id = $3 AND metric_id = 'progress_perc'
-                ) END AS progress
+                ) END AS progress,
+                rs.stream AS \"result_stream: Option<String>\"
            FROM v2_job j
                LEFT JOIN v2_job_queue q USING (id)
                LEFT JOIN v2_job_runtime r USING (id)
                LEFT JOIN v2_job_status f USING (id)
                LEFT JOIN v2_job_completed c USING (id)
+                LEFT JOIN job_result_stream rs ON rs.job_id = $3
                LEFT JOIN job_logs ON job_logs.job_id =  $3
            WHERE j.workspace_id = $2 AND j.id = $3
            AND ($6::text[] IS NULL OR j.tag = ANY($6))",
@@ -6116,6 +6235,8 @@ async fn get_job_update_data(
            get_progress.unwrap_or(false),
            running,
            tags.as_ref().map(|v| v.as_slice()) as Option<&[&str]>,
+            no_logs.unwrap_or(false),
+            stream_offset.unwrap_or(0),
        )
        .fetch_optional(db)
        .await?
@@ -6139,6 +6260,8 @@ async fn get_job_update_data(
            completed: record.completed,
            log_offset: record.log_offset,
            new_logs: record.logs,
+            new_result_stream: record.new_result_stream,
+            stream_offset: record.stream_offset,
            mem_peak: record.mem_peak,
            progress: record.progress,
            workflow_as_code_status: record