Add debug println to inkeep_not_available function

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json

@@ -46,11 +46,11 @@
       ]
     },
     "nullable": [
-      true,
-      true,
-      true,
-      true,
-      true,
+      false,
+      false,
+      false,
+      false,
+      false,
       true,
       true
     ]

backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json

@@ -30,8 +30,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-0f7e01b613a94b29784aae6d7b17b23d6dcf2e5364852a5e85b3c41c417bace2.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-1074c6c98e6a0c83ac04172a39abea21c793f58947051d39931d4da0868a1d77.json

@@ -0,0 +1,21 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        INSERT INTO email_trigger (\n            path, local_part, workspaced_local_part, script_path,\n            is_flow, workspace_id, edited_by, email\n        )\n        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)\n        ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1074c6c98e6a0c83ac04172a39abea21c793f58947051d39931d4da0868a1d77"
+}

backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json

@@ -122,8 +122,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-16d438374b03a9c515f4c2d638366f38ffe2f3a0958adea53e67757c6ac463ec.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json

@@ -40,8 +40,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-2031c5138a785367e5127180ccb6734efa41ee6cb3b4819c1c517798b2b23e4a.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE email_trigger SET script_path = $1, local_part = $2 WHERE workspace_id = $3 AND path = $4",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "2031c5138a785367e5127180ccb6734efa41ee6cb3b4819c1c517798b2b23e4a"
+}

backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json

@@ -0,0 +1,85 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            SELECT id, runnable_path, trigger_kind AS \"trigger_kind: String\",\n                   args AS \"args: sqlx::types::Json<serde_json::Value>\"\n            FROM v2_job\n            WHERE runnable_path = $1\n              AND trigger_kind = $2::job_trigger_kind\n            ORDER BY created_at DESC\n            LIMIT 1\n            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "runnable_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "trigger_kind: String",
+        "type_info": {
+          "Custom": {
+            "name": "job_trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats",
+                "schedule",
+                "app",
+                "ui",
+                "postgres",
+                "sqs",
+                "gcp",
+                "mqtt",
+                "nextcloud"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 3,
+        "name": "args: sqlx::types::Json<serde_json::Value>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        {
+          "Custom": {
+            "name": "job_trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats",
+                "schedule",
+                "app",
+                "ui",
+                "postgres",
+                "sqs",
+                "gcp",
+                "mqtt",
+                "nextcloud"
+              ]
+            }
+          }
+        }
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8"
+}

backend/.sqlx/query-22e0e8a1aa48f8b21763452bd36fbe7db4887c4ac5295052c796bd78a7edc50b.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-23419adcd74c326d716527293eff518b42f4cdb33e034441015494bd26c172d2.json

@@ -40,8 +40,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-302bc55d0c227c5b12458ccde6569c4b531ff494d1ede0a655872ae05215f8a6.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT value FROM variable WHERE workspace_id = $1 AND path = $2 AND is_secret = true",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "302bc55d0c227c5b12458ccde6569c4b531ff494d1ede0a655872ae05215f8a6"
+}

backend/.sqlx/query-33fff66efe810ad7e92b36ca9b287938437182d7817387707e800519d9f5bafc.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3 AND is_secret = true",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "33fff66efe810ad7e92b36ca9b287938437182d7817387707e800519d9f5bafc"
+}

backend/.sqlx/query-3481e65196e500ad914f10b87884c19d8a3636ae955788c3366365194fe2ef57.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "DELETE FROM raw_app WHERE path = $1 AND workspace_id = $2",
+  "query": "DELETE FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2",
   "describe": {
     "columns": [],
     "parameters": {
@@ -11,5 +11,5 @@
     },
     "nullable": []
   },
-  "hash": "b2b2b7251be2b80207f47b10eeff78d61c84161caaf16b8fd25d82b97aac1186"
+  "hash": "3481e65196e500ad914f10b87884c19d8a3636ae955788c3366365194fe2ef57"
 }

backend/.sqlx/query-40971d637c5b4d2af8e67872722880058cc067fca5e807ab3e1ed17d180cb7f7.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM email_trigger WHERE workspace_id = $1 AND path = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "40971d637c5b4d2af8e67872722880058cc067fca5e807ab3e1ed17d180cb7f7"
+}

backend/.sqlx/query-42b4b73e9d60348e2d90fcade9dcad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-45024b932383199974616bba1fc2f7175cc6f2e02d9c565bb5159cae3e0b6835.json

@@ -30,8 +30,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-4d272cf4a77aab7007a5b35589e08532a1020cabaf5e22325a1e05f0491d785c.json

@@ -37,8 +37,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-4f547c0fd54f3bc57212ce87810e35adf640d44d607e62a1fb296e38ac3fdd36.json

@@ -32,8 +32,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }
@@ -71,8 +70,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json

@@ -245,8 +245,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-60a666a7cffce2d7631682095ccfd31e99997e88374496d4800eaa7b5152a464.json

@@ -0,0 +1,39 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT delivery_type AS \"delivery_type: String\",\n               delivery_config\n        FROM gcp_trigger\n        WHERE workspace_id = $1 AND path = $2\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "delivery_type: String",
+        "type_info": {
+          "Custom": {
+            "name": "delivery_mode",
+            "kind": {
+              "Enum": [
+                "push",
+                "pull"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 1,
+        "name": "delivery_config",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      true
+    ]
+  },
+  "hash": "60a666a7cffce2d7631682095ccfd31e99997e88374496d4800eaa7b5152a464"
+}

backend/.sqlx/query-66a0e51cf149ba532463e29dd361a803e1bced2f8e1a12f8933b7598ee85a147.json

@@ -35,8 +35,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-7065f23d04e26831664048f2cfc4f412c57af931f80621aee5012e9cb3535626.json

@@ -29,8 +29,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-73fdd01bad58b8be1a52f89faef8d92a983470adcd3cc850734960c905e61e83.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json

@@ -40,8 +40,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-7fbf72d9059fcd77e4c1112fa4fa22e4276c1da653475628889ce17dc904fbaa.json

@@ -27,8 +27,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-8002f7f1521846056278f18e8b003fe48a4d3d827ac8b212e0f3d3bc2ed42f37.json

@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT script_path, local_part FROM email_trigger WHERE workspace_id = $1 AND path = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "script_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "local_part",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "8002f7f1521846056278f18e8b003fe48a4d3d827ac8b212e0f3d3bc2ed42f37"
+}

backend/.sqlx/query-8364505b9a4d0d6f7406e4a5425360682ac2295bbf6746c7641141ed7bf47bf6.json

@@ -1,12 +1,12 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT EXISTS(SELECT 1 FROM raw_app WHERE path = $1 AND workspace_id = $2)",
+  "query": "SELECT COUNT(*) FROM email_trigger WHERE workspace_id = $1 AND path = $2",
   "describe": {
     "columns": [
       {
         "ordinal": 0,
-        "name": "exists",
-        "type_info": "Bool"
+        "name": "count",
+        "type_info": "Int8"
       }
     ],
     "parameters": {
@@ -19,5 +19,5 @@
       null
     ]
   },
-  "hash": "c4b3ab3d87f158b20f0557f78c6307c50620425aeb43636e19931c142b5ce0d7"
+  "hash": "8364505b9a4d0d6f7406e4a5425360682ac2295bbf6746c7641141ed7bf47bf6"
 }

backend/.sqlx/query-836bac47d89113d90bd03a471446eb9016207975af1e37042d81df8cb6ae2c53.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-84bbd11f7bb0c65dcfb2e12038b8a9cb51d303480ac3c5f8a5e833a53c40ed9b.json

@@ -1,17 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO raw_app\n            (workspace_id, path, summary, extra_perms, data)\n            VALUES ($1, $2, $3, '{}', $4)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "84bbd11f7bb0c65dcfb2e12038b8a9cb51d303480ac3c5f8a5e833a53c40ed9b"
-}

backend/.sqlx/query-87564a196a1662f524407d853db506bf08c28efe82b68b3d44bafbd3d0e91c29.json

@@ -35,8 +35,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-899a162192d93c7fb7d47d38f4fb0fadcafc2bb0c61efa8ee496af7252d18270.json

