feat: add trigger_history table with source tracking (#10696)

* feat: add trigger_history table with source tracking

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

* fix: gate trigger history reads on scopes and harden its writers

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

* fix: filter trigger history scopes in SQL and match the cleared-handler diff

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

* fix: record a trigger restore from the trashbin in its history

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

* fix: record bulk http trigger creates and document the recording boundary

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

* fix: lock the trigger row when capturing its history preimage

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

* fix: only record an auto-disable that actually flipped the schedule

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

* chore: state the auto-disable invariant once instead of at four call sites

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

* feat: render trigger history changes as a structured field diff

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

* fix: make a server-initiated disable atomic with its history row

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

* docs: note that the auto-disable savepoint takes no pool connection

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

* docs: note the flow fallback is the last chance to disable

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

* fix: never leave a trigger enabled because its history row failed

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

* fix: retry the disable history row instead of dropping it on first failure

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

* fix: use the design-system Button for the change-value expander

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

* fix: hold the trigger row lock across its disable history row

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

* fix: keep the history-loss alert out of the listener cancellation race

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

* fix: read the history workspace through the trigger-workspace seam

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

---------

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

backend/.sqlx/query-7947ffe31b8e6f4a38fba9d9caf434fd409ea2806a7c720c2b1d76b2e7db6c32.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
+  "query": "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3 AND enabled = true",
   "describe": {
     "columns": [],
     "parameters": {
@@ -12,5 +12,5 @@
     },
     "nullable": []
   },
-  "hash": "2ee6d24b95cdda151585dcff19f8e7c931785fc21f7bbe9c3a82671943ced0ea"
+  "hash": "7947ffe31b8e6f4a38fba9d9caf434fd409ea2806a7c720c2b1d76b2e7db6c32"
 }

backend/.sqlx/query-928767710fb8b7dc0b1edc897d8ce9b6b59ae2f63e684d3db6eecbcffd767711.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO trigger_history\n            (workspace_id, trigger_kind, path, operation, source, username, changes)\n         SELECT $1, $2, p, $3, $4, $5, $6 FROM unnest($7::text[]) p",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Jsonb",
+        "TextArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "928767710fb8b7dc0b1edc897d8ce9b6b59ae2f63e684d3db6eecbcffd767711"
+}

backend/.sqlx/query-e52a80386b132e53a956458e2eb77d29bd6da4fc8511ed44f21d49f4c0965d36.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO trigger_history\n            (workspace_id, trigger_kind, path, operation, source, username, changes)\n         VALUES ($1, $2, $3, $4, $5, $6, $7)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e52a80386b132e53a956458e2eb77d29bd6da4fc8511ed44f21d49f4c0965d36"
+}

backend/.sqlx/query-fcbbc3b697249c6ee0ca542ea42fadecddead568110dd531fbace439c2263e4f.json

@@ -0,0 +1,71 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, trigger_kind, path, operation, source, username, created_at, changes\n         FROM trigger_history\n         WHERE workspace_id = $1\n           AND ($2::TEXT IS NULL OR trigger_kind = $2)\n           AND ($3::TEXT IS NULL OR path = $3)\n           AND ( $6\n                 OR path = ANY($7)\n                 OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx\n                             WHERE path = pfx\n                                OR left(path, length(pfx) + 1) = pfx || '/' ) )\n         ORDER BY id DESC\n         LIMIT $4 OFFSET $5",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "trigger_kind",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "operation",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "source",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "username",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 6,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 7,
+        "name": "changes",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Int8",
+        "Int8",
+        "Bool",
+        "TextArray",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      true,
+      false,
+      true
+    ]
+  },
+  "hash": "fcbbc3b697249c6ee0ca542ea42fadecddead568110dd531fbace439c2263e4f"
+}

backend/migrations/20260814090221_trigger_history.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS trigger_history;

backend/migrations/20260814090221_trigger_history.up.sql

@@ -0,0 +1,65 @@
+-- Append-only record of every schedule/trigger mutation: who, what changed, and
+-- from which kind of client.
+CREATE TABLE IF NOT EXISTS trigger_history (
+    id BIGSERIAL PRIMARY KEY,
+    workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE,
+    -- 'schedule' or a trigger's TRIGGER_TYPE ('http', 'kafka', ...). Not the
+    -- TRIGGER_KIND enum: that one is capture-oriented and misses 'schedule'.
+    trigger_kind VARCHAR(50) NOT NULL,
+    path VARCHAR(255) NOT NULL,
+    -- 'create' | 'update' | 'delete' | 'enable' | 'disable' | 'suspend'
+    operation VARCHAR(20) NOT NULL,
+    -- 'ui' | 'cli' | 'api' | 'worker'
+    source VARCHAR(20) NOT NULL,
+    -- NULL when the server acted on its own (worker auto-disable).
+    username VARCHAR(255),
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
+    -- {field: {old, new}} for the fields that actually changed. `old` is
+    -- absent where it is not known: a create, and the workspace-wide handler
+    -- override that rewrites every schedule without reading them first. NULL
+    -- when the operation carries no field-level diff at all (delete).
+    changes JSONB
+);
+
+CREATE INDEX IF NOT EXISTS idx_trigger_history_workspace_kind_path
+    ON trigger_history(workspace_id, trigger_kind, path, id DESC);
+
+CREATE INDEX IF NOT EXISTS idx_trigger_history_workspace_id
+    ON trigger_history(workspace_id, id DESC);
+
+GRANT ALL ON TABLE trigger_history TO windmill_user;
+GRANT ALL ON TABLE trigger_history TO windmill_admin;
+GRANT ALL ON SEQUENCE trigger_history_id_seq TO windmill_user;
+GRANT ALL ON SEQUENCE trigger_history_id_seq TO windmill_admin;
+
+ALTER TABLE trigger_history ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY admin_all ON trigger_history FOR ALL TO windmill_admin USING (true) WITH CHECK (true);
+
+-- Every mutating trigger route writes through the RLS pool, so windmill_user
+-- must be able to append.
+CREATE POLICY allow_insert ON trigger_history FOR INSERT TO windmill_user WITH CHECK (true);
+
+-- Reads mirror the path half of the live trigger's own policies: a row can
+-- quote a schedule's `args`, so it must not be readable by anyone who could not
+-- read the trigger it describes. Deliberately narrower than the live row on one
+-- point — the `extra_perms` grants have no counterpart here, since the history
+-- does not carry the row's ACL and must survive its deletion. Narrower is the
+-- safe direction.
+CREATE POLICY see_own ON trigger_history FOR SELECT TO windmill_user
+USING (
+    SPLIT_PART(path::text, '/', 1) = 'u'
+    AND SPLIT_PART(path::text, '/', 2) = current_setting('session.user')
+);
+
+CREATE POLICY see_member ON trigger_history FOR SELECT TO windmill_user
+USING (
+    SPLIT_PART(path::text, '/', 1) = 'g'
+    AND SPLIT_PART(path::text, '/', 2) = ANY(regexp_split_to_array(current_setting('session.groups'), ','))
+);
+
+CREATE POLICY see_folder_extra_perms_user ON trigger_history FOR SELECT TO windmill_user
+USING (
+    SPLIT_PART(path::text, '/', 1) = 'f'
+    AND SPLIT_PART(path::text, '/', 2) = ANY(regexp_split_to_array(current_setting('session.folders_read'), ','))
+);

backend/summarized_schema.txt

@@ -176,6 +176,9 @@ token: token_hash(char), token_prefix(char), token(char), label(char), expiratio
   FK: (workspace_id) -> workspace(id)
 token_expiry_notification: token_hash(char), expiration(ts)
   INDEX: idx_token_expiry_notification_expiration (expiration)
+trigger_history: id(bigint), workspace_id(char), trigger_kind(char), path(char), operation(char), source(char), username(char), created_at(ts), changes(jsonb)
+  FK: (workspace_id) -> workspace(id)
+  INDEX: idx_trigger_history_workspace_kind_path (workspace_id, trigger_kind, path, id), idx_trigger_history_workspace_id (workspace_id, id)
 tutorial_progress: email(char), progress(bit64), skipped_all(bool)
 unique_ext_jwt_token: jwt_hash(bigint), last_used_at(ts), email(text), username(text), is_admin(bool), is_operator(bool), workspace_id(text?), label(text?), scopes(text[]?)
 usage: id(char), is_workspace(bool), month_(int), usage(int)

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

