scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Surface silent fallback warnings in runs and logs (#205)

Browse source 
### Summary
Silent fallback paths now emit stable warnings instead of degrading
without a user-visible signal. The fallback behavior is unchanged; runs
still continue, but worktree, Git, checkpoint, and LLM failover issues
now show up in the run feed and logs.

### Plan Summary
- Emit run notices for workflow Git/worktree fallback paths.
- Reuse the existing failover event for one-shot LLM provider fallback.
- Add tracing for sandbox pipe drain failures.

### What changed
- Added `worktree_skipped_no_git` and gated `sandbox_git_unavailable`
notices during initialization.
- Added `git_push_failed` and `parallel_base_checkpoint_failed` notices,
including redacted output tails where available.
- Logged GitHub token mint failures with a structured `error` field
before the existing notice.
- Plumbed `Emitter` and `StageScope` through `CodergenBackend::one_shot`
so the API backend emits the existing `agent.failover` event instead of
a duplicate tracing-only warning.
- Extracted sandbox pipe draining into a helper that warns on
stdout/stderr read failures, with unit coverage for the error path.
- Updated CLI snapshots for the new worktree warning in stderr and JSON
event output.

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

---------

Co-authored-by: Fabro <noreply@fabro.sh>
Co-authored-by: Bryan Helmkamp <bryan@brynary.com>
This commit is contained in:
fabro-sh-0530[bot] 2026-05-05 09:18:05 -04:00 • committed by GitHub
parent 6e36d8350e
commit e901cd3a81
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 337 additions and 124 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
lib/crates/fabro-cli/src/commands/run/logs.rs
View file

@ -14,7 +14,7 @@ use std::time::Duration;
use anyhow::{Context, Result, bail};
use chrono::{DateTime, Utc};
use fabro_redact::redact_jsonl_line;
use fabro_types::run_event::is_metadata_snapshot_compat_notice_code;
use fabro_types::RunNoticeCode;
use fabro_util::json::normalize_json_value;
use fabro_util::terminal::Styles;
use tokio::time;
@ -801,7 +801,9 @@ fn format_event_pretty_value(envelope: &serde_json::Value, styles: &Styles) -> O
}
fn is_metadata_snapshot_compat_notice(envelope: &serde_json::Value) -> bool {
prop_str_field(envelope, "code").is_some_and(is_metadata_snapshot_compat_notice_code)
prop_str_field(envelope, "code")
.and_then(|code| code.parse::<RunNoticeCode>().ok())
.is_some_and(RunNoticeCode::is_metadata_snapshot_compat)
}
fn str_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> {
@ -1150,14 +1152,27 @@ mod tests {
#[test]
fn pretty_run_notice_warn() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"run.notice","properties":{"level":"warn","code":"sandbox_cleanup_failed","message":"sandbox cleanup failed: boom"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
let code = RunNoticeCode::SandboxCleanupFailed.to_string();
let line = serde_json::json!({
"ts": "2026-01-01T14:25:00Z",
"event": "run.notice",
"properties": {
"level": "warn",
"code": code,
"message": "sandbox cleanup failed: boom",
},
})
.to_string();
let result = format_event_pretty(&line, &styles).unwrap();
assert!(result.contains("Warning:"), "got: {result}");
assert!(
result.contains("sandbox cleanup failed: boom"),
"got: {result}"
);
assert!(result.contains("[sandbox_cleanup_failed]"), "got: {result}");
assert!(
result.contains(&format!("[{}]", RunNoticeCode::SandboxCleanupFailed)),
"got: {result}"
);
}
#[test]
@ -1235,13 +1250,31 @@ mod tests {
fn pretty_stream_suppresses_metadata_compat_notice_only() {
let styles = no_color_styles();
let failed = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.failed","properties":{"phase":"checkpoint","branch":"fabro/meta","duration_ms":900,"failure_kind":"write","error":"write failed"}}"#;
let compat_notice = r#"{"ts":"2026-01-01T14:25:01Z","event":"run.notice","properties":{"level":"warn","code":"checkpoint_metadata_write_failed","message":"legacy metadata warning"}}"#;
let degraded_notice = r#"{"ts":"2026-01-01T14:25:02Z","event":"run.notice","properties":{"level":"warn","code":"checkpoint_metadata_degraded","message":"metadata snapshots disabled"}}"#;
let compat_notice = serde_json::json!({
"ts": "2026-01-01T14:25:01Z",
"event": "run.notice",
"properties": {
"level": "warn",
"code": RunNoticeCode::CheckpointMetadataWriteFailed,
"message": "legacy metadata warning",
},
})
.to_string();
let degraded_notice = serde_json::json!({
"ts": "2026-01-01T14:25:02Z",
"event": "run.notice",
"properties": {
"level": "warn",
"code": RunNoticeCode::CheckpointMetadataDegraded,
"message": "metadata snapshots disabled",
},
})
.to_string();
let mut state = PrettyEventState::default();
assert!(format_event_pretty_streamed(failed, &styles, &mut state).is_some());
assert!(format_event_pretty_streamed(compat_notice, &styles, &mut state).is_none());
let degraded = format_event_pretty_streamed(degraded_notice, &styles, &mut state).unwrap();
assert!(format_event_pretty_streamed(&compat_notice, &styles, &mut state).is_none());
let degraded = format_event_pretty_streamed(&degraded_notice, &styles, &mut state).unwrap();
assert!(
degraded.contains("metadata snapshots disabled"),
"got: {degraded}"

7
lib/crates/fabro-cli/src/commands/run/run_progress/event.rs
View file

@ -527,7 +527,7 @@ fn display_value(value: &Value) -> Option<String> {
mod tests {
use fabro_agent::AgentEvent;
use fabro_types::{MetadataSnapshotFailureKind, MetadataSnapshotPhase, fixtures};
use fabro_workflow::event::{Event, to_run_event};
use fabro_workflow::event::{Event, RunNoticeCode, to_run_event};
use super::*;
@ -804,10 +804,11 @@ mod tests {
fn round_trip_run_notice() {
let event = Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "sandbox_cleanup_failed".into(),
code: RunNoticeCode::SandboxCleanupFailed.to_string(),
message: "sandbox cleanup failed".into(),
exec_output_tail: None,
};
let expected_code = RunNoticeCode::SandboxCleanupFailed.to_string();
let stored = to_run_event(&fixtures::RUN_1, &event);
let parsed = from_run_event(&stored).unwrap();
@ -817,7 +818,7 @@ mod tests {
level: RunNoticeLevel::Warn,
code,
message,
} if code == "sandbox_cleanup_failed" && message == "sandbox cleanup failed"
} if code == expected_code && message == "sandbox cleanup failed"
));
}

14
lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs
View file

@ -3,8 +3,7 @@
reason = "sync CLI run-progress renderer: writes to std::io::stderr directly"
)]
use fabro_types::RunEvent;
use fabro_types::run_event::is_metadata_snapshot_compat_notice_code;
use fabro_types::{RunEvent, RunNoticeCode};
mod event;
mod info_display;
@ -444,7 +443,10 @@ impl ProgressUI {
message,
} => {
if self.saw_metadata_snapshot_failure
&& is_metadata_snapshot_compat_notice_code(&code)
&& code
.parse::<RunNoticeCode>()
.ok()
.is_some_and(RunNoticeCode::is_metadata_snapshot_compat)
{
return;
}
@ -1208,7 +1210,7 @@ mod tests {
emit(&mut ui, Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "sandbox_cleanup_failed".into(),
code: RunNoticeCode::SandboxCleanupFailed.to_string(),
message: "sandbox cleanup failed".into(),
exec_output_tail: None,
});
@ -1279,13 +1281,13 @@ mod tests {
});
emit(&mut ui, Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "checkpoint_metadata_write_failed".into(),
code: RunNoticeCode::CheckpointMetadataWriteFailed.to_string(),
message: "legacy metadata warning".into(),
exec_output_tail: None,
});
emit(&mut ui, Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "checkpoint_metadata_degraded".into(),
code: RunNoticeCode::CheckpointMetadataDegraded.to_string(),
message: "metadata snapshots are disabled for this run".into(),
exec_output_tail: None,
});

