scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-12 23:02:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fabro(01KX1P0VV0DAQTT0N2NADX8J8J): implement (succeeded)

Browse source 
Fabro-Run: 01KX1P0VV0DAQTT0N2NADX8J8J
Fabro-Completed: 5
Fabro-Checkpoint: bf54e53eddd458f2c5ba3914efc012a42e7804f6

⚒️ Generated with [Fabro](https://fabro.sh)
This commit is contained in:
Fabro 2026-07-08 21:13:41 +00:00
parent 79bbcb50eb
commit 48fd46c670
21 changed files with 764 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
docs/public/execution/run-configuration.mdx
View file

@ -306,6 +306,30 @@ SERVICE_URL = "https://api.{{ env.REGION }}.example.com"
Missing host variables produce a hard error pointing at the specific field and unresolved token.
Vault secrets use the same interpolation form with `{{ secrets.NAME }}`. Use
secret tokens for values that should be treated as sensitive even when they do
not look like credentials, such as environment names or deployment labels:
```toml title="run.toml"
[environments.ci.env]
DEPLOY_ENV = "{{ secrets.DEPLOY_ENV }}"
[[run.prepare.steps]]
command = ["./deploy", "--environment", "{{ secrets.DEPLOY_ENV }}"]
```
Declared secret values are registered when the run starts and are redacted by
exact match on structured run surfaces: events, `progress.jsonl`, and setup
failure errors. This redaction applies regardless of the value's shape, so a
low-entropy value such as `staging` is still replaced with `REDACTED` in those
surfaces. Fabro also keeps its content-based redaction pass for credential-shaped
values that were not declared with `{{ secrets.NAME }}`.
The boundary is the structured data Fabro captures. Once a secret is passed into
a sandbox process environment, text the sandbox prints is covered where Fabro
captures it back into structured events or setup errors; this is not a blanket
guarantee for every live process stream or file the sandbox writes.
### `[run.integrations.github.permissions]`
Request a scoped GitHub App token for workflow stages that need `GITHUB_TOKEN` inside the sandbox. Values map directly to GitHub App permission names and access levels.

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

@ -178,12 +178,12 @@ pub(crate) async fn execute(
fabro_run_tools,
};
let execution = async {
let execution = Box::pin(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! {
@ -1457,7 +1457,10 @@ mod tests {
);
let event = running_event(None);
sink.write_run_event(&event).await.unwrap();
// This worker-stamp unit test emits no run-declared secret values.
sink.write_run_event(&event, &fabro_redact::SecretRedactor::default())
.await
.unwrap();
let first = first.lock().await;
let second = second.lock().await;

2
lib/crates/fabro-server/src/server.rs
View file

@ -67,7 +67,7 @@ use fabro_llm::types::{
use fabro_mcp_store::McpServerStore;
use fabro_model::catalog::LlmCatalogSettings;
use fabro_model::{BilledTokenCounts, Catalog, ModelRef, ModelTestMode, ProviderId};
use fabro_redact::redact_jsonl_line;
use fabro_redact::{redact_json_value, redact_jsonl_line};
use fabro_sandbox::daytona::{self, DaytonaSandbox};
use fabro_sandbox::details::sandbox_details;
use fabro_sandbox::reconnect::reconnect_for_run;

41
lib/crates/fabro-server/src/server/handler/events.rs
View file

@ -11,8 +11,8 @@ use super::super::{
EventPayload, HashSet, IntoResponse, Json, KeepAlive, PaginatedEventList, PaginationMeta, Path,
Query, RequireRunManagementTarget, RequireRunScoped, RequireRunStageScoped, RequiredUser,
Response, Router, RunEvent, RunId, Sse, State, StatusCode, StreamExt, UnboundedReceiverStream,
broadcast, get, mpsc, parse_run_id_path, parse_stage_id_path, redact_jsonl_line,
reject_if_archived, update_live_run_from_event,
broadcast, get, mpsc, parse_run_id_path, parse_stage_id_path, redact_json_value,
redact_jsonl_line, reject_if_archived, update_live_run_from_event,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
@ -175,10 +175,18 @@ async fn append_run_event(
))
.into_response();
}
let payload = match EventPayload::new(value, &id) {
let redacted_value = redact_json_value(value);
let payload = match EventPayload::new(redacted_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();
}
};
match state.stores.runs.open_run(&id).await {
Ok(run_store) => match run_store.append_event(&payload).await {
@ -352,16 +360,23 @@ fn event_properties(event: &RunEvent) -> serde_json::Map<String, serde_json::Val
}
fn redacted_event_properties(event: &RunEvent) -> serde_json::Map<String, serde_json::Value> {
build_redacted_event_payload(event, &event.run_id)
.ok()
.and_then(|payload| {
payload
.as_value()
.get("properties")
.and_then(serde_json::Value::as_object)
.cloned()
})
.unwrap_or_else(|| event_properties(event))
// Event detail is a read path. Stored events are trusted to have been
// redacted at worker/source and server ingest; no per-run registry is
// available here, so preserve the existing pattern-only projection.
build_redacted_event_payload(
event,
&event.run_id,
&fabro_redact::SecretRedactor::default(),
)
.ok()
.and_then(|payload| {
payload
.as_value()
.get("properties")
.and_then(serde_json::Value::as_object)
.cloned()
})
.unwrap_or_else(|| event_properties(event))
}
fn truncate_content(value: String, max_content_length: usize) -> (String, bool) {

2
lib/crates/fabro-server/src/server/handler/sessions.rs
View file

@ -1346,7 +1346,7 @@ async fn append_run_session_event(
actor: None,
body,
};
let payload = EventPayload::new(event.to_value()?, &run_id)?;
let payload = EventPayload::new(fabro_redact::redact_json_value(event.to_value()?), &run_id)?;
run_store.append_event_envelope(&payload).await
}

46
lib/crates/fabro-server/src/server/tests.rs
View file

@ -4824,7 +4824,12 @@ async fn append_scoped_stage_event(
parallel_branch_id: None,
};
let stored = fabro_workflow::event::to_run_event_at(&run_id, event, Utc::now(), Some(&scope));
let payload = fabro_workflow::event::build_redacted_event_payload(&stored, &run_id).unwrap();
let payload = fabro_workflow::event::build_redacted_event_payload(
&stored,
&run_id,
&fabro_redact::SecretRedactor::default(),
)
.unwrap();
let run_store = state.stores.runs.open_run(&run_id).await.unwrap();
run_store.append_event(&payload).await.unwrap();
}
@ -9616,6 +9621,45 @@ async fn append_run_event_rejects_reserved_archive_event() {
);
}
#[tokio::test]
async fn append_run_event_redacts_credential_shaped_values_before_storage() {
let state = test_app_state();
let app = crate::test_support::build_test_router(Arc::clone(&state));
let run_id = create_run(&app, MINIMAL_DOT)
.await
.parse::<RunId>()
.unwrap();
let secret = "sk-ant-api03-xK9mZ2vL8nQ5rT1wY4bC7dF0gH3jE6pA";
let req = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/events")))
.header("content-type", "application/json")
.body(Body::from(
json!({
"id": "evt-run-notice-redaction",
"ts": "2026-04-19T12:00:00Z",
"run_id": run_id,
"event": "run.notice",
"properties": {
"level": "warn",
"code": "credential_detected",
"message": format!("token={secret}")
}
})
.to_string(),
))
.unwrap();
let response = app.oneshot(req).await.unwrap();
response_json!(response, StatusCode::OK).await;
let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
let serialized = serde_json::to_string(&run_store.list_events().await.unwrap()).unwrap();
assert!(!serialized.contains(secret));
assert!(serialized.contains("REDACTED"));
}
#[tokio::test]
async fn get_checkpoint_returns_null_initially() {
let state = test_app_state();

247
lib/crates/fabro-workflow/src/event/redaction.rs
View file

@ -1,17 +1,26 @@
use ::fabro_types::{RunEvent, RunId};
use anyhow::{Context, Result};
use fabro_redact::redact_json_value;
use fabro_redact::SecretRedactor;
use fabro_store::EventPayload;
use fabro_util::json::normalize_json_value;
use serde_json::Value;
pub fn build_redacted_event_payload(event: &RunEvent, run_id: &RunId) -> Result<EventPayload> {
let value = redacted_event_value(event)?;
pub fn build_redacted_event_payload(
event: &RunEvent,
run_id: &RunId,
redactor: &SecretRedactor,
) -> Result<EventPayload> {
let value = redacted_event_value(event, redactor)?;
EventPayload::new(value, run_id).map_err(anyhow::Error::from)
}
pub fn redacted_event_json(event: &RunEvent) -> Result<String> {
serde_json::to_string(&redacted_event_value(event)?).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)
}
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")
}
fn normalized_event_value(event: &RunEvent) -> Result<Value> {
@ -19,8 +28,96 @@ fn normalized_event_value(event: &RunEvent) -> Result<Value> {
Ok(normalize_json_value(value))
}
fn redacted_event_value(event: &RunEvent) -> Result<Value> {
Ok(redact_json_value(normalized_event_value(event)?))
fn redacted_event_value(event: &RunEvent, redactor: &SecretRedactor) -> Result<Value> {
let mut value = fabro_redact::redact_json_value(normalized_event_value(event)?);
redact_registered_secrets_in_event_value(&mut value, redactor);
Ok(value)
}
fn redact_registered_secrets_in_event_value(value: &mut Value, redactor: &SecretRedactor) {
if redactor.is_empty() {
return;
}
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;
}
}
if let Some(properties) = value.get_mut("properties") {
redact_registered_secrets_in_properties(properties, redactor);
}
}
fn redact_registered_secrets_in_properties(value: &mut Value, redactor: &SecretRedactor) {
match value {
Value::Object(properties) => {
for (key, child) in properties {
// Registered values can be low-entropy words. Redacting them in
// every event field would corrupt structural values such as ids,
// enum strings, and event names. Until event-field redaction is
// derived from typed field metadata, add new free-form text
// properties here when they are introduced.
if is_free_form_text_property(key) {
*child = redactor.redact_json(std::mem::take(child));
} else {
redact_registered_secrets_in_properties(child, redactor);
}
}
}
Value::Array(items) => {
for item in items {
redact_registered_secrets_in_properties(item, redactor);
}
}
_ => {}
}
}
fn is_free_form_text_property(key: &str) -> bool {
matches!(
key,
// command/script I/O
"command"
| "script"
| "stdout"
| "stderr"
| "output"
| "input"
| "arguments"
| "exec_output_tail"
| "tool_input"
| "tool_output"
// agent/LLM text
| "prompt"
| "response"
| "answer"
| "question"
| "delta"
| "text"
| "message"
| "reason"
| "notes"
| "preview"
// errors
| "error"
| "error_message"
| "failure"
| "causes"
| "details"
| "description"
// diffs/config
| "diff"
| "final_patch"
| "workflow_config"
| "workflow_source"
// metadata text
| "goal"
| "subject"
| "title"
)
}
pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result<EventPayload> {
@ -31,6 +128,7 @@ pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result<Ev
#[cfg(test)]
mod tests {
use ::fabro_types::{fixtures, run_event as fabro_types};
use fabro_redact::SecretRedactor;
use super::*;
use crate::event::{Event, to_run_event};
@ -40,7 +138,9 @@ mod tests {
let stored = to_run_event(&fixtures::RUN_8, &Event::RunSubmitted {
definition_blob: None,
});
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_8).unwrap();
let payload =
build_redacted_event_payload(&stored, &fixtures::RUN_8, &SecretRedactor::default())
.unwrap();
assert_eq!(payload.as_value()["id"], stored.id);
assert_eq!(payload.as_value()["event"], "run.submitted");
}
@ -61,7 +161,9 @@ mod tests {
}),
});
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_8).unwrap();
let payload =
build_redacted_event_payload(&stored, &fixtures::RUN_8, &SecretRedactor::default())
.unwrap();
let payload_text = serde_json::to_string(payload.as_value()).unwrap();
assert!(!payload_text.contains(secret));
@ -72,4 +174,131 @@ mod tests {
"plain stderr"
);
}
#[test]
fn build_redacted_event_payload_redacts_declared_low_entropy_values() {
let stored = to_run_event(&fixtures::RUN_8, &Event::RunNotice {
level: fabro_types::RunNoticeLevel::Warn,
code: "deploy".to_string(),
message: "deploy to staging".to_string(),
exec_output_tail: None,
});
let redactor = SecretRedactor::default();
redactor.register("staging");
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_8, &redactor).unwrap();
let payload_text = serde_json::to_string(payload.as_value()).unwrap();
assert!(!payload_text.contains("staging"));
assert!(payload_text.contains("REDACTED"));
}
#[test]
fn setup_failed_event_redacts_declared_low_entropy_values() {
let stored = to_run_event(&fixtures::RUN_8, &Event::SetupFailed {
command: "deploy staging".to_string(),
index: 0,
exit_code: 7,
stderr: "failed in staging".to_string(),
exec_output_tail: Some(fabro_types::ExecOutputTail {
stdout: None,
stderr: Some("tail staging".to_string()),
stdout_truncated: false,
stderr_truncated: false,
}),
});
let redactor = SecretRedactor::default();
redactor.register("staging");
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_8, &redactor).unwrap();
let payload_text = serde_json::to_string(payload.as_value()).unwrap();
assert_eq!(payload.as_value()["event"], "setup.failed");
assert!(!payload_text.contains("staging"));
assert!(payload_text.contains("REDACTED"));
}
#[test]
fn exact_redaction_preserves_structural_fields() {
let mut stored = to_run_event(&fixtures::RUN_8, &Event::RunNotice {
level: fabro_types::RunNoticeLevel::Warn,
code: "staging".to_string(),
message: "deploy to staging".to_string(),
exec_output_tail: None,
});
stored.node_id = Some("staging".to_string());
stored.node_label = Some("Deploy staging".to_string());
let redactor = SecretRedactor::default();
redactor.register("staging");
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_8, &redactor).unwrap();
let redacted = RunEvent::from_ref(payload.as_value()).unwrap();
assert!(
redacted
.node_id
.as_deref()
.is_some_and(|value| value == "staging")
);
assert!(
redacted
.node_label
.as_deref()
.is_some_and(|value| value == "Deploy REDACTED")
);
assert!(
payload.as_value()["properties"]["code"]
.as_str()
.is_some_and(|value| value == "staging")
);
assert!(
payload.as_value()["properties"]["message"]
.as_str()
.is_some_and(|value| value == "deploy to REDACTED")
);
}
#[test]
fn empty_registry_is_content_only_identity() {
let secret = "sk-ant-api03-xK9mZ2vL8nQ5rT1wY4bC7dF0gH3jE6pA";
let stored = to_run_event(&fixtures::RUN_8, &Event::RunNotice {
level: fabro_types::RunNoticeLevel::Warn,
code: "example".to_string(),
message: format!("token={secret}"),
exec_output_tail: None,
});
let content_only =
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();
assert_eq!(with_empty_registry, content_only_json);
}
#[test]
fn exact_redaction_is_per_registry() {
let stored = to_run_event(&fixtures::RUN_8, &Event::RunNotice {
level: fabro_types::RunNoticeLevel::Warn,
code: "example".to_string(),
message: "deploy staging production".to_string(),
exec_output_tail: None,
});
let staging = SecretRedactor::default();
staging.register("staging");
let production = SecretRedactor::default();
production.register("production");
let staging_payload =
build_redacted_event_payload(&stored, &fixtures::RUN_8, &staging).unwrap();
let production_payload =
build_redacted_event_payload(&stored, &fixtures::RUN_8, &production).unwrap();
let staging_text = serde_json::to_string(staging_payload.as_value()).unwrap();
let production_text = serde_json::to_string(production_payload.as_value()).unwrap();
assert!(!staging_text.contains("staging"));
assert!(staging_text.contains("production"));
assert!(production_text.contains("staging"));
assert!(!production_text.contains("production"));
}
}

