diff --git a/backend/.sqlx/query-38e782610bbabd5902e9b95ea6e026efb056f90e5cc6a505f7164d69ca046a3f.json b/backend/.sqlx/query-03dceb989267d5bea286b83adedf43d53fd14f6f66206a805305e35abbc5a691.json
similarity index 71%
rename from backend/.sqlx/query-38e782610bbabd5902e9b95ea6e026efb056f90e5cc6a505f7164d69ca046a3f.json
rename to backend/.sqlx/query-03dceb989267d5bea286b83adedf43d53fd14f6f66206a805305e35abbc5a691.json
index c921738495..8232e1b35c 100644
--- a/backend/.sqlx/query-38e782610bbabd5902e9b95ea6e026efb056f90e5cc6a505f7164d69ca046a3f.json
+++ b/backend/.sqlx/query-03dceb989267d5bea286b83adedf43d53fd14f6f66206a805305e35abbc5a691.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "WITH t1 AS (UPDATE websocket_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t2 AS (UPDATE kafka_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t3 AS (UPDATE postgres_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t4 AS (UPDATE mqtt_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t5 AS (UPDATE nats_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t6 AS (UPDATE sqs_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4) UPDATE gcp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4",
+  "query": "WITH t1 AS (UPDATE websocket_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t2 AS (UPDATE kafka_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t3 AS (UPDATE postgres_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t4 AS (UPDATE mqtt_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t5 AS (UPDATE nats_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t6 AS (UPDATE sqs_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t7 AS (UPDATE asset_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4 AND is_implicit = false) UPDATE gcp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4",
   "describe": {
     "columns": [],
     "parameters": {
@@ -13,5 +13,5 @@
     },
     "nullable": []
   },
-  "hash": "38e782610bbabd5902e9b95ea6e026efb056f90e5cc6a505f7164d69ca046a3f"
+  "hash": "03dceb989267d5bea286b83adedf43d53fd14f6f66206a805305e35abbc5a691"
 }
diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
index 713ccb9dd3..36ddb8ab9f 100644
--- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
+++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
@@ -15,7 +15,7 @@
       ]
     },
     "nullable": [
-      null
+      true
     ]
   },
   "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
