From 631d9a96e102238366fc183ce4470e43d5c79d68 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 2 Apr 2026 10:02:31 -0700 Subject: [PATCH 1/2] Finish events-as-source-of-truth follow-ups --- ...an-events-as-source-of-truth-follow-ups.md | 457 ++++++++++++++++++ .../fabro-cli/src/commands/run/rewind.rs | 8 + .../src/commands/run/run_progress/mod.rs | 5 + lib/crates/fabro-cli/src/commands/runs/rm.rs | 14 + lib/crates/fabro-cli/src/main.rs | 6 +- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 50 ++ lib/crates/fabro-cli/tests/it/cmd/logs.rs | 22 +- lib/crates/fabro-cli/tests/it/cmd/run.rs | 158 ++++++ lib/crates/fabro-workflow/src/event.rs | 118 ++++- .../fabro-workflow/src/handler/parallel.rs | 1 + .../fabro-workflow/src/lifecycle/disk.rs | 15 +- .../fabro-workflow/src/lifecycle/event.rs | 45 +- .../fabro-workflow/src/lifecycle/git.rs | 9 +- .../fabro-workflow/src/lifecycle/mod.rs | 5 +- .../fabro-workflow/src/operations/create.rs | 13 +- .../fabro-workflow/src/operations/resume.rs | 8 + .../fabro-workflow/src/operations/start.rs | 95 +++- .../src/pipeline/pull_request.rs | 5 + 18 files changed, 1003 insertions(+), 31 deletions(-) create mode 100644 docs-internal/plan-events-as-source-of-truth-follow-ups.md diff --git a/docs-internal/plan-events-as-source-of-truth-follow-ups.md b/docs-internal/plan-events-as-source-of-truth-follow-ups.md new file mode 100644 index 000000000..59ccde68f --- /dev/null +++ b/docs-internal/plan-events-as-source-of-truth-follow-ups.md @@ -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` 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` + +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. diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index 37228c757..928d6e36e 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -4,6 +4,7 @@ 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, @@ -165,6 +166,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 diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index 27af82bba..a38b2da72 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -1079,6 +1079,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, }, ); diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index 82bf91064..c39afe8aa 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -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 { diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index f283e9990..bfd664422 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -165,11 +165,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(); @@ -267,7 +267,7 @@ async fn main_inner() -> (String, Result<()>) { } Ok(()) - } + }) .await; // Print upgrade notice after command completes (non-blocking during execution) diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 9684e979c..5eda6ac9e 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -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]", diff --git a/lib/crates/fabro-cli/tests/it/cmd/logs.rs b/lib/crates/fabro-cli/tests/it/cmd/logs.rs index 115a03aa7..ef9fe6977 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/logs.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/logs.rs @@ -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 ----- diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 9c4885318..c7066fbae 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -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]", diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 447bbf25c..f2ffd2880 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -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,52 @@ pub enum WorkflowRunEvent { #[serde(default, skip_serializing_if = "Option::is_none")] goal: Option, }, + RunSubmitted { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + RunStarting { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + RunRunning { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + RunPaused { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + RunRemoving { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, + RunDead { + #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + }, WorkflowRunCompleted { duration_ms: u64, artifact_count: usize, #[serde(default)] status: String, #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] total_cost: Option, #[serde(default, skip_serializing_if = "Option::is_none")] final_git_commit_sha: Option, #[serde(default, skip_serializing_if = "Option::is_none")] + final_patch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] usage: Option, }, WorkflowRunFailed { error: FabroError, duration_ms: u64, #[serde(default, skip_serializing_if = "Option::is_none")] + reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] git_commit_sha: Option, }, RunNotice { @@ -178,6 +209,8 @@ pub enum WorkflowRunEvent { duration_ms: u64, success_count: usize, failure_count: usize, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + results: Vec, }, InterviewStarted { question: String, @@ -197,8 +230,25 @@ pub enum WorkflowRunEvent { CheckpointCompleted { node_id: String, status: String, + current_node: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + completed_nodes: Vec, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + node_retries: BTreeMap, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + context_values: BTreeMap, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + node_outcomes: BTreeMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + next_node_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] git_commit_sha: Option, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + loop_failure_signatures: BTreeMap, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + restart_failure_signatures: BTreeMap, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + node_visits: BTreeMap, #[serde(default, skip_serializing_if = "Option::is_none")] diff: Option, }, @@ -404,6 +454,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 +530,24 @@ 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::RunPaused { reason } => { + info!(?reason, "Run paused"); + } + Self::RunRemoving { reason } => { + info!(?reason, "Run removing"); + } + Self::RunDead { reason } => { + warn!(?reason, "Run dead"); + } Self::WorkflowRunCompleted { duration_ms, artifact_count, @@ -617,10 +690,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 +716,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"); @@ -882,9 +967,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 +1062,12 @@ 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::RunPaused { .. } => "run.paused", + WorkflowRunEvent::RunRemoving { .. } => "run.removing", + WorkflowRunEvent::RunDead { .. } => "run.dead", WorkflowRunEvent::WorkflowRunCompleted { .. } => "run.completed", WorkflowRunEvent::WorkflowRunFailed { .. } => "run.failed", WorkflowRunEvent::RunNotice { .. } => "run.notice", @@ -1403,6 +1496,20 @@ pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result 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, } @@ -1797,6 +1904,7 @@ mod tests { &WorkflowRunEvent::WorkflowRunFailed { error: FabroError::handler("boom"), duration_ms: 900, + reason: Some(StatusReason::WorkflowError), git_commit_sha: Some("abc123".to_string()), }, ); diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 8ddc1b9e6..0e305df36 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -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 diff --git a/lib/crates/fabro-workflow/src/lifecycle/disk.rs b/lib/crates/fabro-workflow/src/lifecycle/disk.rs index 6b6f7a947..97b98236f 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/disk.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/disk.rs @@ -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 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(()) } diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 5f559c23f..e9aad4a64 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -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>; type WfNodeResult = NodeResult>; @@ -51,6 +51,7 @@ pub(crate) struct EventLifecycle { // Cross-lifecycle data pub checkpoint_git_result: Arc>>, pub last_git_sha: Arc>>, + pub final_patch: Arc>>, pub circuit_breaker: Arc, } @@ -309,8 +310,8 @@ impl RunLifecycle 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 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::>(), + context_values: state + .context + .snapshot() + .into_iter() + .collect::>(), + node_outcomes: state + .node_outcomes + .clone() + .into_iter() + .chain(std::iter::once(( + node.id().to_string(), + result.outcome.clone(), + ))) + .collect::>(), + 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::>(), diff, }); @@ -354,6 +386,7 @@ impl RunLifecycle 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 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 for EventLifecycle { self.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed { error: FabroError::engine(error_msg), duration_ms, + reason: Some(StatusReason::WorkflowError), git_commit_sha: last_sha, }); } diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index eef58a9dd..19f7afe26 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -45,6 +45,7 @@ pub(crate) struct GitLifecycle { // Cross-lifecycle data (shared with EventLifecycle) pub checkpoint_git_result: Arc>>, pub last_git_sha: Arc>>, + pub final_patch: Arc>>, } #[async_trait] @@ -53,6 +54,7 @@ impl RunLifecycle 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 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 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(), diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index ddb1969fc..699c33530 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -96,6 +96,7 @@ impl WorkflowLifecycle { let checkpoint_git_result: Arc>> = Arc::new(Mutex::new(None)); let last_git_sha: Arc>> = Arc::new(Mutex::new(None)); + let final_patch: Arc>> = 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 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; } } diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 66afb487e..d3703c563 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -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 { diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs index 250ddb5fc..e03ca7fb7 100644 --- a/lib/crates/fabro-workflow/src/operations/resume.rs +++ b/lib/crates/fabro-workflow/src/operations/resume.rs @@ -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 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, cancel_token: Option>, active: bool, @@ -577,11 +612,13 @@ struct DetachedRunBootstrapGuard { impl DetachedRunBootstrapGuard { fn arm( + run_id: RunId, _run_dir: &Path, run_store: Arc, cancel_token: Option>, ) -> 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, }); } diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 935744652..f9c658036 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -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()); From c245c4194df58e112dd61b902acab5408677ee73 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 2 Apr 2026 10:26:12 -0700 Subject: [PATCH 2/2] Make provider metadata event-derived and rewind append-only --- .../fabro-cli/src/commands/run/rewind.rs | 83 +++++++++-- .../src/commands/run/run_progress/event.rs | 1 + .../src/commands/run/run_progress/mod.rs | 1 + lib/crates/fabro-cli/tests/it/cmd/rewind.rs | 67 ++++++++- lib/crates/fabro-cli/tests/it/cmd/support.rs | 8 +- lib/crates/fabro-store/src/lib.rs | 2 + lib/crates/fabro-store/src/memory.rs | 12 ++ lib/crates/fabro-store/src/slate/run_store.rs | 38 +++++ lib/crates/fabro-workflow/src/event.rs | 134 ++++++++++++++++-- .../fabro-workflow/src/handler/agent.rs | 78 ++++------ .../fabro-workflow/src/handler/llm/api.rs | 21 +-- .../fabro-workflow/src/handler/llm/cli.rs | 6 + .../fabro-workflow/src/handler/prompt.rs | 45 +++--- .../fabro-workflow/src/pipeline/retro.rs | 1 + .../fabro-workflow/tests/it/integration.rs | 1 + 15 files changed, 395 insertions(+), 103 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index 928d6e36e..b116369b2 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -7,7 +7,7 @@ use fabro_util::terminal::Styles; use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event}; use fabro_workflow::git::MetadataStore; use fabro_workflow::operations::{ - RewindInput, RewindTarget, RunTimeline, build_timeline_or_rebuild, + RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, find_run_id_by_prefix_or_store, rewind, }; use fabro_workflow::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt}; @@ -62,12 +62,14 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs &store, &RewindInput { run_id, - target, + target: target.clone(), push: !args.no_push, }, )?; if let Some(run_info) = run_info.as_ref() { - reset_rewound_run_state(&store, durable_store.as_ref(), &run_id, &run_info.path).await?; + let entry = timeline.resolve(&target)?; + reset_rewound_run_state(&store, durable_store.as_ref(), &run_id, &run_info.path, entry) + .await?; } let run_id_string = run_id.to_string(); @@ -105,6 +107,7 @@ async fn reset_rewound_run_state( durable_store: &dyn fabro_store::Store, run_id: &fabro_types::RunId, run_dir: &std::path::Path, + entry: &TimelineEntry, ) -> Result<()> { let existing_run_store = durable_store .open_run_reader(run_id) @@ -131,18 +134,28 @@ async fn reset_rewound_run_state( .context("failed to restore run record after rewind: missing run metadata")?; let checkpoint = MetadataStore::read_checkpoint(git_store.repo_dir(), &run_id.to_string())? .context("rewound metadata branch is missing checkpoint.json")?; + let previous_status = if let Some(run_store) = existing_run_store.as_ref() { + run_store + .get_status() + .await + .ok() + .flatten() + .map(|status| status.status.to_string()) + } else { + None + }; let _ = std::fs::remove_file(run_dir.join("detached_failure.json")); - durable_store - .delete_run(run_id) - .await - .map_err(|err| anyhow::anyhow!("failed to reset durable store run: {err}"))?; - let run_dir_string = run_dir.to_string_lossy().to_string(); let run_store = durable_store - .create_run(run_id, run_record.created_at, Some(&run_dir_string)) + .open_run(run_id) .await - .map_err(|err| anyhow::anyhow!("failed to recreate durable store run: {err}"))?; + .map_err(|err| anyhow::anyhow!("failed to open durable store run for rewind reset: {err}"))? + .context("failed to reset durable store run after rewind: missing run store")?; + run_store + .reset_for_rewind() + .await + .map_err(|err| anyhow::anyhow!("failed to clear rewound run state: {err}"))?; run_store .put_run(&run_record) .await @@ -159,6 +172,26 @@ async fn reset_rewound_run_state( .await .map_err(|err| anyhow::anyhow!("failed to restore graph after rewind: {err}"))?; } + append_workflow_event( + run_store.as_ref(), + run_id, + &WorkflowRunEvent::RunRewound { + target_checkpoint_ordinal: entry.ordinal, + target_node_id: entry.node_name.clone(), + target_visit: entry.visit, + previous_status, + run_commit_sha: entry.run_commit_sha.clone(), + }, + ) + .await + .map_err(|err| anyhow::anyhow!("failed to append run rewound event: {err}"))?; + append_workflow_event( + run_store.as_ref(), + run_id, + &restored_checkpoint_event(&checkpoint), + ) + .await + .map_err(|err| anyhow::anyhow!("failed to append restored checkpoint event: {err}"))?; run_store .put_status(&fabro_types::RunStatusRecord::new( RunStatus::Submitted, @@ -180,6 +213,36 @@ async fn reset_rewound_run_state( Ok(()) } +fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> WorkflowRunEvent { + let current_status = checkpoint + .node_outcomes + .get(&checkpoint.current_node) + .map_or_else(|| "success".to_string(), |outcome| outcome.status.to_string()); + WorkflowRunEvent::CheckpointCompleted { + node_id: checkpoint.current_node.clone(), + status: current_status, + current_node: checkpoint.current_node.clone(), + completed_nodes: checkpoint.completed_nodes.clone(), + node_retries: checkpoint.node_retries.clone().into_iter().collect(), + context_values: checkpoint.context_values.clone().into_iter().collect(), + node_outcomes: checkpoint.node_outcomes.clone().into_iter().collect(), + next_node_id: checkpoint.next_node_id.clone(), + git_commit_sha: checkpoint.git_commit_sha.clone(), + loop_failure_signatures: checkpoint + .loop_failure_signatures + .iter() + .map(|(sig, count)| (sig.to_string(), *count)) + .collect(), + restart_failure_signatures: checkpoint + .restart_failure_signatures + .iter() + .map(|(sig, count)| (sig.to_string(), *count)) + .collect(), + node_visits: checkpoint.node_visits.clone().into_iter().collect(), + diff: None, + } +} + pub(crate) fn print_timeline(timeline: &RunTimeline, styles: &Styles) { if timeline.entries.is_empty() { eprintln!("No checkpoints found."); diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index 32dc11844..245bf72a3 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -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(), diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index a38b2da72..a65801357 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -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, diff --git a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs index 0e79c3a3f..3543d0e0e 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs @@ -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::>(), + before_events + .iter() + .map(|event| event.payload.as_value()["event"].as_str().unwrap()) + .collect::>(), + "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" + ); +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index e40fb330b..e0f6c2761 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -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 { + 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)) } diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 8d48a6a39..07c3c3fc5 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -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>; + async fn reset_for_rewind(&self) -> Result<()>; + async fn append_event(&self, payload: &EventPayload) -> Result; async fn list_events(&self) -> Result>; async fn list_events_from(&self, seq: u32) -> Result>; diff --git a/lib/crates/fabro-store/src/memory.rs b/lib/crates/fabro-store/src/memory.rs index be1065b55..febc85ffe 100644 --- a/lib/crates/fabro-store/src/memory.rs +++ b/lib/crates/fabro-store/src/memory.rs @@ -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 { payload.validate(&self.run_id)?; diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 1b75cad95..5d704da64 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -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 { 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> { Ok(db.get(key).await?) } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index f2ffd2880..f36fa973a 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicI64, Ordering}; use anyhow::{Context, Result}; use chrono::{SecondsFormat, Utc}; -use fabro_store::{EventPayload, RunStore}; +use fabro_store::{EventPayload, NodeVisitRef, RunStore}; use fabro_types::RunId; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; @@ -95,17 +95,18 @@ pub enum WorkflowRunEvent { #[serde(default, skip_serializing_if = "Option::is_none")] reason: Option, }, - RunPaused { - #[serde(default, skip_serializing_if = "Option::is_none")] - reason: Option, - }, RunRemoving { #[serde(default, skip_serializing_if = "Option::is_none")] reason: Option, }, - RunDead { + RunRewound { + target_checkpoint_ordinal: usize, + target_node_id: String, + target_visit: usize, #[serde(default, skip_serializing_if = "Option::is_none")] - reason: Option, + previous_status: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + run_commit_sha: Option, }, WorkflowRunCompleted { duration_ms: u64, @@ -307,6 +308,7 @@ pub enum WorkflowRunEvent { }, Prompt { stage: String, + visit: u32, text: String, #[serde(default, skip_serializing_if = "Option::is_none")] mode: Option, @@ -326,6 +328,7 @@ pub enum WorkflowRunEvent { /// Forwarded from an agent session, tagged with the workflow stage. Agent { stage: String, + visit: u32, event: AgentEvent, #[serde(default, skip_serializing_if = "Option::is_none")] session_id: Option, @@ -439,6 +442,7 @@ pub enum WorkflowRunEvent { }, AgentCliStarted { node_id: String, + visit: u32, mode: String, provider: String, model: String, @@ -539,14 +543,24 @@ impl WorkflowRunEvent { Self::RunRunning { reason } => { info!(?reason, "Run running"); } - Self::RunPaused { reason } => { - info!(?reason, "Run paused"); - } Self::RunRemoving { reason } => { info!(?reason, "Run removing"); } - Self::RunDead { reason } => { - warn!(?reason, "Run dead"); + Self::RunRewound { + target_checkpoint_ordinal, + target_node_id, + target_visit, + previous_status, + run_commit_sha, + } => { + info!( + target_checkpoint_ordinal, + target_node_id, + target_visit, + previous_status = previous_status.as_deref().unwrap_or(""), + run_commit_sha = run_commit_sha.as_deref().unwrap_or(""), + "Run rewound" + ); } Self::WorkflowRunCompleted { duration_ms, @@ -787,6 +801,7 @@ impl WorkflowRunEvent { mode, provider, model, + .. } => { debug!( stage, @@ -1065,9 +1080,8 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str { WorkflowRunEvent::RunSubmitted { .. } => "run.submitted", WorkflowRunEvent::RunStarting { .. } => "run.starting", WorkflowRunEvent::RunRunning { .. } => "run.running", - WorkflowRunEvent::RunPaused { .. } => "run.paused", WorkflowRunEvent::RunRemoving { .. } => "run.removing", - WorkflowRunEvent::RunDead { .. } => "run.dead", + WorkflowRunEvent::RunRewound { .. } => "run.rewound", WorkflowRunEvent::WorkflowRunCompleted { .. } => "run.completed", WorkflowRunEvent::WorkflowRunFailed { .. } => "run.failed", WorkflowRunEvent::RunNotice { .. } => "run.notice", @@ -1314,12 +1328,16 @@ fn extract_envelope_fields(event: &WorkflowRunEvent) -> EnvelopeFields { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "stage"); let node_label = default_node_label(node_id.as_ref(), None); + let visit = fields.remove("visit"); fields.remove("session_id"); fields.remove("parent_session_id"); - let properties = fields.remove("event").map_or_else( + let mut properties = fields.remove("event").map_or_else( || Value::Object(Map::new()), |value| Value::Object(tagged_variant_fields_from_value(value)), ); + if let (Some(visit), Value::Object(map)) = (visit, &mut properties) { + map.insert("visit".to_string(), visit); + } EnvelopeFields { session_id: session_id.clone(), parent_session_id: parent_session_id.clone(), @@ -1552,6 +1570,15 @@ impl StoreProgressLogger { if let Err(err) = run_store.append_event(&payload).await { tracing::warn!(error = %err, "Failed to append event to run store"); } + if let Err(err) = + project_provider_used_from_event_payload(run_store.as_ref(), &payload) + .await + { + tracing::warn!( + error = %err, + "Failed to project provider metadata from event" + ); + } } StoreProgressCommand::Flush(tx) => { let _ = tx.send(()); @@ -1597,6 +1624,80 @@ impl StoreProgressLogger { } } +async fn project_provider_used_from_event_payload( + run_store: &dyn RunStore, + payload: &EventPayload, +) -> Result<()> { + let value = payload.as_value(); + let Some(event_name) = value.get("event").and_then(Value::as_str) else { + return Ok(()); + }; + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let Some(properties) = value.get("properties").and_then(Value::as_object) else { + return Ok(()); + }; + let Some(visit) = properties + .get("visit") + .and_then(Value::as_u64) + .and_then(|visit| u32::try_from(visit).ok()) + else { + return Ok(()); + }; + + let provider_used = match event_name { + "stage.prompt" => { + let mut provider_used = Map::new(); + if let Some(mode) = properties.get("mode").and_then(Value::as_str) { + provider_used.insert("mode".to_string(), Value::String(mode.to_string())); + } + if let Some(provider) = properties.get("provider").and_then(Value::as_str) { + provider_used.insert("provider".to_string(), Value::String(provider.to_string())); + } + if let Some(model) = properties.get("model").and_then(Value::as_str) { + provider_used.insert("model".to_string(), Value::String(model.to_string())); + } + (!provider_used.is_empty()).then_some(Value::Object(provider_used)) + } + "agent.session.started" => { + let mut provider_used = Map::new(); + provider_used.insert("mode".to_string(), Value::String("agent".to_string())); + if let Some(provider) = properties.get("provider").and_then(Value::as_str) { + provider_used.insert("provider".to_string(), Value::String(provider.to_string())); + } + if let Some(model) = properties.get("model").and_then(Value::as_str) { + provider_used.insert("model".to_string(), Value::String(model.to_string())); + } + Some(Value::Object(provider_used)) + } + "agent.cli.started" => { + let mut provider_used = Map::new(); + provider_used.insert("mode".to_string(), Value::String("cli".to_string())); + if let Some(provider) = properties.get("provider").and_then(Value::as_str) { + provider_used.insert("provider".to_string(), Value::String(provider.to_string())); + } + if let Some(model) = properties.get("model").and_then(Value::as_str) { + provider_used.insert("model".to_string(), Value::String(model.to_string())); + } + if let Some(command) = properties.get("command").and_then(Value::as_str) { + provider_used.insert("command".to_string(), Value::String(command.to_string())); + } + Some(Value::Object(provider_used)) + } + _ => None, + }; + + let Some(provider_used) = provider_used else { + return Ok(()); + }; + + run_store + .put_node_provider_used(&NodeVisitRef { node_id, visit }, &provider_used) + .await + .map_err(anyhow::Error::from) +} + /// Current time as epoch milliseconds. fn epoch_millis() -> i64 { let millis = std::time::SystemTime::now() @@ -1856,6 +1957,7 @@ mod tests { &fixtures::RUN_4, &WorkflowRunEvent::Agent { stage: "code".to_string(), + visit: 2, event: AgentEvent::ToolCallStarted { tool_name: "read_file".to_string(), tool_call_id: "call_1".to_string(), @@ -1873,6 +1975,7 @@ mod tests { assert_eq!(envelope.parent_session_id.as_deref(), Some("ses_parent")); assert_eq!(envelope.properties["tool_name"], "read_file"); assert_eq!(envelope.properties["tool_call_id"], "call_1"); + assert_eq!(envelope.properties["visit"], 2); } #[test] @@ -1975,6 +2078,7 @@ mod tests { assert_eq!( event_name(&WorkflowRunEvent::Agent { stage: "code".to_string(), + visit: 1, event: AgentEvent::SubAgentSpawned { agent_id: "a1".to_string(), depth: 1, diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index bcb7b396b..43242ded9 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -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) { + async fn make_services_with_run_store() -> ( + EngineServices, + Arc, + 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, - stage_dir: &Path, + emitter: &Arc, + _stage_dir: &Path, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { - 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 { diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 066d06164..16f41e63d 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -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 { @@ -38,6 +39,10 @@ fn build_profile(model: &str, provider: Provider) -> Box { } } +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, file_tracking: Arc>, ) { @@ -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, - stage_dir: &std::path::Path, + _stage_dir: &std::path::Path, sandbox: &Arc, tool_hooks: Option>, ) -> Result { @@ -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); diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 5fc21503d..a72ddce4b 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -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(), diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 0f1c71b32..fd7e7b797 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -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) { + async fn make_services_with_run_store() -> ( + EngineServices, + Arc, + 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 { - 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 { diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 58cf394b1..1c687c6ec 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -76,6 +76,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { 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(), diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 33293c152..a896505ba 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -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,