70
lib/crates/fabro-workflow/src/event/sink.rs
View file

@ -4,18 +4,22 @@ use std::sync::Arc;
use ::fabro_types::{RunEvent, RunId};
use anyhow::Result;
use fabro_redact::SecretRedactor;
use fabro_store::RunDatabase;
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};
use super::redaction::{build_redacted_event_payload, redacted_event_json, redacted_run_event};
use super::{Event, to_run_event};
use crate::runtime_store::RunStoreHandle;
pub async fn append_event(run_store: &RunDatabase, run_id: &RunId, event: &Event) -> Result<()> {
let stored = to_run_event(run_id, event);
let payload = build_redacted_event_payload(&stored, run_id)?;
// Direct RunDatabase appends are server/test lifecycle paths with no
// per-run declared-secret registry. Worker run output goes through
// RunEventSink/RunEventLogger with the run redactor.
let payload = build_redacted_event_payload(&stored, run_id, &SecretRedactor::default())?;
run_store
.append_event(&payload)
.await
@ -27,9 +31,10 @@ 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).await
sink.write_run_event(&stored, redactor).await
}
#[derive(Clone)]
@ -99,21 +104,25 @@ impl RunEventSink {
}
}
pub async fn write_run_event(&self, event: &RunEvent) -> Result<()> {
pub async fn write_run_event(&self, event: &RunEvent, redactor: &SecretRedactor) -> Result<()> {
let mut pending = vec![(self, 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?;
}
Self::JsonLines(writer) => {
let line = redacted_event_json(&event)?;
let line = redacted_event_json(&event, redactor)?;
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) => callback(event).await?,
Self::Callback(callback) => {
let event = redacted_run_event(&event, redactor)?;
callback(event).await?;
}
Self::Map { transform, inner } => {
pending.push((inner.as_ref(), transform(event)));
}
@ -144,14 +153,15 @@ pub struct RunEventLogger {
impl RunEventLogger {
#[must_use]
pub fn new(sink: RunEventSink) -> Self {
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).await {
if let Err(err) = sink.write_run_event(&event, &logger_redactor).await {
tracing::warn!(error = %err, "Failed to write run event");
}
}
@ -193,9 +203,9 @@ pub struct StoreProgressLogger {
impl StoreProgressLogger {
#[must_use]
pub fn new(run_store: impl Into<RunStoreHandle>) -> Self {
pub fn new(run_store: impl Into<RunStoreHandle>, redactor: SecretRedactor) -> Self {
Self {
inner: RunEventLogger::new(RunEventSink::backend(run_store.into())),
inner: RunEventLogger::new(RunEventSink::backend(run_store.into()), redactor),
}
}
@ -261,7 +271,9 @@ mod tests {
message: "notice".to_string(),
exec_output_tail: None,
});
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_7).unwrap();
let payload =
build_redacted_event_payload(&stored, &fixtures::RUN_7, &SecretRedactor::default())
.unwrap();
run_store.append_event(&payload).await.unwrap();
let events = run_store.list_events().await.unwrap();
@ -282,8 +294,9 @@ 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).await.unwrap();
sink.write_run_event(&event, &redactor).await.unwrap();
let mut reader = BufReader::new(reader);
let mut line = String::new();
@ -323,8 +336,9 @@ mod tests {
]),
);
let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None });
let redactor = SecretRedactor::default();
sink.write_run_event(&event).await.unwrap();
sink.write_run_event(&event, &redactor).await.unwrap();
let first = first.lock().await;
let second = second.lock().await;
@ -340,7 +354,7 @@ mod tests {
let (writer, reader) = tokio::io::duplex(4096);
let sink = RunEventSink::json_lines(writer);
let logger = RunEventLogger::new(sink);
let logger = RunEventLogger::new(sink, SecretRedactor::default());
let emitter = Emitter::new(fixtures::RUN_8);
logger.register(&emitter);
@ -354,4 +368,32 @@ mod tests {
let payload = event_payload_from_redacted_json(line.trim_end(), &fixtures::RUN_8).unwrap();
assert_eq!(payload.as_value()["event"], "run.paused");
}
#[tokio::test]
async fn run_event_sink_callback_receives_redacted_event() {
let received = Arc::new(AsyncMutex::new(Vec::new()));
let received_events = Arc::clone(&received);
let sink = RunEventSink::callback(move |event| {
let received_events = Arc::clone(&received_events);
async move {
received_events.lock().await.push(event);
Ok(())
}
});
let event = to_run_event(&fixtures::RUN_7, &Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "example".to_string(),
message: "deploy to staging".to_string(),
exec_output_tail: None,
});
let redactor = SecretRedactor::default();
redactor.register("staging");
sink.write_run_event(&event, &redactor).await.unwrap();
let events = received.lock().await;
let text = serde_json::to_string(&events[0].to_value().unwrap()).unwrap();
assert!(!text.contains("staging"));
assert!(text.contains("REDACTED"));
}
}

