feat(pipelines): asset freshness — fresh/stale badge (CE) + watchdog (EE) (#9909)

* feat(pipelines): passive asset freshness tracking on the graph Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(pipelines): drop dead freshness-enforcement stub, document query ordering Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(pipelines): freshness watchdog (EE) — auto re-run stale producers Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): watchdog review fixes — archived workspaces, badge kind parity, scan index Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): CI review — no singlestepflow in freshness, +N parity, completion-time fallback Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): CI review — history completedAt, freshness/asset trigger UI metadata Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore: update ee-repo-ref to 6f5fe0f7f56696fbef5a8349da38496c32e71666 This commit updates the EE repository reference after PR #643 was merged in windmill-ee-private. Previous ee-repo-ref: 1f13380354bf591ae25a2c20d36917534bcc5459 New ee-repo-ref: 6f5fe0f7f56696fbef5a8349da38496c32e71666 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
2026-08-18 16:02:10 +00:00 · 2026-07-04 04:23:44 +00:00
parent af36498432
commit 5d7fb6deca
50 changed files with 998 additions and 59 deletions
@@ -35,7 +35,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM pipeline_freshness_state WHERE workspace_id = $1 AND script_path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0e22295e40bfc173d85f0a962d491f3afd87eeea5e4e8d935b1b86eb02f061e7"
+}
@@ -39,7 +39,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -77,7 +78,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -128,7 +128,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT p.path AS \"path!\",\n               (SELECT c.completed_at\n                  FROM v2_job j\n                  JOIN v2_job_completed c ON c.id = j.id\n                 WHERE j.workspace_id = $1\n                   AND j.runnable_path = p.path\n                   AND j.parent_job IS NULL\n                   -- No 'singlestepflow': flows may share a script's path, and\n                   -- a same-path flow run must not read as the script being\n                   -- fresh (false-fresh). Script retries land as native\n                   -- 'script' jobs; only the rare flow-wrapper fallback is\n                   -- missed, which errs stale. Kept in lockstep with the\n                   -- freshness watchdog's queries (freshness_watchdog_ee).\n                   AND j.kind IN ('script', 'preview')\n                   AND c.status = 'success'\n                 ORDER BY j.created_at DESC\n                 LIMIT 1) AS last_success_at\n          FROM unnest($2::text[]) AS p(path)\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "last_success_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      null,
+      null
+    ]
+  },
+  "hash": "3ea82f16050318a0b3b87c12beb66b49254036ca93f7a6415ceb4e316ed64eed"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT EXISTS(\n            SELECT 1 FROM v2_job_queue q\n            JOIN v2_job j ON j.id = q.id\n            WHERE q.workspace_id = $1\n              AND j.runnable_path = $2\n              AND j.parent_job IS NULL\n              AND j.kind IN ('script', 'preview')\n              AND (q.running = true OR q.scheduled_for <= now())\n        ) AS \"in_flight!\"\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "in_flight!",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "45a539e61a3c35098fb6d59c3d5de8b6fcebe83002c74647666e3a6f5b8490dd"
+}
@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        UPDATE pipeline_freshness_state\n        SET attempts = attempts + 1,\n            last_push_at = now(),\n            next_attempt_at = now()\n                + (LEAST($3::bigint, $4::bigint * (1::bigint << LEAST(attempts + 1, 20)))::text\n                   || ' seconds')::interval\n        WHERE workspace_id = $1 AND script_path = $2 AND next_attempt_at <= now()\n        ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "45c73bd8742601753ad947c88b65ff3ae182ff7c0631d4fa44b67b11a133498f"
+}
@@ -80,7 +80,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT EXISTS(\n            SELECT 1 FROM v2_job j\n            JOIN v2_job_completed c ON c.id = j.id\n            WHERE j.workspace_id = $1\n              AND j.runnable_path = $2\n              AND j.parent_job IS NULL\n              AND j.kind IN ('script', 'preview')\n              AND c.status = 'success'\n              AND c.completed_at > now() - ($3::bigint::text || ' seconds')::interval\n        ) AS \"fresh!\"\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "fresh!",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "572b80f7b16fd5d1a05f8fe9bbb07667fc1b5e802dc2465ffd13178a042e4f89"
+}
@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            INSERT INTO pipeline_freshness_state\n                (workspace_id, script_path, attempts, last_push_at, next_attempt_at)\n            VALUES ($1, $2, 1, now(),\n                    now() + (LEAST($3::bigint, $4::bigint * 2)::text || ' seconds')::interval)\n            ON CONFLICT (workspace_id, script_path) DO NOTHING\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5e0569092c559a1b19f636fae68f6d740a1720f7746c01f9ea693c70271ff55c"
+}
@@ -161,7 +161,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE pipeline_freshness_state SET next_attempt_at = now() - interval '1 second'\n         WHERE workspace_id = $1 AND script_path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6d81570849980f316b08487a03f6017386cf2767c953e0d152e7983a00218c6f"
+}
@@ -0,0 +1,32 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT DISTINCT ON (workspace_id, path)\n               workspace_id AS \"workspace_id!\", path AS \"path!\", content AS \"content!\"\n        FROM script\n        WHERE auto_kind = 'pipeline'\n          AND archived = false\n          AND deleted = false\n          AND content ILIKE '%freshness%'\n          -- Workspace archival stops all execution but leaves script rows\n          -- intact for unarchival; without this the watchdog would keep\n          -- resurrecting runs in a workspace the admin shut down.\n          AND EXISTS (SELECT 1 FROM workspace w\n                      WHERE w.id = script.workspace_id AND w.deleted = false)\n        ORDER BY workspace_id, path, created_at DESC\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "path!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "content!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "7cd0b27f4b4a20b05ac876861b934128c3a6ade024d6662379857905a7447a2a"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        DELETE FROM pipeline_freshness_state s\n        WHERE NOT EXISTS (\n            SELECT 1 FROM unnest($1::text[], $2::text[]) AS w(workspace_id, script_path)\n            WHERE w.workspace_id = s.workspace_id AND w.script_path = s.script_path\n        )\n        ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "TextArray",
+        "TextArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "81d1c56c369c5328d947f1ad21da995a8a555e4a9c2d1387dd8cee9c81efd40b"
+}
@@ -35,7 +35,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -191,7 +191,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT attempts FROM pipeline_freshness_state WHERE workspace_id = $1 AND script_path = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "attempts",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "b10e21ec9edf2d22ff956fdcd085bd601b7dcabdf70e85543455cb4fc3986d74"
+}
@@ -166,7 +166,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO pipeline_freshness_state (workspace_id, script_path, attempts) VALUES ($1, $2, 3)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "bc6f566bc958923045dc3e1ff1639a26f649c960a4ee8d0a584884a236129d43"
+}
@@ -80,7 +80,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -111,7 +111,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO pipeline_freshness_state (workspace_id, script_path) VALUES ($1, $2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d1a1f64cb6b14c561c53c44910cad8bc7a2d226a9f22418782bcc66e46d0f706"
+}
@@ -111,7 +111,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -251,7 +251,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -191,7 +191,8 @@
                "ci_test",
                "github",
                "azure",