diff --git a/backend/.sqlx/query-941212fc8cc6db845d42320e337823424ba8a82b21efce174e1215094342c816.json b/backend/.sqlx/query-941212fc8cc6db845d42320e337823424ba8a82b21efce174e1215094342c816.json
new file mode 100644
index 0000000000..79de2de617
--- /dev/null
+++ b/backend/.sqlx/query-941212fc8cc6db845d42320e337823424ba8a82b21efce174e1215094342c816.json
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT owner_script_path FROM asset_trigger WHERE workspace_id = $1 AND path = $2 AND is_implicit = true",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "owner_script_path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "941212fc8cc6db845d42320e337823424ba8a82b21efce174e1215094342c816"
+}
diff --git a/backend/.sqlx/query-94e3820c19701da35d8f8c3441ffb0b696dee5eb77f74a9635852221528169aa.json b/backend/.sqlx/query-94e3820c19701da35d8f8c3441ffb0b696dee5eb77f74a9635852221528169aa.json
new file mode 100644
index 0000000000..a3fb266c47
--- /dev/null
+++ b/backend/.sqlx/query-94e3820c19701da35d8f8c3441ffb0b696dee5eb77f74a9635852221528169aa.json
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM asset_trigger WHERE workspace_id = $1 AND owner_script_path = $2 AND is_implicit = true",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "94e3820c19701da35d8f8c3441ffb0b696dee5eb77f74a9635852221528169aa"
+}
diff --git a/backend/.sqlx/query-e3d9b2bd73f1864255d27df33f2bdafbc126b6261c0046358022a345c4c8cb34.json b/backend/.sqlx/query-e3d9b2bd73f1864255d27df33f2bdafbc126b6261c0046358022a345c4c8cb34.json
new file mode 100644
index 0000000000..75f73c520d
--- /dev/null
+++ b/backend/.sqlx/query-e3d9b2bd73f1864255d27df33f2bdafbc126b6261c0046358022a345c4c8cb34.json
@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO asset_trigger ( workspace_id, path, script_path, is_flow, owner_script_path, owner_script_hash, is_implicit, on_event, subscription_set, fires, debounce_s, partition_map, cancel_on_new, backlog, edited_by, email ) VALUES ( $1, $2, $3, false, $3, $4, true, $5, $6, $7, $8, $9, $10, $11, $12, $13 )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Int8",
+        "Varchar",
+        "Jsonb",
+        "Varchar",
+        "Int4",
+        "Jsonb",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e3d9b2bd73f1864255d27df33f2bdafbc126b6261c0046358022a345c4c8cb34"
+}
diff --git a/backend/.sqlx/query-f5f544a5b75cc46ad1f418b24859c4a04ef4c74a6df1c5b49c6e0b9a3665060a.json b/backend/.sqlx/query-f5f544a5b75cc46ad1f418b24859c4a04ef4c74a6df1c5b49c6e0b9a3665060a.json
new file mode 100644
index 0000000000..602bfab772
--- /dev/null
+++ b/backend/.sqlx/query-f5f544a5b75cc46ad1f418b24859c4a04ef4c74a6df1c5b49c6e0b9a3665060a.json
@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE asset_trigger SET owner_script_path = $1, script_path = $1, path = $2, server_id = NULL WHERE workspace_id = $3 AND owner_script_path = $4 AND is_implicit = true",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f5f544a5b75cc46ad1f418b24859c4a04ef4c74a6df1c5b49c6e0b9a3665060a"
+}
diff --git a/backend/migrations/20260418150913_asset_event.down.sql b/backend/migrations/20260418150913_asset_event.down.sql
new file mode 100644
index 0000000000..39c11754d7
--- /dev/null
+++ b/backend/migrations/20260418150913_asset_event.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS asset_event;
diff --git a/backend/migrations/20260418150913_asset_event.up.sql b/backend/migrations/20260418150913_asset_event.up.sql
new file mode 100644
index 0000000000..111e6a2de0
--- /dev/null
+++ b/backend/migrations/20260418150913_asset_event.up.sql
@@ -0,0 +1,25 @@
+-- Append-only event log for reactive asset-change triggers.
+-- Written whenever a job produces/updates an asset ('w' or 'rw' access).
+-- Stage 1 of asset-change implicit triggers plan.
+
+CREATE TABLE asset_event (
+    id BIGSERIAL PRIMARY KEY,
+    workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+    asset_kind ASSET_KIND NOT NULL,
+    asset_path VARCHAR(255) NOT NULL,
+    partition_key VARCHAR(255),
+    job_id UUID NOT NULL,
+    script_hash BIGINT,
+    access_type ASSET_ACCESS_TYPE NOT NULL,
+    columns JSONB,
+    metadata JSONB,
+    at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX asset_event_kind_path_at_idx ON asset_event (workspace_id, asset_kind, asset_path, at DESC);
+CREATE INDEX asset_event_job_id_idx ON asset_event (job_id);
+
+GRANT ALL ON asset_event TO windmill_user;
+GRANT ALL ON asset_event TO windmill_admin;
+GRANT ALL ON asset_event_id_seq TO windmill_user;
+GRANT ALL ON asset_event_id_seq TO windmill_admin;
diff --git a/backend/migrations/20260418151526_asset_trigger.down.sql b/backend/migrations/20260418151526_asset_trigger.down.sql
new file mode 100644
index 0000000000..86503be106
--- /dev/null
+++ b/backend/migrations/20260418151526_asset_trigger.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS asset_trigger;
diff --git a/backend/migrations/20260418151526_asset_trigger.up.sql b/backend/migrations/20260418151526_asset_trigger.up.sql
new file mode 100644
index 0000000000..18f6f62918
--- /dev/null
+++ b/backend/migrations/20260418151526_asset_trigger.up.sql
@@ -0,0 +1,91 @@
+-- `asset_trigger` holds one row per reactive asset-change subscription.
+-- Implicit triggers (is_implicit=true) are projected from `#trigger: asset`
+-- annotations on every script save. Never editable via API/CLI.
+
+CREATE TABLE asset_trigger (
+    workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
+    path VARCHAR(255) NOT NULL,
+    script_path VARCHAR(255) NOT NULL,
+    is_flow BOOLEAN NOT NULL DEFAULT false,
+
+    -- implicit-trigger pinning
+    owner_script_path VARCHAR(255),
+    owner_script_hash BIGINT,
+    is_implicit BOOLEAN NOT NULL DEFAULT false,
+
+    -- trigger config
+    on_event VARCHAR(32) NOT NULL,
+    subscription_set JSONB NOT NULL,
+    fires VARCHAR(16) NOT NULL DEFAULT 'all',
+    debounce_s INT NOT NULL DEFAULT 30,
+    partition_map JSONB,
+    cancel_on_new BOOLEAN NOT NULL DEFAULT false,
+    backlog VARCHAR(16) NOT NULL DEFAULT 'coalesce',
+
+    -- boilerplate shared with other trigger types
+    error TEXT,
+    server_id VARCHAR(64),
+    last_server_ping TIMESTAMPTZ,
+    error_handler_path VARCHAR(255),
+    error_handler_args JSONB,
+    retry JSONB,
+    extra_perms JSONB NOT NULL DEFAULT '{}',
+    edited_by VARCHAR(255) NOT NULL,
+    edited_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    email VARCHAR(255) NOT NULL,
+    mode TRIGGER_MODE NOT NULL DEFAULT 'enabled',
+    labels TEXT[],
+
+    PRIMARY KEY (workspace_id, path)
+);
+
+CREATE INDEX asset_trigger_owner_idx ON asset_trigger (workspace_id, owner_script_path) WHERE is_implicit = true;
+CREATE INDEX asset_trigger_subscription_set_idx ON asset_trigger USING GIN (subscription_set);
+
+GRANT ALL ON asset_trigger TO windmill_user;
+GRANT ALL ON asset_trigger TO windmill_admin;
+
+ALTER TABLE asset_trigger ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY admin_policy ON asset_trigger FOR ALL TO windmill_admin USING (true);
+
+CREATE POLICY see_folder_extra_perms_user_select ON asset_trigger FOR SELECT TO windmill_user
+USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_insert ON asset_trigger FOR INSERT TO windmill_user
+WITH CHECK (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_update ON asset_trigger FOR UPDATE TO windmill_user
+USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_delete ON asset_trigger FOR DELETE TO windmill_user
+USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+
+CREATE POLICY see_own ON asset_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'u' AND SPLIT_PART(asset_trigger.path, '/', 2) = current_setting('session.user'));
+CREATE POLICY see_member ON asset_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'g' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
+
+CREATE POLICY see_extra_perms_user_select ON asset_trigger FOR SELECT TO windmill_user
+USING (extra_perms ? CONCAT('u/', current_setting('session.user')));
+CREATE POLICY see_extra_perms_user_insert ON asset_trigger FOR INSERT TO windmill_user
+WITH CHECK ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+CREATE POLICY see_extra_perms_user_update ON asset_trigger FOR UPDATE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+CREATE POLICY see_extra_perms_user_delete ON asset_trigger FOR DELETE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+
+CREATE POLICY see_extra_perms_groups_select ON asset_trigger FOR SELECT TO windmill_user
+USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[]);
+CREATE POLICY see_extra_perms_groups_insert ON asset_trigger FOR INSERT TO windmill_user
+WITH CHECK (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms)
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));
+CREATE POLICY see_extra_perms_groups_update ON asset_trigger FOR UPDATE TO windmill_user
+USING (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms)
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));
+CREATE POLICY see_extra_perms_groups_delete ON asset_trigger FOR DELETE TO windmill_user
+USING (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms)
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));
diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs
index c4b928bcb5..35b4f3955b 100644
--- a/backend/windmill-api-scripts/src/scripts.rs
+++ b/backend/windmill-api-scripts/src/scripts.rs
@@ -1161,6 +1161,13 @@ async fn create_script_internal<'c>(
                     "Error updating triggers due to runnable path change: {e:#}"
                 ))
             })?;
+            windmill_common::implicit_triggers::rename_implicit_asset_triggers_for_script(
+                &mut tx, &w_id, p_path, &ns.path,
+            )
+            .await
+            .map_err(|e| {
+                error::Error::internal_err(format!("Error renaming implicit asset triggers: {e:#}"))
+            })?;
         }
 
         for schedule in schedulables {
@@ -1277,6 +1284,19 @@ async fn create_script_internal<'c>(
             .await?;
     }
 