5
lib/crates/fabro-workflow/src/handler/agent.rs
View file

@ -461,7 +461,10 @@ 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());
let logger = crate::event::StoreProgressLogger::new(
run_store.clone(),
fabro_redact::SecretRedactor::default(),
);
logger.register(services.run.emitter.as_ref());
(services, run_store, logger)
}

5
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -328,7 +328,10 @@ 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());
let logger = crate::event::StoreProgressLogger::new(
run_store.clone(),
fabro_redact::SecretRedactor::default(),
);
logger.register(services.run.emitter.as_ref());
(services, run_store, logger)
}

10
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

@ -783,7 +783,10 @@ 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());
let logger = crate::event::StoreProgressLogger::new(
run_store.clone(),
fabro_redact::SecretRedactor::default(),
);
logger.register(services.run.emitter.as_ref());
let mut node = Node::new("par");
node.attrs.insert(
@ -836,7 +839,10 @@ 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());
let logger = crate::event::StoreProgressLogger::new(
run_store.clone(),
fabro_redact::SecretRedactor::default(),
);
logger.register(services.run.emitter.as_ref());
let mut node = Node::new("par");
node.attrs.insert(

5
lib/crates/fabro-workflow/src/handler/prompt.rs
View file

@ -260,7 +260,10 @@ 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());
let logger = crate::event::StoreProgressLogger::new(
run_store.clone(),
fabro_redact::SecretRedactor::default(),
);
logger.register(services.run.emitter.as_ref());
(services, run_store, logger)
}

