scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-11 22:53:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge remote-tracking branch 'origin/main'

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-02 14:29:31 -07:00
commit 9d3a8243c3
No known key found for this signature in database
30 changed files with 1385 additions and 121 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

457
docs-internal/plan-events-as-source-of-truth-follow-ups.md Normal file
View file

@ -0,0 +1,457 @@
# Plan: Events as Source of Truth Follow-Ups
Close the remaining event-contract gaps required before we can execute `~/.claude/plans/memoized-pondering-knuth.md` and make projected `RunState` the primary read model.
## Context
`docs-internal/plan-events-as-source-of-truth.md` has mostly landed. The event stream is materially stronger now, but it still does not cover every field that `memoized-pondering-knuth.md` wants to derive from events.
The next step is not a broad store refactor. It is a narrow follow-up pass that:
- finishes the missing event coverage
- makes the remaining source-of-truth boundaries explicit
- leaves `memoized-pondering-knuth.md` with no hidden event-contract assumptions
This plan is a prerequisite plan, not the full event-sourced store migration.
## Simplification Rules
These rules govern every follow-up event change in this document:
- enrich an existing semantic event before inventing a new one
- keep run-level summary data on run-level events
- keep handler-specific metadata on handler-specific events
- avoid storage-shaped event names like `*.recorded`, `*.persisted`, or `*.written`
- explicitly retain non-event concerns instead of half-eventizing them
If a proposed event change violates one of those rules, prefer a simpler shape.
## Goal
After this plan lands, the later `memoized-pondering-knuth.md` work should be able to:
- build a projected `RunState` without inventing missing data
- replace direct read-side APIs with projection helpers
- remove duplicated write-side persistence for all event-backed fields
without first having to stop and redesign the event model again.
## Relationship To The Existing Plans
### What `plan-events-as-source-of-truth.md` already solved
- `run.created`
- `stage.completed.response`
- `sandbox.initialized` sandbox metadata
- `checkpoint.completed.diff`
- `command.started` / `command.completed`
- `retro.started.prompt/provider/model`
- `retro.completed.response/retro`
### What is still missing for `memoized-pondering-knuth.md`
- semantic run status events
- checkpoint events that can reconstruct full checkpoint snapshots and history
- full pull request record coverage
- final patch coverage
- provider-used coverage
- parallel-results coverage
- an explicit decision for fields that should remain non-event for now
## Decisions
### 1. Use semantic run lifecycle events, not a generic `run.status_changed`
Status should be reconstructed from explicit run lifecycle events rather than a generic “status changed” envelope.
Add event coverage for:
- `run.submitted`
- `run.starting`
- `run.running`
- `run.paused`
- `run.removing`
- `run.completed`
- `run.failed`
- `run.dead`
Projection rule:
- the latest status-bearing run event defines `RunStatusRecord.status`
- event-specific fields define `RunStatusRecord.reason`
- envelope `ts` defines `RunStatusRecord.updated_at`
`run.started` remains the start-record / execution-metadata event, not the canonical status event.
This separation is intentional:
- `run.started` answers "when and how did execution begin?"
- `run.running` answers "what is the run's status?"
Status mapping table:
| Event | Projected `RunStatus` | `StatusReason` rule |
|---|---|---|
| `run.submitted` | `Submitted` | `None` |
| `run.starting` | `Starting` | optional if the emitter has a concrete reason, otherwise `None` |
| `run.running` | `Running` | `None` |
| `run.paused` | `Paused` | preserve emitted reason if present |
| `run.removing` | `Removing` | `None` |
| `run.completed` | `Succeeded` | preserve emitted reason if present; default should remain `Completed` or `PartialSuccess` based on terminal outcome |
| `run.failed` | `Failed` | preserve emitted reason if present; expected common reasons include workflow/bootstrap/sandbox failures |
| `run.dead` | `Dead` | preserve emitted reason if present; otherwise `None` |
If any current status mutation cannot be represented cleanly by this table, fix the event model in this follow-up plan rather than pushing ambiguity into the later projector.
### 2. Keep the boundary tight: not every stored value must become an event in this pass
This follow-up plan should only eventize the fields that block the later projected-state cutover.
Retain as non-event concerns for now:
- binary assets
- artifact value blobs / offloaded context artifacts
This means `memoized-pondering-knuth.md` should be updated afterwards so `artifact_values` is no longer listed as a required event-backed row for the first cut.
### 3. Prefer complete event payloads over event joins that require hidden store lookups
If a projected record needs fields that do not already exist in another authoritative event, add them directly to the relevant event.
Do not rely on:
- sidecar JSON files
- legacy store records
- “the caller can join this with some other direct read”
Prefer one self-contained semantic event over reconstructing a record from several unrelated low-level events when that reconstruction adds complexity for little value.
## Follow-Up Coverage Matrix
This matrix is the contract for this plan. Each row must be green before `memoized-pondering-knuth.md` starts removing read/write APIs.
| Field / record needed later | Current direct source | Current event state | Follow-up required |
|---|---|---|---|
| `RunStatusRecord` | `put_status` in create/start/resume/finalize/disk paths | Incomplete; no semantic status event family | Add semantic run lifecycle events and a status mapping table |
| `StartRecord` | `put_start` | Mostly covered by `run.started` | Verify `run.started` fully covers `run_branch`, `base_sha`, `start_time`; no shape change if already true |
| latest `Checkpoint` | `put_checkpoint` | Incomplete; `checkpoint.completed` only carries `node_id`, `status`, `git_commit_sha`, `diff` | Enrich `checkpoint.completed` to carry a full checkpoint snapshot payload |
| checkpoint history | `append_checkpoint` / `list_checkpoints` | Incomplete; history cannot be rebuilt from current event payload | Use fully-populated `checkpoint.completed` as append-only checkpoint history |
| `Conclusion` | `put_conclusion` | Partially covered by terminal events plus stage aggregation | Verify the projector can derive full `Conclusion`, including retries/tokens/stage summaries, from existing events; if projection stays awkward, enrich terminal run events instead of adding new conclusion-only events |
| `PullRequestRecord` | `put_pull_request` | Incomplete; `pull_request.created` only carries URL/number/draft | Enrich `pull_request.created` to carry full `PullRequestRecord` fields |
| final patch | `put_final_patch` | Incomplete; terminal run event carries final SHA but not patch text | Enrich `run.completed` with `final_patch` |
| node provider metadata | `put_node_provider_used` | Incomplete; prompt/CLI events carry provider/model, but agent-mode still relies on sidecar sync | Project from existing handler-specific events and enrich forwarded agent session events if needed |
| node parallel results | `put_node_parallel_results` | Incomplete; `parallel.branch.completed.head_sha` is not enough | Enrich `parallel.completed` to carry the final results payload |
| node diff | `put_node_diff` | Partially covered by `checkpoint.completed.diff` | Decide and document whether node diff is sourced from the latest checkpoint event for that node or a dedicated node diff event; keep one canonical rule |
| retro prompt/response/retro payload | `put_retro_prompt`, `put_retro_response`, `put_retro` | Covered | No new event work; just parity-test it |
| sandbox record | `put_sandbox` | Covered | No new event work; just parity-test it |
## Required Event Changes
### 1. Add semantic run lifecycle events
Add new `WorkflowRunEvent` variants for:
- `RunSubmitted`
- `RunStarting`
- `RunRunning`
- `RunPaused`
- `RunRemoving`
- `RunDead`
Existing terminal events remain:
- `run.completed`
- `run.failed`
Required payload fields:
- `reason: Option<StatusReason>` where applicable
- any extra fields already emitted on terminal events should stay there
Emit from the same places that currently call `put_status`:
- `lib/crates/fabro-workflow/src/operations/create.rs`
- `lib/crates/fabro-workflow/src/operations/start.rs`
- `lib/crates/fabro-workflow/src/operations/resume.rs`
- `lib/crates/fabro-workflow/src/pipeline/finalize.rs`
- `lib/crates/fabro-workflow/src/lifecycle/disk.rs`
- CLI administrative flows that directly mutate status:
- `lib/crates/fabro-cli/src/commands/runs/rm.rs`
- `lib/crates/fabro-cli/src/commands/run/rewind.rs`
### 2. Enrich `checkpoint.completed` to carry a full checkpoint snapshot
Current `checkpoint.completed` is not enough to rebuild `Checkpoint`.
Add fields covering:
- `timestamp` is still the envelope `ts`
- `current_node`
- `completed_nodes`
- `node_retries`
- `context_values`
- `node_outcomes`
- `next_node_id`
- `git_commit_sha`
- `loop_failure_signatures`
- `restart_failure_signatures`
- `node_visits`
- `diff`
Emitter seam:
- `lib/crates/fabro-workflow/src/lifecycle/event.rs`
Producer seam for the source checkpoint object:
- `lib/crates/fabro-workflow/src/lifecycle/disk.rs`
Design rule:
- one `checkpoint.completed` event must be sufficient to reconstruct one historical checkpoint record without replaying prior stage events
That keeps checkpoint history export and `rebuild_meta` simple.
Do not split checkpoint reconstruction back across `stage.completed` and other incidental events unless there is a strong size or performance reason. A saved checkpoint is a first-class domain event and should be self-contained.
### 3. Enrich `pull_request.created` to carry the full record
Current event payload is too small for `PullRequestRecord`.
Add:
- `html_url`
- `number`
- `owner`
- `repo`
- `base_branch`
- `head_branch`
- `title`
- `draft`
Producer seam:
- `lib/crates/fabro-workflow/src/pipeline/pull_request.rs`
After this lands, `put_pull_request` should become removable during the later memoized-state cutover.
Do not add a second storage-oriented PR event. The semantic event is already "pull request created"; it just needs the full payload.
### 4. Enrich `run.completed` with the final patch
Current final patch only exists via direct store writes.
Do not add a separate storage-shaped event. The final patch is run-level terminal metadata, so it belongs on the terminal success event.
Enrich `run.completed`, using the patch already computed from:
- `lib/crates/fabro-workflow/src/lifecycle/git.rs`
Required payload:
- `final_patch: Option<String>`
Projection rule:
- `RunState.final_patch` projects from `run.completed.properties.final_patch`
This keeps final run summary data in one place alongside:
- `status`
- `duration_ms`
- `artifact_count`
- `final_git_commit_sha`
If failed runs later need final patch coverage too, extend the terminal failure event deliberately. Do not introduce a separate patch-persistence event unless terminal events prove insufficient.
### 5. Finish provider-used coverage using existing handler-specific events
The current system still reads `provider_used.json` from disk and syncs it into the store. That is not event-sourced.
Replace that with one explicit projection rule based on existing handler-specific events.
Use:
- `stage.prompt` for prompt-mode stages
- forwarded `agent.session.started` for agent-mode stages
- `agent.cli.started` for CLI-backed agent stages
If agent-mode forwarded session events still do not carry enough metadata, enrich `AgentEvent::SessionStarted` rather than adding a new stage-wide event.
Required projected output:
- `mode`
- `provider`
- `model`
- any existing raw provider-used JSON fields that are still needed by consumers
Likely seams:
- `lib/crates/fabro-workflow/src/handler/agent.rs`
- `lib/crates/fabro-workflow/src/handler/llm/api.rs`
- `lib/crates/fabro-workflow/src/pipeline/retro.rs` if retro uses the same forwarded agent session path
- any CLI-backed LLM path if it still produces `provider_used.json`
Do not keep the current “read JSON sidecar, then `put_node_provider_used`” pattern once this event exists.
Do not add a stage-generic provider-used event. Provider metadata is transport-specific and should stay attached to the prompt/agent/CLI events that actually know it.
### 6. Enrich `parallel.completed` with the final results payload
`parallel.branch.completed.head_sha` is useful but not enough to replace `put_node_parallel_results`.
Use the existing terminal parallel event rather than adding a storage-shaped event name.
Add to `parallel.completed`, emitted from:
- `lib/crates/fabro-workflow/src/handler/parallel.rs`
Required new payload:
- `results` as the same JSON array currently persisted to `parallel_results.json`
Projection rule:
- `NodeState.parallel_results` projects from `parallel.completed.properties.results`
Why this is the right event:
- it is emitted after all branch executions have joined
- the final result set has already been assembled
- it represents completion of the parallel node’s branch-collection phase
The workflow-level outcome of the node still comes from `stage.completed`; `parallel.completed` just becomes the canonical source for the branch result set.
Do not add `parallel.results_recorded` or similar. The semantic event already exists.
### 7. Lock down the node diff rule
We already added `checkpoint.completed.diff`, but the memoized plan should not proceed until there is one explicit derivation rule for node diff.
Decision required:
- either `NodeState.diff` is “latest checkpoint diff for that node visit”
- or add a dedicated `node.diff_generated` event
This follow-up plan should pick one and update docs/tests accordingly.
Given the current code, using `checkpoint.completed.diff` is the simpler option unless multiple diffs per node visit need to be preserved.
Prefer `checkpoint.completed.diff` unless a concrete consumer proves that diff generation and checkpoint persistence are semantically different moments.
## File Map
Likely files to touch:
- `docs-internal/events.md`
- `docs-internal/run-directory-keys.md`
- `docs-internal/events-strategy.md`
- `lib/crates/fabro-workflow/src/event.rs`
- `lib/crates/fabro-workflow/src/lifecycle/event.rs`
- `lib/crates/fabro-workflow/src/lifecycle/disk.rs`
- `lib/crates/fabro-workflow/src/lifecycle/git.rs`
- `lib/crates/fabro-workflow/src/operations/create.rs`
- `lib/crates/fabro-workflow/src/operations/start.rs`
- `lib/crates/fabro-workflow/src/operations/resume.rs`
- `lib/crates/fabro-workflow/src/pipeline/finalize.rs`
- `lib/crates/fabro-workflow/src/pipeline/pull_request.rs`
- `lib/crates/fabro-workflow/src/handler/agent.rs`
- `lib/crates/fabro-workflow/src/handler/parallel.rs`
- `lib/crates/fabro-cli/src/commands/runs/rm.rs`
- `lib/crates/fabro-cli/src/commands/run/rewind.rs`
- tests in `fabro-workflow`, `fabro-cli`, and `fabro-store`
## Phases
### Phase 1: Define the missing event contract
- add semantic run lifecycle events
- enrich `checkpoint.completed`
- enrich `pull_request.created`
- enrich `run.completed` with `final_patch`
- add provider-used event coverage
- add parallel-results event coverage
- document the node diff derivation rule
This phase is complete when every row in the follow-up coverage matrix is backed by an explicit event contract.
Priority during this phase:
- first enrich existing semantic events
- only add a truly new event when no existing semantic event owns the data
### Phase 2: Emit the new events everywhere status/data currently writes directly
Replace silent state mutation with canonical event emission first.
Important rule:
- do not remove direct store writes yet
- dual-write is acceptable in this phase
- the purpose is to prove event completeness before the memoized-state migration begins
### Phase 3: Add parity tests against current stored records
Add tests that build real event sequences and verify the future projector contract is now possible for:
- status reconstruction
- start record reconstruction
- checkpoint reconstruction
- checkpoint history reconstruction
- pull request reconstruction
- final patch reconstruction
- provider-used reconstruction
- parallel-results reconstruction
Where a direct legacy store record still exists, compare the event-derived value against the legacy persisted value.
Keep the tests shaped around semantic events, not internal store APIs. The point is to prove the event contract is sufficient.
Minimum parity scenarios:
- create-only run before execution starts
- normal started/running run
- resumed run
- rewound run
- successful git-backed run with final patch
- PR-producing run
- parallel run with branch results
- retro-enabled run
- failed run that exits through terminal/drop-guard paths
### Phase 4: Update the downstream migration plan
Once this follow-up plan lands:
- update `~/.claude/plans/memoized-pondering-knuth.md`
- remove any rows that were intentionally retained as non-event concerns
- mark the newly-completed event-backed rows as ready
- delete stale “likely needs to be added” wording that is no longer true
This keeps the later store-migration plan honest and implementation-ready.
## Verification
1. `cargo build --workspace`
2. `cargo clippy --workspace -- -D warnings`
3. `cargo nextest run -p fabro-workflow`
4. `cargo nextest run -p fabro-cli`
5. `cargo nextest run -p fabro-store`
6. `cargo nextest run --workspace`
7. Manual:
- create a run and inspect `progress.jsonl`
- verify semantic run lifecycle events appear in the expected order
- verify a run with git changes emits `run.completed.properties.final_patch`
- verify a PR-producing run emits full PR metadata in `pull_request.created`
- verify a parallel run emits canonical final results on `parallel.completed` without reading `parallel_results.json`
## Exit Criteria
This follow-up plan is complete when:
- every row in the follow-up coverage matrix is either event-backed or explicitly retained as non-event
- direct store writes are no longer the only source for status, checkpoint history, pull request record, final patch, provider-used, or parallel results
- projector parity tests prove the later `RunState` refactor has the event data it needs
- `memoized-pondering-knuth.md` can be updated to proceed without hidden event-contract gaps
At that point, the later migration work should mostly be mechanical projection and API cleanup, not more event-model design.

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