+    windmill_common::implicit_triggers::sync_implicit_asset_triggers_for_script(
+        &mut tx,
+        &w_id,
+        &ns.path,
+        hash.0,
+        ns.language,
+        &ns.content,
+        ns.assets.as_deref().unwrap_or(&[]),
+        &authed.email,
+        &authed.username,
+    )
+    .await?;
+
     let permissioned_as = username_to_permissioned_as(&authed.username);
     if let Some(parent_hash) = ns.parent_hash {
         tracing::info!(
@@ -2179,6 +2199,10 @@ async fn archive_script_by_path(
     .map_err(|e| Error::internal_err(format!("archiving script in {w_id}: {e:#}")))?;
 
     clear_static_asset_usage(&mut *tx, &w_id, path, AssetUsageKind::Script).await?;
+    windmill_common::implicit_triggers::delete_implicit_asset_triggers_for_script(
+        &mut tx, &w_id, path,
+    )
+    .await?;
 
     audit_log(
         &mut *tx,
@@ -2415,6 +2439,11 @@ async fn delete_script_by_path(
     .await
     .map_err(|e| Error::internal_err(format!("deleting script by path {w_id}: {e:#}")))?;
 
+    windmill_common::implicit_triggers::delete_implicit_asset_triggers_for_script(
+        &mut tx, &w_id, path,
+    )
+    .await?;
+
     if !trash_scripts.is_empty() {
         let mut trash_data = serde_json::json!({"scripts": trash_scripts});
         if !trash_drafts.is_empty() {
diff --git a/backend/windmill-api/src/asset_triggers.rs b/backend/windmill-api/src/asset_triggers.rs
new file mode 100644
index 0000000000..612bd7c862
--- /dev/null
+++ b/backend/windmill-api/src/asset_triggers.rs
@@ -0,0 +1,149 @@
+//! Read-only HTTP surface for `asset_trigger` rows (stage 3).
+//!
+//! v1 of asset-change triggers exposes only implicit triggers (projected
+//! from `#trigger: asset` annotations in script source). Mutations go
+//! through the script save path, so every write endpoint here returns
+//! **409 Conflict** with a pointer back to the owning script.
+
+use axum::{
+    extract::{Extension, Path},
+    http::StatusCode,
+    routing::{delete, get, post},
+    Json, Router,
+};
+use serde::{Deserialize, Serialize};
+use sqlx::types::Json as SqlxJson;
+
+use windmill_common::{
+    db::UserDB,
+    error::{Error, JsonResult, Result},
+    utils::StripPath,
+};
+
+use crate::db::ApiAuthed;
+
+pub fn workspaced_service() -> Router {
+    Router::new()
+        .route("/list", get(list_asset_triggers))
+        .route("/get/{*path}", get(get_asset_trigger))
+        .route("/create", post(reject_mutate))
+        .route("/update/{*path}", post(reject_mutate_path))
+        .route("/delete/{*path}", delete(reject_mutate_path))
+}
+
+#[derive(Serialize, Deserialize, Debug, sqlx::FromRow)]
+pub struct AssetTriggerRow {
+    pub workspace_id: String,
+    pub path: String,
+    pub script_path: String,
+    pub is_flow: bool,
+    pub owner_script_path: Option<String>,
+    pub owner_script_hash: Option<i64>,
+    pub is_implicit: bool,
+    pub on_event: String,
+    pub subscription_set: SqlxJson<serde_json::Value>,
+    pub fires: String,
+    pub debounce_s: i32,
+    pub partition_map: Option<SqlxJson<serde_json::Value>>,
+    pub cancel_on_new: bool,
+    pub backlog: String,
+    pub error: Option<String>,
+    pub edited_by: String,
+    pub edited_at: chrono::DateTime<chrono::Utc>,
+    pub email: String,
+}
+
+async fn list_asset_triggers(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path(w_id): Path<String>,
+) -> JsonResult<Vec<AssetTriggerRow>> {
+    let mut tx = user_db.begin(&authed).await?;
+    let rows = sqlx::query_as::<_, AssetTriggerRow>(
+        "SELECT workspace_id, path, script_path, is_flow, owner_script_path, \
+                owner_script_hash, is_implicit, on_event, subscription_set, fires, \
+                debounce_s, partition_map, cancel_on_new, backlog, error, edited_by, \
+                edited_at, email \
+         FROM asset_trigger \
+         WHERE workspace_id = $1 \
+         ORDER BY edited_at DESC",
+    )
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+    tx.commit().await?;
+    Ok(Json(rows))
+}
+
+async fn get_asset_trigger(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, path)): Path<(String, StripPath)>,
+) -> JsonResult<AssetTriggerRow> {
+    let path = path.to_path();
+    let mut tx = user_db.begin(&authed).await?;
+    let row = sqlx::query_as::<_, AssetTriggerRow>(
+        "SELECT workspace_id, path, script_path, is_flow, owner_script_path, \
+                owner_script_hash, is_implicit, on_event, subscription_set, fires, \
+                debounce_s, partition_map, cancel_on_new, backlog, error, edited_by, \
+                edited_at, email \
+         FROM asset_trigger \
+         WHERE workspace_id = $1 AND path = $2",
+    )
+    .bind(&w_id)
+    .bind(path)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or_else(|| Error::NotFound(format!("asset_trigger {path} not found")))?;
+    tx.commit().await?;
+    Ok(Json(row))
+}
+
+#[derive(Deserialize)]
+struct RejectBody {
+    #[allow(dead_code)]
+    path: Option<String>,
+}
+
+async fn reject_mutate(
+    _authed: ApiAuthed,
+    Extension(_user_db): Extension<UserDB>,
+    Path(_w_id): Path<String>,
+    Json(_body): Json<RejectBody>,
+) -> Result<()> {
+    Err(implicit_mutation_error(None))
+}
+
+async fn reject_mutate_path(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, path)): Path<(String, StripPath)>,
+) -> Result<()> {
+    let path = path.to_path();
+    let mut tx = user_db.begin(&authed).await?;
+    let owner = sqlx::query_scalar!(
+        "SELECT owner_script_path FROM asset_trigger \
+         WHERE workspace_id = $1 AND path = $2 AND is_implicit = true",
+        &w_id,
+        path,
+    )
+    .fetch_optional(&mut *tx)
+    .await?
+    .flatten();
+    tx.commit().await?;
+    Err(implicit_mutation_error(owner))
+}
+
+fn implicit_mutation_error(owner: Option<String>) -> Error {
+    let msg = match owner {
+        Some(script_path) => format!(
+            "Implicit asset trigger — edit `#trigger: asset` annotations in script {script_path}."
+        ),
+        None => {
+            "Asset triggers are implicit-only in v1. Edit `#trigger: asset` annotations in the \
+             owning script to modify them."
+                .to_string()
+        }
+    };
+    Error::Generic(StatusCode::CONFLICT, msg)
+}
diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs
index 9cb517bb44..12299caccf 100644
--- a/backend/windmill-api/src/jobs.rs
+++ b/backend/windmill-api/src/jobs.rs
@@ -5681,6 +5681,8 @@ fn register_potential_assets_on_inline_execution(
                     job_id,
                     workspace_id: w_id.to_string(),
                     created_at: None,
+                    partition_key: None,
+                    script_hash: None,
                 });
             }
         }
diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs
index c11f89a1c4..80d4e1921a 100644
--- a/backend/windmill-api/src/lib.rs
+++ b/backend/windmill-api/src/lib.rs
@@ -68,6 +68,7 @@ use windmill_common::error::AppError;
 mod ai;
 mod apps;
 pub mod args;
