diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs
index 88547c52b..0f3c65885 100644
--- a/lib/crates/fabro-cli/src/commands/run/runner.rs
+++ b/lib/crates/fabro-cli/src/commands/run/runner.rs
@@ -1458,10 +1458,7 @@ mod tests {
         );
         let event = running_event(None);
 
-        // This worker-stamp unit test emits no run-declared secret values.
-        sink.write_run_event(&event, &fabro_redact::SecretRedactor::default())
-            .await
-            .unwrap();
+        sink.write_run_event(&event).await.unwrap();
 
         let first = first.lock().await;
         let second = second.lock().await;
diff --git a/lib/crates/fabro-workflow/src/event/sink.rs b/lib/crates/fabro-workflow/src/event/sink.rs
index 89f6d4020..d4b30e627 100644
--- a/lib/crates/fabro-workflow/src/event/sink.rs
+++ b/lib/crates/fabro-workflow/src/event/sink.rs
@@ -31,14 +31,26 @@ pub async fn append_event_to_sink(
     sink: &RunEventSink,
     run_id: &RunId,
     event: &Event,
-    redactor: &SecretRedactor,
 ) -> Result<()> {
     let stored = to_run_event(run_id, event);
-    sink.write_run_event(&stored, redactor).await
+    sink.write_run_event(&stored).await
+}
+
+/// Destination for run events, carrying the run-scoped [`SecretRedactor`].
+///
+/// The sink owns the exact-match secret registry so every event written
+/// through it — including bootstrap and failure paths that run before a
+/// `RunSession` exists or after one failed to construct — passes the same
+/// registry that run-boundary secret resolution populates via
+/// [`Self::secret_redactor`]. Clones share the registry.
+#[derive(Clone)]
+pub struct RunEventSink {
+    redactor: SecretRedactor,
+    kind:     SinkKind,
 }
 
 #[derive(Clone)]