17
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -160,6 +160,7 @@ fn attach_replays_completed_detached_run() {
----- stdout -----
----- stderr -----
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ Start [TIME]
✓ Run Tests [TIME]
@ -267,6 +268,7 @@ fn attach_before_completion_streams_to_finished_state() {
----- stdout -----
----- stderr -----
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ start [DURATION]
✓ wait [DURATION]
@ -699,6 +701,21 @@ fn attach_json_errors_without_prompting_for_human_input() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"actor": {
"kind": "worker",
"run_id": "[ULID]"
},
"event": "run.notice",
"id": "[EVENT_ID]",
"properties": {
"code": "worktree_skipped_no_git",
"level": "warn",
"message": "Worktree mode `always` requested but no Git repository was found; running without a worktree."
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"actor": {
"kind": "worker",

6
lib/crates/fabro-cli/tests/it/cmd/dump.rs
View file

@ -278,9 +278,9 @@ fn dump_exports_completed_run_snapshot() {
");
assert_snapshot!(dump_file_summary(&output_dir), @"
checkpoints/0013.json
checkpoints/0017.json
checkpoints/0021.json
checkpoints/0014.json
checkpoints/0018.json
checkpoints/0022.json
events.jsonl
graph.fabro
run.json

1
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -684,6 +684,7 @@ fn dry_run_simple() {
Run: [ULID]
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ Start [TIME]
✓ Run Tests [TIME]

5
lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs
View file

@ -21,6 +21,7 @@ fn dry_run_branching() {
warning [node: implement]: Node 'implement' has goal_gate=true but no retry_target or fallback_retry_target (goal_gate_has_retry)
Run: [ULID]
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ Start [TIME]
✓ Plan [TIME]
@ -57,6 +58,7 @@ fn dry_run_conditions() {
Run: [ULID]
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ start [TIME]
✓ Decide [TIME]
@ -91,6 +93,7 @@ fn dry_run_parallel() {
Run: [ULID]
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ start [TIME]
✓ Fork Work [TIME]
@ -126,6 +129,7 @@ fn dry_run_styled() {
Run: [ULID]
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ start [TIME]
✓ Plan [TIME]
@ -161,6 +165,7 @@ fn dry_run_legacy_tool() {
Run: [ULID]
Web UI: http://localhost:3000/runs/[ULID]
Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git]
Sandbox: local (ready in [TIME])
✓ Start [TIME]
✓ Echo [TIME]

59
lib/crates/fabro-sandbox/src/local.rs
View file

@ -126,6 +126,19 @@ fn process_env_vars() -> Vec<(String, String)> {
std::env::vars().collect()
}
async fn drain_pipe<R>(mut pipe: Option<R>, stream: CommandOutputStream) -> String
where
R: AsyncRead + Unpin,
{
let mut buf = String::new();
if let Some(ref mut reader) = pipe {
if let Err(err) = reader.read_to_string(&mut buf).await {
tracing::warn!(error = %err, ?stream, "Failed to drain child output");
}
}
buf
}
#[async_trait]
impl Sandbox for LocalSandbox {
async fn read_file(
@ -277,22 +290,12 @@ impl Sandbox for LocalSandbox {
// it writes more than the OS pipe buffer (~64 KB) the write() syscall
// blocks until the parent drains the pipe, but the parent is blocked
// on child.wait().
let mut stdout_pipe = child.stdout.take();
let mut stderr_pipe = child.stderr.take();
let stdout_task = tokio::spawn(async move {
let mut buf = String::new();
if let Some(ref mut r) = stdout_pipe {
let _ = r.read_to_string(&mut buf).await;
}
buf
});
let stderr_task = tokio::spawn(async move {
let mut buf = String::new();
if let Some(ref mut r) = stderr_pipe {
let _ = r.read_to_string(&mut buf).await;
}
buf
});
let stdout_pipe = child.stdout.take();
let stderr_pipe = child.stderr.take();
let stdout_task =
tokio::spawn(async move { drain_pipe(stdout_pipe, CommandOutputStream::Stdout).await });
let stderr_task =
tokio::spawn(async move { drain_pipe(stderr_pipe, CommandOutputStream::Stderr).await });
let (termination, exit_code) = tokio::select! {
status_result = child.wait() => {
@ -712,7 +715,12 @@ where
)]
mod tests {
use std::collections::HashMap;
use std::io;
use std::path::PathBuf;
use std::pin::Pin;
use std::task::{Context as TaskContext, Poll};
use tokio::io::ReadBuf;
use super::*;
@ -722,6 +730,25 @@ mod tests {
dir
}
#[tokio::test]
async fn drain_pipe_returns_empty_buffer_after_read_failure() {
struct FailingReader;
impl AsyncRead for FailingReader {
fn poll_read(
self: Pin<&mut Self>,
_cx: &mut TaskContext<'_>,
_buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Poll::Ready(Err(io::Error::other("simulated read failure")))
}
}
let output = drain_pipe(Some(FailingReader), CommandOutputStream::Stdout).await;
assert!(output.is_empty());
}
#[tokio::test]
async fn read_file_with_line_numbers() {
let dir = temp_dir();

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

@ -72,7 +72,7 @@ pub use run::{
pub use run_blob_id::RunBlobId;
pub use run_event::{
EventBody, ExecOutputTail, InterviewOption, MetadataSnapshotFailureKind, MetadataSnapshotPhase,
RunEvent, RunNoticeLevel,
RunEvent, RunNoticeCode, RunNoticeLevel,
};
pub use run_id::{RunId, fixtures};
pub use run_projection::{PendingInterviewRecord, RunProjection, StageProjection, first_event_seq};

53
lib/crates/fabro-types/src/run_event/infra.rs
View file

@ -1,17 +1,48 @@
use serde::{Deserialize, Serialize};
/// Legacy `run.notice` codes paired with the new `metadata.snapshot.failed`
/// event for backward compatibility. Display layers suppress these so the
/// typed event renders without a duplicate raw warning.
pub const NOTICE_CODE_CHECKPOINT_METADATA_WRITE_FAILED: &str = "checkpoint_metadata_write_failed";
pub const NOTICE_CODE_CHECKPOINT_METADATA_PUSH_FAILED: &str = "checkpoint_metadata_push_failed";
#[derive(
Debug,
Clone,
Copy,
PartialEq,
Eq,
Hash,
Serialize,
Deserialize,
strum::Display,
strum::EnumString,
strum::IntoStaticStr,
)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum RunNoticeCode {
ArtifactCollectionFailed,
ArtifactOffloadFailed,
ArtifactSyncFailed,
ArtifactUploadFailed,
CheckpointMetadataDegraded,
CheckpointMetadataPushFailed,
CheckpointMetadataWriteFailed,
DirtyWorktree,
GitDiffFailed,
GitPushFailed,
GithubTokenFailed,
ParallelBaseCheckpointFailed,
PullRequestFailed,
SandboxCleanupFailed,
SandboxGitUnavailable,
SandboxPreserved,
WorktreeSkippedNoGit,
}
#[must_use]
pub fn is_metadata_snapshot_compat_notice_code(code: &str) -> bool {
matches!(
code,
NOTICE_CODE_CHECKPOINT_METADATA_WRITE_FAILED | NOTICE_CODE_CHECKPOINT_METADATA_PUSH_FAILED
)
impl RunNoticeCode {
#[must_use]
pub fn is_metadata_snapshot_compat(self) -> bool {
matches!(
self,
Self::CheckpointMetadataWriteFailed | Self::CheckpointMetadataPushFailed
)
}
}
#[derive(

4
lib/crates/fabro-types/src/run_event/mod.rs
View file

@ -1366,7 +1366,7 @@ mod tests {
for body in [
EventBody::RunNotice(RunNoticeProps {
level: RunNoticeLevel::Warn,
code: "git_diff_failed".to_string(),
code: RunNoticeCode::GitDiffFailed.to_string(),
message: "git diff failed".to_string(),
exec_output_tail: Some(tail.clone()),
}),
@ -1402,7 +1402,7 @@ mod tests {
for body in [
EventBody::RunNotice(RunNoticeProps {
level: RunNoticeLevel::Warn,
code: "git_diff_failed".to_string(),
code: RunNoticeCode::GitDiffFailed.to_string(),
message: "git diff failed".to_string(),
exec_output_tail: None,
}),

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

@ -8,7 +8,7 @@ mod stored_fields;
#[cfg(test)]
mod test_support;
pub use fabro_types::{EventBody, RunNoticeLevel};
pub use fabro_types::{EventBody, RunNoticeCode, RunNoticeLevel};
pub use self::convert::{to_run_event, to_run_event_at};
pub use self::emitter::Emitter;

6
lib/crates/fabro-workflow/src/event/convert.rs
View file

@ -1169,8 +1169,8 @@ mod tests {
use std::collections::BTreeMap;
use ::fabro_types::{
EventBody, FailureReason, ParallelBranchId, Principal, RunNoticeLevel, RunProvenance,
StageId, SystemActorKind, fixtures, run_event as fabro_types,
EventBody, FailureReason, ParallelBranchId, Principal, RunNoticeCode, RunNoticeLevel,
RunProvenance, StageId, SystemActorKind, fixtures, run_event as fabro_types,
};
use chrono::Utc;
use fabro_agent::{AgentEvent, SandboxEvent};
@ -1643,7 +1643,7 @@ mod tests {
fn run_notice_maps_exec_output_tail_to_props() {
let stored = to_run_event(&fixtures::RUN_1, &Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "git_diff_failed".to_string(),
code: RunNoticeCode::GitDiffFailed.to_string(),
message: "git diff failed".to_string(),
exec_output_tail: Some(exec_tail()),
});

15
lib/crates/fabro-workflow/src/event/emitter.rs
View file

@ -1,7 +1,7 @@
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use ::fabro_types::{ExecOutputTail, RunEvent, RunId, RunNoticeLevel};
use ::fabro_types::{ExecOutputTail, RunEvent, RunId, RunNoticeCode, RunNoticeLevel};
use chrono::Utc;
use fabro_agent::{WorktreeEvent, WorktreeEventCallback};
@ -76,15 +76,10 @@ impl Emitter {
self.emit_with_scope(event, Some(scope));
}
pub fn notice(
&self,
level: RunNoticeLevel,
code: impl Into<String>,
message: impl Into<String>,
) {
pub fn notice(&self, level: RunNoticeLevel, code: RunNoticeCode, message: impl Into<String>) {
self.emit(&Event::RunNotice {
level,
code: code.into(),
code: code.to_string(),
message: message.into(),
exec_output_tail: None,
});
@ -93,13 +88,13 @@ impl Emitter {
pub fn notice_with_tail(
&self,
level: RunNoticeLevel,
code: impl Into<String>,
code: RunNoticeCode,
message: impl Into<String>,
exec_output_tail: Option<ExecOutputTail>,
) {
self.emit(&Event::RunNotice {
level,
code: code.into(),
code: code.to_string(),
message: message.into(),
exec_output_tail,
});

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

@ -52,6 +52,8 @@ pub trait CodergenBackend: Send + Sync {
_node: &Node,
_prompt: &str,
_system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &StageScope,
) -> Result<CodergenResult, Error> {
Err(Error::Validation(
"one_shot mode not supported by this backend".into(),

20
lib/crates/fabro-workflow/src/handler/llm/api.rs
View file

@ -285,6 +285,8 @@ impl CodergenBackend for AgentApiBackend {
node: &Node,
prompt: &str,
system_prompt: Option<&str>,
emitter: &Arc<Emitter>,
stage_scope: &StageScope,
) -> Result<CodergenResult, Error> {
let client = Client::from_source(self.source.as_ref())
.await
@ -358,14 +360,16 @@ impl CodergenBackend for AgentApiBackend {
let mut found = None;
for target in fallback_chain {
tracing::warn!(
stage = node.id.as_str(),
from_provider = from_provider.as_str(),
from_model = from_model.as_str(),
to_provider = target.provider.as_str(),
to_model = target.model.as_str(),
error = error_msg.as_str(),
"LLM provider failover (prompt)"
emitter.emit_scoped(
&Event::Failover {
stage: node.id.clone(),
from_provider: from_provider.clone(),
from_model: from_model.clone(),
to_provider: target.provider.clone(),
to_model: target.model.clone(),
error: error_msg.clone(),
},
stage_scope,
);
let max_tokens = node.max_tokens().or_else(|| {

6
lib/crates/fabro-workflow/src/handler/llm/cli.rs
View file

@ -810,9 +810,13 @@ impl CodergenBackend for BackendRouter {
node: &Node,
prompt: &str,
system_prompt: Option<&str>,
emitter: &Arc<Emitter>,
stage_scope: &StageScope,
) -> Result<CodergenResult, Error> {
// CLI backend doesn't support one_shot, always route to API
self.api_backend.one_shot(node, prompt, system_prompt).await
self.api_backend
.one_shot(node, prompt, system_prompt, emitter, stage_scope)
.await
}
}

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

@ -12,7 +12,7 @@ use tokio::sync::Semaphore;
use super::{EngineServices, Handler};
use crate::context::{Context, WorkflowContext, keys};
use crate::error::Error;
use crate::event::{Event, StageScope};
use crate::event::{Event, RunNoticeCode, RunNoticeLevel, StageScope};
use crate::git::sanitize_ref_component;
use crate::hook_context::set_hook_node;
use crate::millis_u64;
@ -207,6 +207,12 @@ impl Handler for ParallelHandler {
error = %fabro_sandbox::display_for_log(&e),
"parallel base checkpoint failed"
);
services.run.emitter.notice_with_tail(
RunNoticeLevel::Warn,
RunNoticeCode::ParallelBaseCheckpointFailed,
format!("Could not checkpoint base state before parallel branches: {e}"),
fabro_sandbox::default_redacted_output_tail(&e),
);
None
}
}

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

@ -105,7 +105,13 @@ impl Handler for PromptHandler {
let (response_text, stage_usage, backend_files_touched) =
if let Some(backend) = &self.backend {
let result = backend
.one_shot(node, &prompt, system_prompt.as_deref())
.one_shot(
node,
&prompt,
system_prompt.as_deref(),
&services.run.emitter,
&stage_scope,
)
.await;
match result {
Ok(CodergenResult::Full(outcome)) => return Ok(outcome),
@ -187,6 +193,7 @@ mod tests {
use tempfile::TempDir;
use super::*;
use crate::event::Emitter;
fn make_services() -> EngineServices {
EngineServices::test_default()
@ -211,7 +218,7 @@ mod tests {
let mut services = EngineServices::test_default();
services.run = services
.run
.with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1)))
.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());
logger.register(services.run.emitter.as_ref());
@ -267,7 +274,7 @@ mod tests {
_prompt: &str,
_context: &Context,
_thread_id: Option<&str>,
_emitter: &Arc<crate::event::Emitter>,
_emitter: &Arc<Emitter>,
_sandbox: &Arc<dyn Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, Error> {
@ -279,6 +286,8 @@ mod tests {
_node: &Node,
_prompt: &str,
_system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &StageScope,
) -> Result<CodergenResult, Error> {
Ok(CodergenResult::Text {
text: "one-shot response".to_string(),
@ -327,7 +336,7 @@ mod tests {
_prompt: &str,
_context: &Context,
_thread_id: Option<&str>,
_emitter: &Arc<crate::event::Emitter>,
_emitter: &Arc<Emitter>,
_sandbox: &Arc<dyn Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, Error> {
@ -339,6 +348,8 @@ mod tests {
_node: &Node,
_prompt: &str,
_system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &StageScope,
) -> Result<CodergenResult, Error> {
Ok(CodergenResult::Text {
text: "one-shot response".to_string(),
@ -384,7 +395,7 @@ mod tests {
_prompt: &str,
_context: &Context,
_thread_id: Option<&str>,
_emitter: &Arc<crate::event::Emitter>,
_emitter: &Arc<Emitter>,
_sandbox: &Arc<dyn fabro_agent::Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, Error> {
@ -396,6 +407,8 @@ mod tests {
_node: &Node,
prompt: &str,
system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &StageScope,
) -> Result<CodergenResult, Error> {
*self.captured_prompt.lock().unwrap() = Some(prompt.to_string());
*self.captured_system_prompt.lock().unwrap() = Some(system_prompt.map(String::from));

10
lib/crates/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -16,7 +16,7 @@ use tokio::time::sleep;
use crate::artifact::{normalize_durable_updates, offload_large_values, sync_artifacts_to_env};
use crate::artifact_snapshot::collect_artifacts;
use crate::artifact_upload::ArtifactSink;
use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel};
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::lifecycle::event::{stage_scope_for, stage_visit};
use crate::outcome::BilledModelUsage;
@ -125,7 +125,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
{
self.emitter.notice(
RunNoticeLevel::Warn,
"artifact_upload_failed",
RunNoticeCode::ArtifactUploadFailed,
format!("[node: {node_id}] artifact upload failed: {err}"),
);
return Ok(());
@ -151,7 +151,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
Err(e) => {
self.emitter.notice(
RunNoticeLevel::Warn,
"artifact_collection_failed",
RunNoticeCode::ArtifactCollectionFailed,
format!("[node: {node_id}] artifact collection failed: {e}"),
);
}
@ -174,7 +174,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
{
self.emitter.notice(
RunNoticeLevel::Warn,
"artifact_offload_failed",
RunNoticeCode::ArtifactOffloadFailed,
format!("[node: {node_id}] artifact offload failed: {e}"),
);
}
@ -187,7 +187,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
{
self.emitter.notice(
RunNoticeLevel::Warn,
"artifact_sync_failed",
RunNoticeCode::ArtifactSyncFailed,
format!("[node: {node_id}] artifact sync failed: {e}"),
);
}

43
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -14,7 +14,7 @@ use fabro_util::error::collect_causes;
use fabro_util::time::elapsed_ms;
use crate::artifact;
use crate::event::{Emitter, Event, RunNoticeLevel, StageScope};
use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope};
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::lifecycle::event::stage_scope_for;
use crate::outcome::BilledModelUsage;
@ -128,7 +128,10 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
None,
None,
);
self.emit_metadata_warning("checkpoint_metadata_write_failed", message);
self.emit_metadata_warning(
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
}
},
Err(err) => {
@ -145,7 +148,10 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
None,
None,
);
self.emit_metadata_warning("checkpoint_metadata_write_failed", message);
self.emit_metadata_warning(
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
}
}
}
@ -217,7 +223,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
Some(&scope),
);
self.emit_metadata_warning(
"checkpoint_metadata_write_failed",
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
None
@ -239,7 +245,10 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
None,
Some(&scope),
);
self.emit_metadata_warning("checkpoint_metadata_write_failed", message);
self.emit_metadata_warning(
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
None
}
}
@ -292,6 +301,12 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
error = %fabro_sandbox::display_for_log(&err),
"git push from run lifecycle failed"
);
self.emitter.notice_with_tail(
RunNoticeLevel::Warn,
RunNoticeCode::GitPushFailed,
format!("Failed to push run branch {branch}: {err}"),
exec_output_tail.clone(),
);
(false, exec_output_tail)
}
};
@ -321,7 +336,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
fabro_sandbox::default_redacted_output_tail(&err);
self.emitter.notice_with_tail(
RunNoticeLevel::Warn,
"git_diff_failed",
RunNoticeCode::GitDiffFailed,
format!("[node: {node_id}] git diff failed: {err}"),
exec_output_tail,
);
@ -395,7 +410,10 @@ impl GitLifecycle {
Some(snapshot.bytes),
scope,
);
self.emit_metadata_warning("checkpoint_metadata_push_failed", message);
self.emit_metadata_warning(
RunNoticeCode::CheckpointMetadataPushFailed,
message,
);
} else {
self.emit_metadata_snapshot_completed(
phase,
@ -421,7 +439,7 @@ impl GitLifecycle {
None,
scope,
);
self.emit_metadata_warning("checkpoint_metadata_write_failed", message);
self.emit_metadata_warning(RunNoticeCode::CheckpointMetadataWriteFailed, message);
None
}
}
@ -506,14 +524,9 @@ impl GitLifecycle {
}
}
fn emit_metadata_warning(&self, code: &str, message: String) {
fn emit_metadata_warning(&self, code: RunNoticeCode, message: String) {
if self.metadata_runtime.mark_metadata_degraded() {
self.emitter.emit(&Event::RunNotice {
level: RunNoticeLevel::Warn,
code: code.to_string(),
message,
exec_output_tail: None,
});
self.emitter.notice(RunNoticeLevel::Warn, code, message);
}
}
}

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

@ -10,7 +10,7 @@ use fabro_util::time::elapsed_ms;
use super::types::{Concluded, FinalizeOptions, Retroed};
use crate::error::Error;
use crate::event::{Event, RunNoticeLevel};
use crate::event::{Event, RunNoticeCode, RunNoticeLevel};
use crate::outcome::{Outcome, OutcomeExt, StageOutcome};
use crate::records::{Checkpoint, Conclusion, StageSummary};
use crate::run_metadata::MetadataSnapshot;
@ -239,7 +239,11 @@ pub async fn write_finalize_commit(
None,
None,
);
emit_metadata_warning(services, "checkpoint_metadata_write_failed", message);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
return;
}
};
@ -260,7 +264,11 @@ pub async fn write_finalize_commit(
None,
None,
);
emit_metadata_warning(services, "checkpoint_metadata_write_failed", message);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
return;
}
};
@ -281,7 +289,11 @@ pub async fn write_finalize_commit(
Some(snapshot.entry_count),
Some(snapshot.bytes),
);
emit_metadata_warning(services, "checkpoint_metadata_push_failed", message);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataPushFailed,
message,
);
} else {
emit_metadata_snapshot_completed(services, phase, meta_branch, started, &snapshot);
}
@ -300,7 +312,11 @@ pub async fn write_finalize_commit(
None,
None,
);
emit_metadata_warning(services, "checkpoint_metadata_write_failed", message);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
}
}
}
@ -363,7 +379,7 @@ fn emit_metadata_snapshot_failed(
});
}
fn emit_metadata_warning(services: &RunServices, code: &str, message: String) {
fn emit_metadata_warning(services: &RunServices, code: RunNoticeCode, message: String) {
if services.metadata_runtime.mark_metadata_degraded() {
services.emitter.notice(RunNoticeLevel::Warn, code, message);
}
@ -387,7 +403,7 @@ async fn compute_final_patch(
Err(err) => {
services.emitter.notice(
RunNoticeLevel::Warn,
"git_diff_failed",
RunNoticeCode::GitDiffFailed,
format!("final diff failed: {err}"),
);
None
@ -534,7 +550,7 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result<Con
if services.metadata_runtime.metadata_degraded() {
services.emitter.notice(
RunNoticeLevel::Warn,
"checkpoint_metadata_degraded",
RunNoticeCode::CheckpointMetadataDegraded,
"checkpoint metadata archive writes were degraded for this run".to_string(),
);
}
@ -556,9 +572,11 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result<Con
} else {
format!("sandbox preserved: {info}")
};
services
.emitter
.notice(RunNoticeLevel::Info, "sandbox_preserved", message);
services.emitter.notice(
RunNoticeLevel::Info,
RunNoticeCode::SandboxPreserved,
message,
);
}
if let Err(e) = cleanup_sandbox(
&services,
@ -572,7 +590,7 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result<Con
let exec_output_tail = fabro_sandbox::default_redacted_output_tail(&e);
services.emitter.notice_with_tail(
RunNoticeLevel::Warn,
"sandbox_cleanup_failed",
RunNoticeCode::SandboxCleanupFailed,
format!("sandbox cleanup failed: {}", e.display_with_causes()),
exec_output_tail,
);

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

@ -27,7 +27,7 @@ use tokio::time::timeout as tokio_timeout;
use super::types::{InitOptions, Initialized, LlmSpec, Persisted, SandboxEnvSpec};
use crate::devcontainer_bridge::{devcontainer_to_snapshot_config, run_devcontainer_lifecycle};
use crate::error::Error;
use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel};
use crate::git::RUN_BRANCH_PREFIX;
use crate::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter};
use crate::handler::{HandlerRegistry, default_registry, sandbox_cancel_token};
@ -155,7 +155,7 @@ fn resolve_worktree_plan(options: &mut InitOptions) -> Option<WorktreePlan> {
if let Some(env_name) = env_name {
options.emitter.notice(
RunNoticeLevel::Warn,
"dirty_worktree",
RunNoticeCode::DirtyWorktree,
format!("Uncommitted changes will not be included in the {env_name}."),
);
}
@ -200,6 +200,14 @@ fn resolve_worktree_plan(options: &mut InitOptions) -> Option<WorktreePlan> {
})
}
fn worktree_skipped_notice(mode: Option<WorktreeMode>) -> Option<(RunNoticeCode, &'static str)> {
matches!(mode, Some(WorktreeMode::Always)).then_some((
RunNoticeCode::WorktreeSkippedNoGit,
"Worktree mode `always` requested but no Git repository was found; running without a \
worktree.",
))
}
fn git_setup_intent(run_options: &RunOptions) -> GitSetupIntent {
if let Some(source) = run_options.fork_source_ref.as_ref() {
GitSetupIntent::ForkFromCheckpoint {
@ -239,11 +247,14 @@ async fn build_sandbox_env(
Ok(token) => {
env.insert("GITHUB_TOKEN".to_string(), token);
}
Err(e) => emitter.notice(
RunNoticeLevel::Warn,
"github_token_failed",
format!("Failed to mint GitHub token: {e}"),
),
Err(e) => {
tracing::warn!(error = %e, "Failed to mint GitHub token");
emitter.notice(
RunNoticeLevel::Warn,
RunNoticeCode::GithubTokenFailed,
format!("Failed to mint GitHub token: {e}"),
);
}
}
}
}
@ -516,6 +527,13 @@ pub async fn initialize(
))
};
if worktree_plan.is_some() && !worktree_created {
if let Some((code, message)) = worktree_skipped_notice(options.worktree_mode) {
tracing::warn!(
worktree_mode = ?options.worktree_mode,
"worktree skipped: cwd is not a git repository"
);
options.emitter.notice(RunNoticeLevel::Warn, code, message);
}
options.run_options.git = None;
}
let cleanup_guard = scopeguard::guard(Arc::clone(&sandbox), |sandbox| {
@ -593,7 +611,8 @@ pub async fn initialize(
.is_some();
if !has_run_branch {
let intent = git_setup_intent(&options.run_options);
if sandbox.origin_url().is_some() {
let sandbox_has_origin = sandbox.origin_url().is_some();
if sandbox_has_origin {
sandbox_git
.ensure_git_available(&*sandbox)
.await
@ -619,7 +638,16 @@ pub async fn initialize(
options.run_options.base_branch = info.base_branch;
}
}
Ok(None) => {}
Ok(None) => {
if sandbox_has_origin {
options.emitter.notice(
RunNoticeLevel::Warn,
RunNoticeCode::SandboxGitUnavailable,
"Sandbox could not set up Git despite a configured origin; running \
without checkpointing or PR support.",
);
}
}
Err(e) => {
return Err(Error::engine_with_source("Sandbox git setup failed", &e));
}
@ -702,7 +730,7 @@ pub async fn initialize(
if metadata_runtime.mark_metadata_degraded() {
options.emitter.notice(
RunNoticeLevel::Warn,
"checkpoint_metadata_write_failed",
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
}
@ -1011,6 +1039,17 @@ mod tests {
assert!(options.run_options.git.is_none());
}
#[test]
fn worktree_skipped_notice_only_warns_for_always() {
assert!(worktree_skipped_notice(None).is_none());
assert!(worktree_skipped_notice(Some(WorktreeMode::Clean)).is_none());
assert!(worktree_skipped_notice(Some(WorktreeMode::Dirty)).is_none());
assert!(worktree_skipped_notice(Some(WorktreeMode::Never)).is_none());
let (code, _) = worktree_skipped_notice(Some(WorktreeMode::Always)).unwrap();
assert_eq!(code, RunNoticeCode::WorktreeSkippedNoGit);
}
#[tokio::test]
async fn initialize_prepares_sandbox_and_uses_persisted_run_dir() {
let temp = tempfile::tempdir().unwrap();

4
lib/crates/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -14,7 +14,7 @@ use fabro_util::text::strip_goal_decoration;
use tracing::{debug, info, warn};
use super::types::{Concluded, Finalized, PullRequestOptions};
use crate::event::{Event, RunNoticeLevel};
use crate::event::{Event, RunNoticeCode, RunNoticeLevel};
use crate::outcome::{StageOutcome, format_cost as outcome_format_cost};
use crate::records::{Conclusion, RunSpec};
use crate::runtime_store::RunStoreHandle;
@ -675,7 +675,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) ->
.emit(&Event::PullRequestFailed { error: e.clone() });
services.emitter.notice(
RunNoticeLevel::Warn,
"pull_request_failed",
RunNoticeCode::PullRequestFailed,
format!("PR creation failed: {e}"),
);
}

2
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -6221,6 +6221,8 @@ mod real_llm {
_node: &Node,
prompt: &str,
_system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &fabro_workflow::event::StageScope,
) -> Result<CodergenResult, Error> {
self.complete(prompt).await
}