@ -4,9 +4,10 @@ use cli_table::format::{Border, Separator};
use cli_table::{Cell, CellStruct, Color, Style, Table};
use fabro_checkpoint::git::Store;
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};
@ -61,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();
@ -104,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)
@ -130,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
@ -158,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,
@ -165,6 +199,13 @@ async fn reset_rewound_run_state(
))
.await
.map_err(|err| anyhow::anyhow!("failed to restore run status after rewind: {err}"))?;
append_workflow_event(
run_store.as_ref(),
run_id,
&WorkflowRunEvent::RunSubmitted { reason: None },
)
.await
.map_err(|err| anyhow::anyhow!("failed to append restored run status event: {err}"))?;
run_store
.put_checkpoint(&checkpoint)
.await
@ -172,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(),

6
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,
@ -1079,6 +1080,11 @@ mod tests {
WorkflowRunEvent::PullRequestCreated {
pr_url: "https://github.com/fabro-sh/fabro/pull/42".into(),
pr_number: 42,
owner: "fabro-sh".into(),
repo: "fabro".into(),
base_branch: "main".into(),
head_branch: "fabro/run/42".into(),
title: "Ship the change".into(),
draft: true,
},
);

14
lib/crates/fabro-cli/src/commands/runs/rm.rs
View file