-pub enum RunEventSink {
+enum SinkKind {
     Store(RunStoreHandle),
     JsonLines(Arc<AsyncMutex<Pin<Box<dyn AsyncWrite + Send>>>>),
     Callback(Arc<RunEventSinkCallback>),
@@ -54,14 +66,21 @@ type RunEventSinkCallback = dyn Fn(RunEvent) -> RunEventSinkFuture + Send + Sync
 type RunEventTransform = dyn Fn(RunEvent) -> RunEvent + Send + Sync + 'static;
 
 impl RunEventSink {
+    fn from_kind(kind: SinkKind) -> Self {
+        Self {
+            redactor: SecretRedactor::default(),
+            kind,
+        }
+    }
+
     #[must_use]
     pub fn store(run_store: RunDatabase) -> Self {
-        Self::Store(RunStoreHandle::local(run_store))
+        Self::from_kind(SinkKind::Store(RunStoreHandle::local(run_store)))
     }
 
     #[must_use]
     pub fn backend(run_store: RunStoreHandle) -> Self {
-        Self::Store(run_store)
+        Self::from_kind(SinkKind::Store(run_store))
     }
 
     #[must_use]
@@ -69,7 +88,9 @@ impl RunEventSink {
     where
         W: AsyncWrite + Send + 'static,
     {
-        Self::JsonLines(Arc::new(AsyncMutex::new(Box::pin(writer))))
+        Self::from_kind(SinkKind::JsonLines(Arc::new(AsyncMutex::new(Box::pin(
+            writer,
+        )))))
     }
 
     #[must_use]
@@ -78,51 +99,78 @@ impl RunEventSink {
         F: Fn(RunEvent) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = Result<()>> + Send + 'static,
     {
-        Self::Callback(Arc::new(move |event| Box::pin(callback(event))))
+        Self::from_kind(SinkKind::Callback(Arc::new(move |event| {
+            Box::pin(callback(event))
+        })))
     }
 
+    /// Fan events out to every sink. The composed sink adopts the first
+    /// sink's secret registry; compose sinks before registering secrets.
     #[must_use]
     pub fn fanout(sinks: Vec<Self>) -> Self {
+        let redactor = sinks
+            .first()
+            .map(|sink| sink.redactor.clone())
+            .unwrap_or_default();
         let mut flattened = Vec::new();
         for sink in sinks {
-            match sink {
-                Self::Composite(inner) => flattened.extend(inner),
+            match sink.kind {
+                SinkKind::Composite(inner) => flattened.extend(inner),
                 other => flattened.push(other),
             }
         }
-        Self::Composite(flattened)
+        Self {
+            redactor,
+            kind: SinkKind::Composite(flattened),
+        }
     }
 
+    /// Transform every event before it reaches `inner`. The composed sink
+    /// keeps the inner sink's secret registry.
     #[must_use]
     pub fn map<F>(transform: F, inner: Self) -> Self
     where
         F: Fn(RunEvent) -> RunEvent + Send + Sync + 'static,
     {
-        Self::Map {
-            transform: Arc::new(transform),
-            inner:     Box::new(inner),
+        Self {
+            redactor: inner.redactor,
+            kind:     SinkKind::Map {
+                transform: Arc::new(transform),
+                inner:     Box::new(inner.kind),
+            },
         }
     }
 
-    pub async fn write_run_event(&self, event: &RunEvent, redactor: &SecretRedactor) -> Result<()> {
-        let mut pending = vec![(self, PendingEvent::Raw(event.clone()))];
+    /// Run-scoped exact-match secret registry consulted on every write.
+    ///
+    /// Run-boundary secret resolution registers each resolved value here;
+    /// clones of this sink share the registry, so events written from any
+    /// phase of the run see every registration made before the write.
+    #[must_use]
+    pub fn secret_redactor(&self) -> &SecretRedactor {
+        &self.redactor
+    }
+
+    pub async fn write_run_event(&self, event: &RunEvent) -> Result<()> {
+        let redactor = &self.redactor;
+        let mut pending = vec![(&self.kind, PendingEvent::Raw(event.clone()))];
         while let Some((sink, event)) = pending.pop() {
             match sink {
-                Self::Store(run_store) => {
+                SinkKind::Store(run_store) => {
                     let redacted = event.into_redacted(redactor)?;
                     run_store.append_run_event(&redacted).await?;
                 }
-                Self::JsonLines(writer) => {
+                SinkKind::JsonLines(writer) => {
                     let line = event.into_redacted(redactor)?.json_line()?;
                     let mut writer = writer.lock().await;
                     writer.write_all(line.as_bytes()).await?;
                     writer.write_all(b"\n").await?;
                     writer.flush().await?;
                 }
-                Self::Callback(callback) => {
+                SinkKind::Callback(callback) => {
                     callback(event.into_redacted(redactor)?.event().clone()).await?;
                 }
-                Self::Map { transform, inner } => {
+                SinkKind::Map { transform, inner } => {
                     // A transform can add new fields, so its output re-enters
                     // the pipeline as a raw event and is redacted downstream.
                     pending.push((
@@ -130,7 +178,7 @@ impl RunEventSink {
                         PendingEvent::Raw(transform(event.into_run_event())),
                     ));
                 }
-                Self::Composite(sinks) => {
+                SinkKind::Composite(sinks) => {
                     // Redact once and share the result with every branch.
                     let redacted = event.into_redacted(redactor)?;
                     for sink in sinks.iter().rev() {
@@ -186,14 +234,14 @@ pub struct RunEventLogger {
 
 impl RunEventLogger {
     #[must_use]
-    pub fn new(sink: RunEventSink, redactor: SecretRedactor) -> Self {
+    pub fn new(sink: RunEventSink) -> Self {
         let (tx, mut rx) = mpsc::unbounded_channel();
 
         tokio::spawn(async move {
             while let Some(command) = rx.recv().await {
                 match command {
                     RunEventCommand::Event(event) => {
-                        if let Err(err) = sink.write_run_event(&event, &redactor).await {
+                        if let Err(err) = sink.write_run_event(&event).await {
                             tracing::warn!(error = %err, "Failed to write run event");
                         }
                     }
@@ -235,9 +283,9 @@ pub struct StoreProgressLogger {
 
 impl StoreProgressLogger {
     #[must_use]
-    pub fn new(run_store: impl Into<RunStoreHandle>, redactor: SecretRedactor) -> Self {
+    pub fn new(run_store: impl Into<RunStoreHandle>) -> Self {
         Self {
-            inner: RunEventLogger::new(RunEventSink::backend(run_store.into()), redactor),
+            inner: RunEventLogger::new(RunEventSink::backend(run_store.into())),
         }
     }
 
@@ -326,9 +374,8 @@ mod tests {
         let (writer, reader) = tokio::io::duplex(4096);
         let sink = RunEventSink::json_lines(writer);
         let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None });
-        let redactor = SecretRedactor::default();
 
-        sink.write_run_event(&event, &redactor).await.unwrap();
+        sink.write_run_event(&event).await.unwrap();
 
         let mut reader = BufReader::new(reader);
         let mut line = String::new();
@@ -368,9 +415,8 @@ mod tests {
             ]),
         );
         let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None });
-        let redactor = SecretRedactor::default();
 
-        sink.write_run_event(&event, &redactor).await.unwrap();
+        sink.write_run_event(&event).await.unwrap();
 
         let first = first.lock().await;
         let second = second.lock().await;
@@ -386,7 +432,7 @@ mod tests {
 
         let (writer, reader) = tokio::io::duplex(4096);
         let sink = RunEventSink::json_lines(writer);
-        let logger = RunEventLogger::new(sink, SecretRedactor::default());
+        let logger = RunEventLogger::new(sink);
         let emitter = Emitter::new(fixtures::RUN_8);
         logger.register(&emitter);
 
@@ -418,10 +464,9 @@ mod tests {
             message:          "deploy to staging".to_string(),
             exec_output_tail: None,
         });
-        let redactor = SecretRedactor::default();
-        redactor.register("staging");
+        sink.secret_redactor().register("staging");
 
-        sink.write_run_event(&event, &redactor).await.unwrap();
+        sink.write_run_event(&event).await.unwrap();
 
         let events = received.lock().await;
         let text = serde_json::to_string(&events[0].to_value().unwrap()).unwrap();
diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs
index 50dd32067..16dbe0230 100644
--- a/lib/crates/fabro-workflow/src/handler/agent.rs
+++ b/lib/crates/fabro-workflow/src/handler/agent.rs
@@ -461,10 +461,7 @@ mod tests {
             .run
             .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1)))
             .with_run_store(run_store.clone().into());
-        let logger = crate::event::StoreProgressLogger::new(
-            run_store.clone(),
-            fabro_redact::SecretRedactor::default(),
-        );
+        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
         logger.register(services.run.emitter.as_ref());
         (services, run_store, logger)
     }
diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs
index a22b8f6bc..5bda4282c 100644
--- a/lib/crates/fabro-workflow/src/handler/command.rs
+++ b/lib/crates/fabro-workflow/src/handler/command.rs
@@ -331,10 +331,7 @@ mod tests {
             .run
             .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1)))
             .with_run_store(run_store.clone().into());
-        let logger = crate::event::StoreProgressLogger::new(
-            run_store.clone(),
-            fabro_redact::SecretRedactor::default(),
-        );
+        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
         logger.register(services.run.emitter.as_ref());
         (services, run_store, logger)
     }
diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs
index ffe92961c..22aa2aa9c 100644
--- a/lib/crates/fabro-workflow/src/handler/parallel.rs
+++ b/lib/crates/fabro-workflow/src/handler/parallel.rs
@@ -783,10 +783,7 @@ mod tests {
             .run
             .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1)))
             .with_run_store(run_store.clone().into());
-        let logger = crate::event::StoreProgressLogger::new(
-            run_store.clone(),
-            fabro_redact::SecretRedactor::default(),
-        );
+        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
         logger.register(services.run.emitter.as_ref());
         let mut node = Node::new("par");
         node.attrs.insert(
@@ -839,10 +836,7 @@ mod tests {
             .run
             .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1)))
             .with_run_store(run_store.clone().into());
-        let logger = crate::event::StoreProgressLogger::new(
-            run_store.clone(),
-            fabro_redact::SecretRedactor::default(),
-        );
+        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
         logger.register(services.run.emitter.as_ref());
         let mut node = Node::new("par");
         node.attrs.insert(
diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs
index c56e38470..48cbeeb32 100644
--- a/lib/crates/fabro-workflow/src/handler/prompt.rs
+++ b/lib/crates/fabro-workflow/src/handler/prompt.rs
@@ -260,10 +260,7 @@ mod tests {
             .run
             .with_emitter(Arc::new(Emitter::new(fixtures::RUN_1)))
             .with_run_store(run_store.clone().into());
-        let logger = crate::event::StoreProgressLogger::new(
-            run_store.clone(),
-            fabro_redact::SecretRedactor::default(),
-        );
+        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
         logger.register(services.run.emitter.as_ref());
         (services, run_store, logger)
     }
diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs
index 47dcd9490..beaad87a9 100644
--- a/lib/crates/fabro-workflow/src/operations/resume.rs
+++ b/lib/crates/fabro-workflow/src/operations/resume.rs
@@ -1,7 +1,5 @@
 use std::path::Path;
 
-use fabro_redact::SecretRedactor;
-
 use super::start::{StartServices, Started, execute_persisted_run};
 use crate::error::Error;
 use crate::event::{Event, append_event_to_sink};
@@ -41,14 +39,10 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
     let definition_blob = state.spec.definition_blob;
 
     cleanup_resume_artifacts(run_dir);
-    // Resume submission is a lifecycle marker emitted before run-boundary
-    // interpolation; it cannot contain resolved declared-secret values.
-    let no_run_secret_redactor = SecretRedactor::default();
     append_event_to_sink(
         &services.event_sink,
         &services.run_id,
         &Event::RunSubmitted { definition_blob },
-        &no_run_secret_redactor,
     )
     .await
     .map_err(|err| Error::engine(err.to_string()))?;
diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs
index 720ec5b10..20d2a2791 100644
--- a/lib/crates/fabro-workflow/src/operations/start.rs
+++ b/lib/crates/fabro-workflow/src/operations/start.rs
@@ -82,7 +82,6 @@ struct RunSession {
     vault:             Option<Arc<AsyncRwLock<Vault>>>,
     catalog:           Arc<Catalog>,
     fabro_run_tools:   Option<FabroRunToolServices>,
-    secret_redactor:   SecretRedactor,
 }
 
 struct ResolvedStartLlm {
@@ -147,9 +146,6 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
         )));
     }
     if matches!(status, RunStatus::Submitted) {
-        // These lifecycle request events are emitted before run-boundary secret
-        // interpolation and carry no user free-form output.
-        let no_run_secret_redactor = SecretRedactor::default();
         append_event_to_sink(
             &services.event_sink,
             &services.run_id,
@@ -157,7 +153,6 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
                 resume: false,
                 actor:  None,
             },
-            &no_run_secret_redactor,
         )
         .await
         .map_err(|err| Error::engine(err.to_string()))?;
@@ -168,7 +163,6 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
                 source: RunRunnableSource::StartRequested,
                 actor:  None,
             },
-            &no_run_secret_redactor,
         )
         .await
         .map_err(|err| Error::engine(err.to_string()))?;
@@ -186,9 +180,6 @@ pub(super) async fn execute_persisted_run(
     let run_id = services.run_id;
     let run_store = services.run_store.clone();
     let event_sink = services.event_sink.clone();
-    // Bootstrap events happen before run-boundary secret interpolation, so no
-    // resolved declared-secret value can reach this surface.
-    let bootstrap_redactor = SecretRedactor::default();
     if let Err(err) = run_store.state().await {
         let error = Error::engine(err.to_string());
         let _ = persist_detached_failure(
@@ -199,19 +190,11 @@ pub(super) async fn execute_persisted_run(
             "bootstrap",
             FailureReason::BootstrapFailed,
             &error,
-            &bootstrap_redactor,
         )
         .await;
         return Err(error);
     }
-    if let Err(err) = append_event_to_sink(
-        &event_sink,
-        &run_id,
-        &Event::RunStarting,
-        &bootstrap_redactor,
-    )
-    .await
-    {
+    if let Err(err) = append_event_to_sink(&event_sink, &run_id, &Event::RunStarting).await {
         let error = Error::engine(err.to_string());
         let _ = persist_detached_failure(
             run_id,
@@ -221,7 +204,6 @@ pub(super) async fn execute_persisted_run(
             "bootstrap",
             FailureReason::BootstrapFailed,
             &error,
-            &bootstrap_redactor,
         )
         .await;
         return Err(error);
@@ -245,7 +227,6 @@ pub(super) async fn execute_persisted_run(
                 "bootstrap",
                 FailureReason::BootstrapFailed,
                 &err,
-                &bootstrap_redactor,
             )
             .await;
             bootstrap_guard.defuse();
@@ -264,7 +245,6 @@ pub(super) async fn execute_persisted_run(
                 "bootstrap",
                 FailureReason::BootstrapFailed,
                 &err,
-                &bootstrap_redactor,
             )
             .await;
             bootstrap_guard.defuse();
@@ -273,13 +253,11 @@ pub(super) async fn execute_persisted_run(
     };
 
     bootstrap_guard.defuse();
-    let run_secret_redactor = session.secret_redactor.clone();
     let mut completion_guard = DetachedRunCompletionGuard::arm(
         run_id,
         run_store.clone(),
         event_sink.clone(),
         cancel_token,
-        run_secret_redactor.clone(),
     );
     let run_start = Instant::now();
     let started = Box::pin(session.run(persisted, checkpoint)).await;
@@ -297,7 +275,6 @@ pub(super) async fn execute_persisted_run(
                 run_dir,
                 &err,
                 run_start.elapsed(),
-                &run_secret_redactor,
             )
             .await;
             completion_guard.defuse();
@@ -316,7 +293,6 @@ async fn emit_workflow_run_failed(
     error: &Error,
     reason: FailureReason,
     wall_duration_ms: u64,
-    redactor: &SecretRedactor,
 ) {
     let failure = Some(error::run_failure_from_error(error, reason));
     let conclusion = build_conclusion_from_store(
@@ -338,7 +314,7 @@ async fn emit_workflow_run_failed(
         None,
         conclusion.billing,
     );
-    if let Err(err) = append_event_to_sink(event_sink, &run_id, &failure_event, redactor).await {
+    if let Err(err) = append_event_to_sink(event_sink, &run_id, &failure_event).await {
         tracing::warn!(error = %err, "Failed to append run.failed event");
     }
 }
@@ -350,7 +326,6 @@ async fn persist_terminal_engine_failure(
     _run_dir: &Path,
     error: &Error,
     duration: Duration,
-    redactor: &SecretRedactor,
 ) {
     let engine_result: Result<Outcome, Error> = Err(error.clone());
     let (_, _, run_status) = classify_engine_result(&engine_result);
@@ -365,7 +340,6 @@ async fn persist_terminal_engine_failure(
         error,
         reason,
         crate::millis_u64(duration),
-        redactor,
     )
     .await;
 }
@@ -400,7 +374,10 @@ impl RunSession {
         let configured =
             configured_providers_for_start(services.vault.as_ref(), Arc::clone(&catalog)).await;
         let llm = resolve_start_llm(catalog.as_ref(), &configured, resolved)?;
-        let secret_redactor = SecretRedactor::default();
+        // Register resolved secrets into the event sink's registry so every
+        // event surface — including failure paths that outlive this session or
+        // run when its construction fails partway — redacts them.
+        let secret_redactor = services.event_sink.secret_redactor().clone();
         let vault_guard = match services.vault.as_ref() {
             Some(vault) => Some(vault.read().await),
             None => None,
@@ -537,7 +514,6 @@ impl RunSession {
             vault: services.vault,
             catalog,
             fabro_run_tools: services.fabro_run_tools,
-            secret_redactor,
         })
     }
 }
@@ -868,8 +844,7 @@ impl RunSession {
             });
         }
 
-        let store_progress_logger =
-            RunEventLogger::new(self.event_sink.clone(), self.secret_redactor.clone());
+        let store_progress_logger = RunEventLogger::new(self.event_sink.clone());
         store_progress_logger.register(self.emitter.as_ref());
 
         let init_options = InitOptions {
@@ -896,7 +871,7 @@ impl RunSession {
             checkpoint,
             seed_context: self.seed_context,
             fabro_run_tools: self.fabro_run_tools,
-            secret_redactor: self.secret_redactor.clone(),
+            secret_redactor: self.event_sink.secret_redactor().clone(),
         };
         let mut initialized = match Box::pin(pipeline::initialize(persisted, init_options)).await {
             Ok(initialized) => initialized,
@@ -1020,9 +995,6 @@ impl Drop for DetachedRunBootstrapGuard {
         let event_sink = self.event_sink.clone();
         if let Ok(handle) = Handle::try_current() {
             handle.spawn(async move {
-                // Bootstrap drop failures happen before any run-boundary
-                // declared secret can be resolved into output.
-                let redactor = SecretRedactor::default();
                 emit_workflow_run_failed(
                     run_id,
                     &run_store,
@@ -1030,7 +1002,6 @@ impl Drop for DetachedRunBootstrapGuard {
                     &Error::engine(reason.to_string()),
                     reason,
                     0,
-                    &redactor,
                 )
                 .await;
             });
@@ -1060,12 +1031,11 @@ async fn run_store_reaches_terminal(run_store: &RunStoreHandle, timeout: Duratio
 }
 
 struct DetachedRunCompletionGuard {
-    event_sink:      RunEventSink,
-    run_id:          RunId,
-    run_store:       RunStoreHandle,
-    cancel_token:    CancellationToken,
-    secret_redactor: SecretRedactor,
-    active:          bool,
+    event_sink:   RunEventSink,
+    run_id:       RunId,
+    run_store:    RunStoreHandle,
+    cancel_token: CancellationToken,
+    active:       bool,
 }
 
 impl DetachedRunCompletionGuard {
@@ -1074,14 +1044,12 @@ impl DetachedRunCompletionGuard {
         run_store: RunStoreHandle,
         event_sink: RunEventSink,
         cancel_token: CancellationToken,
-        secret_redactor: SecretRedactor,
     ) -> Self {
         Self {
             event_sink,
             run_id,
             run_store,
             cancel_token,
-            secret_redactor,
             active: true,
         }
     }
@@ -1116,7 +1084,6 @@ impl Drop for DetachedRunCompletionGuard {
         let event_sink = self.event_sink.clone();
         let run_id = self.run_id;
         let run_store = self.run_store.clone();
-        let secret_redactor = self.secret_redactor.clone();
         if let Ok(handle) = Handle::try_current() {
             handle.spawn(async move {
                 if run_store_reaches_terminal(&run_store, DETACHED_COMPLETION_GUARD_TERMINAL_GRACE)
@@ -1131,20 +1098,14 @@ impl Drop for DetachedRunCompletionGuard {
                     &Error::engine(message.to_string()),
                     reason,
                     0,
-                    &secret_redactor,
                 )
                 .await;
-                let _ = append_event_to_sink(
-                    &event_sink,
-                    &run_id,
-                    &Event::RunNotice {
-                        level:            RunNoticeLevel::Error,
-                        code:             code.to_string(),
-                        message:          message.to_string(),
-                        exec_output_tail: None,
-                    },
-                    &secret_redactor,
-                )
+                let _ = append_event_to_sink(&event_sink, &run_id, &Event::RunNotice {
+                    level:            RunNoticeLevel::Error,
+                    code:             code.to_string(),
+                    message:          message.to_string(),
+                    exec_output_tail: None,
+                })
                 .await;
             });
         }
@@ -1159,9 +1120,8 @@ async fn persist_detached_failure(
     phase: &'static str,
     reason: FailureReason,
     error: &Error,
-    redactor: &SecretRedactor,
 ) -> Result<(), Error> {
-    emit_workflow_run_failed(run_id, run_store, event_sink, error, reason, 0, redactor).await;
+    emit_workflow_run_failed(run_id, run_store, event_sink, error, reason, 0).await;
 
     let event = Event::RunNotice {
         level:            RunNoticeLevel::Error,
@@ -1169,7 +1129,7 @@ async fn persist_detached_failure(
         message:          error.to_string(),
         exec_output_tail: None,
     };
-    if let Err(err) = append_event_to_sink(event_sink, &run_id, &event, redactor).await {
+    if let Err(err) = append_event_to_sink(event_sink, &run_id, &event).await {
         tracing::warn!(error = %err, "Failed to append detached failure notice");
     }
 
@@ -1672,17 +1632,12 @@ reasoning = false
         .await
         .unwrap();
 
-        append_event_to_sink(
-            &session.event_sink,
-            &fixtures::RUN_1,
-            &Event::RunNotice {
-                level:            fabro_types::RunNoticeLevel::Warn,
-                code:             "mcp_ready".to_string(),
-                message:          "MCP reported staging before stages".to_string(),
-                exec_output_tail: None,
-            },
-            &session.secret_redactor,
-        )
+        append_event_to_sink(&session.event_sink, &fixtures::RUN_1, &Event::RunNotice {
+            level:            fabro_types::RunNoticeLevel::Warn,
+            code:             "mcp_ready".to_string(),
+            message:          "MCP reported staging before stages".to_string(),
+            exec_output_tail: None,
+        })
         .await
         .unwrap();
 
@@ -1775,10 +1730,12 @@ reasoning = false
         let production_session = session_with_secret("production").await;
 
         let staging_text = staging_session
-            .secret_redactor
+            .event_sink
+            .secret_redactor()
             .redact_into("staging production");
         let production_text = production_session
-            .secret_redactor
+            .event_sink
+            .secret_redactor()
             .redact_into("staging production");
 
         assert!(!staging_text.contains("staging"));
@@ -1787,6 +1744,69 @@ reasoning = false
         assert!(!production_text.contains("production"));
     }
 
+    #[tokio::test]
+    async fn failed_session_construction_keeps_registered_secrets_on_event_sink() {
+        let temp = tempfile::tempdir().unwrap();
+        let (storage_root, _run_dir) = storage_root_and_run_dir(&temp);
+        let mut settings = settings_from_run_layer(RunLayer {
+            execution: Some(RunExecutionLayer {
+                mode: Some(RunMode::DryRun),
+                ..RunExecutionLayer::default()
+            }),
+            ..RunLayer::default()
+        });
+        // MCP env resolves (and registers) the declared secret; the run
+        // environment then references a missing secret, so session
+        // construction fails after partial secret resolution.
+        settings.run.agent.mcps.insert(
+            "vaulted".to_string(),
+            ResolvedMcpEntry::Resolved(ResolvedMcpServerSettings {
+                name: "vaulted".to_string(),
+                transport: ResolvedMcpTransport::Stdio {
+                    command: vec!["mcp-server".to_string()],
+                    env:     HashMap::from([(
+                        "MCP_TOKEN".to_string(),
+                        "{{ secrets.MCP_TOKEN }}".to_string(),
+                    )]),
+                },
+                ..ResolvedMcpServerSettings::default()
+            }),
+        );
+        settings.run.environment.env.insert(
+            "DEPLOY_ENV".to_string(),
+            InterpString::parse("{{ secrets.MISSING }}"),
+        );
+        let (persisted, store) =
+            persisted_workflow_with_settings(MINIMAL_DOT, &storage_root, settings).await;
+        let emitter = Arc::new(Emitter::new(fixtures::RUN_1));
+        let registry = Arc::new(test_registry());
+        let vault = Arc::new(AsyncRwLock::new(token_vault("MCP_TOKEN", "staging")));
+
+        let services = StartServices {
+            vault: Some(vault),
+            ..test_start_services(&store, &storage_root, emitter, registry).await
+        };
+        let event_sink = services.event_sink.clone();
+
+        assert!(RunSession::new(&persisted, services).await.is_err());
+
+        // Bootstrap failure paths write through this same sink; values
+        // registered before the construction failure must still redact.
+        append_event_to_sink(&event_sink, &fixtures::RUN_1, &Event::RunNotice {
+            level:            fabro_types::RunNoticeLevel::Error,
+            code:             "bootstrap_failed".to_string(),
+            message:          "failed after resolving staging".to_string(),
+            exec_output_tail: None,
+        })
+        .await
+        .unwrap();
+
+        let run_store = store.open_run(&fixtures::RUN_1).await.unwrap();
+        let serialized = serde_json::to_string(&run_store.list_events().await.unwrap()).unwrap();
+        assert!(!serialized.contains("staging"));
+        assert!(serialized.contains("REDACTED"));
+    }
+
     #[tokio::test]
     async fn run_session_new_missing_secret_fails_startup() {
         let temp = tempfile::tempdir().unwrap();
@@ -2209,7 +2229,6 @@ reasoning = false
             &run_dir,
             &Error::engine("visit limit exceeded"),
             Duration::from_millis(9_999),
-            &SecretRedactor::default(),
         )
         .await;
 
@@ -2292,7 +2311,6 @@ reasoning = false
                 run_store_handle,
                 event_sink,
                 CancellationToken::new(),
-                SecretRedactor::default(),
             );
         }
 
diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
index 793a444cc..951b5eec8 100644
--- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
@@ -252,7 +252,7 @@ async fn execute_test_run_with_options(
     let run_store = test_run_store(&run_id_value).await;
     seed_created_and_starting(&run_store, &run_options, &graph).await;
     let emitter = test_emitter_arc("test-run");
-    let store_logger = StoreProgressLogger::new(run_store.clone(), SecretRedactor::default());
+    let store_logger = StoreProgressLogger::new(run_store.clone());
     store_logger.register(&emitter);
     let initialized = initialize(
         persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value),
diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs
index c6a61ec6d..587525bfd 100644
--- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs
@@ -1035,8 +1035,7 @@ mod tests {
         let inner_store = test_store().create_run(&test_run_id()).await.unwrap();
         let run_store = inner_store;
         let emitter = Arc::new(Emitter::new(test_run_id()));
-        let store_logger =
-            StoreProgressLogger::new(run_store.clone(), fabro_redact::SecretRedactor::default());
+        let store_logger = StoreProgressLogger::new(run_store.clone());
         store_logger.register(&emitter);
         let sandbox: Arc<dyn fabro_agent::Sandbox> = Arc::new(fabro_agent::LocalSandbox::new(
             std::env::current_dir().unwrap(),
diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs
index 636c054bd..e7b4b6156 100644
--- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs
@@ -1196,7 +1196,7 @@ mod tests {
         let emitter = Arc::new(crate::event::Emitter::new(test_run_id()));
         let store = memory_store();
         let run_store = store.create_run(&test_run_id()).await.unwrap();
-        let store_logger = StoreProgressLogger::new(run_store.clone(), SecretRedactor::default());
+        let store_logger = StoreProgressLogger::new(run_store.clone());
         let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
         emitter.on_event({
             let seen = Arc::clone(&seen);
diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs
index ba3a731d1..1acf410a4 100644
--- a/lib/crates/fabro-workflow/src/test_support.rs
+++ b/lib/crates/fabro-workflow/src/test_support.rs
@@ -201,8 +201,7 @@ async fn initialized(
         .await
         .expect("failed to seed run.starting event in run store");
     let emitter = bound_emitter(run_options.run_id, &emitter);
-    let store_logger =
-        StoreProgressLogger::new(run_store.clone(), fabro_redact::SecretRedactor::default());
+    let store_logger = StoreProgressLogger::new(run_store.clone());
     store_logger.register(emitter.as_ref());
     let artifact_store = ArtifactStore::new(
         Arc::new(