From 631d9a96e102238366fc183ce4470e43d5c79d68 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 2 Apr 2026 10:02:31 -0700 Subject: [PATCH] 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());