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 5

Anchor the run secret registry to the event sink, not the session

Browse source 
The per-run SecretRedactor was created inside RunSession::new and stored
on the session, so every emit path that runs before session construction
succeeds (bootstrap failures, detached guards, lifecycle markers) had to
pass an empty SecretRedactor::default() with a comment asserting that no
secret could have been resolved yet. That assertion was false for the
session-construction failure path: RunSession::new is exactly where
run-boundary secret resolution happens, so a failure after partial
resolution emitted its failure events through an empty registry and lost
the values that had already been registered.

RunEventSink now owns the registry and consults it on every write, and
run-boundary resolution registers values through the sink carried in
StartServices. Registrations made before a construction failure reach
every later write through the same sink, so the gap closes structurally
instead of by per-call-site argument. The redactor parameter threading
(write_run_event, append_event_to_sink, RunEventLogger::new,
StoreProgressLogger::new, the failure helpers, and both drop guards) is
gone along with the justification comments it required.

Adds a regression test that a failed RunSession::new still redacts
values registered before the failure.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Scott Werner 2026-07-09 13:27:08 -04:00
parent cff0b342df
commit f98db10259
12 changed files with 181 additions and 144 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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;

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

@ -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();

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

@ -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)
}

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

@ -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)
}

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

@ -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(

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

@ -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)
}

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

@ -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()))?;

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

@ -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(),
);
}

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

@ -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),

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

@ -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(),

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

@ -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);

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

@ -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(