@@ -0,0 +1,62 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT mqtt_resource_path, client_version AS \"client_version: String\",\n               script_path, mode AS \"mode: String\"\n        FROM mqtt_trigger\n        WHERE workspace_id = $1 AND path = $2\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "mqtt_resource_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "client_version: String",
+        "type_info": {
+          "Custom": {
+            "name": "mqtt_client_version",
+            "kind": {
+              "Enum": [
+                "v3",
+                "v5"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 2,
+        "name": "script_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "mode: String",
+        "type_info": {
+          "Custom": {
+            "name": "trigger_mode",
+            "kind": {
+              "Enum": [
+                "enabled",
+                "disabled",
+                "suspended"
+              ]
+            }
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "899a162192d93c7fb7d47d38f4fb0fadcafc2bb0c61efa8ee496af7252d18270"
+}

backend/.sqlx/query-8c3cc09c1bbb6209467c75723dd02e97dddb99789422012e85e86c3151a9f2e9.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        INSERT INTO kafka_trigger (\n            path, kafka_resource_path, topics, group_id,\n            script_path, is_flow, workspace_id, edited_by, email\n        )\n        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)\n        ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "VarcharArray",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "8c3cc09c1bbb6209467c75723dd02e97dddb99789422012e85e86c3151a9f2e9"
+}

backend/.sqlx/query-940b6d78bab940a37a42492f030d2393e297043e4e58555d872b5c4dd89c196a.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-9c50e3a136a8ee3ec56e083f26d3a960b89e02ec40b292f3b5198baf2a1d3dbf.json

@@ -32,8 +32,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json

@@ -30,8 +30,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-a086a5fa28dcff80911c7e5de73f204fc42625e58b967b0376ff7b382e38fa11.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT script_path FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "script_path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "a086a5fa28dcff80911c7e5de73f204fc42625e58b967b0376ff7b382e38fa11"
+}

backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json

@@ -155,8 +155,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json

@@ -185,8 +185,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-a80a17cda8ad722a97f16526d8576b76c7377066a06dba98a2921a68928aa644.json

@@ -0,0 +1,84 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT gcp_resource_path, topic_id, subscription_id,\n               delivery_type AS \"delivery_type: String\",\n               subscription_mode AS \"subscription_mode: String\",\n               mode AS \"mode: String\"\n        FROM gcp_trigger\n        WHERE workspace_id = $1 AND path = $2\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "gcp_resource_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "topic_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "subscription_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "delivery_type: String",
+        "type_info": {
+          "Custom": {
+            "name": "delivery_mode",
+            "kind": {
+              "Enum": [
+                "push",
+                "pull"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 4,
+        "name": "subscription_mode: String",
+        "type_info": {
+          "Custom": {
+            "name": "gcp_subscription_mode",
+            "kind": {
+              "Enum": [
+                "create_update",
+                "existing"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 5,
+        "name": "mode: String",
+        "type_info": {
+          "Custom": {
+            "name": "trigger_mode",
+            "kind": {
+              "Enum": [
+                "enabled",
+                "disabled",
+                "suspended"
+              ]
+            }
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "a80a17cda8ad722a97f16526d8576b76c7377066a06dba98a2921a68928aa644"
+}

backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json

@@ -160,8 +160,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-b3f0595cacba194e08b9a3e244d9e637e9e156cd85b69126c87dfff89a47711d.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json

@@ -105,8 +105,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-c3b1152b554812d65eb27f95b1fd434f860922fbc021185beffb9827647feb8e.json

@@ -31,8 +31,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-cabef30aa1e308bc656660c4ac0d3e77b020de05465f81a4fb4b0c07bd0d7439.json

@@ -0,0 +1,52 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT local_part, workspaced_local_part, script_path,\n               mode AS \"mode: String\"\n        FROM email_trigger\n        WHERE workspace_id = $1 AND path = $2\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "local_part",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspaced_local_part",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "script_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "mode: String",
+        "type_info": {
+          "Custom": {
+            "name": "trigger_mode",
+            "kind": {
+              "Enum": [
+                "enabled",
+                "disabled",
+                "suspended"
+              ]
+            }
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "cabef30aa1e308bc656660c4ac0d3e77b020de05465f81a4fb4b0c07bd0d7439"
+}

backend/.sqlx/query-ccef7a1bde5cac6c362c5fedb6c13f1f882b695f896f94e5cf91d205633355a1.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json

@@ -105,8 +105,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-d495c94b580fd34d5ae90615ef21a8a9cc35f362197c0766a5787436af141106.json

@@ -25,8 +25,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-d7d11b2c288fa0282539edde0c5450e4b5f8355821617b6c1ce3db64f6ceaf69.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT COUNT(*) as count FROM variable WHERE is_secret = true AND value != 'CLEARED'",
+  "query": "SELECT COUNT(*) FROM variable WHERE is_secret = true",
   "describe": {
     "columns": [
       {
@@ -16,5 +16,5 @@
       null
     ]
   },
-  "hash": "052d42b46d5faba6b41f1fdcbf6a012db51b9e5a255ec0da9a8a0999d668d336"
+  "hash": "d7d11b2c288fa0282539edde0c5450e4b5f8355821617b6c1ce3db64f6ceaf69"
 }

backend/.sqlx/query-e485c82978d10379c6d1b7cd850f3bf764e6c5bf775f3d710ba36a79f5e424eb.json

@@ -0,0 +1,21 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        INSERT INTO websocket_trigger (\n            path, url, script_path, is_flow, workspace_id,\n            edited_by, email, initial_messages\n        )\n        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)\n        ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "JsonbArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e485c82978d10379c6d1b7cd850f3bf764e6c5bf775f3d710ba36a79f5e424eb"
+}

backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json

@@ -185,8 +185,7 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-e80177f3ffd4c1f52cdb4757483f03f72ef81db302d727e18e63a307ac902022.json

@@ -31,8 +31,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-eac595e19e5c8e70f1514ef29dec35c7342ac9a814c73f6290e1d6ebd3a55423.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-ed8facbf29ebb670d05fe8aa34b50d6a6935420fbedc83aa3ad1e9be7465c8dd.json

@@ -24,8 +24,7 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud",
-                "google"
+                "nextcloud"
               ]
             }
           }

backend/.sqlx/query-f0eb9d49021668673880aa20227c24a130cada07fba3f43e32d904fde97106d3.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE mqtt_trigger SET script_path = $1 WHERE workspace_id = $2 AND path = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f0eb9d49021668673880aa20227c24a130cada07fba3f43e32d904fde97106d3"
+}

backend/.sqlx/query-f149d740301a65a62664fd407abe89c972e50837b6c54ac7a3ccd5f60ca05d3c.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT COUNT(*) FROM raw_app WHERE workspace_id = $1",
+  "query": "SELECT COUNT(*) FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2",
   "describe": {
     "columns": [
       {
@@ -11,6 +11,7 @@
     ],
     "parameters": {
       "Left": [
+        "Text",
         "Text"
       ]
     },
@@ -18,5 +19,5 @@
       null
     ]
   },
-  "hash": "3b5295a7c4b99aefa52c9a8ae1e0dd12bf4a0be1bf755caf7a1fa863e7950562"
+  "hash": "f149d740301a65a62664fd407abe89c972e50837b6c54ac7a3ccd5f60ca05d3c"
 }

backend/Cargo.lock

@@ -15656,6 +15656,10 @@ name = "windmill"
 version = "1.628.3"
 dependencies = [
  "anyhow",
+ "async-nats",
+ "aws-config",
+ "aws-credential-types",
+ "aws-sdk-sqs",
  "axum 0.7.9",
  "base64 0.22.1",
  "chrono",
@@ -15669,7 +15673,9 @@ dependencies = [
  "once_cell",
  "prometheus",
  "rand 0.9.0",
+ "rdkafka",
  "reqwest 0.13.1",
+ "rumqttc",
  "rustls 0.23.35",
  "serde",
  "serde_derive",

backend/Cargo.toml

@@ -229,6 +229,12 @@ axum.workspace = true
 serde.workspace = true
 windmill-api-client.workspace = true
 tempfile.workspace = true
+rumqttc.workspace = true
+rdkafka.workspace = true
+async-nats.workspace = true
+aws-sdk-sqs.workspace = true
+aws-config.workspace = true
+aws-credential-types.workspace = true
 
 
 [workspace.dependencies]

backend/parsers/windmill-parser-py/src/lib.rs

@@ -422,11 +422,15 @@ fn parse_expr(
     match e.as_ref() {
         Expr::Name(ExprName { id, .. }) => (parse_typ(id.as_ref(), enums, module), false),
         Expr::Attribute(x) => {
-            if x.value
-                .as_name_expr()
-                .is_some_and(|x| x.id.as_str() == "wmill")
-            {
-                (parse_typ(x.attr.as_str(), enums, module), false)
+            if let Some(name) = x.value.as_name_expr() {
+                match name.id.as_str() {
+                    "wmill" => (parse_typ(x.attr.as_str(), enums, module), false),
+                    "datetime" => {
+                        let full_name = format!("datetime.{}", x.attr.as_str());
+                        (parse_typ(&full_name, enums, module), false)
+                    }
+                    _ => (Typ::Unknown, false),
+                }
             } else {
                 (Typ::Unknown, false)
             }
@@ -493,6 +497,8 @@ fn parse_typ(id: &str, enums: &HashMap<String, EnumInfo>, module: Option<&[Stmt]
         "bytes" => Typ::Bytes,
         "datetime" => Typ::Datetime,
         "datetime.datetime" => Typ::Datetime,
+        "date" => Typ::Date,
+        "datetime.date" => Typ::Date,
         "Sql" | "sql" => Typ::Sql,
         x @ _ if x.starts_with("DynSelect_") => {
             Typ::DynSelect(x.strip_prefix("DynSelect_").unwrap().to_string())
@@ -620,7 +626,7 @@ def main(test1: str, name: datetime.datetime = datetime.now(), byte: bytes = byt
                     Arg {
                         otyp: None,
                         name: "name".to_string(),
-                        typ: Typ::Unknown,
+                        typ: Typ::Datetime,
                         default: Some(json!("<function call>")),
                         has_default: true,
                         oidx: None
@@ -709,7 +715,7 @@ def main(test1: str,
                     Arg {
                         otyp: None,
                         name: "name".to_string(),
-                        typ: Typ::Unknown,
+                        typ: Typ::Datetime,
                         default: Some(json!("<function call>")),
                         has_default: true,
                         oidx: None

backend/parsers/windmill-parser-py/src/pydantic_parser.rs

@@ -295,6 +295,7 @@ fn extract_field_type(
             "bool" => Typ::Bool,
             "bytes" => Typ::Bytes,
             "datetime" => Typ::Datetime,
+            "date" => Typ::Date,
             "Any" => Typ::Unknown, // typing.Any maps to Unknown
             // Custom class - check if it's a model
             custom_type => {

backend/parsers/windmill-parser/src/lib.rs

@@ -65,6 +65,7 @@ pub enum Typ {
     List(Box<Typ>),
     Bytes,
     Datetime,
+    Date,
     Resource(String),
     Email,
     Sql,

backend/parsers/windmill-sql-datatype-parser-wasm/src/lib.rs

@@ -10,6 +10,7 @@ fn to_str(typ: Typ) -> String {
         Typ::List(t) => format!("list-{}", to_str(*t)),
         Typ::Bytes => "bytes".to_string(),
         Typ::Datetime => "datetime".to_string(),
+        Typ::Date => "date".to_string(),
         _ => "unknown".to_string(),
     }
 }

backend/tests/common/mod.rs

@@ -87,6 +87,12 @@ impl ApiServer {
         Self::start_inner(db, true).await
     }
 
+    /// Start the API server with server_mode=true so trigger listeners are active.
+    /// Alias for `start_agent_mode` with a clearer name for trigger e2e tests.
+    pub async fn start_with_listeners(db: Pool<Postgres>) -> anyhow::Result<Self> {
+        Self::start_inner(db, true).await
+    }
+
     async fn start_inner(db: Pool<Postgres>, agent_mode: bool) -> anyhow::Result<Self> {
         let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);
 

backend/tests/fixtures/secret_backend.sql

@@ -14,18 +14,18 @@ INSERT INTO workspace_key(workspace_id, kind, key)
 VALUES ('test-workspace-2', 'cloud', 'test-key-2')
 ON CONFLICT DO NOTHING;
 
--- Insert test secrets for workspace 1
--- Note: The 'value' column stores encrypted values in production,
--- but for tests we'll use plain text that the migration will handle
+-- Insert test variables with placeholder values.
+-- Secret values are encrypted by the test setup using build_crypt + encrypt
+-- with the workspace key, matching production behavior.
 INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms)
 VALUES
-    ('test-workspace', 'u/test-user/db_password', 'encrypted-db-pass-123', true, 'Database password', '{}'),
-    ('test-workspace', 'u/test-user/api_key', 'encrypted-api-key-abc', true, 'API key for external service', '{}'),
+    ('test-workspace', 'u/test-user/db_password', 'PLACEHOLDER', true, 'Database password', '{}'),
+    ('test-workspace', 'u/test-user/api_key', 'PLACEHOLDER', true, 'API key for external service', '{}'),
     ('test-workspace', 'u/test-user/public_var', 'not-a-secret', false, 'A non-secret variable', '{}')
 ON CONFLICT DO NOTHING;
 
 -- Insert test secrets for workspace 2 (to test isolation)
 INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms)
 VALUES
-    ('test-workspace-2', 'u/test-user/other_secret', 'encrypted-other-secret', true, 'Secret in workspace 2', '{}')
+    ('test-workspace-2', 'u/test-user/other_secret', 'PLACEHOLDER', true, 'Secret in workspace 2', '{}')
 ON CONFLICT DO NOTHING;

backend/tests/fixtures/start_all_triggers.sh

@@ -0,0 +1,58 @@
+#!/usr/bin/env bash
+# Starts (or stops) all external services needed for the trigger e2e tests.
+#
+# Usage:
+#   ./tests/fixtures/start_all_triggers.sh        # start everything
+#   ./tests/fixtures/start_all_triggers.sh stop   # stop everything
+#   ./tests/fixtures/start_all_triggers.sh oss    # start only OSS services
+set -euo pipefail
+
+DIR="$(cd "$(dirname "$0")" && pwd)"
+ACTION="${1:-start}"
+
+SCRIPTS_OSS=(
+    "$DIR/start_mqtt.sh"
+    "$DIR/start_websocket.sh"
+    "$DIR/start_postgres_replication.sh"
+)
+
+SCRIPTS_EE=(
+    "$DIR/start_kafka.sh"
+    "$DIR/start_nats.sh"
+    "$DIR/start_sqs.sh"
+    "$DIR/start_gcp_pubsub.sh"
+)
+
+if [[ "$ACTION" == "stop" ]]; then
+    for s in "${SCRIPTS_OSS[@]}" "${SCRIPTS_EE[@]}"; do
+        echo "--- $(basename "$s" .sh) stop ---"
+        bash "$s" stop
+    done
+    exit 0
+fi
+
+if [[ "$ACTION" == "oss" ]]; then
+    SCRIPTS=("${SCRIPTS_OSS[@]}")
+else
+    SCRIPTS=("${SCRIPTS_OSS[@]}" "${SCRIPTS_EE[@]}")
+fi
+
+for s in "${SCRIPTS[@]}"; do
+    echo "--- $(basename "$s" .sh) ---"
+    bash "$s"
+    echo ""
+done
+
+echo "============================================"
+echo "All services ready. Run the e2e tests with:"
+echo ""
+
+if [[ "$ACTION" == "oss" ]]; then
+    echo "  cargo test --test trigger_e2e --features mqtt_trigger,websocket,postgres_trigger -- --ignored --nocapture"
+else
+    echo "  # OSS triggers"
+    echo "  cargo test --test trigger_e2e --features mqtt_trigger,websocket,postgres_trigger -- --ignored --nocapture"
+    echo ""
+    echo "  # Enterprise triggers"
+    echo "  AWS_ENDPOINT_URL=http://localhost:4566 PUBSUB_EMULATOR_HOST=localhost:8085 cargo test --test trigger_e2e --features kafka,nats,sqs_trigger,gcp_trigger,enterprise,private -- --ignored --nocapture"
+fi

backend/tests/fixtures/start_gcp_pubsub.sh

@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+# Starts the GCP Pub/Sub emulator for trigger_e2e::test_gcp_e2e (Enterprise)
+#
+# Usage:
+#   ./tests/fixtures/start_gcp_pubsub.sh        # start
+#   ./tests/fixtures/start_gcp_pubsub.sh stop   # stop & remove
+set -euo pipefail
+
+NAME="windmill-test-pubsub"
+PORT=8085
+
+if [[ "${1:-}" == "stop" ]]; then
+    docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running"
+    exit 0
+fi
+
+if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then
+    echo "$NAME is already running"
+    exit 0
+fi
+
+docker rm -f "$NAME" 2>/dev/null || true
+
+docker run -d --name "$NAME" -p "${PORT}:8085" \
+    gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators \
+    gcloud beta emulators pubsub start --host-port="0.0.0.0:${PORT}"
+
+echo "Waiting for Pub/Sub emulator to become ready..."
+for i in $(seq 1 30); do
+    if curl -sf "http://localhost:${PORT}" &>/dev/null; then
+        break
+    fi
+    sleep 1
+done
+
+# Create the test topic and subscription
+curl -sX PUT "http://localhost:${PORT}/v1/projects/local-project/topics/windmill-e2e-test" >/dev/null
+curl -sX PUT "http://localhost:${PORT}/v1/projects/local-project/subscriptions/windmill-e2e-sub" \
+    -H "Content-Type: application/json" \
+    -d '{"topic": "projects/local-project/topics/windmill-e2e-test"}' >/dev/null
+
+echo "GCP Pub/Sub emulator listening on localhost:${PORT}"
+echo "  topic:        windmill-e2e-test"
+echo "  subscription: windmill-e2e-sub"
+echo ""
+echo "Run the test:"
+echo "  PUBSUB_EMULATOR_HOST=localhost:${PORT} cargo test --test trigger_e2e test_gcp_e2e --features gcp_trigger,enterprise,private -- --ignored --nocapture"

backend/tests/fixtures/start_kafka.sh

@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+# Starts a Kafka broker for trigger_e2e::test_kafka_e2e (Enterprise)
+#
+# Usage:
+#   ./tests/fixtures/start_kafka.sh        # start
+#   ./tests/fixtures/start_kafka.sh stop   # stop & remove
+set -euo pipefail
+
+NAME="windmill-test-kafka"
+PORT=9092
+
+if [[ "${1:-}" == "stop" ]]; then
+    docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running"
+    exit 0
+fi
+
+if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then
+    echo "$NAME is already running"
+    exit 0
+fi
+
+docker rm -f "$NAME" 2>/dev/null || true
+
+docker run -d --name "$NAME" -p "${PORT}:9092" \
+    -e KAFKA_NODE_ID=1 \
+    -e KAFKA_PROCESS_ROLES=broker,controller \
+    -e KAFKA_LISTENERS="PLAINTEXT://0.0.0.0:${PORT},CONTROLLER://0.0.0.0:9093" \
+    -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
+    -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
+    -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
+    -e KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://localhost:${PORT}" \
+    -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
+    -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
+    apache/kafka:latest
+
+echo "Waiting for Kafka to become ready..."
+for i in $(seq 1 60); do
+    if docker exec "$NAME" /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server "localhost:${PORT}" &>/dev/null; then
+        break
+    fi
+    sleep 1
+done
+
+docker exec "$NAME" /opt/kafka/bin/kafka-topics.sh --create \
+    --topic windmill-e2e-test \
+    --bootstrap-server "localhost:${PORT}" \
+    --partitions 1 --replication-factor 1 2>/dev/null || true
+
+echo "Kafka broker listening on localhost:${PORT} with topic 'windmill-e2e-test'"
+echo ""
+echo "Run the test:"
+echo "  cargo test --test trigger_e2e test_kafka_e2e --features kafka,enterprise,private -- --ignored --nocapture"

backend/tests/fixtures/start_mqtt.sh

@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+# Starts a Mosquitto MQTT broker for trigger_e2e::test_mqtt_e2e
+#
+# Usage:
+#   ./tests/fixtures/start_mqtt.sh        # start
+#   ./tests/fixtures/start_mqtt.sh stop   # stop & remove
+set -euo pipefail
+
+NAME="windmill-test-mqtt"
+PORT=1883
+
+if [[ "${1:-}" == "stop" ]]; then
+    docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running"
+    exit 0
+fi
+
+if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then
+    echo "$NAME is already running"
+    exit 0
+fi
+
+docker rm -f "$NAME" 2>/dev/null || true
+
+docker run -d --name "$NAME" -p "${PORT}:1883" \
+    eclipse-mosquitto:latest \
+    mosquitto -c /mosquitto-no-auth.conf
+
+echo "MQTT broker listening on localhost:${PORT}"
+echo ""
+echo "Run the test:"
+echo "  cargo test --test trigger_e2e test_mqtt_e2e --features mqtt_trigger -- --ignored --nocapture"

backend/tests/fixtures/start_nats.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# Starts a NATS server for trigger_e2e::test_nats_e2e (Enterprise)
+#
+# Usage:
+#   ./tests/fixtures/start_nats.sh        # start
+#   ./tests/fixtures/start_nats.sh stop   # stop & remove
+set -euo pipefail
+
+NAME="windmill-test-nats"
+PORT=4222
+
+if [[ "${1:-}" == "stop" ]]; then
+    docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running"
+    exit 0
+fi
+
+if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then
+    echo "$NAME is already running"
+    exit 0
+fi
+
+docker rm -f "$NAME" 2>/dev/null || true
+
+docker run -d --name "$NAME" -p "${PORT}:4222" nats:latest
+
+echo "NATS server listening on localhost:${PORT}"
+echo ""
+echo "Run the test:"
+echo "  cargo test --test trigger_e2e test_nats_e2e --features nats,enterprise,private -- --ignored --nocapture"

backend/tests/fixtures/start_postgres_replication.sh

@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+# Configures the local PostgreSQL for logical replication trigger tests.
+#
+# Prerequisites: wal_level=logical must be set (requires PG restart).
+# Check with:  SHOW wal_level;
+#
+# Usage:
+#   ./tests/fixtures/start_postgres_replication.sh          # setup
+#   ./tests/fixtures/start_postgres_replication.sh stop     # teardown
+set -euo pipefail
+
+PGURL="${DATABASE_URL:-postgres://postgres:changeme@localhost:5432/windmill}"
+
+if [[ "${1:-}" == "stop" ]]; then
+    psql "$PGURL" <<'SQL'
+SELECT pg_drop_replication_slot('test_e2e_slot')
+    WHERE EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'test_e2e_slot');
+DROP PUBLICATION IF EXISTS test_e2e_pub;
+DROP TABLE IF EXISTS test_trigger_table;
+SQL
+    echo "Postgres replication teardown complete"
+    exit 0
+fi
+
+# Check wal_level
+WAL_LEVEL=$(psql "$PGURL" -tAc "SHOW wal_level;")
+if [[ "$WAL_LEVEL" != "logical" ]]; then
+    echo "ERROR: wal_level is '$WAL_LEVEL', must be 'logical'"
+    echo ""
+    echo "Fix with:"
+    echo "  psql \"$PGURL\" -c \"ALTER SYSTEM SET wal_level = logical;\""
+    echo "  # then restart PostgreSQL"
+    exit 1
+fi
+
+psql "$PGURL" <<'SQL'
+CREATE TABLE IF NOT EXISTS test_trigger_table (id serial PRIMARY KEY, data text);
+
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = 'test_e2e_pub') THEN
+        CREATE PUBLICATION test_e2e_pub FOR TABLE test_trigger_table;
+    END IF;
+END $$;
+
+SELECT pg_create_logical_replication_slot('test_e2e_slot', 'pgoutput')
+    WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'test_e2e_slot');
+SQL
+
+echo "Postgres logical replication ready (publication=test_e2e_pub, slot=test_e2e_slot)"
+echo ""
+echo "Run the test:"
+echo "  cargo test --test trigger_e2e test_postgres_e2e --features postgres_trigger -- --ignored --nocapture"

backend/tests/fixtures/start_sqs.sh

@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+# Starts LocalStack for SQS trigger_e2e::test_sqs_e2e (Enterprise)
+#
+# Usage:
+#   ./tests/fixtures/start_sqs.sh        # start
+#   ./tests/fixtures/start_sqs.sh stop   # stop & remove
+set -euo pipefail
+
+NAME="windmill-test-localstack"
+PORT=4566
+
+if [[ "${1:-}" == "stop" ]]; then
+    docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running"
+    exit 0
+fi
+
+if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then
+    echo "$NAME is already running"
+    exit 0
+fi
+
+docker rm -f "$NAME" 2>/dev/null || true
+
+docker run -d --name "$NAME" -p "${PORT}:4566" \
+    -e SERVICES=sqs \
+    localstack/localstack
+
+echo "Waiting for LocalStack to become ready..."
+for i in $(seq 1 30); do
+    if curl -sf "http://localhost:${PORT}/_localstack/health" &>/dev/null; then
+        break
+    fi
+    sleep 1
+done
+
+# Create the test queue
+aws --endpoint-url="http://localhost:${PORT}" \
+    --region us-east-1 \
+    --no-sign-request \
+    sqs create-queue --queue-name windmill-e2e-test 2>/dev/null || true
+
+echo "LocalStack SQS listening on localhost:${PORT} with queue 'windmill-e2e-test'"
+echo ""
+echo "Run the test:"
+echo "  AWS_ENDPOINT_URL=http://localhost:${PORT} cargo test --test trigger_e2e test_sqs_e2e --features sqs_trigger,enterprise,private -- --ignored --nocapture"

backend/tests/fixtures/start_websocket.sh

@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+# Starts a WebSocket echo server for trigger_e2e::test_websocket_e2e
+#
+# Usage:
+#   ./tests/fixtures/start_websocket.sh        # start
+#   ./tests/fixtures/start_websocket.sh stop   # stop & remove
+set -euo pipefail
+
+NAME="windmill-test-ws-echo"
+PORT=8765
+
+if [[ "${1:-}" == "stop" ]]; then
+    docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running"
+    exit 0
+fi
+
+if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then
+    echo "$NAME is already running"
+    exit 0
+fi
+
+docker rm -f "$NAME" 2>/dev/null || true
+
+docker run -d --name "$NAME" -p "${PORT}:8080" \
+    -e PORT=8080 \
+    jmalloc/echo-server
+
+echo "WebSocket echo server listening on localhost:${PORT}"
+echo ""
+echo "Run the test:"
+echo "  cargo test --test trigger_e2e test_websocket_e2e --features websocket -- --ignored --nocapture"

backend/tests/secret_backend_integration.rs

@@ -26,28 +26,56 @@
 #[cfg(all(feature = "private", feature = "enterprise"))]
 mod tests {
     use sqlx::{Pool, Postgres};
-    use std::collections::HashMap;
     use windmill_common::secret_backend::{
         migrate_secrets_to_database, migrate_secrets_to_vault, test_vault_connection,
         SecretBackend, VaultBackend, VaultSettings,
     };
+    use windmill_common::variables::{build_crypt, decrypt, encrypt};
+
+    /// Plaintext values for test secrets. The fixture inserts PLACEHOLDERs that
+    /// `encrypt_fixture_secrets` replaces with properly encrypted values.
+    const TEST_SECRETS: &[(&str, &str, &str)] = &[
+        ("test-workspace", "u/test-user/db_password", "db-pass-123"),
+        ("test-workspace", "u/test-user/api_key", "api-key-abc"),
+        (
+            "test-workspace-2",
+            "u/test-user/other_secret",
+            "other-secret-value",
+        ),
+    ];
+
+    /// Encrypt the PLACEHOLDER values inserted by the fixture using the real
+    /// workspace encryption keys, so migration tests can decrypt them correctly.
+    async fn encrypt_fixture_secrets(db: &Pool<Postgres>) {
+        for &(workspace_id, path, plaintext) in TEST_SECRETS {
+            let mc = build_crypt(db, workspace_id).await.unwrap();
+            let encrypted = encrypt(&mc, plaintext);
+            sqlx::query!(
+                "UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3 AND is_secret = true",
+                encrypted,
+                workspace_id,
+                path,
+            )
+            .execute(db)
+            .await
+            .unwrap();
+        }
+    }
 
-    /// Check if vault tests should run (requires RUN_VAULT_TESTS=1 env var)
     fn should_run_vault_tests() -> bool {
         std::env::var("RUN_VAULT_TESTS")
             .map(|v| v == "1" || v.to_lowercase() == "true")
             .unwrap_or(false)
     }
 
-    /// Set up BASE_URL for JWT tests (required for OIDC issuer URL generation)
+    #[cfg(feature = "openidconnect")]
     async fn setup_base_url() {
-        let base_url = std::env::var("BASE_URL")
-            .unwrap_or_else(|_| "http://localhost:8000".to_string());
+        let base_url =
+            std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost:8000".to_string());
         let mut url = windmill_common::BASE_URL.write().await;
         *url = base_url;
     }
 
-    /// Skip test if RUN_VAULT_TESTS is not set
     macro_rules! skip_if_no_vault {
         () => {
             if !should_run_vault_tests() {
@@ -71,6 +99,7 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "openidconnect")]
     fn vault_settings_jwt() -> VaultSettings {
         VaultSettings {
             address: std::env::var("VAULT_ADDR")
@@ -163,6 +192,7 @@ mod tests {
     // ==================== JWT Auth Tests ====================
 
     /// Test Vault connection with JWT authentication
+    #[cfg(feature = "openidconnect")]
     #[sqlx::test(fixtures("base", "secret_backend"))]
     async fn test_vault_connection_jwt(db: Pool<Postgres>) {
         skip_if_no_vault!();
@@ -236,70 +266,51 @@ mod tests {
 
         let settings = vault_settings_static_token();
 
-        // Verify Vault connection
         test_vault_connection(&settings, Some(&db))
             .await
             .expect("Failed to connect to Vault");
 
-        // Check initial state
-        let secrets_before = sqlx::query!(
-            "SELECT workspace_id, path, value FROM variable WHERE is_secret = true ORDER BY workspace_id, path"
-        )
-        .fetch_all(&db)
-        .await
-        .expect("Failed to query secrets");
+        // Encrypt fixture placeholders with real workspace keys
+        encrypt_fixture_secrets(&db).await;
 
-        println!(
-            "Found {} secrets in database before migration:",
-            secrets_before.len()
-        );
-        for s in &secrets_before {
-            println!("  - {}/{}: {} chars", s.workspace_id, s.path, s.value.len());
-        }
+        let secret_count = sqlx::query_scalar!(
+            "SELECT COUNT(*) FROM variable WHERE is_secret = true"
+        )
+        .fetch_one(&db)
+        .await
+        .expect("Failed to count secrets");
+        println!("Found {} secrets in database before migration", secret_count.unwrap_or(0));
 
         // Run migration
-        println!("\nMigrating secrets to Vault...");
+        println!("Migrating secrets to Vault...");
         let report = migrate_secrets_to_vault(&db, &settings)
             .await
             .expect("Migration to Vault failed");
 
-        println!("Migration report:");
-        println!("  Total secrets: {}", report.total_secrets);
-        println!("  Migrated: {}", report.migrated_count);
-        println!("  Failed: {}", report.failed_count);
+        println!("Migration report: total={}, migrated={}, failed={}",
+            report.total_secrets, report.migrated_count, report.failed_count);
 
         if !report.failures.is_empty() {
-            println!("  Failures:");
             for f in &report.failures {
-                println!("    - {}/{}: {}", f.workspace_id, f.path, f.error);
+                println!("  FAIL: {}/{}: {}", f.workspace_id, f.path, f.error);
             }
         }
 
         assert_eq!(report.failed_count, 0, "Migration had failures");
         assert!(report.migrated_count > 0, "No secrets were migrated");
 
-        // Verify secrets in Vault
-        println!("\nVerifying secrets in Vault...");
+        // Verify decrypted values in Vault match original plaintexts
         let vault_backend = VaultBackend::new(settings.clone());
-
-        for secret in &secrets_before {
-            let result = vault_backend
-                .get_secret(&secret.workspace_id, &secret.path)
-                .await;
-            assert!(
-                result.is_ok(),
-                "Failed to read secret {}/{} from Vault: {:?}",
-                secret.workspace_id,
-                secret.path,
-                result.err()
-            );
-            println!(
-                "  ✓ {}/{} exists in Vault",
-                secret.workspace_id, secret.path
-            );
+        for &(ws, path, expected_plaintext) in TEST_SECRETS {
+            let value = vault_backend
+                .get_secret(ws, path)
+                .await
+                .unwrap_or_else(|e| panic!("Failed to read {}/{} from Vault: {:?}", ws, path, e));
+            assert_eq!(value, expected_plaintext, "Vault value mismatch for {}/{}", ws, path);
+            println!("  ✓ {}/{} correct in Vault", ws, path);
         }
 
-        println!("\n✓ Migration to Vault completed successfully");
+        println!("✓ Migration to Vault completed successfully");
     }
 
     /// Test migration from Vault back to database
@@ -313,49 +324,50 @@ mod tests {
             .await
             .expect("Failed to connect to Vault");
 
+        encrypt_fixture_secrets(&db).await;
+
         // First migrate TO Vault
         println!("Setting up: migrating secrets to Vault first...");
         let to_vault = migrate_secrets_to_vault(&db, &settings)
             .await
             .expect("Initial migration to Vault failed");
         assert!(to_vault.migrated_count > 0, "No secrets to test with");
-        println!("  Migrated {} secrets to Vault", to_vault.migrated_count);
 
         // Clear database values
-        println!("\nClearing database secret values...");
         sqlx::query!("UPDATE variable SET value = 'CLEARED' WHERE is_secret = true")
             .execute(&db)
             .await
             .expect("Failed to clear values");
 
         // Migrate back from Vault
-        println!("\nMigrating secrets from Vault to database...");
+        println!("Migrating secrets from Vault to database...");
         let report = migrate_secrets_to_database(&db, &settings)
             .await
             .expect("Migration to database failed");
 
-        println!("Migration report:");
-        println!("  Total secrets: {}", report.total_secrets);
-        println!("  Migrated: {}", report.migrated_count);
-        println!("  Failed: {}", report.failed_count);
+        println!("Migration report: total={}, migrated={}, failed={}",
+            report.total_secrets, report.migrated_count, report.failed_count);
 
         assert_eq!(report.failed_count, 0, "Migration had failures");
         assert!(report.migrated_count > 0, "No secrets were migrated");
 
-        // Verify restored
-        let restored = sqlx::query!(
-            "SELECT COUNT(*) as count FROM variable WHERE is_secret = true AND value != 'CLEARED'"
-        )
-        .fetch_one(&db)
-        .await
-        .expect("Failed to count restored");
+        // Verify restored values decrypt to the original plaintexts
+        for &(ws, path, expected_plaintext) in TEST_SECRETS {
+            let row = sqlx::query_scalar!(
+                "SELECT value FROM variable WHERE workspace_id = $1 AND path = $2 AND is_secret = true",
+                ws, path
+            )
+            .fetch_one(&db)
+            .await
+            .unwrap_or_else(|_| panic!("Secret {}/{} not found in DB after migration", ws, path));
 
-        assert!(
-            restored.count.unwrap_or(0) > 0,
-            "No secrets were restored in database"
-        );
+            let mc = build_crypt(&db, ws).await.unwrap();
+            let decrypted = decrypt(&mc, row).expect("Failed to decrypt restored value");
+            assert_eq!(decrypted, expected_plaintext, "Restored value mismatch for {}/{}", ws, path);
+            println!("  ✓ {}/{} correctly restored in DB", ws, path);
+        }
 
-        println!("\n✓ Migration to database completed successfully");
+        println!("✓ Migration to database completed successfully");
     }
 
     // ==================== Variable Rename Tests ====================
@@ -443,68 +455,47 @@ mod tests {
             .await
             .expect("Failed to connect to Vault");
 
-        // Get original secrets
-        let original: HashMap<(String, String), String> = sqlx::query!(
-            "SELECT workspace_id, path, value FROM variable WHERE is_secret = true"
-        )
-        .fetch_all(&db)
-        .await
-        .expect("Failed to query")
-        .into_iter()
-        .map(|r| ((r.workspace_id, r.path), r.value))
-        .collect();
+        encrypt_fixture_secrets(&db).await;
 
-        println!("Original secrets: {} entries", original.len());
-
-        // Step 1: DB -> Vault
-        println!("\n=== Step 1: Migrate DB -> Vault ===");
+        println!("=== Step 1: Migrate DB -> Vault ===");
         let to_vault = migrate_secrets_to_vault(&db, &settings)
             .await
             .expect("Migration to Vault failed");
-        println!("Migrated {} secrets to Vault", to_vault.migrated_count);
         assert_eq!(to_vault.failed_count, 0);
+        println!("Migrated {} secrets to Vault", to_vault.migrated_count);
 
-        // Step 2: Clear DB
-        println!("\n=== Step 2: Clear database values ===");
+        println!("=== Step 2: Clear database values ===");
         sqlx::query!("UPDATE variable SET value = 'ROUND_TRIP_CLEARED' WHERE is_secret = true")
             .execute(&db)
             .await
             .expect("Failed to clear");
 
-        // Step 3: Vault -> DB
-        println!("\n=== Step 3: Migrate Vault -> DB ===");
+        println!("=== Step 3: Migrate Vault -> DB ===");
         let to_db = migrate_secrets_to_database(&db, &settings)
             .await
             .expect("Migration to database failed");
-        println!("Migrated {} secrets to database", to_db.migrated_count);
         assert_eq!(to_db.failed_count, 0);
+        println!("Migrated {} secrets to database", to_db.migrated_count);
 
-        // Step 4: Verify
-        println!("\n=== Step 4: Verify round-trip integrity ===");
-        let restored: HashMap<(String, String), String> = sqlx::query!(
-            "SELECT workspace_id, path, value FROM variable WHERE is_secret = true"
-        )
-        .fetch_all(&db)
-        .await
-        .expect("Failed to query")
-        .into_iter()
-        .map(|r| ((r.workspace_id, r.path), r.value))
-        .collect();
-
-        for ((ws, path), _) in &original {
-            let restored_value = restored
-                .get(&(ws.clone(), path.clone()))
-                .expect(&format!("Secret {}/{} not found after round-trip", ws, path));
-
-            assert_ne!(
-                restored_value, "ROUND_TRIP_CLEARED",
-                "Secret {}/{} was not restored",
+        println!("=== Step 4: Verify round-trip integrity ===");
+        for &(ws, path, expected_plaintext) in TEST_SECRETS {
+            let encrypted = sqlx::query_scalar!(
+                "SELECT value FROM variable WHERE workspace_id = $1 AND path = $2 AND is_secret = true",
                 ws, path
-            );
-            println!("  ✓ {}/{}: restored", ws, path);
+            )
+            .fetch_one(&db)
+            .await
+            .unwrap_or_else(|_| panic!("Secret {}/{} not found after round-trip", ws, path));
+
+            assert_ne!(encrypted, "ROUND_TRIP_CLEARED", "Secret {}/{} was not restored", ws, path);
+
+            let mc = build_crypt(&db, ws).await.unwrap();
+            let decrypted = decrypt(&mc, encrypted).expect("Failed to decrypt");
+            assert_eq!(decrypted, expected_plaintext, "Round-trip value mismatch for {}/{}", ws, path);
+            println!("  ✓ {}/{}: round-trip OK", ws, path);
         }
 
-        println!("\n✓ Full round-trip completed successfully!");
+        println!("✓ Full round-trip completed successfully!");
     }
 
     // ==================== Workspace Isolation Test ====================
@@ -517,7 +508,8 @@ mod tests {
         let settings = vault_settings_static_token();
         let backend = VaultBackend::new(settings.clone());
 
-        // First migrate secrets to Vault
+        encrypt_fixture_secrets(&db).await;
+
         migrate_secrets_to_vault(&db, &settings)
             .await
             .expect("Migration failed");
@@ -542,7 +534,7 @@ mod tests {
         assert!(ws1.is_ok(), "Same-workspace access should work");
         println!("✓ Same-workspace access works");
 
-        println!("\n✓ Workspace isolation verified!");
+        println!("✓ Workspace isolation verified!");
     }
 }
 

backend/tests/trigger_e2e.rs

@@ -0,0 +1,726 @@
+/*!
+ * End-to-end integration tests for Windmill trigger listeners.
+ *
+ * Each test is `#[ignore]` because it requires a running external service
+ * (MQTT broker, NATS server, Kafka broker, etc.). See individual test doc
+ * comments for setup instructions.
+ *
+ * Quick start — use the helper scripts in `tests/fixtures/`:
+ * ```bash
+ * ./tests/fixtures/start_all_triggers.sh          # start all services
+ * ./tests/fixtures/start_all_triggers.sh oss      # start OSS services only
+ * ./tests/fixtures/start_all_triggers.sh stop     # tear down everything
+ * ```
+ *
+ * The general pattern:
+ * 1. Insert a test script + trigger row + resource into the DB
+ * 2. Start the API server with listeners enabled (server_mode=true)
+ * 3. Connect to the external service and send a test message
+ * 4. Poll `v2_job` for a job matching the trigger path + trigger_kind
+ * 5. Verify the args shape/content
+ */
+
+use serde_json::json;
+use sqlx::{Pool, Postgres};
+use std::time::Duration;
+
+mod common;
+use common::*;
+
+// ============================================================================
+// Helpers
+// ============================================================================
+
+/// Row shape for polling v2_job.
+#[derive(Debug)]
+#[allow(dead_code)]
+struct TriggerJobRow {
+    id: uuid::Uuid,
+    runnable_path: Option<String>,
+    trigger_kind: Option<String>,
+    args: Option<sqlx::types::Json<serde_json::Value>>,
+}
+
+/// Poll `v2_job` every 500ms for up to `timeout` for a job whose
+/// `runnable_path` and `trigger_kind` match the expected values.
+async fn poll_for_trigger_job(
+    db: &Pool<Postgres>,
+    script_path: &str,
+    trigger_kind: &str,
+    timeout: Duration,
+) -> anyhow::Result<TriggerJobRow> {
+    let deadline = tokio::time::Instant::now() + timeout;
+    loop {
+        let row = sqlx::query_as!(
+            TriggerJobRow,
+            r#"
+            SELECT id, runnable_path, trigger_kind AS "trigger_kind: String",
+                   args AS "args: sqlx::types::Json<serde_json::Value>"
+            FROM v2_job
+            WHERE runnable_path = $1
+              AND trigger_kind = $2::job_trigger_kind
+            ORDER BY created_at DESC
+            LIMIT 1
+            "#,
+            script_path,
+            trigger_kind as _,
+        )
+        .fetch_optional(db)
+        .await?;
+
+        if let Some(job) = row {
+            return Ok(job);
+        }
+
+        if tokio::time::Instant::now() >= deadline {
+            anyhow::bail!(
+                "timed out waiting for trigger job (script_path={}, trigger_kind={})",
+                script_path,
+                trigger_kind
+            );
+        }
+
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+}
+
+/// Insert a minimal test script row that trigger listeners can reference.
+async fn insert_test_script(db: &Pool<Postgres>, path: &str) -> anyhow::Result<i64> {
+    let hash: i64 = rand::random::<i64>().unsigned_abs() as i64;
+    sqlx::query(
+        "INSERT INTO script (workspace_id, hash, path, summary, description, content,
+                  created_by, language, kind, lock)
+         VALUES ('test-workspace', $1, $2, '', '', 'def main(): pass',
+                  'test-user', 'python3', 'script', '')",
+    )
+    .bind(hash)
+    .bind(path)
+    .execute(db)
+    .await?;
+    Ok(hash)
+}
+
+/// Insert a resource row for triggers that resolve connection details from the
+/// `resource` table.
+async fn insert_resource(
+    db: &Pool<Postgres>,
+    path: &str,
+    resource_type: &str,
+    value: serde_json::Value,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by)
+         VALUES ('test-workspace', $1, $2::jsonb, $3, '{}'::jsonb, 'test-user')",
+    )
+    .bind(path)
+    .bind(value)
+    .bind(resource_type)
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+// ============================================================================
+// MQTT Trigger E2E
+// ============================================================================
+
+/// End-to-end test for MQTT trigger.
+///
+/// Requires a running MQTT broker. Setup:
+/// ```bash
+/// ./tests/fixtures/start_mqtt.sh
+/// ```
+///
+/// Run:
+/// ```bash
+/// cargo test --test trigger_e2e test_mqtt_e2e --features mqtt_trigger \
+///     -- --ignored --nocapture
+/// ```
+#[ignore = "requires running MQTT broker on localhost:1883"]
+#[sqlx::test(fixtures("base"))]
+async fn test_mqtt_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let script_path = "f/test/mqtt_e2e_handler";
+    insert_test_script(&db, script_path).await?;
+
+    insert_resource(
+        &db,
+        "u/test-user/mqtt_res",
+        "mqtt",
+        json!({
+            "broker": "localhost",
+            "port": 1883
+        }),
+    )
+    .await?;
+
+    sqlx::query(
+        r#"
+        INSERT INTO mqtt_trigger (
+            path, mqtt_resource_path, subscribe_topics, client_version,
+            script_path, is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9)
+        "#,
+    )
+    .bind("f/test/mqtt_e2e_trigger")
+    .bind("u/test-user/mqtt_res")
+    .bind(json!({"topic": "windmill/test/e2e", "qos": "qos0"}))
+    .bind("v5")
+    .bind(script_path)
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .execute(&db)
+    .await?;
+
+    let _server = ApiServer::start_with_listeners(db.clone()).await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
+
+    // Publish a message using rumqttc
+    let mut mqtt_opts = rumqttc::MqttOptions::new("windmill-e2e-test", "localhost", 1883);
+    mqtt_opts.set_keep_alive(Duration::from_secs(5));
+    let (client, mut eventloop) = rumqttc::AsyncClient::new(mqtt_opts, 10);
+
+    // Drive the event loop in the background
+    let el_handle = tokio::spawn(async move {
+        loop {
+            match eventloop.poll().await {
+                Ok(_) => {}
+                Err(_) => break,
+            }
+        }
+    });
+
+    tokio::time::sleep(Duration::from_millis(500)).await;
+    client
+        .publish(
+            "windmill/test/e2e",
+            rumqttc::QoS::AtLeastOnce,
+            false,
+            b"hello from e2e test".to_vec(),
+        )
+        .await?;
+
+    let job = poll_for_trigger_job(&db, script_path, "mqtt", Duration::from_secs(30)).await?;
+    assert!(job.args.is_some(), "job should have args");
+
+    client.disconnect().await.ok();
+    el_handle.abort();
+
+    Ok(())
+}
+
+// ============================================================================
+// WebSocket Trigger E2E
+// ============================================================================
+
+/// End-to-end test for WebSocket trigger.
+///
+/// Requires a WebSocket echo server. Setup:
+/// ```bash
+/// ./tests/fixtures/start_websocket.sh
+/// ```
+///
+/// Run:
+/// ```bash
+/// cargo test --test trigger_e2e test_websocket_e2e --features websocket \
+///     -- --ignored --nocapture
+/// ```
+#[ignore = "requires running WebSocket echo server on localhost:8765"]
+#[sqlx::test(fixtures("base"))]
+async fn test_websocket_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let script_path = "f/test/ws_e2e_handler";
+    insert_test_script(&db, script_path).await?;
+
+    sqlx::query!(
+        r#"
+        INSERT INTO websocket_trigger (
+            path, url, script_path, is_flow, workspace_id,
+            edited_by, email, initial_messages
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+        "#,
+        "f/test/ws_e2e_trigger",
+        "ws://localhost:8765",
+        script_path,
+        false,
+        "test-workspace",
+        "test-user",
+        "test@windmill.dev",
+        &[json!({"type": "RawMessage", "content": "hello from e2e test"})] as &[serde_json::Value],
+    )
+    .execute(&db)
+    .await?;
+
+    let _server = ApiServer::start_with_listeners(db.clone()).await?;
+
+    // The WebSocket trigger connects to the server and sends initial_messages,
+    // and each received message triggers a job.
+    let job = poll_for_trigger_job(&db, script_path, "websocket", Duration::from_secs(30)).await?;
+    assert!(job.args.is_some(), "job should have args");
+
+    Ok(())
+}
+
+// ============================================================================
+// Postgres Trigger E2E
+// ============================================================================
+
+/// End-to-end test for Postgres trigger (logical replication).
+///
+/// Requires PostgreSQL with `wal_level=logical`. Setup:
+/// ```bash
+/// ./tests/fixtures/start_postgres_replication.sh
+/// ```
+/// (The script checks wal_level and creates the table/publication/slot in the
+/// main DB. This test re-creates them in its isolated sqlx::test database.)
+///
+/// Run:
+/// ```bash
+/// cargo test --test trigger_e2e test_postgres_e2e --features postgres_trigger \
+///     -- --ignored --nocapture
+/// ```
+#[ignore = "requires PostgreSQL with wal_level=logical"]
+#[sqlx::test(fixtures("base"))]
+async fn test_postgres_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let script_path = "f/test/pg_e2e_handler";
+    insert_test_script(&db, script_path).await?;
+
+    // Create the tracked table + publication + replication slot inside the
+    // isolated test database (sqlx::test gives us a fresh DB each run).
+    // Replication slots are server-wide so we use a random suffix.
+    let suffix: u32 = rand::random();
+    let slot_name = format!("test_e2e_slot_{suffix}");
+    let pub_name = format!("test_e2e_pub_{suffix}");
+
+    sqlx::query("CREATE TABLE test_trigger_table (id serial PRIMARY KEY, data text)")
+        .execute(&db)
+        .await?;
+    sqlx::query(&format!("CREATE PUBLICATION {pub_name} FOR TABLE test_trigger_table"))
+        .execute(&db)
+        .await?;
+    sqlx::query(&format!(
+        "SELECT pg_create_logical_replication_slot('{slot_name}', 'pgoutput')"
+    ))
+    .execute(&db)
+    .await?;
+
+    // Extract the test DB name from the pool so the resource points here,
+    // not at the main windmill database.
+    let test_db_name: String =
+        sqlx::query_scalar("SELECT current_database()")
+            .fetch_one(&db)
+            .await?;
+
+    insert_resource(
+        &db,
+        "u/test-user/pg_res",
+        "postgresql",
+        json!({
+            "user": "postgres",
+            "password": "changeme",
+            "host": "localhost",
+            "port": 5432,
+            "dbname": test_db_name,
+            "sslmode": "disable"
+        }),
+    )
+    .await?;
+
+    sqlx::query(
+        r#"
+        INSERT INTO postgres_trigger (
+            path, script_path, is_flow, workspace_id, edited_by, email,
+            postgres_resource_path, replication_slot_name, publication_name
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+        "#,
+    )
+    .bind("f/test/pg_e2e_trigger")
+    .bind(script_path)
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .bind("u/test-user/pg_res")
+    .bind(&slot_name)
+    .bind(&pub_name)
+    .execute(&db)
+    .await?;
+
+    let _server = ApiServer::start_with_listeners(db.clone()).await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
+
+    // Insert a row into the tracked table to trigger a change event
+    sqlx::query("INSERT INTO test_trigger_table (data) VALUES ('e2e test data')")
+        .execute(&db)
+        .await?;
+
+    let job = poll_for_trigger_job(&db, script_path, "postgres", Duration::from_secs(30)).await?;
+    assert!(job.args.is_some(), "job should have args");
+
+    Ok(())
+}
+
+// ============================================================================
+// Kafka Trigger E2E (Enterprise)
+// ============================================================================
+
+/// End-to-end test for Kafka trigger (Enterprise only).
+///
+/// Requires a running Kafka broker with the test topic. Setup:
+/// ```bash
+/// ./tests/fixtures/start_kafka.sh
+/// ```
+///
+/// Run:
+/// ```bash
+/// cargo test --test trigger_e2e test_kafka_e2e \
+///     --features kafka,enterprise,private -- --ignored --nocapture
+/// ```
+#[cfg(all(feature = "enterprise", feature = "private"))]
+#[ignore = "requires running Kafka broker on localhost:9092"]
+#[sqlx::test(fixtures("base"))]
+async fn test_kafka_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let script_path = "f/test/kafka_e2e_handler";
+    insert_test_script(&db, script_path).await?;
+
+    insert_resource(
+        &db,
+        "u/test-user/kafka_res",
+        "kafka",
+        json!({
+            "brokers": ["localhost:9092"],
+            "security": { "label": "PLAINTEXT" }
+        }),
+    )
+    .await?;
+
+    sqlx::query!(
+        r#"
+        INSERT INTO kafka_trigger (
+            path, kafka_resource_path, topics, group_id,
+            script_path, is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+        "#,
+        "f/test/kafka_e2e_trigger",
+        "u/test-user/kafka_res",
+        &["windmill-e2e-test"] as &[&str],
+        "windmill-e2e-test-group",
+        script_path,
+        false,
+        "test-workspace",
+        "test-user",
+        "test@windmill.dev",
+    )
+    .execute(&db)
+    .await?;
+
+    let _server = ApiServer::start_with_listeners(db.clone()).await?;
+    tokio::time::sleep(Duration::from_secs(5)).await;
+
+    // Produce messages using rdkafka. The consumer starts with auto.offset.reset=latest
+    // and needs time for group rebalance, so we send repeatedly until a job appears.
+    use rdkafka::config::ClientConfig;
+    use rdkafka::producer::{FutureProducer, FutureRecord};
+
+    let producer: FutureProducer = ClientConfig::new()
+        .set("bootstrap.servers", "localhost:9092")
+        .create()?;
+
+    let db2 = db.clone();
+    let produce_handle = tokio::spawn(async move {
+        for _ in 0..30 {
+            let _ = producer
+                .send(
+                    FutureRecord::to("windmill-e2e-test")
+                        .payload("hello from kafka e2e test")
+                        .key("test-key"),
+                    Duration::from_secs(5),
+                )
+                .await;
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    });
+
+    let job = poll_for_trigger_job(&db2, script_path, "kafka", Duration::from_secs(30)).await?;
+    produce_handle.abort();
+    assert!(job.args.is_some(), "job should have args");
+
+    Ok(())
+}
+
+// ============================================================================
+// NATS Trigger E2E (Enterprise)
+// ============================================================================
+
+/// End-to-end test for NATS trigger (Enterprise only).
+///
+/// Requires a running NATS server. Setup:
+/// ```bash
+/// ./tests/fixtures/start_nats.sh
+/// ```
+///
+/// Run:
+/// ```bash
+/// cargo test --test trigger_e2e test_nats_e2e \
+///     --features nats,enterprise,private -- --ignored --nocapture
+/// ```
+#[cfg(all(feature = "enterprise", feature = "private"))]
+#[ignore = "requires running NATS server on localhost:4222"]
+#[sqlx::test(fixtures("base"))]
+async fn test_nats_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let script_path = "f/test/nats_e2e_handler";
+    insert_test_script(&db, script_path).await?;
+
+    insert_resource(
+        &db,
+        "u/test-user/nats_res",
+        "nats",
+        json!({
+            "servers": ["nats://localhost:4222"],
+            "auth": { "label": "NO_AUTH" },
+            "require_tls": false
+        }),
+    )
+    .await?;
+
+    sqlx::query!(
+        r#"
+        INSERT INTO nats_trigger (
+            path, nats_resource_path, subjects, script_path,
+            is_flow, workspace_id, edited_by, email, use_jetstream
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+        "#,
+        "f/test/nats_e2e_trigger",
+        "u/test-user/nats_res",
+        &["windmill.e2e.test"] as &[&str],
+        script_path,
+        false,
+        "test-workspace",
+        "test-user",
+        "test@windmill.dev",
+        false,
+    )
+    .execute(&db)
+    .await?;
+
+    let _server = ApiServer::start_with_listeners(db.clone()).await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
+
+    // Publish a message using async-nats
+    let nats_client = async_nats::connect("localhost:4222").await?;
+    nats_client
+        .publish("windmill.e2e.test", "hello from nats e2e test".into())
+        .await?;
+    nats_client.flush().await?;
+
+    let job = poll_for_trigger_job(&db, script_path, "nats", Duration::from_secs(30)).await?;
+    assert!(job.args.is_some(), "job should have args");
+
+    Ok(())
+}
+
+// ============================================================================
+// SQS Trigger E2E (Enterprise)
+// ============================================================================
+
+/// End-to-end test for SQS trigger (Enterprise only).
+///
+/// Requires LocalStack with the test queue. Setup:
+/// ```bash
+/// ./tests/fixtures/start_sqs.sh
+/// ```
+///
+/// Run:
+/// ```bash
+/// AWS_ENDPOINT_URL=http://localhost:4566 \
+/// cargo test --test trigger_e2e test_sqs_e2e \
+///     --features sqs_trigger,enterprise,private -- --ignored --nocapture
+/// ```
+#[cfg(all(feature = "enterprise", feature = "private"))]
+#[ignore = "requires LocalStack SQS on localhost:4566"]
+#[sqlx::test(fixtures("base"))]
+async fn test_sqs_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    // The SQS listener uses aws_config which respects AWS_ENDPOINT_URL for LocalStack.
+    std::env::set_var("AWS_ENDPOINT_URL", "http://localhost:4566");
+
+    let script_path = "f/test/sqs_e2e_handler";
+    insert_test_script(&db, script_path).await?;
+
+    insert_resource(
+        &db,
+        "u/test-user/aws_res",
+        "aws",
+        json!({
+            "awsAccessKeyId": "test",
+            "awsSecretAccessKey": "test",
+            "region": "us-east-1"
+        }),
+    )
+    .await?;
+
+    sqlx::query!(
+        r#"
+        INSERT INTO sqs_trigger (
+            path, queue_url, aws_resource_path, script_path,
+            is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+        "#,
+        "f/test/sqs_e2e_trigger",
+        "http://localhost:4566/000000000000/windmill-e2e-test",
+        "u/test-user/aws_res",
+        script_path,
+        false,
+        "test-workspace",
+        "test-user",
+        "test@windmill.dev",
+    )
+    .execute(&db)
+    .await?;
+
+    let _server = ApiServer::start_with_listeners(db.clone()).await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
+
+    // Send a message using aws-sdk-sqs
+    let config = aws_config::defaults(aws_config::BehaviorVersion::latest())
+        .endpoint_url("http://localhost:4566")
+        .region(aws_config::Region::new("us-east-1"))
+        .credentials_provider(aws_credential_types::Credentials::new(
+            "test", "test", None, None, "test",
+        ))
+        .load()
+        .await;
+    let sqs_client = aws_sdk_sqs::Client::new(&config);
+
+    sqs_client
+        .send_message()
+        .queue_url("http://localhost:4566/000000000000/windmill-e2e-test")
+        .message_body("hello from sqs e2e test")
+        .send()
+        .await?;
+
+    let job = poll_for_trigger_job(&db, script_path, "sqs", Duration::from_secs(30)).await?;
+    assert!(job.args.is_some(), "job should have args");
+
+    Ok(())
+}
+
+// ============================================================================
+// GCP Pub/Sub Trigger E2E (Enterprise)
+// ============================================================================
+
+/// End-to-end test for GCP Pub/Sub trigger (Enterprise only).
+///
+/// Requires the GCP Pub/Sub emulator with test topic/subscription. Setup:
+/// ```bash
+/// ./tests/fixtures/start_gcp_pubsub.sh
+/// ```
+///
+/// Run:
+/// ```bash
+/// PUBSUB_EMULATOR_HOST=localhost:8085 \
+/// cargo test --test trigger_e2e test_gcp_e2e \
+///     --features gcp_trigger,enterprise,private -- --ignored --nocapture
+/// ```
+#[cfg(all(feature = "enterprise", feature = "private"))]
+#[ignore = "requires GCP Pub/Sub emulator on localhost:8085"]
+#[sqlx::test(fixtures("base"))]
+async fn test_gcp_e2e(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let script_path = "f/test/gcp_e2e_handler";
+    insert_test_script(&db, script_path).await?;
+
+    // The GCP emulator doesn't require real credentials, but the resource
+    // row must still exist for the listener to resolve it.
+    // The private_key must use literal \n (backslash-n) as in real GCP service
+    // account JSON files. The trigger code re-parses it through serde_json to
+    // convert those escape sequences to actual newlines.
+    insert_resource(
+        &db,
+        "u/test-user/gcp_res",
+        "google",
+        json!({
+            "project_id": "test-project",
+            "private_key_id": "test",
+            "private_key": "-----BEGIN RSA PRIVATE KEY-----\\nMIIBogIBAAJBALRiMLAH\\n-----END RSA PRIVATE KEY-----\\n",
+            "client_email": "test@test-project.iam.gserviceaccount.com",
+            "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+            "token_uri": "https://oauth2.googleapis.com/token",
+            "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs"
+        }),
+    )
+    .await?;
+
+    sqlx::query(
+        r#"
+        INSERT INTO gcp_trigger (
+            path, gcp_resource_path, topic_id, subscription_id,
+            delivery_type, subscription_mode, script_path, is_flow,
+            workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::gcp_subscription_mode, $7, $8, $9, $10, $11)
+        "#,
+    )
+    .bind("f/test/gcp_e2e_trigger")
+    .bind("u/test-user/gcp_res")
+    .bind("windmill-e2e-test")
+    .bind("windmill-e2e-sub")
+    .bind("pull")
+    .bind("existing")
+    .bind(script_path)
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .execute(&db)
+    .await?;
+
+    let _server = ApiServer::start_with_listeners(db.clone()).await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
+
+    // Publish a message to the emulator via HTTP
+    let client = reqwest::Client::new();
+    let emulator_host =
+        std::env::var("PUBSUB_EMULATOR_HOST").unwrap_or_else(|_| "localhost:8085".to_string());
+    // The google-cloud-pubsub crate uses "local-project" as the default project ID
+    // when PUBSUB_EMULATOR_HOST is set, so we must publish to that project's topic.
+    let publish_url = format!(
+        "http://{}/v1/projects/local-project/topics/windmill-e2e-test:publish",
+        emulator_host
+    );
+
+    let message_data = base64::Engine::encode(
+        &base64::engine::general_purpose::STANDARD,
+        "hello from gcp e2e test",
+    );
+    client
+        .post(&publish_url)
+        .json(&json!({
+            "messages": [{ "data": message_data }]
+        }))
+        .send()
+        .await?;
+
+    let job = poll_for_trigger_job(&db, script_path, "gcp", Duration::from_secs(30)).await?;
+    assert!(job.args.is_some(), "job should have args");
+
+    Ok(())
+}

backend/tests/triggers.rs

@@ -1272,3 +1272,431 @@ async fn test_schedule_insert_and_query(db: Pool<Postgres>) -> anyhow::Result<()
 
     Ok(())
 }
+
+// ============================================================================
+// MQTT Trigger Tests (DB-level)
+// ============================================================================
+
+#[sqlx::test(fixtures("base"))]
+async fn test_mqtt_trigger_insert(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO mqtt_trigger (
+            path, mqtt_resource_path, subscribe_topics, client_version,
+            script_path, is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9)
+        "#,
+    )
+    .bind("f/test/mqtt_trigger")
+    .bind("u/admin/mqtt_resource")
+    .bind(json!({"topic": "test/+", "qos": "qos1"}))
+    .bind("v5")
+    .bind("f/test/mqtt_handler")
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .execute(&db)
+    .await?;
+
+    let trigger = sqlx::query!(
+        r#"
+        SELECT mqtt_resource_path, client_version AS "client_version: String",
+               script_path, mode AS "mode: String"
+        FROM mqtt_trigger
+        WHERE workspace_id = $1 AND path = $2
+        "#,
+        "test-workspace",
+        "f/test/mqtt_trigger",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(trigger.mqtt_resource_path, "u/admin/mqtt_resource");
+    assert_eq!(trigger.client_version, "v5");
+    assert_eq!(trigger.script_path, "f/test/mqtt_handler");
+    assert_eq!(trigger.mode, "enabled");
+
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn test_mqtt_trigger_update(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO mqtt_trigger (
+            path, mqtt_resource_path, subscribe_topics, client_version,
+            script_path, is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9)
+        "#,
+    )
+    .bind("f/test/mqtt_trigger")
+    .bind("u/admin/mqtt_resource")
+    .bind(json!({"topic": "test/+", "qos": "qos1"}))
+    .bind("v5")
+    .bind("f/test/old_handler")
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .execute(&db)
+    .await?;
+
+    sqlx::query!(
+        "UPDATE mqtt_trigger SET script_path = $1 WHERE workspace_id = $2 AND path = $3",
+        "f/test/new_handler",
+        "test-workspace",
+        "f/test/mqtt_trigger",
+    )
+    .execute(&db)
+    .await?;
+
+    let trigger = sqlx::query!(
+        "SELECT script_path FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2",
+        "test-workspace",
+        "f/test/mqtt_trigger",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(trigger.script_path, "f/test/new_handler");
+
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn test_mqtt_trigger_delete(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO mqtt_trigger (
+            path, mqtt_resource_path, subscribe_topics, client_version,
+            script_path, is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9)
+        "#,
+    )
+    .bind("f/test/mqtt_trigger")
+    .bind("u/admin/mqtt_resource")
+    .bind(json!({"topic": "test/+", "qos": "qos1"}))
+    .bind("v5")
+    .bind("f/test/mqtt_handler")
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .execute(&db)
+    .await?;
+
+    sqlx::query!(
+        "DELETE FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2",
+        "test-workspace",
+        "f/test/mqtt_trigger",
+    )
+    .execute(&db)
+    .await?;
+
+    let count = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2",
+        "test-workspace",
+        "f/test/mqtt_trigger",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(count, Some(0));
+
+    Ok(())
+}
+
+// ============================================================================
+// GCP Trigger Tests (DB-level)
+// ============================================================================
+
+#[sqlx::test(fixtures("base"))]
+async fn test_gcp_trigger_insert_pull(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO gcp_trigger (
+            path, gcp_resource_path, topic_id, subscription_id,
+            delivery_type, subscription_mode, script_path, is_flow,
+            workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::gcp_subscription_mode, $7, $8, $9, $10, $11)
+        "#,
+    )
+    .bind("f/test/gcp_trigger_pull")
+    .bind("u/admin/gcp_resource")
+    .bind("my-topic")
+    .bind("my-subscription")
+    .bind("pull")
+    .bind("create_update")
+    .bind("f/test/gcp_handler")
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .execute(&db)
+    .await?;
+
+    let trigger = sqlx::query!(
+        r#"
+        SELECT gcp_resource_path, topic_id, subscription_id,
+               delivery_type AS "delivery_type: String",
+               subscription_mode AS "subscription_mode: String",
+               mode AS "mode: String"
+        FROM gcp_trigger
+        WHERE workspace_id = $1 AND path = $2
+        "#,
+        "test-workspace",
+        "f/test/gcp_trigger_pull",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(trigger.gcp_resource_path, "u/admin/gcp_resource");
+    assert_eq!(trigger.topic_id, "my-topic");
+    assert_eq!(trigger.subscription_id, "my-subscription");
+    assert_eq!(trigger.delivery_type, "pull");
+    assert_eq!(trigger.subscription_mode, "create_update");
+    assert_eq!(trigger.mode, "enabled");
+
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn test_gcp_trigger_insert_push(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO gcp_trigger (
+            path, gcp_resource_path, topic_id, subscription_id,
+            delivery_type, delivery_config, subscription_mode,
+            script_path, is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::jsonb, $7::gcp_subscription_mode, $8, $9, $10, $11, $12)
+        "#,
+    )
+    .bind("f/test/gcp_trigger_push")
+    .bind("u/admin/gcp_resource")
+    .bind("my-topic")
+    .bind("my-push-subscription")
+    .bind("push")
+    .bind(json!({"endpoint": "https://example.com/push"}))
+    .bind("create_update")
+    .bind("f/test/gcp_handler")
+    .bind(false)
+    .bind("test-workspace")
+    .bind("test-user")
+    .bind("test@windmill.dev")
+    .execute(&db)
+    .await?;
+
+    let trigger = sqlx::query!(
+        r#"
+        SELECT delivery_type AS "delivery_type: String",
+               delivery_config
+        FROM gcp_trigger
+        WHERE workspace_id = $1 AND path = $2
+        "#,
+        "test-workspace",
+        "f/test/gcp_trigger_push",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(trigger.delivery_type, "push");
+    assert!(trigger.delivery_config.is_some());
+    assert_eq!(
+        trigger.delivery_config.unwrap()["endpoint"],
+        "https://example.com/push"
+    );
+
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn test_gcp_trigger_unique_constraint(db: Pool<Postgres>) -> anyhow::Result<()> {
+    let insert_query = r#"
+        INSERT INTO gcp_trigger (
+            path, gcp_resource_path, topic_id, subscription_id,
+            delivery_type, subscription_mode, script_path, is_flow,
+            workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::gcp_subscription_mode, $7, $8, $9, $10, $11)
+    "#;
+
+    sqlx::query(insert_query)
+        .bind("f/test/gcp_trigger_1")
+        .bind("u/admin/gcp_resource")
+        .bind("my-topic")
+        .bind("shared-subscription")
+        .bind("pull")
+        .bind("create_update")
+        .bind("f/test/gcp_handler")
+        .bind(false)
+        .bind("test-workspace")
+        .bind("test-user")
+        .bind("test@windmill.dev")
+        .execute(&db)
+        .await?;
+
+    // Inserting a second trigger with same (subscription_id, gcp_resource_path, workspace_id) should fail
+    let result = sqlx::query(insert_query)
+        .bind("f/test/gcp_trigger_2")
+        .bind("u/admin/gcp_resource")
+        .bind("my-topic")
+        .bind("shared-subscription")
+        .bind("pull")
+        .bind("create_update")
+        .bind("f/test/gcp_handler_2")
+        .bind(false)
+        .bind("test-workspace")
+        .bind("test-user")
+        .bind("test@windmill.dev")
+        .execute(&db)
+        .await;
+
+    assert!(
+        result.is_err(),
+        "should fail due to unique constraint on (subscription_id, gcp_resource_path, workspace_id)"
+    );
+
+    Ok(())
+}
+
+// ============================================================================
+// Email Trigger Tests (DB-level)
+// ============================================================================
+
+#[sqlx::test(fixtures("base"))]
+async fn test_email_trigger_insert(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query!(
+        r#"
+        INSERT INTO email_trigger (
+            path, local_part, workspaced_local_part, script_path,
+            is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+        "#,
+        "f/test/email_trigger",
+        "support",
+        true,
+        "f/test/email_handler",
+        false,
+        "test-workspace",
+        "test-user",
+        "test@windmill.dev",
+    )
+    .execute(&db)
+    .await?;
+
+    let trigger = sqlx::query!(
+        r#"
+        SELECT local_part, workspaced_local_part, script_path,
+               mode AS "mode: String"
+        FROM email_trigger
+        WHERE workspace_id = $1 AND path = $2
+        "#,
+        "test-workspace",
+        "f/test/email_trigger",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(trigger.local_part, "support");
+    assert_eq!(trigger.workspaced_local_part, true);
+    assert_eq!(trigger.script_path, "f/test/email_handler");
+    assert_eq!(trigger.mode, "enabled");
+
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn test_email_trigger_update(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query!(
+        r#"
+        INSERT INTO email_trigger (
+            path, local_part, workspaced_local_part, script_path,
+            is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+        "#,
+        "f/test/email_trigger",
+        "support",
+        true,
+        "f/test/old_handler",
+        false,
+        "test-workspace",
+        "test-user",
+        "test@windmill.dev",
+    )
+    .execute(&db)
+    .await?;
+
+    sqlx::query!(
+        "UPDATE email_trigger SET script_path = $1, local_part = $2 WHERE workspace_id = $3 AND path = $4",
+        "f/test/new_handler",
+        "billing",
+        "test-workspace",
+        "f/test/email_trigger",
+    )
+    .execute(&db)
+    .await?;
+
+    let trigger = sqlx::query!(
+        "SELECT script_path, local_part FROM email_trigger WHERE workspace_id = $1 AND path = $2",
+        "test-workspace",
+        "f/test/email_trigger",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(trigger.script_path, "f/test/new_handler");
+    assert_eq!(trigger.local_part, "billing");
+
+    Ok(())
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn test_email_trigger_delete(db: Pool<Postgres>) -> anyhow::Result<()> {
+    sqlx::query!(
+        r#"
+        INSERT INTO email_trigger (
+            path, local_part, workspaced_local_part, script_path,
+            is_flow, workspace_id, edited_by, email
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+        "#,
+        "f/test/email_trigger",
+        "support",
+        true,
+        "f/test/email_handler",
+        false,
+        "test-workspace",
+        "test-user",
+        "test@windmill.dev",
+    )
+    .execute(&db)
+    .await?;
+
+    sqlx::query!(
+        "DELETE FROM email_trigger WHERE workspace_id = $1 AND path = $2",
+        "test-workspace",
+        "f/test/email_trigger",
+    )
+    .execute(&db)
+    .await?;
+
+    let count = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM email_trigger WHERE workspace_id = $1 AND path = $2",
+        "test-workspace",
+        "f/test/email_trigger",
+    )
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(count, Some(0));
+
+    Ok(())
+}

backend/windmill-api/src/inkeep_oss.rs

@@ -15,6 +15,7 @@ pub fn global_service() -> Router {
 
 #[cfg(not(feature = "private"))]
 async fn inkeep_not_available() -> windmill_common::error::Result<()> {
+    println!("Inkeep not available - Enterprise Edition feature");
     Err(Error::Generic(
         http::StatusCode::FORBIDDEN,
         "Inkeep AI documentation assistant is only available in Windmill Enterprise Edition"

backend/windmill-worker/src/ai/types.rs

@@ -460,6 +460,11 @@ impl OpenAPISchema {
             Typ::Bool => Self::from_str("boolean"),
             Typ::Bytes => Self::from_str("string"),
             Typ::Datetime => Self::datetime(),
+            Typ::Date => Self {
+                r#type: Some(SchemaType::Single("string".to_string())),
+                format: Some("date".to_string()),
+                ..Default::default()
+            },
             Typ::Resource(_) => Self::from_str("string"),
             Typ::Email => Self::from_str("string"),
             Typ::Sql => Self::from_str("string"),

backend/windmill-worker/src/python_executor.rs

@@ -1016,6 +1016,21 @@ async fn prepare_wrapper(
                                      kwargs[\"{name}\"] = datetime.fromisoformat(kwargs[\"{name}\"])\n",
                 )
             }
+            windmill_parser::Typ::Date => {
+                let name = &x.name;
+                format!(
+                    "if \"{name}\" in kwargs and kwargs[\"{name}\"] is not None:\n    \
+                                     try:\n        \
+                                         kwargs[\"{name}\"] = date.fromisoformat(kwargs[\"{name}\"])\n    \
+                                     except ValueError:\n        \
+                                         for _fmt in (\"%d-%m-%Y\", \"%m/%d/%Y\", \"%d/%m/%Y\", \"%Y/%m/%d\"):\n            \
+                                             try:\n                \
+                                                 kwargs[\"{name}\"] = datetime.strptime(kwargs[\"{name}\"], _fmt).date()\n                \
+                                                 break\n            \
+                                             except ValueError:\n                \
+                                                 continue\n",
+                )
+            }
             _ => "".to_string(),
         })
         .collect::<Vec<String>>()
@@ -1035,14 +1050,19 @@ async fn prepare_wrapper(
     } else {
         ""
     };
-    let import_datetime = if init_sig
+    let has_datetime = init_sig
         .args
         .iter()
-        .any(|x| x.typ == windmill_parser::Typ::Datetime)
-    {
-        "from datetime import datetime"
-    } else {
-        ""
+        .any(|x| x.typ == windmill_parser::Typ::Datetime);
+    let has_date = init_sig
+        .args
+        .iter()
+        .any(|x| x.typ == windmill_parser::Typ::Date);
+    let import_datetime = match (has_datetime, has_date) {
+        (true, true) => "from datetime import datetime, date",
+        (true, false) => "from datetime import datetime",
+        (false, true) => "from datetime import datetime, date",
+        (false, false) => "",
     };
     let spread = if sig.star_kwargs {
         "args = kwargs".to_string()

backend/windmill-worker/src/schema.rs

@@ -34,7 +34,7 @@ fn make_rules_for_arg_typ(typ: &Typ) -> Vec<SchemaValidationRule> {
             rules.push(SchemaValidationRule::IsString);
             rules.push(SchemaValidationRule::IsBytes);
         }
-        Typ::Datetime => {
+        Typ::Datetime | Typ::Date => {
             rules.push(SchemaValidationRule::IsString);
             rules.push(SchemaValidationRule::IsDatetime);
         }

cli/windmill-utils-internal/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "windmill-utils-internal",
-  "version": "1.3.2",
+  "version": "1.3.4",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "windmill-utils-internal",
-      "version": "1.3.2",
+      "version": "1.3.4",
       "license": "Apache 2.0",
       "devDependencies": {
         "@types/node": "^24.2.0",

cli/windmill-utils-internal/package.json

@@ -1,6 +1,6 @@
 {
   "name": "windmill-utils-internal",
-  "version": "1.3.3",
+  "version": "1.3.4",
   "description": "Internal utility functions for Windmill",
   "main": "dist/cjs/index.js",
   "module": "dist/esm/index.js",

cli/windmill-utils-internal/src/parse/parse-schema.ts

@@ -108,6 +108,9 @@ export function argSigToJsonSchemaType(
   } else if (t === "datetime") {
     newS.type = "string";
     newS.format = "date-time";
+  } else if (t === "date") {
+    newS.type = "string";
+    newS.format = "date";
   } else if (typeof t !== "string" && "oneof" in t) {
     newS.type = "object";
     if (t.oneof) {

flake.nix

@@ -55,8 +55,11 @@
           nodejs
           postgresql
           pkg-config
-          clang
+          llvmPackages_18.clang
+          mold
           cmake
+          cyrus_sasl
+          krb5
         ];
         coursier = pkgs.fetchFromGitHub {
           owner = "coursier";
@@ -66,7 +69,7 @@
         };
 
         PKG_CONFIG_PATH = pkgs.lib.makeSearchPath "lib/pkgconfig"
-          (with pkgs; [ openssl.dev libxml2.dev xmlsec.dev libxslt.dev ]);
+          (with pkgs; [ openssl.dev libxml2.dev xmlsec.dev libxslt.dev cyrus_sasl.dev krb5.dev ]);
         RUSTY_V8_ARCHIVE = let
           # NOTE: needs to be same as in Cargo.toml
           version = "130.0.7";
@@ -303,6 +306,14 @@
           # RUST_LOG = "debug";
           # RUST_LOG = "kube=debug";
 
+          # Override cargo linker to use clang 18 (stdenv brings clang 21 which causes SIGSEGV with mold)
+          CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_LINKER = "${pkgs.llvmPackages_18.clang}/bin/clang";
+          CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER = "${pkgs.llvmPackages_18.clang}/bin/clang";
+          CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS = "-C link-arg=-fuse-ld=mold -C link-arg=-Wl,-rpath,${pkgs.lib.makeLibraryPath [ pkgs.openssl pkgs.libffi pkgs.cyrus_sasl pkgs.krb5 pkgs.libxml2 pkgs.xmlsec pkgs.libxslt stdenv.cc.cc.lib ]}";
+          CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_RUSTFLAGS = "-C link-arg=-fuse-ld=mold -C link-arg=-Wl,-rpath,${pkgs.lib.makeLibraryPath [ pkgs.openssl pkgs.libffi pkgs.cyrus_sasl pkgs.krb5 pkgs.libxml2 pkgs.xmlsec pkgs.libxslt stdenv.cc.cc.lib ]}";
+          # rpath for build scripts and proc macros (host compilation)
+          CARGO_HOST_RUSTFLAGS = "-C link-arg=-Wl,-rpath,${pkgs.lib.makeLibraryPath [ pkgs.openssl pkgs.libffi pkgs.cyrus_sasl pkgs.krb5 pkgs.libxml2 pkgs.xmlsec pkgs.libxslt stdenv.cc.cc.lib ]}";
+
           # See this issue: https://github.com/NixOS/nixpkgs/issues/370494
           # Allows to build jemalloc on nixos
           CFLAGS = "-Wno-error=int-conversion";
@@ -310,6 +321,7 @@
           # Need to tell bindgen where to find libclang
           LIBCLANG_PATH = "${pkgs.llvmPackages.libclang.lib}/lib";
 
+          # LD_LIBRARY_PATH set in shellHook with a wrapper to avoid leaking into git/ssh
           # LD_LIBRARY_PATH = "${pkgs.gcc.lib}/lib";
 
           # Set C flags for Rust's bindgen program. Unlike ordinary C

frontend/package-lock.json

@@ -79,14 +79,14 @@
 				"windmill-parser-wasm-java": "1.510.1",
 				"windmill-parser-wasm-nu": "1.510.1",
 				"windmill-parser-wasm-php": "1.574.1",
-				"windmill-parser-wasm-py": "1.623.1",
+				"windmill-parser-wasm-py": "^1.628.3",
 				"windmill-parser-wasm-regex": "1.625.0",
 				"windmill-parser-wasm-ruby": "1.526.1",
 				"windmill-parser-wasm-rust": "1.558.1",
 				"windmill-parser-wasm-ts": "1.623.1",
 				"windmill-parser-wasm-yaml": "1.593.0",
 				"windmill-sql-datatype-parser-wasm": "1.512.0",
-				"windmill-utils-internal": "^1.3.2",
+				"windmill-utils-internal": "^1.3.4",
 				"xterm": "^5.3.0",
 				"xterm-readline": "^1.1.2",
 				"y-monaco": "^0.1.4",
@@ -834,7 +834,6 @@
 			"version": "1.7.1",
 			"resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.7.1.tgz",
 			"integrity": "sha512-o1uhUASyo921r2XtHYOHy7gdkGLge8ghBEQHMWmyJFoXlpU58kIrhhN3w26lpQb6dspetweapMn2CSNwQ8I4wg==",
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"dependencies": {
@@ -846,7 +845,6 @@
 			"version": "1.7.1",
 			"resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.7.1.tgz",
 			"integrity": "sha512-PVtJr5CmLwYAU9PZDMITZoR5iAOShYREoR45EyyLrbntV50mdePTgUn4AmOw90Ifcj+x2kRjdzr1HP3RrNiHGA==",
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"dependencies": {
@@ -857,7 +855,6 @@
 			"version": "1.1.0",
 			"resolved": "https://registry.npmjs.org/@emnapi/wasi-threads/-/wasi-threads-1.1.0.tgz",
 			"integrity": "sha512-WI0DdZ8xFSbgMjR1sFsKABJ/C5OnRrjT06JXbZKexJGrDuPTzZdDYfFlsgcCXCyf+suG5QU2e/y1Wo2V/OapLQ==",
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"dependencies": {
@@ -1347,7 +1344,6 @@
 			"version": "1.1.0",
 			"resolved": "https://registry.npmjs.org/@napi-rs/wasm-runtime/-/wasm-runtime-1.1.0.tgz",
 			"integrity": "sha512-Fq6DJW+Bb5jaWE69/qOE0D1TUN9+6uWhCeZpdnSBk14pjLcCWR7Q8n49PTSPHazM37JqrsdpEthXy2xn6jWWiA==",
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"dependencies": {
@@ -1502,7 +1498,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1519,7 +1514,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1536,7 +1530,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1553,7 +1546,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1570,7 +1562,6 @@
 			"cpu": [
 				"arm"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1587,7 +1578,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1604,7 +1594,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1621,7 +1610,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1638,7 +1626,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1655,7 +1642,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1672,7 +1658,6 @@
 			"cpu": [
 				"wasm32"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"dependencies": {
@@ -1689,7 +1674,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -1706,7 +1690,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"os": [
@@ -2327,7 +2310,6 @@
 			"version": "0.10.1",
 			"resolved": "https://registry.npmjs.org/@tybys/wasm-util/-/wasm-util-0.10.1.tgz",
 			"integrity": "sha512-9tTaPJLSiejZKx+Bmog4uSubteqTvFrVrURwkmHixBo0G4seD0zUxp98E1DzUBJxLQ3NPwXrGKDiVjwx/DpPsg==",
-			"dev": true,
 			"license": "MIT",
 			"optional": true,
 			"dependencies": {
@@ -2955,115 +2937,6 @@
 			"integrity": "sha512-WmoN8qaIAo7WTYWbAZuG8PYEhn5fkz7dZrqTBZ7dtt//lL2Gwms1IcnQ5yHqjDfX8Ft5j4YzDM23f87zBfDe9g==",
 			"license": "ISC"
 		},
-		"node_modules/@vitest/browser": {
-			"version": "4.0.15",
-			"resolved": "https://registry.npmjs.org/@vitest/browser/-/browser-4.0.15.tgz",
-			"integrity": "sha512-zedtczX688KehaIaAv7m25CeDLb0gBtAOa2Oi1G1cqvSO5aLSVfH6lpZMJLW8BKYuWMxLQc9/5GYoM+jgvGIrw==",
-			"dev": true,
-			"license": "MIT",
-			"optional": true,
-			"peer": true,
-			"dependencies": {
-				"@vitest/mocker": "4.0.15",
-				"@vitest/utils": "4.0.15",
-				"magic-string": "^0.30.21",
-				"pixelmatch": "7.1.0",
-				"pngjs": "^7.0.0",
-				"sirv": "^3.0.2",
-				"tinyrainbow": "^3.0.3",
-				"ws": "^8.18.3"
-			},
-			"funding": {
-				"url": "https://opencollective.com/vitest"
-			},
-			"peerDependencies": {
-				"vitest": "4.0.15"
-			}
-		},
-		"node_modules/@vitest/browser-playwright": {
-			"version": "4.0.15",
-			"resolved": "https://registry.npmjs.org/@vitest/browser-playwright/-/browser-playwright-4.0.15.tgz",
-			"integrity": "sha512-94yVpDbb+ykiT7mK6ToonGnq2GIHEQGBTZTAzGxBGQXcVNCh54YKC2/WkfaDzxy0m6Kgw05kq3FYHKHu+wRdIA==",
-			"dev": true,
-			"license": "MIT",
-			"optional": true,
-			"peer": true,
-			"dependencies": {
-				"@vitest/browser": "4.0.15",
-				"@vitest/mocker": "4.0.15",
-				"tinyrainbow": "^3.0.3"
-			},
-			"funding": {
-				"url": "https://opencollective.com/vitest"
-			},
-			"peerDependencies": {
-				"playwright": "*",
-				"vitest": "4.0.15"
-			},
-			"peerDependenciesMeta": {
-				"playwright": {
-					"optional": false
-				}
-			}
-		},
-		"node_modules/@vitest/browser-playwright/node_modules/@vitest/mocker": {
-			"version": "4.0.15",
-			"resolved": "https://registry.npmjs.org/@vitest/mocker/-/mocker-4.0.15.tgz",
-			"integrity": "sha512-CZ28GLfOEIFkvCFngN8Sfx5h+Se0zN+h4B7yOsPVCcgtiO7t5jt9xQh2E1UkFep+eb9fjyMfuC5gBypwb07fvQ==",
-			"dev": true,
-			"license": "MIT",
-			"optional": true,
-			"peer": true,
-			"dependencies": {
-				"@vitest/spy": "4.0.15",
-				"estree-walker": "^3.0.3",
-				"magic-string": "^0.30.21"
-			},
-			"funding": {
-				"url": "https://opencollective.com/vitest"
-			},
-			"peerDependencies": {
-				"msw": "^2.4.9",
-				"vite": "^6.0.0 || ^7.0.0-0"
-			},
-			"peerDependenciesMeta": {
-				"msw": {
-					"optional": true
-				},
-				"vite": {
-					"optional": true
-				}
-			}
-		},
-		"node_modules/@vitest/browser/node_modules/@vitest/mocker": {
-			"version": "4.0.15",
-			"resolved": "https://registry.npmjs.org/@vitest/mocker/-/mocker-4.0.15.tgz",
-			"integrity": "sha512-CZ28GLfOEIFkvCFngN8Sfx5h+Se0zN+h4B7yOsPVCcgtiO7t5jt9xQh2E1UkFep+eb9fjyMfuC5gBypwb07fvQ==",
-			"dev": true,
-			"license": "MIT",
-			"optional": true,
-			"peer": true,
-			"dependencies": {
-				"@vitest/spy": "4.0.15",
-				"estree-walker": "^3.0.3",
-				"magic-string": "^0.30.21"
-			},
-			"funding": {
-				"url": "https://opencollective.com/vitest"
-			},
-			"peerDependencies": {
-				"msw": "^2.4.9",
-				"vite": "^6.0.0 || ^7.0.0-0"
-			},
-			"peerDependenciesMeta": {
-				"msw": {
-					"optional": true
-				},
-				"vite": {
-					"optional": true
-				}
-			}
-		},
 		"node_modules/@vitest/expect": {
 			"version": "4.0.15",
 			"resolved": "https://registry.npmjs.org/@vitest/expect/-/expect-4.0.15.tgz",
@@ -7185,7 +7058,7 @@
 			"version": "1.21.7",
 			"resolved": "https://registry.npmjs.org/jiti/-/jiti-1.21.7.tgz",
 			"integrity": "sha512-/imKNG4EbWNrVjoNC/1H5/9GFy+tqjGBHCaSsN+P2RnPqjsLmv6UD3Ej+Kj8nBWaRAwyk7kK5ZUc+OEatnTR3A==",
-			"dev": true,
+			"devOptional": true,
 			"license": "MIT",
 			"bin": {
 				"jiti": "bin/jiti.js"
@@ -7684,7 +7557,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7705,7 +7577,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7726,7 +7597,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7747,7 +7617,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7768,7 +7637,6 @@
 			"cpu": [
 				"arm"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7789,7 +7657,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7810,7 +7677,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7831,7 +7697,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7852,7 +7717,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7873,7 +7737,6 @@
 			"cpu": [
 				"arm64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -7894,7 +7757,6 @@
 			"cpu": [
 				"x64"
 			],
-			"dev": true,
 			"license": "MPL-2.0",
 			"optional": true,
 			"os": [
@@ -9950,21 +9812,6 @@
 				"node": ">= 6"
 			}
 		},
-		"node_modules/pixelmatch": {
-			"version": "7.1.0",
-			"resolved": "https://registry.npmjs.org/pixelmatch/-/pixelmatch-7.1.0.tgz",
-			"integrity": "sha512-1wrVzJ2STrpmONHKBy228LM1b84msXDUoAzVEl0R8Mz4Ce6EPr+IVtxm8+yvrqLYMHswREkjYFaMxnyGnaY3Ng==",
-			"dev": true,
-			"license": "ISC",
-			"optional": true,
-			"peer": true,
-			"dependencies": {
-				"pngjs": "^7.0.0"
-			},
-			"bin": {
-				"pixelmatch": "bin/pixelmatch"
-			}
-		},
 		"node_modules/pkg-types": {
 			"version": "1.3.1",
 			"resolved": "https://registry.npmjs.org/pkg-types/-/pkg-types-1.3.1.tgz",
@@ -10040,18 +9887,6 @@
 				"node": ">=4"
 			}
 		},
-		"node_modules/pngjs": {
-			"version": "7.0.0",
-			"resolved": "https://registry.npmjs.org/pngjs/-/pngjs-7.0.0.tgz",
-			"integrity": "sha512-LKWqWJRhstyYo9pGvgor/ivk2w94eSjE3RGVuzLGlr3NmD8bf7RcYGze1mNdEHRP6TRP6rMuDHk5t44hnTRyow==",
-			"dev": true,
-			"license": "MIT",
-			"optional": true,
-			"peer": true,
-			"engines": {
-				"node": ">=14.19.0"
-			}
-		},
 		"node_modules/postcss": {
 			"version": "8.5.6",
 			"resolved": "https://registry.npmjs.org/postcss/-/postcss-8.5.6.tgz",
@@ -12548,21 +12383,6 @@
 				}
 			}
 		},
-		"node_modules/svelte-check/node_modules/picomatch": {
-			"version": "4.0.3",
-			"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
-			"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
-			"dev": true,
-			"license": "MIT",
-			"optional": true,
-			"peer": true,
-			"engines": {
-				"node": ">=12"
-			},
-			"funding": {
-				"url": "https://github.com/sponsors/jonschlinkert"
-			}
-		},
 		"node_modules/svelte-eslint-parser": {
 			"version": "0.43.0",
 			"resolved": "https://registry.npmjs.org/svelte-eslint-parser/-/svelte-eslint-parser-0.43.0.tgz",
@@ -14603,9 +14423,9 @@
 			"integrity": "sha512-COyid6B1RYs+bpzUCInsA4HY/WZkpDLfkQ90+AqU/TVTpzYSbAC2JCbIwy0cRElBvlhI4bQ+9Wg6hSQKMpEkpA=="
 		},
 		"node_modules/windmill-parser-wasm-py": {
-			"version": "1.623.1",
-			"resolved": "https://registry.npmjs.org/windmill-parser-wasm-py/-/windmill-parser-wasm-py-1.623.1.tgz",
-			"integrity": "sha512-lFBlZg6hvhHzsU5oPJq0478UyMTZ9UKVd8Hc8ggmxPIHZaJBeJ+56NR75hmGwg0VJcRff8ed+zEm1PgmjVhD+w=="
+			"version": "1.628.3",
+			"resolved": "https://registry.npmjs.org/windmill-parser-wasm-py/-/windmill-parser-wasm-py-1.628.3.tgz",
+			"integrity": "sha512-TlluqknZpg8cZ+A3m6JFLPseY2PpKtDsxdj26fAnCUzKPtse8TxQR+n0dwC80rfW5TwdWSulvNGRDgcNuf7CTw=="
 		},
 		"node_modules/windmill-parser-wasm-regex": {
 			"version": "1.625.0",
@@ -14638,9 +14458,9 @@
 			"integrity": "sha512-uHNL8F72/Tf96xF3hOHnPDjkEyqXw7fNjcPJiUhth9sTQkcwUIoJMOdwm8/cs+j9kKVRJ4tgNYMHEBLylazp6g=="
 		},
 		"node_modules/windmill-utils-internal": {
-			"version": "1.3.2",
-			"resolved": "https://registry.npmjs.org/windmill-utils-internal/-/windmill-utils-internal-1.3.2.tgz",
-			"integrity": "sha512-sLE7sfmN3vBwPReqKorBrusnn83Qnm8vVFqh/7+kz8mf98ojMVDMgBginjlh9KBtVfWu9NhyjGS+WXRcnWalzA==",
+			"version": "1.3.4",
+			"resolved": "https://registry.npmjs.org/windmill-utils-internal/-/windmill-utils-internal-1.3.4.tgz",
+			"integrity": "sha512-XVypDKIZ6P4fwIjZwvuvq1m+j0rtAA7BDp1rI2F7hQ+VBKZUHsLskP+jgstXs+kN1LqGGsQJj4ecMYDImpIZ6A==",
 			"license": "Apache 2.0"
 		},
 		"node_modules/word-wrap": {
@@ -14785,29 +14605,6 @@
 				"node": "^14.17.0 || ^16.13.0 || >=18.0.0"
 			}
 		},
-		"node_modules/ws": {
-			"version": "8.19.0",
-			"resolved": "https://registry.npmjs.org/ws/-/ws-8.19.0.tgz",
-			"integrity": "sha512-blAT2mjOEIi0ZzruJfIhb3nps74PRWTCz1IjglWEEpQl5XS/UNama6u2/rjFkDDouqr4L67ry+1aGIALViWjDg==",
-			"license": "MIT",
-			"optional": true,
-			"peer": true,
-			"engines": {
-				"node": ">=10.0.0"
-			},
-			"peerDependencies": {
-				"bufferutil": "^4.0.1",
-				"utf-8-validate": ">=5.0.2"
-			},
-			"peerDependenciesMeta": {
-				"bufferutil": {
-					"optional": true
-				},
-				"utf-8-validate": {
-					"optional": true
-				}
-			}
-		},
 		"node_modules/xml-utils": {
 			"version": "1.10.2",
 			"resolved": "https://registry.npmjs.org/xml-utils/-/xml-utils-1.10.2.tgz",

frontend/package.json

@@ -149,14 +149,14 @@
 		"windmill-parser-wasm-java": "1.510.1",
 		"windmill-parser-wasm-nu": "1.510.1",
 		"windmill-parser-wasm-php": "1.574.1",
-		"windmill-parser-wasm-py": "1.623.1",
+		"windmill-parser-wasm-py": "^1.628.3",
 		"windmill-parser-wasm-regex": "1.625.0",
 		"windmill-parser-wasm-ruby": "1.526.1",
 		"windmill-parser-wasm-rust": "1.558.1",
 		"windmill-parser-wasm-ts": "1.623.1",
 		"windmill-parser-wasm-yaml": "1.593.0",
 		"windmill-sql-datatype-parser-wasm": "1.512.0",
-		"windmill-utils-internal": "^1.3.2",
+		"windmill-utils-internal": "^1.3.4",
 		"xterm": "^5.3.0",
 		"xterm-readline": "^1.1.2",
 		"y-monaco": "^0.1.4",

frontend/src/lib/components/ArgInput.svelte

@@ -258,7 +258,11 @@
 			nvalue = structuredClone($state.snapshot(defaultValue))
 			if (defaultValue === undefined || defaultValue === null) {
 				if (inputCat === 'string') {
-					nvalue = nullable ? null : format === 'uuid' && extra?.['x-auto-generate'] ? randomUUID() : ''
+					nvalue = nullable
+						? null
+						: format === 'uuid' && extra?.['x-auto-generate']
+							? randomUUID()
+							: ''
 				} else if (inputCat == 'enum' && required) {
 					let firstV = enum_?.[0]
 					if (typeof firstV === 'string') {
@@ -1177,7 +1181,6 @@
 										titleClass="text-2xs"
 									/>
 								{/if}
-	
 							{:else if disabled}
 								<textarea disabled></textarea>
 							{:else}

frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts

@@ -899,11 +899,6 @@ class AIChatManager {
 	}
 
 	saveAndClear = async () => {
-		console.log('saveAndClear called', {
-			hasAbortController: !!this.abortController,
-			isLoading: this.loading,
-			stack: new Error().stack
-		})
 		this.cancel('saveAndClear')
 		await this.historyManager.save(this.displayMessages, this.messages)
 		this.displayMessages = []