@@ -67,8 +67,9 @@ pub struct ApiAuthed {
     /// `label-*` string. Only `username_override_from_label` sets it.
     pub username_override_is_token_label: bool,
     /// Whether the request authenticated with the session token minted at browser login.
-    /// Only `trigger_or_fallback` reads it — see `is_session_label` for why it attributes
-    /// rather than proves, and must not gate authority.
+    /// Read by `trigger_or_fallback` and by `TriggerSource::of_request` (which attributes a
+    /// trigger mutation to the UI) — see `is_session_label` for why it attributes rather than
+    /// proves, and must not gate authority.
     pub is_session_token: bool,
     pub token_prefix: Option<String>,
     pub read_only: bool,

backend/windmill-api-auth/src/scopes.rs

@@ -274,6 +274,7 @@ pub enum ScopeDomain {
 
     // Native trigger domains
     NativeTriggers,
+    TriggersHistory,
 
     // System domains
     Audit,
@@ -335,6 +336,7 @@ impl ScopeDomain {
             Self::PostgresTriggers => "postgres_triggers",
             Self::EmailTriggers => "email_triggers",
             Self::NativeTriggers => "native_triggers",
+            Self::TriggersHistory => "triggers_history",
             Self::Audit => "audit",
             Self::Settings => "settings",
             Self::Workers => "workers",
@@ -401,6 +403,7 @@ impl ScopeDomain {
             "indexer" | "srch" => Some(Self::Indexer),
             "teams" => Some(Self::Teams),
             "native_triggers" => Some(Self::NativeTriggers),
+            "triggers_history" => Some(Self::TriggersHistory),
             "git_sync" | "github_app" => Some(Self::GitSync),
             "capture" => Some(Self::Capture),
             "drafts" => Some(Self::Drafts),

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

@@ -27,6 +27,9 @@ use windmill_common::{
     db::UserDB,
     error::{Error, JsonResult, Result},
     schedule::Schedule,
+    trigger_history::{
+        self, TriggerHistoryEvent, TriggerOperation, TriggerSource, SCHEDULE_TRIGGER_KIND,
+    },
     user_drafts::{
         delete_all_drafts_for_path, fetch_draft_only_list_rows, overlay_or_draft_only,
         UserDraftItemKind, WithDraftOverlay, WithDraftQuery,
@@ -85,6 +88,40 @@ fn resolve_edited_by(authed: &ApiAuthed) -> String {
     authed.username.clone()
 }
 
+/// Append this mutation to `trigger_history`, diffing the row against `before`.
+///
+/// Call it on the transaction that made the change, after the change: the
+/// snapshot it takes is the "after" side of the diff, and the two commit or roll
+/// back together.
+async fn record_schedule_history(
+    tx: &mut sqlx::PgConnection,
+    authed: &ApiAuthed,
+    w_id: &str,
+    path: &str,
+    operation: TriggerOperation,
+    before: Option<serde_json::Value>,
+) -> Result<()> {
+    let after = trigger_history::snapshot_row(&mut *tx, "schedule", w_id, path).await?;
+    // Nothing to describe when the row is not there after the write: the same
+    // guard the trigger side needs, kept here so the two read alike.
+    if after.is_none() {
+        return Ok(());
+    }
+    trigger_history::record(
+        &mut *tx,
+        TriggerHistoryEvent {
+            workspace_id: w_id,
+            trigger_kind: SCHEDULE_TRIGGER_KIND,
+            path,
+            operation,
+            source: TriggerSource::of_request(authed.is_session_token),
+            username: Some(&authed.username),
+            changes: trigger_history::summarize_changes(before.as_ref(), after.as_ref()),
+        },
+    )
+    .await
+}
+
 pub fn workspaced_service() -> Router {
     Router::new()
         .route("/list", get(list_schedule))
@@ -417,6 +454,16 @@ async fn create_schedule(
     .await
     .map_err(|e| Error::internal_err(format!("inserting schedule in {w_id}: {e:#}")))?;
 
+    record_schedule_history(
+        &mut *tx,
+        &authed,
+        &w_id,
+        &ns.path,
+        TriggerOperation::Create,
+        None,
+    )
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,
@@ -524,6 +571,8 @@ async fn edit_schedule(
         authed.email.clone()
     };
 
+    let before = trigger_history::snapshot_row(&mut *tx, "schedule", &w_id, path).await?;
+
     let schedule = sqlx::query_as!(
         Schedule,
         r#"
@@ -632,6 +681,16 @@ async fn edit_schedule(
     // like set_enabled, flow updates, and worker job completions.
     clear_schedule(&mut tx, path, &w_id).await?;
 
+    record_schedule_history(
+        &mut *tx,
+        &authed,
+        &w_id,
+        path,
+        TriggerOperation::Update,
+        before,
+    )
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,
@@ -1084,6 +1143,8 @@ pub async fn set_enabled(
             }
         }
     }
+    let before = trigger_history::snapshot_row(&mut *tx, "schedule", &w_id, path).await?;
+
     // email is still written for backwards compat with old workers that don't know about permissioned_as
     let schedule_o = sqlx::query_as!(
         Schedule,
@@ -1139,6 +1200,20 @@ pub async fn set_enabled(
 
     clear_schedule(&mut tx, path, &w_id).await?;
 
+    record_schedule_history(
+        &mut *tx,
+        &authed,
+        &w_id,
+        path,
+        if payload.enabled {
+            TriggerOperation::Enable
+        } else {
+            TriggerOperation::Disable
+        },
+        before,
+    )
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,
@@ -1285,6 +1360,22 @@ async fn delete_schedule(
         .await?;
     }
 
+    // No diff: the row is gone, and the trashbin above already keeps its full
+    // contents for a restore.
+    trigger_history::record(
+        &mut *tx,
+        TriggerHistoryEvent {
+            workspace_id: &w_id,
+            trigger_kind: SCHEDULE_TRIGGER_KIND,
+            path,
+            operation: TriggerOperation::Delete,
+            source: TriggerSource::of_request(authed.is_session_token),
+            username: Some(&authed.username),
+            changes: None,
+        },
+    )
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,
@@ -1373,6 +1464,11 @@ async fn set_default_error_handler(
     }
 
     if payload.override_existing {
+        // The rewrite and its history rows go in one transaction: on separate
+        // connections a concurrent edit could interleave, leaving the
+        // id-ordered drawer showing the wrong latest change, and a failed
+        // insert would leave the schedules rewritten with nothing recording it.
+        let mut tx = db.begin().await?;
         let updated_schedules: Vec<String>;
         match payload.handler_type {
             HandlerType::Error => {
@@ -1386,14 +1482,14 @@ async fn set_default_error_handler(
                         payload.number_of_occurence_exact,
                         w_id,
                     )
-                    .fetch_all(&db)
+                    .fetch_all(&mut *tx)
                     .await?;
                 } else {
                     updated_schedules = sqlx::query_scalar!(
                         "UPDATE schedule SET ws_error_handler_muted = false, on_failure = NULL, on_failure_extra_args = NULL, on_failure_times = NULL, on_failure_exact = NULL WHERE workspace_id = $1 RETURNING path",
                         w_id,
                     )
-                    .fetch_all(&db)
+                    .fetch_all(&mut *tx)
                     .await?;
                 }
             }
@@ -1406,14 +1502,14 @@ async fn set_default_error_handler(
                         payload.number_of_occurence,
                         w_id,
                     )
-                    .fetch_all(&db)
+                    .fetch_all(&mut *tx)
                     .await?;
                 } else {
                     updated_schedules = sqlx::query_scalar!(
                         "UPDATE schedule SET on_recovery = NULL, on_recovery_extra_args = NULL, on_recovery_times = NULL WHERE workspace_id = $1 RETURNING path",
                         w_id,
                     )
-                    .fetch_all(&db)
+                    .fetch_all(&mut *tx)
                     .await?;
                 }
             }
@@ -1425,18 +1521,70 @@ async fn set_default_error_handler(
                         payload.extra_args,
                         w_id,
                     )
-                    .fetch_all(&db)
+                    .fetch_all(&mut *tx)
                     .await?;
                 } else {
                     updated_schedules = sqlx::query_scalar!(
                         "UPDATE schedule SET on_success = NULL, on_success_extra_args = NULL WHERE workspace_id = $1 RETURNING path",
                         w_id,
                     )
-                    .fetch_all(&db)
+                    .fetch_all(&mut *tx)
                     .await?;
                 }
             }
         }
+        // One row per schedule the workspace-wide override rewrote, so a handler
+        // that appeared on a schedule nobody edited is traceable. Every column
+        // the UPDATE above wrote, not just the handler path: the mute flag and
+        // the occurrence thresholds are what someone auditing a surprise
+        // notification change most needs. No `old` side and no
+        // already-had-this-value filter — the UPDATE rewrites the whole
+        // workspace unconditionally, so these rows record the write rather than
+        // a delta.
+        // Built from the same values the branch that ran actually bound: a reset
+        // (`payload.path` absent) hardcodes NULL / false in SQL while the request
+        // still carries the form's other fields, so reading them here would name
+        // values the write never produced.
+        let cleared = payload.path.is_none();
+        let handler_path = payload.path.clone();
+        let extra_args = (!cleared).then(|| payload.extra_args.clone()).flatten();
+        let times = (!cleared).then_some(payload.number_of_occurence).flatten();
+        let handler_fields = match payload.handler_type {
+            HandlerType::Error => serde_json::json!({
+                "on_failure": { "new": handler_path },
+                "on_failure_extra_args": { "new": extra_args },
+                "on_failure_times": { "new": times },
+                "on_failure_exact": {
+                    "new": (!cleared).then_some(payload.number_of_occurence_exact).flatten()
+                },
+                "ws_error_handler_muted": {
+                    "new": !cleared && payload.workspace_handler_muted.unwrap_or(false)
+                },
+            }),
+            HandlerType::Recovery => serde_json::json!({
+                "on_recovery": { "new": handler_path },
+                "on_recovery_extra_args": { "new": extra_args },
+                "on_recovery_times": { "new": times },
+            }),
+            HandlerType::Success => serde_json::json!({
+                "on_success": { "new": handler_path },
+                "on_success_extra_args": { "new": extra_args },
+            }),
+        };
+        trigger_history::record_bulk(
+            &mut tx,
+            &w_id,
+            SCHEDULE_TRIGGER_KIND,
+            &updated_schedules,
+            TriggerOperation::Update,
+            TriggerSource::of_request(authed.is_session_token),
+            Some(&authed.username),
+            Some(handler_fields),
+        )
+        .await?;
+
+        tx.commit().await?;
+
         for updated_schedule_path in updated_schedules {
             // managed ducklake-maintenance rows get the handler update (their
             // failures should reach workspace handlers) but must not be

backend/windmill-api/openapi.yaml

@@ -20342,6 +20342,36 @@ paths:
                       type: string
                       nullable: true
 
+  /w/{workspace}/triggers_history/list:
+    get:
+      summary: list the history of schedule and trigger modifications
+      operationId: listTriggerHistory
+      tags:
+        - trigger
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/Page"
+        - $ref: "#/components/parameters/PerPage"
+        - name: trigger_kind
+          description: "'schedule' or a trigger type (http, kafka, ...)"
+          in: query
+          schema:
+            type: string
+        - name: path
+          description: only return the history of the trigger at this path
+          in: query
+          schema:
+            type: string
+      responses:
+        "200":
+          description: trigger history
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/TriggerHistoryEntry"
+
   /w/{workspace}/folders/list:
     get:
       summary: list folders
@@ -28245,6 +28275,44 @@ components:
         is_fileset:
           type: boolean
 
+    TriggerHistoryEntry:
+      type: object
+      properties:
+        id:
+          type: integer
+          format: int64
+        trigger_kind:
+          type: string
+          description: "'schedule' or a trigger type (http, kafka, ...)"
+        path:
+          type: string
+        operation:
+          type: string
+          enum: [create, update, delete, enable, disable, suspend]
+        source:
+          type: string
+          description: The kind of client the change came from. `worker` means the server disabled the trigger on its own after a failure.
+          enum: [ui, cli, api, worker]
+        username:
+          type: string
+          nullable: true
+          description: Unset when the server acted on its own.
+        created_at:
+          type: string
+          format: date-time
+        changes:
+          type: object
+          nullable: true
+          additionalProperties: true
+          description: "{field: {old, new}} for the fields that actually changed. Unset for a delete."
+      required:
+        - id
+        - trigger_kind
+        - path
+        - operation
+        - source
+        - created_at
+
     Schedule:
       type: object
       properties:

backend/windmill-api/src/lib.rs

@@ -178,6 +178,7 @@ mod teams_oss;
 mod token;
 mod tracing_init;
 mod trash;
+mod trigger_history;
 pub mod triggers;
 mod users;
 #[cfg(feature = "private")]
@@ -280,6 +281,23 @@ async fn set_deploy_origin(
     windmill_common::deploy_origin::scope(origin, next.run(req)).await
 }
 
+/// Scope the request in the client kind it declares, so a trigger mutation can
+/// be attributed to the CLI rather than to a bare API call. Entered for every
+/// request, undeclared ones included: `TriggerSource::of_request` reads the
+/// scope's absence as "no request is being served", which is what separates a
+/// caller from a worker disabling a trigger on its own.
+async fn set_request_client(
+    req: axum::extract::Request,
+    next: axum::middleware::Next,
+) -> axum::response::Response {
+    let client = req
+        .headers()
+        .get(windmill_common::trigger_history::CLIENT_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .and_then(windmill_common::trigger_history::client_from_header);
+    windmill_common::trigger_history::scope_client(client, next.run(req)).await
+}
+
 #[cfg(not(feature = "tantivy"))]
 type IndexReader = ();
 
@@ -639,6 +657,7 @@ pub async fn run_server(
                         .nest("/folders_history", folder_history::workspaced_service())
                         .nest("/groups", groups::workspaced_service())
                         .nest("/groups_history", group_history::workspaced_service())
+                        .nest("/triggers_history", trigger_history::workspaced_service())
                         .nest("/inputs", windmill_api_inputs::workspaced_service())
                         .nest("/internal_db", internal_db::workspaced_service())
                         .route("/labels/list", get(list_workspace_labels))
@@ -1136,6 +1155,8 @@ pub async fn run_server(
 
     let app = app.layer(axum::middleware::from_fn(set_deploy_origin));
 
+    let app = app.layer(axum::middleware::from_fn(set_request_client));
+
     let app = app.layer(CatchPanicLayer::custom(|err| {
         tracing::error!("panic in handler, returning 500: {:?}", err);
         Response::builder()

backend/windmill-api/src/token.rs

@@ -236,6 +236,26 @@ lazy_static! {
             }],
         });
 
+        // Read-only: `trigger_history` is append-only and written by the server
+        // alone, so there is no `triggers_history:write`. Its own domain rather
+        // than a `schedules`/`*_triggers` alias: one listing spans every kind,
+        // and a history row quotes the whole trigger row (a schedule's `args`
+        // included), so reading it is an explicit grant rather than a side
+        // effect of being able to read the trigger. Path-selectable because the
+        // route filters rows by the caller's path grants.
+        groups.push(ScopeDomain {
+            name: "Trigger History".to_string(),
+            description: Some(
+                "Read-only access to the modification history of schedules and triggers"
+                    .to_string(),
+            ),
+            scopes: vec![ScopeOption {
+                value: "triggers_history:read".to_string(),
+                label: "Read".to_string(),
+                requires_resource_path: true,
+            }],
+        });
+
         groups.extend(build_standard_scope_domains());
         groups.extend(build_trigger_scope_domains());
 

backend/windmill-api/src/trash.rs

@@ -9,6 +9,9 @@ use windmill_common::{
     db::UserDB,
     error::{Error, Result},
     trashbin::{self, TrashItem, TrashItemWithData},
+    trigger_history::{
+        self, TriggerHistoryEvent, TriggerOperation, TriggerSource, SCHEDULE_TRIGGER_KIND,
+    },
     utils::require_admin,
 };
 
@@ -88,6 +91,31 @@ async fn restore_trash_item(
         .execute(&mut *tx)
         .await?;
 
+    // A restore puts the trigger back, so the history has to say so: otherwise
+    // the last thing it records for a live trigger is its own deletion. The
+    // trashed row is the snapshot, so this needs no extra read.
+    let restored_trigger_kind = match item.item_kind.as_str() {
+        SCHEDULE_TRIGGER_KIND => Some(SCHEDULE_TRIGGER_KIND),
+        // `<type>_trigger` is what `delete_trigger` trashes it under, and the
+        // stem is the `TRIGGER_TYPE` the history records against.
+        kind => kind.strip_suffix("_trigger"),
+    };
+    if let Some(trigger_kind) = restored_trigger_kind {
+        trigger_history::record(
+            &mut *tx,
+            TriggerHistoryEvent {
+                workspace_id: &w_id,
+                trigger_kind,
+                path: &item.item_path,
+                operation: TriggerOperation::Create,
+                source: TriggerSource::of_request(authed.is_session_token),
+                username: Some(&authed.username),
+                changes: trigger_history::summarize_changes(None, item.item_data.get("row")),
+            },
+        )
+        .await?;
+    }
+
     audit_log(
         &mut *tx,
         &authed,

backend/windmill-api/src/trigger_history.rs

@@ -0,0 +1,105 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+use axum::{
+    extract::{Extension, Path, Query},
+    routing::get,
+    Router,
+};
+use serde::{Deserialize, Serialize};
+use windmill_api_auth::{build_scope_path_filter, check_scopes, ApiAuthed, ScopePathFilter};
+use windmill_common::{
+    db::UserDB,
+    error::JsonResult,
+    utils::{paginate, Pagination},
+};
+
+pub fn workspaced_service() -> Router {
+    Router::new().route("/list", get(list_trigger_history))
+}
+
+#[derive(Serialize)]
+pub struct TriggerHistoryEntry {
+    pub id: i64,
+    pub trigger_kind: String,
+    pub path: String,
+    pub operation: String,
+    pub source: String,
+    pub username: Option<String>,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    pub changes: Option<serde_json::Value>,
+}
+
+#[derive(Deserialize)]
+pub struct ListTriggerHistoryQuery {
+    pub page: Option<usize>,
+    pub per_page: Option<usize>,
+    /// `"schedule"` or a trigger type (`"http"`, `"kafka"`, …).
+    pub trigger_kind: Option<String>,
+    pub path: Option<String>,
+}
+
+/// Two gates, because they answer different questions: the RLS policies on
+/// `trigger_history` bound the rows to what the *user* may read, and
+/// `triggers_history:read:<path>` bounds them further to what this *token* may
+/// read. Without the second, a token scoped to one path could read the diffs of
+/// every trigger its user can see, and a `create` row quotes the whole trigger
+/// row, a schedule's `args` included.
+async fn list_trigger_history(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path(w_id): Path<String>,
+    Query(query): Query<ListTriggerHistoryQuery>,
+) -> JsonResult<Vec<TriggerHistoryEntry>> {
+    if let Some(path) = query.path.as_deref() {
+        check_scopes(&authed, || format!("triggers_history:read:{}", path))?;
+    }
+
+    // In the WHERE, not a retain after the fetch: the result is paginated, and a
+    // post-fetch filter would let a page's size report how many rows the token
+    // may not read — and return short pages that read as "no history".
+    let (scope_all, scope_exact, scope_prefix) =
+        match build_scope_path_filter(&authed, "triggers_history", "read") {
+            ScopePathFilter::AllowAll => (true, Vec::new(), Vec::new()),
+            ScopePathFilter::Restricted { exact, prefix } => (false, exact, prefix),
+        };
+
+    let mut tx = user_db.begin(&authed).await?;
+
+    let (per_page, offset) = paginate(Pagination { page: query.page, per_page: query.per_page });
+
+    let history = sqlx::query_as!(
+        TriggerHistoryEntry,
+        "SELECT id, trigger_kind, path, operation, source, username, created_at, changes
+         FROM trigger_history
+         WHERE workspace_id = $1
+           AND ($2::TEXT IS NULL OR trigger_kind = $2)
+           AND ($3::TEXT IS NULL OR path = $3)
+           AND ( $6
+                 OR path = ANY($7)
+                 OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx
+                             WHERE path = pfx
+                                OR left(path, length(pfx) + 1) = pfx || '/' ) )
+         ORDER BY id DESC
+         LIMIT $4 OFFSET $5",
+        w_id,
+        query.trigger_kind,
+        query.path,
+        per_page as i64,
+        offset as i64,
+        scope_all,
+        &scope_exact[..],
+        &scope_prefix[..],
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
+    tx.commit().await?;
+
+    Ok(axum::Json(history))
+}

backend/windmill-common/src/lib.rs

@@ -121,6 +121,7 @@ pub mod teams_ee;
 pub mod teams_oss;
 pub mod tracing_init;
 pub mod trashbin;
+pub mod trigger_history;
 pub mod triggers;
 pub mod user_drafts;
 pub mod usernames;

backend/windmill-common/src/trigger_history.rs

@@ -0,0 +1,551 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+//! Append-only history of schedule and trigger mutations (`trigger_history`).
+//!
+//! Every field of a row is derived by the server at write time: the caller
+//! passes what it is doing, never who it claims to be or where it claims to
+//! come from. **Who** (the authed username, or nobody for a server-initiated
+//! change) and **what** (a field-level diff computed from the row before and
+//! after the write) are derived by the server and cannot be forged. **From what
+//! kind of client** ([`TriggerSource`]) is weaker on purpose: a first-party
+//! client declares itself in a header, so it attributes rather than proves —
+//! see [`TriggerSource::of_request`].
+//!
+//! # What is recorded
+//!
+//! Authoring a single trigger through its own surface — create, update, delete,
+//! enable/disable/suspend, restore from the trashbin, and the workspace-wide
+//! default-handler override — plus the server disabling one after a failure.
+//! **Adding a route that authors a trigger means adding a `record` call to it**;
+//! nothing enforces that, because the alternative (a database trigger) cannot
+//! see who or which client asked, and would fire on every listener ping.
+//!
+//! Deliberately outside that line, and not a gap to be closed one call site at a
+//! time:
+//!
+//! - **Cascades of renaming or deleting something else** — a script/flow rename
+//!   rewriting `script_path` (`triggers::update_triggers_script_path`), a user
+//!   being removed rewriting ownership. The event belongs to the runnable or the
+//!   user, not to the trigger.
+//! - **Workspace-level bulk operations** — archive, fork clone, cross-workspace
+//!   deploy. They move whole workspaces; a per-trigger row per path would say
+//!   nothing the workspace event does not.
+//! - **Runtime housekeeping** — clearing `paused_until` / `error` after a run,
+//!   consumer-offset state (`reset_offset`, `server_id`), the managed
+//!   ducklake-maintenance schedule. The same category as the `server_id` and
+//!   `last_server_ping` columns the diff already drops.
+//!
+//! # The server-initiated disables: the disable wins
+//!
+//! When the server disables a trigger it could not run, two things want to be
+//! true and cannot both be guaranteed: the trigger ends up disabled, and the
+//! history says who disabled it. The disable wins, every time.
+//!
+//! A trigger left enabled reads as healthy while never firing again, and for a
+//! flow schedule nothing comes back to retry — it arms its next occurrence when
+//! the flow *starts*, so once the runnable is gone that code is never reached
+//! again. Enabled-and-dead is silent; disabled-without-an-audit-row is not, and
+//! the trigger's own `error` column still says why.
+//!
+//! So each writer puts the disabling `UPDATE` and the record in one
+//! transaction, with only the insert inside a savepoint
+//! ([`record_in_disable_tx`]). Both land on the same commit, and the trigger's
+//! row lock is held across the pair — so the row cannot end up describing a
+//! trigger deleted and recreated at that path in between. If the insert alone
+//! fails it rolls back to the savepoint, the disable still commits, and the lost
+//! row is reported to the workspace error handler and the critical alert
+//! channel — loud, never silent.
+//!
+//! # Authorization contract
+//!
+//! None of the helpers here authorize anything: they take a connection and
+//! write what they are given, exactly like `audit_log`. A caller must already
+//! have authorized the mutation *and* performed it, and must derive `username`
+//! from the request's `ApiAuthed` and `source` from
+//! [`TriggerSource::of_request`] — never from anything the request body
+//! carries. Reads are gated separately, by the RLS policies on the table and by
+//! the token scopes the listing route checks.
+
+use sqlx::{Acquire, PgConnection};
+
+use crate::error::Result;
+
+/// Header a first-party client sets to name itself. Only `cli`, `ui` and `api`
+/// mean anything; any other value, and the header being absent, falls back to
+/// what the credentials say.
+pub const CLIENT_HEADER: &str = "x-windmill-client";
+
+/// `trigger_kind` a schedule is recorded under. Triggers use their own
+/// `TriggerCrud::TRIGGER_TYPE`.
+pub const SCHEDULE_TRIGGER_KIND: &str = "schedule";
+
+/// The kind of client a trigger mutation came from.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TriggerSource {
+    /// A browser session in the Windmill app.
+    Ui,
+    /// The `wmill` CLI (including the git-sync pull that shells out to it).
+    Cli,
+    /// A direct API call with a token: user scripts, CI, third-party clients.
+    Api,
+    /// No request at all: a worker or a trigger listener disabling something
+    /// after a failure.
+    Worker,
+}
+
+impl TriggerSource {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            TriggerSource::Ui => "ui",
+            TriggerSource::Cli => "cli",
+            TriggerSource::Api => "api",
+            TriggerSource::Worker => "worker",
+        }
+    }
+
+    fn from_client_header(value: &str) -> Option<Self> {
+        match value.trim().to_ascii_lowercase().as_str() {
+            "cli" => Some(TriggerSource::Cli),
+            "ui" => Some(TriggerSource::Ui),
+            "api" => Some(TriggerSource::Api),
+            _ => None,
+        }
+    }
+
+    /// The source of the request currently being served.
+    ///
+    /// The declared client wins when it is one we know; otherwise the token
+    /// decides, and only the session token minted at browser login attributes
+    /// to the UI. Both inputs are attribution, never authority — nothing reads
+    /// a history row to make an access decision, so a caller lying about either
+    /// only mislabels its own row.
+    pub fn of_request(is_session_token: bool) -> Self {
+        match REQUEST_CLIENT.try_with(|client| *client) {
+            Ok(Some(source)) => source,
+            Ok(None) if is_session_token => TriggerSource::Ui,
+            Ok(None) => TriggerSource::Api,
+            // Outside a request there is no caller to attribute to. The
+            // server-initiated paths pass `Worker` themselves; this is what
+            // keeps a stray call from inventing one.
+            Err(_) => TriggerSource::Worker,
+        }
+    }
+}
+
+/// What a mutation did to the trigger it is recorded against.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TriggerOperation {
+    Create,
+    Update,
+    Delete,
+    Enable,
+    Disable,
+    Suspend,
+}
+
+impl TriggerOperation {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            TriggerOperation::Create => "create",
+            TriggerOperation::Update => "update",
+            TriggerOperation::Delete => "delete",
+            TriggerOperation::Enable => "enable",
+            TriggerOperation::Disable => "disable",
+            TriggerOperation::Suspend => "suspend",
+        }
+    }
+}
+
+tokio::task_local! {
+    static REQUEST_CLIENT: Option<TriggerSource>;
+}
+
+/// Run `f` with `client` as the declared client of every trigger mutation it
+/// causes. Entered for every request, unmarked ones included, so that having no
+/// scope at all means "not serving a request" — which is what
+/// [`TriggerSource::Worker`] records.
+pub async fn scope_client<F: std::future::Future>(
+    client: Option<TriggerSource>,
+    f: F,
+) -> F::Output {
+    REQUEST_CLIENT.scope(client, f).await
+}
+
+/// Parse the declared client of the request being served, if any.
+pub fn client_from_header(value: &str) -> Option<TriggerSource> {
+    TriggerSource::from_client_header(value)
+}
+
+/// Row fields that say nothing about the change itself: bookkeeping the history
+/// row already carries, and listener runtime state that moves on its own.
+const IGNORED_FIELDS: &[&str] = &[
+    "workspace_id",
+    "edited_at",
+    "edited_by",
+    "extra_perms",
+    "last_server_ping",
+    "server_id",
+    // Listener runtime state like the two above: every trigger update clears it,
+    // so keeping it here would tag an ordinary edit with the failure it had
+    // before. The server-initiated disables put the error in `changes`
+    // themselves, so nothing is lost.
+    "error",
+    // Written from the requester on every schedule mutation, purely for workers
+    // that predate `permissioned_as`; it tracks the editor, not the schedule.
+    "email",
+];
+
+/// A `changes` payload bigger than this is replaced by the list of field names
+/// it would have held. A schedule's `args` is caller-supplied and bounded only
+/// by the API's request-size limit, and a history row is not worth a
+/// multi-megabyte write.
+const MAX_CHANGES_BYTES: usize = 32 * 1024;
+
+/// The row at `path` as JSON, or `None` when there is none — which, on an RLS
+/// connection, also covers a row the caller cannot see.
+///
+/// `FOR UPDATE`, so the preimage and the mutation that follows it see the same
+/// row: without the lock another request can commit between the two, and its
+/// change then lands in this caller's diff under this caller's name.
+///
+/// Two things follow from taking the lock here rather than at the write:
+///
+/// - The only row locked is the one the caller is about to write, and the
+///   schedule paths reach the job queue only afterwards, so their documented
+///   schedule-then-queue order is unchanged.
+/// - The lock is held for whatever the caller does before its own `UPDATE`. For
+///   `TriggerCrud::update_trigger` that includes the impl's external work — the
+///   postgres impl opens a replication slot on a user-supplied host, the gcp and
+///   azure impls call their subscription APIs — so a concurrent `setmode`, a
+///   listener error write, or a script rename's bulk `script_path` update waits
+///   on that call. Bounded by those APIs, not by us; the alternative is a
+///   preimage inside each impl next to its own `UPDATE`.
+///
+/// `table` is interpolated: pass a compile-time constant, never anything a
+/// caller can reach.
+pub async fn snapshot_row(
+    conn: &mut PgConnection,
+    table: &'static str,
+    workspace_id: &str,
+    path: &str,
+) -> Result<Option<serde_json::Value>> {
+    // SAFETY: `table` is a compile-time constant.
+    let snapshot: Option<serde_json::Value> = sqlx::query_scalar(&format!(
+        "SELECT to_jsonb(t) FROM {table} t WHERE workspace_id = $1 AND path = $2 FOR UPDATE"
+    ))
+    .bind(workspace_id)
+    .bind(path)
+    .fetch_optional(&mut *conn)
+    .await?;
+    Ok(snapshot)
+}
+
+/// A field-level diff of two row snapshots, as `{field: {"old": …, "new": …}}`,
+/// with `"old"` omitted where there is none to report.
+///
+/// A create (`before` absent) keeps every non-null column of the new row, which
+/// is its initial shape including whatever the column defaults supplied —
+/// `to_jsonb` cannot tell a caller-set column from a defaulted one. Returns
+/// `None` when nothing meaningful changed.
+pub fn summarize_changes(
+    before: Option<&serde_json::Value>,
+    after: Option<&serde_json::Value>,
+) -> Option<serde_json::Value> {
+    let empty = serde_json::Map::new();
+    let before = before.and_then(|v| v.as_object()).unwrap_or(&empty);
+    let after = after.and_then(|v| v.as_object())?;
+
+    // Names of the changed fields, and the running size of what has been cloned
+    // so far. Measured as it goes rather than by serializing the finished map:
+    // a caller-sized `args` would otherwise be cloned in full and then copied
+    // again just to learn it was too big.
+    let mut fields = Vec::new();
+    let mut changes = serde_json::Map::new();
+    let mut bytes = 0usize;
+    for (field, new_value) in after {
+        if IGNORED_FIELDS.contains(&field.as_str()) {
+            continue;
+        }
+        let old_value = before.get(field);
+        match old_value {
+            Some(old_value) if old_value == new_value => continue,
+            None if new_value.is_null() => continue,
+            _ => {}
+        }
+        fields.push(field.clone());
+        if bytes <= MAX_CHANGES_BYTES {
+            bytes += json_len(new_value) + old_value.map_or(0, json_len) + field.len();
+        }
+        if bytes > MAX_CHANGES_BYTES {
+            continue;
+        }
+        let mut entry = serde_json::Map::new();
+        if let Some(old_value) = old_value {
+            entry.insert("old".to_string(), old_value.clone());
+        }
+        entry.insert("new".to_string(), new_value.clone());
+        changes.insert(field.clone(), serde_json::Value::Object(entry));
+    }
+
+    if fields.is_empty() {
+        return None;
+    }
+    if bytes > MAX_CHANGES_BYTES {
+        return Some(serde_json::json!({ "truncated_fields": fields }));
+    }
+    Some(serde_json::Value::Object(changes))
+}
+
+/// Serialized size of `value` without building the string for it.
+fn json_len(value: &serde_json::Value) -> usize {
+    struct Counter(usize);
+    impl std::io::Write for Counter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0 += buf.len();
+            Ok(buf.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+    let mut counter = Counter(0);
+    let _ = serde_json::to_writer(&mut counter, value);
+    counter.0
+}
+
+/// jsonb rejects `\u0000` inside a string, and `changes` quotes caller-supplied
+/// text — a schedule's `args`, a worker's error message. One NUL anywhere in
+/// there would fail the insert and cost the row.
+fn strip_nuls(value: &mut serde_json::Value) {
+    match value {
+        serde_json::Value::String(s) if s.contains('\0') => *s = s.replace('\0', ""),
+        serde_json::Value::Array(items) => items.iter_mut().for_each(strip_nuls),
+        serde_json::Value::Object(map) => map.values_mut().for_each(strip_nuls),
+        _ => {}
+    }
+}
+
+/// The last word on what reaches the column, applied at the write itself so a
+/// hand-built `changes` (the server-initiated disables carry an error string of
+/// unknown length and origin) gets it too, not just a computed diff.
+fn cap_changes(changes: Option<serde_json::Value>) -> Option<serde_json::Value> {
+    let mut changes = changes?;
+    strip_nuls(&mut changes);
+    if json_len(&changes) <= MAX_CHANGES_BYTES {
+        return Some(changes);
+    }
+    let fields = changes
+        .as_object()
+        .map(|o| o.keys().cloned().collect::<Vec<_>>())
+        .unwrap_or_default();
+    Some(serde_json::json!({ "truncated_fields": fields }))
+}
+
+/// One trigger mutation, as it is about to be recorded.
+#[derive(Clone)]
+pub struct TriggerHistoryEvent<'a> {
+    pub workspace_id: &'a str,
+    /// `"schedule"`, or the trigger's `TRIGGER_TYPE` (`"http"`, `"kafka"`, …).
+    pub trigger_kind: &'a str,
+    pub path: &'a str,
+    pub operation: TriggerOperation,
+    pub source: TriggerSource,
+    /// `None` when the server acted on its own.
+    pub username: Option<&'a str>,
+    pub changes: Option<serde_json::Value>,
+}
+
+impl<'a> TriggerHistoryEvent<'a> {
+    /// The event for a trigger the server disabled on its own after a failure.
+    ///
+    /// `forced_state` is the column the disable wrote, in the same
+    /// `{field: {old, new}}` shape as a diff — the two disable paths write
+    /// different columns (`enabled` for a schedule, `mode` for a trigger).
+    ///
+    /// Record this only when the disabling `UPDATE` reported an affected row,
+    /// and only when that `UPDATE` was itself predicated on the trigger still
+    /// being enabled. The server reads the trigger long before it writes, so
+    /// without both the row describes a transition a user had already made.
+    pub fn server_disable(
+        workspace_id: &'a str,
+        trigger_kind: &'a str,
+        path: &'a str,
+        mut forced_state: serde_json::Value,
+        error: &str,
+    ) -> Self {
+        if let Some(obj) = forced_state.as_object_mut() {
+            obj.insert("error".to_string(), serde_json::json!({ "new": error }));
+        }
+        Self {
+            workspace_id,
+            trigger_kind,
+            path,
+            operation: TriggerOperation::Disable,
+            source: TriggerSource::Worker,
+            username: None,
+            changes: Some(forced_state),
+        }
+    }
+}
+
+/// Record a disable inside the transaction that made it, without letting a
+/// failed insert take the disable down with it.
+///
+/// The caller's `UPDATE` holds the trigger's row lock until that transaction
+/// commits, and this runs inside that window — so the row cannot end up
+/// describing a trigger that was deleted and recreated at the same path in
+/// between, which is the whole point of doing it here rather than on a second
+/// connection afterwards.
+///
+/// The insert itself goes in a savepoint. If it fails it rolls back alone, the
+/// caller still commits the disable, and the reason comes back here so the
+/// caller can alert: a trigger left enabled reads as healthy while never firing
+/// again, which is worse than a missing audit row.
+pub async fn record_in_disable_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    event: TriggerHistoryEvent<'_>,
+) -> Option<String> {
+    let mut savepoint = match tx.begin().await {
+        Ok(savepoint) => savepoint,
+        Err(e) => return Some(e.to_string()),
+    };
+    match record(&mut savepoint, event).await {
+        Ok(()) => savepoint.commit().await.err().map(|e| e.to_string()),
+        Err(e) => {
+            savepoint.rollback().await.ok();
+            Some(e.to_string())
+        }
+    }
+}
+
+/// Append `event` to the history.
+///
+/// Pass the same connection as the mutation for the two to commit together.
+/// Does not authorize — see the module docs.
+pub async fn record(conn: &mut PgConnection, event: TriggerHistoryEvent<'_>) -> Result<()> {
+    sqlx::query!(
+        "INSERT INTO trigger_history
+            (workspace_id, trigger_kind, path, operation, source, username, changes)
+         VALUES ($1, $2, $3, $4, $5, $6, $7)",
+        event.workspace_id,
+        event.trigger_kind,
+        event.path,
+        event.operation.as_str(),
+        event.source.as_str(),
+        event.username,
+        cap_changes(event.changes) as _,
+    )
+    .execute(&mut *conn)
+    .await?;
+    Ok(())
+}
+
+/// Append one row per path, all describing the same change.
+///
+/// For the workspace-wide operations that rewrite every schedule at once, where
+/// a per-path diff would cost a snapshot per row and say the same thing each
+/// time. Does not authorize — see the module docs.
+pub async fn record_bulk(
+    conn: &mut PgConnection,
+    workspace_id: &str,
+    trigger_kind: &str,
+    paths: &[String],
+    operation: TriggerOperation,
+    source: TriggerSource,
+    username: Option<&str>,
+    changes: Option<serde_json::Value>,
+) -> Result<()> {
+    if paths.is_empty() {
+        return Ok(());
+    }
+    sqlx::query!(
+        "INSERT INTO trigger_history
+            (workspace_id, trigger_kind, path, operation, source, username, changes)
+         SELECT $1, $2, p, $3, $4, $5, $6 FROM unnest($7::text[]) p",
+        workspace_id,
+        trigger_kind,
+        operation.as_str(),
+        source.as_str(),
+        username,
+        cap_changes(changes) as _,
+        paths,
+    )
+    .execute(&mut *conn)
+    .await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    /// The whole worker side of the attribution rests on this: a mutation made
+    /// outside a request records `worker` without each call site saying so.
+    #[tokio::test]
+    async fn client_is_absent_outside_a_request() {
+        assert_eq!(TriggerSource::of_request(false), TriggerSource::Worker);
+        assert_eq!(
+            scope_client(None, async { TriggerSource::of_request(true) }).await,
+            TriggerSource::Ui
+        );
+        assert_eq!(
+            scope_client(None, async { TriggerSource::of_request(false) }).await,
+            TriggerSource::Api
+        );
+        assert_eq!(
+            scope_client(Some(TriggerSource::Cli), async {
+                TriggerSource::of_request(true)
+            })
+            .await,
+            TriggerSource::Cli
+        );
+    }
+
+    /// `error` and `edited_at` stand in for the whole ignore list: every trigger
+    /// update clears `error`, so without it an ordinary edit would carry the
+    /// failure the trigger had before it.
+    #[test]
+    fn diff_keeps_only_what_changed() {
+        let before = json!({"schedule": "0 0 * * *", "enabled": true, "edited_at": "a", "error": "boom"});
+        let after = json!({"schedule": "0 1 * * *", "enabled": true, "edited_at": "b", "error": null});
+        assert_eq!(
+            summarize_changes(Some(&before), Some(&after)),
+            Some(json!({"schedule": {"old": "0 0 * * *", "new": "0 1 * * *"}}))
+        );
+        assert_eq!(summarize_changes(Some(&before), Some(&before)), None);
+    }
+
+    #[test]
+    fn create_drops_null_columns_and_bookkeeping() {
+        let after = json!({"schedule": "0 0 * * *", "summary": null, "workspace_id": "w"});
+        assert_eq!(
+            summarize_changes(None, Some(&after)),
+            Some(json!({"schedule": {"new": "0 0 * * *"}}))
+        );
+    }
+
+    /// A NUL reaching the column fails the insert, and `changes` quotes
+    /// caller-supplied text — so this is the difference between a recorded
+    /// disable and a lost one.
+    #[test]
+    fn nul_bytes_never_reach_the_column() {
+        let changes = cap_changes(Some(json!({ "error": { "new": "boom\u{0}tail" } })));
+        assert_eq!(changes, Some(json!({ "error": { "new": "boomtail" } })));
+    }
+
+    #[test]
+    fn oversized_changes_keep_the_field_names() {
+        let after = json!({ "args": "x".repeat(MAX_CHANGES_BYTES + 1) });
+        assert_eq!(
+            summarize_changes(None, Some(&after)),
+            Some(json!({"truncated_fields": ["args"]}))
+        );
+    }
+}

backend/windmill-queue/src/jobs.rs

@@ -2503,6 +2503,68 @@ pub async fn send_success_to_workspace_handler<'a, 'c, T: Serialize + Send + Syn
     Ok(())
 }
 
+/// The event for a schedule the server disabled on its own.
+pub fn schedule_auto_disable_event<'a>(
+    workspace_id: &'a str,
+    path: &'a str,
+    error: &str,
+) -> windmill_common::trigger_history::TriggerHistoryEvent<'a> {
+    windmill_common::trigger_history::TriggerHistoryEvent::server_disable(
+        workspace_id,
+        windmill_common::trigger_history::SCHEDULE_TRIGGER_KIND,
+        path,
+        serde_json::json!({ "enabled": { "old": true, "new": false } }),
+        error,
+    )
+}
+
+/// Disable a schedule the server can no longer arm, and record that it did.
+///
+/// Contract on `record_in_disable_tx`. Here `tx` is the job-completion
+/// transaction, so the savepoint also keeps a failed insert from poisoning it.
+///
+/// Returns `Err` only when the disable itself failed; a lost history row comes
+/// back through `history_lost` for the caller to report.
+async fn disable_schedule_and_record(
+    tx: &mut Transaction<'_, Postgres>,
+    schedule: &Schedule,
+    err: &Error,
+    history_lost: &mut Option<String>,
+) -> Result<u64, Error> {
+    let disable_result = sqlx::query!(
+        "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3 AND enabled = true",
+        err.to_string(),
+        &schedule.workspace_id,
+        &schedule.path
+    )
+    .execute(&mut **tx)
+    .await;
+
+    #[cfg(feature = "failpoints")]
+    let disable_result = if schedule_failpoints::is_active(
+        schedule_failpoints::ScheduleFailPoint::ScheduleDisable,
+    ) {
+        Err(sqlx::Error::Protocol(
+            "failpoint: schedule disable".to_string(),
+        ))
+    } else {
+        disable_result
+    };
+
+    let rows = disable_result?.rows_affected();
+    // Zero rows means a user disabled the schedule first: no transition of ours
+    // to record.
+    if rows == 0 {
+        return Ok(0);
+    }
+
+    let event =
+        schedule_auto_disable_event(&schedule.workspace_id, &schedule.path, &err.to_string());
+    *history_lost = windmill_common::trigger_history::record_in_disable_tx(tx, event).await;
+
+    Ok(rows)
+}
+
 pub async fn try_schedule_next_job<'c>(
     db: &Pool<Postgres>,
     mut tx: Transaction<'c, Postgres>,
@@ -2657,36 +2719,31 @@ pub async fn try_schedule_next_job<'c>(
                 "Could not push next scheduled job for {}: {err}. Disabling schedule.",
                 schedule.path
             );
-            let disable_result = sqlx::query!(
-                "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
-                err.to_string(),
-                &schedule.workspace_id,
-                &schedule.path
-            )
-            .execute(&mut *tx)
-            .await;
-            #[cfg(feature = "failpoints")]
-            let disable_result = if schedule_failpoints::is_active(
-                schedule_failpoints::ScheduleFailPoint::ScheduleDisable,
-            ) {
-                Err(sqlx::Error::Protocol(
-                    "failpoint: schedule disable".to_string(),
-                ))
-            } else {
-                disable_result
-            };
-            if let Err(disable_err) = disable_result {
+            let mut history_lost = None;
+            match disable_schedule_and_record(&mut tx, schedule, err, &mut history_lost).await {
+                Err(disable_err) => {
+                    report_error_to_workspace_handler_or_critical_side_channel(
+                        job,
+                        db,
+                        format!(
+                            "Could not push next scheduled job for {} and could not disable schedule: {disable_err}",
+                            schedule.path,
+                        ),
+                    )
+                    .await;
+                }
+                Ok(_) => push_err = None,
+            }
+            if let Some(history_err) = history_lost {
                 report_error_to_workspace_handler_or_critical_side_channel(
                     job,
                     db,
                     format!(
-                        "Could not push next scheduled job for {} and could not disable schedule: {disable_err}",
+                        "Disabled schedule {} but could not record it in the trigger history: {history_err}",
                         schedule.path,
                     ),
                 )
                 .await;
-            } else {
-                push_err = None;
             }
         }
     }

backend/windmill-trigger-http/src/handler.rs

@@ -307,6 +307,37 @@ pub async fn create_many_http_triggers(
             .map_err(|err| error_wrapper(&new_http_trigger.config.route_path, err.into()))?;
         }
 
+        // Bulk create is still authoring, so it records like the single-create
+        // route rather than being the one way to make a trigger appear with no
+        // history behind it.
+        let created = windmill_common::trigger_history::snapshot_row(
+            &mut *tx,
+            "http_trigger",
+            &w_id,
+            &new_http_trigger.base.path,
+        )
+        .await
+        .map_err(|err| error_wrapper(&new_http_trigger.config.route_path, err))?;
+        windmill_common::trigger_history::record(
+            &mut *tx,
+            windmill_common::trigger_history::TriggerHistoryEvent {
+                workspace_id: &w_id,
+                trigger_kind: HttpTrigger::TRIGGER_TYPE,
+                path: &new_http_trigger.base.path,
+                operation: windmill_common::trigger_history::TriggerOperation::Create,
+                source: windmill_common::trigger_history::TriggerSource::of_request(
+                    authed.is_session_token,
+                ),
+                username: Some(&authed.username),
+                changes: windmill_common::trigger_history::summarize_changes(
+                    None,
+                    created.as_ref(),
+                ),
+            },
+        )
+        .await
+        .map_err(|err| error_wrapper(&new_http_trigger.config.route_path, err))?;
+
         audit_log(
             &mut *tx,
             &authed,

backend/windmill-trigger/src/handler.rs

@@ -16,6 +16,7 @@ use windmill_api_auth::{build_scope_path_predicate, check_scopes, ApiAuthed};
 use windmill_common::{
     db::UserDB,
     error::{Error, JsonResult, Result},
+    trigger_history::{self, TriggerHistoryEvent, TriggerOperation, TriggerSource},
     user_drafts::{
         delete_all_drafts_for_path, delete_own_draft_for_path, fetch_draft_only_list_rows,
         overlay_or_draft_only, UserDraftItemKind, WithDraftOverlay, WithDraftQuery,
@@ -458,6 +459,45 @@ pub trait TriggerCrud: Send + Sync + 'static {
     }
 }
 
+/// Append this mutation to `trigger_history`, diffing the row at `path` against
+/// `before`.
+///
+/// Call it on the transaction that made the change, after the change: the
+/// snapshot it takes is the "after" side of the diff, and the two commit or roll
+/// back together.
+///
+/// Records nothing when the snapshots say no row was written. `TriggerCrud::update_trigger`
+/// returns `Result<()>` and several impls do not check `rows_affected`, so an
+/// update aimed at a path that does not exist — or that RLS hides from the
+/// caller — reaches here having changed nothing; without this the caller could
+/// forge history rows at any path, since the insert policy is `WITH CHECK (true)`.
+async fn record_trigger_history<T: TriggerCrud>(
+    tx: &mut PgConnection,
+    authed: &ApiAuthed,
+    workspace_id: &str,
+    path: &str,
+    operation: TriggerOperation,
+    before: Option<serde_json::Value>,
+) -> Result<()> {
+    let after = trigger_history::snapshot_row(&mut *tx, T::TABLE_NAME, workspace_id, path).await?;
+    if after.is_none() || (operation == TriggerOperation::Update && before.is_none()) {
+        return Ok(());
+    }
+    trigger_history::record(
+        &mut *tx,
+        TriggerHistoryEvent {
+            workspace_id,
+            trigger_kind: T::TRIGGER_TYPE,
+            path,
+            operation,
+            source: TriggerSource::of_request(authed.is_session_token),
+            username: Some(&authed.username),
+            changes: trigger_history::summarize_changes(before.as_ref(), after.as_ref()),
+        },
+    )
+    .await
+}
+
 pub fn trigger_routes<T: TriggerCrud + 'static>() -> Router {
     let mut router = Router::new()
         .route("/create", post(create_trigger::<T>))
@@ -556,6 +596,16 @@ async fn create_trigger<T: TriggerCrud>(
         .await?;
     }
 
+    record_trigger_history::<T>(
+        &mut *tx,
+        &authed,
+        &workspace_id,
+        &new_path,
+        TriggerOperation::Create,
+        None,
+    )
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,
@@ -782,6 +832,9 @@ async fn update_trigger<T: TriggerCrud>(
         &authed.username,
     );
 
+    let before =
+        trigger_history::snapshot_row(&mut *tx, T::TABLE_NAME, &workspace_id, path).await?;
+
     handler
         .update_trigger(&db, &mut *tx, &authed, &workspace_id, path, edit_trigger)
         .await?;
@@ -799,6 +852,18 @@ async fn update_trigger<T: TriggerCrud>(
         .await?;
     }
 
+    // Recorded at the new path, so a rename reads as one event there with
+    // `path` among the changed fields rather than a delete plus a create.
+    record_trigger_history::<T>(
+        &mut *tx,
+        &authed,
+        &workspace_id,
+        &new_path,
+        TriggerOperation::Update,
+        before,
+    )
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,
@@ -913,6 +978,22 @@ async fn delete_trigger<T: TriggerCrud>(
         .await?;
     }
 
+    // No diff: the row is gone, and the trashbin above already keeps its full
+    // contents for a restore.
+    trigger_history::record(
+        &mut *tx,
+        TriggerHistoryEvent {
+            workspace_id: &workspace_id,
+            trigger_kind: T::TRIGGER_TYPE,
+            path,
+            operation: TriggerOperation::Delete,
+            source: TriggerSource::of_request(authed.is_session_token),
+            username: Some(&authed.username),
+            changes: None,
+        },
+    )
+    .await?;
+
     audit_log(
         &mut *tx,
         &authed,
@@ -1052,6 +1133,9 @@ async fn set_trigger_mode<T: TriggerCrud>(
         }
     }
 
+    let before =
+        trigger_history::snapshot_row(&mut *tx, T::TABLE_NAME, &workspace_id, path).await?;
+
     let updated = handler
         .set_trigger_mode(&authed, &mut *tx, &workspace_id, path, &payload.mode)
         .await?;
@@ -1063,6 +1147,20 @@ async fn set_trigger_mode<T: TriggerCrud>(
         )));
     }
 
