fix: show scheduled singlestepflow runs in flow history sidebar (#10312)

* fix: show scheduled singlestepflow runs in flow input history

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: only match flow-wrapped singlestepflow rows in flow history

singlestepflow wraps either a script or a flow; a script and flow may share
a runnable_path, so filter flow history to flow-wrapped rows via the wrapped
module type. Extends the regression test to cover the same-path collision.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: surface scheduled singlestepflow runs in script history too

Scheduled scripts with a dynamic-skip handler or native retry also run as
singlestepflow. Include that kind for ScriptPath history, filtered to
script-wrapped rows so a same-path flow run does not leak in. ScriptHash is
untouched (these wrappers carry no runnable_id). Test covers both directions.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/windmill-api-inputs/src/lib.rs

@@ -73,14 +73,6 @@ impl Display for RunnableType {
 }
 
 impl RunnableType {
-    fn job_kind(&self) -> JobKind {
-        match self {
-            RunnableType::ScriptHash => JobKind::Script,
-            RunnableType::ScriptPath => JobKind::Script,
-            RunnableType::FlowPath => JobKind::Flow,
-        }
-    }
-
     fn column_name(&self) -> &'static str {
         match self {
             RunnableType::ScriptHash => "runnable_id",
@@ -159,6 +151,28 @@ async fn get_input_history(
         "AND parent_job IS NULL"
     };
 
+    // A scheduled runnable with a dynamic-skip handler (or a scheduled script with
+    // native retry) is wrapped in a synthetic `singlestepflow`, so its runs land under
+    // that kind rather than `flow`/`script` (see windmill-queue schedule.rs). Such a
+    // wrapper holds either a script or a flow, and the two may share a runnable_path, so
+    // match only the wrapped kind that belongs to the queried runnable. The wrapped type
+    // lives in raw_flow.modules[id in ('a','main')].value.type (mirrors the projection in
+    // windmill-api jobs.rs). runnable_id is NULL on these rows, so ScriptHash never matches.
+    let singlestepflow_filter = match r.runnable_type {
+        RunnableType::FlowPath => {
+            "AND (kind <> 'singlestepflow' OR EXISTS (\
+                SELECT 1 FROM jsonb_array_elements(v2_job.raw_flow->'modules') m \
+                WHERE m->>'id' IN ('a', 'main') AND m->'value'->>'type' = 'flow'))"
+        }
+        RunnableType::ScriptPath => {
+            "AND (kind <> 'singlestepflow' OR EXISTS (\
+                SELECT 1 FROM jsonb_array_elements(v2_job.raw_flow->'modules') m \
+                WHERE m->>'id' IN ('a', 'main') \
+                  AND COALESCE(m->'value'->>'type', 'script') <> 'flow'))"
+        }
+        RunnableType::ScriptHash => "",
+    };
+
     // Two-step approach: first fetch 2*(per_page+offset) rows using created_at ordering
     // (which leverages the ix_job_root_job_index_by_path_2 index on v2_job), then sort
     // the small result set by completed_at. This works because created_at and completed_at
@@ -172,7 +186,7 @@ async fn get_input_history(
             kind IN ('preview', 'flowpreview') as is_preview \
             FROM v2_job JOIN v2_job_completed USING (id) \
             WHERE v2_job.workspace_id = $3 AND {} = $1 AND kind = any($2) \
-            AND v2_job.script_entrypoint_override IS NULL \
+            AND v2_job.script_entrypoint_override IS NULL {singlestepflow_filter} \
             {args_query} AND v2_job_completed.status != 'skipped' {include_non_root} \
             ORDER BY v2_job.created_at DESC LIMIT $4\
         ) t ORDER BY completed_at DESC LIMIT $5 OFFSET $6",
@@ -186,14 +200,32 @@ async fn get_input_history(
         _ => query.bind(&r.runnable_id),
     };
 
-    let job_kinds = match r.runnable_type.job_kind() {
-        kind @ JobKind::Script if g.include_preview.unwrap_or(false) => {
-            vec![kind, JobKind::Preview]
+    // Include SingleStepFlow so scheduled runs surface (see `singlestepflow_filter`
+    // above, which restricts it to the wrapped kind matching the runnable). ScriptHash
+    // is omitted: those wrappers carry no runnable_id, so it can never match.
+    let include_preview = g.include_preview.unwrap_or(false);
+    let job_kinds = match r.runnable_type {
+        RunnableType::ScriptHash => {
+            let mut kinds = vec![JobKind::Script];
+            if include_preview {
+                kinds.push(JobKind::Preview);
+            }
+            kinds
         }
-        kind @ JobKind::Flow if g.include_preview.unwrap_or(false) => {
-            vec![kind, JobKind::FlowPreview]
+        RunnableType::ScriptPath => {
+            let mut kinds = vec![JobKind::Script, JobKind::SingleStepFlow];
+            if include_preview {
+                kinds.push(JobKind::Preview);
+            }
+            kinds
+        }
+        RunnableType::FlowPath => {
+            let mut kinds = vec![JobKind::Flow, JobKind::SingleStepFlow];
+            if include_preview {
+                kinds.push(JobKind::FlowPreview);
+            }
+            kinds
         }
-        kind => vec![kind],
     };
 
     let rows = query

backend/windmill-api-integration-tests/tests/inputs.rs

@@ -74,3 +74,94 @@ async fn test_inputs_endpoints(db: Pool<Postgres>) -> anyhow::Result<()> {
 
     Ok(())
 }
+
+// A scheduled runnable with a dynamic-skip handler (or a scheduled script with native
+// retry) runs as a `singlestepflow`, not `flow`/`script` (see windmill-queue schedule.rs).
+// Both a flow's and a script's history must surface their own singlestepflow runs, but a
+// script and flow may share a path, so each side must match only the wrapped kind that
+// belongs to it — the other must not leak in.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_input_history_singlestepflow_flow_vs_script(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    // Both rows share runnable_path 'f/test/scheduled'.
+    let flow_job = insert_singlestepflow(&db, "flow").await?;
+    let script_job = insert_singlestepflow(&db, "script").await?;
+
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let base = format!("http://localhost:{port}/api/w/test-workspace/inputs");
+
+    let history_ids = |runnable_type: &'static str| {
+        let base = base.clone();
+        async move {
+            let resp = authed(client().get(format!(
+                "{base}/history?runnable_id=f/test/scheduled&runnable_type={runnable_type}"
+            )))
+            .send()
+            .await?;
+            let status = resp.status().as_u16();
+            let body = resp.text().await?;
+            assert_2xx(status, &body, "GET /inputs/history");
+            let inputs: Vec<serde_json::Value> = serde_json::from_str(&body)?;
+            anyhow::Ok(
+                inputs
+                    .iter()
+                    .filter_map(|i| i.get("id").and_then(|v| v.as_str()).map(String::from))
+                    .collect::<std::collections::HashSet<_>>(),
+            )
+        }
+    };
+
+    let flow_hist = history_ids("FlowPath").await?;
+    assert!(
+        flow_hist.contains(&flow_job.to_string()),
+        "flow-wrapped singlestepflow missing from flow history: {flow_hist:?}",
+    );
+    assert!(
+        !flow_hist.contains(&script_job.to_string()),
+        "script-wrapped singlestepflow leaked into flow history: {flow_hist:?}",
+    );
+
+    let script_hist = history_ids("ScriptPath").await?;
+    assert!(
+        script_hist.contains(&script_job.to_string()),
+        "script-wrapped singlestepflow missing from script history: {script_hist:?}",
+    );
+    assert!(
+        !script_hist.contains(&flow_job.to_string()),
+        "flow-wrapped singlestepflow leaked into script history: {script_hist:?}",
+    );
+
+    Ok(())
+}
+
+// Insert a completed root singlestepflow at path f/test/scheduled wrapping `wrapped_type`
+// ('flow' or 'script') as its single module — mirrors the schedule.rs wrapper shape.
+async fn insert_singlestepflow(
+    db: &Pool<Postgres>,
+    wrapped_type: &str,
+) -> anyhow::Result<uuid::Uuid> {
+    let id = uuid::Uuid::new_v4();
+    let raw_flow = json!({ "modules": [{ "id": "a", "value": { "type": wrapped_type } }] });
+    sqlx::query(
+        "INSERT INTO v2_job (id, workspace_id, tag, created_by, permissioned_as, \
+         permissioned_as_email, kind, runnable_path, raw_flow, same_worker, visible_to_owner) \
+         VALUES ($1, 'test-workspace', 'flow', 'test-user', 'u/test-user', \
+         'test@windmill.dev', 'singlestepflow', 'f/test/scheduled', $2, false, true)",
+    )
+    .bind(id)
+    .bind(sqlx::types::Json(&raw_flow))
+    .execute(db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO v2_job_completed (id, workspace_id, duration_ms, deleted, status) \
+         VALUES ($1, 'test-workspace', 1, false, 'success')",
+    )
+    .bind(id)
+    .execute(db)
+    .await?;
+    Ok(id)
+}