mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-13 23:14:17 +00:00
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.
This commit is contained in:
Bryan Helmkamp
parent
28a1fc72f4
commit
e79f337077
7 changed files with 237 additions and 79 deletions
|
|
@ -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`
|
||||
|
||||
|
|
|
|||
|
|
@ -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` |
|
||||
|
||||
---
|
||||
|
||||
|
|
|
|||
|
|
@ -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<Arc<dyn Fn(SessionEvent) + Send + Sync>>,
|
||||
) -> anyhow::Result<RetroNarrative> {
|
||||
) -> anyhow::Result<RetroAgentResult> {
|
||||
// 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<Mutex<Option<RetroNarrative>>> = 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
|
||||
|
|
|
|||
|
|
@ -109,8 +109,6 @@ pub enum WorkflowRunEvent {
|
|||
name: String,
|
||||
index: usize,
|
||||
handler_type: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
script: Option<String>,
|
||||
attempt: usize,
|
||||
max_attempts: usize,
|
||||
},
|
||||
|
|
@ -139,6 +137,8 @@ pub enum WorkflowRunEvent {
|
|||
loop_failure_signatures: Option<BTreeMap<String, usize>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
restart_failure_signatures: Option<BTreeMap<String, usize>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
response: Option<String>,
|
||||
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<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
provider: Option<String>,
|
||||
#[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<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
retro: Option<serde_json::Value>,
|
||||
},
|
||||
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,
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -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<Mutex<Option<GitCheckpointResult>>>,
|
||||
pub last_git_sha: Arc<Mutex<Option<String>>>,
|
||||
pub circuit_breaker: Arc<CircuitBreakerLifecycle>,
|
||||
}
|
||||
|
||||
fn snapshot_failure_signatures(
|
||||
circuit_breaker: &CircuitBreakerLifecycle,
|
||||
) -> (
|
||||
Option<BTreeMap<String, usize>>,
|
||||
Option<BTreeMap<String, usize>>,
|
||||
) {
|
||||
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::<BTreeMap<_, _>>()
|
||||
});
|
||||
let restart_sigs = (!restart_sigs.is_empty()).then(|| {
|
||||
restart_sigs
|
||||
.into_iter()
|
||||
.map(|(sig, count)| (sig.to_string(), count))
|
||||
.collect::<BTreeMap<_, _>>()
|
||||
});
|
||||
(loop_sigs, restart_sigs)
|
||||
}
|
||||
|
||||
fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option<String> {
|
||||
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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> for EventLifecycle {
|
|||
.into_iter()
|
||||
.collect::<BTreeMap<_, _>>()
|
||||
}),
|
||||
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,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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<Retro> {
|
|||
}
|
||||
|
||||
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<Arc<dyn Fn(SessionEvent) + Send + Sync>> =
|
||||
|
|
@ -89,32 +93,33 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
|
|||
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");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue