fix: register the job token with the sensitive log masking system (#10943)

* fix(worker): register the job token with the log masking system

The masking system covered secrets fetched through `get_value_internal` and
`$encrypted:` args, but not the job's own token, so a script that echoed
`$WM_TOKEN` wrote it verbatim into logs that are persisted to the database
and, when configured, to object storage.

`run_worker` now registers the token for the job it just pulled, alongside
the existing `register_running_job` call, so it is redacted like any other
registered secret.

That makes every job carry at least one registered value, where before the
per-batch mask snapshot was skipped entirely for the majority of jobs that
touched no secret. Cache the compiled Aho-Corasick automaton per job and
invalidate it when a new secret is registered, so a chatty job no longer
rebuilds it once per log batch.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P316wKe2QCYNcdsx1PwAJ3

* fix(nativets): mask secrets in the in-process log path

NativeTS hands `console.log` output to a task that drains a channel into
`append_logs`, so it never reaches the masking in `handle_child::write_lines`
and a script logging `$WM_TOKEN` persisted the raw JWT.

That drain can still be flushing after the job is unregistered, so a plain
per-line `snapshot` would leave the tail unmasked. `JobMasker` keeps the last
masks it saw for exactly that window, and refreshes while the job is alive so
secrets fetched mid-run are covered too.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P316wKe2QCYNcdsx1PwAJ3

* fix(nativets): seed the job masker at construction

A `JobMasker` that only looked up its masks on the first `mask` call had the
same hole at the head of the log that its retention closes at the tail: if the
drain task's first productive poll landed after the job was unregistered, the
registry was already gone and every line was written raw.

`new` now takes the snapshot, and its callers construct it from the job's own
execution while the job is still registered.

Also cover the nativets sink with an integration test, gated on `deno_core` the
way the CI test build is.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P316wKe2QCYNcdsx1PwAJ3

* fix(nativets): mask on the producing side of the log channel

Masking as the drain task wrote to `append_logs` left two holes, because that
task is detached and outlives the job: a secret registered mid-run could still
be queued when the job was unregistered and would then be written raw, and the
`windmill:job_log` tracing emission that EE forwards job logs on never went
through the mask at all.

Mask where the line is produced instead. That loop is joined before the job
completes, so the job's secrets are always still registered, and one call now
covers both the tracing mirror and the channel. The result stream keeps reading
the raw text, the way `handle_child` keeps its raw `line` for results.

`JobMasker` is no longer load-bearing for the post-unregistration window, so it
is documented for what it now does: keep the security notice to once per set of
secrets for a sink that masks line by line.

Also drop the nativets test's tag override — `DEFAULT_TAGS` does advertise
`nativets`, so the comment justifying it was wrong — and pin the automaton
cache invalidation, whose failure mode is an unmasked secret.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P316wKe2QCYNcdsx1PwAJ3

* fix(worker): hold the log-masking lifecycle at the job boundary

Registering the job around the poller's call left every other way of running a
job uncovered: the interactive worker shell and inline AI agent tools both call
`handle_queued_job` directly, and a script logging `$WM_TOKEN` from either
persisted the live credential. Register from inside `handle_queued_job` instead,
under a drop guard, so each path is covered by construction rather than by
remembering to add a call. Nothing is lost by unregistering earlier: the writes
that follow go through `append_logs`, which never consulted the registry.

In nativets, decide the stream/log routing before masking. `MaskSnapshot`'s
notice is one-shot, so a secret-bearing `WM_STREAM:` chunk used to spend it on
text that is then discarded, leaving later redactions in `job_logs` unexplained.

Restore the masker's post-unregistration test: the memory-limit path never joins
the producing loop, so that fallback is still load-bearing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P316wKe2QCYNcdsx1PwAJ3

* docs: correct the nativets masking comments

The producer loop is not joined on the memory-limit path, so it does not
"always" run while the job is registered — say normally, which is what
`JobMasker`'s fallback is there for. Name the reason a stream chunk stays raw
everywhere it goes, including the tracing mirror: it is result data that no log
sink persists, so masking it would be masking a result. State the masker test's
invariant without asserting a mechanism behind it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P316wKe2QCYNcdsx1PwAJ3

* fix(masking): keep the security notice on a line of its own

`mask` appended the notice as a newline plus the notice text, which
assumes the caller hands it a bare log line. nativets hands it a chunk
that already ends in a newline, and its sink concatenates chunks
verbatim, so the notice arrived after a blank line and the next log line
was welded onto the end of it.

Emit the notice as its own line for either shape. `handle_child` is
unaffected: its input never ends in a newline, so it keeps the original
path.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P316wKe2QCYNcdsx1PwAJ3

---------

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

backend/tests/job_token_log_masking.rs

@@ -0,0 +1,108 @@
+/*
+ * The job's own token (`$WM_TOKEN`) stays valid well past the job it was minted
+ * for, and job logs are persisted to `job_logs` and optionally to object storage,
+ * so a script that echoes the token would otherwise park a live credential in
+ * durable storage. `run_worker` registers the token with `sensitive_log_masks`
+ * for the job it pulled; this pins that the persisted log carries the masked form.
+ */
+
+use sqlx::{Pool, Postgres};
+use windmill_common::{
+    jobs::{JobPayload, RawCode},
+    scripts::ScriptLang,
+};
+use windmill_test_utils::*;
+
+/// Prefix of a serialized job token: `jwt_` plus the base64 of a JWT header.
+/// The masked form keeps only `jwt` + the last three characters, so it never matches.
+const RAW_TOKEN_PREFIX: &str = "jwt_ey";
+
+#[sqlx::test(fixtures("base"))]
+async fn test_job_token_masked_in_persisted_logs(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    let job = RunJob::from(JobPayload::Code(RawCode {
+        hash: None,
+        content: "echo \"running with --token $WM_TOKEN\"".to_string(),
+        path: None,
+        lock: None,
+        language: ScriptLang::Bash,
+        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,
+        tag: None,
+    }))
+    .run_until_complete(&db, false, port)
+    .await;
+    assert!(job.success, "job should have succeeded");
+
+    let logs =
+        sqlx::query_scalar::<_, Option<String>>("SELECT logs FROM job_logs WHERE job_id = $1")
+            .bind(job.id)
+            .fetch_one(&db)
+            .await?
+            .unwrap_or_default();
+
+    assert!(
+        !logs.contains(RAW_TOKEN_PREFIX),
+        "an unmasked job token reached the persisted logs: {logs}"
+    );
+    assert!(
+        logs.contains("secret value was masked"),
+        "expected the masking notice in logs: {logs}"
+    );
+    Ok(())
+}
+
+/// nativets runs V8 in-process and persists `console.log` output through its own
+/// channel, so it is masked by a different mechanism than the bash case above and
+/// needs its own guard.
+#[cfg(feature = "deno_core")]
+#[sqlx::test(fixtures("base"))]
+async fn test_job_token_masked_in_nativets_logs(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    let job = RunJob::from(JobPayload::Code(RawCode {
+        hash: None,
+        content: "export async function main() {\n  console.log('running with --token ' + process.env.WM_TOKEN);\n  return 'ok';\n}".to_string(),
+        path: None,
+        lock: None,
+        language: ScriptLang::Nativets,
+        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,
+        tag: None,
+    }))
+    .run_until_complete(&db, false, port)
+    .await;
+    assert!(job.success, "job should have succeeded");
+
+    let logs =
+        sqlx::query_scalar::<_, Option<String>>("SELECT logs FROM job_logs WHERE job_id = $1")
+            .bind(job.id)
+            .fetch_one(&db)
+            .await?
+            .unwrap_or_default();
+
+    assert!(
+        !logs.contains(RAW_TOKEN_PREFIX),
+        "an unmasked job token reached the persisted logs: {logs}"
+    );
+    assert!(
+        logs.contains("secret value was masked"),
+        "expected the masking notice in logs: {logs}"
+    );
+    Ok(())
+}

backend/windmill-common/src/sensitive_log_masks.rs

@@ -10,7 +10,7 @@
 
 use std::borrow::Cow;
 use std::collections::{HashMap, HashSet};
-use std::sync::RwLock;
+use std::sync::{Arc, RwLock};
 use uuid::Uuid;
 
 /// Minimum length for a secret to be registered for masking.
@@ -20,9 +20,27 @@ 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";
 
+/// The secrets registered for one job, plus the automaton compiled from them.
+#[derive(Default)]
+struct JobMasks {
+    secrets: HashSet<String>,
+    /// Built on the first `snapshot` after a change and shared by every later
+    /// snapshot. Every job registers at least its own token, so without this
+    /// cache each log batch of each job would rebuild the automaton.
+    compiled: Option<Arc<CompiledMasks>>,
+}
+
+/// Aho-Corasick automaton for O(m) multi-pattern matching in a single pass,
+/// regardless of the number of secrets registered, with the replacement
+/// strings indexed to match the automaton's pattern order.
+struct CompiledMasks {
+    ac: aho_corasick::AhoCorasick,
+    replacements: Vec<String>,
+}
+
 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>>> =
+    /// Map of job_id -> secret values that should be masked in that job's logs.
+    static ref SENSITIVE_MASKS: RwLock<HashMap<Uuid, JobMasks>> =
         RwLock::new(HashMap::new());
 
     /// Set of currently running job IDs on this worker process.
@@ -32,13 +50,8 @@ lazy_static::lazy_static! {
 }
 
 /// 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>,
+    compiled: Arc<CompiledMasks>,
     /// 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>,
@@ -53,34 +66,104 @@ impl MaskSnapshot {
         }
 
         // Single-pass check + replace using the pre-built automaton
-        if !self.ac.is_match(text) {
+        if !self.compiled.ac.is_match(text) {
             return Cow::Borrowed(text);
         }
 
-        let mut result = self.ac.replace_all(text, &self.replacements);
+        let mut result = self
+            .compiled
+            .ac
+            .replace_all(text, &self.compiled.replacements);
 
-        // Append the notice only once per snapshot (i.e. per batch)
+        // Append the notice only once per snapshot (i.e. per batch), as its own line.
+        // Callers pass either a bare line (`handle_child`) or a chunk that already ends
+        // in a newline (nativets), and the sinks concatenate what they get verbatim:
+        // assuming either shape welds the notice onto a neighbouring line.
         if !self.notice_shown.get() {
             self.notice_shown.set(true);
-            result.push('\n');
-            result.push_str(MASKED_NOTICE);
+            if result.ends_with('\n') {
+                result.push_str(MASKED_NOTICE);
+                result.push('\n');
+            } else {
+                result.push('\n');
+                result.push_str(MASKED_NOTICE);
+            }
         }
 
         Cow::Owned(result)
     }
 }
 
+/// A masker for sinks that mask line by line rather than in batches, like nativets
+/// masking each `console.log` chunk as V8 produces it. `snapshot` per line would
+/// re-arm the security notice on every one; this keeps it to once per distinct set
+/// of secrets while still picking up secrets registered mid-run.
+///
+/// Masks by job id alone — the caller is the one that knows the text it passes
+/// belongs to that job.
+pub struct JobMasker {
+    job_id: Uuid,
+    snapshot: Option<MaskSnapshot>,
+}
+
+impl JobMasker {
+    pub fn new(job_id: Uuid) -> Self {
+        JobMasker { job_id, snapshot: snapshot(&job_id) }
+    }
+
+    /// Mask every secret registered for the job. Returns `Cow::Borrowed` when no match.
+    /// Falls back to the masks it last saw once the job is unregistered, so a sink
+    /// still draining past the end of a run does not start emitting secrets.
+    pub fn mask<'a>(&mut self, text: &'a str) -> Cow<'a, str> {
+        if let Some(fresh) = snapshot(&self.job_id) {
+            // Replacing an equivalent snapshot would re-arm the notice, so only take
+            // one built from a secret set we have not seen.
+            let unchanged = self
+                .snapshot
+                .as_ref()
+                .is_some_and(|cur| Arc::ptr_eq(&cur.compiled, &fresh.compiled));
+            if !unchanged {
+                self.snapshot = Some(fresh);
+            }
+        }
+        match self.snapshot.as_ref() {
+            Some(snapshot) => snapshot.mask(text),
+            None => Cow::Borrowed(text),
+        }
+    }
+}
+
 /// 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;
+    {
+        let masks = SENSITIVE_MASKS.read().unwrap_or_else(|e| e.into_inner());
+        let job = masks.get(job_id)?;
+        if job.secrets.is_empty() {
+            return None;
+        }
+        if let Some(compiled) = job.compiled.as_ref() {
+            return Some(MaskSnapshot {
+                compiled: compiled.clone(),
+                notice_shown: std::cell::Cell::new(false),
+            });
+        }
     }
 
+    let mut masks = SENSITIVE_MASKS.write().unwrap_or_else(|e| e.into_inner());
+    let job = masks.get_mut(job_id)?;
+    if job.secrets.is_empty() {
+        return None;
+    }
+    let compiled = job
+        .compiled
+        .get_or_insert_with(|| Arc::new(compile(&job.secrets)));
+    Some(MaskSnapshot { compiled: compiled.clone(), notice_shown: std::cell::Cell::new(false) })
+}
+
+fn compile(secrets: &HashSet<String>) -> CompiledMasks {
     // 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()));
@@ -106,7 +189,7 @@ pub fn snapshot(job_id: &Uuid) -> Option<MaskSnapshot> {
         .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) })
+    CompiledMasks { ac, replacements }
 }
 
 /// Register a job as currently running. Call this before `handle_queued_job`.
@@ -148,20 +231,110 @@ pub fn register_secret_for_all_running_jobs(secret: &str) {
 
     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());
+        if let Some(job) = masks.get_mut(&job_id) {
+            if job.secrets.insert(secret.to_string()) {
+                job.compiled = None;
+            }
         }
     }
 }
 
 /// Register a secret value for a specific job.
-/// Used for `$encrypted:` args where we know the job ID.
+/// Used for the job's own token and 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());
+    if let Some(job) = masks.get_mut(&job_id) {
+        if job.secrets.insert(secret.to_string()) {
+            job.compiled = None;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The compiled automaton is cached per job, so a secret registered after the
+    /// first snapshot only gets masked if the cache is invalidated.
+    #[test]
+    fn snapshot_rebuilds_after_a_new_secret_is_registered() {
+        let job_id = Uuid::new_v4();
+        register_running_job(job_id);
+        register_secret_for_job(job_id, "firstsecretvalue");
+        let _ = snapshot(&job_id)
+            .expect("secret registered")
+            .mask("firstsecretvalue");
+
+        register_secret_for_job(job_id, "secondsecretvalue");
+
+        let snap = snapshot(&job_id).expect("secrets registered");
+        let masked = snap.mask("firstsecretvalue then secondsecretvalue");
+        assert!(!masked.contains("firstsecretvalue"), "{masked}");
+        assert!(!masked.contains("secondsecretvalue"), "{masked}");
+        unregister_running_job(job_id);
+    }
+
+    /// A line-by-line sink must not repeat the notice on every line, and must still
+    /// pick up a secret registered after the masker was built.
+    #[test]
+    fn job_masker_notices_once_per_secret_set() {
+        let job_id = Uuid::new_v4();
+        register_running_job(job_id);
+        register_secret_for_job(job_id, "firstsecretvalue");
+        let mut masker = JobMasker::new(job_id);
+
+        let first = masker.mask("saw firstsecretvalue").into_owned();
+        assert!(!first.contains("firstsecretvalue"), "{first}");
+        assert!(first.contains(MASKED_NOTICE), "{first}");
+
+        let second = masker.mask("saw firstsecretvalue again").into_owned();
+        assert!(!second.contains("firstsecretvalue"), "{second}");
+        assert!(!second.contains(MASKED_NOTICE), "{second}");
+
+        register_secret_for_job(job_id, "secondsecretvalue");
+        let third = masker.mask("saw secondsecretvalue").into_owned();
+        assert!(!third.contains("secondsecretvalue"), "{third}");
+        unregister_running_job(job_id);
+    }
+
+    /// Unregistration must not turn masking off under a sink that is still emitting:
+    /// the masker keeps working off the masks it last saw rather than going quiet.
+    #[test]
+    fn job_masker_masks_after_the_job_is_unregistered() {
+        let job_id = Uuid::new_v4();
+        register_running_job(job_id);
+        register_secret_for_job(job_id, "supersecretvalue");
+        let mut masker = JobMasker::new(job_id);
+
+        unregister_running_job(job_id);
+
+        let masked = masker.mask("logged supersecretvalue here");
+        assert!(!masked.contains("supersecretvalue"), "{masked}");
+    }
+
+    /// The notice has to end up on a line of its own for both shapes callers pass:
+    /// a bare line (`handle_child`) and a newline-terminated chunk (nativets). The
+    /// sinks concatenate what they are given verbatim, so getting this wrong welds
+    /// the notice onto whichever line follows it.
+    #[test]
+    fn notice_lands_on_its_own_line_for_both_caller_shapes() {
+        let job_id = Uuid::new_v4();
+        register_running_job(job_id);
+        register_secret_for_job(job_id, "supersecretvalue");
+
+        let line = snapshot(&job_id)
+            .expect("secret registered")
+            .mask("tok supersecretvalue");
+        assert_eq!(line, format!("tok s*****e\n{MASKED_NOTICE}"));
+
+        let chunk = snapshot(&job_id)
+            .expect("secret registered")
+            .mask("tok supersecretvalue\n");
+        assert_eq!(chunk, format!("tok s*****e\n{MASKED_NOTICE}\n"));
+
+        unregister_running_job(job_id);
     }
 }

backend/windmill-runtime-nativets/src/lib.rs

@@ -838,6 +838,11 @@ pub async fn eval_fetch_timeout(
                 }
             }
             let w_id_for_tracing = w_id_for_tracing;
+            // nativets delivers logs in-process, so they never reach the masking in
+            // `handle_child::write_lines` and a `console.log` of `$WM_TOKEN` would be
+            // persisted verbatim. Mask here rather than in the detached task draining into
+            // `append_logs`: this loop normally runs while the job is still registered.
+            let mut masker = windmill_common::sensitive_log_masks::JobMasker::new(job_id);
             let handle = tokio::spawn(async move {
                 let mut result_stream = String::new();
                 let mut is_stream = false;
@@ -845,10 +850,20 @@ pub async fn eval_fetch_timeout(
                     use windmill_common::result_stream::extract_stream_from_logs;
                     use windmill_common::tracing_init::{OTEL_JOB_LOGS, OTEL_PREFIX};
 
+                    let stream = extract_stream_from_logs(&log.trim_end_matches("\n"));
+
+                    // A stream chunk is result data, not a log line — it never reaches
+                    // `job_logs`, and `merge_result_stream` can make it the job's result —
+                    // so it stays raw wherever it goes, here and in the mirror below.
+                    // Deliberately unlike `handle_child`, which streams the masked text.
+                    // Routed before masking because the notice is one-shot: spent on a chunk
+                    // no sink persists, a later redaction in `job_logs` would go unexplained.
+                    let logged = stream.is_none().then(|| masker.mask(&log).into_owned());
+
                     // Mirror `process_streaming_log_lines` (EE) + the OTEL_JOB_LOGS
                     // hook from handle_child.rs, neither of which runs for nativets
                     // since nativets delivers logs in-process via the log channel.
-                    for line in log.lines() {
+                    for line in logged.as_deref().unwrap_or(&log).lines() {
                         tracing::info!(
                             target: "windmill:job_log",
                             job_id = ?job_id,
@@ -862,7 +877,7 @@ pub async fn eval_fetch_timeout(
                         }
                     }
 
-                    if let Some(stream) = extract_stream_from_logs(&log.trim_end_matches("\n")) {
+                    if let Some(stream) = stream {
                         if !is_stream {
                             is_stream = true;
                             if let Some(ref f) = stream_notifier_update {
@@ -874,8 +889,8 @@ pub async fn eval_fetch_timeout(
                         if let Err(e) = result_stream_sender.send(stream) {
                             tracing::error!("failed to send result stream: {e}");
                         }
-                    } else {
-                        if let Err(e) = append_logs_sender.send(log) {
+                    } else if let Some(logged) = logged {
+                        if let Err(e) = append_logs_sender.send(logged) {
                             tracing::error!("failed to send log: {e}");
                         }
                     }

backend/windmill-worker/src/worker.rs

@@ -3859,8 +3859,6 @@ 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 log_ctx = log_context_for_job(&arc_job, &worker_name, Some(hostname));
 
@@ -3976,8 +3974,6 @@ 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(
@@ -4512,6 +4508,30 @@ async fn detect_and_store_runtime_assets_from_job_args(
     }
 }
 
+/// Holds a job's entry in the log-masking registry for as long as it executes, so
+/// that secrets it fetches can be registered against it, and masks the job's own
+/// token from the start: `$WM_TOKEN` stays valid well past the run, and a script
+/// that echoes it would otherwise leave a live credential in the persisted logs.
+///
+/// Lives here rather than at the call sites so that every way of running a job —
+/// the poller, the interactive worker shell, an inline AI agent tool — is covered
+/// by construction.
+struct RunningJobMasks(Uuid);
+
+impl RunningJobMasks {
+    fn register(job_id: Uuid, token: &str) -> Self {
+        windmill_common::sensitive_log_masks::register_running_job(job_id);
+        windmill_common::sensitive_log_masks::register_secret_for_job(job_id, token);
+        RunningJobMasks(job_id)
+    }
+}
+
+impl Drop for RunningJobMasks {
+    fn drop(&mut self) {
+        windmill_common::sensitive_log_masks::unregister_running_job(self.0);
+    }
+}
+
 pub async fn handle_queued_job(
     job: Arc<MiniPulledJob>,
     raw_code: Option<String>,
@@ -4533,6 +4553,8 @@ pub async fn handle_queued_job(
     flow_runners: Option<Arc<FlowRunners>>,
     #[cfg(feature = "benchmark")] _bench: &mut BenchmarkIter,
 ) -> windmill_common::error::Result<JobOutcome> {
+    let _masks = RunningJobMasks::register(job.id, &client.token);
+
     if job.canceled_by.is_some() {
         return Err(Error::JsonErr(canceled_job_to_result(&job)));
     }