-                "asset"
+                "asset",
+                "freshness"
              ]
            }
          }
@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT runnable_path AS \"runnable_path!\", created_by AS \"created_by!\",\n                  args AS \"args: sqlx::types::Json<serde_json::Value>\"\n             FROM v2_job\n             WHERE workspace_id = $1 AND trigger_kind = 'freshness'\n             ORDER BY created_at",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "runnable_path!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "created_by!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "args: sqlx::types::Json<serde_json::Value>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      false,
+      true
+    ]
+  },
+  "hash": "e50bce4bba01c90171973862bcfa088b5c0c1ad601146ff1c57d4a3ddc30556a"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COUNT(*) AS \"count!\" FROM pipeline_freshness_state WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count!",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "f26cde14890cc754e04341264c4dba01f2f3a02bb36a8d7c20e1990dc81e0472"
+}
@@ -1 +1 @@
-81d8a449effdc540a6e8810668dd5d4aea0c485a
+6f5fe0f7f56696fbef5a8349da38496c32e71666
@@ -0,0 +1,5 @@
+-- Postgres has no ALTER TYPE ... DROP VALUE for enums. The 'freshness' value
+-- stays even on rollback, consistent with prior job_trigger_kind additions
+-- (see 20260510174213_asset_trigger_dispatch).
+DROP INDEX IF EXISTS idx_script_pipeline_freshness_scan;
+DROP TABLE IF EXISTS pipeline_freshness_state;
@@ -0,0 +1,35 @@
+-- Attribution for runs pushed by the pipeline freshness watchdog (the EE
+-- background loop that re-runs a `// freshness`-annotated producer whose
+-- output aged past its window).
+ALTER TYPE job_trigger_kind ADD VALUE IF NOT EXISTS 'freshness';
+
+-- Per-(workspace, script) watchdog state: exponential-backoff bookkeeping so
+-- a persistently failing producer isn't re-pushed on every scan tick, and an
+-- atomic claim so concurrent servers can't double-push in the same tick
+-- (claim = the UPDATE/INSERT that advances next_attempt_at; only the winner
+-- pushes). Rows exist only while a script is stale — observing it fresh (or
+-- its annotation gone) deletes the row, resetting the backoff.
+CREATE TABLE pipeline_freshness_state (
+    workspace_id VARCHAR(50) NOT NULL REFERENCES workspace (id) ON DELETE CASCADE,
+    script_path VARCHAR(510) NOT NULL,
+    attempts INTEGER NOT NULL DEFAULT 1,
+    last_push_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (workspace_id, script_path)
+);
+
+-- Written only by the server monitor loop on the raw (non-RLS) pool, but
+-- granted like every other app table so a future user-transaction reader
+-- doesn't hit the recurring missing-GRANT class of bug.
+GRANT ALL ON pipeline_freshness_state TO windmill_user;
+GRANT ALL ON pipeline_freshness_state TO windmill_admin;
+
+-- The watchdog's ~60s candidate scan (latest deployed pipeline members)
+-- filters on this exact predicate and orders by (workspace_id, path,
+-- created_at DESC); without a matching partial index it seq-scans the whole
+-- script-version heap on every tick, on instances that mostly have zero
+-- pipeline scripts. (idx_script_pipeline_path is text_pattern_ops for
+-- prefix LIKE — it can't serve this ordering.)
+CREATE INDEX idx_script_pipeline_freshness_scan
+    ON script (workspace_id, path, created_at DESC)
+    WHERE auto_kind = 'pipeline' AND archived = false AND deleted = false;
@@ -77,10 +77,12 @@ pub struct ParseAssetsOutput {
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pub partition: Option<PartitionSpec>,
    // `// freshness <duration>` — SLA stating outputs must be at most
-    // `duration` old. Active backstop: when no other trigger has fired the
-    // script within the window, a watchdog re-runs it. Distinct from
-    // schedule (which is producer cadence); freshness is consumer SLA and
-    // applies regardless of which trigger last fired.
+    // `duration` old. Drives passive monitoring in CE (the asset graph
+    // colors the node's badge fresh/stale against its last successful run)
+    // and the enterprise watchdog (windmill-queue `freshness_watchdog`),
+    // which re-runs a stale unpartitioned producer. Distinct from schedule
+    // (which is producer cadence); freshness is consumer SLA and applies
+    // regardless of which trigger last fired.
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pub freshness: Option<FreshnessSpec>,
    // `// trigger all` → AND join barrier; default (`any`) = OR (current
@@ -637,7 +639,7 @@ fn parse_kv_opts(s: &str) -> BTreeMap<String, String> {
 //   - `on <trigger-spec>`        → asset / native trigger edge (including
 //                                  the marker-only `on schedule` form)
 //   - `partitioned <kind> [opts]` → partition declaration
-//   - `freshness <duration>`     → SLA / active backstop
+//   - `freshness <duration>`     → SLA window (badge + EE watchdog)
 //   - `tag <name>`               → worker-tag override (annotation wins
 //                                  over UI-set value at deploy)
 //   - `retry <count> [<delay>]`  → cascade-only retry policy
@@ -178,6 +178,14 @@ lazy_static::lazy_static! {
    .and_then(|x| x.parse::<bool>().ok())
    .unwrap_or(false);

+    // Ops kill switch for the pipeline freshness watchdog (a background
+    // pusher — being able to stop it without a redeploy matters more than
+    // for read-only monitors).
+    pub static ref DISABLE_FRESHNESS_WATCHDOG: bool = std::env::var("DISABLE_FRESHNESS_WATCHDOG")
+    .ok()
+    .and_then(|x| x.parse::<bool>().ok())
+    .unwrap_or(false);
+
    pub static ref WORKERS_NAMES: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(Vec::new()));

    static ref QUEUE_COUNT_TAGS: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(Vec::new()));
@@ -2994,6 +3002,28 @@ pub async fn monitor_db(
        }
    };

+    // run every ~60s (2 iterations * 30s). Enterprise feature: the active
+    // `// freshness` backstop lives in windmill-queue's `freshness_watchdog`
+    // (`private`); OSS gets a no-op stub. Runtime-gated on an Enterprise
+    // license like the audit export above. Safe on concurrent servers — the
+    // watchdog claims per-script state rows atomically before pushing.
+    let pipeline_freshness_watchdog_f = async {
+        if server_mode
+            && !*DISABLE_FRESHNESS_WATCHDOG
+            && iteration.is_some()
+            && iteration.as_ref().unwrap().should_run(2)
+        {
+            if let Some(db) = conn.as_sql() {
+                if matches!(
+                    windmill_common::ee_oss::get_license_plan().await,
+                    windmill_common::ee_oss::LicensePlan::Enterprise
+                ) {
+                    windmill_queue::freshness_watchdog::tick(db).await;
+                }
+            }
+        }
+    };
+
    join!(
        expired_items_f,
        zombie_jobs_f,
@@ -3021,6 +3051,7 @@ pub async fn monitor_db(
        manage_audit_partitions_f,
        export_audit_logs_to_object_store_f,
        cleanup_scheduled_job_deletions_f,
+        pipeline_freshness_watchdog_f,
    );
 }

@@ -0,0 +1,294 @@
+//! End-to-end tests for the pipeline freshness watchdog (Enterprise).
+//!
+//! `windmill_queue::freshness_watchdog::tick` is called directly against
+//! seeded `script` / `v2_job(_completed)` rows — no worker or API server is
+//! needed, since the watchdog's job ends at the push (the pushed job sitting
+//! in `v2_job_queue` is itself part of the assertions). Covers: staleness on
+//! never-ran and aged-out members, the fresh short-circuit + state reset,
+//! the in-flight suppression, the backoff claim, and the skip rules
+//! (partitioned, malformed window, non-pipeline scripts).
+
+#![cfg(feature = "private")]
+
+use sqlx::{Pool, Postgres};
+use windmill_queue::freshness_watchdog::tick;
+use windmill_test_utils::initialize_tracing;
+
+const WS: &str = "test-workspace";
+const PATH: &str = "u/test-user/freshness_producer";
+
+/// Seed a deployed pipeline-member script. Mirrors the deploy path's output:
+/// `auto_kind = 'pipeline'`, empty (non-NULL) lock so run-by-path resolution
+/// treats it as deployed, hash derived from path+content for uniqueness.
+async fn seed_pipeline_script(
+    db: &Pool<Postgres>,
+    path: &str,
+    content: &str,
+) -> anyhow::Result<()> {
+    let mut h = 0i64;
+    for b in path.bytes().chain(content.bytes()) {
+        h = h.wrapping_mul(31).wrapping_add(b as i64);
+    }
+    sqlx::query(
+        r#"INSERT INTO script (workspace_id, hash, path, summary, description, content,
+                                created_by, language, tag, lock, auto_kind)
+           VALUES ($1, $2, $3, '', '', $4, 'test-user', 'bash'::script_lang, 'bash', '', 'pipeline')
+           ON CONFLICT DO NOTHING"#,
+    )
+    .bind(WS)
+    .bind(h)
+    .bind(path)
+    .bind(content)
+    .execute(db)
+    .await?;
+    // Process-global deployed-script caches are keyed by (workspace, path) /
+    // (workspace, hash) and would leak between #[sqlx::test] isolated DBs
+    // that reuse both — resolve everything from this test's own DB.
+    windmill_common::DEPLOYED_SCRIPT_CACHE_DISABLED
+        .store(true, std::sync::atomic::Ordering::Relaxed);
+    Ok(())
+}
+
+/// Seed a completed root run of `path` that finished `age_s` seconds ago.
+async fn seed_completed_run(
+    db: &Pool<Postgres>,
+    path: &str,
+    age_s: i64,
+    success: bool,
+) -> anyhow::Result<()> {
+    let id = uuid::Uuid::new_v4();
+    sqlx::query(
+        r#"INSERT INTO v2_job (id, workspace_id, runnable_path, kind, created_at,
+                                created_by, permissioned_as, permissioned_as_email, tag)
+           VALUES ($1, $2, $3, 'script'::job_kind,
+                   now() - ($4::bigint::text || ' seconds')::interval,
+                   'test-user', 'u/test-user', 'test@windmill.dev', 'bash')"#,
+    )
+    .bind(id)
+    .bind(WS)
+    .bind(path)
+    .bind(age_s)
+    .execute(db)
+    .await?;
+    sqlx::query(
+        r#"INSERT INTO v2_job_completed (id, workspace_id, duration_ms, status, started_at, completed_at)
+           VALUES ($1, $2, 0, CASE WHEN $3 THEN 'success'::job_status ELSE 'failure'::job_status END,
+                   now() - ($4::bigint::text || ' seconds')::interval,
+                   now() - ($4::bigint::text || ' seconds')::interval)"#,
+    )
+    .bind(id)
+    .bind(WS)
+    .bind(success)
+    .bind(age_s)
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+/// Jobs the watchdog pushed: (path, created_by, args) rows attributed to
+/// `trigger_kind = 'freshness'`.
+async fn fetch_pushed(
+    db: &Pool<Postgres>,
+) -> anyhow::Result<Vec<(String, String, Option<serde_json::Value>)>> {
+    let rows = sqlx::query!(
+        r#"SELECT runnable_path AS "runnable_path!", created_by AS "created_by!",
+                  args AS "args: sqlx::types::Json<serde_json::Value>"
+             FROM v2_job
+             WHERE workspace_id = $1 AND trigger_kind = 'freshness'
+             ORDER BY created_at"#,
+        WS,
+    )
+    .fetch_all(db)
+    .await?;
+    Ok(rows
+        .into_iter()
+        .map(|r| (r.runnable_path, r.created_by, r.args.map(|a| a.0)))
+        .collect())
+}
+
+async fn state_row(db: &Pool<Postgres>, path: &str) -> anyhow::Result<Option<i32>> {
+    Ok(sqlx::query_scalar!(
+        "SELECT attempts FROM pipeline_freshness_state WHERE workspace_id = $1 AND script_path = $2",
+        WS,
+        path,
+    )
+    .fetch_optional(db)
+    .await?)
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn never_ran_member_is_pushed_once(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 30s\necho hi\n").await?;
+
+    tick(&db).await;
+
+    let pushed = fetch_pushed(&db).await?;
+    assert_eq!(pushed.len(), 1, "one watchdog push expected");
+    let (path, created_by, args) = &pushed[0];
+    assert_eq!(path, PATH);
+    assert_eq!(created_by, &format!("freshness-{PATH}"));
+    let args = args.as_ref().expect("args recorded");
+    assert_eq!(
+        args.get("_wmill_skip_asset_dispatch"),
+        Some(&serde_json::json!(true)),
+        "watchdog runs must not re-fire the cascade"
+    );
+    assert_eq!(
+        args.pointer("/trigger/kind"),
+        Some(&serde_json::json!("freshness"))
+    );
+    assert_eq!(state_row(&db, PATH).await?, Some(1), "claim row recorded");
+
+    // Second tick: the pushed job is queued-and-due, so the in-flight guard
+    // suppresses a duplicate regardless of backoff.
+    tick(&db).await;
+    assert_eq!(
+        fetch_pushed(&db).await?.len(),
+        1,
+        "no duplicate while queued"
+    );
+
+    // Simulate the queued job vanishing without a completion: the backoff
+    // claim (next_attempt_at in the future) now carries the suppression.
+    sqlx::query!("DELETE FROM v2_job_queue WHERE workspace_id = $1", WS)
+        .execute(&db)
+        .await?;
+    tick(&db).await;
+    assert_eq!(fetch_pushed(&db).await?.len(), 1, "backoff holds the retry");
+
+    // Force the backoff window open: the watchdog retries and escalates.
+    sqlx::query!(
+        "UPDATE pipeline_freshness_state SET next_attempt_at = now() - interval '1 second'
+         WHERE workspace_id = $1 AND script_path = $2",
+        WS,
+        PATH,
+    )
+    .execute(&db)
+    .await?;
+    tick(&db).await;
+    assert_eq!(fetch_pushed(&db).await?.len(), 2, "due retry pushed");
+    assert_eq!(state_row(&db, PATH).await?, Some(2), "attempts escalated");
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn fresh_member_is_skipped_and_state_reset(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 1h\necho hi\n").await?;
+    seed_completed_run(&db, PATH, 10, true).await?;
+    // Leftover backoff row from an earlier staleness episode.
+    sqlx::query!(
+        "INSERT INTO pipeline_freshness_state (workspace_id, script_path, attempts) VALUES ($1, $2, 3)",
+        WS,
+        PATH,
+    )
+    .execute(&db)
+    .await?;
+
+    tick(&db).await;
+
+    assert!(
+        fetch_pushed(&db).await?.is_empty(),
+        "fresh member not pushed"
+    );
+    assert_eq!(state_row(&db, PATH).await?, None, "backoff reset on fresh");
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn aged_out_member_is_pushed(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 1h\necho hi\n").await?;
+    // Old success outside the window + a recent failure: still stale.
+    seed_completed_run(&db, PATH, 7200, true).await?;
+    seed_completed_run(&db, PATH, 60, false).await?;
+
+    tick(&db).await;
+
+    assert_eq!(fetch_pushed(&db).await?.len(), 1, "aged-out member pushed");
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn partitioned_malformed_and_plain_members_are_skipped(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    // Partitioned: freshness means partition-gap detection, out of scope.
+    seed_pipeline_script(
+        &db,
+        "u/test-user/partitioned",
+        "# pipeline\n# partitioned daily\n# freshness 1h\necho hi\n",
+    )
+    .await?;
+    // Malformed window: fails safe to unwatched.
+    seed_pipeline_script(
+        &db,
+        "u/test-user/malformed",
+        "# pipeline\n# freshness soonish\necho hi\n",
+    )
+    .await?;
+    // Freshness only in prose (parser must reject; ILIKE prefilter passes).
+    seed_pipeline_script(
+        &db,
+        "u/test-user/prose",
+        "# pipeline\n# ensure freshness of data below\necho hi\n",
+    )
+    .await?;
+
+    tick(&db).await;
+
+    assert!(fetch_pushed(&db).await?.is_empty(), "no member is watched");
+    let rows = sqlx::query_scalar!(
+        r#"SELECT COUNT(*) AS "count!" FROM pipeline_freshness_state WHERE workspace_id = $1"#,
+        WS,
+    )
+    .fetch_one(&db)
+    .await?;
+    assert_eq!(rows, 0, "no state rows for unwatched members");
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn archived_workspace_is_not_resurrected(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    // Workspace archival stops all execution but keeps script rows for
+    // unarchival — the watchdog must not keep pushing runs there.
+    seed_pipeline_script(&db, PATH, "# pipeline\n# freshness 30s\necho hi\n").await?;
+    sqlx::query!("UPDATE workspace SET deleted = true WHERE id = $1", WS)
+        .execute(&db)
+        .await?;
+
+    tick(&db).await;
+
+    assert!(
+        fetch_pushed(&db).await?.is_empty(),
+        "no pushes into an archived workspace"
+    );
+    assert_eq!(
+        state_row(&db, PATH).await?,
+        None,
+        "no state bookkeeping either"
+    );
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn state_of_unwatched_member_is_cleaned_up(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    // A stale-bookkeeping row whose script no longer declares freshness
+    // (e.g. annotation removed and redeployed) must not survive the sweep.
+    sqlx::query!(
+        "INSERT INTO pipeline_freshness_state (workspace_id, script_path) VALUES ($1, $2)",
+        WS,
+        "u/test-user/gone",
+    )
+    .execute(&db)
+    .await?;
+
+    tick(&db).await;
+
+    assert_eq!(state_row(&db, "u/test-user/gone").await?, None);
+    Ok(())
+}
@@ -632,6 +632,14 @@ struct GraphRunnableNode {
    partition_kind: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none", default)]
    freshness: Option<String>,
+    // Completion time of the most recently started successful run of this
+    // pipeline member. The canvas checks it against the `// freshness` window
+    // to color the badge fresh/stale. The badge itself is passive; on EE the
+    // freshness watchdog (windmill-queue) separately re-runs stale
+    // unpartitioned producers. Absent when no successful run is visible to
+    // the caller (job RLS applies).
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    last_success_at: Option<chrono::DateTime<chrono::Utc>>,
    #[serde(skip_serializing_if = "Option::is_none", default)]
    tag: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none", default)]
@@ -920,6 +928,46 @@ async fn asset_graph(
    .fetch_all(&mut *tx)
    .await?;

+    // Newest successful completed run per pipeline member, for the passive
+    // freshness status on the canvas. Correlated per-path lookup walks
+    // ix_job_root_job_index_by_path_2 newest-first until the first success,
+    // so cost is bounded by the member count, not run history. "Newest" is
+    // by created_at (the index order), not completed_at: with overlapping
+    // runs of one path this can pick an earlier completion, erring toward
+    // stale — never toward false-fresh. Inside the user tx so job-visibility
+    // RLS applies — a caller who can't see the runs gets no timestamp rather
+    // than leaked completion times.
+    let member_paths: Vec<String> = pipeline_member_paths
+        .iter()
+        .map(|r| r.path.clone())
+        .collect();
+    let last_success_rows = sqlx::query!(
+        r#"
+        SELECT p.path AS "path!",
+               (SELECT c.completed_at
+                  FROM v2_job j
+                  JOIN v2_job_completed c ON c.id = j.id
+                 WHERE j.workspace_id = $1
+                   AND j.runnable_path = p.path
+                   AND j.parent_job IS NULL
+                   -- No 'singlestepflow': flows may share a script's path, and
+                   -- a same-path flow run must not read as the script being
+                   -- fresh (false-fresh). Script retries land as native
+                   -- 'script' jobs; only the rare flow-wrapper fallback is
+                   -- missed, which errs stale. Kept in lockstep with the
+                   -- freshness watchdog's queries (freshness_watchdog_ee).
+                   AND j.kind IN ('script', 'preview')
+                   AND c.status = 'success'
+                 ORDER BY j.created_at DESC
+                 LIMIT 1) AS last_success_at
+          FROM unnest($2::text[]) AS p(path)
+        "#,
+        &w_id,
+        &member_paths,
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
    // Existing scripts / flows in the workspace. Used to filter out
    // orphan trigger rows whose `script_path` no longer resolves — those
    // would otherwise be added to `runnable_set` below and surface as
@@ -1010,6 +1058,11 @@ async fn asset_graph(
            (r.path.clone(), lineage)
        })
        .collect();
+    let last_success_by_path: std::collections::HashMap<String, chrono::DateTime<chrono::Utc>> =
+        last_success_rows
+            .into_iter()
+            .filter_map(|r| r.last_success_at.map(|t| (r.path, t)))
+            .collect();
    let pipeline_member_script_paths: std::collections::HashSet<String> =
        pipeline_member_paths.into_iter().map(|r| r.path).collect();
    let existing_script_paths: std::collections::HashSet<String> =
@@ -1215,6 +1268,10 @@ async fn asset_graph(
                freshness: ann
                    .and_then(|a| a.freshness.as_ref())
                    .map(|f| f.duration.clone()),
+                last_success_at: (usage_kind == AssetUsageKind::Script)
+                    .then(|| last_success_by_path.get(&path))
+                    .flatten()
+                    .copied(),
                tag: ann.and_then(|a| a.tag.clone()),
                retry: ann.and_then(|a| a.retry.clone()),
                data_tests: ann.map(|a| a.data_tests.clone()).unwrap_or_default(),
@@ -1252,16 +1252,6 @@ async fn create_script_internal<'c>(
    // membership; parsed writes tell us what is produced (we don't record
    // them in auto_kind itself).
    let pipeline_annotations = parse_pipeline_annotations(&ns.content);
-    // `// freshness` is parsed but enforcement is a not-yet-implemented
-    // enterprise feature (skeleton in windmill_common::pipeline_advanced).
-    // Surface a clear TODO at deploy rather than silently accepting an
-    // annotation that does nothing.
-    if pipeline_annotations.freshness.is_some() {
-        tracing::warn!(
-            "{}",
-            windmill_common::pipeline_advanced::freshness_enforcement_todo()
-        );
-    }
    // `// materialize` materializes a `ducklake://<name>/<table>` target from a
    // DuckDB script. These two constraints hold for *both* modes: a non-DuckLake
    // target would otherwise deploy, register a producer in the asset graph, then
@@ -25592,6 +25592,7 @@ components:
        - google
        - github
        - asset
+        - freshness

    TriggerMode:
      description: job trigger mode
@@ -352,6 +352,10 @@ mod debounce_duration_tests {
        assert_eq!(parse_duration_secs("5m"), Some(300));
        assert_eq!(parse_duration_secs("2h"), Some(7200));
        assert_eq!(parse_duration_secs(" 1d "), Some(86400));
+        // Explicit plus sign comes free with i64 parsing; the TS mirror
+        // (parseDurationSecs) matches it — keep the two in lockstep.
+        assert_eq!(parse_duration_secs("+5m"), Some(300));
+        assert_eq!(parse_duration_secs("+45"), Some(45));
    }

    #[test]
@@ -1,16 +1,12 @@
-//! OSS fallback: pipeline freshness/SLA enforcement and partition backfills
-//! are enterprise features; their implementations live in windmill-ee-private
-//! (see `pipeline_advanced_ee`). In the public build the entry points report
-//! that the enterprise edition is required.
+//! OSS fallback: pipeline partition backfills are an enterprise feature;
+//! their implementations live in windmill-ee-private (see
+//! `pipeline_advanced_ee`). In the public build the entry points report that
+//! the enterprise edition is required. (Freshness lives elsewhere: the
+//! fresh/stale badge is CE in the assets API, the active watchdog is
+//! windmill-queue's `freshness_watchdog`.)

 use crate::error::Error;

-pub fn freshness_enforcement_todo() -> Error {
-    Error::internal_err(
-        "Pipeline freshness/SLA enforcement requires the enterprise edition".to_string(),
-    )
-}
-
 pub fn backfill_todo() -> Error {
    Error::internal_err("Pipeline partition backfill requires the enterprise edition".to_string())
 }
@@ -0,0 +1,9 @@
+//! OSS fallback for the pipeline freshness watchdog. The active backstop —
+//! re-running a `// freshness`-annotated producer whose output aged past its
+//! window — is an enterprise feature (see `freshness_watchdog_ee`). In the
+//! public build the tick is a no-op; CE keeps the passive fresh/stale badge
+//! on the asset graph.
+
+use windmill_common::DB;
+
+pub async fn tick(_db: &DB) {}
@@ -14,6 +14,13 @@ pub mod cascade_oss;
 pub use cascade_ee as cascade;
 #[cfg(not(feature = "private"))]
 pub use cascade_oss as cascade;
+#[cfg(feature = "private")]
+pub mod freshness_watchdog_ee;
+pub mod freshness_watchdog_oss;
+#[cfg(feature = "private")]
+pub use freshness_watchdog_ee as freshness_watchdog;
+#[cfg(not(feature = "private"))]
+pub use freshness_watchdog_oss as freshness_watchdog;
 pub mod jobs;
 #[cfg(feature = "private")]
 pub mod jobs_ee;
@@ -46,6 +46,9 @@ pub enum JobTriggerKind {
    // A run dispatched because an upstream pipeline script wrote an asset
    // this runnable subscribes to via `// on s3://...` annotations.
    Asset,
+    // A run pushed by the pipeline freshness watchdog (EE) because the
+    // script's `// freshness` window elapsed without a successful run.
+    Freshness,
 }

 impl std::fmt::Display for JobTriggerKind {
@@ -68,6 +71,7 @@ impl std::fmt::Display for JobTriggerKind {
            JobTriggerKind::Github => "github",
            JobTriggerKind::CiTest => "ci_test",
            JobTriggerKind::Asset => "asset",
+            JobTriggerKind::Freshness => "freshness",
        };
        write!(f, "{}", kind)
    }
@@ -401,6 +401,7 @@
 					in_pipeline: r.in_pipeline ?? false,
 					partition_kind: r.partition_kind,
 					freshness: r.freshness,
+					last_success_at: r.last_success_at,
 					tag: r.tag,
 					retry: r.retry,
 					macros: r.macros,
@@ -22,12 +22,13 @@
 	import { preventDefault, stopPropagation } from 'svelte/legacy'
 	import type { GraphUsageKind } from './types'
 	import type { RunnableRunState } from './activeRunnables.svelte'
+	import { parseDurationSecs } from './parsePipelineAnnotations'
 	import { NODE } from '$lib/components/graph/util'
 	import DropdownV2 from '$lib/components/DropdownV2.svelte'
 	import Popover from '$lib/components/meltComponents/Popover.svelte'
 	import type { Item } from '$lib/utils'
 	import { workspaceStore } from '$lib/stores'
-	import { sendUserToast } from '$lib/utils'
+	import { sendUserToast, msToReadableTimeShort } from '$lib/utils'

 	interface Props {
 		data: {
@@ -36,6 +37,9 @@
 			in_pipeline?: boolean
 			partition_kind?: 'daily' | 'hourly' | 'weekly' | 'monthly' | 'dynamic'
 			freshness?: string
+			// Completion time (ISO) of the newest successful run visible to
+			// the caller. With `freshness`, drives the fresh/stale chip state.
+			last_success_at?: string
 			tag?: string
 			retry?: { count: number; delay?: string }
 			// Macros this script provides (deployed/drafted `// macros` library).
@@ -122,6 +126,44 @@
 		}
 	}

+	// Freshness verdict: newest successful run (server `last_success_at`,
+	// or a newer one the session poll observed) vs the `// freshness`
+	// window. No verdict (undefined) for drafts — no run history — and for
+	// unparseable windows; the chip then stays neutral like the other
+	// annotation chips.
+	let freshnessWindowS = $derived(data.freshness ? parseDurationSecs(data.freshness) : undefined)
+	// Ticks so a node crosses fresh→stale while the canvas stays open (the
+	// graph payload is static between refetches). Armed only when a verdict
+	// is rendered.
+	let nowMs = $state(Date.now())
+	$effect(() => {
+		if (freshnessWindowS === undefined || data.unsaved) return
+		const id = setInterval(() => (nowMs = Date.now()), 30_000)
+		return () => clearInterval(id)
+	})
+	let lastSuccessMs = $derived.by(() => {
+		const server = data.last_success_at ? new Date(data.last_success_at).getTime() : undefined
+		const polled = data.runState?.lastSuccessAt
+			? new Date(data.runState.lastSuccessAt).getTime()
+			: undefined
+		if (server === undefined) return polled
+		return polled === undefined ? server : Math.max(server, polled)
+	})
+	let freshnessState = $derived.by((): 'fresh' | 'stale' | undefined => {
+		if (freshnessWindowS === undefined || data.unsaved) return undefined
+		if (lastSuccessMs === undefined) return 'stale'
+		return nowMs - lastSuccessMs <= freshnessWindowS * 1000 ? 'fresh' : 'stale'
+	})
+	let freshnessTooltip = $derived.by(() => {
+		const base = `// freshness ${data.freshness}`
+		if (freshnessState === undefined) return base
+		if (lastSuccessMs === undefined) return `${base} — stale: no successful run yet`
+		const ago = msToReadableTimeShort(Math.max(0, nowMs - lastSuccessMs))
+		return freshnessState === 'fresh'
+			? `${base} — fresh: last successful run ${ago} ago`
+			: `${base} — stale: last successful run ${ago} ago`
+	})
+
 	// Cascade + bounded-run options live on the Run button's caret popover
 	// (whenever there's a cascade OR a bounded-run start — see `hasCaret`
 	// below), so the kebab menu stays focused on lifecycle actions only.
@@ -164,8 +206,8 @@
 		</span>
 		<!-- Annotation chips share one neutral treatment — the icon carries
 		     the meaning (colors are reserved for feedback, per the brand
-		     guidelines). Only the run-state chip below keeps semantic
-		     colors. -->
+		     guidelines). Only the freshness chip (when it has a verdict)
+		     and the run-state chip below use semantic colors. -->
 		{#if data.partition_kind}
 			<div
 				class="shrink-0 flex items-center gap-0.5 px-1 py-0.5 mr-1 rounded-sm bg-surface-secondary text-secondary"
@@ -175,10 +217,22 @@
 				<span class="text-3xs leading-none">{data.partition_kind}</span>
 			</div>
 		{/if}
+		<!-- Freshness is the one annotation chip that carries feedback (a
+		     fresh/stale verdict against real run history), so like the
+		     run-state chip it uses semantic colors: emerald = within window,
+		     amber = stale. Neutral when there's no verdict (drafts, bad
+		     window value). -->
 		{#if data.freshness}
 			<div
-				class="shrink-0 flex items-center gap-0.5 px-1 py-0.5 mr-1 rounded-sm bg-surface-secondary text-secondary"
-				title={`// freshness ${data.freshness}`}
+				class={twMerge(
+					'shrink-0 flex items-center gap-0.5 px-1 py-0.5 mr-1 rounded-sm',
+					freshnessState === 'fresh'
+						? 'bg-emerald-100 dark:bg-emerald-900/40 text-emerald-700 dark:text-emerald-300'
+						: freshnessState === 'stale'
+							? 'bg-amber-100 dark:bg-amber-900/40 text-amber-700 dark:text-amber-300'
+							: 'bg-surface-secondary text-secondary'
+				)}
+				title={freshnessTooltip}
 			>
 				<Timer size={10} />
 				<span class="text-3xs leading-none">{data.freshness}</span>
@@ -1,8 +1,15 @@
 import { JobService } from '$lib/gen'

 export type RunStatus = 'running' | 'success' | 'failure'
-/** Per-runnable badge state: latest run status + runs observed this session. */
-export type RunnableRunState = { status: RunStatus; runs: number }
+/**
+ * Per-runnable badge state: latest run status + runs observed this session.
+ * `lastSuccessAt` is the completion time (start + duration when the listing
+ * carries it, else start as a conservative lower bound) of the newest
+ * successful run seen by the poll — lets the freshness chip go green right
+ * after an in-session run, ahead of the next graph refetch (whose
+ * `last_success_at` would carry it).
+ */
+export type RunnableRunState = { status: RunStatus; runs: number; lastSuccessAt?: string }

 export type EventStatus = 'queued' | 'running' | 'success' | 'failure'
 /** One folder activity-log row (a job observed by the poll). */
@@ -14,6 +21,12 @@ export type PipelineEvent = {
 	/** What started it, as far as the job listing reveals. */
 	source: 'schedule' | 'run'
 	at: string
+	/**
+	 * Completion time (start + duration) for completed rows. The freshness
+	 * chip compares against completion — `at` is the start time and would
+	 * read a long run as older than its output actually is.
+	 */
+	completedAt?: string
 	/**
 	 * Queued jobs: when the job is due to start. A future value means a
 	 * scheduled run waiting for its cron tick, not pipeline activity.
@@ -45,7 +58,8 @@ function statesEq(a: Map<string, RunnableRunState>, b: Map<string, RunnableRunSt
 	if (a.size !== b.size) return false
 	for (const [k, v] of a) {
 		const w = b.get(k)
-		if (!w || w.status !== v.status || w.runs !== v.runs) return false
+		if (!w || w.status !== v.status || w.runs !== v.runs || w.lastSuccessAt !== v.lastSuccessAt)
+			return false
 	}
 	return true
 }
@@ -128,7 +142,7 @@ export function useActiveRunnableIds(
 	// keeps showing the last status while idle; only `dispose()` clears them.
 	const completedHistory = new Map<
 		string,
-		{ runs: number; lastStatus: RunStatus; lastTs: string }
+		{ runs: number; lastStatus: RunStatus; lastTs: string; lastSuccessTs?: string }
 	>()
 	const countedJobIds = new Set<string>()
 	// Job ids we've observed in-flight at least once. The catch-up pulse is
@@ -241,10 +255,25 @@ export function useActiveRunnableIds(
 						const prev = completedHistory.get(id)
 						const status: RunStatus = (j as any).success === true ? 'success' : 'failure'
 						const ts = startedTs ?? new Date(pollStartedMs).toISOString()
+						// Freshness compares against COMPLETION time (that's
+						// when the output materialized — the server-side
+						// last_success_at is completed_at too). The listing
+						// only carries started_at, so add duration_ms; when
+						// absent, the start is a conservative lower bound
+						// (errs stale, never false-fresh).
+						const durationMs = (j as any).duration_ms
+						const doneTs =
+							typeof durationMs === 'number' && startedTs
+								? new Date(new Date(startedTs).getTime() + durationMs).toISOString()
+								: ts
 						completedHistory.set(id, {
 							runs: (prev?.runs ?? 0) + 1,
 							lastStatus: !prev || ts >= prev.lastTs ? status : prev.lastStatus,
-							lastTs: !prev || ts >= prev.lastTs ? ts : prev.lastTs
+							lastTs: !prev || ts >= prev.lastTs ? ts : prev.lastTs,
+							lastSuccessTs:
+								status === 'success' && (!prev?.lastSuccessTs || doneTs >= prev.lastSuccessTs)
+									? doneTs
+									: prev?.lastSuccessTs
 						})
 					}
 				}
@@ -267,6 +296,10 @@ export function useActiveRunnableIds(
 								: 'failure',
 						source: (j as any).schedule_path ? 'schedule' : 'run',
 						at: startedTs ?? new Date(pollStartedMs).toISOString(),
+						completedAt:
+							!isQueued && typeof (j as any).duration_ms === 'number' && startedTs
+								? new Date(new Date(startedTs).getTime() + (j as any).duration_ms).toISOString()
+								: undefined,
 						scheduledFor: isQueued ? ((j as any).scheduled_for as string | undefined) : undefined
 					})
 				}
@@ -289,7 +322,11 @@ export function useActiveRunnableIds(
 		// previous badge state until a worker picks the job up.
 		const snap = new Map<string, RunnableRunState>()
 		for (const [id, h] of completedHistory) {
-			snap.set(id, { status: runningThisTick.has(id) ? 'running' : h.lastStatus, runs: h.runs })
+			snap.set(id, {
+				status: runningThisTick.has(id) ? 'running' : h.lastStatus,
+				runs: h.runs,
+				lastSuccessAt: h.lastSuccessTs
+			})
 		}
 		for (const id of runningThisTick) {
 			if (!snap.has(id)) snap.set(id, { status: 'running', runs: 0 })
@@ -1,6 +1,7 @@
 import { describe, expect, it } from 'vitest'
 import {
 	mergeColumnLineage,
+	parseDurationSecs,
 	parsePipelineAnnotations,
 	type ColumnLineage
 } from './parsePipelineAnnotations'
@@ -188,3 +189,40 @@ describe('mergeColumnLineage', () => {
 		expect(mergeColumnLineage([], annotated)).toEqual(annotated)
 	})
 })
+
+// Mirror of the Rust `parse_duration_secs` tests (windmill-common assets.rs)
+// — the freshness chip's staleness verdict depends on identical parsing.
+describe('parseDurationSecs', () => {
+	it('parses suffixed durations', () => {
+		expect(parseDurationSecs('30s')).toBe(30)
+		expect(parseDurationSecs('5m')).toBe(300)
+		expect(parseDurationSecs('2h')).toBe(7200)
+		expect(parseDurationSecs('1d')).toBe(86400)
+	})
+
+	it('bare integer means seconds', () => {
+		expect(parseDurationSecs('45')).toBe(45)
+	})
+
+	it('tolerates surrounding whitespace', () => {
+		expect(parseDurationSecs(' 5 m ')).toBe(300)
+	})
+
+	it('accepts an explicit plus sign (Rust i64 parsing does)', () => {
+		expect(parseDurationSecs('+5m')).toBe(300)
+		expect(parseDurationSecs('+45')).toBe(45)
+	})
+
+	it('rejects malformed / non-positive input', () => {
+		expect(parseDurationSecs('')).toBeUndefined()
+		expect(parseDurationSecs('h')).toBeUndefined()
+		expect(parseDurationSecs('1.5h')).toBeUndefined()
+		expect(parseDurationSecs('-5m')).toBeUndefined()
+		expect(parseDurationSecs('0')).toBeUndefined()
+		expect(parseDurationSecs('fast')).toBeUndefined()
+	})
+
+	it('rejects values beyond i32 seconds (mirrors backend cap)', () => {
+		expect(parseDurationSecs('999999999d')).toBeUndefined()
+	})
+})
@@ -67,6 +67,28 @@ export type FreshnessSpec = {
 	duration: string
 }

+// Mirrors backend `parse_duration_secs` (windmill-common assets.rs): a bare
+// integer means seconds, otherwise `<n>` with an `s`/`m`/`h`/`d` suffix
+// (e.g. `30s`, `5m`, `2h`, `1d`). Returns undefined for malformed or
+// non-positive input so a typo'd `// freshness` window fails safe (the chip
+// stays neutral instead of guessing a staleness verdict).
+export function parseDurationSecs(s: string): number | undefined {
+	const t = s.trim()
+	if (!t) return undefined
+	const last = t[t.length - 1]
+	const mult =
+		last === 's' ? 1 : last === 'm' ? 60 : last === 'h' ? 3600 : last === 'd' ? 86400 : undefined
+	const num = (mult !== undefined ? t.slice(0, -1) : t).trim()
+	// `+?`: Rust's i64 parsing accepts an explicit plus sign (`+5m`), so the
+	// mirror must too — divergence here would leave the chip neutral for a
+	// window the deploy path and watchdog honor.
+	if (mult === undefined && !/^\+?\d+$/.test(t)) return undefined
+	if (!/^\+?\d+$/.test(num)) return undefined
+	const secs = Number(num) * (mult ?? 1)
+	if (!Number.isSafeInteger(secs) || secs <= 0 || secs > 2147483647) return undefined
+	return secs
+}
+
 // `// retry <count> [<delay>]` — see backend RetrySpec. Delay is kept as the
 // raw duration string and resolved to seconds at deploy.
 export type RetrySpec = {
@@ -80,7 +80,15 @@ export function usePipelineHistory(
 						kind: j.job_kind.startsWith('flow') ? 'flow' : 'script',
 						status: j.success ? 'success' : 'failure',
 						source: j.schedule_path ? 'schedule' : 'run',
-						at: j.started_at ?? j.created_at
+						at: j.started_at ?? j.created_at,
+						// Same completion-time derivation as the live poll —
+						// the freshness chip compares against completion, and
+						// `at` (start) would read a long run as older than its
+						// output actually is.
+						completedAt:
+							j.started_at != undefined
+								? new Date(new Date(j.started_at).getTime() + j.duration_ms).toISOString()
+								: undefined
 					})
 				}
 				sawFullPage = rows.length === PER_PAGE
@@ -22,6 +22,11 @@ export interface AssetGraphRunnableNode {
 	// Raw `// freshness <duration>` value, e.g. "1h", "30m". Surfaced for
 	// the badge; the runtime parses it as needed.
 	freshness?: string
+	// Completion time (ISO) of the newest successful run of this pipeline
+	// member visible to the caller. The freshness chip compares it against
+	// the `// freshness` window to render fresh/stale. Absent = no
+	// successful run found (or none visible under job RLS).
+	last_success_at?: string
 	// `// tag <name>` worker-tag override. Surfaced for the badge so users
 	// can see which worker pool will pick this script up at a glance.
 	tag?: string
@@ -1,4 +1,14 @@
-import { Webhook, Mail, Calendar, Route, Unplug, Database, Terminal } from 'lucide-svelte'
+import {
+	Webhook,
+	Mail,
+	Calendar,
+	Route,
+	Unplug,
+	Database,
+	Terminal,
+	Timer,
+	Zap
+} from 'lucide-svelte'
 import KafkaIcon from '$lib/components/icons/KafkaIcon.svelte'
 import NatsIcon from '$lib/components/icons/NatsIcon.svelte'
 import MqttIcon from '$lib/components/icons/MqttIcon.svelte'
@@ -82,7 +92,8 @@ export const jobTriggerKinds: JobTriggerKind[] = [
 	'azure',
 	'google',
 	'github',
-	'asset'
+	'asset',
+	'freshness'
 ]

 export type Trigger = {
@@ -118,7 +129,12 @@ export const triggerIconMap = {
 	cli: Terminal,
 	nextcloud: NextcloudIcon,
 	google: GoogleIcon,
-	github: GithubIcon
+	github: GithubIcon,
+	// Job-attribution-only kinds (no trigger CRUD page): the pipeline asset
+	// cascade and the freshness watchdog. Needed so the Runs filter and job
+	// detail render these trigger kinds instead of a blank label / no icon.
+	asset: Zap,
+	freshness: Timer
 }

 export const triggerDisplayNamesMap = {
@@ -139,8 +155,12 @@ export const triggerDisplayNamesMap = {
 	cli: 'CLI',
 	nextcloud: 'Nextcloud',
 	google: 'Google',
-	github: 'GitHub'
-} as const satisfies Record<TriggerType, string>
+	github: 'GitHub',
+	asset: 'Asset cascade',
+	freshness: 'Freshness'
+	// `asset` / `freshness` are job-attribution-only (JobTriggerKind, not
+	// TriggerType) — hence the union in the satisfies below.
+} as const satisfies Record<TriggerType | 'asset' | 'freshness', string>

 /**
 * Converts a TriggerType to a CaptureTriggerKind when a mapping exists
@@ -1337,8 +1337,17 @@
 						? 'failure'
 						: 'success'
 			const cur = m.get(id)
-			if (cur) cur.runs += 1
-			else m.set(id, { status, runs: 1 })
+			// Freshness compares against completion; `at` (start) is the
+			// fallback lower bound for rows without a duration.
+			const successAt = e.status === 'success' ? (e.completedAt ?? e.at) : undefined
+			if (cur) {
+				cur.runs += 1
+				// Newest-first, so the first success per id is the latest one —
+				// it feeds the freshness chip between graph refetches.
+				if (successAt && !cur.lastSuccessAt) cur.lastSuccessAt = successAt
+			} else {
+				m.set(id, { status, runs: 1, lastSuccessAt: successAt })
+			}
 		}
 		return m
 	})