From e901cd3a813dc82f16738ab6bba9154b55ae1fea Mon Sep 17 00:00:00 2001 From: "fabro-sh-0530[bot]" <281434857+fabro-sh-0530[bot]@users.noreply.github.com> Date: Tue, 5 May 2026 09:18:05 -0400 Subject: [PATCH] Surface silent fallback warnings in runs and logs (#205) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### 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 Co-authored-by: Bryan Helmkamp --- lib/crates/fabro-cli/src/commands/run/logs.rs | 51 +++++++++++++--- .../src/commands/run/run_progress/event.rs | 7 ++- .../src/commands/run/run_progress/mod.rs | 14 +++-- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 17 ++++++ lib/crates/fabro-cli/tests/it/cmd/dump.rs | 6 +- lib/crates/fabro-cli/tests/it/cmd/run.rs | 1 + .../tests/it/workflow/dry_run_examples.rs | 5 ++ lib/crates/fabro-sandbox/src/local.rs | 59 ++++++++++++++----- lib/crates/fabro-types/src/lib.rs | 2 +- lib/crates/fabro-types/src/run_event/infra.rs | 53 +++++++++++++---- lib/crates/fabro-types/src/run_event/mod.rs | 4 +- lib/crates/fabro-workflow/src/event.rs | 2 +- .../fabro-workflow/src/event/convert.rs | 6 +- .../fabro-workflow/src/event/emitter.rs | 15 ++--- .../fabro-workflow/src/handler/agent.rs | 2 + .../fabro-workflow/src/handler/llm/api.rs | 20 ++++--- .../fabro-workflow/src/handler/llm/cli.rs | 6 +- .../fabro-workflow/src/handler/parallel.rs | 8 ++- .../fabro-workflow/src/handler/prompt.rs | 23 ++++++-- .../fabro-workflow/src/lifecycle/artifact.rs | 10 ++-- .../fabro-workflow/src/lifecycle/git.rs | 43 +++++++++----- .../fabro-workflow/src/pipeline/finalize.rs | 42 +++++++++---- .../fabro-workflow/src/pipeline/initialize.rs | 59 +++++++++++++++---- .../src/pipeline/pull_request.rs | 4 +- .../fabro-workflow/tests/it/integration.rs | 2 + 25 files changed, 337 insertions(+), 124 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 8f01378cc..7d7e9063b 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -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::().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(°raded_notice, &styles, &mut state).unwrap(); assert!( degraded.contains("metadata snapshots disabled"), "got: {degraded}" diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index 9e40f0d8b..ec56dce46 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -527,7 +527,7 @@ fn display_value(value: &Value) -> Option { 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" )); } diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index 2004ac0ba..7e7df7f2d 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -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::() + .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, }); diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index e703e3207..7c6728f39 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -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", diff --git a/lib/crates/fabro-cli/tests/it/cmd/dump.rs b/lib/crates/fabro-cli/tests/it/cmd/dump.rs index 12b13ee9a..c541b59eb 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/dump.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/dump.rs @@ -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 diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 19d876cfc..2aa614979 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -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] diff --git a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs index bb8a50c85..c3f8d836c 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs @@ -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] diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index 8b69b7377..66d084b5f 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/lib/crates/fabro-sandbox/src/local.rs @@ -126,6 +126,19 @@ fn process_env_vars() -> Vec<(String, String)> { std::env::vars().collect() } +async fn drain_pipe(mut pipe: Option, 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> { + 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(); diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 22675855c..e5d3b65a5 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -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}; diff --git a/lib/crates/fabro-types/src/run_event/infra.rs b/lib/crates/fabro-types/src/run_event/infra.rs index 8ee1bc32c..77689087b 100644 --- a/lib/crates/fabro-types/src/run_event/infra.rs +++ b/lib/crates/fabro-types/src/run_event/infra.rs @@ -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( diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index 9d5fbf11d..647efc381 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -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, }), diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index f5120e62e..e64b2ae30 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -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; diff --git a/lib/crates/fabro-workflow/src/event/convert.rs b/lib/crates/fabro-workflow/src/event/convert.rs index 4aa9df5b1..255d235ea 100644 --- a/lib/crates/fabro-workflow/src/event/convert.rs +++ b/lib/crates/fabro-workflow/src/event/convert.rs @@ -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()), }); diff --git a/lib/crates/fabro-workflow/src/event/emitter.rs b/lib/crates/fabro-workflow/src/event/emitter.rs index a1bba727a..dc50a56dd 100644 --- a/lib/crates/fabro-workflow/src/event/emitter.rs +++ b/lib/crates/fabro-workflow/src/event/emitter.rs @@ -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, - message: impl Into, - ) { + pub fn notice(&self, level: RunNoticeLevel, code: RunNoticeCode, message: impl Into) { 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, + code: RunNoticeCode, message: impl Into, exec_output_tail: Option, ) { self.emit(&Event::RunNotice { level, - code: code.into(), + code: code.to_string(), message: message.into(), exec_output_tail, }); diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 676e2f142..7e3a4be0c 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -52,6 +52,8 @@ pub trait CodergenBackend: Send + Sync { _node: &Node, _prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { Err(Error::Validation( "one_shot mode not supported by this backend".into(), diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 5d21c7d32..9247c19ec 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -285,6 +285,8 @@ impl CodergenBackend for AgentApiBackend { node: &Node, prompt: &str, system_prompt: Option<&str>, + emitter: &Arc, + stage_scope: &StageScope, ) -> Result { 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(|| { diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 5036eec97..7fcf60ed8 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -810,9 +810,13 @@ impl CodergenBackend for BackendRouter { node: &Node, prompt: &str, system_prompt: Option<&str>, + emitter: &Arc, + stage_scope: &StageScope, ) -> Result { // 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 } } diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 39862586c..c58104f83 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -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 } } diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index a76099b80..caad1e058 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -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, + _emitter: &Arc, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { @@ -279,6 +286,8 @@ mod tests { _node: &Node, _prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { 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, + _emitter: &Arc, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { @@ -339,6 +348,8 @@ mod tests { _node: &Node, _prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { 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, + _emitter: &Arc, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { @@ -396,6 +407,8 @@ mod tests { _node: &Node, prompt: &str, system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { *self.captured_prompt.lock().unwrap() = Some(prompt.to_string()); *self.captured_system_prompt.lock().unwrap() = Some(system_prompt.map(String::from)); diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index f65fcee86..50d6ccded 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -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 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 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 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 for ArtifactLifecycle { { self.emitter.notice( RunNoticeLevel::Warn, - "artifact_sync_failed", + RunNoticeCode::ArtifactSyncFailed, format!("[node: {node_id}] artifact sync failed: {e}"), ); } diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 0e0e9f195..2f9d5bfdb 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -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 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 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 for GitLifecycle { Some(&scope), ); self.emit_metadata_warning( - "checkpoint_metadata_write_failed", + RunNoticeCode::CheckpointMetadataWriteFailed, message, ); None @@ -239,7 +245,10 @@ impl RunLifecycle 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 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 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); } } } diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 927d2d833..cb58bd345 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -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 Result Result Option { 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 { }) } +fn worktree_skipped_notice(mode: Option) -> 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(); diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 35f27aa92..37aec323e 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -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}"), ); } diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index f6ece331b..b89fc7f5a 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -6221,6 +6221,8 @@ mod real_llm { _node: &Node, prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &fabro_workflow::event::StageScope, ) -> Result { self.complete(prompt).await }