diff --git a/backend/Cargo.lock b/backend/Cargo.lock
index c903490c8d..38095e4037 100644
--- a/backend/Cargo.lock
+++ b/backend/Cargo.lock
@@ -16197,6 +16197,7 @@ dependencies = [
  "aws-credential-types",
  "aws-sdk-sqs",
  "base64 0.22.1",
+ "futures",
  "rand 0.9.0",
  "rdkafka",
  "reqwest 0.13.1",
@@ -16477,6 +16478,7 @@ name = "windmill-common"
 version = "1.664.0"
 dependencies = [
  "aes-gcm",
+ "aho-corasick",
  "anyhow",
  "async-recursion",
  "async-stream",
diff --git a/backend/windmill-api-integration-tests/Cargo.toml b/backend/windmill-api-integration-tests/Cargo.toml
index 35dd64f995..cb4827e857 100644
--- a/backend/windmill-api-integration-tests/Cargo.toml
+++ b/backend/windmill-api-integration-tests/Cargo.toml
@@ -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
diff --git a/backend/windmill-api-integration-tests/tests/sensitive_log_masking.rs b/backend/windmill-api-integration-tests/tests/sensitive_log_masking.rs
new file mode 100644
index 0000000000..fb759ee4b5
--- /dev/null
+++ b/backend/windmill-api-integration-tests/tests/sensitive_log_masking.rs
@@ -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(())
+}
diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml
index a0aa27342d..d593787def 100644
--- a/backend/windmill-common/Cargo.toml
+++ b/backend/windmill-common/Cargo.toml
@@ -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
diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs
index 0093f1088f..b5ec518315 100644
--- a/backend/windmill-common/src/lib.rs
+++ b/backend/windmill-common/src/lib.rs
@@ -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")]
diff --git a/backend/windmill-common/src/sensitive_log_masks.rs b/backend/windmill-common/src/sensitive_log_masks.rs
new file mode 100644
index 0000000000..8123d6d55b
--- /dev/null
+++ b/backend/windmill-common/src/sensitive_log_masks.rs
@@ -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());
+    }
+}
diff --git a/backend/windmill-store/src/variables.rs b/backend/windmill-store/src/variables.rs
index 9db4f010a6..964f81809c 100644
--- a/backend/windmill-store/src/variables.rs
+++ b/backend/windmill-store/src/variables.rs
@@ -1115,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());
diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs
index 0c81c45286..c6d737becb 100644
--- a/backend/windmill-worker/src/common.rs
+++ b/backend/windmill-worker/src/common.rs
@@ -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}"
diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs
index 46cc047bf0..2edaa4f3a1 100644
--- a/backend/windmill-worker/src/handle_child.rs
+++ b/backend/windmill-worker/src/handle_child.rs
@@ -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);
diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs
index bdeea1c076..b75403bef7 100644
--- a/backend/windmill-worker/src/worker.rs
+++ b/backend/windmill-worker/src/worker.rs
@@ -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(