11
lib/crates/fabro-workflow/src/operations/fork.rs
View file

@ -208,8 +208,15 @@ async fn replay_historical_projection_events(
let mut event = envelope.event.clone();
event.id = format!("{new_run_id}-fork-{}", envelope.seq);
event.run_id = new_run_id;
let payload = event::build_redacted_event_payload(&event, &new_run_id)
.map_err(|err| Error::engine(err.to_string()))?;
// Fork replay copies events that were already persisted through the
// source run's redaction boundary; no live run secret registry exists
// for the historical projection replay.
let payload = event::build_redacted_event_payload(
&event,
&new_run_id,
&fabro_redact::SecretRedactor::default(),
)
.map_err(|err| Error::engine(err.to_string()))?;
run_store
.append_event(&payload)
.await

6
lib/crates/fabro-workflow/src/operations/resume.rs
View file

@ -1,5 +1,7 @@
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};
@ -39,10 +41,14 @@ 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()))?;

257
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -8,6 +8,7 @@ use fabro_interview::{AutoApproveInterviewer, Interviewer};
use fabro_llm::client::Client as LlmClient;
use fabro_mcp::config::McpServerSettings;
use fabro_model::{Catalog, FallbackTarget, ProviderId};
use fabro_redact::SecretRedactor;
use fabro_sandbox::daytona::DaytonaConfig;
use fabro_sandbox::from_environment::{
daytona_config_from_environment, docker_config_from_environment_with_secrets,
@ -81,6 +82,7 @@ struct RunSession {
vault: Option<Arc<AsyncRwLock<Vault>>>,
catalog: Arc<Catalog>,
fabro_run_tools: Option<FabroRunToolServices>,
secret_redactor: SecretRedactor,
}
struct ResolvedStartLlm {
@ -145,6 +147,9 @@ 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,
@ -152,6 +157,7 @@ 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()))?;
@ -162,6 +168,7 @@ 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()))?;
@ -179,6 +186,9 @@ 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(
@ -189,11 +199,19 @@ 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).await {
if let Err(err) = append_event_to_sink(
&event_sink,
&run_id,
&Event::RunStarting,
&bootstrap_redactor,
)
.await
{
let error = Error::engine(err.to_string());
let _ = persist_detached_failure(
run_id,
@ -203,6 +221,7 @@ pub(super) async fn execute_persisted_run(
"bootstrap",
FailureReason::BootstrapFailed,
&error,
&bootstrap_redactor,
)
.await;
return Err(error);
@ -226,6 +245,7 @@ pub(super) async fn execute_persisted_run(
"bootstrap",
FailureReason::BootstrapFailed,
&err,
&bootstrap_redactor,
)
.await;
bootstrap_guard.defuse();
@ -244,6 +264,7 @@ pub(super) async fn execute_persisted_run(
"bootstrap",
FailureReason::BootstrapFailed,
&err,
&bootstrap_redactor,
)
.await;
bootstrap_guard.defuse();
@ -252,11 +273,13 @@ 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;
@ -274,6 +297,7 @@ pub(super) async fn execute_persisted_run(
run_dir,
&err,
run_start.elapsed(),
&run_secret_redactor,
)
.await;
completion_guard.defuse();
@ -292,6 +316,7 @@ 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(
@ -313,7 +338,7 @@ async fn emit_workflow_run_failed(
None,
conclusion.billing,
);
if let Err(err) = append_event_to_sink(event_sink, &run_id, &failure_event).await {
if let Err(err) = append_event_to_sink(event_sink, &run_id, &failure_event, redactor).await {
tracing::warn!(error = %err, "Failed to append run.failed event");
}
}
@ -325,6 +350,7 @@ 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);
@ -339,6 +365,7 @@ async fn persist_terminal_engine_failure(
error,
reason,
crate::millis_u64(duration),
redactor,
)
.await;
}
@ -373,6 +400,7 @@ 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();
let vault_guard = match services.vault.as_ref() {
Some(vault) => Some(vault.read().await),
None => None,
@ -380,7 +408,9 @@ impl RunSession {
// Token-only secrets lookup over the vault read guard, shared across
// every run-boundary resolver. A missing or non-Token secret becomes
// `None`, so resolution fails closed with a secret error.
let secret_lookup = |name: &str| vault_token_lookup(vault_guard.as_deref(), name);
let secret_lookup = |name: &str| {
registered_vault_token_lookup(vault_guard.as_deref(), &secret_redactor, name)
};
let mcp_servers = resolved
.agent
.mcps
@ -507,6 +537,7 @@ impl RunSession {
vault: services.vault,
catalog,
fabro_run_tools: services.fabro_run_tools,
secret_redactor,
})
}
}
@ -566,6 +597,16 @@ fn vault_token_lookup(vault: Option<&Vault>, name: &str) -> Option<String> {
vault.and_then(|vault| fabro_auth::vault_get_token(vault, name).ok().flatten())
}
fn registered_vault_token_lookup(
vault: Option<&Vault>,
redactor: &SecretRedactor,
name: &str,
) -> Option<String> {
let value = vault_token_lookup(vault, name)?;
redactor.register(value.clone());
Some(value)
}
async fn load_accepted_run_definition(
run_store: &RunStoreHandle,
blob_id: fabro_types::RunBlobId,
@ -827,7 +868,8 @@ impl RunSession {
});
}
let store_progress_logger = RunEventLogger::new(self.event_sink.clone());
let store_progress_logger =
RunEventLogger::new(self.event_sink.clone(), self.secret_redactor.clone());
store_progress_logger.register(self.emitter.as_ref());
let init_options = InitOptions {
@ -854,8 +896,17 @@ impl RunSession {
checkpoint,
seed_context: self.seed_context,
fabro_run_tools: self.fabro_run_tools,
secret_redactor: self.secret_redactor.clone(),
};
let mut initialized = match Box::pin(pipeline::initialize(persisted, init_options)).await {
Ok(initialized) => initialized,
Err(err) => {
// Initialize can emit setup-failure events before returning;
// flush them before the terminal failure path appends run.failed.
store_progress_logger.flush().await;
return Err(err);
}
};
let mut initialized = Box::pin(pipeline::initialize(persisted, init_options)).await?;
initialized.on_node = on_node;
let sandbox_for_cleanup = Arc::clone(&initialized.engine.run.sandbox);
@ -969,6 +1020,9 @@ 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,
@ -976,6 +1030,7 @@ impl Drop for DetachedRunBootstrapGuard {
&Error::engine(reason.to_string()),
reason,
0,
&redactor,
)
.await;
});
@ -1005,11 +1060,12 @@ 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,
active: bool,
event_sink: RunEventSink,
run_id: RunId,
run_store: RunStoreHandle,
cancel_token: CancellationToken,
secret_redactor: SecretRedactor,
active: bool,
}
impl DetachedRunCompletionGuard {
@ -1018,12 +1074,14 @@ 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,
}
}
@ -1058,6 +1116,7 @@ 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)
@ -1072,14 +1131,20 @@ 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,
})
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,
)
.await;
});
}
@ -1094,8 +1159,9 @@ 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).await;
emit_workflow_run_failed(run_id, run_store, event_sink, error, reason, 0, redactor).await;
let event = Event::RunNotice {
level: RunNoticeLevel::Error,
@ -1103,7 +1169,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).await {
if let Err(err) = append_event_to_sink(event_sink, &run_id, &event, redactor).await {
tracing::warn!(error = %err, "Failed to append detached failure notice");
}
@ -1568,6 +1634,159 @@ reasoning = false
);
}
#[tokio::test]
async fn run_session_registered_secret_redacts_pre_stage_stored_event() {
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()
});
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()
}),
);
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 session = RunSession::new(&persisted, StartServices {
vault: Some(vault),
..test_start_services(&store, &storage_root, emitter, registry).await
})
.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,
)
.await
.unwrap();
let events = session.run_store.list_events().await.unwrap();
let serialized = serde_json::to_string(&events).unwrap();
assert!(!serialized.contains("staging"));
assert!(serialized.contains("REDACTED"));
}
#[tokio::test]
async fn start_redacts_declared_secret_from_setup_failure_events_and_error() {
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()
});
settings.run.prepare = prepare_with_step(command_step(
&["sh", "-c", "printf %s \"$DEPLOY_ENV\" >&2; exit 7"],
HashMap::from([(
"DEPLOY_ENV".to_string(),
"{{ secrets.DEPLOY_ENV }}".to_string(),
)]),
));
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("DEPLOY_ENV", "staging")));
let Err(err) = start(&run_dir, StartServices {
vault: Some(vault),
..test_start_services(&store, &storage_root, emitter, registry).await
})
.await
else {
panic!("setup failure should fail the run");
};
let error_text = err.to_string();
assert!(!error_text.contains("staging"));
assert!(error_text.contains("REDACTED"));
let run_store = store.open_run(&fixtures::RUN_1).await.unwrap();
let events = run_store.list_events().await.unwrap();
let serialized = serde_json::to_string(&events).unwrap();
assert!(
events
.iter()
.any(|event| event.event.event_name() == "setup.failed")
);
assert!(!serialized.contains("staging"));
assert!(serialized.contains("REDACTED"));
}
#[tokio::test]
async fn run_session_secret_redactors_are_isolated_per_run() {
async fn session_with_secret(value: &str) -> RunSession {
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()
});
settings.run.environment.env.insert(
"DEPLOY_ENV".to_string(),
InterpString::parse("{{ secrets.DEPLOY_ENV }}"),
);
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("DEPLOY_ENV", value)));
RunSession::new(&persisted, StartServices {
vault: Some(vault),
..test_start_services(&store, &storage_root, emitter, registry).await
})
.await
.unwrap()
}
let staging_session = session_with_secret("staging").await;
let production_session = session_with_secret("production").await;
let staging_text = staging_session
.secret_redactor
.redact_into("staging production");
let production_text = production_session
.secret_redactor
.redact_into("staging production");
assert!(!staging_text.contains("staging"));
assert!(staging_text.contains("production"));
assert!(production_text.contains("staging"));
assert!(!production_text.contains("production"));
}
#[tokio::test]
async fn run_session_new_missing_secret_fails_startup() {
let temp = tempfile::tempdir().unwrap();
@ -1990,6 +2209,7 @@ reasoning = false
&run_dir,
&Error::engine("visit limit exceeded"),
Duration::from_millis(9_999),
&SecretRedactor::default(),
)
.await;
@ -2072,6 +2292,7 @@ reasoning = false
run_store_handle,
event_sink,
CancellationToken::new(),
SecretRedactor::default(),
);
}