+mod asset_triggers;
 mod audit;
 pub mod auth;
 #[cfg(all(feature = "private", feature = "parquet"))]
@@ -546,6 +547,7 @@ pub async fn run_server(
                         .nest("/acls", granular_acls::workspaced_service())
                         .nest("/apps", apps::workspaced_service(request_size_limit * 5))
                         .nest("/assets", windmill_api_assets::workspaced_service())
+                        .nest("/asset_triggers", asset_triggers::workspaced_service())
                         .nest("/audit", audit::workspaced_service())
                         .nest("/capture", capture::workspaced_service())
                         .nest(
diff --git a/backend/windmill-common/src/implicit_triggers.rs b/backend/windmill-common/src/implicit_triggers.rs
new file mode 100644
index 0000000000..221165ff53
--- /dev/null
+++ b/backend/windmill-common/src/implicit_triggers.rs
@@ -0,0 +1,245 @@
+//! Implicit asset triggers: projected from `#trigger: asset` annotations
+//! in script source onto rows in `asset_trigger` with `is_implicit = true`.
+//!
+//! Every script save re-parses annotations and calls
+//! [`sync_implicit_asset_triggers_for_script`], which computes the
+//! subscription set from the script's inferred read assets and upserts one
+//! row per asset annotation (v1 allows at most one). Implicit rows are never
+//! edited via API or CLI — the trigger stays in perfect sync with the script.
+
+use serde::{Deserialize, Serialize};
+use sqlx::{types::Json, Postgres, Transaction};
+
+use windmill_types::assets::{AssetKind, AssetUsageAccessType, AssetWithAltAccessType};
+
+use crate::{
+    error::Error,
+    scripts::ScriptLang,
+    trigger_annotations::{
+        parse_trigger_annotations, AssetTriggerAnnotation, TriggerAnnotation,
+        TriggerAnnotationError,
+    },
+};
+
+/// Deterministic path used by implicit asset triggers. Keeps the original
+/// script path prefix so row-level-security policies on `asset_trigger`
+/// work unchanged (`SPLIT_PART(path, '/', 1)` — 'u' / 'f' / 'g').
+pub const IMPLICIT_TRIGGER_PATH_SUFFIX: &str = "/__asset_trigger__";
+
+pub fn implicit_trigger_path(script_path: &str) -> String {
+    format!("{script_path}{IMPLICIT_TRIGGER_PATH_SUFFIX}")
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SubscriptionEntry {
+    pub kind: AssetKind,
+    pub path: String,
+}
+
+/// JSONB shape stored in `asset_trigger.subscription_set`. Uses an object
+/// wrapper with a `paths` array so stage 8 (partitioning) can add sibling
+/// fields without a migration. Indexed with GIN — containment queries like
+/// `subscription_set -> 'paths' @> ...` remain indexable.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SubscriptionSet {
+    pub paths: Vec<SubscriptionEntry>,
+}
+
+#[derive(thiserror::Error, Debug)]
+pub enum SyncError {
+    #[error("trigger annotation: {0}")]
+    Annotation(#[from] TriggerAnnotationError),
+    #[error("sql: {0}")]
+    Sql(#[from] sqlx::Error),
+}
+
+impl From<SyncError> for Error {
+    fn from(e: SyncError) -> Self {
+        Error::BadRequest(e.to_string())
+    }
+}
+
+/// Re-sync implicit asset triggers for a script. Called inside the script-
+/// save transaction after the script row has been inserted.
+///
+/// `inferred_read_assets` is the static asset-usage list captured at save
+/// time (`ns.assets` on the save handler). Only entries with `r` or `rw`
+/// access contribute to the subscription set — reactive triggers subscribe
+/// to *upstreams*, not the script's own writes.
+pub async fn sync_implicit_asset_triggers_for_script(
+    tx: &mut Transaction<'_, Postgres>,
+    workspace_id: &str,
+    script_path: &str,
+    script_hash: i64,
+    language: ScriptLang,
+    content: &str,
+    inferred_read_assets: &[AssetWithAltAccessType],
+    author_email: &str,
+    author_username: &str,
+) -> std::result::Result<(), SyncError> {
+    let anns = parse_trigger_annotations(content, language)?;
+    let asset_annotations: Vec<&AssetTriggerAnnotation> = anns
+        .iter()
+        .map(|a| match a {
+            TriggerAnnotation::Asset(inner) => inner,
+        })
+        .collect();
+
+    // Delete all existing implicit rows for this owner — we'll re-insert the
+    // current ones below. Safer than a diff since v1 is single-annotation.
+    sqlx::query!(
+        "DELETE FROM asset_trigger \
+         WHERE workspace_id = $1 AND owner_script_path = $2 AND is_implicit = true",
+        workspace_id,
+        script_path,
+    )
+    .execute(&mut **tx)
+    .await?;
+
+    for ann in asset_annotations {
+        let paths = resolve_subscription_set(ann, inferred_read_assets);
+        let subscription_set = SubscriptionSet { paths };
+        let trigger_path = implicit_trigger_path(script_path);
+        let on_event = match ann.on {
+            crate::trigger_annotations::AssetTriggerEvent::Change => "change",
+        };
+        let fires = match ann.fires {
+            crate::trigger_annotations::FiresMode::All => "all",
+            crate::trigger_annotations::FiresMode::Any => "any",
+        };
+        let backlog = match ann.backlog {
+            crate::trigger_annotations::BacklogMode::Coalesce => "coalesce",
+            crate::trigger_annotations::BacklogMode::Replay => "replay",
+            crate::trigger_annotations::BacklogMode::Skip => "skip",
+        };
+        sqlx::query!(
+            "INSERT INTO asset_trigger ( \
+                workspace_id, path, script_path, is_flow, \
+                owner_script_path, owner_script_hash, is_implicit, \
+                on_event, subscription_set, fires, debounce_s, partition_map, \
+                cancel_on_new, backlog, \
+                edited_by, email \
+             ) VALUES ( \
+                $1, $2, $3, false, \
+                $3, $4, true, \
+                $5, $6, $7, $8, $9, \
+                $10, $11, \
+                $12, $13 \
+             )",
+            workspace_id,
+            trigger_path,
+            script_path,
+            script_hash,
+            on_event,
+            Json(&subscription_set) as _,
+            fires,
+            ann.debounce_s,
+            Json(&ann.partition_map) as _,
+            ann.cancel_on_new,
+            backlog,
+            author_username,
+            author_email,
+        )
+        .execute(&mut **tx)
+        .await?;
+    }
+
+    Ok(())
+}
+
+/// Resolve the reactive subscription set from the annotation and the
+/// script's inferred reads. `only` overrides the inferred reads entirely;
+/// otherwise we take inferred reads minus `exclude` plus `extra`, filtered
+/// by `kinds` if specified. Writes are ignored.
+fn resolve_subscription_set(
+    ann: &AssetTriggerAnnotation,
+    inferred_reads: &[AssetWithAltAccessType],
+) -> Vec<SubscriptionEntry> {
+    let read_paths: Vec<SubscriptionEntry> = if let Some(only) = &ann.only {
+        // `only` specifies bare paths without kind — default to Resource.
+        // Kinds filter (below) still applies, so `kinds=[s3object]` with
+        // `only=[f/foo]` yields an s3object subscription.
+        let kind = ann
+            .kinds
+            .as_ref()
+            .and_then(|ks| ks.first().copied())
+            .unwrap_or(AssetKind::Resource);
+        only.iter()
+            .map(|p| SubscriptionEntry { kind, path: p.clone() })
+            .collect()
+    } else {
+        inferred_reads
+            .iter()
+            .filter(|a| {
+                matches!(
+                    a.access_type.or(a.alt_access_type),
+                    Some(AssetUsageAccessType::R) | Some(AssetUsageAccessType::RW)
+                )
+            })
+            .filter(|a| !ann.exclude.iter().any(|e| e == &a.path))
+            .map(|a| SubscriptionEntry { kind: a.kind, path: a.path.clone() })
+            .collect()
+    };
+
+    let extras = ann.extra.iter().map(|p| SubscriptionEntry {
+        kind: ann
+            .kinds
+            .as_ref()
+            .and_then(|ks| ks.first().copied())
+            .unwrap_or(AssetKind::Resource),
+        path: p.clone(),
+    });
+
+    let mut all: Vec<SubscriptionEntry> = read_paths.into_iter().chain(extras).collect();
+
+    if let Some(kinds) = &ann.kinds {
+        all.retain(|e| kinds.contains(&e.kind));
+    }
+
+    // Dedup (preserve first-seen order).
+    let mut seen = std::collections::HashSet::new();
+    all.retain(|e| seen.insert((e.kind, e.path.clone())));
+    all
+}
+
+/// Delete all implicit asset triggers owned by a script. Called when a
+/// script is archived or deleted.
+pub async fn delete_implicit_asset_triggers_for_script(
+    tx: &mut Transaction<'_, Postgres>,
+    workspace_id: &str,
+    script_path: &str,
+) -> std::result::Result<(), sqlx::Error> {
+    sqlx::query!(
+        "DELETE FROM asset_trigger \
+         WHERE workspace_id = $1 AND owner_script_path = $2 AND is_implicit = true",
+        workspace_id,
+        script_path,
+    )
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
+
+/// Rename implicit triggers when the owning script is renamed. Keeps the
+/// trigger path aligned with the new script path so the deterministic
+/// addressing remains stable.
+pub async fn rename_implicit_asset_triggers_for_script(
+    tx: &mut Transaction<'_, Postgres>,
+    workspace_id: &str,
+    old_script_path: &str,
+    new_script_path: &str,
+) -> std::result::Result<(), sqlx::Error> {
+    let new_trigger_path = implicit_trigger_path(new_script_path);
+    sqlx::query!(
+        "UPDATE asset_trigger \
+         SET owner_script_path = $1, script_path = $1, path = $2, server_id = NULL \
+         WHERE workspace_id = $3 AND owner_script_path = $4 AND is_implicit = true",
+        new_script_path,
+        new_trigger_path,
+        workspace_id,
+        old_script_path,
+    )
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs
index 52e57ea67f..80f88251c5 100644
--- a/backend/windmill-common/src/lib.rs
+++ b/backend/windmill-common/src/lib.rs
@@ -63,6 +63,7 @@ pub mod flow_status;
 pub mod flows;
 pub mod folders;
 pub mod global_settings;
+pub mod implicit_triggers;
 pub mod indexer;
 pub mod instance_config;
 pub mod job_metrics;
@@ -70,6 +71,7 @@ pub mod log_context;
 pub mod min_version;
 pub mod notify_events;
 pub mod runtime_assets;
+pub mod trigger_annotations;
 pub mod workspace_dependencies;
 
 #[cfg(feature = "private")]
diff --git a/backend/windmill-common/src/runtime_assets.rs b/backend/windmill-common/src/runtime_assets.rs
index ecd863f482..fc601353a2 100644
--- a/backend/windmill-common/src/runtime_assets.rs
+++ b/backend/windmill-common/src/runtime_assets.rs
@@ -69,6 +69,9 @@ pub struct InsertRuntimeAssetParams {
     pub access_type: Option<AssetUsageAccessType>,
     pub created_at: Option<chrono::DateTime<chrono::Utc>>,
     pub columns: Option<BTreeMap<String, AssetUsageAccessType>>,
+    // Forward-compatibility fields for asset-change triggers (v1 unused: partitioning lands in stage 8).
+    pub partition_key: Option<String>,
+    pub script_hash: Option<i64>,
 }
 
 async fn insert_runtime_assets(
@@ -90,6 +93,57 @@ async fn insert_runtime_assets(
         query_builder.push(" ON CONFLICT DO NOTHING");
         query_builder.build().execute(executor).await?;
     }
+    insert_asset_events(executor, assets).await?;
+    Ok(())
+}
+
+/// Append rows to `asset_event` for every write/read-write asset access,
+/// and pg_notify on channel `asset_event` so the asset-trigger dispatcher
+/// can react in near-real-time. No-op for read-only accesses.
+async fn insert_asset_events(
+    executor: &Pool<Postgres>,
+    assets: &[InsertRuntimeAssetParams],
+) -> error::Result<()> {
+    let write_assets: Vec<&InsertRuntimeAssetParams> = assets
+        .iter()
+        .filter(|a| {
+            matches!(
+                a.access_type,
+                Some(AssetUsageAccessType::W) | Some(AssetUsageAccessType::RW)
+            )
+        })
+        .collect();
+    if write_assets.is_empty() {
+        return Ok(());
+    }
+    for chunk in write_assets.chunks(1000) {
+        let mut query_builder = QueryBuilder::new(
+            "WITH inserted AS (INSERT INTO asset_event \
+             (workspace_id, asset_kind, asset_path, partition_key, job_id, script_hash, access_type, columns, at) ",
+        );
+        query_builder.push_values(chunk, |mut b, asset| {
+            b.push_bind(&asset.workspace_id)
+                .push_bind(&asset.asset_kind)
+                .push_bind(&asset.asset_path)
+                .push_bind(&asset.partition_key)
+                .push_bind(asset.job_id)
+                .push_bind(&asset.script_hash)
+                .push_bind(asset.access_type.expect("filtered above"))
+                .push_bind(Json(&asset.columns))
+                .push_bind(asset.created_at.unwrap_or_else(chrono::Utc::now));
+        });
+        query_builder.push(
+            " RETURNING id, workspace_id, asset_kind::text AS asset_kind, asset_path, partition_key) \
+             SELECT pg_notify('asset_event', json_build_object( \
+                'id', id, \
+                'workspace_id', workspace_id, \
+                'asset_kind', asset_kind, \
+                'asset_path', asset_path, \
+                'partition_key', partition_key \
+             )::text) FROM inserted",
+        );
+        query_builder.build().execute(executor).await?;
+    }
     Ok(())
 }
 
diff --git a/backend/windmill-common/src/trigger_annotations.rs b/backend/windmill-common/src/trigger_annotations.rs
new file mode 100644
index 0000000000..a3ffba6afa
--- /dev/null
+++ b/backend/windmill-common/src/trigger_annotations.rs
@@ -0,0 +1,489 @@
+//! Parse `#trigger: asset ...` annotations out of script source.
+//!
+//! This is the source-of-truth for implicit asset triggers. A script's
+//! annotations are re-parsed on every save and projected into rows in the
+//! `asset_trigger` table.
+
+use std::str::FromStr;
+
+use serde::{Deserialize, Serialize};
+
+use crate::{assets::AssetKind, scripts::ScriptLang};
+
+#[derive(thiserror::Error, Debug, PartialEq, Eq)]
+pub enum TriggerAnnotationError {
+    #[error("trigger annotation is not valid: {0}")]
+    Invalid(String),
+    #[error("trigger option `{key}` has unsupported value `{value}`: {reason}")]
+    UnsupportedValue { key: String, value: String, reason: String },
+    #[error("unknown trigger option `{0}`")]
+    UnknownOption(String),
+    #[error("script has more than one `trigger: asset` annotation — v1 supports a single asset trigger per script")]
+    MultipleAssetAnnotations,
+}
+
+pub type Result<T> = std::result::Result<T, TriggerAnnotationError>;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum AssetTriggerEvent {
+    Change,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum FiresMode {
+    All,
+    Any,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum BacklogMode {
+    Coalesce,
+    Replay,
+    Skip,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "mode", rename_all = "lowercase")]
+pub enum PartitionMap {
+    Identity,
+    None,
+    All,
+    Window { back_days: i32, forward_days: i32 },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AssetTriggerAnnotation {
+    pub on: AssetTriggerEvent,
+    pub fires: FiresMode,
+    pub debounce_s: i32,
+    /// If `Some`, only these upstream paths (filtered by kinds) are subscribed
+    /// to; overrides the script's inferred read set.
+    pub only: Option<Vec<String>>,
+    pub exclude: Vec<String>,
+    pub extra: Vec<String>,
+    /// If `Some`, the subscription set is narrowed to these asset kinds.
+    pub kinds: Option<Vec<AssetKind>>,
+    pub partition_map: PartitionMap,
+    pub cancel_on_new: bool,
+    pub backlog: BacklogMode,
+}
+
+impl Default for AssetTriggerAnnotation {
+    fn default() -> Self {
+        Self {
+            on: AssetTriggerEvent::Change,
+            fires: FiresMode::All,
+            debounce_s: 30,
+            only: None,
+            exclude: Vec::new(),
+            extra: Vec::new(),
+            kinds: None,
+            partition_map: PartitionMap::Identity,
+            cancel_on_new: false,
+            backlog: BacklogMode::Coalesce,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum TriggerAnnotation {
+    Asset(AssetTriggerAnnotation),
+}
+
+/// Parse every `<comment> trigger: ...` annotation found anywhere in the file.
+///
+/// Returns `Ok(vec![])` when no trigger annotation is present — this is the
+/// common case. Errors are only returned for malformed annotations so a
+/// missing trigger never blocks a save.
+pub fn parse_trigger_annotations(code: &str, lang: ScriptLang) -> Result<Vec<TriggerAnnotation>> {
+    let prefix = lang.as_comment_lit();
+    let marker = format!("{prefix} trigger:");
+    let mut out = Vec::new();
+    let mut asset_seen = false;
+    for raw in code.lines() {
+        let line = raw.trim_end();
+        let trimmed = line.trim_start();
+        let Some(rest) = trimmed.strip_prefix(&marker) else {
+            continue;
+        };
+        let rest = rest.trim();
+        if let Some(body) = rest.strip_prefix("asset").map(|r| r.trim()) {
+            if asset_seen {
+                return Err(TriggerAnnotationError::MultipleAssetAnnotations);
+            }
+            asset_seen = true;
+            out.push(TriggerAnnotation::Asset(parse_asset_body(body)?));
+        } else {
+            return Err(TriggerAnnotationError::Invalid(format!(
+                "expected `trigger: asset ...`, got `trigger: {rest}`"
+            )));
+        }
+    }
+    Ok(out)
+}
+
+fn parse_asset_body(body: &str) -> Result<AssetTriggerAnnotation> {
+    let mut ann = AssetTriggerAnnotation::default();
+    for token in split_top_level(body)? {
+        let Some((key, value)) = token.split_once('=') else {
+            return Err(TriggerAnnotationError::Invalid(format!(
+                "expected `key=value`, got `{token}`"
+            )));
+        };
+        let key = key.trim();
+        let value = value.trim();
+        match key {
+            "on" => {
+                if value != "change" {
+                    return Err(TriggerAnnotationError::UnsupportedValue {
+                        key: key.to_string(),
+                        value: value.to_string(),
+                        reason: "only `on=change` is supported in v1".to_string(),
+                    });
+                }
+                ann.on = AssetTriggerEvent::Change;
+            }
+            "fires" => {
+                ann.fires = match value {
+                    "all" => FiresMode::All,
+                    "any" => FiresMode::Any,
+                    other => {
+                        return Err(TriggerAnnotationError::UnsupportedValue {
+                            key: key.to_string(),
+                            value: other.to_string(),
+                            reason: "expected `all` or `any`".to_string(),
+                        });
+                    }
+                };
+            }
+            "debounce" => {
+                ann.debounce_s = parse_duration_seconds(value).map_err(|e| {
+                    TriggerAnnotationError::UnsupportedValue {
+                        key: key.to_string(),
+                        value: value.to_string(),
+                        reason: e,
+                    }
+                })?;
+            }
+            "only" => ann.only = Some(parse_list(value)),
+            "exclude" => ann.exclude = parse_list(value),
+            "extra" => ann.extra = parse_list(value),
+            "kinds" => {
+                let list = parse_list(value);
+                let mut kinds = Vec::with_capacity(list.len());
+                for k in list {
+                    kinds.push(parse_asset_kind(&k).ok_or_else(|| {
+                        TriggerAnnotationError::UnsupportedValue {
+                            key: "kinds".to_string(),
+                            value: k.clone(),
+                            reason: "expected one of datatable|ducklake|s3object|resource|volume"
+                                .to_string(),
+                        }
+                    })?);
+                }
+                ann.kinds = Some(kinds);
+            }
+            "partition_map" => {
+                ann.partition_map = parse_partition_map(value).map_err(|reason| {
+                    TriggerAnnotationError::UnsupportedValue {
+                        key: key.to_string(),
+                        value: value.to_string(),
+                        reason,
+                    }
+                })?;
+            }
+            "cancel_on_new" => {
+                ann.cancel_on_new =
+                    parse_bool(value).ok_or_else(|| TriggerAnnotationError::UnsupportedValue {
+                        key: key.to_string(),
+                        value: value.to_string(),
+                        reason: "expected `true` or `false`".to_string(),
+                    })?;
+            }
+            "backlog" => {
+                ann.backlog = match value {
+                    "coalesce" => BacklogMode::Coalesce,
+                    "replay" => BacklogMode::Replay,
+                    "skip" => BacklogMode::Skip,
+                    other => {
+                        return Err(TriggerAnnotationError::UnsupportedValue {
+                            key: key.to_string(),
+                            value: other.to_string(),
+                            reason: "expected `coalesce`, `replay`, or `skip`".to_string(),
+                        });
+                    }
+                };
+            }
+            other => return Err(TriggerAnnotationError::UnknownOption(other.to_string())),
+        }
+    }
+    Ok(ann)
+}
+
+/// Split a body like `on=change fires=all only=[a,b] debounce=30s` into
+/// tokens, preserving bracketed list values as a single token.
+fn split_top_level(body: &str) -> Result<Vec<String>> {
+    let mut tokens = Vec::new();
+    let mut current = String::new();
+    let mut depth: i32 = 0;
+    for ch in body.chars() {
+        match ch {
+            '[' => {
+                depth += 1;
+                current.push(ch);
+            }
+            ']' => {
+                depth -= 1;
+                if depth < 0 {
+                    return Err(TriggerAnnotationError::Invalid(
+                        "unbalanced `]` in trigger annotation".to_string(),
+                    ));
+                }
+                current.push(ch);
+            }
+            c if c.is_whitespace() && depth == 0 => {
+                if !current.is_empty() {
+                    tokens.push(std::mem::take(&mut current));
+                }
+            }
+            _ => current.push(ch),
+        }
+    }
+    if depth != 0 {
+        return Err(TriggerAnnotationError::Invalid(
+            "unbalanced `[` in trigger annotation".to_string(),
+        ));
+    }
+    if !current.is_empty() {
+        tokens.push(current);
+    }
+    Ok(tokens)
+}
+
+fn parse_list(value: &str) -> Vec<String> {
+    let trimmed = value.trim();
+    let inner = trimmed
+        .strip_prefix('[')
+        .and_then(|s| s.strip_suffix(']'))
+        .unwrap_or(trimmed);
+    inner
+        .split(',')
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .map(str::to_string)
+        .collect()
+}
+
+fn parse_bool(value: &str) -> Option<bool> {
+    match value {
+        "true" => Some(true),
+        "false" => Some(false),
+        _ => None,
+    }
+}
+
+fn parse_asset_kind(s: &str) -> Option<AssetKind> {
+    match s {
+        "s3object" => Some(AssetKind::S3Object),
+        "resource" => Some(AssetKind::Resource),
+        "variable" => Some(AssetKind::Variable),
+        "ducklake" => Some(AssetKind::Ducklake),
+        "datatable" => Some(AssetKind::DataTable),
+        "volume" => Some(AssetKind::Volume),
+        _ => None,
+    }
+}
+
+/// Parse durations like `30s`, `5m`, `2h`, `1d`. Returns whole seconds.
+/// Bare integers are accepted as seconds for backward-compatibility with
+/// simple `debounce=30`.
+fn parse_duration_seconds(s: &str) -> std::result::Result<i32, String> {
+    let s = s.trim();
+    if s.is_empty() {
+        return Err("empty duration".to_string());
+    }
+    let (num, unit) = match s
+        .char_indices()
+        .find(|(_, c)| !c.is_ascii_digit() && *c != '-')
+    {
+        Some((i, _)) => (&s[..i], &s[i..]),
+        None => (s, "s"),
+    };
+    let n: i32 = i32::from_str(num).map_err(|_| format!("not a number: `{num}`"))?;
+    let mult = match unit {
+        "s" | "" => 1,
+        "m" => 60,
+        "h" => 60 * 60,
+        "d" => 24 * 60 * 60,
+        other => return Err(format!("unknown duration unit `{other}`")),
+    };
+    Ok(n * mult)
+}
+
+fn parse_partition_map(value: &str) -> std::result::Result<PartitionMap, String> {
+    match value {
+        "identity" => Ok(PartitionMap::Identity),
+        "none" => Ok(PartitionMap::None),
+        "all" => Ok(PartitionMap::All),
+        other if other.starts_with("window(") && other.ends_with(')') => {
+            let inside = &other["window(".len()..other.len() - 1];
+            let mut parts = inside.split(',').map(str::trim);
+            let back = parts
+                .next()
+                .ok_or_else(|| "window() needs two arguments".to_string())?;
+            let forward = parts
+                .next()
+                .ok_or_else(|| "window() needs two arguments".to_string())?;
+            if parts.next().is_some() {
+                return Err("window() takes exactly two arguments".to_string());
+            }
+            let back_days = parse_signed_days(back)?;
+            let forward_days = parse_signed_days(forward)?;
+            Ok(PartitionMap::Window { back_days, forward_days })
+        }
+        other => Err(format!(
+            "expected `identity`, `none`, `all`, or `window(-Nd,Md)`, got `{other}`"
+        )),
+    }
+}
+
+fn parse_signed_days(s: &str) -> std::result::Result<i32, String> {
+    let s = s.trim();
+    let rest = s
+        .strip_suffix('d')
+        .ok_or_else(|| format!("expected a signed day literal like `-7d` or `3d`, got `{s}`"))?;
+    i32::from_str(rest).map_err(|_| format!("not a number: `{rest}`"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_none_when_no_annotation() {
+        let out =
+            parse_trigger_annotations("let x = 1;\n// just a comment", ScriptLang::Bun).unwrap();
+        assert!(out.is_empty());
+    }
+
+    #[test]
+    fn parses_minimal_python_annotation() {
+        let src = "# trigger: asset on=change\nprint('hi')";
+        let out = parse_trigger_annotations(src, ScriptLang::Python3).unwrap();
+        let TriggerAnnotation::Asset(ann) = &out[0];
+        assert_eq!(ann.on, AssetTriggerEvent::Change);
+        assert_eq!(ann.fires, FiresMode::All);
+        assert_eq!(ann.debounce_s, 30);
+        assert_eq!(ann.cancel_on_new, false);
+        assert_eq!(ann.backlog, BacklogMode::Coalesce);
+        assert_eq!(ann.partition_map, PartitionMap::Identity);
+    }
+
+    #[test]
+    fn parses_bun_annotation_anywhere_in_file() {
+        let src = "import x from 'y';\n\n// trigger: asset on=change fires=any\nexport async function main() {}";
+        let out = parse_trigger_annotations(src, ScriptLang::Bun).unwrap();
+        assert_eq!(out.len(), 1);
+        let TriggerAnnotation::Asset(ann) = &out[0];
+        assert_eq!(ann.fires, FiresMode::Any);
+    }
+
+    #[test]
+    fn parses_sql_annotation() {
+        let src = "-- trigger: asset on=change debounce=5m\nSELECT 1;";
+        let out = parse_trigger_annotations(src, ScriptLang::Postgresql).unwrap();
+        let TriggerAnnotation::Asset(ann) = &out[0];
+        assert_eq!(ann.debounce_s, 300);
+    }
+
+    #[test]
+    fn parses_rust_annotation_with_its_bespoke_prefix() {
+        let src = "//! trigger: asset on=change backlog=skip\nfn main() {}";
+        let out = parse_trigger_annotations(src, ScriptLang::Rust).unwrap();
+        let TriggerAnnotation::Asset(ann) = &out[0];
+        assert_eq!(ann.backlog, BacklogMode::Skip);
+    }
+
+    #[test]
+    fn parses_full_option_set() {
+        let src = "# trigger: asset on=change fires=any debounce=2h only=[f/a,f/b] exclude=[f/c] extra=[f/d] kinds=[datatable,s3object] partition_map=window(-7d,0d) cancel_on_new=true backlog=replay\n";
+        let out = parse_trigger_annotations(src, ScriptLang::Python3).unwrap();
+        let TriggerAnnotation::Asset(ann) = &out[0];
+        assert_eq!(ann.fires, FiresMode::Any);
+        assert_eq!(ann.debounce_s, 2 * 60 * 60);
+        assert_eq!(ann.only, Some(vec!["f/a".to_string(), "f/b".to_string()]));
+        assert_eq!(ann.exclude, vec!["f/c".to_string()]);
+        assert_eq!(ann.extra, vec!["f/d".to_string()]);
+        assert_eq!(
+            ann.kinds,
+            Some(vec![AssetKind::DataTable, AssetKind::S3Object])
+        );
+        assert_eq!(
+            ann.partition_map,
+            PartitionMap::Window { back_days: -7, forward_days: 0 }
+        );
+        assert_eq!(ann.cancel_on_new, true);
+        assert_eq!(ann.backlog, BacklogMode::Replay);
+    }
+
+    #[test]
+    fn rejects_unsupported_on_value() {
+        let src = "# trigger: asset on=freshness\n";
+        let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
+        assert!(matches!(
+            err,
+            TriggerAnnotationError::UnsupportedValue { .. }
+        ));
+    }
+
+    #[test]
+    fn rejects_duplicate_asset_annotation() {
+        let src = "# trigger: asset on=change\n# trigger: asset on=change\n";
+        let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
+        assert_eq!(err, TriggerAnnotationError::MultipleAssetAnnotations);
+    }
+
+    #[test]
+    fn rejects_unknown_option() {
+        let src = "# trigger: asset on=change zombie=true\n";
+        let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
+        assert!(matches!(err, TriggerAnnotationError::UnknownOption(o) if o == "zombie"));
+    }
+
+    #[test]
+    fn rejects_malformed_kind() {
+        let src = "# trigger: asset on=change kinds=[s3object,foobar]\n";
+        let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
+        assert!(
+            matches!(err, TriggerAnnotationError::UnsupportedValue { key, .. } if key == "kinds")
+        );
+    }
+
+    #[test]
+    fn rejects_bad_partition_window() {
+        let src = "# trigger: asset on=change partition_map=window(abc,def)\n";
+        let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
+        assert!(
+            matches!(err, TriggerAnnotationError::UnsupportedValue { key, .. } if key == "partition_map")
+        );
+    }
+
+    #[test]
+    fn accepts_bare_integer_debounce_as_seconds() {
+        let src = "# trigger: asset on=change debounce=45\n";
+        let out = parse_trigger_annotations(src, ScriptLang::Python3).unwrap();
+        let TriggerAnnotation::Asset(ann) = &out[0];
+        assert_eq!(ann.debounce_s, 45);
+    }
+
+    #[test]
+    fn ignores_lines_with_other_comment_prefix() {
+        // `# trigger: ...` in a bun file uses `//` as prefix, so a `#` line is ignored.
+        let src = "# trigger: asset on=change\nconsole.log('hi');";
+        let out = parse_trigger_annotations(src, ScriptLang::Bun).unwrap();
+        assert!(out.is_empty());
+    }
+}
diff --git a/backend/windmill-common/src/triggers.rs b/backend/windmill-common/src/triggers.rs
index b4f8f472cc..0580dd7c84 100644
--- a/backend/windmill-common/src/triggers.rs
+++ b/backend/windmill-common/src/triggers.rs
@@ -39,7 +39,8 @@ pub async fn update_triggers_script_path(
          t3 AS (UPDATE postgres_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
          t4 AS (UPDATE mqtt_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
          t5 AS (UPDATE nats_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
-         t6 AS (UPDATE sqs_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4) \
+         t6 AS (UPDATE sqs_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
+         t7 AS (UPDATE asset_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4 AND is_implicit = false) \
          UPDATE gcp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4",
         new_path,
         old_path,
diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs
index 264dd31eac..39ce62c022 100644
--- a/backend/windmill-worker/src/worker.rs
+++ b/backend/windmill-worker/src/worker.rs
@@ -3313,6 +3313,8 @@ async fn detect_and_store_runtime_assets_from_job_args(
             access_type: None,
             created_at: None,
             columns: None,
+            partition_key: None,
+            script_hash: None,
         };
         register_runtime_asset(asset);
     }
diff --git a/cli/src/commands/sync/sync.ts b/cli/src/commands/sync/sync.ts
index 668e21b82b..68da5354f7 100644
--- a/cli/src/commands/sync/sync.ts
+++ b/cli/src/commands/sync/sync.ts
@@ -1457,6 +1457,11 @@ export async function elementsToMap(
     ) {
       continue;
     }
+    // Asset triggers are implicit: source-of-truth is `#trigger: asset` in
+    // script annotations, never editable via CLI. Skip unconditionally.
+    if (path.endsWith(".asset_trigger" + ext)) {
+      continue;
+    }
     if (!skips.includeUsers && path.endsWith(".user" + ext)) continue;
     if (!skips.includeGroups && path.endsWith(".group" + ext)) continue;
     if (!skips.includeSettings && path === "settings" + ext) continue;