fix(flows): mint fresh orchestration token so long steps don't expire the result-fetch JWT (#10415)

* fix(flows): mint fresh orchestration token so long steps don't expire the result-fetch JWT

A flow step's ephemeral JWT is minted at step pull time with a lifetime of
SCRIPT_TOKEN_EXPIRY (900s on cloud) and reused to drive post-completion flow
orchestration — including the next step's input-transform isolated-eval, which
fetches prior steps' results (e.g. `[...results.x]`). If the step whose
completion triggers that fetch ran longer than the token's lifetime minus the
60s JWT leeway (~16min on cloud), the reused token is already expired and the
fetch is rejected as anonymous:

    Failed to fetch results for step 'x':
    Bad request: As a non logged in user, you can only see jobs ran by anonymous users

This surfaces as an intermittent, hard-to-diagnose failure of long-running
flows (per-step duration, not total flow duration).

Mint a fresh token for flow-step completions so the orchestration client's
lifetime is independent of how long the finished step ran (falls back to the
step token on error).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* address review: derive end-user label, guard mint on step staleness, trim comment

- Derive the token label the same way as create_token
  (ephemeral-script-end-user-{created_by} when permissioned_as differs from
  created_by) so run-on-behalf-of flows keep the end-user override that
  username_override_from_label relies on, instead of hardcoding "ephemeral-script".
- Only mint the fresh token when the finished step could actually have expired it
  (duration >= SCRIPT_TOKEN_EXPIRY/2), so the common short-step path keeps the
  pull-time token and avoids an extra get_job_perms query per completion.
- Add warn_after_seconds(5) on the mint, matching create_token.
- Trim the comment to the durable invariant and drop the internal ticket id
  (comment + log line) per AGENTS.md.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

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

backend/windmill-worker/src/result_processor.rs

@@ -689,8 +689,52 @@ pub async fn handle_receive_completed_job(
     killpill_rx: &tokio::sync::broadcast::Receiver<()>,
     #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
 ) -> Option<Arc<MiniPulledJob>> {
-    let token = jc.token.clone();
     let workspace = jc.job.workspace_id.clone();
+    // The client built here drives post-completion orchestration (the next step's input
+    // transforms fetch prior step results), which outlives the finished step. The step's own
+    // token has a `SCRIPT_TOKEN_EXPIRY` lifetime, so reusing it would fail that orchestration
+    // once the step itself ran longer than the token lives; refresh it when the step is old enough.
+    let token_maybe_expired = jc
+        .duration
+        .is_some_and(|d| d as u64 >= *windmill_common::worker::SCRIPT_TOKEN_EXPIRY * 1000 / 2);
+    let token = if jc.job.is_flow_step() && token_maybe_expired {
+        // Mirror `create_token`'s label so run-on-behalf-of flows keep their end-user override.
+        let label = if jc.job.permissioned_as != format!("u/{}", jc.job.created_by)
+            && jc.job.permissioned_as != jc.job.created_by
+        {
+            format!("ephemeral-script-end-user-{}", jc.job.created_by)
+        } else {
+            "ephemeral-script".to_string()
+        };
+        match windmill_common::auth::create_token_for_owner(
+            db,
+            &jc.job.workspace_id,
+            &jc.job.permissioned_as,
+            &label,
+            *windmill_common::worker::SCRIPT_TOKEN_EXPIRY,
+            &jc.job.permissioned_as_email,
+            &jc.job.id,
+            None,
+            Some(format!(
+                "job-span-{}",
+                jc.job.flow_innermost_root_job.unwrap_or(jc.job.id)
+            )),
+        )
+        .warn_after_seconds(5)
+        .await
+        {
+            Ok(t) => t,
+            Err(e) => {
+                tracing::warn!(
+                    "could not mint fresh flow-orchestration token for job {}, reusing step token: {e:#}",
+                    jc.job.id
+                );
+                jc.token.clone()
+            }
+        }
+    } else {
+        jc.token.clone()
+    };
     let client = AuthedClient::new(base_internal_url.to_string(), workspace, token, None);
     let job = jc.job.clone();
     let mem_peak = jc.mem_peak.clone();