6
lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -15,6 +15,7 @@ use fabro_agent::Sandbox;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_hooks::HookSettings;
use fabro_interview::AutoApproveInterviewer;
use fabro_redact::SecretRedactor;
use fabro_sandbox::SandboxSpec;
use fabro_store::Database;
use fabro_types::settings::run::RunModelControls;
@ -251,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());
let store_logger = StoreProgressLogger::new(run_store.clone(), SecretRedactor::default());
store_logger.register(&emitter);
let initialized = initialize(
persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value),
@ -295,6 +296,7 @@ async fn execute_test_run_with_options(
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor: SecretRedactor::default(),
},
)
.await
@ -358,6 +360,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor: SecretRedactor::default(),
},
)
.await
@ -429,6 +432,7 @@ async fn run_with_lifecycle(
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor: SecretRedactor::default(),
},
)
.await?;

3
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -1035,7 +1035,8 @@ 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());
let store_logger =
StoreProgressLogger::new(run_store.clone(), fabro_redact::SecretRedactor::default());
store_logger.register(&emitter);
let sandbox: Arc<dyn fabro_agent::Sandbox> = Arc::new(fabro_agent::LocalSandbox::new(
std::env::current_dir().unwrap(),

99
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -559,10 +559,12 @@ pub async fn initialize(
stderr: result.stderr.clone(),
exec_output_tail,
});
return Err(Error::engine(format!(
let message = format!(
"Setup command failed (exit code {}): {command}\n{}",
exit_code, result.stderr,
)));
);
let message = fabro_redact::redact_string(&message);
return Err(Error::engine(options.secret_redactor.redact_into(&message)));
}
let exit_code = result.exit_code.unwrap_or(0);
options.emitter.emit(&Event::SetupCommandCompleted {
@ -647,6 +649,7 @@ mod tests {
use fabro_acp::test_support::fake_acp_agent_script;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_interview::AutoApproveInterviewer;
use fabro_redact::SecretRedactor;
use fabro_sandbox::SandboxSpec;
use fabro_store::Database;
use fabro_types::settings::run::RunModelControls;
@ -798,6 +801,13 @@ mod tests {
async fn initialize_with_setup_step(
setup: crate::run_options::SetupCommand,
) -> (crate::error::Result<Initialized>, Vec<RunEvent>) {
initialize_with_setup_step_and_redactor(setup, SecretRedactor::default()).await
}
async fn initialize_with_setup_step_and_redactor(
setup: crate::run_options::SetupCommand,
secret_redactor: SecretRedactor,
) -> (crate::error::Result<Initialized>, Vec<RunEvent>) {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
@ -812,18 +822,18 @@ mod tests {
});
let result = initialize(persisted, InitOptions {
run_id: test_run_id(),
run_store: {
run_id: test_run_id(),
run_store: {
let store = memory_store();
let inner = store.create_run(&test_run_id()).await.unwrap();
inner.into()
},
dry_run: false,
emitter: emitter.clone(),
sandbox: SandboxSpec::Local {
dry_run: false,
emitter: emitter.clone(),
sandbox: SandboxSpec::Local {
working_directory: std::env::current_dir().unwrap(),
},
llm: LlmSpec {
llm: LlmSpec {
model: "test-model".to_string(),
provider_id: fabro_model::ProviderId::anthropic(),
fallback_chain: Vec::new(),
@ -831,30 +841,31 @@ mod tests {
model_controls: RunModelControls::default(),
dry_run: true,
},
interviewer: Arc::new(AutoApproveInterviewer::engine()),
steering_hub: Arc::new(crate::steering_hub::SteeringHub::new(emitter.clone())),
catalog: test_catalog(),
lifecycle: crate::run_options::LifecycleOptions {
interviewer: Arc::new(AutoApproveInterviewer::engine()),
steering_hub: Arc::new(crate::steering_hub::SteeringHub::new(emitter.clone())),
catalog: test_catalog(),
lifecycle: crate::run_options::LifecycleOptions {
setup_commands: vec![setup],
setup_command_timeout_ms: 1_000,
},
run_options: test_settings(&run_dir),
workflow_path: None,
workflow_bundle: None,
hooks: fabro_hooks::HookSettings { hooks: vec![] },
sandbox_env: SandboxEnvSpec {
run_options: test_settings(&run_dir),
workflow_path: None,
workflow_bundle: None,
hooks: fabro_hooks::HookSettings { hooks: vec![] },
sandbox_env: SandboxEnvSpec {
toml_env: HashMap::new(),
github_permissions: None,
origin_url: None,
},
vault: None,
git: None,
run_control: None,
vault: None,
git: None,
run_control: None,
registry_override: None,
artifact_sink: None,
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
artifact_sink: None,
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor,
})
.await;
let events = seen.lock().unwrap().clone();
@ -937,6 +948,7 @@ mod tests {
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor: SecretRedactor::default(),
})
.await
.unwrap();
@ -1135,6 +1147,7 @@ mod tests {
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor: SecretRedactor::default(),
})
.await
.unwrap();
@ -1183,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());
let store_logger = StoreProgressLogger::new(run_store.clone(), SecretRedactor::default());
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
emitter.on_event({
let seen = Arc::clone(&seen);
@ -1231,6 +1244,7 @@ mod tests {
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor: SecretRedactor::default(),
})
.await
.unwrap();
@ -1317,6 +1331,40 @@ mod tests {
}
}
#[tokio::test]
async fn initialize_setup_failure_redacts_registered_secret_in_error_text() {
let redactor = SecretRedactor::default();
redactor.register("staging");
let setup = crate::run_options::SetupCommand {
command: "printf %s \"$DEPLOY_ENV\" >&2; exit 7".to_string(),
env: HashMap::from([("DEPLOY_ENV".to_string(), "staging".to_string())]),
};
let (result, _events) = initialize_with_setup_step_and_redactor(setup, redactor).await;
let Err(err) = result else {
panic!("setup should fail");
};
let text = err.to_string();
assert!(!text.contains("staging"));
assert!(text.contains("REDACTED"));
}
#[tokio::test]
async fn initialize_setup_failure_keeps_content_redaction_with_empty_registry() {
let secret = "sk-ant-api03-xK9mZ2vL8nQ5rT1wY4bC7dF0gH3jE6pA";
let command = format!("printf %s {secret} >&2; exit 7");
let (result, _events) = initialize_with_setup_command(&command).await;
let Err(err) = result else {
panic!("setup should fail");
};
let text = err.to_string();
assert!(!text.contains(secret));
assert!(text.contains("REDACTED"));
}
#[tokio::test]
async fn initialize_cancelled_setup_command_returns_cancelled() {
let temp = tempfile::tempdir().unwrap();
@ -1374,6 +1422,7 @@ mod tests {
checkpoint: None,
seed_context: None,
fabro_run_tools: None,
secret_redactor: SecretRedactor::default(),
})
.await;

2
lib/crates/fabro-workflow/src/pipeline/types.rs
View file

@ -6,6 +6,7 @@ use fabro_graphviz::graph::Graph;
use fabro_interview::Interviewer;
use fabro_mcp::config::McpServerSettings;
use fabro_model::{Catalog, FallbackTarget, ProviderId};
use fabro_redact::SecretRedactor;
use fabro_sandbox::SandboxSpec;
use fabro_template::TemplateContext;
use fabro_types::settings::run::{PullRequestSettings, RunModelControls};
@ -272,6 +273,7 @@ pub struct InitOptions {
pub checkpoint: Option<Checkpoint>,
pub seed_context: Option<Context>,
pub fabro_run_tools: Option<FabroRunToolServices>,
pub secret_redactor: SecretRedactor,
}
/// Output of the INITIALIZE phase.

7
lib/crates/fabro-workflow/src/runtime_store.rs
View file

@ -3,6 +3,7 @@ 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};
@ -83,7 +84,11 @@ impl RunStoreBackend for LocalRunStoreBackend {
}
async fn append_run_event(&self, event: &RunEvent) -> Result<()> {
let payload = build_redacted_event_payload(event, &event.run_id)?;
// 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())?;
self.run_store
.append_event(&payload)
.await

3
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -201,7 +201,8 @@ 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());
let store_logger =
StoreProgressLogger::new(run_store.clone(), fabro_redact::SecretRedactor::default());
store_logger.register(emitter.as_ref());
let artifact_store = ArtifactStore::new(
Arc::new(