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 1

Make provider metadata event-derived and rewind append-only

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-02 10:26:12 -07:00
parent 12e316b785
commit f786f91fe7
15 changed files with 395 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
lib/crates/fabro-cli/src/commands/run/rewind.rs
View file

@ -7,7 +7,7 @@ use fabro_util::terminal::Styles;
use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event};
use fabro_workflow::git::MetadataStore;
use fabro_workflow::operations::{
RewindInput, RewindTarget, RunTimeline, build_timeline_or_rebuild,
RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild,
find_run_id_by_prefix_or_store, rewind,
};
use fabro_workflow::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt};
@ -62,12 +62,14 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs
&store,
&RewindInput {
run_id,
target,
target: target.clone(),
push: !args.no_push,
},
)?;
if let Some(run_info) = run_info.as_ref() {
reset_rewound_run_state(&store, durable_store.as_ref(), &run_id, &run_info.path).await?;
let entry = timeline.resolve(&target)?;
reset_rewound_run_state(&store, durable_store.as_ref(), &run_id, &run_info.path, entry)
.await?;
}
let run_id_string = run_id.to_string();
@ -105,6 +107,7 @@ async fn reset_rewound_run_state(
durable_store: &dyn fabro_store::Store,
run_id: &fabro_types::RunId,
run_dir: &std::path::Path,
entry: &TimelineEntry,
) -> Result<()> {
let existing_run_store = durable_store
.open_run_reader(run_id)
@ -131,18 +134,28 @@ async fn reset_rewound_run_state(
.context("failed to restore run record after rewind: missing run metadata")?;
let checkpoint = MetadataStore::read_checkpoint(git_store.repo_dir(), &run_id.to_string())?
.context("rewound metadata branch is missing checkpoint.json")?;
let previous_status = if let Some(run_store) = existing_run_store.as_ref() {
run_store
.get_status()
.await
.ok()
.flatten()
.map(|status| status.status.to_string())
} else {
None
};
let _ = std::fs::remove_file(run_dir.join("detached_failure.json"));
durable_store
.delete_run(run_id)
.await
.map_err(|err| anyhow::anyhow!("failed to reset durable store run: {err}"))?;
let run_dir_string = run_dir.to_string_lossy().to_string();
let run_store = durable_store
.create_run(run_id, run_record.created_at, Some(&run_dir_string))
.open_run(run_id)
.await
.map_err(|err| anyhow::anyhow!("failed to recreate durable store run: {err}"))?;
.map_err(|err| anyhow::anyhow!("failed to open durable store run for rewind reset: {err}"))?
.context("failed to reset durable store run after rewind: missing run store")?;
run_store
.reset_for_rewind()
.await
.map_err(|err| anyhow::anyhow!("failed to clear rewound run state: {err}"))?;
run_store
.put_run(&run_record)
.await
@ -159,6 +172,26 @@ async fn reset_rewound_run_state(
.await
.map_err(|err| anyhow::anyhow!("failed to restore graph after rewind: {err}"))?;
}
append_workflow_event(
run_store.as_ref(),
run_id,
&WorkflowRunEvent::RunRewound {
target_checkpoint_ordinal: entry.ordinal,
target_node_id: entry.node_name.clone(),
target_visit: entry.visit,
previous_status,
run_commit_sha: entry.run_commit_sha.clone(),
},
)
.await
.map_err(|err| anyhow::anyhow!("failed to append run rewound event: {err}"))?;
append_workflow_event(
run_store.as_ref(),
run_id,
&restored_checkpoint_event(&checkpoint),
)
.await
.map_err(|err| anyhow::anyhow!("failed to append restored checkpoint event: {err}"))?;
run_store
.put_status(&fabro_types::RunStatusRecord::new(
RunStatus::Submitted,
@ -180,6 +213,36 @@ async fn reset_rewound_run_state(
Ok(())
}
fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> WorkflowRunEvent {
let current_status = checkpoint
.node_outcomes
.get(&checkpoint.current_node)
.map_or_else(|| "success".to_string(), |outcome| outcome.status.to_string());
WorkflowRunEvent::CheckpointCompleted {
node_id: checkpoint.current_node.clone(),
status: current_status,
current_node: checkpoint.current_node.clone(),
completed_nodes: checkpoint.completed_nodes.clone(),
node_retries: checkpoint.node_retries.clone().into_iter().collect(),
context_values: checkpoint.context_values.clone().into_iter().collect(),
node_outcomes: checkpoint.node_outcomes.clone().into_iter().collect(),
next_node_id: checkpoint.next_node_id.clone(),
git_commit_sha: checkpoint.git_commit_sha.clone(),
loop_failure_signatures: checkpoint
.loop_failure_signatures
.iter()
.map(|(sig, count)| (sig.to_string(), *count))
.collect(),
restart_failure_signatures: checkpoint
.restart_failure_signatures
.iter()
.map(|(sig, count)| (sig.to_string(), *count))
.collect(),
node_visits: checkpoint.node_visits.clone().into_iter().collect(),
diff: None,
}
}
pub(crate) fn print_timeline(timeline: &RunTimeline, styles: &Styles) {
if timeline.entries.is_empty() {
eprintln!("No checkpoints found.");

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

@ -631,6 +631,7 @@ mod tests {
fn round_trip_agent_tool_call() {
let event = WorkflowRunEvent::Agent {
stage: "code".into(),
visit: 1,
event: AgentEvent::ToolCallStarted {
tool_name: "read_file".into(),
tool_call_id: "tc1".into(),

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

@ -482,6 +482,7 @@ mod tests {
fn agent_event(stage: &str, event: AgentEvent) -> WorkflowRunEvent {
WorkflowRunEvent::Agent {
stage: stage.into(),
visit: 1,
event,
session_id: None,
parent_session_id: None,

67
lib/crates/fabro-cli/tests/it/cmd/rewind.rs
View file

@ -4,7 +4,7 @@ use fabro_test::{fabro_snapshot, run_and_format, test_context};
use super::support::{
git_filters, git_stdout, output_stderr as support_stderr, run_branch_commits_since_base,
setup_git_backed_changed_run,
run_events, run_snapshot, setup_git_backed_changed_run,
};
#[test]
@ -122,3 +122,68 @@ fn rewind_target_updates_metadata_and_resume_hint() {
"rewound timeline should drop @2: {list}"
);
}
#[test]
fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() {
let context = test_context!();
let setup = setup_git_backed_changed_run(&context);
let before_events = run_events(&setup.run.run_dir);
assert!(
before_events.iter().any(|event| event.payload.as_value()["event"] == "run.completed"),
"setup run should be completed before rewind"
);
let mut cmd = context.command();
cmd.current_dir(&setup.repo_dir);
cmd.args(["rewind", &setup.run.run_id, "@1", "--no-push"]);
let output = cmd.output().expect("rewind should execute");
assert!(
output.status.success(),
"rewind should succeed\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
support_stderr(&output),
);
let after_events = run_events(&setup.run.run_dir);
assert_eq!(
after_events.len(),
before_events.len() + 3,
"rewind should append run.rewound, checkpoint.completed, and run.submitted"
);
assert_eq!(
after_events[..before_events.len()]
.iter()
.map(|event| event.payload.as_value()["event"].as_str().unwrap())
.collect::<Vec<_>>(),
before_events
.iter()
.map(|event| event.payload.as_value()["event"].as_str().unwrap())
.collect::<Vec<_>>(),
"rewind should preserve the prior event prefix"
);
assert_eq!(
after_events[before_events.len()].payload.as_value()["event"],
"run.rewound"
);
assert_eq!(
after_events[before_events.len() + 1].payload.as_value()["event"],
"checkpoint.completed"
);
assert_eq!(
after_events[before_events.len() + 2].payload.as_value()["event"],
"run.submitted"
);
let snapshot = run_snapshot(&setup.run.run_dir);
assert_eq!(
snapshot.status.as_ref().map(|status| &status.status),
Some(&fabro_types::RunStatus::Submitted)
);
assert!(snapshot.conclusion.is_none(), "rewind should clear conclusion");
assert!(snapshot.final_patch.is_none(), "rewind should clear final patch");
assert!(snapshot.pull_request.is_none(), "rewind should clear pull request");
assert!(
snapshot.nodes.is_empty(),
"rewind should clear node snapshots that belonged to the prior execution"
);
}

8
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -4,7 +4,7 @@ use std::process::Output;
use std::sync::Arc;
use std::time::{Duration, Instant};
use fabro_store::{RunSnapshot, RunStore, SlateStore, Store};
use fabro_store::{EventEnvelope, RunSnapshot, RunStore, SlateStore, Store};
use fabro_test::TestContext;
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
@ -494,6 +494,12 @@ pub(crate) fn run_snapshot(run_dir: &Path) -> RunSnapshot {
.expect("run store snapshot should exist")
}
pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
run_store(run_dir)
.and_then(|store| block_on(store.list_events()).ok())
.expect("run store events should exist")
}
pub(crate) fn git_stdout(repo_dir: &Path, args: &[&str]) -> String {
stdout(&git_success(repo_dir, args))
}

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

@ -122,6 +122,8 @@ pub trait RunStore: Send + Sync {
async fn put_pull_request(&self, record: &PullRequestRecord) -> Result<()>;
async fn get_pull_request(&self) -> Result<Option<PullRequestRecord>>;
async fn reset_for_rewind(&self) -> Result<()>;
async fn append_event(&self, payload: &EventPayload) -> Result<u32>;
async fn list_events(&self) -> Result<Vec<EventEnvelope>>;
async fn list_events_from(&self, seq: u32) -> Result<Vec<EventEnvelope>>;

12
lib/crates/fabro-store/src/memory.rs
View file

@ -535,6 +535,18 @@ impl RunStore for InMemoryRunStore {
self.get_json(keys::pull_request()).await
}
async fn reset_for_rewind(&self) -> Result<()> {
let mut data = self.data.lock().await;
data.retain(|key, _| {
key == keys::init()
|| key == keys::run()
|| key == keys::start()
|| key == keys::graph()
|| key.starts_with(keys::EVENTS_PREFIX)
});
Ok(())
}
async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
payload.validate(&self.run_id)?;

38
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -468,6 +468,32 @@ impl RunStore for SlateRunStore {
self.inner.db.get_json(keys::pull_request()).await
}
async fn reset_for_rewind(&self) -> Result<()> {
let db = self.inner.db.writer()?;
for key in [
keys::status(),
keys::checkpoint(),
keys::conclusion(),
keys::retro(),
keys::sandbox(),
keys::final_patch(),
keys::pull_request(),
keys::retro_prompt(),
keys::retro_response(),
] {
db.delete(key).await?;
}
for prefix in [
b"nodes/".as_slice(),
keys::CHECKPOINTS_PREFIX.as_bytes(),
keys::ARTIFACT_VALUES_PREFIX.as_bytes(),
keys::ARTIFACT_NODES_PREFIX.as_bytes(),
] {
delete_prefix(db, prefix).await?;
}
Ok(())
}
async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
payload.validate(&self.inner.run_id)?;
let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst);
@ -762,6 +788,18 @@ async fn put_bytes(db: &slatedb::Db, key: &str, value: &[u8]) -> Result<()> {
Ok(())
}
async fn delete_prefix(db: &slatedb::Db, prefix: &[u8]) -> Result<()> {
let mut iter = db.scan_prefix(prefix).await?;
let mut keys = Vec::new();
while let Some(entry) = iter.next().await? {
keys.push(key_to_string(&entry.key)?);
}
for key in keys {
db.delete(key).await?;
}
Ok(())
}
async fn get_bytes(db: &slatedb::Db, key: &str) -> Result<Option<Bytes>> {
Ok(db.get(key).await?)
}

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

@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicI64, Ordering};
use anyhow::{Context, Result};
use chrono::{SecondsFormat, Utc};
use fabro_store::{EventPayload, RunStore};
use fabro_store::{EventPayload, NodeVisitRef, RunStore};
use fabro_types::RunId;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
@ -95,17 +95,18 @@ pub enum WorkflowRunEvent {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunPaused {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunRemoving {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunDead {
RunRewound {
target_checkpoint_ordinal: usize,
target_node_id: String,
target_visit: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
previous_status: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
run_commit_sha: Option<String>,
},
WorkflowRunCompleted {
duration_ms: u64,
@ -307,6 +308,7 @@ pub enum WorkflowRunEvent {
},
Prompt {
stage: String,
visit: u32,
text: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
mode: Option<String>,
@ -326,6 +328,7 @@ pub enum WorkflowRunEvent {
/// Forwarded from an agent session, tagged with the workflow stage.
Agent {
stage: String,
visit: u32,
event: AgentEvent,
#[serde(default, skip_serializing_if = "Option::is_none")]
session_id: Option<String>,
@ -439,6 +442,7 @@ pub enum WorkflowRunEvent {
},
AgentCliStarted {
node_id: String,
visit: u32,
mode: String,
provider: String,
model: String,
@ -539,14 +543,24 @@ impl WorkflowRunEvent {
Self::RunRunning { reason } => {
info!(?reason, "Run running");
}
Self::RunPaused { reason } => {
info!(?reason, "Run paused");
}
Self::RunRemoving { reason } => {
info!(?reason, "Run removing");
}
Self::RunDead { reason } => {
warn!(?reason, "Run dead");
Self::RunRewound {
target_checkpoint_ordinal,
target_node_id,
target_visit,
previous_status,
run_commit_sha,
} => {
info!(
target_checkpoint_ordinal,
target_node_id,
target_visit,
previous_status = previous_status.as_deref().unwrap_or(""),
run_commit_sha = run_commit_sha.as_deref().unwrap_or(""),
"Run rewound"
);
}
Self::WorkflowRunCompleted {
duration_ms,
@ -787,6 +801,7 @@ impl WorkflowRunEvent {
mode,
provider,
model,
..
} => {
debug!(
stage,
@ -1065,9 +1080,8 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str {
WorkflowRunEvent::RunSubmitted { .. } => "run.submitted",
WorkflowRunEvent::RunStarting { .. } => "run.starting",
WorkflowRunEvent::RunRunning { .. } => "run.running",
WorkflowRunEvent::RunPaused { .. } => "run.paused",
WorkflowRunEvent::RunRemoving { .. } => "run.removing",
WorkflowRunEvent::RunDead { .. } => "run.dead",
WorkflowRunEvent::RunRewound { .. } => "run.rewound",
WorkflowRunEvent::WorkflowRunCompleted { .. } => "run.completed",
WorkflowRunEvent::WorkflowRunFailed { .. } => "run.failed",
WorkflowRunEvent::RunNotice { .. } => "run.notice",
@ -1314,12 +1328,16 @@ fn extract_envelope_fields(event: &WorkflowRunEvent) -> EnvelopeFields {
let mut fields = tagged_variant_fields(event);
let node_id = remove_string(&mut fields, "stage");
let node_label = default_node_label(node_id.as_ref(), None);
let visit = fields.remove("visit");
fields.remove("session_id");
fields.remove("parent_session_id");
let properties = fields.remove("event").map_or_else(
let mut properties = fields.remove("event").map_or_else(
|| Value::Object(Map::new()),
|value| Value::Object(tagged_variant_fields_from_value(value)),
);
if let (Some(visit), Value::Object(map)) = (visit, &mut properties) {
map.insert("visit".to_string(), visit);
}
EnvelopeFields {
session_id: session_id.clone(),
parent_session_id: parent_session_id.clone(),
@ -1552,6 +1570,15 @@ impl StoreProgressLogger {
if let Err(err) = run_store.append_event(&payload).await {
tracing::warn!(error = %err, "Failed to append event to run store");
}
if let Err(err) =
project_provider_used_from_event_payload(run_store.as_ref(), &payload)
.await
{
tracing::warn!(
error = %err,
"Failed to project provider metadata from event"
);
}
}
StoreProgressCommand::Flush(tx) => {
let _ = tx.send(());
@ -1597,6 +1624,80 @@ impl StoreProgressLogger {
}
}
async fn project_provider_used_from_event_payload(
run_store: &dyn RunStore,
payload: &EventPayload,
) -> Result<()> {
let value = payload.as_value();
let Some(event_name) = value.get("event").and_then(Value::as_str) else {
return Ok(());
};
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let Some(properties) = value.get("properties").and_then(Value::as_object) else {
return Ok(());
};
let Some(visit) = properties
.get("visit")
.and_then(Value::as_u64)
.and_then(|visit| u32::try_from(visit).ok())
else {
return Ok(());
};
let provider_used = match event_name {
"stage.prompt" => {
let mut provider_used = Map::new();
if let Some(mode) = properties.get("mode").and_then(Value::as_str) {
provider_used.insert("mode".to_string(), Value::String(mode.to_string()));
}
if let Some(provider) = properties.get("provider").and_then(Value::as_str) {
provider_used.insert("provider".to_string(), Value::String(provider.to_string()));
}
if let Some(model) = properties.get("model").and_then(Value::as_str) {
provider_used.insert("model".to_string(), Value::String(model.to_string()));
}
(!provider_used.is_empty()).then_some(Value::Object(provider_used))
}
"agent.session.started" => {
let mut provider_used = Map::new();
provider_used.insert("mode".to_string(), Value::String("agent".to_string()));
if let Some(provider) = properties.get("provider").and_then(Value::as_str) {
provider_used.insert("provider".to_string(), Value::String(provider.to_string()));
}
if let Some(model) = properties.get("model").and_then(Value::as_str) {
provider_used.insert("model".to_string(), Value::String(model.to_string()));
}
Some(Value::Object(provider_used))
}
"agent.cli.started" => {
let mut provider_used = Map::new();
provider_used.insert("mode".to_string(), Value::String("cli".to_string()));
if let Some(provider) = properties.get("provider").and_then(Value::as_str) {
provider_used.insert("provider".to_string(), Value::String(provider.to_string()));
}
if let Some(model) = properties.get("model").and_then(Value::as_str) {
provider_used.insert("model".to_string(), Value::String(model.to_string()));
}
if let Some(command) = properties.get("command").and_then(Value::as_str) {
provider_used.insert("command".to_string(), Value::String(command.to_string()));
}
Some(Value::Object(provider_used))
}
_ => None,
};
let Some(provider_used) = provider_used else {
return Ok(());
};
run_store
.put_node_provider_used(&NodeVisitRef { node_id, visit }, &provider_used)
.await
.map_err(anyhow::Error::from)
}
/// Current time as epoch milliseconds.
fn epoch_millis() -> i64 {
let millis = std::time::SystemTime::now()
@ -1856,6 +1957,7 @@ mod tests {
&fixtures::RUN_4,
&WorkflowRunEvent::Agent {
stage: "code".to_string(),
visit: 2,
event: AgentEvent::ToolCallStarted {
tool_name: "read_file".to_string(),
tool_call_id: "call_1".to_string(),
@ -1873,6 +1975,7 @@ mod tests {
assert_eq!(envelope.parent_session_id.as_deref(), Some("ses_parent"));
assert_eq!(envelope.properties["tool_name"], "read_file");
assert_eq!(envelope.properties["tool_call_id"], "call_1");
assert_eq!(envelope.properties["visit"], 2);
}
#[test]
@ -1975,6 +2078,7 @@ mod tests {
assert_eq!(
event_name(&WorkflowRunEvent::Agent {
stage: "code".to_string(),
visit: 1,
event: AgentEvent::SubAgentSpawned {
agent_id: "a1".to_string(),
depth: 1,

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

@ -200,33 +200,6 @@ pub(crate) fn truncate(s: &str, max_chars: usize) -> &str {
}
}
pub(crate) async fn sync_provider_used_to_store(
stage_dir: &Path,
node_ref: &NodeVisitRef<'_>,
services: &EngineServices,
) -> Result<(), FabroError> {
let Some(ref store) = services.run_store else {
return Ok(());
};
let path = stage_dir.join("provider_used.json");
let json = match fs::read_to_string(&path).await {
Ok(json) => json,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(err) => {
return Err(FabroError::handler(format!(
"Failed to read provider_used.json: {err}"
)));
}
};
let value: serde_json::Value = serde_json::from_str(&json)
.map_err(|err| FabroError::handler(format!("Failed to parse provider_used.json: {err}")))?;
store
.put_node_provider_used(node_ref, &value)
.await
.map_err(|err| FabroError::handler(err.to_string()))
}
/// Shared simulate implementation for LLM-backed handlers (agent & prompt).
/// Produces a simulated outcome with standard context updates.
pub(crate) fn simulate_llm_handler(node: &Node) -> Outcome {
@ -329,10 +302,7 @@ impl Handler for AgentHandler {
)
.await;
match result {
Ok(CodergenResult::Full(outcome)) => {
sync_provider_used_to_store(&stage_dir, &node_ref, services).await?;
return Ok(outcome);
}
Ok(CodergenResult::Full(outcome)) => return Ok(outcome),
Ok(CodergenResult::Text {
text,
usage,
@ -364,8 +334,6 @@ impl Handler for AgentHandler {
} else {
fs::write(stage_dir.join("response.md"), &response_text).await?;
}
sync_provider_used_to_store(&stage_dir, &node_ref, services).await?;
// 7. Build and write status
let mut outcome = Outcome::success();
outcome.notes = Some(format!("Stage completed: {}", node.id));
@ -433,7 +401,11 @@ mod tests {
EngineServices::test_default()
}
async fn make_services_with_run_store() -> (EngineServices, Arc<dyn RunStore>) {
async fn make_services_with_run_store() -> (
EngineServices,
Arc<dyn RunStore>,
crate::event::StoreProgressLogger,
) {
let store = InMemoryStore::default();
let run_store = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
@ -443,7 +415,9 @@ mod tests {
run_store: Some(Arc::clone(&run_store)),
..EngineServices::test_default()
};
(services, run_store)
let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store));
logger.register(services.emitter.as_ref());
(services, run_store, logger)
}
fn test_context() -> Context {
@ -706,27 +680,32 @@ mod tests {
}
#[tokio::test]
async fn codergen_handler_persists_provider_used_in_run_store() {
struct ProviderUsedBackend;
async fn codergen_handler_projects_provider_used_from_agent_session_events() {
struct ProviderEventBackend;
#[async_trait]
impl CodergenBackend for ProviderUsedBackend {
impl CodergenBackend for ProviderEventBackend {
async fn run(
&self,
_node: &Node,
node: &Node,
_prompt: &str,
_context: &Context,
context: &Context,
_thread_id: Option<&str>,
_emitter: &Arc<EventEmitter>,
stage_dir: &Path,
emitter: &Arc<EventEmitter>,
_stage_dir: &Path,
_sandbox: &Arc<dyn fabro_agent::Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, FabroError> {
std::fs::write(
stage_dir.join("provider_used.json"),
r#"{"mode":"agent","provider":"openai","model":"gpt-5.4"}"#,
)
.unwrap();
emitter.emit(&crate::event::WorkflowRunEvent::Agent {
stage: node.id.clone(),
visit: crate::run_dir::visit_from_context(context) as u32,
event: fabro_agent::AgentEvent::SessionStarted {
provider: Some("openai".to_string()),
model: Some("gpt-5.4".to_string()),
},
session_id: Some("session_123".to_string()),
parent_session_id: None,
});
Ok(CodergenResult::Text {
text: "done".to_string(),
usage: None,
@ -736,17 +715,18 @@ mod tests {
}
}
let handler = AgentHandler::new(Some(Box::new(ProviderUsedBackend)));
let handler = AgentHandler::new(Some(Box::new(ProviderEventBackend)));
let node = Node::new("step");
let context = test_context();
let graph = Graph::new("test");
let tmp = TempDir::new().unwrap();
let (services, run_store) = make_services_with_run_store().await;
let (services, run_store, logger) = make_services_with_run_store().await;
handler
.execute(&node, &context, &graph, tmp.path(), &services)
.await
.unwrap();
logger.flush().await;
let snapshot = run_store
.get_node(&NodeVisitRef {

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

@ -23,6 +23,7 @@ use crate::error::FabroError;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::outcome::StageUsage;
use crate::outcome::compute_stage_cost;
use crate::run_dir::visit_from_context;
use fabro_graphviz::graph::Node;
fn build_profile(model: &str, provider: Provider) -> Box<dyn AgentProfile> {
@ -38,6 +39,10 @@ fn build_profile(model: &str, provider: Provider) -> Box<dyn AgentProfile> {
}
}
fn current_visit(context: &Context) -> u32 {
u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX)
}
/// Shared state for tracking file modifications from agent tool calls.
struct FileTracking {
/// Maps tool_call_id → file_path for in-flight write/edit calls.
@ -83,6 +88,7 @@ fn track_file_event(event: &AgentEvent, state: &mut FileTracking) {
fn spawn_event_forwarder(
session: &Session,
node_id: String,
visit: u32,
emitter: Arc<EventEmitter>,
file_tracking: Arc<Mutex<FileTracking>>,
) {
@ -101,6 +107,7 @@ fn spawn_event_forwarder(
{
emitter.emit(&WorkflowRunEvent::Agent {
stage: node_id.clone(),
visit,
event: event.event.clone(),
session_id: Some(event.session_id.clone()),
parent_session_id: event.parent_session_id.clone(),
@ -424,7 +431,7 @@ impl CodergenBackend for AgentApiBackend {
context: &Context,
thread_id: Option<&str>,
emitter: &Arc<EventEmitter>,
stage_dir: &std::path::Path,
_stage_dir: &std::path::Path,
sandbox: &Arc<dyn Sandbox>,
tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, FabroError> {
@ -479,6 +486,7 @@ impl CodergenBackend for AgentApiBackend {
spawn_event_forwarder(
&session,
node.id.clone(),
current_visit(context),
Arc::clone(emitter),
Arc::clone(&file_tracking),
);
@ -486,6 +494,7 @@ impl CodergenBackend for AgentApiBackend {
// Emit Prompt event before processing
emitter.emit(&WorkflowRunEvent::Prompt {
stage: node.id.clone(),
visit: current_visit(context),
text: prompt.to_string(),
mode: Some("agent".to_string()),
provider: Some(actual_provider.as_str().to_string()),
@ -552,6 +561,7 @@ impl CodergenBackend for AgentApiBackend {
spawn_event_forwarder(
&session,
node.id.clone(),
current_visit(context),
Arc::clone(emitter),
Arc::clone(&file_tracking),
);
@ -633,15 +643,6 @@ impl CodergenBackend for AgentApiBackend {
(v, s.last.clone())
};
let provider_used = serde_json::json!({
"mode": "agent",
"provider": actual_provider.as_str(),
"model": &actual_model,
});
if let Ok(json) = serde_json::to_string_pretty(&provider_used) {
let _ = std::fs::write(stage_dir.join("provider_used.json"), json);
}
// Cache session back for reuse on success.
if let Some(key) = reuse_key {
self.sessions.lock().unwrap().insert(key, session);

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

@ -15,6 +15,7 @@ use crate::error::FabroError;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::outcome::StageUsage;
use crate::outcome::compute_stage_cost;
use crate::run_dir::visit_from_context;
use fabro_graphviz::graph::Node;
/// Maps a provider to its corresponding CLI tool metadata.
@ -56,6 +57,10 @@ impl AgentCli {
}
}
fn current_visit(context: &Context) -> u32 {
u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX)
}
/// Ensure the CLI tool for the given provider is installed in the sandbox.
///
/// Checks if the CLI binary exists; if not, installs Node.js (if missing) and
@ -496,6 +501,7 @@ impl CodergenBackend for AgentCliBackend {
let command = cli_command_for_provider(provider, model, &prompt_path);
emitter.emit(&WorkflowRunEvent::AgentCliStarted {
node_id: node.id.clone(),
visit: current_visit(_context),
mode: "cli".to_string(),
provider: provider.as_str().to_string(),
model: model.to_string(),

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

@ -15,8 +15,7 @@ use fabro_graphviz::graph::{Graph, Node};
use tokio::fs;
use super::agent::{
CodergenBackend, CodergenResult, expand_variables, extract_status_fields,
sync_provider_used_to_store, truncate,
CodergenBackend, CodergenResult, expand_variables, extract_status_fields, truncate,
};
use super::{EngineServices, Handler};
@ -107,6 +106,20 @@ impl Handler for PromptHandler {
fs::write(stage_dir.join("prompt.md"), &prompt).await?;
}
let prompt_provider = node
.provider()
.map(String::from)
.or_else(|| Some(Provider::default_from_env().as_str().to_string()));
let prompt_model = node.model().map(String::from);
services.emitter.emit(&WorkflowRunEvent::Prompt {
stage: node.id.clone(),
visit: u32::try_from(visit).unwrap_or(u32::MAX),
text: prompt.clone(),
mode: Some("prompt".to_string()),
provider: prompt_provider.clone(),
model: prompt_model.clone(),
});
// 3. Call LLM backend (one_shot)
let (response_text, stage_usage, backend_files_touched) =
if let Some(backend) = &self.backend {
@ -114,10 +127,7 @@ impl Handler for PromptHandler {
.one_shot(node, &prompt, system_prompt.as_deref(), &stage_dir)
.await;
match result {
Ok(CodergenResult::Full(outcome)) => {
sync_provider_used_to_store(&stage_dir, &node_ref, services).await?;
return Ok(outcome);
}
Ok(CodergenResult::Full(outcome)) => return Ok(outcome),
Ok(CodergenResult::Text {
text,
usage,
@ -167,7 +177,6 @@ impl Handler for PromptHandler {
} else {
fs::write(stage_dir.join("response.md"), &response_text).await?;
}
sync_provider_used_to_store(&stage_dir, &node_ref, services).await?;
// 5. Build and write status
let mut outcome = Outcome::success();
@ -205,7 +214,11 @@ mod tests {
EngineServices::test_default()
}
async fn make_services_with_run_store() -> (EngineServices, Arc<dyn RunStore>) {
async fn make_services_with_run_store() -> (
EngineServices,
Arc<dyn RunStore>,
crate::event::StoreProgressLogger,
) {
let store = InMemoryStore::default();
let run_store = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
@ -215,7 +228,9 @@ mod tests {
run_store: Some(Arc::clone(&run_store)),
..EngineServices::test_default()
};
(services, run_store)
let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store));
logger.register(services.emitter.as_ref());
(services, run_store, logger)
}
#[tokio::test]
@ -318,7 +333,7 @@ mod tests {
}
#[tokio::test]
async fn prompt_handler_persists_provider_used_in_run_store() {
async fn prompt_handler_projects_provider_used_from_prompt_events() {
use fabro_agent::Sandbox;
struct ProviderOneShotBackend;
@ -344,13 +359,8 @@ mod tests {
_node: &Node,
_prompt: &str,
_system_prompt: Option<&str>,
stage_dir: &Path,
_stage_dir: &Path,
) -> Result<CodergenResult, FabroError> {
std::fs::write(
stage_dir.join("provider_used.json"),
r#"{"mode":"prompt","provider":"openai","model":"gpt-5.4"}"#,
)
.unwrap();
Ok(CodergenResult::Text {
text: "one-shot response".to_string(),
usage: None,
@ -369,12 +379,13 @@ mod tests {
let context = Context::new();
let graph = Graph::new("test");
let tmp = TempDir::new().unwrap();
let (services, run_store) = make_services_with_run_store().await;
let (services, run_store, logger) = make_services_with_run_store().await;
handler
.execute(&node, &context, &graph, tmp.path(), &services)
.await
.unwrap();
logger.flush().await;
let snapshot = run_store
.get_node(&NodeVisitRef {

1
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -76,6 +76,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
if !event.event.is_streaming_noise() {
emitter.emit(&WorkflowRunEvent::Agent {
stage: "retro".to_string(),
visit: 1,
event: event.event.clone(),
session_id: Some(event.session_id.clone()),
parent_session_id: event.parent_session_id.clone(),

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

@ -12063,6 +12063,7 @@ impl Handler for KeepaliveHandler {
tokio::time::sleep(std::time::Duration::from_millis(self.interval_ms)).await;
services.emitter.emit(&WorkflowRunEvent::Prompt {
stage: node.id.clone(),
visit: 1,
text: "keepalive".to_string(),
mode: None,
provider: None,