+    record_trigger_history::<T>(
+        &mut *tx,
+        &authed,
+        &workspace_id,
+        path,
+        match payload.mode {
+            TriggerMode::Enabled => TriggerOperation::Enable,
+            TriggerMode::Disabled => TriggerOperation::Disable,
+            TriggerMode::Suspended => TriggerOperation::Suspend,
+        },
+        before,
+    )
+    .await?;
+
     tx.commit().await?;
 
     handle_deployment_metadata(

backend/windmill-trigger/src/listener.rs

@@ -386,9 +386,14 @@ pub trait Listener: TriggerCrud + TriggerJobArgs {
         error: String,
     ) {
         if listening_trigger.trigger_mode {
-            // SAFETY: Self::TABLE_NAME is a compile-time constant.
-            let report_status = sqlx::query(&format!(
-                r#"
+            // Contract on `record_in_disable_tx`: one transaction so the row
+            // lock spans both writes.
+            let mut history_err = None;
+            let report_status = async {
+                let mut tx = db.begin().await?;
+                // SAFETY: Self::TABLE_NAME is a compile-time constant.
+                let rows = sqlx::query(&format!(
+                    r#"
                     UPDATE
                         {}
                     SET
@@ -398,18 +403,60 @@ pub trait Listener: TriggerCrud + TriggerJobArgs {
                         last_server_ping = NULL
                     WHERE
                         workspace_id = $2 AND
-                        path = $3
+                        path = $3 AND
+                        mode <> 'disabled'::TRIGGER_MODE
                 "#,
-                Self::TABLE_NAME
-            ))
-            .bind(&error)
-            .bind(&listening_trigger.workspace_id)
-            .bind(&listening_trigger.path)
-            .execute(db)
+                    Self::TABLE_NAME
+                ))
+                .bind(&error)
+                .bind(&listening_trigger.workspace_id)
+                .bind(&listening_trigger.path)
+                .execute(&mut *tx)
+                .await?
+                .rows_affected();
+
+                // Zero rows: deleted, or a user disabled it first — no
+                // transition of ours to record.
+                if rows > 0 {
+                    // `to_key`, not `Display`: it is what lines up with the
+                    // `TRIGGER_TYPE` the API records under.
+                    let trigger_kind = Self::TRIGGER_KIND.to_key();
+                    history_err = windmill_common::trigger_history::record_in_disable_tx(
+                        &mut tx,
+                        windmill_common::trigger_history::TriggerHistoryEvent::server_disable(
+                            &listening_trigger.workspace_id,
+                            &trigger_kind,
+                            &listening_trigger.path,
+                            serde_json::json!({ "mode": { "new": "disabled" } }),
+                            &error,
+                        ),
+                    )
+                    .await;
+                }
+                tx.commit().await?;
+                Ok::<(), Error>(())
+            }
             .await;
 
+            if let Some(history_err) = history_err {
+                // Spawned: the commit above made the cleared `server_id` visible,
+                // so the ping branch of the enclosing `select!` is about to
+                // finish and drop everything left in this future. Awaiting the
+                // alert here would lose the one signal that the row is missing.
+                let message = format!(
+                    "Disabled {} trigger {} but could not record it in the trigger history: {}",
+                    Self::TRIGGER_KIND,
+                    listening_trigger.path,
+                    history_err
+                );
+                let (db, workspace_id) = (db.clone(), listening_trigger.workspace_id.clone());
+                tokio::spawn(async move {
+                    report_critical_error(message, db, Some(&workspace_id), None).await;
+                });
+            }
+
             match report_status {
-                Ok(_) => {
+                Ok(()) => {
                     report_critical_error(
                         format!(
                             "Disabling {} trigger {} because of error: {}",

backend/windmill-worker/src/worker_flow.rs

@@ -2923,15 +2923,39 @@ pub async fn handle_flow(
                     // its own disable write failed. Retry it: rearm_schedule turns
                     // these into NoOp, so without disabling here the schedule would
                     // stay enabled yet never run.
-                    if let Err(disable_err) = sqlx::query!(
-                        "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
-                        err.to_string(),
-                        &flow_job.workspace_id,
-                        &schedule.path
-                    )
-                    .execute(db)
-                    .await
-                    {
+                    // Contract on `record_in_disable_tx`. Worth knowing here:
+                    // this is the last chance to disable, because a flow
+                    // schedule arms its next occurrence when the flow *starts*,
+                    // so once the flow is gone nothing reaches this code again.
+                    let mut history_lost = None;
+                    let disable_result = async {
+                        let mut tx = db.begin().await?;
+                        let rows = sqlx::query!(
+                            "UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3 AND enabled = true",
+                            err.to_string(),
+                            &flow_job.workspace_id,
+                            &schedule.path
+                        )
+                        .execute(&mut *tx)
+                        .await?
+                        .rows_affected();
+                        if rows > 0 {
+                            history_lost = windmill_common::trigger_history::record_in_disable_tx(
+                                &mut tx,
+                                windmill_queue::jobs::schedule_auto_disable_event(
+                                    &flow_job.workspace_id,
+                                    &schedule.path,
+                                    &err.to_string(),
+                                ),
+                            )
+                            .await;
+                        }
+                        tx.commit().await?;
+                        Ok::<(), Error>(())
+                    }
+                    .await;
+
+                    if let Err(disable_err) = disable_result {
                         report_error_to_workspace_handler_or_critical_side_channel(
                             &mini_job,
                             db,
@@ -2942,6 +2966,17 @@ pub async fn handle_flow(
                         )
                         .await;
                     }
+                    if let Some(history_err) = history_lost {
+                        report_error_to_workspace_handler_or_critical_side_channel(
+                            &mini_job,
+                            db,
+                            format!(
+                                "Disabled schedule {} but could not record it in the trigger history: {history_err}",
+                                schedule.path,
+                            ),
+                        )
+                        .await;
+                    }
                 } else {
                     // Transient error (DB contention, timeout) after retry exhaustion:
                     // not the schedule's fault. Report it but leave the schedule

cli/src/core/client.ts

@@ -14,6 +14,17 @@ export function markRequestsAsSyncOrigin() {
   OpenAPI.HEADERS = { ...existing, "X-Windmill-Deploy-Origin": "sync" };
 }
 
+/**
+ * Name this process as the CLI on every subsequent request, so a trigger the
+ * CLI created or disabled is attributed to `cli` rather than to a bare API
+ * call in `trigger_history`. Attribution only — nothing on the server grants
+ * anything on the strength of it.
+ */
+export function markRequestsAsCliClient() {
+  const existing = typeof OpenAPI.HEADERS === "object" ? OpenAPI.HEADERS : {};
+  OpenAPI.HEADERS = { ...existing, "X-Windmill-Client": "cli" };
+}
+
 export function setClient(token?: string, baseUrl?: string) {
   if (baseUrl === undefined) {
     baseUrl = process.env["BASE_INTERNAL_URL"] ??

cli/src/main.ts

@@ -32,6 +32,7 @@ import { OpenAPI } from "../gen/index.ts";
 import { getHeaders } from "./utils/utils.ts";
 import { detectAuthGatewayChallenge } from "./utils/http_guards.ts";
 import { setShowDiffs } from "./core/conf.ts";
+import { markRequestsAsCliClient } from "./core/client.ts";
 import { NpmProvider } from "./utils/upgrade.ts";
 import { pull as hubPull } from "./commands/hub/hub.ts";
 import { pull, push } from "./commands/sync/sync.ts";
@@ -300,6 +301,7 @@ async function main() {
     if (extraHeaders) {
       OpenAPI.HEADERS = extraHeaders;
     }
+    markRequestsAsCliClient();
     OpenAPI.interceptors.response.use(async (response) => {
       await detectAuthGatewayChallenge(response);
       return response;

frontend/src/lib/components/triggers/TriggerEditorToolbar.svelte

@@ -5,7 +5,7 @@
 
 	import { Tooltip } from '../meltComponents'
 	import DeleteTriggerButton from './DeleteTriggerButton.svelte'
-	import { type Trigger } from './utils'
+	import { type Trigger, type TriggerType } from './utils'
 	import TriggerSuspendedJobsModal from './TriggerSuspendedJobsModal.svelte'
 	import type { TriggerMode } from '$lib/gen'
 	import TriggerModeToggle from './TriggerModeToggle.svelte'
@@ -14,6 +14,7 @@
 	import { pageDrawerSessionSource } from '../sessions/pageDrawerSession'
 	import { page } from '$app/state'
 	import { workspaceStore } from '$lib/stores'
+	import TriggerHistoryButton from './TriggerHistoryButton.svelte'
 
 	interface Props {
 		saveDisabled: any
@@ -35,6 +36,10 @@
 		/** Path of the trigger being edited, used to deep-link "Open in AI session"
 		 * at this trigger. Empty while creating one. */
 		triggerPath?: string
+		/** Kind the modification history is recorded under. Only the schedule
+		 * editor has to pass it: every other editor renders with a `trigger`,
+		 * whose `type` is the same value. */
+		triggerKind?: TriggerType
 	}
 
 	let {
@@ -54,7 +59,8 @@
 		trigger,
 		suspendedJobsModal,
 		disableSuspendedMode = false,
-		triggerPath
+		triggerPath,
+		triggerKind
 	}: Props = $props()
 
 	const canSave = $derived((permissions === 'write' && edit) || permissions === 'create')
@@ -77,10 +83,20 @@
 				)
 			: undefined
 	)
+
+	// Only a deployed trigger has a history: a draft has never been written.
+	// `triggerKind` is what opts an editor in, so the kinds `trigger_history`
+	// does not record (native triggers) simply never pass it.
+	const historyPath = $derived(
+		triggerKind && edit && !trigger?.isDraft ? triggerPath || trigger?.path : undefined
+	)
 </script>
 
 {#if !allowDraft}
 	{@render extra?.()}
+	{#if triggerKind && historyPath}
+		<TriggerHistoryButton {triggerKind} path={historyPath} />
+	{/if}
 	<OpenInSessionButton source={sessionSource} />
 	{#if edit}
 		<TriggerModeToggle
@@ -107,6 +123,9 @@
 	{/if}
 {:else}
 	<div class="flex flex-row gap-2 items-center">
+		{#if triggerKind && historyPath}
+			<TriggerHistoryButton {triggerKind} path={historyPath} />
+		{/if}
 		<OpenInSessionButton source={sessionSource} />
 		{#if !trigger?.draftConfig}
 			<div class="center-center">

frontend/src/lib/components/triggers/TriggerHistoryButton.svelte

@@ -0,0 +1,122 @@
+<script lang="ts">
+	import { History } from 'lucide-svelte'
+	import { TriggerService, type TriggerHistoryEntry } from '$lib/gen'
+	import { workspaceStore } from '$lib/stores'
+	import { displayDate } from '$lib/utils'
+	import Button from '../common/button/Button.svelte'
+	import Drawer from '../common/drawer/Drawer.svelte'
+	import DrawerContent from '../common/drawer/DrawerContent.svelte'
+	import Badge from '../common/badge/Badge.svelte'
+	import Skeleton from '../common/skeleton/Skeleton.svelte'
+	import TriggerHistoryChanges from './TriggerHistoryChanges.svelte'
+	import { getTriggerWorkspace } from './triggerWorkspace'
+	import type { TriggerType } from './utils'
+
+	interface Props {
+		/** Trigger kind as the backend records it: `schedule`, `http`, `kafka`, … */
+		triggerKind: TriggerType
+		path: string
+	}
+
+	let { triggerKind, path }: Props = $props()
+
+	// An AI session can edit a trigger in a workspace that is not the nav one;
+	// the whole trigger subtree reads its workspace through this seam.
+	const triggerWs = getTriggerWorkspace()
+	const wsId = $derived(triggerWs?.() ?? $workspaceStore)
+
+	let drawer: Drawer | undefined = $state()
+	let entries: TriggerHistoryEntry[] | undefined = $state(undefined)
+	let loading = $state(false)
+	let error: string | undefined = $state(undefined)
+
+	const PER_PAGE = 50
+
+	async function load() {
+		if (!wsId) return
+		loading = true
+		error = undefined
+		try {
+			entries = await TriggerService.listTriggerHistory({
+				workspace: wsId,
+				triggerKind,
+				path,
+				perPage: PER_PAGE
+			})
+		} catch (e) {
+			error = e?.body ?? e?.message ?? 'Could not load history'
+			entries = []
+		} finally {
+			loading = false
+		}
+	}
+
+	// `worker` is the one value that is not a person: the server disabled the
+	// trigger on its own, so it reads as a warning rather than as attribution.
+	const sourceColor = {
+		ui: 'blue',
+		cli: 'gray',
+		api: 'indigo',
+		worker: 'yellow'
+	} as const
+
+	const operationColor = {
+		create: 'green',
+		update: 'blue',
+		delete: 'red',
+		enable: 'green',
+		disable: 'red',
+		suspend: 'yellow'
+	} as const
+</script>
+
+<Drawer bind:this={drawer} size="800px">
+	<DrawerContent title="Trigger history" on:close={() => drawer?.closeDrawer()}>
+		{#if loading}
+			<div class="flex flex-col gap-2">
+				{#each new Array(3) as _, i (i)}
+					<Skeleton layout={[[4], 0.7]} />
+				{/each}
+			</div>
+		{:else if error}
+			<p class="text-sm text-red-500">{error}</p>
+		{:else if !entries || entries.length === 0}
+			<p class="text-sm text-secondary">No modification recorded yet</p>
+		{:else}
+			<div class="flex flex-col gap-2">
+				{#each entries as entry (entry.id)}
+					<div class="flex flex-col gap-2 border border-border-light rounded-md p-3">
+						<div class="flex flex-row gap-2 items-center flex-wrap">
+							<Badge color={operationColor[entry.operation] ?? 'gray'}>{entry.operation}</Badge>
+							<Badge color={sourceColor[entry.source] ?? 'gray'}>{entry.source}</Badge>
+							<span class="text-sm text-primary">
+								{entry.username ?? 'Windmill'}
+							</span>
+							<span class="text-xs text-secondary ml-auto">{displayDate(entry.created_at)}</span>
+						</div>
+						{#if entry.changes}
+							<TriggerHistoryChanges changes={entry.changes} />
+						{/if}
+					</div>
+				{/each}
+			</div>
+			{#if entries.length === PER_PAGE}
+				<p class="text-xs text-secondary mt-2">
+					Showing the {PER_PAGE} most recent modifications
+				</p>
+			{/if}
+		{/if}
+	</DrawerContent>
+</Drawer>
+
+<Button
+	unifiedSize="sm"
+	variant="subtle"
+	startIcon={{ icon: History }}
+	iconOnly
+	title="Modification history"
+	on:click={() => {
+		drawer?.openDrawer()
+		load()
+	}}
+/>

frontend/src/lib/components/triggers/TriggerHistoryChanges.svelte

@@ -0,0 +1,90 @@
+<script lang="ts">
+	import { ChevronDown, ChevronRight, Minus, Pencil, Plus } from 'lucide-svelte'
+	import Badge from '../common/badge/Badge.svelte'
+	import Button from '../common/button/Button.svelte'
+	import ObjectViewer from '../propertyPicker/ObjectViewer.svelte'
+	import { formatScalar, isComplex, parseChanges, type FieldChange } from './triggerHistoryChanges'
+
+	interface Props {
+		changes: unknown
+	}
+
+	let { changes }: Props = $props()
+
+	const parsed = $derived(parseChanges(changes))
+
+	let expanded: Record<string, boolean> = $state({})
+
+	const marker = {
+		added: { icon: Plus, class: 'text-green-500' },
+		removed: { icon: Minus, class: 'text-red-500' },
+		changed: { icon: Pencil, class: 'text-yellow-500' }
+	} as const
+
+	/** A row needs the tree viewer when either side is an object or an array. */
+	function needsTree(change: FieldChange): boolean {
+		return (
+			('next' in change && isComplex(change.next)) || ('prev' in change && isComplex(change.prev))
+		)
+	}
+</script>
+
+{#if parsed.kind === 'truncated'}
+	<div class="flex flex-col gap-1.5">
+		<span class="text-xs text-secondary">
+			{parsed.fields.length} fields changed, too large to store in full
+		</span>
+		<div class="flex flex-row flex-wrap gap-1">
+			{#each parsed.fields as field (field)}
+				<Badge color="gray" small>{field}</Badge>
+			{/each}
+		</div>
+	</div>
+{:else if parsed.kind === 'fields'}
+	<!-- One grid for the whole entry, so field names and values line up in
+	     columns however many rows it has. -->
+	<div class="grid grid-cols-[auto_max-content_minmax(0,1fr)] gap-x-2 items-baseline">
+		{#each parsed.changes as change (change.field)}
+			{@const Icon = marker[change.kind].icon}
+			{@const tree = needsTree(change)}
+			<Icon size={12} class={`${marker[change.kind].class} shrink-0 translate-y-0.5`} />
+			<span class="text-2xs font-mono text-emphasis py-0.5">{change.field}</span>
+			{#if tree}
+				<Button
+					unifiedSize="2xs"
+					variant="subtle"
+					wrapperClasses="w-fit"
+					btnClasses="!text-2xs !font-normal !px-1"
+					startIcon={{ icon: expanded[change.field] ? ChevronDown : ChevronRight }}
+					on:click={() => (expanded[change.field] = !expanded[change.field])}
+				>
+					{change.kind === 'changed' ? 'value changed' : 'value'}
+				</Button>
+			{:else}
+				<span class="text-2xs font-mono py-0.5 break-all">
+					{#if change.kind === 'changed' || (change.kind === 'removed' && change.prev !== undefined)}
+						<span class="text-tertiary line-through">{formatScalar(change.prev)}</span>
+						<span class="text-tertiary px-1">→</span>
+					{/if}
+					{#if change.kind === 'removed'}
+						<span class="text-tertiary italic">unset</span>
+					{:else}
+						<span class="text-emphasis">{formatScalar(change.next)}</span>
+					{/if}
+				</span>
+			{/if}
+			{#if tree && expanded[change.field]}
+				<div class="col-span-3 pl-6 pb-1 flex flex-col gap-1 min-w-0">
+					{#if change.kind !== 'added' && change.prev !== undefined}
+						<div class="text-2xs text-tertiary">before</div>
+						<ObjectViewer pureViewer json={change.prev} />
+					{/if}
+					{#if change.kind !== 'removed'}
+						<div class="text-2xs text-tertiary">after</div>
+						<ObjectViewer pureViewer json={change.next} />
+					{/if}
+				</div>
+			{/if}
+		{/each}
+	</div>
+{/if}

frontend/src/lib/components/triggers/amqp/AmqpTriggerEditorInner.svelte

@@ -445,6 +445,7 @@
 {#snippet actionsSnippet()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="amqp"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}

frontend/src/lib/components/triggers/azure/AzureTriggerEditorInner.svelte

@@ -404,6 +404,7 @@
 {#snippet actionsButtons()}
 	{#if !drawerLoading && can_write}
 		<TriggerEditorToolbar
+			triggerKind="azure"
 			triggerPath={initialPath}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}
 			{saveDisabled}

frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte

@@ -467,6 +467,7 @@
 {#snippet saveButton()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="email"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : can_write && isAdmin ? 'create' : 'write'}

frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte

@@ -415,6 +415,7 @@
 {#snippet actionsButtons()}
 	{#if !drawerLoading && can_write}
 		<TriggerEditorToolbar
+			triggerKind="gcp"
 			triggerPath={initialPath}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}
 			{saveDisabled}

frontend/src/lib/components/triggers/http/RouteEditorInner.svelte

@@ -991,6 +991,7 @@
 {#snippet saveButton()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="http"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}

frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte

@@ -465,6 +465,7 @@
 {#snippet actionsButtons(size: 'xs' | 'sm' = 'sm')}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="kafka"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}

frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte

@@ -439,6 +439,7 @@
 {#snippet actionsSnippet()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="mqtt"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}

frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte

@@ -436,6 +436,7 @@
 {#snippet actionsSnippet()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="nats"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}

frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte

@@ -618,6 +618,7 @@
 {#snippet actionsSnippet()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="postgres"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}

frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte

@@ -736,6 +736,7 @@
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
 			triggerPath={initialPath}
+			triggerKind="schedule"
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}
 			{saveDisabled}

frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte

@@ -418,6 +418,7 @@
 {#snippet actionsSnippet()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="sqs"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={drawerLoading || !can_write ? 'none' : 'create'}

frontend/src/lib/components/triggers/triggerHistoryChanges.test.ts

@@ -0,0 +1,35 @@
+import { describe, it, expect } from 'vitest'
+import { parseChanges } from './triggerHistoryChanges'
+
+// The viewer's whole layout rests on `changes` only ever arriving in these three
+// shapes; a backend that starts writing a fourth would otherwise render as an
+// empty entry rather than fail.
+describe('parseChanges', () => {
+	it('classifies each side by whether it holds a value', () => {
+		expect(
+			parseChanges({
+				schedule: { old: '0 0 * * *', new: '0 1 * * *' },
+				summary: { new: 'created' },
+				on_failure: { old: 'u/admin/handler', new: null },
+				description: { old: null, new: 'set now' }
+			})
+		).toEqual({
+			kind: 'fields',
+			changes: [
+				{ kind: 'added', field: 'description', next: 'set now' },
+				{ kind: 'removed', field: 'on_failure', prev: 'u/admin/handler' },
+				{ kind: 'changed', field: 'schedule', prev: '0 0 * * *', next: '0 1 * * *' },
+				{ kind: 'added', field: 'summary', next: 'created' }
+			]
+		})
+	})
+
+	it('reads the truncation marker and the empty cases', () => {
+		expect(parseChanges({ truncated_fields: ['args', 'schedule'] })).toEqual({
+			kind: 'truncated',
+			fields: ['args', 'schedule']
+		})
+		expect(parseChanges(null)).toEqual({ kind: 'none' })
+		expect(parseChanges({})).toEqual({ kind: 'none' })
+	})
+})

frontend/src/lib/components/triggers/triggerHistoryChanges.ts

@@ -0,0 +1,70 @@
+/**
+ * The `changes` column of `trigger_history`, in the three shapes the backend
+ * writes it (see `windmill-common/src/trigger_history.rs`):
+ *
+ *   null                                 no field-level diff — a delete
+ *   { truncated_fields: [...] }          the diff exceeded the 32 KiB cap
+ *   { <field>: { old?, new } }           the fields that changed
+ *
+ * Parsing it into this closed set up front is what lets the viewer lay every
+ * entry out the same way, instead of rendering whatever tree came back.
+ */
+
+/**
+ * How one field moved. A missing or null side is treated as the field having no
+ * value: the writer omits `old` where it never read one, and the backend writes
+ * `new: null` for a cleared column, so both collapse to added/removed.
+ */
+export type FieldChange =
+	| { kind: 'added'; field: string; next: unknown }
+	| { kind: 'removed'; field: string; prev: unknown | undefined }
+	| { kind: 'changed'; field: string; prev: unknown; next: unknown }
+
+export type ParsedChanges =
+	| { kind: 'none' }
+	| { kind: 'truncated'; fields: string[] }
+	| { kind: 'fields'; changes: FieldChange[] }
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+	return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+export function parseChanges(changes: unknown): ParsedChanges {
+	if (!isRecord(changes)) return { kind: 'none' }
+
+	const truncated = changes['truncated_fields']
+	if (Array.isArray(truncated)) {
+		return { kind: 'truncated', fields: truncated.map(String) }
+	}
+
+	const parsed: FieldChange[] = []
+	for (const [field, entry] of Object.entries(changes)) {
+		if (!isRecord(entry)) continue
+		// A null on either side is the field having no value, which reads as
+		// added/removed rather than as a change to or from `null`.
+		const prev = entry['old'] ?? undefined
+		const next = entry['new'] ?? undefined
+		if (next === undefined) {
+			parsed.push({ kind: 'removed', field, prev })
+		} else if (prev === undefined) {
+			parsed.push({ kind: 'added', field, next })
+		} else {
+			parsed.push({ kind: 'changed', field, prev, next })
+		}
+	}
+	// Alphabetical: jsonb hands back its own key order, which is by length then
+	// bytes and would shuffle as values change.
+	parsed.sort((a, b) => a.field.localeCompare(b.field))
+
+	return parsed.length === 0 ? { kind: 'none' } : { kind: 'fields', changes: parsed }
+}
+
+/** True when a value needs the tree viewer rather than a one-line rendering. */
+export function isComplex(value: unknown): boolean {
+	return typeof value === 'object' && value !== null
+}
+
+/** One-line rendering of a scalar, quoted so `""` and `"0"` stay visible. */
+export function formatScalar(value: unknown): string {
+	return typeof value === 'string' ? JSON.stringify(value) : String(value)
+}

frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte

@@ -506,6 +506,7 @@
 {#snippet actionsButtons()}
 	{#if !drawerLoading}
 		<TriggerEditorToolbar
+			triggerKind="websocket"
 			triggerPath={initialPath}
 			{trigger}
 			permissions={!drawerLoading && can_write ? 'create' : 'none'}