Merge branch 'main' into cli-onbehaflof

2026-08-21 00:02:23 +00:00 · 2026-03-31 18:47:03 +02:00
parent 84c2817520 1a73012e07
commit ecd02fa322
87 changed files with 3005 additions and 345 deletions
@@ -1,5 +1,31 @@
 # Changelog

+## [1.665.0](https://github.com/windmill-labs/windmill/compare/v1.664.0...v1.665.0) (2026-03-26)
+
+
+### Features
+
+* add instance setting to enforce workspace prefix for HTTP routes ([#8528](https://github.com/windmill-labs/windmill/issues/8528)) ([9b3e558](https://github.com/windmill-labs/windmill/commit/9b3e558d84f15052e9c32695a467f8ef7e4ad1f5))
+* add trashbin system for soft-deleting items ([#8519](https://github.com/windmill-labs/windmill/issues/8519)) ([69ce946](https://github.com/windmill-labs/windmill/commit/69ce946241d98ea90bc7135d44ca0c87f928be88))
+* mask sensitive values in job logs ([#8520](https://github.com/windmill-labs/windmill/issues/8520)) ([0885d8c](https://github.com/windmill-labs/windmill/commit/0885d8c986f13ac210e4db3ad38febe9be391ba4))
+* move basic git sync from EE to CE with runtime user count gating ([#8493](https://github.com/windmill-labs/windmill/issues/8493)) ([79d2bd5](https://github.com/windmill-labs/windmill/commit/79d2bd51a00654162754046308d7670242120df6))
+* runner groups for shared-process multi-script dedicated workers ([#8434](https://github.com/windmill-labs/windmill/issues/8434)) ([c28314f](https://github.com/windmill-labs/windmill/commit/c28314f424ea0e04b86565ce88e6c91e0df1a0cf))
+* SCIM user deprovisioning (active:false) + instance-level user disable ([#8484](https://github.com/windmill-labs/windmill/issues/8484)) ([0bd7568](https://github.com/windmill-labs/windmill/commit/0bd756839c0261f255111d62088bdaaecb838085))
+* show groups and notes in flow status viewer ([#8535](https://github.com/windmill-labs/windmill/issues/8535)) ([167084a](https://github.com/windmill-labs/windmill/commit/167084a0ebe73384fa0d31f0b24017a47686a072))
+
+
+### Bug Fixes
+
+* auto-generate datatable SDK reference for app mode system prompt ([#8522](https://github.com/windmill-labs/windmill/issues/8522)) ([8a32322](https://github.com/windmill-labs/windmill/commit/8a32322c187ccc60ec7eafb61a9678f267a82282))
+* consider wmill.yaml environments alias in git sync ([#8532](https://github.com/windmill-labs/windmill/issues/8532)) ([b7475c7](https://github.com/windmill-labs/windmill/commit/b7475c73094a28f520f798f6cb1a0c6b4807ccb7))
+* GitHub Enterprise Server support for self-managed GitHub Apps ([#8507](https://github.com/windmill-labs/windmill/issues/8507)) ([935fb44](https://github.com/windmill-labs/windmill/commit/935fb44c848b8bf9430b5600dd3c3bedb2f89efd))
+* raw apps bundle not found during deployment error ([#8515](https://github.com/windmill-labs/windmill/issues/8515)) ([34e3115](https://github.com/windmill-labs/windmill/commit/34e3115bcbd19a8e0b6f483435586a2ab43d0a8e))
+* require admin for workspace encryption key export ([#8523](https://github.com/windmill-labs/windmill/issues/8523)) ([0317668](https://github.com/windmill-labs/windmill/commit/031766808945aefc926f0836d011c0b2a5d2243d))
+* restrict logout redirect to whitelisted domains ([#8524](https://github.com/windmill-labs/windmill/issues/8524)) ([4c8edd5](https://github.com/windmill-labs/windmill/commit/4c8edd5e944d77ed2d41c2b87171c1115c0fdcdc))
+* serve index disk storage sizes from /srch/ endpoint ([#8511](https://github.com/windmill-labs/windmill/issues/8511)) ([e3620e0](https://github.com/windmill-labs/windmill/commit/e3620e074e1bdb46b2b8d732f35a91d300589663))
+* use /apps_raw/get/ redirect URL for raw apps set as workspace default ([#8508](https://github.com/windmill-labs/windmill/issues/8508)) ([85c52e2](https://github.com/windmill-labs/windmill/commit/85c52e2cded10606cc895d0d3b717e13c69bc9b3))
+* use resource-level scope overrides during OAuth2 token refresh ([#8540](https://github.com/windmill-labs/windmill/issues/8540)) ([55ad0ff](https://github.com/windmill-labs/windmill/commit/55ad0ff5c499c33b766f47c6f32ba5d3eeb14763))
+
 ## [1.664.0](https://github.com/windmill-labs/windmill/compare/v1.663.0...v1.664.0) (2026-03-24)


@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM trashbin WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "08522e494e34f4ecae21460262bf0ed3c5a197dd744c87cb760aaf47001febbd"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM trashbin WHERE workspace_id = $1 AND id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1d995dd5a094631ae96c16d68026fdeb22714af38162e87c02b052a5b8ec2645"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM trashbin WHERE expires_at <= now()",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "3453c0b7dd3c4d2c9bc639f379901741955502c9345e82a9b7fbbf3d3c7ab517"
+}
@@ -0,0 +1,65 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, workspace_id, item_kind, item_path, item_data, deleted_by, deleted_at, expires_at\n         FROM trashbin\n         WHERE workspace_id = $1 AND id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "item_kind",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "item_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "item_data",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 5,
+        "name": "deleted_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 6,
+        "name": "deleted_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 7,
+        "name": "expires_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "446404eda9b9632c9a1384af6bf2f88594825dbaa647290a58bd63df61b531a7"
+}
@@ -0,0 +1,61 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, workspace_id, item_kind, item_path, deleted_by, deleted_at, expires_at\n             FROM trashbin\n             WHERE workspace_id = $1 AND item_kind = $2\n             ORDER BY deleted_at DESC\n             LIMIT $3 OFFSET $4",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "item_kind",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "item_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "deleted_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "deleted_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 6,
+        "name": "expires_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "51c3274a8092d80503a6b97ef3896cc3ba1957042a48ac5f9629ada25b3e78ef"
+}
@@ -15,7 +15,7 @@
      ]
    },
    "nullable": [
-      null
+      true
    ]
  },
  "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "SELECT client, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url FROM account WHERE workspace_id = $1 AND id = $2",
+  "query": "SELECT client, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url, scopes FROM account WHERE workspace_id = $1 AND id = $2",
  "describe": {
    "columns": [
      {
@@ -32,6 +32,11 @@
        "ordinal": 5,
        "name": "cc_token_url",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 6,
+        "name": "scopes",
+        "type_info": "TextArray"
      }
    ],
    "parameters": {
@@ -46,8 +51,9 @@
      false,
      true,
      true,
+      true,
      true
    ]
  },
-  "hash": "cc269052ffc1e613d7edc31f0f7bb84f6e6301ad1afb028813105a121a69fa7e"
+  "hash": "63c48fde8c0c0fff9abffc3be27e9948556b636b70b818cc31c2d50921a27366"
 }
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "INSERT INTO account (workspace_id, client, expires_at, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url, mcp_server_url) VALUES ($1, $2, now() + ($3 || ' seconds')::interval, $4, $5, $6, $7, $8, $9) RETURNING id",
+  "query": "INSERT INTO account (workspace_id, client, expires_at, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url, mcp_server_url, scopes) VALUES ($1, $2, now() + ($3 || ' seconds')::interval, $4, $5, $6, $7, $8, $9, $10) RETURNING id",
  "describe": {
    "columns": [
      {
@@ -19,12 +19,13 @@
        "Varchar",
        "Varchar",
        "Varchar",
-        "Text"
+        "Text",
+        "TextArray"
      ]
    },
    "nullable": [
      false
    ]
  },
-  "hash": "b1bd088c2e1aca3104bede7d0953369b6b17ad3ad62692ae6f2303be890e6391"
+  "hash": "870e1c3f0dc1aaa07ac74a2e37721ce352ad4fb67d36c19dce09d841e36f85dd"
 }
@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO trashbin (workspace_id, item_kind, item_path, item_data, deleted_by)\n         VALUES ($1, $2, $3, $4, $5) RETURNING id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Jsonb",
+        "Varchar"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "8b25c4252da77cd2fe1b3916b518251dbb3c6d4c095efa015823f0324ab27d7f"
+}
@@ -0,0 +1,60 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, workspace_id, item_kind, item_path, deleted_by, deleted_at, expires_at\n             FROM trashbin\n             WHERE workspace_id = $1\n             ORDER BY deleted_at DESC\n             LIMIT $2 OFFSET $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "item_kind",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "item_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "deleted_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "deleted_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 6,
+        "name": "expires_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "92fb6afe3b7041b2954340094c08e702fc1577d3fa4ff1ff2f1e089971ff5e32"
+}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM trashbin WHERE workspace_id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "bae31609123da68d16bea8e0f1c4624403b6f97e13f13f056501fe2f4efb0f06"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM script WHERE path = $1 AND workspace_id = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "c1fd495abb4353b46361ec94fd4ae8d224457171b1b73fe145d28e67f1fe03af"
+}
@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.664.0"
+version = "1.665.0"
 authors.workspace = true
 edition.workspace = true

@@ -82,7 +82,7 @@ members = [
 exclude = ["./windmill-duckdb-ffi-internal"]

 [workspace.package]
-version = "1.664.0"
+version = "1.665.0"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"

@@ -1 +1 @@
-6bb0ff0c40300dfc6049f8e027d8161a9f104d50
+6db424512b0d02f86489e85f0026581b7637d6e6
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS trashbin;
@@ -0,0 +1,16 @@
+CREATE TABLE trashbin (
+    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE,
+    item_kind VARCHAR(50) NOT NULL,
+    item_path TEXT NOT NULL,
+    item_data JSONB NOT NULL,
+    deleted_by VARCHAR(255) NOT NULL,
+    deleted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    expires_at TIMESTAMPTZ NOT NULL DEFAULT now() + INTERVAL '3 days'
+);
+
+CREATE INDEX idx_trashbin_expires_at ON trashbin(expires_at);
+CREATE INDEX idx_trashbin_workspace_kind ON trashbin(workspace_id, item_kind);
+
+GRANT ALL ON trashbin TO windmill_user;
+GRANT ALL ON trashbin TO windmill_admin;
@@ -0,0 +1 @@
+ALTER TABLE account DROP COLUMN IF EXISTS scopes;
@@ -0,0 +1 @@
+ALTER TABLE account ADD COLUMN scopes TEXT[];
@@ -1099,6 +1099,9 @@ Windmill Community Edition {GIT_VERSION}
        }

        let addr = SocketAddr::from((server_bind_address, port));
+        let listener = tokio::net::TcpListener::bind(addr)
+            .await
+            .context("binding main windmill server")?;

        let (base_internal_tx, base_internal_rx) = tokio::sync::oneshot::channel::<String>();

@@ -1232,7 +1235,7 @@ Windmill Community Edition {GIT_VERSION}
                        db.clone(),
                        index_reader,
                        log_index_reader,
-                        addr,
+                        listener,
                        server_killpill_rx,
                        base_internal_tx,
                        server_mode,
@@ -1178,6 +1178,15 @@ pub async fn delete_expired_items(db: &DB) -> () {
            tracing::error!("Error deleting custom concurrency key: {:?}", e);
        }
    }
+
+    match windmill_common::trashbin::delete_expired_trash(db).await {
+        Ok(count) => {
+            if count > 0 {
+                tracing::info!("deleted {} expired trash items", count);
+            }
+        }
+        Err(e) => tracing::error!("Error deleting expired trash items: {}", e.to_string()),
+    }
 }

 pub async fn check_expiring_tokens(db: &DB) {
@@ -33,7 +33,7 @@ workspace_key_kind: cloud

 ## Tables
 _sqlx_migrations: version(bigint), description(text), installed_on(ts), success(bool), checksum(bytes), execution_time(bigint)
-account: workspace_id(char), id(int), expires_at(ts), refresh_token(char), client(char), refresh_error(text), grant_type(char), cc_client_id(char), cc_client_secret(char), cc_token_url(char), mcp_server_url(text)
+account: workspace_id(char), id(int), expires_at(ts), refresh_token(char), client(char), refresh_error(text), grant_type(char), cc_client_id(char), cc_client_secret(char), cc_token_url(char), mcp_server_url(text), scopes(text[])
  FK: (workspace_id) -> workspace(id)
 agent_token_blacklist: token(char), expires_at(ts), blacklisted_at(ts), blacklisted_by(char)
 ai_agent_memory: workspace_id(char), conversation_id(uuid), step_id(char), messages(jsonb), created_at(ts), updated_at(ts)
@@ -151,6 +151,9 @@ script: workspace_id(char), hash(bigint), path(char), parent_hashes(bigint[]), s
 skip_workspace_diff_tally: workspace_id(char), added_at(ts)
 sqs_trigger: path(char), queue_url(char), aws_resource_path(char), message_attributes(text[]), script_path(char), is_flow(bool), workspace_id(char), edited_by(char), email(char), edited_at(ts), extra_perms(jsonb), error(text), server_id(char), last_server_ping(ts), aws_auth_resource_type(aws_auth_resource_type), error_handler_path(char), error_handler_args(jsonb), retry(jsonb), mode(trigger_mode)
  FK: (workspace_id) -> workspace(id)
+trashbin: id(bigint), workspace_id(char), item_kind(char), item_path(char), item_data(jsonb), deleted_by(char), deleted_at(ts), expires_at(ts)
+  FK: (workspace_id) -> workspace(id)
+  INDEX: idx_trashbin_expires_at (expires_at), idx_trashbin_workspace_kind (workspace_id, item_kind)
 token: token_hash(char), token_prefix(char), token(char), label(char), expiration(ts), workspace_id(char), owner(char), email(char), super_admin(bool), created_at(ts), last_used_at(ts), scopes(text[]), job(uuid)
  FK: (workspace_id) -> workspace(id)
 token_expiry_notification: token_hash(char), expiration(ts)
@@ -1657,6 +1657,38 @@ async fn delete_flow_by_path(
    }
    let mut tx = user_db.begin(&authed).await?;

+    // Capture all related data for trashbin before deleting (CASCADE will remove flow_version, flow_node)
+    let trash_flow: Option<serde_json::Value> =
+        sqlx::query_scalar("SELECT to_jsonb(t) FROM flow t WHERE path = $1 AND workspace_id = $2")
+            .bind(path)
+            .bind(&w_id)
+            .fetch_optional(&mut *tx)
+            .await?;
+
+    let trash_flow_versions: Vec<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM flow_version t WHERE path = $1 AND workspace_id = $2",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let trash_flow_nodes: Vec<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM flow_node t WHERE path = $1 AND workspace_id = $2",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let trash_drafts: Vec<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM draft t WHERE path = $1 AND workspace_id = $2 AND typ = 'flow'",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+
    sqlx::query!(
        "DELETE FROM draft WHERE path = $1 AND workspace_id = $2 AND typ = 'flow'",
        path,
@@ -1673,6 +1705,28 @@ async fn delete_flow_by_path(
    .execute(&mut *tx)
    .await?;

+    if let Some(flow_data) = trash_flow {
+        let mut trash_data = serde_json::json!({"row": flow_data});
+        if !trash_flow_versions.is_empty() {
+            trash_data["flow_versions"] = serde_json::Value::Array(trash_flow_versions);
+        }
+        if !trash_flow_nodes.is_empty() {
+            trash_data["flow_nodes"] = serde_json::Value::Array(trash_flow_nodes);
+        }
+        if !trash_drafts.is_empty() {
+            trash_data["drafts"] = serde_json::Value::Array(trash_drafts);
+        }
+        windmill_common::trashbin::move_to_trash(
+            &mut *tx,
+            &w_id,
+            "flow",
+            path,
+            trash_data,
+            &authed.username,
+        )
+        .await?;
+    }
+
    if !query.keep_captures.unwrap_or(false) {
        sqlx::query!(
            "DELETE FROM capture_config WHERE path = $1 AND workspace_id = $2 AND is_flow IS TRUE",
@@ -656,6 +656,7 @@ async fn delete_group(
    )
    .execute(&mut *tx)
    .await?;
+
    audit_log(
        &mut *tx,
        &authed,
@@ -31,6 +31,7 @@ reqwest.workspace = true
 tokio.workspace = true
 anyhow.workspace = true
 uuid.workspace = true
+futures.workspace = true
 rand.workspace = true
 rumqttc.workspace = true
 rdkafka.workspace = true
@@ -0,0 +1,465 @@
+//! Integration tests for sensitive log masking.
+//!
+//! A single comprehensive test that runs real bun scripts through real workers,
+//! covering all masking scenarios: secret variables, non-secret variables,
+//! multiple secrets, mid-string secrets, `$encrypted:` args, resources
+//! referencing secret variables, and cross-job isolation.
+//!
+//! Run with:
+//!   cargo test -p windmill-api-integration-tests --test sensitive_log_masking -- --nocapture
+//!
+//! Requires: bun runtime, live database (migrations applied by sqlx::test).
+
+use futures::StreamExt;
+use serde_json::json;
+use sqlx::{Pool, Postgres};
+use uuid::Uuid;
+use windmill_common::jobs::{JobPayload, RawCode};
+use windmill_common::scripts::ScriptLang;
+use windmill_common::worker::to_raw_value;
+use windmill_test_utils::*;
+
+fn client() -> reqwest::Client {
+    reqwest::Client::new()
+}
+
+fn authed(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
+    builder.header("Authorization", "Bearer SECRET_TOKEN")
+}
+
+/// Helper: create a variable via the API.
+async fn create_variable(port: u16, path: &str, value: &str, is_secret: bool) {
+    let base = format!("http://localhost:{port}/api/w/test-workspace/variables");
+    let resp = authed(client().post(format!("{base}/create")))
+        .json(&json!({
+            "path": path,
+            "value": value,
+            "is_secret": is_secret,
+            "description": "test variable for log masking"
+        }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        resp.status(),
+        201,
+        "failed to create variable {path}: {}",
+        resp.text().await.unwrap_or_default()
+    );
+}
+
+/// Helper: create a resource via the API.
+async fn create_resource(port: u16, path: &str, value: serde_json::Value) {
+    let base = format!("http://localhost:{port}/api/w/test-workspace/resources");
+    let resp = authed(client().post(format!("{base}/create")))
+        .json(&json!({
+            "path": path,
+            "value": value,
+            "resource_type": "object",
+            "description": "test resource for log masking"
+        }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        resp.status(),
+        201,
+        "failed to create resource {path}: {}",
+        resp.text().await.unwrap_or_default()
+    );
+}
+
+/// Helper: fetch job logs from the job_logs table.
+async fn get_job_logs(db: &Pool<Postgres>, job_id: Uuid) -> Option<String> {
+    sqlx::query_scalar!(
+        r#"SELECT logs as "logs!" FROM job_logs WHERE job_id = $1"#,
+        job_id,
+    )
+    .fetch_optional(db)
+    .await
+    .unwrap()
+}
+
+/// Helper: push a bun preview job and return its UUID.
+async fn push_bun_job(db: &Pool<Postgres>, code: String) -> Uuid {
+    RunJob::from(JobPayload::Code(RawCode {
+        hash: None,
+        content: code,
+        path: None,
+        language: ScriptLang::Bun,
+        lock: None,
+        cache_ttl: None,
+        cache_ignore_s3_path: None,
+        dedicated_worker: None,
+        concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
+            .into(),
+        debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
+        modules: None,
+    }))
+    .push(db)
+    .await
+}
+
+/// Helper: push a bun preview job with encrypted args.
+async fn push_bun_job_with_encrypted_arg(
+    db: &Pool<Postgres>,
+    code: String,
+    arg_name: &str,
+    plaintext_value: &str,
+) -> Uuid {
+    // We need to know the job_id in advance to encrypt with the right key suffix.
+    let job_id = Uuid::new_v4();
+
+    // Encrypt the value the same way the frontend does:
+    // build_crypt_with_key_suffix(db, workspace, root_job_id)
+    let mc = windmill_common::variables::build_crypt_with_key_suffix(
+        db,
+        "test-workspace",
+        &job_id.to_string(),
+    )
+    .await
+    .expect("build_crypt_with_key_suffix");
+
+    // Encrypt the JSON-serialized string value
+    let json_str = serde_json::to_string(plaintext_value).unwrap();
+    let encrypted = windmill_common::variables::encrypt(&mc, &json_str);
+    let arg_value = format!("$encrypted:{encrypted}");
+
+    let mut args = std::collections::HashMap::new();
+    args.insert(arg_name.to_string(), to_raw_value(&json!(arg_value)));
+
+    RunJob::from(JobPayload::Code(RawCode {
+        hash: None,
+        content: code,
+        path: None,
+        language: ScriptLang::Bun,
+        lock: None,
+        cache_ttl: None,
+        cache_ignore_s3_path: None,
+        dedicated_worker: None,
+        concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
+            .into(),
+        debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
+        modules: None,
+    }))
+    .job_id(job_id)
+    .arg(arg_name, json!(arg_value))
+    .push(db)
+    .await
+}
+
+/// Comprehensive test covering all sensitive log masking scenarios in a single
+/// test function to amortize server/worker startup cost.
+///
+/// Scenarios covered (each as a separate job inside the same worker):
+///   1. Secret variable fetched and logged → masked
+///   2. Non-secret variable fetched and logged → NOT masked (no false positives)
+///   3. Two different secrets fetched and logged in the same job → both masked
+///   4. Secret embedded mid-string (e.g. "token=SECRET&user=bob") → masked
+///   5. Same secret logged 3 times → all occurrences masked
+///   6. `$encrypted:` password arg logged → masked
+///   7. Resource referencing a secret variable via `$var:` → secret masked when logged
+///   8. Cross-job isolation: job A's secret does NOT leak into job B's logs
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_sensitive_log_masking(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    // === Setup: create variables and resources ===
+    let secret1 = "alpha_secret_value_9x7k2m";
+    let secret2 = "beta_secret_token_4j8n3p";
+    let plain_val = "plain_visible_value_12345";
+    let encrypted_password = "encrypted_pass_w0rd_zq5r";
+    let resource_secret = "resource_db_password_h7t2";
+
+    create_variable(port, "u/test-user/secret_alpha", secret1, true).await;
+    create_variable(port, "u/test-user/secret_beta", secret2, true).await;
+    create_variable(port, "u/test-user/plain_var", plain_val, false).await;
+    // Secret variable that will be referenced by a resource via $var:
+    create_variable(port, "u/test-user/res_secret_var", resource_secret, true).await;
+    // Resource whose "password" field references the secret variable
+    create_resource(
+        port,
+        "u/test-user/db_with_secret",
+        json!({"host": "db.example.com", "password": "$var:u/test-user/res_secret_var"}),
+    )
+    .await;
+
+    let mut completed = listen_for_completed_jobs(&db).await;
+    let db2 = db.clone();
+    in_test_worker(
+        db.clone(),
+        async move {
+            // ================================================================
+            // Scenario 1: Secret variable fetched and console.logged → masked
+            // ================================================================
+            let job1 = push_bun_job(
+                &db2,
+                r#"import * as wmill from "windmill-client";
+export async function main() {
+  const secret = await wmill.getVariable("u/test-user/secret_alpha");
+  console.log("The secret value is: " + secret);
+  return "ok";
+}"#
+                .into(),
+            )
+            .await;
+            completed.next().await;
+            let cjob1 = completed_job(job1, &db2).await;
+            assert!(cjob1.success, "scenario 1 job failed");
+            let logs1 = get_job_logs(&db2, job1).await.expect("scenario 1: no logs");
+
+            assert!(
+                !logs1.contains(secret1),
+                "scenario 1: secret value leaked in logs\nLogs:\n{logs1}"
+            );
+            assert!(
+                logs1.contains("The secret value is: alp*****"),
+                "scenario 1: expected masked output with first 3 chars\nLogs:\n{logs1}"
+            );
+            assert!(
+                logs1.contains("[windmill] secret value was masked for security reasons, use string transformations to display full value"),
+                "scenario 1: expected security notice\nLogs:\n{logs1}"
+            );
+
+            // ================================================================
+            // Scenario 2: Non-secret variable → NOT masked (no false positives)
+            // ================================================================
+            let job2 = push_bun_job(
+                &db2,
+                r#"import * as wmill from "windmill-client";
+export async function main() {
+  const val = await wmill.getVariable("u/test-user/plain_var");
+  console.log("The plain value is: " + val);
+  return "ok";
+}"#
+                .into(),
+            )
+            .await;
+            completed.next().await;
+            let cjob2 = completed_job(job2, &db2).await;
+            assert!(cjob2.success, "scenario 2 job failed");
+            let logs2 = get_job_logs(&db2, job2).await.expect("scenario 2: no logs");
+
+            assert!(
+                logs2.contains(plain_val),
+                "scenario 2: plain value should appear unmasked\nLogs:\n{logs2}"
+            );
+
+            // ================================================================
+            // Scenario 3: Two different secrets fetched in the same job → both masked
+            // ================================================================
+            let job3 = push_bun_job(
+                &db2,
+                r#"import * as wmill from "windmill-client";
+export async function main() {
+  const s1 = await wmill.getVariable("u/test-user/secret_alpha");
+  const s2 = await wmill.getVariable("u/test-user/secret_beta");
+  console.log("secret1=" + s1);
+  console.log("secret2=" + s2);
+  return "ok";
+}"#
+                .into(),
+            )
+            .await;
+            completed.next().await;
+            let cjob3 = completed_job(job3, &db2).await;
+            assert!(cjob3.success, "scenario 3 job failed");
+            let logs3 = get_job_logs(&db2, job3).await.expect("scenario 3: no logs");
+
+            assert!(
+                !logs3.contains(secret1),
+                "scenario 3: secret1 leaked\nLogs:\n{logs3}"
+            );
+            assert!(
+                !logs3.contains(secret2),
+                "scenario 3: secret2 leaked\nLogs:\n{logs3}"
+            );
+            assert!(
+                logs3.contains("secret1=alp*****"),
+                "scenario 3: secret1 not masked\nLogs:\n{logs3}"
+            );
+            assert!(
+                logs3.contains("secret2=bet*****"),
+                "scenario 3: secret2 not masked\nLogs:\n{logs3}"
+            );
+
+            // ================================================================
+            // Scenario 4: Secret embedded mid-string → masked in place
+            // ================================================================
+            let job4 = push_bun_job(
+                &db2,
+                r#"import * as wmill from "windmill-client";
+export async function main() {
+  const secret = await wmill.getVariable("u/test-user/secret_alpha");
+  console.log("token=" + secret + "&user=bob&format=json");
+  return "ok";
+}"#
+                .into(),
+            )
+            .await;
+            completed.next().await;
+            let cjob4 = completed_job(job4, &db2).await;
+            assert!(cjob4.success, "scenario 4 job failed");
+            let logs4 = get_job_logs(&db2, job4).await.expect("scenario 4: no logs");
+
+            assert!(
+                !logs4.contains(secret1),
+                "scenario 4: secret leaked mid-string\nLogs:\n{logs4}"
+            );
+            assert!(
+                logs4.contains("token=alp*****&user=bob&format=json"),
+                "scenario 4: mid-string masking failed\nLogs:\n{logs4}"
+            );
+
+            // ================================================================
+            // Scenario 5: Same secret logged 3 times → all occurrences masked
+            // ================================================================
+            let job5 = push_bun_job(
+                &db2,
+                r#"import * as wmill from "windmill-client";
+export async function main() {
+  const secret = await wmill.getVariable("u/test-user/secret_beta");
+  console.log("First: " + secret);
+  console.log("Second: " + secret);
+  console.log("Third: " + secret);
+  return "ok";
+}"#
+                .into(),
+            )
+            .await;
+            completed.next().await;
+            let cjob5 = completed_job(job5, &db2).await;
+            assert!(cjob5.success, "scenario 5 job failed");
+            let logs5 = get_job_logs(&db2, job5).await.expect("scenario 5: no logs");
+
+            assert!(
+                !logs5.contains(secret2),
+                "scenario 5: secret leaked\nLogs:\n{logs5}"
+            );
+            let mask_count = logs5.matches("bet*****").count();
+            assert!(
+                mask_count >= 3,
+                "scenario 5: expected >= 3 masked occurrences, found {mask_count}\nLogs:\n{logs5}"
+            );
+            // Security notice should appear only once even though masking happened 3 times
+            let notice_count = logs5.matches("[windmill] secret value was masked").count();
+            assert_eq!(
+                notice_count, 1,
+                "scenario 5: security notice should appear exactly once, found {notice_count}\nLogs:\n{logs5}"
+            );
+
+            // ================================================================
+            // Scenario 6: $encrypted: password arg → masked when logged
+            // ================================================================
+            let job6 = push_bun_job_with_encrypted_arg(
+                &db2,
+                r#"export async function main(password: string) {
+  console.log("password is: " + password);
+  return "ok";
+}"#
+                .into(),
+                "password",
+                encrypted_password,
+            )
+            .await;
+            completed.next().await;
+            let cjob6 = completed_job(job6, &db2).await;
+            assert!(cjob6.success, "scenario 6 job failed");
+            let logs6 = get_job_logs(&db2, job6).await.expect("scenario 6: no logs");
+
+            assert!(
+                !logs6.contains(encrypted_password),
+                "scenario 6: encrypted password leaked\nLogs:\n{logs6}"
+            );
+            assert!(
+                logs6.contains("password is: enc*****"),
+                "scenario 6: encrypted password not masked\nLogs:\n{logs6}"
+            );
+
+            // ================================================================
+            // Scenario 7: Resource with $var: referencing a secret → masked
+            // ================================================================
+            let job7 = push_bun_job(
+                &db2,
+                r#"import * as wmill from "windmill-client";
+export async function main() {
+  const res = await wmill.getResource("u/test-user/db_with_secret");
+  console.log("db password: " + res.password);
+  console.log("db host: " + res.host);
+  return "ok";
+}"#
+                .into(),
+            )
+            .await;
+            completed.next().await;
+            let cjob7 = completed_job(job7, &db2).await;
+            assert!(cjob7.success, "scenario 7 job failed");
+            let logs7 = get_job_logs(&db2, job7).await.expect("scenario 7: no logs");
+
+            assert!(
+                !logs7.contains(resource_secret),
+                "scenario 7: resource secret leaked\nLogs:\n{logs7}"
+            );
+            assert!(
+                logs7.contains("db password: res*****"),
+                "scenario 7: resource secret not masked\nLogs:\n{logs7}"
+            );
+            // Non-secret field should remain visible
+            assert!(
+                logs7.contains("db host: db.example.com"),
+                "scenario 7: non-secret resource field should be visible\nLogs:\n{logs7}"
+            );
+
+            // ================================================================
+            // Scenario 8: Cross-job isolation — job A fetches secret_alpha,
+            //              then job B logs "alpha_secret_value_9x7k2m" as a
+            //              literal string (not fetched as a secret).
+            //              Job B should NOT mask it because the secret belongs
+            //              to job A which already completed.
+            // ================================================================
+            // Job A: fetch the secret (registers it) then completes
+            let job_a = push_bun_job(
+                &db2,
+                r#"import * as wmill from "windmill-client";
+export async function main() {
+  const s = await wmill.getVariable("u/test-user/secret_alpha");
+  console.log("fetched secret");
+  return "ok";
+}"#
+                .into(),
+            )
+            .await;
+            completed.next().await;
+            let cjob_a = completed_job(job_a, &db2).await;
+            assert!(cjob_a.success, "scenario 8 job A failed");
+
+            // Job B: logs the same string as a hardcoded literal (NOT fetched as secret)
+            // Since job A already completed and unregistered, and job B never
+            // fetched the secret, it should NOT be masked.
+            let job_b_code = format!(
+                r#"export async function main() {{
+  console.log("literal value: {secret1}");
+  return "ok";
+}}"#
+            );
+            let job_b = push_bun_job(&db2, job_b_code).await;
+            completed.next().await;
+            let cjob_b = completed_job(job_b, &db2).await;
+            assert!(cjob_b.success, "scenario 8 job B failed");
+            let logs_b = get_job_logs(&db2, job_b)
+                .await
+                .expect("scenario 8 job B: no logs");
+
+            assert!(
+                logs_b.contains(secret1),
+                "scenario 8: job B should show the literal string unmasked (it never fetched a secret)\nLogs:\n{logs_b}"
+            );
+        },
+        port,
+    )
+    .await;
+
+    Ok(())
+}
@@ -963,6 +963,15 @@ async fn delete_schedule(
        )));
    }

+    // Capture row for trashbin before deleting
+    let trash_data: Option<serde_json::Value> = sqlx::query_scalar(
+        "SELECT jsonb_build_object('row', to_jsonb(t)) FROM schedule t WHERE path = $1 AND workspace_id = $2",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+
    let del = sqlx::query_scalar!(
        "DELETE FROM schedule WHERE path = $1 AND workspace_id = $2 RETURNING 1",
        path,
@@ -979,6 +988,18 @@ async fn delete_schedule(
        )));
    }

+    if let Some(data) = trash_data {
+        windmill_common::trashbin::move_to_trash(
+            &mut *tx,
+            &w_id,
+            "schedule",
+            path,
+            data,
+            &authed.username,
+        )
+        .await?;
+    }
+
    audit_log(
        &mut *tx,
        &authed,
@@ -2240,33 +2240,58 @@ async fn delete_script_by_path(
    .await?
    .unwrap_or(false);

-    let script = if !draft_only {
+    if !draft_only {
        require_admin(authed.is_admin, &authed.username)?;
-        sqlx::query_scalar!(
-            "DELETE FROM script WHERE path = $1 AND workspace_id = $2 RETURNING path",
+    }
+
+    // Capture all script versions and drafts for trashbin before deleting
+    let trash_scripts: Vec<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM script t WHERE path = $1 AND workspace_id = $2",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let trash_drafts: Vec<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM draft t WHERE path = $1 AND workspace_id = $2 AND typ = 'script'",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let script = sqlx::query_scalar!(
+        "DELETE FROM script WHERE path = $1 AND workspace_id = $2 RETURNING path",
+        path,
+        w_id
+    )
+    .fetch_one(&mut *tx)
+    .await
+    .map_err(|e| Error::internal_err(format!("deleting script by path {w_id}: {e:#}")))?;
+
+    if !trash_scripts.is_empty() {
+        let mut trash_data = serde_json::json!({"scripts": trash_scripts});
+        if !trash_drafts.is_empty() {
+            trash_data["drafts"] = serde_json::Value::Array(trash_drafts);
+        }
+        windmill_common::trashbin::move_to_trash(
+            &mut *tx,
+            &w_id,
+            "script",
            path,
-            w_id
+            trash_data,
+            &authed.username,
        )
-        .fetch_one(&db)
-        .await
-        .map_err(|e| Error::internal_err(format!("deleting script by path {w_id}: {e:#}")))?
-    } else {
-        sqlx::query_scalar!(
-            "DELETE FROM script WHERE path = $1 AND workspace_id = $2 RETURNING path",
-            path,
-            w_id
-        )
-        .fetch_one(&mut *tx)
-        .await
-        .map_err(|e| Error::internal_err(format!("deleting script by path {w_id}: {e:#}")))?
-    };
+        .await?;
+    }

    sqlx::query!(
        "DELETE FROM draft WHERE path = $1 AND workspace_id = $2 AND typ = 'script'",
        path,
        w_id
    )
-    .execute(&db)
+    .execute(&mut *tx)
    .await?;

    if !query.keep_captures.unwrap_or(false) {
@@ -2275,7 +2300,7 @@ async fn delete_script_by_path(
            path,
            w_id
        )
-        .execute(&db)
+        .execute(&mut *tx)
        .await?;

        sqlx::query!(
@@ -2283,7 +2308,7 @@ async fn delete_script_by_path(
            path,
            w_id
        )
-        .execute(&db)
+        .execute(&mut *tx)
        .await?;
    }

@@ -2370,6 +2395,30 @@ async fn delete_scripts_bulk(

    let mut tx = db.begin().await?;

+    // Capture scripts for trashbin per path before bulk delete
+    for path in &request.paths {
+        let trash_scripts: Vec<serde_json::Value> = sqlx::query_scalar(
+            "SELECT to_jsonb(t) FROM script t WHERE path = $1 AND workspace_id = $2",
+        )
+        .bind(path)
+        .bind(&w_id)
+        .fetch_all(&mut *tx)
+        .await?;
+
+        if !trash_scripts.is_empty() {
+            let trash_data = serde_json::json!({"scripts": trash_scripts});
+            windmill_common::trashbin::move_to_trash(
+                &mut *tx,
+                &w_id,
+                "script",
+                path,
+                trash_data,
+                &authed.username,
+            )
+            .await?;
+        }
+    }
+
    let mut deleted_paths = sqlx::query_scalar!(
        "DELETE FROM script WHERE workspace_id = $1 AND path = ANY($2) RETURNING path",
        w_id,
@@ -1,7 +1,7 @@
 openapi: "3.0.3"

 info:
-  version: 1.664.0
+  version: 1.665.0
  title: Windmill API

  contact:
@@ -4834,6 +4834,11 @@ paths:
                mcp_server_url:
                  type: string
                  description: "MCP server URL for MCP OAuth token refresh"
+                scopes:
+                  type: array
+                  items:
+                    type: string
+                  description: "OAuth scopes to use for token refresh. Overrides instance-level scopes."
              required:
                - refresh_token
                - expires_in
@@ -1451,6 +1451,30 @@ async fn delete_app(

    let mut tx = user_db.begin(&authed).await?;

+    // Capture all related data for trashbin before deleting (CASCADE will remove app_version, etc.)
+    let trash_app: Option<serde_json::Value> =
+        sqlx::query_scalar("SELECT to_jsonb(t) FROM app t WHERE path = $1 AND workspace_id = $2")
+            .bind(path)
+            .bind(&w_id)
+            .fetch_optional(&mut *tx)
+            .await?;
+
+    let trash_app_versions: Vec<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM app_version t WHERE app_id = (SELECT id FROM app WHERE path = $1 AND workspace_id = $2)",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let trash_drafts: Vec<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM draft t WHERE path = $1 AND workspace_id = $2 AND typ = 'app'",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_all(&mut *tx)
+    .await?;
+
    sqlx::query!(
        "DELETE FROM draft WHERE path = $1 AND workspace_id = $2 AND typ = 'app'",
        path,
@@ -1467,6 +1491,25 @@ async fn delete_app(
    .execute(&mut *tx)
    .await?;

+    if let Some(app_data) = trash_app {
+        let mut trash_data = serde_json::json!({"row": app_data});
+        if !trash_app_versions.is_empty() {
+            trash_data["app_versions"] = serde_json::Value::Array(trash_app_versions);
+        }
+        if !trash_drafts.is_empty() {
+            trash_data["drafts"] = serde_json::Value::Array(trash_drafts);
+        }
+        windmill_common::trashbin::move_to_trash(
+            &mut *tx,
+            &w_id,
+            "app",
+            path,
+            trash_data,
+            &authed.username,
+        )
+        .await?;
+    }
+
    audit_log(
        &mut *tx,
        &authed,
@@ -45,8 +45,8 @@ use windmill_common::global_settings::EMAIL_DOMAIN_SETTING;
 use windmill_common::worker::HUB_CACHE_DIR;

 use std::fs::DirBuilder;
+use std::sync::Arc;
 use std::time::Duration;
-use std::{net::SocketAddr, sync::Arc};
 use tokio::sync::RwLock;
 use tower::ServiceBuilder;
 use tower_cookies::CookieManagerLayer;
@@ -165,6 +165,7 @@ pub mod teams_ee;
 mod teams_oss;
 mod token;
 mod tracing_init;
+mod trash;
 pub mod triggers;
 mod users;
 #[cfg(feature = "private")]
@@ -325,7 +326,7 @@ pub async fn run_server(
    db: DB,
    job_index_reader: Option<IndexReader>,
    log_index_reader: Option<ServiceLogIndexReader>,
-    addr: SocketAddr,
+    listener: tokio::net::TcpListener,
    mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
    port_tx: tokio::sync::oneshot::Sender<String>,
    server_mode: bool,
@@ -411,6 +412,9 @@ pub async fn run_server(
                    auth_cache: auth_cache.clone(),
                    base_internal_url: _base_internal_url.clone(),
                });
+                let addr = listener
+                    .local_addr()
+                    .unwrap_or_else(|_| std::net::SocketAddr::from(([127, 0, 0, 1], 0)));
                if let Err(err) = smtp_server.start_listener_thread(addr).await {
                    tracing::error!("Error starting SMTP server: {err:#}");
                }
@@ -451,9 +455,6 @@ pub async fn run_server(
        health::start_health_check_loop(db.clone(), killpill_rx.resubscribe());
    }

-    let listener = tokio::net::TcpListener::bind(addr)
-        .await
-        .context("binding main windmill server")?;
    let port = listener.local_addr().map(|x| x.port()).unwrap_or(8000);
    let ip = listener
        .local_addr()
@@ -597,6 +598,7 @@ pub async fn run_server(
                        .nest("/resources", resources::workspaced_service())
                        .nest("/schedules", windmill_api_schedule::workspaced_service())
                        .nest("/scripts", scripts::workspaced_service())
+                        .nest("/trash", trash::workspaced_service())
                        .nest(
                            "/users",
                            users::workspaced_service().layer(Extension(argon2.clone())),
@@ -0,0 +1,511 @@
+use axum::{
+    extract::{Extension, Json, Path, Query},
+    routing::{delete, get, post},
+    Router,
+};
+use serde::Deserialize;
+use windmill_audit::{audit_oss::audit_log, ActionKind};
+use windmill_common::{
+    db::UserDB,
+    error::{Error, Result},
+    trashbin::{self, TrashItem, TrashItemWithData},
+    utils::require_admin,
+};
+
+use crate::db::{ApiAuthed, DB};
+
+pub fn workspaced_service() -> Router {
+    Router::new()
+        .route("/list", get(list_trash))
+        .route("/get/:id", get(get_trash_item))
+        .route("/restore/:id", post(restore_trash_item))
+        .route("/delete/:id", delete(permanently_delete_item))
+        .route("/empty", post(empty_trash))
+}
+
+#[derive(Deserialize)]
+struct ListTrashQuery {
+    item_kind: Option<String>,
+    page: Option<i64>,
+    per_page: Option<i64>,
+}
+
+async fn list_trash(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    Query(query): Query<ListTrashQuery>,
+) -> Result<Json<Vec<TrashItem>>> {
+    require_admin(authed.is_admin, &authed.username)?;
+    let items = trashbin::list_trash(
+        &db,
+        &w_id,
+        query.item_kind.as_deref(),
+        query.page,
+        query.per_page,
+    )
+    .await?;
+    Ok(Json(items))
+}
+
+async fn get_trash_item(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path((w_id, id)): Path<(String, i64)>,
+) -> Result<Json<TrashItemWithData>> {
+    require_admin(authed.is_admin, &authed.username)?;
+    let item = trashbin::get_trash_item(&db, &w_id, id).await?;
+    Ok(Json(item))
+}
+
+async fn restore_trash_item(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, id)): Path<(String, i64)>,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+    let item = trashbin::get_trash_item(&db, &w_id, id).await?;
+    let mut tx = user_db.begin(&authed).await?;
+
+    match item.item_kind.as_str() {
+        "script" => restore_script(&mut tx, &item).await?,
+        "flow" => restore_flow(&mut tx, &item).await?,
+        "app" => restore_app(&mut tx, &item).await?,
+        "schedule" => restore_schedule(&mut tx, &item).await?,
+        "variable" => restore_variable(&mut tx, &item).await?,
+        "resource" => restore_resource(&mut tx, &item).await?,
+        kind if kind.ends_with("_trigger") => restore_trigger(&mut tx, &item).await?,
+        _ => {
+            return Err(Error::BadRequest(format!(
+                "Unknown item kind: {}",
+                item.item_kind
+            )))
+        }
+    }
+
+    sqlx::query!("DELETE FROM trashbin WHERE id = $1", item.id)
+        .execute(&mut *tx)
+        .await?;
+
+    audit_log(
+        &mut *tx,
+        &authed,
+        "trash.restore",
+        ActionKind::Create,
+        &w_id,
+        Some(&item.item_path),
+        None,
+    )
+    .await?;
+
+    tx.commit().await?;
+    Ok(format!("{} '{}' restored", item.item_kind, item.item_path))
+}
+
+async fn permanently_delete_item(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, id)): Path<(String, i64)>,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+    let item = trashbin::get_trash_item(&db, &w_id, id).await?;
+    let mut tx = user_db.begin(&authed).await?;
+    trashbin::permanently_delete_item(&mut *tx, &w_id, id).await?;
+    audit_log(
+        &mut *tx,
+        &authed,
+        "trashbin.permanently_delete",
+        ActionKind::Delete,
+        &w_id,
+        Some(&item.item_path),
+        None,
+    )
+    .await?;
+    tx.commit().await?;
+    Ok("permanently deleted".to_string())
+}
+
+async fn empty_trash(
+    authed: ApiAuthed,
+    Extension(_db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Path(w_id): Path<String>,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+    let mut tx = user_db.begin(&authed).await?;
+    let count = trashbin::empty_trash(&mut *tx, &w_id).await?;
+    audit_log(
+        &mut *tx,
+        &authed,
+        "trashbin.empty",
+        ActionKind::Delete,
+        &w_id,
+        None,
+        None,
+    )
+    .await?;
+    tx.commit().await?;
+    Ok(format!("{} items permanently deleted", count))
+}
+
+// --- Restore functions per item kind ---
+
+async fn restore_script(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> {
+    let data = &item.item_data;
+
+    // Check for path conflict
+    let exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM script WHERE path = $1 AND workspace_id = $2)",
+        &item.item_path,
+        &item.workspace_id,
+    )
+    .fetch_one(&mut *tx)
+    .await?
+    .unwrap_or(false);
+
+    if exists {
+        return Err(Error::BadRequest(format!(
+            "A script already exists at path '{}'",
+            item.item_path
+        )));
+    }
+
+    // Scripts are stored as an array (all versions for the path)
+    let scripts = data
+        .get("scripts")
+        .and_then(|v| v.as_array())
+        .ok_or_else(|| Error::internal_err("Invalid trash data for script"))?;
+
+    for script in scripts {
+        sqlx::query("INSERT INTO script SELECT * FROM jsonb_populate_record(null::script, $1)")
+            .bind(script)
+            .execute(&mut *tx)
+            .await
+            .map_err(|e| Error::internal_err(format!("restoring script: {e:#}")))?;
+    }
+
+    // Restore drafts if present
+    if let Some(drafts) = data.get("drafts").and_then(|v| v.as_array()) {
+        for draft in drafts {
+            sqlx::query(
+                "INSERT INTO draft SELECT * FROM jsonb_populate_record(null::draft, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(draft)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn restore_flow(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> {
+    let data = &item.item_data;
+
+    let exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM flow WHERE path = $1 AND workspace_id = $2)",
+        &item.item_path,
+        &item.workspace_id,
+    )
+    .fetch_one(&mut *tx)
+    .await?
+    .unwrap_or(false);
+
+    if exists {
+        return Err(Error::BadRequest(format!(
+            "A flow already exists at path '{}'",
+            item.item_path
+        )));
+    }
+
+    let row = data
+        .get("row")
+        .ok_or_else(|| Error::internal_err("Invalid trash data for flow"))?;
+
+    sqlx::query("INSERT INTO flow SELECT * FROM jsonb_populate_record(null::flow, $1)")
+        .bind(row)
+        .execute(&mut *tx)
+        .await?;
+
+    // Restore flow_versions
+    if let Some(versions) = data.get("flow_versions").and_then(|v| v.as_array()) {
+        for version in versions {
+            sqlx::query(
+                "INSERT INTO flow_version SELECT * FROM jsonb_populate_record(null::flow_version, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(version)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    // Restore flow_nodes
+    if let Some(nodes) = data.get("flow_nodes").and_then(|v| v.as_array()) {
+        for node in nodes {
+            sqlx::query(
+                "INSERT INTO flow_node SELECT * FROM jsonb_populate_record(null::flow_node, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(node)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    // Restore drafts
+    if let Some(drafts) = data.get("drafts").and_then(|v| v.as_array()) {
+        for draft in drafts {
+            sqlx::query(
+                "INSERT INTO draft SELECT * FROM jsonb_populate_record(null::draft, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(draft)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn restore_app(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> {
+    let data = &item.item_data;
+
+    let exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM app WHERE path = $1 AND workspace_id = $2)",
+        &item.item_path,
+        &item.workspace_id,
+    )
+    .fetch_one(&mut *tx)
+    .await?
+    .unwrap_or(false);
+
+    if exists {
+        return Err(Error::BadRequest(format!(
+            "An app already exists at path '{}'",
+            item.item_path
+        )));
+    }
+
+    let row = data
+        .get("row")
+        .ok_or_else(|| Error::internal_err("Invalid trash data for app"))?;
+
+    // Insert app first (app_version has FK to app.id)
+    sqlx::query("INSERT INTO app SELECT * FROM jsonb_populate_record(null::app, $1)")
+        .bind(row)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| Error::internal_err(format!("restoring app row: {e:#}")))?;
+
+    // Then restore app_versions
+    if let Some(versions) = data.get("app_versions").and_then(|v| v.as_array()) {
+        for version in versions {
+            sqlx::query(
+                "INSERT INTO app_version SELECT * FROM jsonb_populate_record(null::app_version, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(version)
+            .execute(&mut *tx)
+            .await
+            .map_err(|e| Error::internal_err(format!("restoring app_version: {e:#}")))?;
+        }
+    }
+
+    // Restore drafts
+    if let Some(drafts) = data.get("drafts").and_then(|v| v.as_array()) {
+        for draft in drafts {
+            sqlx::query(
+                "INSERT INTO draft SELECT * FROM jsonb_populate_record(null::draft, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(draft)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn restore_schedule(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> {
+    let data = &item.item_data;
+
+    let exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2)",
+        &item.item_path,
+        &item.workspace_id,
+    )
+    .fetch_one(&mut *tx)
+    .await?
+    .unwrap_or(false);
+
+    if exists {
+        return Err(Error::BadRequest(format!(
+            "A schedule already exists at path '{}'",
+            item.item_path
+        )));
+    }
+
+    let row = data
+        .get("row")
+        .ok_or_else(|| Error::internal_err("Invalid trash data for schedule"))?;
+
+    sqlx::query("INSERT INTO schedule SELECT * FROM jsonb_populate_record(null::schedule, $1)")
+        .bind(row)
+        .execute(&mut *tx)
+        .await?;
+
+    Ok(())
+}
+
+async fn restore_variable(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> {
+    let data = &item.item_data;
+
+    let exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM variable WHERE path = $1 AND workspace_id = $2)",
+        &item.item_path,
+        &item.workspace_id,
+    )
+    .fetch_one(&mut *tx)
+    .await?
+    .unwrap_or(false);
+
+    if exists {
+        return Err(Error::BadRequest(format!(
+            "A variable already exists at path '{}'",
+            item.item_path
+        )));
+    }
+
+    let row = data
+        .get("row")
+        .ok_or_else(|| Error::internal_err("Invalid trash data for variable"))?;
+
+    sqlx::query("INSERT INTO variable SELECT * FROM jsonb_populate_record(null::variable, $1)")
+        .bind(row)
+        .execute(&mut *tx)
+        .await?;
+
+    // Restore linked resource if present
+    if let Some(linked_resource) = data.get("linked_resource") {
+        if !linked_resource.is_null() {
+            sqlx::query(
+                "INSERT INTO resource SELECT * FROM jsonb_populate_record(null::resource, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(linked_resource)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn restore_resource(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> {
+    let data = &item.item_data;
+
+    let exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM resource WHERE path = $1 AND workspace_id = $2)",
+        &item.item_path,
+        &item.workspace_id,
+    )
+    .fetch_one(&mut *tx)
+    .await?
+    .unwrap_or(false);
+
+    if exists {
+        return Err(Error::BadRequest(format!(
+            "A resource already exists at path '{}'",
+            item.item_path
+        )));
+    }
+
+    let row = data
+        .get("row")
+        .ok_or_else(|| Error::internal_err("Invalid trash data for resource"))?;
+
+    sqlx::query("INSERT INTO resource SELECT * FROM jsonb_populate_record(null::resource, $1)")
+        .bind(row)
+        .execute(&mut *tx)
+        .await?;
+
+    // Restore linked variables if present
+    if let Some(linked_vars) = data.get("linked_variables").and_then(|v| v.as_array()) {
+        for var in linked_vars {
+            sqlx::query(
+                "INSERT INTO variable SELECT * FROM jsonb_populate_record(null::variable, $1)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(var)
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn restore_trigger(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> {
+    let data = &item.item_data;
+
+    let table_name = data
+        .get("table_name")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| Error::internal_err("Invalid trash data for trigger: missing table_name"))?;
+
+    // Validate table name to prevent SQL injection
+    let valid_tables = [
+        "http_trigger",
+        "websocket_trigger",
+        "kafka_trigger",
+        "nats_trigger",
+        "postgres_trigger",
+        "mqtt_trigger",
+        "sqs_trigger",
+        "gcp_trigger",
+        "email_trigger",
+    ];
+
+    if !valid_tables.contains(&table_name) {
+        return Err(Error::BadRequest(format!(
+            "Invalid trigger table: {}",
+            table_name
+        )));
+    }
+
+    let exists: bool = sqlx::query_scalar(&format!(
+        "SELECT EXISTS(SELECT 1 FROM {} WHERE path = $1 AND workspace_id = $2)",
+        table_name
+    ))
+    .bind(&item.item_path)
+    .bind(&item.workspace_id)
+    .fetch_one(&mut *tx)
+    .await?;
+
+    if exists {
+        return Err(Error::BadRequest(format!(
+            "A trigger already exists at path '{}'",
+            item.item_path
+        )));
+    }
+
+    let row = data
+        .get("row")
+        .ok_or_else(|| Error::internal_err("Invalid trash data for trigger"))?;
+
+    sqlx::query(&format!(
+        "INSERT INTO {} SELECT * FROM jsonb_populate_record(null::{}, $1)",
+        table_name, table_name
+    ))
+    .bind(row)
+    .execute(&mut *tx)
+    .await?;
+
+    Ok(())
+}
@@ -56,6 +56,7 @@ tokio-util.workspace = true
 datafusion = { workspace = true, optional = true}
 reqwest = { workspace = true }
 tracing-subscriber = { workspace = true }
+aho-corasick = "1"
 lazy_static.workspace = true
 tracing-appender.workspace = true
 gethostname.workspace = true
@@ -11,8 +11,9 @@ use aws_config::BehaviorVersion;
 use aws_credential_types::provider::token::ProvideToken;
 use aws_credential_types::provider::ProvideCredentials;
 use aws_sdk_bedrockruntime::types::{
-    ContentBlock, ConversationRole, ConverseStreamOutput, ImageBlock, ImageFormat, ImageSource,
-    InferenceConfiguration, Message, SystemContentBlock, Tool, ToolInputSchema, ToolSpecification,
+    ContentBlock, ConversationRole, ConverseStreamOutput, DocumentBlock, DocumentFormat,
+    DocumentSource, ImageBlock, ImageFormat, ImageSource, InferenceConfiguration, Message,
+    SystemContentBlock, Tool, ToolInputSchema, ToolSpecification,
 };
 use aws_sdk_bedrockruntime::Client as BedrockRuntimeClient;
 use serde::{Deserialize, Serialize};
@@ -356,13 +357,12 @@ pub fn content_to_text(content: &OpenAIContent) -> String {
    }
 }

-/// Parse image data URL and extract format and base64 data
-fn parse_image_data_url(url: &str) -> Result<(ImageFormat, Vec<u8>), Error> {
+/// Parse a data URL and extract MIME type and decoded bytes.
+fn parse_data_url_bytes(url: &str) -> Result<(String, Vec<u8>), Error> {
    if !url.starts_with("data:") {
-        return Err(Error::internal_err("Image URL must be a data URL"));
+        return Err(Error::internal_err("URL must be a data URL"));
    }

-    // Parse data:image/png;base64,<data>
    let base64_start = url
        .find("base64,")
        .ok_or_else(|| Error::internal_err("Invalid data URL format"))?;
@@ -372,30 +372,51 @@ fn parse_image_data_url(url: &str) -> Result<(ImageFormat, Vec<u8>), Error> {
        .split(';')
        .next()
        .and_then(|s| s.strip_prefix("data:"))
-        .unwrap_or("image/png");
+        .unwrap_or("application/octet-stream");
+
+    let bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, base64_data)
+        .map_err(|e| Error::internal_err(format!("Failed to decode base64 data: {}", e)))?;
+
+    Ok((mime_type.to_string(), bytes))
+}
+
+/// Parse an image data URL and extract ImageFormat and decoded bytes.
+fn parse_image_data_url(url: &str) -> Result<(ImageFormat, Vec<u8>), Error> {
+    let (mime_type, bytes) = parse_data_url_bytes(url)?;

-    // Extract format from MIME type (e.g., "image/png" -> "png")
    let format_str = mime_type
        .rsplit_once('/')
        .map(|(_, format)| format)
        .unwrap_or("png");

-    // Map to ImageFormat enum
    let format = match format_str {
        "png" => ImageFormat::Png,
        "jpeg" | "jpg" => ImageFormat::Jpeg,
        "gif" => ImageFormat::Gif,
        "webp" => ImageFormat::Webp,
-        _ => ImageFormat::Png, // Default to PNG
+        _ => ImageFormat::Png,
    };

-    // Decode base64
-    let bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, base64_data)
-        .map_err(|e| Error::internal_err(format!("Failed to decode base64 image: {}", e)))?;
-
    Ok((format, bytes))
 }

+/// Map a MIME type to a Bedrock DocumentFormat.
+fn mime_to_document_format(mime_type: &str) -> DocumentFormat {
+    match mime_type {
+        "application/pdf" => DocumentFormat::Pdf,
+        "text/csv" => DocumentFormat::Csv,
+        "text/html" => DocumentFormat::Html,
+        "text/plain" => DocumentFormat::Txt,
+        "application/vnd.openxmlformats-officedocument.wordprocessingml.document" => {
+            DocumentFormat::Docx
+        }
+        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => {
+            DocumentFormat::Xlsx
+        }
+        _ => DocumentFormat::Pdf,
+    }
+}
+
 /// Convert a ContentPart to Bedrock ContentBlock
 fn content_part_to_block(part: &ContentPart) -> Result<Option<ContentBlock>, Error> {
    match part {
@@ -418,8 +439,21 @@ fn content_part_to_block(part: &ContentPart) -> Result<Option<ContentBlock>, Err

            Ok(Some(ContentBlock::Image(image_block)))
        }
+        ContentPart::File { file } => {
+            let (mime_type, bytes) = parse_data_url_bytes(&file.file_data)?;
+            let doc_source = DocumentSource::Bytes(bytes.into());
+            let doc_block = DocumentBlock::builder()
+                .format(mime_to_document_format(&mime_type))
+                .name(file.filename.replace('.', "_"))
+                .source(doc_source)
+                .build()
+                .map_err(|e| {
+                    Error::internal_err(format!("Failed to build document block: {}", e))
+                })?;
+            Ok(Some(ContentBlock::Document(doc_block)))
+        }
        ContentPart::S3Object { .. } => {
-            // S3Objects should be converted to ImageUrl before calling this function
+            // S3Objects should be converted before calling this function
            Ok(None)
        }
    }
@@ -354,6 +354,13 @@ pub fn convert_content_to_gemini_parts(content: &OpenAIContent) -> Vec<GeminiPar
                        inline_data: GeminiInlineData { mime_type, data },
                    })
                }
+                ContentPart::File { file } => {
+                    parse_data_url(&file.file_data).map(|(mime_type, data)| {
+                        GeminiPart::InlineData {
+                            inline_data: GeminiInlineData { mime_type, data },
+                        }
+                    })
+                }
                // S3Objects are handled by the worker
                _ => None,
            })
@@ -33,6 +33,11 @@ pub enum ContentPart {
    ImageUrl {
        image_url: ImageUrlData,
    },
+    /// File content block for OpenAI Chat Completions format (PDFs, etc.)
+    #[serde(rename = "file")]
+    File {
+        file: FileData,
+    },
    #[serde(rename = "s3_object")]
    S3Object {
        s3_object: S3Object,
@@ -44,6 +49,25 @@ pub struct ImageUrlData {
    pub url: String, // data:image/png;base64,... or https://...
 }

+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct FileData {
+    pub filename: String,
+    pub file_data: String, // data:application/pdf;base64,...
+}
+
+/// Check if a MIME type represents a document (as opposed to an image).
+pub fn is_document_mime(mime_type: &str) -> bool {
+    matches!(
+        mime_type,
+        "application/pdf"
+            | "text/csv"
+            | "text/html"
+            | "text/plain"
+            | "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
+            | "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+    )
+}
+
 #[derive(Serialize, Deserialize, Clone, Debug)]
 #[serde(untagged)]
 pub enum OpenAIContent {
@@ -86,6 +86,7 @@ pub mod schedule;
 pub mod schema;
 pub mod scripts;
 pub mod secret_backend;
+pub mod sensitive_log_masks;
 pub mod server;
 pub mod ssrf;
 #[cfg(feature = "private")]
@@ -96,6 +97,7 @@ pub mod stream;
 pub mod teams_ee;
 pub mod teams_oss;
 pub mod tracing_init;
+pub mod trashbin;
 pub mod triggers;
 pub mod usernames;
 pub mod users;
@@ -0,0 +1,159 @@
+//! In-memory store for masking sensitive values (secrets, password args) in job logs.
+//!
+//! Workers run an embedded server in the same process, so we use global state to track:
+//! - Which jobs are currently running
+//! - Which secret values each job should mask in its stdout
+//!
+//! When a secret is fetched via `get_value_internal` (embedded server handler), we don't know
+//! which job triggered the request (auth is user-based, not job-based), so we register the
+//! secret for ALL currently running jobs on this worker process.
+
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::sync::RwLock;
+use uuid::Uuid;
+
+/// Minimum length for a secret to be registered for masking.
+/// Short strings (e.g. "true", "1234") would cause too many false positives.
+const MIN_SECRET_LENGTH: usize = 8;
+
+const MASKED_NOTICE: &str =
+    "[windmill] secret value was masked for security reasons, use string transformations to display full value";
+
+lazy_static::lazy_static! {
+    /// Map of job_id -> set of secret values that should be masked in that job's logs.
+    static ref SENSITIVE_MASKS: RwLock<HashMap<Uuid, HashSet<String>>> =
+        RwLock::new(HashMap::new());
+
+    /// Set of currently running job IDs on this worker process.
+    static ref RUNNING_JOBS: RwLock<HashSet<Uuid>> =
+        RwLock::new(HashSet::new());
+
+}
+
+/// A lock-free snapshot of secrets for a job, taken once per log batch.
+/// Uses Aho-Corasick for O(m) multi-pattern matching in a single pass,
+/// regardless of the number of secrets registered.
+pub struct MaskSnapshot {
+    /// Aho-Corasick automaton for fast matching.
+    ac: aho_corasick::AhoCorasick,
+    /// Replacement strings, indexed to match the automaton's pattern order.
+    replacements: Vec<String>,
+    /// Whether the security notice has already been appended for this snapshot.
+    /// Tracked locally to avoid a global write lock on every masked line.
+    notice_shown: std::cell::Cell<bool>,
+}
+
+impl MaskSnapshot {
+    /// Mask all secrets in `text`. Returns `Cow::Borrowed` when no match (zero allocation).
+    /// The Aho-Corasick scan is O(text_len) regardless of how many secrets are registered.
+    pub fn mask<'a>(&self, text: &'a str) -> Cow<'a, str> {
+        if text.is_empty() {
+            return Cow::Borrowed(text);
+        }
+
+        // Single-pass check + replace using the pre-built automaton
+        if !self.ac.is_match(text) {
+            return Cow::Borrowed(text);
+        }
+
+        let mut result = self.ac.replace_all(text, &self.replacements);
+
+        // Append the notice only once per snapshot (i.e. per batch)
+        if !self.notice_shown.get() {
+            self.notice_shown.set(true);
+            result.push('\n');
+            result.push_str(MASKED_NOTICE);
+        }
+
+        Cow::Owned(result)
+    }
+}
+
+/// Take a snapshot of the current secrets for a job. Returns `None` if no secrets
+/// are registered (the caller can then skip masking entirely for the whole batch).
+///
+/// Call this once per log batch in `write_lines`, not per line.
+pub fn snapshot(job_id: &Uuid) -> Option<MaskSnapshot> {
+    let masks = SENSITIVE_MASKS.read().unwrap_or_else(|e| e.into_inner());
+    let secrets = masks.get(job_id)?;
+    if secrets.is_empty() {
+        return None;
+    }
+
+    // Sort longest-first so longer secrets are matched before shorter substrings
+    let mut sorted: Vec<&String> = secrets.iter().collect();
+    sorted.sort_by(|a, b| b.len().cmp(&a.len()));
+
+    let replacements: Vec<String> = sorted
+        .iter()
+        .map(|s| {
+            let prefix: String = s.chars().take(3).collect();
+            format!("{}*****", prefix)
+        })
+        .collect();
+
+    let ac = aho_corasick::AhoCorasickBuilder::new()
+        .match_kind(aho_corasick::MatchKind::LeftmostLongest)
+        .build(sorted.iter().map(|s| s.as_str()))
+        .expect("failed to build aho-corasick automaton");
+
+    Some(MaskSnapshot { ac, replacements, notice_shown: std::cell::Cell::new(false) })
+}
+
+/// Register a job as currently running. Call this before `handle_queued_job`.
+pub fn register_running_job(job_id: Uuid) {
+    {
+        let mut jobs = RUNNING_JOBS.write().unwrap_or_else(|e| e.into_inner());
+        jobs.insert(job_id);
+    }
+    {
+        let mut masks = SENSITIVE_MASKS.write().unwrap_or_else(|e| e.into_inner());
+        masks.entry(job_id).or_default();
+    }
+}
+
+/// Unregister a job when it completes. Removes both the running job entry and its mask set.
+pub fn unregister_running_job(job_id: Uuid) {
+    {
+        let mut jobs = RUNNING_JOBS.write().unwrap_or_else(|e| e.into_inner());
+        jobs.remove(&job_id);
+    }
+    {
+        let mut masks = SENSITIVE_MASKS.write().unwrap_or_else(|e| e.into_inner());
+        masks.remove(&job_id);
+    }
+}
+
+/// Register a secret value for ALL currently running jobs.
+/// Used when a secret is fetched via the embedded server (we don't know which job triggered it).
+pub fn register_secret_for_all_running_jobs(secret: &str) {
+    if secret.len() < MIN_SECRET_LENGTH {
+        return;
+    }
+    let jobs = RUNNING_JOBS.read().unwrap_or_else(|e| e.into_inner());
+    if jobs.is_empty() {
+        return;
+    }
+    let job_ids: Vec<Uuid> = jobs.iter().copied().collect();
+    drop(jobs);
+
+    let mut masks = SENSITIVE_MASKS.write().unwrap_or_else(|e| e.into_inner());
+    for job_id in job_ids {
+        if let Some(set) = masks.get_mut(&job_id) {
+            set.insert(secret.to_string());
+        }
+    }
+}
+
+/// Register a secret value for a specific job.
+/// Used for `$encrypted:` args where we know the job ID.
+pub fn register_secret_for_job(job_id: Uuid, secret: &str) {
+    if secret.len() < MIN_SECRET_LENGTH {
+        return;
+    }
+    let mut masks = SENSITIVE_MASKS.write().unwrap_or_else(|e| e.into_inner());
+    if let Some(set) = masks.get_mut(&job_id) {
+        set.insert(secret.to_string());
+    }
+}
@@ -0,0 +1,155 @@
+use serde::Serialize;
+use sqlx::PgConnection;
+
+use crate::error::Result;
+
+#[derive(Serialize, sqlx::FromRow)]
+pub struct TrashItem {
+    pub id: i64,
+    pub workspace_id: String,
+    pub item_kind: String,
+    pub item_path: String,
+    pub deleted_by: String,
+    pub deleted_at: chrono::DateTime<chrono::Utc>,
+    pub expires_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Serialize, sqlx::FromRow)]
+pub struct TrashItemWithData {
+    pub id: i64,
+    pub workspace_id: String,
+    pub item_kind: String,
+    pub item_path: String,
+    pub item_data: serde_json::Value,
+    pub deleted_by: String,
+    pub deleted_at: chrono::DateTime<chrono::Utc>,
+    pub expires_at: chrono::DateTime<chrono::Utc>,
+}
+
+pub async fn move_to_trash(
+    tx: &mut PgConnection,
+    workspace_id: &str,
+    item_kind: &str,
+    item_path: &str,
+    item_data: serde_json::Value,
+    deleted_by: &str,
+) -> Result<i64> {
+    let id = sqlx::query_scalar!(
+        "INSERT INTO trashbin (workspace_id, item_kind, item_path, item_data, deleted_by)
+         VALUES ($1, $2, $3, $4, $5) RETURNING id",
+        workspace_id,
+        item_kind,
+        item_path,
+        item_data,
+        deleted_by,
+    )
+    .fetch_one(&mut *tx)
+    .await?;
+
+    Ok(id)
+}
+
+pub async fn list_trash<'e, E: sqlx::PgExecutor<'e>>(
+    db: E,
+    workspace_id: &str,
+    kind_filter: Option<&str>,
+    page: Option<i64>,
+    per_page: Option<i64>,
+) -> Result<Vec<TrashItem>> {
+    let per_page = per_page.unwrap_or(100).min(1000);
+    let offset = page.unwrap_or(0) * per_page;
+
+    let items = if let Some(kind) = kind_filter {
+        sqlx::query_as!(
+            TrashItem,
+            "SELECT id, workspace_id, item_kind, item_path, deleted_by, deleted_at, expires_at
+             FROM trashbin
+             WHERE workspace_id = $1 AND item_kind = $2
+             ORDER BY deleted_at DESC
+             LIMIT $3 OFFSET $4",
+            workspace_id,
+            kind,
+            per_page,
+            offset,
+        )
+        .fetch_all(db)
+        .await?
+    } else {
+        sqlx::query_as!(
+            TrashItem,
+            "SELECT id, workspace_id, item_kind, item_path, deleted_by, deleted_at, expires_at
+             FROM trashbin
+             WHERE workspace_id = $1
+             ORDER BY deleted_at DESC
+             LIMIT $2 OFFSET $3",
+            workspace_id,
+            per_page,
+            offset,
+        )
+        .fetch_all(db)
+        .await?
+    };
+
+    Ok(items)
+}
+
+pub async fn get_trash_item<'e, E: sqlx::PgExecutor<'e>>(
+    db: E,
+    workspace_id: &str,
+    id: i64,
+) -> Result<TrashItemWithData> {
+    let item = sqlx::query_as!(
+        TrashItemWithData,
+        "SELECT id, workspace_id, item_kind, item_path, item_data, deleted_by, deleted_at, expires_at
+         FROM trashbin
+         WHERE workspace_id = $1 AND id = $2",
+        workspace_id,
+        id,
+    )
+    .fetch_optional(db)
+    .await?
+    .ok_or_else(|| crate::error::Error::NotFound("Trash item not found".to_string()))?;
+
+    Ok(item)
+}
+
+pub async fn permanently_delete_item<'e, E: sqlx::PgExecutor<'e>>(
+    db: E,
+    workspace_id: &str,
+    id: i64,
+) -> Result<()> {
+    let rows = sqlx::query!(
+        "DELETE FROM trashbin WHERE workspace_id = $1 AND id = $2",
+        workspace_id,
+        id,
+    )
+    .execute(db)
+    .await?
+    .rows_affected();
+
+    if rows == 0 {
+        return Err(crate::error::Error::NotFound(
+            "Trash item not found".to_string(),
+        ));
+    }
+
+    Ok(())
+}
+
+pub async fn empty_trash<'e, E: sqlx::PgExecutor<'e>>(db: E, workspace_id: &str) -> Result<i64> {
+    let rows = sqlx::query!("DELETE FROM trashbin WHERE workspace_id = $1", workspace_id,)
+        .execute(db)
+        .await?
+        .rows_affected();
+
+    Ok(rows as i64)
+}
+
+pub async fn delete_expired_trash<'e, E: sqlx::PgExecutor<'e>>(db: E) -> Result<i64> {
+    let rows = sqlx::query!("DELETE FROM trashbin WHERE expires_at <= now()")
+        .execute(db)
+        .await?
+        .rows_affected();
+
+    Ok(rows as i64)
+}
@@ -544,14 +544,22 @@ pub async fn exchange_token(
    grant_type: &str,
    oauth_client_info: Option<&ClientWithScopes>,
    http_client: &reqwest::Client,
+    scopes: Option<&[String]>,
 ) -> Result<TokenResponse, Error> {
    let token_json = match grant_type {
-        "authorization_code" => client
-            .exchange_refresh_token(&RefreshToken::from(refresh_token))
-            .with_client(http_client)
-            .execute::<serde_json::Value>()
-            .await
-            .map_err(to_anyhow)?,
+        "authorization_code" | "" => {
+            let mut request = client.exchange_refresh_token(&RefreshToken::from(refresh_token));
+            if let Some(scopes) = scopes {
+                if !scopes.is_empty() {
+                    request = request.param("scope", scopes.join(" "));
+                }
+            }
+            request
+                .with_client(http_client)
+                .execute::<serde_json::Value>()
+                .await
+                .map_err(to_anyhow)?
+        }
        "client_credentials" => {
            let mut token_request = client.exchange_client_credentials();

@@ -569,12 +577,6 @@ pub async fn exchange_token(
                .await
                .map_err(to_anyhow)?
        }
-        "" | _ if grant_type.is_empty() => client
-            .exchange_refresh_token(&RefreshToken::from(refresh_token))
-            .with_client(http_client)
-            .execute::<serde_json::Value>()
-            .await
-            .map_err(to_anyhow)?,
        _ => {
            return Err(Error::BadRequest(format!(
                "Unsupported grant type: {}",
@@ -599,6 +601,7 @@ pub struct OAuthAccountInfo {
    pub cc_client_id: Option<String>,
    pub cc_client_secret: Option<String>,
    pub cc_token_url: Option<String>,
+    pub scopes: Option<Vec<String>>,
 }

 /// Refresh an OAuth token and update the database.
@@ -615,7 +618,7 @@ pub async fn refresh_token<'c>(
 ) -> error::Result<String> {
    let account = sqlx::query_as!(
        OAuthAccountInfo,
-        "SELECT client, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url FROM account WHERE workspace_id = $1 AND id = $2",
+        "SELECT client, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url, scopes FROM account WHERE workspace_id = $1 AND id = $2",
        w_id,
        id,
    )
@@ -679,8 +682,15 @@ pub async fn refresh_token_for_account<'c>(
        oauth_client_info.client.to_owned()
    };

+    // Account-level scopes override instance-level scopes
+    let effective_scopes = account
+        .scopes
+        .as_deref()
+        .filter(|s| !s.is_empty())
+        .unwrap_or(&oauth_client_info.scopes);
+
    if account.grant_type == "client_credentials" {
-        for scope in oauth_client_info.scopes.iter() {
+        for scope in effective_scopes.iter() {
            client.add_scope(scope);
        }
    }
@@ -699,6 +709,7 @@ pub async fn refresh_token_for_account<'c>(
        &account.grant_type,
        Some(&oauth_client_info),
        http_client,
+        Some(effective_scopes),
    )
    .await;

@@ -877,6 +877,15 @@ async fn delete_resource(
    }
    let mut tx = user_db.begin(&authed).await?;

+    // Capture resource data for trashbin before deleting
+    let trash_resource: Option<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM resource t WHERE path = $1 AND workspace_id = $2",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+
    // Fetch the resource value before deleting, so we can find linked $var: references
    let resource_value: Option<Option<serde_json::Value>> =
        sqlx::query_scalar("SELECT value FROM resource WHERE path = $1 AND workspace_id = $2")
@@ -885,6 +894,32 @@ async fn delete_resource(
            .fetch_optional(&mut *tx)
            .await?;

+    // Collect all $var: paths referenced in the resource value
+    let mut linked_var_paths: Vec<String> = Vec::new();
+    if let Some(Some(ref value)) = resource_value {
+        collect_var_refs(value, &mut linked_var_paths);
+    }
+
+    // Capture linked variables for trashbin before deleting them
+    let trash_linked_vars: Vec<serde_json::Value> = if linked_var_paths.is_empty() {
+        Vec::new()
+    } else {
+        let placeholders: Vec<String> = linked_var_paths
+            .iter()
+            .enumerate()
+            .map(|(i, _)| format!("${}", i + 2))
+            .collect();
+        let query = format!(
+            "SELECT to_jsonb(t) FROM variable t WHERE workspace_id = $1 AND path IN ({})",
+            placeholders.join(", ")
+        );
+        let mut q = sqlx::query_scalar::<_, serde_json::Value>(&query).bind(&w_id);
+        for var_path in &linked_var_paths {
+            q = q.bind(var_path);
+        }
+        q.fetch_all(&mut *tx).await?
+    };
+
    let deleted_path = sqlx::query_scalar!(
        "DELETE FROM resource WHERE path = $1 AND workspace_id = $2 RETURNING path",
        path,
@@ -894,12 +929,6 @@ async fn delete_resource(
    .await?;
    not_found_if_none(deleted_path, "Resource", &path)?;

-    // Collect all $var: paths referenced in the resource value
-    let mut linked_var_paths: Vec<String> = Vec::new();
-    if let Some(Some(value)) = resource_value {
-        collect_var_refs(&value, &mut linked_var_paths);
-    }
-
    // Delete linked variables that are actually referenced in the resource value
    let deleted_linked_variables: Vec<String> = if linked_var_paths.is_empty() {
        Vec::new()
@@ -919,6 +948,23 @@ async fn delete_resource(
        }
        q.fetch_all(&mut *tx).await?
    };
+
+    if let Some(res_data) = trash_resource {
+        let mut trash_data = serde_json::json!({"row": res_data});
+        if !trash_linked_vars.is_empty() {
+            trash_data["linked_variables"] = serde_json::Value::Array(trash_linked_vars);
+        }
+        windmill_common::trashbin::move_to_trash(
+            &mut *tx,
+            &w_id,
+            "resource",
+            path,
+            trash_data,
+            &authed.username,
+        )
+        .await?;
+    }
+
    audit_log(
        &mut *tx,
        &authed,
@@ -1025,6 +1071,30 @@ async fn delete_resources_bulk(

    let mut tx = user_db.begin(&authed).await?;

+    // Capture resources for trashbin per path before bulk delete
+    for path in &request.paths {
+        let trash_resource: Option<serde_json::Value> = sqlx::query_scalar(
+            "SELECT to_jsonb(t) FROM resource t WHERE path = $1 AND workspace_id = $2",
+        )
+        .bind(path)
+        .bind(&w_id)
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        if let Some(res_data) = trash_resource {
+            let trash_data = serde_json::json!({"row": res_data});
+            windmill_common::trashbin::move_to_trash(
+                &mut *tx,
+                &w_id,
+                "resource",
+                path,
+                trash_data,
+                &authed.username,
+            )
+            .await?;
+        }
+    }
+
    let deleted_paths = sqlx::query_scalar!(
        "DELETE FROM resource WHERE path = ANY($1) AND workspace_id = $2 RETURNING path",
        &request.paths,
@@ -520,6 +520,23 @@ async fn delete_variable(

    let mut tx = user_db.begin(&authed).await?;

+    // Capture data for trashbin before deleting
+    let trash_var: Option<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM variable t WHERE path = $1 AND workspace_id = $2",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    let trash_linked_resource: Option<serde_json::Value> = sqlx::query_scalar(
+        "SELECT to_jsonb(t) FROM resource t WHERE path = $1 AND workspace_id = $2",
+    )
+    .bind(path)
+    .bind(&w_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+
    sqlx::query!(
        "DELETE FROM variable WHERE path = $1 AND workspace_id = $2",
        path,
@@ -534,6 +551,23 @@ async fn delete_variable(
    )
    .fetch_optional(&mut *tx)
    .await?;
+
+    if let Some(var_data) = trash_var {
+        let mut trash_data = serde_json::json!({"row": var_data});
+        if let Some(linked) = trash_linked_resource {
+            trash_data["linked_resource"] = linked;
+        }
+        windmill_common::trashbin::move_to_trash(
+            &mut *tx,
+            &w_id,
+            "variable",
+            path,
+            trash_data,
+            &authed.username,
+        )
+        .await?;
+    }
+
    audit_log(
        &mut *tx,
        &authed,
@@ -633,6 +667,41 @@ async fn delete_variables_bulk(

    let mut tx = user_db.begin(&authed).await?;

+    // Capture variables for trashbin per path before bulk delete
+    for path in &request.paths {
+        let trash_var: Option<serde_json::Value> = sqlx::query_scalar(
+            "SELECT to_jsonb(t) FROM variable t WHERE path = $1 AND workspace_id = $2",
+        )
+        .bind(path)
+        .bind(&w_id)
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        if let Some(var_data) = trash_var {
+            let trash_linked: Option<serde_json::Value> = sqlx::query_scalar(
+                "SELECT to_jsonb(t) FROM resource t WHERE path = $1 AND workspace_id = $2",
+            )
+            .bind(path)
+            .bind(&w_id)
+            .fetch_optional(&mut *tx)
+            .await?;
+
+            let mut trash_data = serde_json::json!({"row": var_data});
+            if let Some(linked) = trash_linked {
+                trash_data["linked_resource"] = linked;
+            }
+            windmill_common::trashbin::move_to_trash(
+                &mut *tx,
+                &w_id,
+                "variable",
+                path,
+                trash_data,
+                &authed.username,
+            )
+            .await?;
+        }
+    }
+
    let deleted_paths = sqlx::query_scalar!(
        "DELETE FROM variable WHERE path = ANY($1) AND workspace_id = $2 RETURNING path",
        &request.paths,
@@ -1046,6 +1115,10 @@ pub async fn get_value_internal<'a>(
        variable.value
    };

+    if variable.is_secret && !r.is_empty() {
+        windmill_common::sensitive_log_masks::register_secret_for_all_running_jobs(&r);
+    }
+
    // Cache the result when explicitly allowed and caching appropriate
    if allow_cache {
        cache_variable(&w_id, &path, db_with_opt_authed.email(), r.clone());
@@ -97,14 +97,13 @@ impl ApiServer {
    async fn start_inner(db: Pool<Postgres>, agent_mode: bool) -> anyhow::Result<Self> {
        let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);

-        let sock = tokio::net::TcpListener::bind("127.0.0.1:0")
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .map_err(|e| anyhow::anyhow!("failed to bind TCP listener: {}", e))?;

-        let addr = sock
+        let addr = listener
            .local_addr()
            .map_err(|e| anyhow::anyhow!("failed to get local address: {}", e))?;
-        drop(sock);
        let (port_tx, _port_rx) = tokio::sync::oneshot::channel::<String>();
        let name = next_worker_name();
        tracing::info!("starting api server for name={name}");
@@ -112,7 +111,7 @@ impl ApiServer {
            db.clone(),
            None,
            None,
-            addr,
+            listener,
            rx,
            port_tx,
            agent_mode,
@@ -610,6 +610,17 @@ async fn delete_trigger<T: TriggerCrud>(
    })?;

    let mut tx = user_db.begin(&authed).await?;
+
+    // Capture trigger data for trashbin before deleting
+    let trash_data: Option<serde_json::Value> = sqlx::query_scalar(&format!(
+        "SELECT jsonb_build_object('row', to_jsonb(t), 'table_name', '{table}') FROM {table} t WHERE path = $1 AND workspace_id = $2",
+        table = T::TABLE_NAME
+    ))
+    .bind(path)
+    .bind(&workspace_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+
    let deleted = handler
        .delete_by_path(&mut *tx, &workspace_id, path)
        .await?;
@@ -621,6 +632,19 @@ async fn delete_trigger<T: TriggerCrud>(
        )));
    }

+    if let Some(data) = trash_data {
+        let item_kind = format!("{}_trigger", T::TRIGGER_TYPE);
+        windmill_common::trashbin::move_to_trash(
+            &mut *tx,
+            &workspace_id,
+            &item_kind,
+            path,
+            data,
+            &authed.username,
+        )
+        .await?;
+    }
+
    audit_log(
        &mut *tx,
        &authed,
@@ -70,6 +70,29 @@ pub async fn download_and_encode_s3_image(
    Ok((mime_type.to_string(), base64_data))
 }

+/// Convert an S3Object to the appropriate ContentPart based on MIME type.
+pub async fn s3_object_to_content_part(
+    s3_object: &S3Object,
+    client: &AuthedClient,
+    workspace_id: &str,
+) -> Result<ContentPart, Error> {
+    let (mime_type, file_bytes) =
+        download_and_encode_s3_image(s3_object, client, workspace_id).await?;
+    let data_url = format!("data:{};base64,{}", mime_type, file_bytes);
+
+    if windmill_common::ai_types::is_document_mime(&mime_type) {
+        let filename = s3_object
+            .s3
+            .rsplit('/')
+            .next()
+            .unwrap_or("document.pdf")
+            .to_string();
+        Ok(ContentPart::File { file: FileData { filename, file_data: data_url } })
+    } else {
+        Ok(ContentPart::ImageUrl { image_url: ImageUrlData { url: data_url } })
+    }
+}
+
 /// Prepare messages for API by converting S3Objects to base64 ImageUrls
 pub async fn prepare_messages_for_api(
    messages: &[OpenAIMessage],
@@ -92,15 +115,10 @@ pub async fn prepare_messages_for_api(
                    for part in parts {
                        match part {
                            ContentPart::S3Object { s3_object } => {
-                                // Convert S3Object to base64 image URL
-                                let (mime_type, image_bytes) =
-                                    download_and_encode_s3_image(s3_object, client, workspace_id)
-                                        .await?;
-                                prepared_content.push(ContentPart::ImageUrl {
-                                    image_url: ImageUrlData {
-                                        url: format!("data:{};base64,{}", mime_type, image_bytes),
-                                    },
-                                });
+                                prepared_content.push(
+                                    s3_object_to_content_part(s3_object, client, workspace_id)
+                                        .await?,
+                                );
                            }
                            other => {
                                // Keep Text and ImageUrl as-is
@@ -92,7 +92,9 @@ pub enum AnthropicRequestContent {
        cache_control: Option<CacheControl>,
    },
    #[serde(rename = "image")]
-    Image { source: AnthropicImageSource },
+    Image { source: AnthropicBase64Source },
+    #[serde(rename = "document")]
+    Document { source: AnthropicBase64Source },
    #[serde(rename = "tool_use")]
    ToolUse { id: String, name: String, input: Box<RawValue> },
    #[serde(rename = "tool_result")]
@@ -104,9 +106,9 @@ pub enum AnthropicRequestContent {
    },
 }

-/// Image source for Anthropic API
+/// Base64 source for Anthropic API (used by both Image and Document content blocks)
 #[derive(Serialize, Debug)]
-pub struct AnthropicImageSource {
+pub struct AnthropicBase64Source {
    pub r#type: String,
    pub media_type: String,
    pub data: String,
@@ -270,10 +272,20 @@ fn convert_content_to_anthropic(content: &Option<OpenAIContent>) -> Vec<Anthropi
                        }
                    }
                    ContentPart::ImageUrl { image_url } => {
-                        // Handle base64 images
                        if let Some((media_type, data)) = parse_data_url(&image_url.url) {
                            result.push(AnthropicRequestContent::Image {
-                                source: AnthropicImageSource {
+                                source: AnthropicBase64Source {
+                                    r#type: "base64".to_string(),
+                                    media_type,
+                                    data,
+                                },
+                            });
+                        }
+                    }
+                    ContentPart::File { file } => {
+                        if let Some((media_type, data)) = parse_data_url(&file.file_data) {
+                            result.push(AnthropicRequestContent::Document {
+                                source: AnthropicBase64Source {
                                    r#type: "base64".to_string(),
                                    media_type,
                                    data,
@@ -84,13 +84,13 @@ impl GoogleAIQueryBuilder {
                );
            }

-            if let Some(images) = args.images {
-                for image in images.iter() {
-                    if !image.s3.is_empty() {
-                        let (mime_type, image_bytes) =
-                            download_and_encode_s3_image(image, client, workspace_id).await?;
+            if let Some(attachments) = args.attachments {
+                for attachment in attachments.iter() {
+                    if !attachment.s3.is_empty() {
+                        let (mime_type, file_bytes) =
+                            download_and_encode_s3_image(attachment, client, workspace_id).await?;
                        parts.push(GeminiPart::InlineData {
-                            inline_data: GeminiInlineData { mime_type, data: image_bytes },
+                            inline_data: GeminiInlineData { mime_type, data: file_bytes },
                        });
                    }
                }
@@ -4,7 +4,7 @@ use serde_json::value::RawValue;
 use windmill_common::{ai_providers::AIProvider, client::AuthedClient, error::Error};

 use crate::ai::{
-    image_handler::{download_and_encode_s3_image, prepare_messages_for_api},
+    image_handler::{prepare_messages_for_api, s3_object_to_content_part},
    query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventProcessor},
    sse::{OpenAIResponsesSSEParser, SSEParser},
    types::*,
@@ -103,6 +103,8 @@ pub enum ImageGenerationContent {
    InputText { text: String },
    #[serde(rename = "input_image")]
    InputImage { image_url: String },
+    #[serde(rename = "input_file")]
+    InputFile { filename: String, file_data: String },
 }

 /// Output content for assistant messages in Responses API
@@ -240,6 +242,12 @@ fn convert_content_to_responses_format(
                            image_url: image_url.url.clone(),
                        })
                    }
+                    ContentPart::File { file } => {
+                        Some(ImageGenerationContent::InputFile {
+                            filename: file.filename.clone(),
+                            file_data: file.file_data.clone(),
+                        })
+                    }
                    // S3 objects should have been resolved earlier, but handle gracefully
                    ContentPart::S3Object { .. } => None,
                })
@@ -421,15 +429,26 @@ impl OpenAIQueryBuilder {
        let mut content =
            vec![ImageGenerationContent::InputText { text: args.user_message.to_string() }];

-        // Add images if provided
-        if let Some(images) = args.images {
-            for image in images.iter() {
-                if !image.s3.is_empty() {
-                    let (mime_type, image_bytes) =
-                        download_and_encode_s3_image(image, client, workspace_id).await?;
-                    content.push(ImageGenerationContent::InputImage {
-                        image_url: format!("data:{};base64,{}", mime_type, image_bytes),
-                    });
+        // Add attachments (images, PDFs, etc.) if provided
+        if let Some(attachments) = args.attachments {
+            for attachment in attachments.iter() {
+                if !attachment.s3.is_empty() {
+                    let part =
+                        s3_object_to_content_part(attachment, client, workspace_id).await?;
+                    match part {
+                        ContentPart::File { file } => {
+                            content.push(ImageGenerationContent::InputFile {
+                                filename: file.filename,
+                                file_data: file.file_data,
+                            });
+                        }
+                        ContentPart::ImageUrl { image_url } => {
+                            content.push(ImageGenerationContent::InputImage {
+                                image_url: image_url.url,
+                            });
+                        }
+                        _ => {}
+                    }
                }
            }
        }
@@ -28,7 +28,7 @@ pub struct BuildRequestArgs<'a> {
    pub output_type: &'a OutputType,
    pub system_prompt: Option<&'a str>,
    pub user_message: &'a str,
-    pub images: Option<&'a [S3Object]>,
+    pub attachments: Option<&'a [S3Object]>,
    pub has_websearch: bool,
 }

@@ -26,7 +26,8 @@ use windmill_types::s3::S3Object;
 // Re-export shared types from windmill_common
 pub use windmill_common::ai_providers::AIPlatform;
 pub use windmill_common::ai_types::{
-    ContentPart, ImageUrlData, OpenAIContent, OpenAIMessage, ToolDef, ToolDefFunction, UrlCitation,
+    ContentPart, FileData, ImageUrlData, OpenAIContent, OpenAIMessage, ToolDef, ToolDefFunction,
+    UrlCitation,
 };

 /// same as OpenAIMessage but with agent_action field included in the serialization
@@ -96,7 +97,8 @@ struct AIAgentArgsRaw {
    max_completion_tokens: Option<u32>,
    output_schema: Option<OpenAPISchema>,
    output_type: Option<OutputType>,
-    user_images: Option<Vec<S3Object>>,
+    #[serde(alias = "user_images")]
+    user_attachments: Option<Vec<S3Object>>,
    streaming: Option<bool>,
    max_iterations: Option<usize>,
    memory: Option<Memory>,
@@ -116,7 +118,7 @@ pub struct AIAgentArgs {
    pub max_completion_tokens: Option<u32>,
    pub output_schema: Option<OpenAPISchema>,
    pub output_type: Option<OutputType>,
-    pub user_images: Option<Vec<S3Object>>,
+    pub user_attachments: Option<Vec<S3Object>>,
    pub streaming: Option<bool>,
    pub max_iterations: Option<usize>,
    pub memory: Option<Memory>,
@@ -148,7 +150,7 @@ impl From<AIAgentArgsRaw> for AIAgentArgs {
            max_completion_tokens: raw.max_completion_tokens,
            output_schema: raw.output_schema,
            output_type: raw.output_type,
-            user_images: raw.user_images,
+            user_attachments: raw.user_attachments,
            streaming: raw.streaming,
            max_iterations: raw.max_iterations,
            memory,
@@ -714,24 +714,45 @@ pub async fn run_agent(
        }
    };

-    // Add user message if provided and non-empty
-    if let Some(ref user_message) = args.user_message {
-        if !user_message.is_empty() {
+    // Add user message and attachments as a single user message
+    // (Bedrock requires a text block alongside document blocks in the same message)
+    {
+        let has_message = args
+            .user_message
+            .as_ref()
+            .map(|m| !m.is_empty())
+            .unwrap_or(false);
+        let has_attachments = args
+            .user_attachments
+            .as_ref()
+            .map(|a| !a.is_empty())
+            .unwrap_or(false);
+
+        if has_message && has_attachments {
+            let mut parts = vec![ContentPart::Text {
+                text: args.user_message.clone().unwrap(),
+            }];
+            for attachment in args.user_attachments.as_ref().unwrap() {
+                if !attachment.s3.is_empty() {
+                    parts.push(ContentPart::S3Object { s3_object: attachment.clone() });
+                }
+            }
            messages.push(OpenAIMessage {
                role: "user".to_string(),
-                content: Some(OpenAIContent::Text(user_message.clone())),
+                content: Some(OpenAIContent::Parts(parts)),
                ..Default::default()
            });
-        }
-    }
-
-    // Add user images if provided
-    if let Some(ref user_images) = args.user_images {
-        if !user_images.is_empty() {
+        } else if has_message {
+            messages.push(OpenAIMessage {
+                role: "user".to_string(),
+                content: Some(OpenAIContent::Text(args.user_message.clone().unwrap())),
+                ..Default::default()
+            });
+        } else if has_attachments {
            let mut parts = vec![];
-            for image in user_images.iter() {
-                if !image.s3.is_empty() {
-                    parts.push(ContentPart::S3Object { s3_object: image.clone() });
+            for attachment in args.user_attachments.as_ref().unwrap() {
+                if !attachment.s3.is_empty() {
+                    parts.push(ContentPart::S3Object { s3_object: attachment.clone() });
                }
            }
            messages.push(OpenAIMessage {
@@ -882,7 +903,7 @@ pub async fn run_agent(
                output_type,
                system_prompt: args.system_prompt.as_deref(),
                user_message: args.user_message.as_deref().unwrap_or(""),
-                images: args.user_images.as_deref(),
+                attachments: args.user_attachments.as_deref(),
                has_websearch,
            };

@@ -286,6 +286,18 @@ pub async fn transform_json_value(
                    )
                    .await?;
                    decrypt(&mc, encrypted.to_string()).and_then(|x| {
+                        // Register the raw decrypted string for log masking.
+                        // This covers both string values and their JSON representations
+                        // (numbers, objects, etc.) that could appear in logs.
+                        windmill_common::sensitive_log_masks::register_secret_for_job(job.id, &x);
+                        if let serde_json::Value::String(ref s) =
+                            serde_json::from_str::<serde_json::Value>(&x).unwrap_or_default()
+                        {
+                            // Also register the inner string value (without JSON quotes)
+                            windmill_common::sensitive_log_masks::register_secret_for_job(
+                                job.id, s,
+                            );
+                        }
                        serde_json::from_str(&x).map_err(|e| {
                            Error::internal_err(format!(
                                "Failed to decrypt '$encrypted:' value: {e}"
@@ -432,12 +432,26 @@ pub async fn write_lines(
        let job_id = job_id.clone();
        let mut nstream = String::new();

+        // Snapshot secrets once per batch — no lock needed per line.
+        // Trade-off: secrets registered mid-batch (between snapshot and log line)
+        // won't be masked until the next batch. In practice the async HTTP round-trip
+        // to fetch a secret completes before the script's log line arrives.
+        let mask_snapshot = windmill_common::sensitive_log_masks::snapshot(&job_id);
+
        while let Some(line) = read_lines.next().await {
            match line {
                Ok(line) => {
                    if line.is_empty() {
                        continue;
                    }
+                    let line = if let Some(ref snap) = mask_snapshot {
+                        match snap.mask(&line) {
+                            std::borrow::Cow::Owned(masked) => masked,
+                            std::borrow::Cow::Borrowed(_) => line,
+                        }
+                    } else {
+                        line
+                    };
                    if *OTEL_JOB_LOGS {
                        if let Some(otel_suffix) = line.strip_prefix(OTEL_PREFIX) {
                            tracing::event!(tracing::Level::INFO, otel_suffix);
@@ -2745,6 +2745,8 @@ pub async fn run_worker(

                    let arc_job = Arc::new(job);

+                    windmill_common::sensitive_log_masks::register_running_job(arc_job.id);
+
                    let span = create_span_with_name(&arc_job, &worker_name, Some(hostname), "job");

                    let job_result = handle_queued_job(
@@ -2844,6 +2846,8 @@ pub async fn run_worker(
                        _ => {}
                    }

+                    windmill_common::sensitive_log_masks::unregister_running_job(job_id);
+
                    #[cfg(feature = "prometheus")]
                    if let Some(duration) = _timer.map(|x| x.stop_and_record()) {
                        register_metric(
@@ -2,7 +2,7 @@ import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts";
 import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
 import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts";

-export const VERSION = "v1.664.0";
+export const VERSION = "v1.665.0";

 export async function login(email: string, password: string): Promise<string> {
  return await windmill.UserService.login({
@@ -68,7 +68,7 @@ export {
  workspaceAdd,
 };

-export const VERSION = "1.664.0";
+export const VERSION = "1.665.0";

 // Re-exported from constants.ts to maintain backwards compatibility
 export { WM_FORK_PREFIX } from "./core/constants.ts";
@@ -1,12 +1,12 @@
 {
 	"name": "windmill-components",
-	"version": "1.664.0",
+	"version": "1.665.0",
 	"lockfileVersion": 3,
 	"requires": true,
 	"packages": {
 		"": {
 			"name": "windmill-components",
-			"version": "1.664.0",
+			"version": "1.665.0",
 			"hasInstallScript": true,
 			"license": "AGPL-3.0",
 			"dependencies": {
@@ -1,6 +1,6 @@
 {
 	"name": "windmill-components",
-	"version": "1.664.0",
+	"version": "1.665.0",
 	"scripts": {
 		"dev": "vite dev",
 		"build": "vite build",
@@ -475,6 +475,11 @@
 					grant_type: valueToken.grant_type || 'authorization_code'
 				}

+				// Store scopes so token refresh uses the same scopes
+				if (scopes.length > 0) {
+					accountData.scopes = scopes
+				}
+
 				// Add client credentials if using client_credentials flow
 				if (useClientCredentials) {
 					accountData.cc_client_id = clientId.trim()
@@ -15,6 +15,7 @@
 		type?: 'danger' | 'reload'
 		showIcon?: boolean
 		id?: string
+		trashbin?: boolean
 		children?: Snippet
 		onConfirmed?: () => void | Promise<void>
 		onCanceled?: () => void
@@ -29,6 +30,7 @@
 		type: _type,
 		showIcon = true,
 		id,
+		trashbin = false,
 		children,
 		onConfirmed,
 		onCanceled
@@ -120,6 +122,12 @@
 							<div class="mt-2 text-sm text-secondary">
 								{@render children?.()}
 							</div>
+							{#if trashbin}
+								<p class="mt-3 text-xs text-tertiary"
+									>This item will be moved to the trashbin and can be restored by a workspace admin
+									within 3 days.</p
+								>
+							{/if}
 						</div>
 					</div>
 					<div class="flex items-center space-x-2 flex-row-reverse space-x-reverse mt-4">
@@ -92,7 +92,7 @@

 	let initialS3 = $derived(
 		Array.isArray(initialValue)
-			? initialValue?.map((v) => v.s3)
+			? initialValue?.filter((v) => v != null).map((v) => v.s3)
 			: initialValue?.s3
 				? [initialValue?.s3]
 				: undefined
@@ -112,7 +112,7 @@
 				if (!$fileUploads.find((fileUpload) => fileUpload.path === s3)) {
 					let initialFileUploads = initialValue
 						? Array.isArray(initialValue)
-							? initialValue.map(transform)
+							? initialValue.filter((v) => v != null).map(transform)
 							: [transform(initialValue)]
 						: []
 					$fileUploads = [...$fileUploads, ...initialFileUploads]
@@ -140,10 +140,10 @@ export const AI_AGENT_SCHEMA: Schema = {
 			format: 'json-schema',
 			showExpr: "fields.output_type === 'text'"
 		},
-		user_images: {
+		user_attachments: {
 			type: 'array',
 			description:
-				'Array of images to give as input to the AI agent. Requires a configured workspace S3 storage.',
+				'Array of files (images or PDFs) to give as input to the AI agent. Requires a configured workspace S3 storage.',
 			items: {
 				type: 'object',
 				resourceType: 's3object'
@@ -176,7 +176,7 @@ export const AI_AGENT_SCHEMA: Schema = {
 		'streaming',
 		'memory',
 		'output_schema',
-		'user_images',
+		'user_attachments',
 		'max_completion_tokens',
 		'temperature',
 		'max_iterations'
@@ -186,6 +186,12 @@ export const AI_AGENT_SCHEMA: Schema = {
 function migrateAiAgentInputTransforms(
 	inputTransforms: Record<string, InputTransform>
 ): Record<string, InputTransform> {
+	// Migrate user_images → user_attachments
+	if ('user_images' in inputTransforms && !('user_attachments' in inputTransforms)) {
+		inputTransforms.user_attachments = inputTransforms.user_images
+		delete inputTransforms.user_images
+	}
+
 	// Check if this has the legacy format
 	if ('messages_context_length' in inputTransforms && !('memory' in inputTransforms)) {
 		const legacyValue = inputTransforms.messages_context_length
@@ -93,6 +93,7 @@
 		open={Boolean(deleteConfirmedCallback)}
 		title="Remove"
 		confirmationText="Remove"
+		trashbin
 		on:canceled={() => {
 			deleteConfirmedCallback = undefined
 		}}
@@ -0,0 +1,249 @@
+<script lang="ts">
+	import { Button, Skeleton } from '$lib/components/common'
+	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
+	import DataTable from '$lib/components/table/DataTable.svelte'
+	import Head from '$lib/components/table/Head.svelte'
+	import Cell from '$lib/components/table/Cell.svelte'
+	import Row from '$lib/components/table/Row.svelte'
+	import { workspaceStore } from '$lib/stores'
+	import { sendUserToast } from '$lib/toast'
+	import { type TrashItem, TrashService } from '$lib/services/trashService'
+	import {
+		Trash2,
+		RotateCcw,
+		FileCode2,
+		GitFork,
+		AppWindow,
+		Clock,
+		Variable,
+		Database,
+		Zap,
+		RefreshCw
+	} from 'lucide-svelte'
+	import { untrack } from 'svelte'
+
+	let items: TrashItem[] | undefined = $state(undefined)
+	let deleteConfirmedCallback: (() => void) | undefined = $state(undefined)
+	let deleteOpen = $derived(Boolean(deleteConfirmedCallback))
+	let emptyConfirmOpen = $state(false)
+
+	function getKindIcon(kind: string) {
+		if (kind === 'script') return FileCode2
+		if (kind === 'flow') return GitFork
+		if (kind === 'app') return AppWindow
+		if (kind === 'schedule') return Clock
+		if (kind === 'variable') return Variable
+		if (kind === 'resource') return Database
+		if (kind.endsWith('_trigger')) return Zap
+		return Trash2
+	}
+
+	function getKindLabel(kind: string) {
+		if (kind === 'script') return 'Script'
+		if (kind === 'flow') return 'Flow'
+		if (kind === 'app') return 'App'
+		if (kind === 'schedule') return 'Schedule'
+		if (kind === 'variable') return 'Variable'
+		if (kind === 'resource') return 'Resource'
+		if (kind.endsWith('_trigger')) {
+			return (
+				kind
+					.replace('_trigger', '')
+					.replace(/_/g, ' ')
+					.replace(/\b\w/g, (c) => c.toUpperCase()) + ' Trigger'
+			)
+		}
+		return kind
+	}
+
+	function timeAgo(dateStr: string): string {
+		const now = new Date()
+		const date = new Date(dateStr)
+		const diffMs = now.getTime() - date.getTime()
+		const diffMins = Math.floor(diffMs / 60000)
+		if (diffMins < 1) return 'just now'
+		if (diffMins < 60) return `${diffMins}m ago`
+		const diffHours = Math.floor(diffMins / 60)
+		if (diffHours < 24) return `${diffHours}h ago`
+		const diffDays = Math.floor(diffHours / 24)
+		return `${diffDays}d ago`
+	}
+
+	function timeRemaining(dateStr: string): string {
+		const now = new Date()
+		const expires = new Date(dateStr)
+		const diffMs = expires.getTime() - now.getTime()
+		if (diffMs <= 0) return 'expired'
+		const diffHours = Math.floor(diffMs / 3600000)
+		if (diffHours < 1) return '< 1h remaining'
+		if (diffHours < 24) return `${diffHours}h remaining`
+		const diffDays = Math.floor(diffHours / 24)
+		const remainingHours = diffHours % 24
+		if (remainingHours === 0) return `${diffDays}d remaining`
+		return `${diffDays}d ${remainingHours}h remaining`
+	}
+
+	async function loadItems() {
+		items = await TrashService.listTrash({
+			workspace: $workspaceStore!
+		})
+	}
+
+	async function restoreItem(item: TrashItem) {
+		try {
+			await TrashService.restoreTrashItem({
+				workspace: $workspaceStore!,
+				id: item.id
+			})
+			sendUserToast(`Restored ${getKindLabel(item.item_kind)} '${item.item_path}'`)
+			loadItems()
+		} catch (e) {
+			sendUserToast(`Failed to restore: ${e}`, true)
+		}
+	}
+
+	async function permanentlyDelete(item: TrashItem) {
+		try {
+			await TrashService.permanentlyDeleteTrashItem({
+				workspace: $workspaceStore!,
+				id: item.id
+			})
+			sendUserToast(`Permanently deleted '${item.item_path}'`)
+			loadItems()
+		} catch (e) {
+			sendUserToast(`Failed to delete: ${e}`, true)
+		}
+	}
+
+	async function emptyAll() {
+		try {
+			const result = await TrashService.emptyTrash({ workspace: $workspaceStore! })
+			sendUserToast(result)
+			loadItems()
+		} catch (e) {
+			sendUserToast(`Failed to empty trash: ${e}`, true)
+		}
+	}
+
+	$effect(() => {
+		$workspaceStore
+		untrack(() => loadItems())
+	})
+</script>
+
+<div class="flex justify-end mb-4 gap-2">
+	<Button startIcon={{ icon: RefreshCw }} variant="default" size="xs" onclick={loadItems}>
+		Refresh
+	</Button>
+	<Button
+		startIcon={{ icon: Trash2 }}
+		variant="default"
+		size="xs"
+		onclick={() => {
+			emptyConfirmOpen = true
+		}}
+		disabled={!items || items.length === 0}
+	>
+		Empty Trashbin
+	</Button>
+</div>
+
+{#if items === undefined}
+	<Skeleton layout={[20, 8, 8, 8]} />
+{:else if items.length === 0}
+	<div class="flex flex-col items-center justify-center py-12 text-tertiary">
+		<Trash2 size={40} class="mb-3 opacity-50" />
+		<p class="text-base">Trashbin is empty</p>
+		<p class="text-sm mt-1">No recently deleted items.</p>
+	</div>
+{:else}
+	<DataTable size="sm">
+		<Head>
+			<tr>
+				<Cell head first>Type</Cell>
+				<Cell head>Path</Cell>
+				<Cell head>Deleted by</Cell>
+				<Cell head>Deleted</Cell>
+				<Cell head>Expires</Cell>
+				<Cell head last>Actions</Cell>
+			</tr>
+		</Head>
+		{#each items as item (item.id)}
+			{@const Icon = getKindIcon(item.item_kind)}
+			<Row>
+				<Cell first>
+					<div class="flex items-center gap-2">
+						<Icon size={14} />
+						<span class="text-xs">{getKindLabel(item.item_kind)}</span>
+					</div>
+				</Cell>
+				<Cell>
+					<span class="font-mono text-xs">{item.item_path}</span>
+				</Cell>
+				<Cell>
+					<span class="text-xs">{item.deleted_by}</span>
+				</Cell>
+				<Cell>
+					<span class="text-xs text-tertiary">{timeAgo(item.deleted_at)}</span>
+				</Cell>
+				<Cell>
+					<span class="text-xs text-tertiary">{timeRemaining(item.expires_at)}</span>
+				</Cell>
+				<Cell last>
+					<div class="flex gap-1">
+						<Button
+							startIcon={{ icon: RotateCcw }}
+							variant="default"
+							size="xs2"
+							onclick={() => restoreItem(item)}
+						>
+							Restore
+						</Button>
+						<Button
+							startIcon={{ icon: Trash2 }}
+							variant="default"
+							size="xs2"
+							onclick={() => {
+								deleteConfirmedCallback = () => permanentlyDelete(item)
+							}}
+						>
+							Delete
+						</Button>
+					</div>
+				</Cell>
+			</Row>
+		{/each}
+	</DataTable>
+{/if}
+
+<ConfirmationModal
+	open={deleteOpen}
+	title="Permanently delete"
+	confirmationText="Delete forever"
+	onCanceled={() => {
+		deleteConfirmedCallback = undefined
+	}}
+	onConfirmed={() => {
+		if (deleteConfirmedCallback) {
+			deleteConfirmedCallback()
+		}
+		deleteConfirmedCallback = undefined
+	}}
+>
+	<p>This item will be permanently deleted. This action cannot be undone.</p>
+</ConfirmationModal>
+
+<ConfirmationModal
+	open={emptyConfirmOpen}
+	title="Empty trashbin"
+	confirmationText="Empty trashbin"
+	onCanceled={() => {
+		emptyConfirmOpen = false
+	}}
+	onConfirmed={() => {
+		emptyAll()
+		emptyConfirmOpen = false
+	}}
+>
+	<p>All items in the trashbin will be permanently deleted. This action cannot be undone.</p>
+</ConfirmationModal>
@@ -21,6 +21,7 @@
 	title={`Are you sure you want to delete this ${trigger?.isDraft ? 'draft' : 'deployed'} trigger ?`}
 	confirmationText="Delete"
 	open={confirmationModalOpen}
+	trashbin
 	on:canceled={() => {
 		confirmationModalOpen = false
 	}}
@@ -0,0 +1,69 @@
+import { OpenAPI } from '$lib/gen/core/OpenAPI'
+import { request as __request } from '$lib/gen/core/request'
+
+export type TrashItem = {
+	id: number
+	workspace_id: string
+	item_kind: string
+	item_path: string
+	deleted_by: string
+	deleted_at: string
+	expires_at: string
+}
+
+export class TrashService {
+	public static listTrash(data: {
+		workspace: string
+		itemKind?: string
+		page?: number
+		perPage?: number
+	}): Promise<TrashItem[]> {
+		return __request(OpenAPI, {
+			method: 'GET',
+			url: '/w/{workspace}/trash/list',
+			path: {
+				workspace: data.workspace
+			},
+			query: {
+				item_kind: data.itemKind,
+				page: data.page,
+				per_page: data.perPage
+			}
+		})
+	}
+
+	public static restoreTrashItem(data: { workspace: string; id: number }): Promise<string> {
+		return __request(OpenAPI, {
+			method: 'POST',
+			url: '/w/{workspace}/trash/restore/{id}',
+			path: {
+				workspace: data.workspace,
+				id: data.id
+			}
+		})
+	}
+
+	public static permanentlyDeleteTrashItem(data: {
+		workspace: string
+		id: number
+	}): Promise<string> {
+		return __request(OpenAPI, {
+			method: 'DELETE',
+			url: '/w/{workspace}/trash/delete/{id}',
+			path: {
+				workspace: data.workspace,
+				id: data.id
+			}
+		})
+	}
+
+	public static emptyTrash(data: { workspace: string }): Promise<string> {
+		return __request(OpenAPI, {
+			method: 'POST',
+			url: '/w/{workspace}/trash/empty',
+			path: {
+				workspace: data.workspace
+			}
+		})
+	}
+}
@@ -30,13 +30,14 @@
 	import Toggle from '$lib/components/Toggle.svelte'
 	import Tooltip from '$lib/components/Tooltip.svelte'
 	import type { ResourceType, WorkspaceDeployUISettings } from '$lib/gen'
-	import { FolderService, OauthService, ResourceService, WorkspaceService, type ListableResource } from '$lib/gen'
 	import {
-		enterpriseLicense,
-		userStore,
-		workspaceStore,
-		userWorkspaces
-	} from '$lib/stores'
+		FolderService,
+		OauthService,
+		ResourceService,
+		WorkspaceService,
+		type ListableResource
+	} from '$lib/gen'
+	import { enterpriseLicense, userStore, workspaceStore, userWorkspaces } from '$lib/stores'
 	import { sendUserToast } from '$lib/toast'
 	import {
 		canWrite,
@@ -151,7 +152,12 @@
 	let folderPresets = $derived([
 		...folders.map((f) => ({ name: `f/${f}`, value: `path_start:\\ f/${f}/` })),
 		...(resourcesFilterSchema.user_folders_only
-			? [{ name: resourcesFilterSchema.user_folders_only.label ?? '?', value: 'user_folders_only:\\ true' }]
+			? [
+					{
+						name: resourcesFilterSchema.user_folders_only.label ?? '?',
+						value: 'user_folders_only:\\ true'
+					}
+				]
 			: [])
 	])

@@ -577,6 +583,7 @@
 	open={Boolean(deleteConfirmedCallback)}
 	title="Remove resource"
 	confirmationText="Remove"
+	trashbin
 	on:canceled={() => {
 		deleteConfirmedCallback = undefined
 	}}
@@ -74,7 +74,12 @@
 	let folderPresets = $derived([
 		...folders.map((f) => ({ name: `f/${f}`, value: `path_start:\\ f/${f}/` })),
 		...(variablesFilterSchema.user_folders_only
-			? [{ name: variablesFilterSchema.user_folders_only.label ?? '?', value: 'user_folders_only:\\ true' }]
+			? [
+					{
+						name: variablesFilterSchema.user_folders_only.label ?? '?',
+						value: 'user_folders_only:\\ true'
+					}
+				]
 			: [])
 	])
 	let contextualVariables: ContextualVariable[] = $state([])
@@ -576,6 +581,7 @@
 	{open}
 	title="Remove variable"
 	confirmationText="Remove"
+	trashbin
 	on:canceled={() => {
 		deleteConfirmedCallback = undefined
 	}}
@@ -56,6 +56,7 @@
 	import StorageSettings from '$lib/components/workspaceSettings/StorageSettings.svelte'
 	import VolumeStorageSettings from '$lib/components/workspaceSettings/VolumeStorageSettings.svelte'
 	import GitSyncSection from '$lib/components/git_sync/GitSyncSection.svelte'
+	import Trashbin from '$lib/components/settings/Trashbin.svelte'
 	import { untrack } from 'svelte'
 	import { getHandlerType } from '$lib/components/triggers/utils'
 	import DucklakeSettings, {
@@ -473,17 +474,15 @@
 	}

 	async function loadSettings(): Promise<void> {
-		const [settings, copilotSettingsState]: [
-			GetSettingsResponse,
-			GetCopilotSettingsStateResponse
-		] = await Promise.all([
-			WorkspaceService.getSettings({
-				workspace: $workspaceStore!
-			}),
-			WorkspaceService.getCopilotSettingsState({
-				workspace: $workspaceStore!
-			})
-		])
+		const [settings, copilotSettingsState]: [GetSettingsResponse, GetCopilotSettingsStateResponse] =
+			await Promise.all([
+				WorkspaceService.getSettings({
+					workspace: $workspaceStore!
+				}),
+				WorkspaceService.getCopilotSettingsState({
+					workspace: $workspaceStore!
+				})
+			])
 		slack_team_name = settings.slack_name
 		teams_team_id = settings.teams_team_id
 		teams_team_name = settings.teams_team_name
@@ -1193,6 +1192,12 @@
 					label: 'Encryption',
 					aiId: 'workspace-settings-encryption',
 					aiDescription: 'Encryption workspace settings'
+				},
+				{
+					id: 'trashbin',
+					label: 'Trashbin',
+					aiId: 'workspace-settings-trashbin',
+					aiDescription: 'Trashbin for recently deleted items'
 				}
 			]
 		}
@@ -1927,6 +1932,14 @@ export async function main(
 								saveLabel="Save & Re-encrypt workspace"
 								disabled={!!encryptionKeyValidationError || workspaceReencryptionInProgress}
 							/>
+						{:else if tab == 'trashbin'}
+							<SettingsPageHeader
+								title="Trashbin"
+								description="When scripts, flows, apps, resources, variables, schedules, or triggers are deleted, they are moved to the trashbin and kept for 3 days before being permanently removed. Admins can restore or permanently delete items from here."
+							/>
+							<div class="mt-4">
+								<Trashbin />
+							</div>
 						{/if}
 					</div>
 				</div>
@@ -136,11 +136,13 @@ ALL_PROVIDERS = [
    OPENROUTER,
 ]

-# Vision-capable providers for user_images tests
+# Vision-capable providers for user_images/user_attachments tests
 VISION_PROVIDERS = [
-    OPENAI,      # gpt-4o-mini supports vision
-    ANTHROPIC,   # claude-3 supports vision
-    GOOGLE_AI,   # gemini supports vision
+    OPENAI,       # gpt-4o-mini supports vision
+    ANTHROPIC,    # claude-3 supports vision
+    GOOGLE_AI,    # gemini supports vision
+    OPENROUTER,   # openai-compatible, vision depends on model
+    BEDROCK,      # bedrock converse API supports vision and documents
 ]


@@ -0,0 +1,21 @@
+%PDF-1.4
+1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj
+2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj
+3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 612 792]/Contents 4 0 R/Resources<</Font<</F1 5 0 R>>>>>>endobj
+4 0 obj<</Length 44>>stream
+BT /F1 24 Tf 100 700 Td (Hello PDF World) Tj ET
+endstream
+endobj
+5 0 obj<</Type/Font/Subtype/Type1/BaseFont/Helvetica>>endobj
+xref
+0 6
+0000000000 65535 f 
+0000000009 00000 n 
+0000000058 00000 n 
+0000000115 00000 n 
+0000000266 00000 n 
+0000000360 00000 n 
+trailer<</Size 6/Root 1 0 R>>
+startxref
+431
+%%EOF
@@ -0,0 +1,96 @@
+"""Tests for AI agent user_attachments (PDF) functionality with S3 storage.
+
+Prerequisites:
+- MinIO running on localhost:9000 with bucket 'wmill'
+- Test files uploaded to MinIO (test_images/test_image.webp, test_documents/test_document.pdf)
+- S3 resource and storage configured in the integration-tests workspace
+"""
+
+import pytest
+
+from .conftest import AIAgentTestClient, create_ai_agent_flow, TEST_IMAGE_S3_KEY
+from .providers import VISION_PROVIDERS, get_provider_ids
+
+TEST_PDF_S3_KEY = "test_documents/test_document.pdf"
+
+
+class TestUserAttachments:
+    """Test AI agent with PDF attachments from S3 storage."""
+
+    @pytest.mark.parametrize(
+        "provider_config",
+        VISION_PROVIDERS,
+        ids=get_provider_ids(VISION_PROVIDERS),
+    )
+    def test_pdf_analysis(
+        self,
+        client: AIAgentTestClient,
+        setup_providers,
+        provider_config,
+    ):
+        """Test that AI can analyze a PDF uploaded to S3."""
+        flow_value = create_ai_agent_flow(
+            provider_input_transform=provider_config["input_transform"],
+            system_prompt="You are a helpful assistant that reads documents. Be concise.",
+            include_user_images=True,
+        )
+
+        # Run the flow with the PDF (test_document.pdf contains "Hello PDF World")
+        result = client.run_preview_flow(
+            flow_value=flow_value,
+            args={
+                "user_message": "What text does this PDF document contain? Reply with just the text.",
+                "user_images": [
+                    {
+                        "s3": TEST_PDF_S3_KEY,
+                        "storage": None,
+                        "filename": "test_document.pdf",
+                    }
+                ],
+            },
+        )
+
+        assert result is not None
+        assert isinstance(result, (dict, str))
+        result_text = str(result).lower()
+        assert "hello" in result_text or "pdf" in result_text, (
+            f"Expected AI to read PDF content containing 'Hello PDF World', "
+            f"got: {result}"
+        )
+        print(f"PDF analysis result from {provider_config['name']}: {result}")
+
+    @pytest.mark.parametrize(
+        "provider_config",
+        VISION_PROVIDERS,
+        ids=get_provider_ids(VISION_PROVIDERS),
+    )
+    def test_backward_compat_user_images(
+        self,
+        client: AIAgentTestClient,
+        setup_providers,
+        provider_config,
+    ):
+        """Test that the old user_images field name still works for images."""
+        flow_value = create_ai_agent_flow(
+            provider_input_transform=provider_config["input_transform"],
+            system_prompt="You are a helpful assistant that describes images. Be concise.",
+            include_user_images=True,
+        )
+
+        result = client.run_preview_flow(
+            flow_value=flow_value,
+            args={
+                "user_message": "Describe what you see in this image in one sentence.",
+                "user_images": [
+                    {
+                        "s3": TEST_IMAGE_S3_KEY,
+                        "storage": None,
+                        "filename": "test_image.webp",
+                    }
+                ],
+            },
+        )
+
+        assert result is not None
+        assert isinstance(result, (dict, str))
+        print(f"Backward compat image result from {provider_config['name']}: {result}")
@@ -4,7 +4,7 @@ verify_ssl = true
 name = "pypi"

 [packages]
-wmill = ">=1.664.0"
+wmill = ">=1.665.0"
 sendgrid = "*"
 mysql-connector-python = "*"
 pymongo = "*"
@@ -1,7 +1,7 @@
 openapi: '3.0.3'

 info:
-  version: 1.664.0
+  version: 1.665.0
  title: OpenFlow Spec
  contact:
    name: Ruben Fiszel
@@ -1027,13 +1027,13 @@ components:
                JSON Schema object defining structured output format. Used when you need the AI to return data in a specific shape.
                Supports standard JSON Schema properties: type, properties, required, items, enum, pattern, minLength, maxLength, minimum, maximum, etc.
                Example: { type: 'object', properties: { name: { type: 'string' }, age: { type: 'integer' } }, required: ['name'] }
-            user_images:
+            user_attachments:
              allOf:
                - $ref: '#/components/schemas/InputTransform'
              description: |
-                Array of image references for vision-capable models.
+                Array of file references (images or PDFs) for the AI agent.
                Format: Array<{ bucket: string, key: string }> - S3 object references
-                Example: [{ bucket: 'my-bucket', key: 'images/photo.jpg' }]
+                Example: [{ bucket: 'my-bucket', key: 'documents/report.pdf' }]
            max_completion_tokens:
              allOf:
                - $ref: '#/components/schemas/InputTransform'
@@ -12,7 +12,7 @@
    RootModule        = 'WindmillClient.psm1'

    # Version number of this module.
-    ModuleVersion     = '1.664.0'
+    ModuleVersion     = '1.665.0'

    # Supported PSEditions
    # CompatiblePSEditions = @()
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "wmill"
-version = "1.664.0"
+version = "1.665.0"
 description = "A client library for accessing Windmill server wrapping the Windmill client API"
 license = "Apache-2.0"
 homepage = "https://windmill.dev"
@@ -1,6 +1,6 @@
 {
  "name": "@windmill/windmill",
-  "version": "1.664.0",
+  "version": "1.665.0",
  "exports": "./src/index.ts",
  "publish": {
    "exclude": ["!src", "./s3Types.ts", "./sqlUtils.ts", "./client.ts"]
@@ -1,7 +1,7 @@
 {
  "name": "windmill-client",
  "description": "Windmill SDK client for browsers and Node.js",
-  "version": "1.664.0",
+  "version": "1.665.0",
  "author": "Ruben Fiszel",
  "license": "Apache 2.0",
  "sideEffects": false,
@@ -1 +1 @@
-1.664.0
+1.665.0
				`@@ -0,0 +1 @@`
				`ALTER TABLE account DROP COLUMN IF EXISTS scopes;`
				`@@ -0,0 +1 @@`
				`ALTER TABLE account ADD COLUMN scopes TEXT[];`
@@ -1 +1 @@
 .664.0
 .665.0