@ -5,6 +5,7 @@ use fabro_store::Store;
use tracing::warn;
use fabro_sandbox::reconnect::reconnect as reconnect_sandbox;
use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event};
use fabro_workflow::run_lookup::RunInfo;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use fabro_workflow::run_status::{RunStatus, RunStatusRecord};
@ -136,6 +137,19 @@ async fn remove_run_dir_with_cleanup(store: &dyn Store, run: &RunInfo) -> Result
"failed to save removing status to store"
);
}
if let Err(err) = append_workflow_event(
run_store.as_ref(),
&run.run_id,
&WorkflowRunEvent::RunRemoving { reason: None },
)
.await
{
warn!(
run_id = %run.run_id,
error = %err,
"failed to append removing status event"
);
}
}
if let Some(record) = load_sandbox_record(&run.path, run_store.as_deref()).await {

6
lib/crates/fabro-cli/src/main.rs
View file

@ -171,11 +171,11 @@ async fn main_inner() -> (String, Result<()>) {
None
};
let result = async move {
let result = Box::pin(async move {
match *command {
Commands::Llm(ns) => commands::llm::dispatch(ns, &globals).await?,
Commands::Exec(args) => commands::exec::execute(args, &globals).await?,
Commands::RunCmd(cmd) => commands::run::dispatch(cmd, &globals).await?,
Commands::RunCmd(cmd) => Box::pin(commands::run::dispatch(cmd, &globals)).await?,
Commands::Preflight(args) => commands::preflight::execute(args, &globals).await?,
Commands::Validate(args) => {
let styles = Styles::detect_stderr();
@ -270,7 +270,7 @@ async fn main_inner() -> (String, Result<()>) {
}
Ok(())
}
})
.await;
// Print upgrade notice after command completes (non-blocking during execution)

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

@ -423,6 +423,22 @@ fn attach_json_errors_without_prompting_for_human_input() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.submitted",
"id": "[EVENT_ID]",
"properties": {},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.starting",
"id": "[EVENT_ID]",
"properties": {
"reason": "sandbox_initializing"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.initializing",
"id": "[EVENT_ID]",
@ -466,6 +482,13 @@ fn attach_json_errors_without_prompting_for_human_input() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.running",
"id": "[EVENT_ID]",
"properties": {},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.started",
"id": "[EVENT_ID]",
@ -533,6 +556,33 @@ fn attach_json_errors_without_prompting_for_human_input() {
"node_id": "start",
"node_label": "start",
"properties": {
"completed_nodes": [
"start"
],
"context_values": {
"current.preamble": "Goal: Wait for approval/n",
"current_node": "start",
"failure_class": "",
"failure_signature": "",
"graph.goal": "Wait for approval",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.retry_count.start": 0,
"internal.run_id": "[ULID]",
"internal.thread_id": null,
"outcome": "success"
},
"current_node": "start",
"next_node_id": "approve",
"node_outcomes": {
"start": {
"status": "success",
"usage": null
}
},
"node_visits": {
"start": 1
},
"status": "success"
},
"run_id": "[ULID]",

22
lib/crates/fabro-cli/tests/it/cmd/logs.rs
View file

@ -63,25 +63,28 @@ fn logs_completed_run_outputs_raw_ndjson() {
exit_code: 0
----- stdout -----
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"start","node_label":"start","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"start","node_label":"start","properties":{"completed_nodes":["start"],"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":null,"outcome":"success"},"current_node":"start","next_node_id":"run_tests","node_outcomes":{"start":{"notes":"[Simulated] start","status":"success","usage":null}},"node_visits":{"start":1},"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"handler_type":"agent","index":1,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"run_tests","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"start","outcome":"success","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":1,"max_attempts":1,"node_visits":{"run_tests":1,"start":1},"notes":"[Simulated] run_tests","preferred_label":null,"response":"[Simulated] Response for stage: run_tests","status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"run_tests","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"report"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"run_tests","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"run_tests","properties":{"completed_nodes":["start","run_tests"],"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"run_tests","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"start","last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","outcome":"success","response.run_tests":"[Simulated] Response for stage: run_tests","thread.start.current_node":"run_tests"},"current_node":"run_tests","next_node_id":"report","node_outcomes":{"run_tests":{"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"notes":"[Simulated] run_tests","status":"success","usage":null},"start":{"notes":"[Simulated] start","status":"success","usage":null}},"node_visits":{"run_tests":1,"start":1},"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"handler_type":"agent","index":2,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: report","last_stage":"report","response.report":"[Simulated] Response for stage: report"},"context_values":{"current.preamble":"Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n","current_node":"report","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"run_tests","last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","outcome":"success","response.run_tests":"[Simulated] Response for stage: run_tests","thread.run_tests.current_node":"report","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":2,"max_attempts":1,"node_visits":{"report":1,"run_tests":1,"start":1},"notes":"[Simulated] report","preferred_label":null,"response":"[Simulated] Response for stage: report","status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"report","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"exit"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"report","node_label":"report","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"report","node_label":"report","properties":{"completed_nodes":["start","run_tests","report"],"context_values":{"current.preamble":"Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n","current_node":"report","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.report":0,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"run_tests","last_response":"[Simulated] Response for stage: report","last_stage":"report","outcome":"success","response.report":"[Simulated] Response for stage: report","response.run_tests":"[Simulated] Response for stage: run_tests","thread.run_tests.current_node":"report","thread.start.current_node":"run_tests"},"current_node":"report","next_node_id":"exit","node_outcomes":{"report":{"context_updates":{"last_response":"[Simulated] Response for stage: report","last_stage":"report","response.report":"[Simulated] Response for stage: report"},"notes":"[Simulated] report","status":"success","usage":null},"run_tests":{"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"notes":"[Simulated] run_tests","status":"success","usage":null},"start":{"notes":"[Simulated] start","status":"success","usage":null}},"node_visits":{"report":1,"run_tests":1,"start":1},"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"handler_type":"exit","index":3,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"duration_ms": [DURATION_MS],"files_touched":[],"index":3,"max_attempts":1,"notes":null,"preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.completed","id":"[EVENT_ID]","properties":{"artifact_count":0,"duration_ms": [DURATION_MS],"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.completed","id":"[EVENT_ID]","properties":{"artifact_count":0,"duration_ms": [DURATION_MS],"reason":"completed","status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
----- stderr -----
@ -225,25 +228,28 @@ fn logs_follow_detached_run_streams_until_completion() {
exit_code: 0
----- stdout -----
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"start","node_label":"start","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"start","node_label":"start","properties":{"completed_nodes":["start"],"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":null,"outcome":"success"},"current_node":"start","next_node_id":"run_tests","node_outcomes":{"start":{"notes":"[Simulated] start","status":"success","usage":null}},"node_visits":{"start":1},"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"handler_type":"agent","index":1,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"run_tests","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"start","outcome":"success","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":1,"max_attempts":1,"node_visits":{"run_tests":1,"start":1},"notes":"[Simulated] run_tests","preferred_label":null,"response":"[Simulated] Response for stage: run_tests","status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"run_tests","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"report"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"run_tests","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"run_tests","properties":{"completed_nodes":["start","run_tests"],"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"run_tests","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"start","last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","outcome":"success","response.run_tests":"[Simulated] Response for stage: run_tests","thread.start.current_node":"run_tests"},"current_node":"run_tests","next_node_id":"report","node_outcomes":{"run_tests":{"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"notes":"[Simulated] run_tests","status":"success","usage":null},"start":{"notes":"[Simulated] start","status":"success","usage":null}},"node_visits":{"run_tests":1,"start":1},"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"handler_type":"agent","index":2,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: report","last_stage":"report","response.report":"[Simulated] Response for stage: report"},"context_values":{"current.preamble":"Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n","current_node":"report","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"run_tests","last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","outcome":"success","response.run_tests":"[Simulated] Response for stage: run_tests","thread.run_tests.current_node":"report","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":2,"max_attempts":1,"node_visits":{"report":1,"run_tests":1,"start":1},"notes":"[Simulated] report","preferred_label":null,"response":"[Simulated] Response for stage: report","status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"report","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"exit"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"report","node_label":"report","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"report","node_label":"report","properties":{"completed_nodes":["start","run_tests","report"],"context_values":{"current.preamble":"Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n","current_node":"report","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.report":0,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"run_tests","last_response":"[Simulated] Response for stage: report","last_stage":"report","outcome":"success","response.report":"[Simulated] Response for stage: report","response.run_tests":"[Simulated] Response for stage: run_tests","thread.run_tests.current_node":"report","thread.start.current_node":"run_tests"},"current_node":"report","next_node_id":"exit","node_outcomes":{"report":{"context_updates":{"last_response":"[Simulated] Response for stage: report","last_stage":"report","response.report":"[Simulated] Response for stage: report"},"notes":"[Simulated] report","status":"success","usage":null},"run_tests":{"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"notes":"[Simulated] run_tests","status":"success","usage":null},"start":{"notes":"[Simulated] start","status":"success","usage":null}},"node_visits":{"report":1,"run_tests":1,"start":1},"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"handler_type":"exit","index":3,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"duration_ms": [DURATION_MS],"files_touched":[],"index":3,"max_attempts":1,"notes":null,"preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.completed","id":"[EVENT_ID]","properties":{"artifact_count":0,"duration_ms": [DURATION_MS],"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.completed","id":"[EVENT_ID]","properties":{"artifact_count":0,"duration_ms": [DURATION_MS],"reason":"completed","status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
----- stderr -----

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"
);
}

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

@ -370,6 +370,22 @@ fn json_run_implies_auto_approve_for_human_gates() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.submitted",
"id": "[EVENT_ID]",
"properties": {},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.starting",
"id": "[EVENT_ID]",
"properties": {
"reason": "sandbox_initializing"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.initializing",
"id": "[EVENT_ID]",
@ -413,6 +429,13 @@ fn json_run_implies_auto_approve_for_human_gates() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.running",
"id": "[EVENT_ID]",
"properties": {},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.started",
"id": "[EVENT_ID]",
@ -480,6 +503,33 @@ fn json_run_implies_auto_approve_for_human_gates() {
"node_id": "start",
"node_label": "start",
"properties": {
"completed_nodes": [
"start"
],
"context_values": {
"current.preamble": "Goal: Route through the default approval path/n",
"current_node": "start",
"failure_class": "",
"failure_signature": "",
"graph.goal": "Route through the default approval path",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.retry_count.start": 0,
"internal.run_id": "[ULID]",
"internal.thread_id": null,
"outcome": "success"
},
"current_node": "start",
"next_node_id": "approve",
"node_outcomes": {
"start": {
"status": "success",
"usage": null
}
},
"node_visits": {
"start": 1
},
"status": "success"
},
"run_id": "[ULID]",
@ -568,6 +618,52 @@ fn json_run_implies_auto_approve_for_human_gates() {
"node_id": "approve",
"node_label": "approve",
"properties": {
"completed_nodes": [
"start",
"approve"
],
"context_values": {
"current.preamble": "Goal: Route through the default approval path/n",
"current_node": "approve",
"failure_class": "",
"failure_signature": "",
"graph.goal": "Route through the default approval path",
"human.gate.label": "[A] Approve",
"human.gate.selected": "A",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.retry_count.approve": 0,
"internal.retry_count.start": 0,
"internal.run_id": "[ULID]",
"internal.thread_id": "start",
"outcome": "success",
"preferred_label": "[A] Approve",
"thread.start.current_node": "approve"
},
"current_node": "approve",
"next_node_id": "ship",
"node_outcomes": {
"approve": {
"context_updates": {
"human.gate.label": "[A] Approve",
"human.gate.selected": "A"
},
"preferred_label": "[A] Approve",
"status": "success",
"suggested_next_ids": [
"ship"
],
"usage": null
},
"start": {
"status": "success",
"usage": null
}
},
"node_visits": {
"approve": 1,
"start": 1
},
"status": "success"
},
"run_id": "[ULID]",
@ -683,6 +779,67 @@ fn json_run_implies_auto_approve_for_human_gates() {
"node_id": "ship",
"node_label": "ship",
"properties": {
"completed_nodes": [
"start",
"approve",
"ship"
],
"context_values": {
"command.output": "shipped/n",
"command.stderr": "",
"current.preamble": "Goal: Route through the default approval path/n/n## Completed stages/n- **approve**: success/n/n## Context/n- human.gate.label: [A] Approve/n- human.gate.selected: A/n",
"current_node": "ship",
"failure_class": "",
"failure_signature": "",
"graph.goal": "Route through the default approval path",
"human.gate.label": "[A] Approve",
"human.gate.selected": "A",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.retry_count.approve": 0,
"internal.retry_count.ship": 0,
"internal.retry_count.start": 0,
"internal.run_id": "[ULID]",
"internal.thread_id": "approve",
"outcome": "success",
"preferred_label": "[A] Approve",
"thread.approve.current_node": "ship",
"thread.start.current_node": "approve"
},
"current_node": "ship",
"next_node_id": "exit",
"node_outcomes": {
"approve": {
"context_updates": {
"human.gate.label": "[A] Approve",
"human.gate.selected": "A"
},
"preferred_label": "[A] Approve",
"status": "success",
"suggested_next_ids": [
"ship"
],
"usage": null
},
"ship": {
"context_updates": {
"command.output": "shipped/n",
"command.stderr": ""
},
"notes": "Script completed: echo shipped",
"status": "success",
"usage": null
},
"start": {
"status": "success",
"usage": null
}
},
"node_visits": {
"approve": 1,
"ship": 1,
"start": 1
},
"status": "success"
},
"run_id": "[ULID]",
@ -728,6 +885,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
"properties": {
"artifact_count": 0,
"duration_ms": "[DURATION_MS]",
"reason": "completed",
"status": "success"
},
"run_id": "[ULID]",

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

226
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};
@ -14,9 +14,10 @@ use tokio::sync::{mpsc, oneshot};
use uuid::Uuid;
use crate::error::FabroError;
use crate::outcome::{FailureDetail, StageUsage};
use crate::outcome::{FailureDetail, Outcome, StageUsage};
use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback};
use fabro_llm::types::Usage as LlmUsage;
use fabro_types::StatusReason;
use fabro_util::redact::redact_jsonl_line;
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
@ -82,22 +83,53 @@ pub enum WorkflowRunEvent {
#[serde(default, skip_serializing_if = "Option::is_none")]
goal: Option<String>,
},
RunSubmitted {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunStarting {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunRunning {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunRemoving {
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
},
RunRewound {
target_checkpoint_ordinal: usize,
target_node_id: String,
target_visit: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
previous_status: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
run_commit_sha: Option<String>,
},
WorkflowRunCompleted {
duration_ms: u64,
artifact_count: usize,
#[serde(default)]
status: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
#[serde(default, skip_serializing_if = "Option::is_none")]
total_cost: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
final_git_commit_sha: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
final_patch: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
usage: Option<LlmUsage>,
},
WorkflowRunFailed {
error: FabroError,
duration_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<StatusReason>,
#[serde(default, skip_serializing_if = "Option::is_none")]
git_commit_sha: Option<String>,
},
RunNotice {
@ -178,6 +210,8 @@ pub enum WorkflowRunEvent {
duration_ms: u64,
success_count: usize,
failure_count: usize,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
results: Vec<serde_json::Value>,
},
InterviewStarted {
question: String,
@ -197,8 +231,25 @@ pub enum WorkflowRunEvent {
CheckpointCompleted {
node_id: String,
status: String,
current_node: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
completed_nodes: Vec<String>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
node_retries: BTreeMap<String, u32>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
context_values: BTreeMap<String, serde_json::Value>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
node_outcomes: BTreeMap<String, Outcome>,
#[serde(default, skip_serializing_if = "Option::is_none")]
next_node_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
git_commit_sha: Option<String>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
loop_failure_signatures: BTreeMap<String, usize>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
restart_failure_signatures: BTreeMap<String, usize>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
node_visits: BTreeMap<String, usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
diff: Option<String>,
},
@ -257,6 +308,7 @@ pub enum WorkflowRunEvent {
},
Prompt {
stage: String,
visit: u32,
text: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
mode: Option<String>,
@ -276,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>,
@ -389,6 +442,7 @@ pub enum WorkflowRunEvent {
},
AgentCliStarted {
node_id: String,
visit: u32,
mode: String,
provider: String,
model: String,
@ -404,6 +458,11 @@ pub enum WorkflowRunEvent {
PullRequestCreated {
pr_url: String,
pr_number: u64,
owner: String,
repo: String,
base_branch: String,
head_branch: String,
title: String,
draft: bool,
},
PullRequestFailed {
@ -475,6 +534,34 @@ impl WorkflowRunEvent {
Self::WorkflowRunStarted { name, run_id, .. } => {
info!(workflow = name.as_str(), run_id = %run_id, "Workflow run started");
}
Self::RunSubmitted { reason } => {
info!(?reason, "Run submitted");
}
Self::RunStarting { reason } => {
info!(?reason, "Run starting");
}
Self::RunRunning { reason } => {
info!(?reason, "Run running");
}
Self::RunRemoving { reason } => {
info!(?reason, "Run removing");
}
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,
artifact_count,
@ -617,10 +704,14 @@ impl WorkflowRunEvent {
duration_ms,
success_count,
failure_count,
results,
} => {
debug!(
duration_ms,
success_count, failure_count, "Parallel execution completed"
success_count,
failure_count,
result_count = results.len(),
"Parallel execution completed"
);
}
Self::InterviewStarted {
@ -639,9 +730,17 @@ impl WorkflowRunEvent {
warn!(stage, duration_ms, "Interview timeout");
}
Self::CheckpointCompleted {
node_id, status, ..
node_id,
status,
completed_nodes,
..
} => {
debug!(node_id, status, "Checkpoint completed");
debug!(
node_id,
status,
completed_count = completed_nodes.len(),
"Checkpoint completed"
);
}
Self::CheckpointFailed { node_id, error } => {
error!(node_id, error, "Checkpoint failed");
@ -702,6 +801,7 @@ impl WorkflowRunEvent {
mode,
provider,
model,
..
} => {
debug!(
stage,
@ -882,9 +982,11 @@ impl WorkflowRunEvent {
pr_url,
pr_number,
draft,
owner,
repo,
..
} => {
info!(pr_url = %pr_url, pr_number, draft, "Pull request created");
info!(pr_url = %pr_url, pr_number, draft, owner, repo, "Pull request created");
}
Self::PullRequestFailed { error, .. } => {
error!(error = %error, "Pull request creation failed");
@ -975,6 +1077,11 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str {
match event {
WorkflowRunEvent::RunCreated { .. } => "run.created",
WorkflowRunEvent::WorkflowRunStarted { .. } => "run.started",
WorkflowRunEvent::RunSubmitted { .. } => "run.submitted",
WorkflowRunEvent::RunStarting { .. } => "run.starting",
WorkflowRunEvent::RunRunning { .. } => "run.running",
WorkflowRunEvent::RunRemoving { .. } => "run.removing",
WorkflowRunEvent::RunRewound { .. } => "run.rewound",
WorkflowRunEvent::WorkflowRunCompleted { .. } => "run.completed",
WorkflowRunEvent::WorkflowRunFailed { .. } => "run.failed",
WorkflowRunEvent::RunNotice { .. } => "run.notice",
@ -1221,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(),
@ -1403,6 +1514,20 @@ pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result<Ev
EventPayload::new(value, run_id).map_err(anyhow::Error::from)
}
pub async fn append_workflow_event(
run_store: &dyn RunStore,
run_id: &RunId,
event: &WorkflowRunEvent,
) -> Result<()> {
let envelope = canonicalize_event(run_id, event);
let payload = build_redacted_event_payload(&envelope, run_id)?;
run_store
.append_event(&payload)
.await
.map(|_| ())
.map_err(anyhow::Error::from)
}
pub struct ProgressLogger {
run_dir: PathBuf,
}
@ -1445,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(());
@ -1490,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()
@ -1749,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(),
@ -1766,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]
@ -1797,6 +2007,7 @@ mod tests {
&WorkflowRunEvent::WorkflowRunFailed {
error: FabroError::handler("boom"),
duration_ms: 900,
reason: Some(StatusReason::WorkflowError),
git_commit_sha: Some("abc123".to_string()),
},
);
@ -1867,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(),

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

@ -499,6 +499,7 @@ impl Handler for ParallelHandler {
duration_ms: millis_u64(parallel_start.elapsed()),
success_count,
failure_count: fail_count,
results: results_json.clone(),
});
{
let run_id = context

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 {

15
lib/crates/fabro-workflow/src/lifecycle/disk.rs
View file

@ -12,7 +12,7 @@ use fabro_core::outcome::NodeResult;
use fabro_core::state::RunState;
use super::circuit_breaker::CircuitBreakerLifecycle;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent, append_workflow_event};
use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode;
use crate::outcome::{OutcomeExt, StageUsage};
@ -64,6 +64,19 @@ impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
message: format!("failed to save running status to store: {err}"),
});
}
if let Err(err) = append_workflow_event(
self.run_store.as_ref(),
&self.run_id,
&WorkflowRunEvent::RunRunning { reason: None },
)
.await
{
self.emitter.emit(&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Warn,
code: "status_event_append_failed".to_string(),
message: format!("failed to append running status event: {err}"),
});
}
Ok(())
}

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

@ -23,7 +23,7 @@ use crate::graph::WorkflowNode;
use crate::outcome::{
FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage, stage_usage_to_llm,
};
use fabro_types::RunId;
use fabro_types::{RunId, StatusReason};
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
@ -51,6 +51,7 @@ pub(crate) struct EventLifecycle {
// Cross-lifecycle data
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
pub last_git_sha: Arc<Mutex<Option<String>>>,
pub final_patch: Arc<Mutex<Option<String>>>,
pub circuit_breaker: Arc<CircuitBreakerLifecycle>,
}
@ -309,8 +310,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
&self,
node: &WorkflowNode,
result: &WfNodeResult,
_next_node_id: Option<&str>,
_state: &WfRunState,
next_node_id: Option<&str>,
state: &WfRunState,
) -> CoreResult<()> {
let status = result.outcome.status.to_string();
@ -319,11 +320,42 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
let git_sha = git_result.as_ref().and_then(|r| r.commit_sha.clone());
let diff = git_result.as_ref().and_then(|r| r.diff.clone());
let (loop_failure_signatures, restart_failure_signatures) =
snapshot_failure_signatures(&self.circuit_breaker);
self.emitter.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: node.id().to_string(),
status,
current_node: node.id().to_string(),
completed_nodes: state.completed_nodes.clone(),
node_retries: state
.node_retries
.clone()
.into_iter()
.collect::<BTreeMap<_, _>>(),
context_values: state
.context
.snapshot()
.into_iter()
.collect::<BTreeMap<_, _>>(),
node_outcomes: state
.node_outcomes
.clone()
.into_iter()
.chain(std::iter::once((
node.id().to_string(),
result.outcome.clone(),
)))
.collect::<BTreeMap<_, _>>(),
next_node_id: next_node_id.map(ToOwned::to_owned),
git_commit_sha: git_sha.clone(),
loop_failure_signatures: loop_failure_signatures.unwrap_or_default(),
restart_failure_signatures: restart_failure_signatures.unwrap_or_default(),
node_visits: state
.node_visits
.clone()
.into_iter()
.collect::<BTreeMap<_, _>>(),
diff,
});
@ -354,6 +386,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
u64::try_from(self.run_start.lock().unwrap().elapsed().as_millis()).unwrap();
let artifact_count = self.artifact_store.lock().unwrap().list().len();
let last_sha = self.last_git_sha.lock().unwrap().clone();
let final_patch = self.final_patch.lock().unwrap().clone();
let total_cost = {
let sum: f64 = state
.node_outcomes
@ -373,8 +406,13 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
duration_ms,
artifact_count,
status: outcome.status.to_string(),
reason: Some(match outcome.status {
StageStatus::PartialSuccess => StatusReason::PartialSuccess,
_ => StatusReason::Completed,
}),
total_cost,
final_git_commit_sha: last_sha,
final_patch,
usage: run_usage,
});
} else {
@ -385,6 +423,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
self.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed {
error: FabroError::engine(error_msg),
duration_ms,
reason: Some(StatusReason::WorkflowError),
git_commit_sha: last_sha,
});
}

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

@ -45,6 +45,7 @@ pub(crate) struct GitLifecycle {
// Cross-lifecycle data (shared with EventLifecycle)
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
pub last_git_sha: Arc<Mutex<Option<String>>>,
pub final_patch: Arc<Mutex<Option<String>>>,
}
#[async_trait]
@ -53,6 +54,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
// Reset last_git_sha (diff base parity)
*self.last_git_sha.lock().unwrap() = None;
*self.checkpoint_git_result.lock().unwrap() = None;
*self.final_patch.lock().unwrap() = None;
// Init metadata branch (best-effort)
if let (Some(_), Some(repo_path)) = (
@ -350,6 +352,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
{
match git_diff(&*self.sandbox, &base_sha).await {
Ok(patch) if !patch.is_empty() => {
*self.final_patch.lock().unwrap() = Some(patch.clone());
if let Err(err) = self.run_store.put_final_patch(&patch).await {
self.emitter.emit(&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Warn,
@ -358,11 +361,13 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
"failed to persist final diff in run store: {err}"
),
});
return;
}
}
Ok(_) => {}
Ok(_) => {
*self.final_patch.lock().unwrap() = None;
}
Err(err) => {
*self.final_patch.lock().unwrap() = None;
self.emitter.emit(&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Warn,
code: "git_diff_failed".to_string(),

5
lib/crates/fabro-workflow/src/lifecycle/mod.rs
View file

@ -96,6 +96,7 @@ impl WorkflowLifecycle {
let checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>> =
Arc::new(Mutex::new(None));
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
let final_patch: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
let artifact_store = Arc::new(Mutex::new(ArtifactStore::new(Some(
runtime_state.artifact_values_dir(),
))));
@ -127,6 +128,7 @@ impl WorkflowLifecycle {
goal: (!graph.goal().is_empty()).then(|| graph.goal().to_string()),
artifact_store: Arc::clone(&artifact_store),
last_git_sha: Arc::clone(&last_git_sha),
final_patch: Arc::clone(&final_patch),
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
circuit_breaker: Arc::clone(&circuit_breaker),
};
@ -165,6 +167,7 @@ impl WorkflowLifecycle {
start_node_id,
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
last_git_sha: Arc::clone(&last_git_sha),
final_patch,
};
let artifact = ArtifactLifecycle::new(
@ -425,8 +428,8 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
if state.cancelled {
return;
}
self.git.on_run_end(outcome, state).await;
self.event.on_run_end(outcome, state).await;
self.hook.on_run_end(outcome, state).await;
self.git.on_run_end(outcome, state).await;
}
}

13
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -18,7 +18,9 @@ use crate::transforms::{Transform, expand_vars};
use fabro_sandbox::daytona::detect_repo_info;
use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow};
use crate::event::{WorkflowRunEvent, canonicalize_event_at, normalize_json_value};
use crate::event::{
WorkflowRunEvent, append_workflow_event, canonicalize_event_at, normalize_json_value,
};
const RUN_CONFIG_FILE: &str = "workflow.toml";
@ -199,7 +201,14 @@ async fn persist_created_run(
.append_event(&payload)
.await
.map(|_| ())
.map_err(store_error)
.map_err(store_error)?;
append_workflow_event(
run_store.as_ref(),
&record.run_id,
&WorkflowRunEvent::RunSubmitted { reason: None },
)
.await
.map_err(store_error)
}
fn store_error(err: impl std::fmt::Display) -> FabroError {

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

@ -3,6 +3,7 @@ use std::path::Path;
use fabro_store::RuntimeState;
use crate::error::FabroError;
use crate::event::{WorkflowRunEvent, append_workflow_event};
use crate::outcome::StageStatus;
use crate::run_status::{self, RunStatus};
@ -54,6 +55,13 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
))
.await
.map_err(|err| FabroError::engine(err.to_string()))?;
append_workflow_event(
services.run_store.as_ref(),
&services.run_id,
&WorkflowRunEvent::RunSubmitted { reason: None },
)
.await
.map_err(|err| FabroError::engine(err.to_string()))?;
Box::pin(execute_persisted_run(run_dir, Some(checkpoint), services)).await
}

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

@ -17,8 +17,8 @@ use serde::Serialize;
use crate::context::Context;
use crate::error::FabroError;
use crate::event::{
EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent, canonicalize_event,
event_payload_from_redacted_json, redacted_event_json,
EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent, append_workflow_event,
canonicalize_event, event_payload_from_redacted_json, redacted_event_json,
};
use crate::git::MetadataStore;
use crate::handler::HandlerRegistry;
@ -138,9 +138,22 @@ pub(super) async fn execute_persisted_run(
.await;
return Err(error);
}
append_workflow_event(
run_store.as_ref(),
&run_id,
&WorkflowRunEvent::RunStarting {
reason: Some(StatusReason::SandboxInitializing),
},
)
.await
.map_err(|err| FabroError::engine(err.to_string()))?;
let mut bootstrap_guard =
DetachedRunBootstrapGuard::arm(run_dir, Arc::clone(&run_store), cancel_token.clone());
let mut bootstrap_guard = DetachedRunBootstrapGuard::arm(
run_id,
run_dir,
Arc::clone(&run_store),
cancel_token.clone(),
);
let persisted = match Persisted::load_from_store(services.run_store.as_ref(), run_dir).await {
Ok(persisted) => persisted,
@ -188,8 +201,14 @@ pub(super) async fn execute_persisted_run(
Ok(started)
}
Err(err) => {
persist_terminal_engine_failure(run_store.as_ref(), run_dir, &err, run_start.elapsed())
.await;
persist_terminal_engine_failure(
run_id,
run_store.as_ref(),
run_dir,
&err,
run_start.elapsed(),
)
.await;
completion_guard.defuse();
Err(err)
}
@ -197,6 +216,7 @@ pub(super) async fn execute_persisted_run(
}
async fn persist_terminal_engine_failure(
run_id: RunId,
run_store: &dyn RunStore,
_run_dir: &Path,
error: &FabroError,
@ -222,6 +242,20 @@ async fn persist_terminal_engine_failure(
{
tracing::warn!(error = %err, "Failed to save terminal engine failure status to store");
}
if let Err(err) = append_workflow_event(
run_store,
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
error: error.clone(),
duration_ms: u64::try_from(duration.as_millis()).unwrap(),
reason: status_reason,
git_commit_sha: None,
},
)
.await
{
tracing::warn!(error = %err, "Failed to append terminal engine failure event");
}
}
impl RunSession {
@ -570,6 +604,7 @@ impl RunSession {
}
struct DetachedRunBootstrapGuard {
run_id: RunId,
run_store: Arc<dyn RunStore>,
cancel_token: Option<Arc<AtomicBool>>,
active: bool,
@ -577,11 +612,13 @@ struct DetachedRunBootstrapGuard {
impl DetachedRunBootstrapGuard {
fn arm(
run_id: RunId,
_run_dir: &Path,
run_store: Arc<dyn RunStore>,
cancel_token: Option<Arc<AtomicBool>>,
) -> Self {
Self {
run_id,
run_store,
cancel_token,
active: true,
@ -605,6 +642,7 @@ impl Drop for DetachedRunBootstrapGuard {
} else {
StatusReason::SandboxInitFailed
};
let run_id = self.run_id;
let run_store = Arc::clone(&self.run_store);
if let Ok(handle) = Handle::try_current() {
handle.spawn(async move {
@ -614,6 +652,17 @@ impl Drop for DetachedRunBootstrapGuard {
Some(reason),
))
.await;
let _ = append_workflow_event(
run_store.as_ref(),
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
error: FabroError::engine(format!("{reason:?}")),
duration_ms: 0,
reason: Some(reason),
git_commit_sha: None,
},
)
.await;
});
}
}
@ -707,6 +756,17 @@ impl Drop for DetachedRunCompletionGuard {
Some(reason),
))
.await;
let _ = append_workflow_event(
run_store.as_ref(),
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
error: FabroError::engine(message.to_string()),
duration_ms: 0,
reason: Some(reason),
git_commit_sha: None,
},
)
.await;
if let Err(err) = run_store
.put_conclusion(&build_failure_conclusion(message))
.await
@ -789,6 +849,20 @@ async fn persist_detached_failure(
{
tracing::warn!(error = %err, "Failed to save detached failure status to store");
}
if let Err(err) = append_workflow_event(
run_store,
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
error: error.clone(),
duration_ms: 0,
reason: Some(reason),
git_commit_sha: None,
},
)
.await
{
tracing::warn!(error = %err, "Failed to append detached failure event");
}
let event = WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Error,
@ -929,7 +1003,16 @@ mod tests {
emitter_for_injection.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: "start".to_string(),
status: "success".to_string(),
current_node: "start".to_string(),
completed_nodes: Vec::new(),
node_retries: HashMap::new().into_iter().collect(),
context_values: HashMap::new().into_iter().collect(),
node_outcomes: HashMap::new().into_iter().collect(),
next_node_id: None,
git_commit_sha: Some("sha-test".to_string()),
loop_failure_signatures: HashMap::new().into_iter().collect(),
restart_failure_signatures: HashMap::new().into_iter().collect(),
node_visits: HashMap::new().into_iter().collect(),
diff: None,
});
}

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

@ -590,6 +590,11 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) ->
emitter.emit(&WorkflowRunEvent::PullRequestCreated {
pr_url: record.html_url.clone(),
pr_number: record.number,
owner: record.owner.clone(),
repo: record.repo.clone(),
base_branch: record.base_branch.clone(),
head_branch: record.head_branch.clone(),
title: record.title.clone(),
draft: pr_cfg.draft,
});
pr_url = Some(record.html_url.clone());

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,