From e79f337077ada1b7819793c4e915a9469507e594 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 2 Apr 2026 08:33:42 -0700 Subject: [PATCH] Make workflow state fully derivable from events Complete the remaining event coverage from the events-as-source-of-truth plan. Add response and failure-signature snapshots to stage.completed, enrich retro.started and retro.completed with prompt/response data, and remove the stale script field from stage.started. Also update the internal event and run-directory docs so they match current event payloads and derivation rules. --- docs-internal/events.md | 50 ++++++++++++--- docs-internal/run-directory-keys.md | 52 +++++++-------- lib/crates/fabro-retro/src/retro_agent.rs | 41 ++++++++---- lib/crates/fabro-workflow/src/event.rs | 57 ++++++++++++++++- .../fabro-workflow/src/lifecycle/event.rs | 51 +++++++++++++-- .../fabro-workflow/src/lifecycle/mod.rs | 1 + .../fabro-workflow/src/pipeline/retro.rs | 64 +++++++++++++------ 7 files changed, 237 insertions(+), 79 deletions(-) diff --git a/docs-internal/events.md b/docs-internal/events.md index 94966ec2e..bf29a9ef7 100644 --- a/docs-internal/events.md +++ b/docs-internal/events.md @@ -169,7 +169,6 @@ Emitted when a workflow node begins execution. "properties": { "index": 1, "handler_type": "agent", - "script": null, "attempt": 1, "max_attempts": 3 } @@ -179,8 +178,7 @@ Emitted when a workflow node begins execution. | Property | Type | Description | |----------|------|-------------| | `index` | number | Stage execution order index | -| `handler_type` | string? | Handler type (`"agent"`, `"prompt"`, `"command"`, `"conditional"`, `"human"`, `"parallel"`, etc.) | -| `script` | string? | Script body (command nodes only) | +| `handler_type` | string | Handler type (`"agent"`, `"prompt"`, `"command"`, `"conditional"`, `"human"`, `"parallel"`, etc.) | | `attempt` | number | Current attempt number (1-based) | | `max_attempts` | number | Maximum attempts allowed | @@ -213,6 +211,13 @@ Emitted when a workflow node finishes execution. "error": "lint failed", "failure_class": "deterministic", "failure_signature": "clippy::unused_import", + "context_updates": {"response.code": "done"}, + "jump_to_node": "review", + "context_values": {"response.code": "done"}, + "node_visits": {"code": 1}, + "loop_failure_signatures": {"code|deterministic|clippy::unused_import": 2}, + "restart_failure_signatures": {"code|transient_infra|timeout": 1}, + "response": "done", "notes": "All tests passing", "files_touched": ["src/main.rs", "src/lib.rs"], "attempt": 1, @@ -240,6 +245,13 @@ Emitted when a workflow node finishes execution. | `error` | string? | Error message (flattened from failure detail) | | `failure_class` | string? | `"transient_infra"`, `"deterministic"`, `"budget_exhausted"`, `"compilation_loop"`, `"canceled"`, `"structural"` | | `failure_signature` | string? | Dedup key for repeated failures | +| `context_updates` | object? | Context delta written by this stage | +| `jump_to_node` | string? | Non-edge jump target | +| `context_values` | object? | Full context snapshot after the stage | +| `node_visits` | object? | Node visit counts after the stage | +| `loop_failure_signatures` | object? | Loop failure signature counts | +| `restart_failure_signatures` | object? | Restart failure signature counts | +| `response` | string? | Full LLM or agent response text when produced by the stage | | `notes` | string? | Free-text notes | | `files_touched` | string[] | File paths modified | | `attempt` | number | Attempt number (1-based) | @@ -497,7 +509,8 @@ Emitted after a checkpoint is saved. "node_label": "code", "properties": { "status": "success", - "git_commit_sha": "abc123..." + "git_commit_sha": "abc123...", + "diff": "diff --git a/src/lib.rs b/src/lib.rs\n..." } } ``` @@ -506,6 +519,7 @@ Emitted after a checkpoint is saved. |----------|------|-------------| | `status` | string | Checkpoint status | | `git_commit_sha` | string? | Commit SHA at checkpoint time | +| `diff` | string? | Git diff captured for the checkpointed node | ### `checkpoint.failed` @@ -1433,7 +1447,11 @@ Emitted after the engine completes sandbox initialization (distinct from `sandbo "id": "...", "ts": "...", "run_id": "...", "event": "sandbox.initialized", "properties": { - "working_directory": "/workspace/my-project" + "working_directory": "/workspace/my-project", + "provider": "daytona", + "identifier": "sandbox-123", + "host_working_directory": "/tmp/fabro-run/worktree", + "container_mount_point": "/workspace" } } ``` @@ -1441,6 +1459,10 @@ Emitted after the engine completes sandbox initialization (distinct from `sandbo | Property | Type | Description | |----------|------|-------------| | `working_directory` | string | Working directory inside sandbox | +| `provider` | string | Sandbox provider | +| `identifier` | string? | Provider-specific sandbox identifier | +| `host_working_directory` | string? | Host-side working directory | +| `container_mount_point` | string? | Container mount point inside the sandbox | ### `sandbox.cleanup.started` @@ -2074,11 +2096,19 @@ Emitted when the stall watchdog detects no progress. { "id": "...", "ts": "...", "run_id": "...", "event": "retro.started", - "properties": {} + "properties": { + "prompt": "Analyze the workflow run data at `/tmp/retro_data/` ...", + "provider": "anthropic", + "model": "claude-sonnet-4-20250514" + } } ``` -No properties. +| Property | Type | Description | +|----------|------|-------------| +| `prompt` | string? | Prompt sent to the retro agent | +| `provider` | string? | LLM provider for the retro agent | +| `model` | string? | Model used for the retro agent | ### `retro.completed` @@ -2087,7 +2117,9 @@ No properties. "id": "...", "ts": "...", "run_id": "...", "event": "retro.completed", "properties": { - "duration_ms": 5000 + "duration_ms": 5000, + "response": "The run was mostly smooth...", + "retro": {"smoothness": "smooth"} } } ``` @@ -2095,6 +2127,8 @@ No properties. | Property | Type | Description | |----------|------|-------------| | `duration_ms` | number | Retro duration | +| `response` | string? | Raw assistant response from the retro agent | +| `retro` | object? | Parsed `Retro` payload | ### `retro.failed` diff --git a/docs-internal/run-directory-keys.md b/docs-internal/run-directory-keys.md index 202139e09..1f36edbdf 100644 --- a/docs-internal/run-directory-keys.md +++ b/docs-internal/run-directory-keys.md @@ -118,22 +118,20 @@ Final run summary. Written when the run finishes. Retrospective analysis. Written after the retro agent completes. -No direct event mapping — this is generated by the retro agent's LLM response. The `retro.completed` event only carries `duration_ms`. - | Field | Description | Event Source | |-------|-------------|--------------| -| `run_id` | ULID string | — | -| `workflow_name` | workflow name | — | -| `goal` | workflow goal text | — | -| `timestamp` | RFC 3339 timestamp | — | -| `smoothness` | rating (optional) | — (LLM-generated) | -| `stages` | list of stage retro objects | — (LLM-generated) | -| `stats` | aggregate stats object | — (computed from stage data) | -| `intent` | what the run intended to do (optional) | — (LLM-generated) | -| `outcome` | what actually happened (optional) | — (LLM-generated) | -| `learnings` | list of learnings (optional) | — (LLM-generated) | -| `friction_points` | list of friction points (optional) | — (LLM-generated) | -| `open_items` | list of open items (optional) | — (LLM-generated) | +| `run_id` | ULID string | `retro.completed` → `envelope.run_id` | +| `workflow_name` | workflow name | `run.started` → `properties.name` | +| `goal` | workflow goal text | `run.started` → `properties.goal` | +| `timestamp` | RFC 3339 timestamp | `retro.completed` → `envelope.ts` | +| `smoothness` | rating (optional) | `retro.completed` → `properties.retro.smoothness` | +| `stages` | list of stage retro objects | `retro.completed` → `properties.retro.stages` | +| `stats` | aggregate stats object | `retro.completed` → `properties.retro.stats` | +| `intent` | what the run intended to do (optional) | `retro.completed` → `properties.retro.intent` | +| `outcome` | what actually happened (optional) | `retro.completed` → `properties.retro.outcome` | +| `learnings` | list of learnings (optional) | `retro.completed` → `properties.retro.learnings` | +| `friction_points` | list of friction points (optional) | `retro.completed` → `properties.retro.friction_points` | +| `open_items` | list of open items (optional) | `retro.completed` → `properties.retro.open_items` | ## 7. `sandbox.json` @@ -141,23 +139,23 @@ Sandbox environment details. Written when the sandbox is ready. | Field | Description | Event Source | |-------|-------------|--------------| -| `provider` | provider name | `sandbox.ready` → `properties.provider` | +| `provider` | provider name | `sandbox.initialized` → `properties.provider` | | `working_directory` | working directory in sandbox | `sandbox.initialized` → `properties.working_directory` | -| `identifier` | instance identifier (optional) | `sandbox.ready` → `properties.name` | -| `host_working_directory` | host-side path (optional) | — | -| `container_mount_point` | container mount point (optional) | — | +| `identifier` | instance identifier (optional) | `sandbox.initialized` → `properties.identifier` | +| `host_working_directory` | host-side path (optional) | `sandbox.initialized` → `properties.host_working_directory` | +| `container_mount_point` | container mount point (optional) | `sandbox.initialized` → `properties.container_mount_point` | ## 8. `workflow.fabro` Raw Graphviz dot source for the workflow graph. Plain text, not JSON. -No event source — written directly from the parsed graph. +Event source: `run.created` → `properties.workflow_source` ## 9. `workflow.toml` Workflow configuration in TOML format. Same schema as `settings` in `run.json`. -No event source — copied from the workflow definition. +Event source: `run.created` → `properties.workflow_config` ## 10. `checkpoints/{seq:04}-{epoch_ms}.json` @@ -261,19 +259,19 @@ Array of objects: |-------|-------------|--------------| | `id` | branch node id | `parallel.branch.completed` → `envelope.node_id` | | `status` | status string | `parallel.branch.completed` → `properties.status` | -| `head_sha` | git HEAD SHA (optional) | — | +| `head_sha` | git HEAD SHA (optional) | `parallel.branch.completed` → `properties.head_sha` | ## 23. `retro/prompt.md` Prompt sent to the retro agent. Plain text/markdown, not JSON. -No event source. +Event source: `retro.started` → `properties.prompt` ## 24. `retro/response.md` Response received from the retro agent. Plain text/markdown, not JSON. -No event source. +Event source: `retro.completed` → `properties.response` ## 25. `retro/status.json` @@ -289,13 +287,11 @@ Retro agent execution status. Retro agent LLM provider metadata. -No event source — written directly by the retro agent. - | Field | Description | Event Source | |-------|-------------|--------------| -| `mode` | always `"agent"` | — | -| `provider` | provider name | — | -| `model` | model identifier | — | +| `mode` | execution mode | constant `"agent"` plus `retro.started` context | +| `provider` | LLM provider | `retro.started` → `properties.provider` | +| `model` | model identifier | `retro.started` → `properties.model` | --- diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index 2ecacdae0..3e63350af 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/lib/crates/fabro-retro/src/retro_agent.rs @@ -114,6 +114,24 @@ const SUBMIT_RETRO_SCHEMA: &str = r#"{ "required": ["smoothness", "intent", "outcome"] }"#; +pub const RETRO_DATA_DIR: &str = "/tmp/retro_data"; + +pub struct RetroAgentResult { + pub narrative: RetroNarrative, + pub response: String, +} + +#[must_use] +pub fn build_retro_prompt(retro_data_dir: &str) -> String { + format!( + "Analyze the workflow run data at `{retro_data_dir}/` and generate a retrospective. \ + The key file is `{retro_data_dir}/progress.jsonl` which contains the full event stream. \ + Also check `{retro_data_dir}/checkpoint.json` for stage outcomes. \ + Use grep to search for interesting signals (failures, retries, errors, approach changes) \ + rather than reading the entire file. When done, call the `submit_retro` tool with your analysis." + ) +} + /// Run a retro agent session that analyzes workflow run data and produces /// a structured narrative. The agent explores `progress.jsonl` and other /// files via tool access, then calls `submit_retro` with its analysis. @@ -125,11 +143,10 @@ pub async fn run_retro_agent( provider: Provider, model: &str, event_callback: Option>, -) -> anyhow::Result { +) -> anyhow::Result { // Upload data files into sandbox (needed for Daytona; no-op effect for local // since the agent can also read from the original paths via tools). - let retro_data_dir = "/tmp/retro_data"; - upload_data_files(sandbox, run_store, run_dir, retro_data_dir).await?; + upload_data_files(sandbox, run_store, run_dir, RETRO_DATA_DIR).await?; // Build provider profile with the submit_retro tool let captured: Arc>> = Arc::new(Mutex::new(None)); @@ -188,13 +205,7 @@ pub async fn run_retro_agent( session.initialize().await; - let prompt = format!( - "Analyze the workflow run data at `{retro_data_dir}/` and generate a retrospective. \ - The key file is `{retro_data_dir}/progress.jsonl` which contains the full event stream. \ - Also check `{retro_data_dir}/checkpoint.json` for stage outcomes. \ - Use grep to search for interesting signals (failures, retries, errors, approach changes) \ - rather than reading the entire file. When done, call the `submit_retro` tool with your analysis." - ); + let prompt = build_retro_prompt(RETRO_DATA_DIR); write_retro_prompt(run_store, &retro_dir, &prompt).await?; @@ -213,7 +224,8 @@ pub async fn run_retro_agent( Turn::Assistant { content, .. } => Some(content.as_str()), _ => None, }) - .unwrap_or_default(); + .unwrap_or_default() + .to_string(); // Extract result / determine outcome let (outcome, failure_reason, narrative_result) = match process_result { @@ -238,7 +250,7 @@ pub async fn run_retro_agent( }; // Write artifacts (on both success and failure) - write_retro_response(run_store, &retro_dir, response_text).await?; + write_retro_response(run_store, &retro_dir, &response_text).await?; write_retro_artifacts( &retro_dir, provider.as_str(), @@ -254,7 +266,10 @@ pub async fn run_retro_agent( let _ = handle.await; } - narrative_result + narrative_result.map(|narrative| RetroAgentResult { + narrative, + response: response_text, + }) } /// Return a placeholder narrative for dry-run mode. Exercises the full diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index ec11c5e72..06c5c2ea4 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -109,8 +109,6 @@ pub enum WorkflowRunEvent { name: String, index: usize, handler_type: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - script: Option, attempt: usize, max_attempts: usize, }, @@ -139,6 +137,8 @@ pub enum WorkflowRunEvent { loop_failure_signatures: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] restart_failure_signatures: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + response: Option, attempt: usize, max_attempts: usize, }, @@ -442,6 +442,8 @@ pub enum WorkflowRunEvent { stderr: String, }, RetroStarted { + #[serde(default, skip_serializing_if = "Option::is_none")] + prompt: Option, #[serde(default, skip_serializing_if = "Option::is_none")] provider: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -450,6 +452,8 @@ pub enum WorkflowRunEvent { RetroCompleted { duration_ms: u64, #[serde(default, skip_serializing_if = "Option::is_none")] + response: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] retro: Option, }, RetroFailed { @@ -945,7 +949,11 @@ impl WorkflowRunEvent { command, index, exit_code, "Devcontainer lifecycle command failed" ); } - Self::RetroStarted { provider, model } => { + Self::RetroStarted { + prompt: _, + provider, + model, + } => { info!( provider = provider.as_deref().unwrap_or(""), model = model.as_deref().unwrap_or(""), @@ -1660,6 +1668,7 @@ mod tests { node_visits: None, loop_failure_signatures: None, restart_failure_signatures: None, + response: None, attempt: 1, max_attempts: 1, }, @@ -1674,6 +1683,42 @@ mod tests { assert!(envelope.session_id.is_none()); } + #[test] + fn canonicalize_stage_completed_keeps_response_and_signature_snapshots() { + let envelope = canonicalize_event( + &fixtures::RUN_2, + &WorkflowRunEvent::StageCompleted { + node_id: "plan".to_string(), + name: "Plan".to_string(), + index: 0, + duration_ms: 5000, + status: "success".to_string(), + preferred_label: None, + suggested_next_ids: Vec::new(), + usage: None, + failure: None, + notes: None, + files_touched: Vec::new(), + context_updates: None, + jump_to_node: None, + context_values: None, + node_visits: None, + loop_failure_signatures: Some(BTreeMap::from([("sig-a".to_string(), 2usize)])), + restart_failure_signatures: Some(BTreeMap::from([("sig-b".to_string(), 1usize)])), + response: Some("done".to_string()), + attempt: 1, + max_attempts: 1, + }, + ); + + assert_eq!(envelope.properties["response"], "done"); + assert_eq!(envelope.properties["loop_failure_signatures"]["sig-a"], 2); + assert_eq!( + envelope.properties["restart_failure_signatures"]["sig-b"], + 1 + ); + } + #[test] fn canonicalize_stage_failure_flattens_failure_detail() { let envelope = canonicalize_event( @@ -1786,6 +1831,7 @@ mod tests { let envelope = canonicalize_event( &fixtures::RUN_8, &WorkflowRunEvent::RetroStarted { + prompt: Some("Analyze the run".to_string()), provider: None, model: None, }, @@ -1794,12 +1840,17 @@ mod tests { let payload = build_redacted_event_payload(&envelope, &fixtures::RUN_8).unwrap(); assert_eq!(payload.as_value()["id"], envelope.id); assert_eq!(payload.as_value()["event"], "retro.started"); + assert_eq!( + payload.as_value()["properties"]["prompt"], + "Analyze the run" + ); } #[test] fn event_name_matches_new_dot_notation() { assert_eq!( event_name(&WorkflowRunEvent::RetroStarted { + prompt: None, provider: None, model: None, }), diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index d4217d16b..159995801 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -12,8 +12,10 @@ use fabro_core::lifecycle::{ use fabro_core::outcome::NodeResult; use fabro_core::state::RunState; +use super::circuit_breaker::CircuitBreakerLifecycle; use super::git::GitCheckpointResult; use crate::artifact::ArtifactStore; +use crate::context; use crate::error::FabroError; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::graph::WorkflowGraph; @@ -45,6 +47,36 @@ pub(crate) struct EventLifecycle { // Cross-lifecycle data pub checkpoint_git_result: Arc>>, pub last_git_sha: Arc>>, + pub circuit_breaker: Arc, +} + +fn snapshot_failure_signatures( + circuit_breaker: &CircuitBreakerLifecycle, +) -> ( + Option>, + Option>, +) { + let (loop_sigs, restart_sigs) = circuit_breaker.snapshot(); + let loop_sigs = (!loop_sigs.is_empty()).then(|| { + loop_sigs + .into_iter() + .map(|(sig, count)| (sig.to_string(), count)) + .collect::>() + }); + let restart_sigs = (!restart_sigs.is_empty()).then(|| { + restart_sigs + .into_iter() + .map(|(sig, count)| (sig.to_string(), count)) + .collect::>() + }); + (loop_sigs, restart_sigs) +} + +fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { + outcome + .context_updates + .get(&context::keys::response_key(node_id)) + .and_then(|value| value.as_str().map(ToOwned::to_owned)) } #[async_trait] @@ -87,12 +119,13 @@ impl RunLifecycle for EventLifecycle { } let gv = node.inner(); let stage_index = state.stage_index; + let (loop_failure_signatures, restart_failure_signatures) = + snapshot_failure_signatures(&self.circuit_breaker); self.emitter.emit(&WorkflowRunEvent::StageStarted { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, handler_type: gv.handler_type().unwrap_or_default().to_string(), - script: None, attempt: 1, max_attempts: 1, }); @@ -112,8 +145,12 @@ impl RunLifecycle for EventLifecycle { jump_to_node: None, context_values: None, node_visits: None, - loop_failure_signatures: None, - restart_failure_signatures: None, + loop_failure_signatures, + restart_failure_signatures, + response: state + .context + .get(&context::keys::response_key(&gv.id)) + .and_then(|value| value.as_str().map(ToOwned::to_owned)), attempt: 1, max_attempts: 1, }); @@ -130,7 +167,6 @@ impl RunLifecycle for EventLifecycle { name: gv.label().to_string(), index: state.stage_index, handler_type: gv.handler_type().unwrap_or_default().to_string(), - script: None, attempt: ctx.attempt as usize, max_attempts: ctx.max_attempts as usize, }); @@ -185,6 +221,8 @@ impl RunLifecycle for EventLifecycle { let gv = node.inner(); let stage_index = state.stage_index; let duration_ms = u64::try_from(result.duration.as_millis()).unwrap(); + let (loop_failure_signatures, restart_failure_signatures) = + snapshot_failure_signatures(&self.circuit_breaker); if outcome.status == StageStatus::Fail { self.emitter.emit(&WorkflowRunEvent::StageFailed { @@ -228,8 +266,9 @@ impl RunLifecycle for EventLifecycle { .into_iter() .collect::>() }), - loop_failure_signatures: None, - restart_failure_signatures: None, + loop_failure_signatures, + restart_failure_signatures, + response: response_from_outcome(&gv.id, outcome), attempt: result.attempts as usize, max_attempts: result.max_attempts as usize, }); diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index cbbca5aa0..ddb1969fc 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -128,6 +128,7 @@ impl WorkflowLifecycle { artifact_store: Arc::clone(&artifact_store), last_git_sha: Arc::clone(&last_git_sha), checkpoint_git_result: Arc::clone(&checkpoint_git_result), + circuit_breaker: Arc::clone(&circuit_breaker), }; let hook = HookLifecycle { diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 4d27b7b13..58cf394b1 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -2,7 +2,9 @@ use std::sync::Arc; use fabro_agent::SessionEvent; use fabro_retro::retro::{Retro, derive_retro}; -use fabro_retro::retro_agent::{dry_run_narrative, run_retro_agent}; +use fabro_retro::retro_agent::{ + RETRO_DATA_DIR, build_retro_prompt, dry_run_narrative, run_retro_agent, +}; use super::types::{Executed, RetroOptions, Retroed}; use crate::event::WorkflowRunEvent; @@ -54,15 +56,17 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { } let retro_start = std::time::Instant::now(); + let retro_prompt = build_retro_prompt(RETRO_DATA_DIR); if let Some(ref emitter) = options.emitter { emitter.emit(&WorkflowRunEvent::RetroStarted { + prompt: Some(retro_prompt), provider: Some(options.provider.as_str().to_string()), model: Some(options.model.clone()), }); } - let narrative_result = if dry_run { - Ok(dry_run_narrative()) + let retro_result = if dry_run { + Ok((dry_run_narrative(), String::new())) } else if let Some(client) = options.llm_client.as_ref() { let emitter_clone = options.emitter.clone(); let event_callback: Option> = @@ -89,32 +93,33 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { event_callback, ) .await + .map(|result| (result.narrative, result.response)) } else { Err(anyhow::anyhow!("No LLM client available")) }; let duration_ms = u64::try_from(retro_start.elapsed().as_millis()).unwrap(); - if let Some(ref emitter) = options.emitter { - match &narrative_result { - Ok(_) => emitter.emit(&WorkflowRunEvent::RetroCompleted { - duration_ms, - retro: serde_json::to_value(&retro).ok(), - }), - Err(e) => emitter.emit(&WorkflowRunEvent::RetroFailed { - error: e.to_string(), - duration_ms, - }), - } - } - - match narrative_result { - Ok(narrative) => { + match retro_result { + Ok((narrative, response)) => { retro.apply_narrative(narrative); + if let Some(ref emitter) = options.emitter { + emitter.emit(&WorkflowRunEvent::RetroCompleted { + duration_ms, + response: Some(response), + retro: serde_json::to_value(&retro).ok(), + }); + } if let Err(err) = options.run_store.put_retro(&retro).await { tracing::warn!(error = %err, "Failed to save retro with narrative to store"); } } Err(e) => { + if let Some(ref emitter) = options.emitter { + emitter.emit(&WorkflowRunEvent::RetroFailed { + error: e.to_string(), + duration_ms, + }); + } tracing::debug!(error = %e, "Retro agent skipped"); } } @@ -298,7 +303,7 @@ mod tests { let seen = Arc::new(Mutex::new(Vec::new())); emitter.on_event({ let seen = Arc::clone(&seen); - move |event| seen.lock().unwrap().push(event.event.clone()) + move |event| seen.lock().unwrap().push(event.clone()) }); let retro = run_retro( @@ -325,7 +330,24 @@ mod tests { assert!(retro.is_some()); let seen = seen.lock().unwrap(); - assert!(seen.iter().any(|event| event == "retro.started")); - assert!(seen.iter().any(|event| event == "retro.completed")); + let retro_started = seen + .iter() + .find(|event| event.event == "retro.started") + .unwrap(); + assert_eq!(retro_started.properties["provider"], "anthropic"); + assert_eq!(retro_started.properties["model"], "test-model"); + assert!( + retro_started.properties["prompt"] + .as_str() + .is_some_and(|prompt| prompt.contains("/tmp/retro_data/progress.jsonl")) + ); + + let retro_completed = seen + .iter() + .find(|event| event.event == "retro.completed") + .unwrap(); + assert_eq!(retro_completed.properties["response"], ""); + assert!(retro_completed.properties.get("retro").is_some()); + assert_eq!(retro_completed.properties["retro"]["smoothness"], "smooth"); } }