fabro(01KX1P0VV0DAQTT0N2NADX8J8J): simplify_fable (succeeded)

Fabro-Run: 01KX1P0VV0DAQTT0N2NADX8J8J
Fabro-Completed: 6

⚒️ Generated with [Fabro](https://fabro.sh)

lib/crates/fabro-cli/src/commands/run/runner.rs

@@ -26,7 +26,7 @@ use fabro_types::{
 };
 use fabro_vault::Vault;
 use fabro_workflow::artifact_upload::{ArtifactSink, StageArtifactUploader};
-use fabro_workflow::event::{Emitter, RunEventSink};
+use fabro_workflow::event::{Emitter, RedactedRunEvent, RunEventSink};
 use fabro_workflow::operations::{self, StartServices};
 use fabro_workflow::run_control::RunControlState;
 use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle};
@@ -178,12 +178,12 @@ pub(crate) async fn execute(
         fabro_run_tools,
     };
 
-    let execution = Box::pin(async {
+    let execution = async {
         match mode {
             RunWorkerMode::Start => operations::start(&run_dir, services).await,
             RunWorkerMode::Resume => operations::resume(&run_dir, services).await,
         }
-    });
+    };
 
     if let Some(mut control_manager) = control_manager {
         tokio::select! {
@@ -1005,7 +1005,8 @@ impl RunStoreBackend for HttpRunStore {
         Ok(events)
     }
 
-    async fn append_run_event(&self, event: &RunEvent) -> Result<()> {
+    async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()> {
+        let event = event.event();
         let seq = Box::pin(self.with_retries("append run event", || {
             let client = self.client.clone_for_reuse();
             let run_id = self.run_id;

lib/crates/fabro-redact/src/secret_registry.rs

@@ -40,9 +40,10 @@ impl SecretRedactor {
 
     /// Redact all registered secret values from `s`.
     pub fn redact_into(&self, s: &str) -> String {
-        let Some(values) = self.values_snapshot() else {
+        let values = self.read();
+        if values.is_empty() {
             return s.to_string();
-        };
+        }
         redact_string_values(s, &values)
     }
 
@@ -50,11 +51,10 @@ impl SecretRedactor {
     ///
     /// Object keys and non-string values are left unchanged.
     pub fn redact_json(&self, mut value: Value) -> Value {
-        let Some(values) = self.values_snapshot() else {
-            return value;
-        };
-
-        redact_json_leaves(&mut value, &values);
+        let values = self.read();
+        if !values.is_empty() {
+            redact_json_leaves(&mut value, &values);
+        }
         value
     }
 
@@ -65,14 +65,6 @@ impl SecretRedactor {
     fn write(&self) -> RwLockWriteGuard<'_, Vec<String>> {
         self.values.write().unwrap_or_else(PoisonError::into_inner)
     }
-
-    fn values_snapshot(&self) -> Option<Vec<String>> {
-        let values = self.read();
-        if values.is_empty() {
-            return None;
-        }
-        Some(values.clone())
-    }
 }
 
 fn redact_json_leaves(value: &mut Value, values: &[String]) {

lib/crates/fabro-server/src/server/handler/events.rs

@@ -160,33 +160,26 @@ async fn append_run_event(
     if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
         return response;
     }
-    let event = match RunEvent::from_value(value.clone()) {
-        Ok(event) => event,
-        Err(err) => {
-            return ApiError::bad_request(format!("Invalid run event: {err}")).into_response();
-        }
-    };
-    if event.run_id != id {
-        return ApiError::bad_request("Event run_id does not match path run ID.").into_response();
-    }
-    if let Some(denied) = denied_lifecycle_event_name(&event.body) {
-        return ApiError::bad_request(format!(
-            "{denied} is a lifecycle event; clients must call the corresponding operation endpoint instead of injecting it via append_run_event"
-        ))
-        .into_response();
-    }
-    let redacted_value = redact_json_value(value);
-    let payload = match EventPayload::new(redacted_value, &id) {
+    // Ingest trust boundary: content-redact before validation and storage so
+    // credential-shaped values from arbitrary clients never reach the store.
+    // `EventPayload::new` validates required fields and the run_id match, and
+    // the single parse below feeds both storage checks and the live-run update.
+    let payload = match EventPayload::new(redact_json_value(value), &id) {
         Ok(payload) => payload,
         Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
     };
     let event = match RunEvent::from_ref(payload.as_value()) {
         Ok(event) => event,
         Err(err) => {
-            return ApiError::bad_request(format!("Invalid redacted run event: {err}"))
-                .into_response();
+            return ApiError::bad_request(format!("Invalid run event: {err}")).into_response();
         }
     };
+    if let Some(denied) = denied_lifecycle_event_name(&event.body) {
+        return ApiError::bad_request(format!(
+            "{denied} is a lifecycle event; clients must call the corresponding operation endpoint instead of injecting it via append_run_event"
+        ))
+        .into_response();
+    }
 
     match state.stores.runs.open_run(&id).await {
         Ok(run_store) => match run_store.append_event(&payload).await {

lib/crates/fabro-workflow/src/event.rs

@@ -15,7 +15,7 @@ pub use self::emitter::Emitter;
 pub use self::events::Event;
 pub use self::names::event_name;
 pub use self::redaction::{
-    build_redacted_event_payload, event_payload_from_redacted_json, redacted_event_json,
+    RedactedRunEvent, build_redacted_event_payload, event_payload_from_redacted_json,
 };
 pub use self::sink::{
     RunEventLogger, RunEventSink, StoreProgressLogger, append_event, append_event_to_sink,

lib/crates/fabro-workflow/src/event/redaction.rs

@@ -14,13 +14,36 @@ pub fn build_redacted_event_payload(
     EventPayload::new(value, run_id).map_err(anyhow::Error::from)
 }
 
-pub fn redacted_event_json(event: &RunEvent, redactor: &SecretRedactor) -> Result<String> {
-    serde_json::to_string(&redacted_event_value(event, redactor)?).map_err(anyhow::Error::from)
+/// A run event passed through the full redaction pipeline exactly once.
+///
+/// The redacted JSON value feeds payload and JSON-line surfaces, and the typed
+/// event is reparsed from that same value, so every downstream consumer
+/// observes identical redacted data without re-running the redaction passes.
+pub struct RedactedRunEvent {
+    event: RunEvent,
+    value: Value,
 }
 
-pub(super) fn redacted_run_event(event: &RunEvent, redactor: &SecretRedactor) -> Result<RunEvent> {
-    let value = redacted_event_value(event, redactor)?;
-    RunEvent::from_ref(&value).context("Failed to reparse redacted event payload")
+impl RedactedRunEvent {
+    pub fn new(event: &RunEvent, redactor: &SecretRedactor) -> Result<Self> {
+        let value = redacted_event_value(event, redactor)?;
+        let event =
+            RunEvent::from_ref(&value).context("Failed to reparse redacted event payload")?;
+        Ok(Self { event, value })
+    }
+
+    #[must_use]
+    pub fn event(&self) -> &RunEvent {
+        &self.event
+    }
+
+    pub fn payload(&self) -> Result<EventPayload> {
+        EventPayload::new(self.value.clone(), &self.event.run_id).map_err(anyhow::Error::from)
+    }
+
+    pub fn json_line(&self) -> Result<String> {
+        serde_json::to_string(&self.value).map_err(anyhow::Error::from)
+    }
 }
 
 fn normalized_event_value(event: &RunEvent) -> Result<Value> {
@@ -40,10 +63,7 @@ fn redact_registered_secrets_in_event_value(value: &mut Value, redactor: &Secret
     }
 
     if let Some(Value::String(node_label)) = value.get_mut("node_label") {
-        let redacted = redactor.redact_into(node_label);
-        if redacted != *node_label {
-            *node_label = redacted;
-        }
+        *node_label = redactor.redact_into(node_label);
     }
 
     if let Some(properties) = value.get_mut("properties") {
@@ -271,7 +291,10 @@ mod tests {
             fabro_redact::redact_json_value(normalized_event_value(&stored).unwrap());
         let content_only_json = serde_json::to_string(&content_only).unwrap();
 
-        let with_empty_registry = redacted_event_json(&stored, &SecretRedactor::default()).unwrap();
+        let with_empty_registry = RedactedRunEvent::new(&stored, &SecretRedactor::default())
+            .unwrap()
+            .json_line()
+            .unwrap();
 
         assert_eq!(with_empty_registry, content_only_json);
     }

lib/crates/fabro-workflow/src/event/sink.rs

@@ -10,7 +10,7 @@ use tokio::io::{AsyncWrite, AsyncWriteExt};
 use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot};
 
 use super::emitter::Emitter;
-use super::redaction::{build_redacted_event_payload, redacted_event_json, redacted_run_event};
+use super::redaction::{RedactedRunEvent, build_redacted_event_payload};
 use super::{Event, to_run_event};
 use crate::runtime_store::RunStoreHandle;
 
@@ -105,30 +105,36 @@ impl RunEventSink {
     }
 
     pub async fn write_run_event(&self, event: &RunEvent, redactor: &SecretRedactor) -> Result<()> {
-        let mut pending = vec![(self, event.clone())];
+        let mut pending = vec![(self, PendingEvent::Raw(event.clone()))];
         while let Some((sink, event)) = pending.pop() {
             match sink {
                 Self::Store(run_store) => {
-                    let event = redacted_run_event(&event, redactor)?;
-                    run_store.append_run_event(&event).await?;
+                    let redacted = event.into_redacted(redactor)?;
+                    run_store.append_run_event(&redacted).await?;
                 }
                 Self::JsonLines(writer) => {
-                    let line = redacted_event_json(&event, redactor)?;
+                    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) => {
-                    let event = redacted_run_event(&event, redactor)?;
-                    callback(event).await?;
+                    callback(event.into_redacted(redactor)?.event().clone()).await?;
                 }
                 Self::Map { transform, inner } => {
-                    pending.push((inner.as_ref(), transform(event)));
+                    // A transform can add new fields, so its output re-enters
+                    // the pipeline as a raw event and is redacted downstream.
+                    pending.push((
+                        inner.as_ref(),
+                        PendingEvent::Raw(transform(event.into_run_event())),
+                    ));
                 }
                 Self::Composite(sinks) => {
+                    // Redact once and share the result with every branch.
+                    let redacted = event.into_redacted(redactor)?;
                     for sink in sinks.iter().rev() {
-                        pending.push((sink, event.clone()));
+                        pending.push((sink, PendingEvent::Redacted(Arc::clone(&redacted))));
                     }
                 }
             }
@@ -137,6 +143,33 @@ impl RunEventSink {
     }
 }
 
+/// Worklist entry for [`RunEventSink::write_run_event`]: events are redacted at
+/// most once per fan-out and the shared result reused by every consumer.
+#[allow(
+    clippy::large_enum_variant,
+    reason = "Worklist entries stay inline to avoid boxing hot-path payloads."
+)]
+enum PendingEvent {
+    Raw(RunEvent),
+    Redacted(Arc<RedactedRunEvent>),
+}
+
+impl PendingEvent {
+    fn into_redacted(self, redactor: &SecretRedactor) -> Result<Arc<RedactedRunEvent>> {
+        match self {
+            Self::Raw(event) => Ok(Arc::new(RedactedRunEvent::new(&event, redactor)?)),
+            Self::Redacted(redacted) => Ok(redacted),
+        }
+    }
+
+    fn into_run_event(self) -> RunEvent {
+        match self {
+            Self::Raw(event) => event,
+            Self::Redacted(redacted) => redacted.event().clone(),
+        }
+    }
+}
+
 #[allow(
     clippy::large_enum_variant,
     reason = "Logger queue messages stay inline to avoid boxing hot-path payloads."
@@ -155,13 +188,12 @@ impl RunEventLogger {
     #[must_use]
     pub fn new(sink: RunEventSink, redactor: SecretRedactor) -> Self {
         let (tx, mut rx) = mpsc::unbounded_channel();
-        let logger_redactor = redactor;
 
         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, &logger_redactor).await {
+                        if let Err(err) = sink.write_run_event(&event, &redactor).await {
                             tracing::warn!(error = %err, "Failed to write run event");
                         }
                     }

lib/crates/fabro-workflow/src/handler/command.rs

@@ -263,7 +263,10 @@ mod tests {
             Ok(Vec::new())
         }
 
-        async fn append_run_event(&self, _event: &fabro_types::RunEvent) -> anyhow::Result<()> {
+        async fn append_run_event(
+            &self,
+            _event: &crate::event::RedactedRunEvent,
+        ) -> anyhow::Result<()> {
             Ok(())
         }
 

lib/crates/fabro-workflow/src/lifecycle/git.rs

@@ -1305,7 +1305,7 @@ mod tests {
             Ok(Vec::new())
         }
 
-        async fn append_run_event(&self, _event: &RunEvent) -> Result<()> {
+        async fn append_run_event(&self, _event: &crate::event::RedactedRunEvent) -> Result<()> {
             Ok(())
         }
 

lib/crates/fabro-workflow/src/pipeline/finalize.rs

@@ -1416,7 +1416,7 @@ mod tests {
             Ok(Vec::new())
         }
 
-        async fn append_run_event(&self, _event: &RunEvent) -> Result<()> {
+        async fn append_run_event(&self, _event: &crate::event::RedactedRunEvent) -> Result<()> {
             Ok(())
         }
 

lib/crates/fabro-workflow/src/runtime_store.rs

@@ -3,17 +3,19 @@ use std::sync::Arc;
 use anyhow::Result;
 use async_trait::async_trait;
 use bytes::Bytes;
-use fabro_redact::SecretRedactor;
 use fabro_store::{EventEnvelope, RunDatabase, RunProjection};
-use fabro_types::{RunBlobId, RunEvent};
+use fabro_types::RunBlobId;
 
-use crate::event::build_redacted_event_payload;
+use crate::event::RedactedRunEvent;
 
 #[async_trait]
 pub trait RunStoreBackend: Send + Sync {
     async fn load_state(&self) -> Result<RunProjection>;
     async fn list_events(&self) -> Result<Vec<EventEnvelope>>;
-    async fn append_run_event(&self, event: &RunEvent) -> Result<()>;
+    /// Append an event that already passed the redaction pipeline. Taking
+    /// [`RedactedRunEvent`] keeps unredacted events out of every backend and
+    /// lets backends reuse the single redaction pass instead of re-running it.
+    async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()>;
     async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId>;
     async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>>;
     async fn read_run_log(&self) -> Result<Option<Vec<u8>>>;
@@ -43,7 +45,7 @@ impl RunStoreHandle {
         self.backend.list_events().await
     }
 
-    pub async fn append_run_event(&self, event: &RunEvent) -> Result<()> {
+    pub async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()> {
         self.backend.append_run_event(event).await
     }
 
@@ -83,14 +85,9 @@ impl RunStoreBackend for LocalRunStoreBackend {
             .map_err(anyhow::Error::from)
     }
 
-    async fn append_run_event(&self, event: &RunEvent) -> Result<()> {
-        // Local backends can be called from server/test paths that do not have a
-        // worker run registry. RunEventSink applies the per-run redactor before
-        // calling this backend during workflow execution.
-        let payload =
-            build_redacted_event_payload(event, &event.run_id, &SecretRedactor::default())?;
+    async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()> {
         self.run_store
-            .append_event(&payload)
+            .append_event(&event.payload()?)
             .await
             .map(|_| ())
             .map_err(anyhow::Error::from)
@@ -129,7 +126,7 @@ mod tests {
     use object_store::memory::InMemory;
 
     use super::RunStoreHandle;
-    use crate::event::{Event, append_event};
+    use crate::event::{Event, RedactedRunEvent, append_event};
     use crate::records::RunSpec;
 
     async fn test_run_store() -> fabro_store::RunDatabase {
@@ -223,7 +220,9 @@ mod tests {
                 definition_blob: None,
             }),
         };
-        handle.append_run_event(&event).await.unwrap();
+        let redacted =
+            RedactedRunEvent::new(&event, &fabro_redact::SecretRedactor::default()).unwrap();
+        handle.append_run_event(&redacted).await.unwrap();
 
         let blob_id = handle.write_blob(br#"{"ok":true}"#).await.unwrap();
         let blob = handle.read_blob(&blob_id).await.unwrap().unwrap();