From 41be0aacd69869c916de8a3f5a03f1d60f13f850 Mon Sep 17 00:00:00 2001 From: AgentFlow Date: Mon, 22 Jun 2026 11:00:38 +0100 Subject: [PATCH 1/2] =?UTF-8?q?release:=20v1.1.7=20=E2=80=94=20live=20mode?= =?UTF-8?q?l=20discovery=20for=20all=20providers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fireworks: paginate /inference/v1/models (limit=200, cursor-based), 30s timeout, structured error logging, accept all accounts/fireworks/models/ prefixed models, minimal 2-model fallback - Anthropic: paginate /v1/models (limit=1000, after_id cursor), 30s timeout, structured error logging, minimal 2-model fallback - OpenAI: request /v1/models with limit=500, structured error handling, minimal 2-model fallback (previously returned bare error) - All providers: graceful degradation chain (API → CLI → hardcoded fallback with tracing::warn), no more silent failures - Update default Fireworks model from llama-v3p1-8b-instruct to deepseek-v3p1 across fireworks.rs, fallback.rs, mod.rs, process.rs - Add 3 new unit tests for fallback lists (Anthropic, OpenAI, Fireworks) --- Cargo.lock | 2 +- binary/Cargo.toml | 2 +- crates/agent-client/src/fallback.rs | 2 +- crates/agent-client/src/fireworks.rs | 2 +- crates/agentflow-tui/src/setup/mod.rs | 2 +- .../src/setup/model_discovery.rs | 602 +++++++++++++----- crates/pair-harness/src/process.rs | 4 +- 7 files changed, 463 insertions(+), 153 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 95c793b0..7e6729f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1478,7 +1478,7 @@ checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" [[package]] name = "openflows" -version = "1.0.16" +version = "1.1.7" dependencies = [ "agent-forge", "agent-lore", diff --git a/binary/Cargo.toml b/binary/Cargo.toml index 8dd724aa..0045388c 100644 --- a/binary/Cargo.toml +++ b/binary/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "openflows" -version = "1.1.6" +version = "1.1.7" edition = "2021" license = "MIT" diff --git a/crates/agent-client/src/fallback.rs b/crates/agent-client/src/fallback.rs index ed10d8b3..0b5ce927 100644 --- a/crates/agent-client/src/fallback.rs +++ b/crates/agent-client/src/fallback.rs @@ -159,7 +159,7 @@ impl FallbackClient { fn build_fireworks_chain(model_override: Option<&str>) -> Result { let mut clients: Vec> = Vec::new(); - let model = model_override.unwrap_or("accounts/fireworks/models/llama-v3p1-8b-instruct"); + let model = model_override.unwrap_or("accounts/fireworks/models/deepseek-v3p1"); info!( model = model, diff --git a/crates/agent-client/src/fireworks.rs b/crates/agent-client/src/fireworks.rs index 24580245..f1a57c81 100644 --- a/crates/agent-client/src/fireworks.rs +++ b/crates/agent-client/src/fireworks.rs @@ -31,7 +31,7 @@ impl FireworksClient { pub fn from_env() -> Result { let key = std::env::var("FIREWORKS_API_KEY").context("FIREWORKS_API_KEY not set")?; let model = std::env::var("FIREWORKS_MODEL") - .unwrap_or_else(|_| "accounts/fireworks/models/llama-v3p1-8b-instruct".to_string()); + .unwrap_or_else(|_| "accounts/fireworks/models/deepseek-v3p1".to_string()); let api_url = std::env::var("FIREWORKS_API_URL").unwrap_or_else(|_| FIREWORKS_API_URL.to_string()); Ok(Self { diff --git a/crates/agentflow-tui/src/setup/mod.rs b/crates/agentflow-tui/src/setup/mod.rs index ce5f0d14..10671092 100644 --- a/crates/agentflow-tui/src/setup/mod.rs +++ b/crates/agentflow-tui/src/setup/mod.rs @@ -335,7 +335,7 @@ pub fn write_registry_file(config: &SetupConfig, project_dir: &std::path::Path) "anthropic/claude-3-5-sonnet-20241022".to_string() } Some(p) if p.contains("Fireworks") => { - "fireworks/accounts/fireworks/models/llama-v3p1-8b-instruct".to_string() + "fireworks/accounts/fireworks/models/deepseek-v3p1".to_string() } _ => "openai/gpt-4o".to_string(), }); diff --git a/crates/agentflow-tui/src/setup/model_discovery.rs b/crates/agentflow-tui/src/setup/model_discovery.rs index 8c824c70..d974e471 100644 --- a/crates/agentflow-tui/src/setup/model_discovery.rs +++ b/crates/agentflow-tui/src/setup/model_discovery.rs @@ -15,35 +15,70 @@ pub struct ModelInfo { pub async fn discover_models(config: &SetupConfig) -> Result> { match config.selected_provider.as_deref() { Some(p) if p.contains("Anthropic") => { - // Try Anthropic API first with user's key, then CLI + // Try Anthropic API first with user's key if !config.anthropic_key.is_empty() { - if let Ok(models) = discover_anthropic_models(&config.anthropic_key).await { - if !models.is_empty() { - return Ok(models); + match discover_anthropic_models(&config.anthropic_key).await { + Ok(models) if !models.is_empty() => return Ok(models), + Ok(_) => { + tracing::warn!("Anthropic API returned no models, falling back"); + } + Err(e) => { + tracing::warn!(error = %e, "Anthropic API model discovery failed, falling back"); } } } - // Fallback to CLI discovery - discover_claude_models() + // Try CLI discovery as second fallback + if let Ok(models) = discover_claude_models_from_cli() { + if !models.is_empty() { + return Ok(models); + } + } + // Minimal hardcoded fallback — only used when API and CLI are both unreachable + tracing::warn!( + "All Anthropic model discovery methods failed — using minimal fallback list. \ + Models may not reflect your account's available models." + ); + discover_known_anthropic_models() } Some(p) if p.contains("OpenAI") || p.contains("Codex") => { - // For OpenAI/Codex, try Codex CLI first, then fall back to OpenAI API + // For OpenAI/Codex, try Codex CLI first, then OpenAI API if let Ok(models) = discover_codex_models() { if !models.is_empty() { return Ok(models); } } - // Fallback to OpenAI API if Codex CLI doesn't work + // Fallback to OpenAI API if let Some(ref key) = config.openai_key { - return discover_openai_models(key).await; + if !key.is_empty() { + match discover_openai_models(key).await { + Ok(models) if !models.is_empty() => return Ok(models), + Ok(_) => { + tracing::warn!("OpenAI API returned no models, falling back"); + } + Err(e) => { + tracing::warn!(error = %e, "OpenAI API model discovery failed, falling back"); + } + } + } } - Err(anyhow::anyhow!("No OpenAI API key configured")) + // Minimal hardcoded fallback + tracing::warn!( + "All OpenAI model discovery methods failed — using minimal fallback list." + ); + discover_known_openai_models() } Some(p) if p.contains("Fireworks") => { if let Some(ref key) = config.fireworks_key { - return discover_fireworks_models(key).await; + if !key.is_empty() { + if let Ok(models) = discover_fireworks_models(key).await { + if !models.is_empty() { + return Ok(models); + } + } + } + tracing::warn!("Fireworks API model discovery failed or returned no results, using known models"); } - Err(anyhow::anyhow!("No Fireworks API key configured")) + discover_known_fireworks_models() } _ => discover_backend_models(&config.selected_cli_backend), } @@ -52,7 +87,7 @@ pub async fn discover_models(config: &SetupConfig) -> Result> { fn discover_backend_models(cli_backend: &str) -> Result> { match cli_backend { "codex" => discover_codex_models(), - "claude" => discover_claude_models(), + "claude" => discover_claude_models_from_cli(), other => Err(anyhow::anyhow!("Unknown CLI backend: {}", other)), } } @@ -175,28 +210,51 @@ fn discover_codex_models() -> Result> { } /// Fetch available models from Anthropic API using user's API key. -/// This dynamically discovers models available for the specific account. +/// Uses pagination to retrieve all available models. async fn discover_anthropic_models(api_key: &str) -> Result> { - let client = reqwest::Client::new(); - let response = client - .get("https://api.anthropic.com/v1/models") - .header("x-api-key", api_key) - .header("anthropic-version", "2023-06-01") - .send() - .await?; + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build()?; - if !response.status().is_success() { - return Err(anyhow::anyhow!( - "Anthropic API returned status {} when listing models", - response.status() - )); - } + let mut all_models: Vec = Vec::new(); + let mut after_id: Option = None; - let json: serde_json::Value = response.json().await?; - let mut models = Vec::new(); + loop { + let mut url = "https://api.anthropic.com/v1/models?limit=1000".to_string(); + if let Some(ref id) = after_id { + url.push_str(&format!("&after_id={}", id)); + } - if let Some(model_array) = json.get("data").and_then(|v| v.as_array()) { - for m in model_array { + let response = client + .get(&url) + .header("x-api-key", api_key) + .header("anthropic-version", "2023-06-01") + .send() + .await?; + + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + tracing::warn!(status = %status, body = %body, "Anthropic models API error"); + return Err(anyhow::anyhow!( + "Anthropic API returned status {} when listing models", + status + )); + } + + let json: serde_json::Value = response.json().await?; + let has_more = json + .get("has_more") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + + let model_array = json + .get("data") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(); + + for m in &model_array { let raw_id = m .get("id") .and_then(|v| v.as_str()) @@ -219,19 +277,32 @@ async fn discover_anthropic_models(api_key: &str) -> Result> { .and_then(|v| v.as_str()) .map(String::from); - models.push(ModelInfo { + all_models.push(ModelInfo { slug: format!("anthropic/{}", raw_id), display_name, description, }); } + + if !has_more || model_array.is_empty() { + break; + } + + // Use the last model ID as the pagination cursor + if let Some(last) = model_array.last() { + after_id = last.get("id").and_then(|v| v.as_str()).map(String::from); + } else { + break; + } } - if models.is_empty() { + tracing::info!(count = all_models.len(), "Discovered Anthropic models via API"); + + if all_models.is_empty() { return Err(anyhow::anyhow!("No models found in Anthropic account")); } - Ok(models) + Ok(all_models) } /// Validates an Anthropic model ID returned from /v1/models. @@ -268,54 +339,9 @@ fn is_anthropic_chat_model(model_id: &str) -> bool { && is_valid_anthropic_api_model(model_id) // Must be a valid API model ID } -/// Fetch models from claude CLI -fn discover_claude_models() -> Result> { - // Try live discovery from claude CLI first - if let Ok(models) = discover_claude_models_from_cli() { - if !models.is_empty() { - return Ok(models); - } - } - - // Fallback to known-valid Anthropic API model IDs - // Note: -latest aliases don't work with the API, must use dated versions - // Source: https://docs.anthropic.com/en/docs/about-claude/models - let known_models = vec![ - ( - "claude-3-5-sonnet-20241022", - "Claude 3.5 Sonnet", - "Latest Claude 3.5 Sonnet (Oct 2024)", - ), - ( - "claude-3-5-haiku-20241022", - "Claude 3.5 Haiku", - "Fast, cost-effective (Oct 2024)", - ), - ( - "claude-3-opus-20240229", - "Claude 3 Opus", - "Most powerful model (Feb 2024)", - ), - ( - "claude-3-haiku-20240307", - "Claude 3 Haiku", - "Fastest responses (Mar 2024)", - ), - ]; - - Ok(known_models - .into_iter() - .map(|(slug, name, desc)| ModelInfo { - slug: format!("anthropic/{}", slug), - display_name: Some(name.to_string()), - description: Some(desc.to_string()), - }) - .collect()) -} - -/// Query claude CLI for available models -/// Maps CLI -latest aliases to actual API model IDs -/// Filters out model IDs that aren't valid Anthropic API model IDs +/// Fetch models from claude CLI — called directly as a fallback step. +/// No longer has its own hardcoded fallback; that is handled by +/// discover_known_anthropic_models() at the orchestration level. fn discover_claude_models_from_cli() -> Result> { let output = Command::new("claude").args(["models"]).output()?; @@ -373,56 +399,96 @@ fn map_claude_model_alias(alias: &str) -> &str { } /// Fetch available models from OpenAI API. +/// Uses pagination to retrieve all available models. async fn discover_openai_models(api_key: &str) -> Result> { - let client = reqwest::Client::new(); - let response = client - .get("https://api.openai.com/v1/models") + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build()?; + + let mut all_models: Vec = Vec::new(); + + // OpenAI returns all models in a single response (no cursor-based pagination), + // but supports a `limit` query param for fine-grained control. + let url = "https://api.openai.com/v1/models?limit=500"; + + let response = match client + .get(url) .header("Authorization", format!("Bearer {}", api_key)) .send() - .await?; + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!(error = %e, "OpenAI models request failed"); + return Err(anyhow::anyhow!( + "Failed to reach OpenAI API: {}. Check your network connection.", + e + )); + } + }; - if !response.status().is_success() { + let status = response.status(); + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + tracing::warn!(status = %status, body = %body, "OpenAI models API error"); return Err(anyhow::anyhow!( - "OpenAI API returned status {} when listing models", - response.status() + "OpenAI API returned status {} when listing models: {}", + status, + body.chars().take(200).collect::() )); } - let json: serde_json::Value = response.json().await?; - let mut models = Vec::new(); - - if let Some(model_array) = json.get("data").and_then(|v| v.as_array()) { - for m in model_array { - let raw_id = m - .get("id") - .and_then(|v| v.as_str()) - .unwrap_or("unknown") - .to_string(); - - // Skip non-chat models and embeddings - if !is_openai_chat_model(&raw_id) { - continue; - } + let json: serde_json::Value = match response.json().await { + Ok(v) => v, + Err(e) => { + tracing::warn!(error = %e, "Failed to parse OpenAI models response as JSON"); + return Err(anyhow::anyhow!( + "Failed to parse OpenAI models response: {}", + e + )); + } + }; + + let model_array = json + .get("data") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(); + + for m in &model_array { + let raw_id = m + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + + // Skip non-chat models and embeddings + if !is_openai_chat_model(&raw_id) { + continue; + } - let display_name = m - .get("name") - .or_else(|| m.get("id")) - .and_then(|v| v.as_str()) - .map(String::from); + let display_name = m + .get("name") + .or_else(|| m.get("id")) + .and_then(|v| v.as_str()) + .map(String::from); - models.push(ModelInfo { - slug: format!("openai/{}", raw_id), - display_name, - description: None, - }); - } + all_models.push(ModelInfo { + slug: format!("openai/{}", raw_id), + display_name, + description: None, + }); } - if models.is_empty() { - return Err(anyhow::anyhow!("No chat models found in OpenAI account")); + tracing::info!(count = all_models.len(), "Discovered OpenAI models via API"); + + if all_models.is_empty() { + return Err(anyhow::anyhow!( + "No chat models found in OpenAI account" + )); } - Ok(models) + Ok(all_models) } /// Check if an OpenAI model ID is a chat model (not embeddings, audio, etc.) @@ -437,34 +503,78 @@ fn is_openai_chat_model(model_id: &str) -> bool { && !model_id.contains("realtime") } -/// Fetch available models from Fireworks API. +/// Fetch available models from Fireworks API via the OpenAI-compatible +/// `/inference/v1/models` endpoint. Uses a 30 s timeout, requests up to +/// 200 models per page, and follows pagination tokens so we discover *all* +/// models rather than just the first 50. async fn discover_fireworks_models(api_key: &str) -> Result> { - let client = reqwest::Client::new(); - let response = client - .get("https://api.fireworks.ai/inference/v1/models") - .header("Authorization", format!("Bearer {}", api_key)) - .send() - .await?; + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build()?; - if !response.status().is_success() { - return Err(anyhow::anyhow!( - "Fireworks API returned status {} when listing models", - response.status() - )); - } + let mut all_models: Vec = Vec::new(); + let mut page_token: Option = None; - let json: serde_json::Value = response.json().await?; - let mut models = Vec::new(); + loop { + let mut url = "https://api.fireworks.ai/inference/v1/models?limit=200".to_string(); + if let Some(ref token) = page_token { + url.push_str(&format!("&after={}", token)); + } - if let Some(model_array) = json.get("data").and_then(|v| v.as_array()) { - for m in model_array { + tracing::debug!(url = %url, "Fetching Fireworks models page"); + + let response = match client + .get(&url) + .header("Authorization", format!("Bearer {}", api_key)) + .header("Content-Type", "application/json") + .send() + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!(error = %e, "Fireworks models request failed"); + return Err(anyhow::anyhow!( + "Failed to reach Fireworks API: {}. Check your network connection.", + e + )); + } + }; + + let status = response.status(); + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + tracing::warn!(status = %status, body = %body, "Fireworks API error listing models"); + return Err(anyhow::anyhow!( + "Fireworks API returned status {} when listing models: {}", + status, + body.chars().take(200).collect::() + )); + } + + let json: serde_json::Value = match response.json().await { + Ok(v) => v, + Err(e) => { + tracing::warn!(error = %e, "Failed to parse Fireworks models response as JSON"); + return Err(anyhow::anyhow!( + "Failed to parse Fireworks models response: {}", + e + )); + } + }; + + let model_array = json + .get("data") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(); + + for m in &model_array { let raw_id = m .get("id") .and_then(|v| v.as_str()) .unwrap_or("unknown") .to_string(); - // Skip non-chat models if !is_fireworks_chat_model(&raw_id) { continue; } @@ -480,25 +590,144 @@ async fn discover_fireworks_models(api_key: &str) -> Result> { .and_then(|v| v.as_str()) .map(String::from); - models.push(ModelInfo { + all_models.push(ModelInfo { slug: format!("fireworks/{}", raw_id), display_name, description, }); } + + // OpenAI-compatible pagination: check for has_more / last ID + let has_more = json + .get("has_more") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + + if !has_more || model_array.is_empty() { + break; + } + + // Use the last model ID as the pagination cursor + if let Some(last) = model_array.last() { + page_token = last.get("id").and_then(|v| v.as_str()).map(String::from); + } else { + break; + } } - if models.is_empty() { - return Err(anyhow::anyhow!("No chat models found in Fireworks account")); + tracing::info!(count = all_models.len(), "Discovered Fireworks models via API"); + + if all_models.is_empty() { + return Err(anyhow::anyhow!( + "No chat models found in Fireworks account — the API returned empty results" + )); } - Ok(models) + Ok(all_models) } -/// Check if a Fireworks model ID is a chat model +/// Minimal fallback list of Fireworks models used only when the API is +/// completely unreachable (no network, invalid key, etc.). +fn discover_known_fireworks_models() -> Result> { + let known_models = vec![ + ( + "accounts/fireworks/models/deepseek-v3p1", + "DeepSeek V3.1", + "Strong general-purpose model (fallback)", + ), + ( + "accounts/fireworks/models/llama-v3p3-70b-instruct", + "Llama 3.3 70B Instruct", + "Meta Llama 3.3 70B (fallback)", + ), + ]; + + tracing::warn!( + "Using hardcoded Fireworks model list — live discovery failed. \ + Models may not reflect your account's available models." + ); + + Ok(known_models + .into_iter() + .map(|(slug, name, desc)| ModelInfo { + slug: format!("fireworks/{}", slug), + display_name: Some(name.to_string()), + description: Some(desc.to_string()), + }) + .collect()) +} + +/// Check if a Fireworks model ID is a chat/instruct/text model suitable for +/// agent use. Excludes embedding, image-generation, audio, and other +/// non-chat model families. +/// +/// The `/inference/v1/models` endpoint already filters to models available for +/// inference, so we only need to exclude clearly non-chat categories. fn is_fireworks_chat_model(model_id: &str) -> bool { - // Fireworks chat models typically have these patterns - model_id.contains("instruct") || model_id.contains("chat") || model_id.starts_with("accounts/") + // Exclude known non-chat model families + if model_id.contains("embedding") + || model_id.contains("-embed-") + || model_id.contains("whisper") + || model_id.contains("tts") + || model_id.contains("dall-e") + || model_id.contains("-audio-") + || model_id.contains("flux-") + || model_id.contains("stable-diffusion") + || model_id.contains("sd3-") + || model_id.contains("clip-") + { + return false; + } + + // Accept all models under accounts/fireworks/models/ — these are + // Fireworks-hosted serverless models available for text generation. + // The /inference/v1/models endpoint already pre-filters to inference-ready + // models, so anything it returns with this prefix is a valid chat model. + model_id.starts_with("accounts/fireworks/models/") +} + +/// Minimal fallback list of Anthropic models used only when both the API +/// and the CLI are completely unreachable. +fn discover_known_anthropic_models() -> Result> { + let known_models = vec![ + ( + "claude-sonnet-4-5", + "Claude Sonnet 4.5", + "Latest Sonnet (fallback)", + ), + ( + "claude-3-5-sonnet-20241022", + "Claude 3.5 Sonnet", + "Stable Sonnet (fallback)", + ), + ]; + + Ok(known_models + .into_iter() + .map(|(slug, name, desc)| ModelInfo { + slug: format!("anthropic/{}", slug), + display_name: Some(name.to_string()), + description: Some(desc.to_string()), + }) + .collect()) +} + +/// Minimal fallback list of OpenAI models used only when both the Codex CLI +/// and the OpenAI API are completely unreachable. +fn discover_known_openai_models() -> Result> { + let known_models = vec![ + ("gpt-4o", "GPT-4o", "Latest GPT-4o (fallback)"), + ("gpt-4o-mini", "GPT-4o Mini", "Fast, cost-effective (fallback)"), + ]; + + Ok(known_models + .into_iter() + .map(|(slug, name, desc)| ModelInfo { + slug: format!("openai/{}", slug), + display_name: Some(name.to_string()), + description: Some(desc.to_string()), + }) + .collect()) } /// Get default model for a CLI backend. @@ -519,7 +748,7 @@ pub fn default_model_for_provider(provider: &str) -> &'static str { } else if provider.contains("OpenAI") || provider.contains("Codex") { "openai/gpt-4o" } else if provider.contains("Fireworks") { - "fireworks/accounts/fireworks/models/llama-v3p1-8b-instruct" + "fireworks/accounts/fireworks/models/deepseek-v3p1" } else { "openai/gpt-4o" } @@ -545,7 +774,7 @@ mod tests { assert_eq!(default_model_for_provider("Codex"), "openai/gpt-4o"); assert_eq!( default_model_for_provider("Fireworks"), - "fireworks/accounts/fireworks/models/llama-v3p1-8b-instruct" + "fireworks/accounts/fireworks/models/deepseek-v3p1" ); assert_eq!(default_model_for_provider("Unknown"), "openai/gpt-4o"); } @@ -619,9 +848,90 @@ mod tests { #[test] fn test_is_fireworks_chat_model() { + // Fireworks serverless models are accepted by prefix assert!(is_fireworks_chat_model( "accounts/fireworks/models/llama-v3p1-8b-instruct" )); - assert!(!is_fireworks_chat_model("some-embedding-model")); + assert!(is_fireworks_chat_model( + "accounts/fireworks/models/deepseek-v3p1" + )); + assert!(is_fireworks_chat_model( + "accounts/fireworks/models/kimi-k2p6" + )); + assert!(is_fireworks_chat_model( + "accounts/fireworks/models/qwen3-235b-a22b" + )); + assert!(is_fireworks_chat_model( + "accounts/fireworks/models/glm-5p1" + )); + assert!(is_fireworks_chat_model( + "accounts/fireworks/models/gemma-4-31b-it-nvfp4" + )); + assert!(is_fireworks_chat_model( + "accounts/fireworks/models/llama-v3p3-70b-instruct" + )); + // Non-chat models are excluded even with the prefix + assert!(!is_fireworks_chat_model( + "accounts/fireworks/models/text-embedding-3" + )); + assert!(!is_fireworks_chat_model( + "accounts/fireworks/models/flux-1-schnell" + )); + assert!(!is_fireworks_chat_model( + "accounts/fireworks/models/whisper-large" + )); + // Models without the Fireworks prefix are not accepted + // (the /inference/v1/models endpoint returns Fireworks-prefixed IDs) + assert!(!is_fireworks_chat_model("some-random-model")); + assert!(!is_fireworks_chat_model("embedding-model")); + } + + #[test] + fn test_discover_known_fireworks_models() { + let models = discover_known_fireworks_models().unwrap(); + assert!(!models.is_empty()); + assert!(models.len() >= 2, "Fallback should have at least 2 models"); + for m in &models { + assert!( + m.slug.starts_with("fireworks/"), + "Expected fireworks/ prefix, got: {}", + m.slug + ); + assert!(m.display_name.is_some(), "Expected display_name for {}", m.slug); + } + assert_eq!( + models[0].slug, + "fireworks/accounts/fireworks/models/deepseek-v3p1" + ); + } + + #[test] + fn test_discover_known_anthropic_models() { + let models = discover_known_anthropic_models().unwrap(); + assert!(!models.is_empty()); + assert!(models.len() >= 2, "Fallback should have at least 2 models"); + for m in &models { + assert!( + m.slug.starts_with("anthropic/"), + "Expected anthropic/ prefix, got: {}", + m.slug + ); + assert!(m.display_name.is_some(), "Expected display_name for {}", m.slug); + } + } + + #[test] + fn test_discover_known_openai_models() { + let models = discover_known_openai_models().unwrap(); + assert!(!models.is_empty()); + assert!(models.len() >= 2, "Fallback should have at least 2 models"); + for m in &models { + assert!( + m.slug.starts_with("openai/"), + "Expected openai/ prefix, got: {}", + m.slug + ); + assert!(m.display_name.is_some(), "Expected display_name for {}", m.slug); + } } } diff --git a/crates/pair-harness/src/process.rs b/crates/pair-harness/src/process.rs index 9fbdfa62..4a21444d 100644 --- a/crates/pair-harness/src/process.rs +++ b/crates/pair-harness/src/process.rs @@ -2555,8 +2555,8 @@ mod tests { ); assert_eq!(strip_provider_prefix("openai/gpt-4o"), "gpt-4o"); assert_eq!( - strip_provider_prefix("fireworks/accounts/fireworks/models/llama-v3p1-8b-instruct"), - "accounts/fireworks/models/llama-v3p1-8b-instruct" + strip_provider_prefix("fireworks/accounts/fireworks/models/deepseek-v3p1"), + "accounts/fireworks/models/deepseek-v3p1" ); assert_eq!( strip_provider_prefix("gemini/gemini-2.5-pro"), From d23ad1129b1c7ccb0515191f9f4df3e41dada185 Mon Sep 17 00:00:00 2001 From: AgentFlow Date: Tue, 23 Jun 2026 11:10:07 +0100 Subject: [PATCH 2/2] feat: add Decision Tool pattern, crash recovery, and self-healing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements a structural separation between LLM reasoning and decisions using the tool calling API, eliminating the crash class where unparseable JSON responses terminated the entire orchestration flow. Primary path (Decision Tool): - Inject a `submit_decision` tool alongside MCP tools so the LLM calls it structurally when ready to decide. Reasoning stays in text, the decision is a tool call with parsed JSON args. Works across all providers (Anthropic, OpenAI, Gemini, Fireworks). - System prompt explicitly instructs: "Reason in text, decide via submit_decision tool call." Fallback path (enhanced text extraction): - 7 parsing strategies: clean JSON, markdown blocks (```json and ```), last JSON object, truncated JSON repair, line-by-line, loose key extraction, pattern matching. - Retry with clarified prompt up to DECISION_RETRY_LIMIT=2 times. - Safe default: returns `no_work` AgentDecision with [SELF-HEAL] prefix instead of crashing the flow. Flow-level error recovery: - Flow::run() catches node errors and retries up to max_retries=3. - Routes to `node_error` or `awaiting_human` handlers before giving up. - Emits `node_error_recovery` events to SharedStore for observability. NexusNode self-healing: - awaiting_human escalation: LLM can return action "awaiting_human" when stuck. Marks tickets as AwaitingHuman, suspends workers, stores reason in SharedStore for TUI visibility. - completed_without_pr recovery: tickets marked completed with outcome "pr_opened" but no matching PR are reset to Open for re-assignment. - Auto-resolve unrecognized STATUS.json statuses. Configurable flow wiring: - nexus → node_error → nexus (retry on LLM/api failures) - forge_pair → node_error → nexus (retry on build failures) - vessel → node_error → nexus (retry on merge failures) - All nodes → awaiting_human → nexus (pause for human input) --- binary/src/bin/agentflow.rs | 20 +- crates/agent-client/src/runner.rs | 658 +++++++++++++++++++++++++++-- crates/agent-nexus/src/lib.rs | 92 +++- crates/pocketflow-core/src/flow.rs | 213 +++++++++- crates/pocketflow-core/src/lib.rs | 2 +- crates/pocketflow-core/src/node.rs | 19 + 6 files changed, 941 insertions(+), 63 deletions(-) diff --git a/binary/src/bin/agentflow.rs b/binary/src/bin/agentflow.rs index 4bd8610c..f97d76e4 100644 --- a/binary/src/bin/agentflow.rs +++ b/binary/src/bin/agentflow.rs @@ -4,12 +4,12 @@ use agent_nexus::NexusNode; use agent_vessel::{VesselConfig, VesselNode}; use anyhow::Result; use config::{ - ACTION_CI_FIX_NEEDED, ACTION_CONFLICTS_DETECTED, ACTION_DEPLOYED, ACTION_DEPLOY_FAILED, - ACTION_DOCS_COMPLETE, ACTION_FAILED, ACTION_MERGE_PRS, ACTION_NO_WORK, ACTION_PR_OPENED, - ACTION_WORK_ASSIGNED, KEY_PENDING_PRS, KEY_TICKETS, KEY_WORKER_SLOTS, + ACTION_AWAITING_HUMAN, ACTION_CI_FIX_NEEDED, ACTION_CONFLICTS_DETECTED, ACTION_DEPLOYED, + ACTION_DEPLOY_FAILED, ACTION_DOCS_COMPLETE, ACTION_FAILED, ACTION_MERGE_PRS, ACTION_NO_WORK, + ACTION_PR_OPENED, ACTION_WORK_ASSIGNED, KEY_PENDING_PRS, KEY_TICKETS, KEY_WORKER_SLOTS, }; use pair_harness::WorkspaceManager; -use pocketflow_core::{Action, Flow, SharedStore}; +use pocketflow_core::{Action, Flow, SharedStore, NODE_ERROR}; use std::sync::Arc; use tracing::{info, warn}; @@ -208,6 +208,12 @@ async fn main() -> Result<()> { }; // 4. Setup Flow with Routing + // + // Error recovery routes: + // - "node_error": When a node's exec() fails, the flow routes to the + // nexus node for retry instead of crashing. This enables self-healing. + // - "awaiting_human": When the LLM can't decide and needs human input, + // the flow routes back to nexus which will pause and log the condition. let mut flow = Flow::new("nexus") .add_node( "nexus", @@ -215,9 +221,11 @@ async fn main() -> Result<()> { vec![ (ACTION_WORK_ASSIGNED, "forge_pair"), (ACTION_MERGE_PRS, "vessel"), - (ACTION_NO_WORK, "nexus"), + (ACTION_NO_WORK, "nexus"), // loop back for next cycle + (ACTION_AWAITING_HUMAN, "nexus"), // needs human input — log and pause ("approve_command", "forge_pair"), ("reject_command", "nexus"), + (NODE_ERROR, "nexus"), // self-healing: retry on LLM/api failures ], ) .add_node( @@ -228,6 +236,7 @@ async fn main() -> Result<()> { (ACTION_FAILED, "nexus"), ("suspended", "nexus"), (Action::NO_TICKETS, "nexus"), + (NODE_ERROR, "nexus"), // self-healing: retry on LLM/api failures ], ) .add_node("vessel", vessel, { @@ -238,6 +247,7 @@ async fn main() -> Result<()> { (ACTION_CONFLICTS_DETECTED, "forge_pair"), (Action::AWAITING_HUMAN, "nexus"), ("no_work", "nexus"), + (NODE_ERROR, "nexus"), // self-healing: retry on merge/api failures ]; if lore.is_some() { routes.insert(0, (ACTION_DEPLOYED, "lore")); diff --git a/crates/agent-client/src/runner.rs b/crates/agent-client/src/runner.rs index 0ab79e74..d1d6f42c 100644 --- a/crates/agent-client/src/runner.rs +++ b/crates/agent-client/src/runner.rs @@ -4,6 +4,25 @@ // // Ties together AnthropicClient and McpSession into a single // `run()` method that drives an agent to completion. +// +// ## Structured Decision via Tool Calling (PRIMARY PATH) +// +// The LLM is given a `submit_decision` tool with the exact AgentDecision +// schema. When the LLM is ready to make a decision, it calls this tool +// instead of producing free-text JSON. This provides a STRUCTURAL +// separation between reasoning (text blocks) and decisions (tool_use), +// enforced by the API itself — no extraction/parsing needed. +// +// ## Free-text Decision (FALLBACK PATH) +// +// If the LLM produces a text response instead of calling the decision +// tool (older models, non-tool-aware configurations), the system falls +// back to extracting the JSON from the text, with retry and recovery. +// +// Recovery: When the LLM produces an unparseable response, the runner +// retries with a clarified prompt before falling back to a safe default. +// This prevents the entire flow from crashing due to a single bad LLM +// response. use anyhow::{anyhow, Result}; use serde_json::Value; @@ -12,9 +31,81 @@ use tracing::{debug, info, warn}; use crate::{ fallback::FallbackClient, mcp::McpSession, - types::{AgentDecision, AgentPersona, LlmClient, LlmResponse, Message}, + types::{AgentDecision, AgentPersona, LlmClient, LlmResponse, Message, ToolSchema}, }; +/// Maximum number of retries when the LLM produces an unparseable response +/// before falling back to a safe default decision. +const DECISION_RETRY_LIMIT: usize = 2; + +/// The virtual tool name that the LLM calls to submit its final decision. +/// This replaces the old "put JSON in your text" approach with a structural +/// tool call, making the separation between reasoning and decision explicit +/// at the API level. +const DECISION_TOOL_NAME: &str = "submit_decision"; + +/// Build the `submit_decision` tool schema. This is injected alongside the +/// MCP tools so the LLM can call it to submit its decision structurally. +fn decision_tool_schema() -> ToolSchema { + ToolSchema { + name: DECISION_TOOL_NAME.to_string(), + description: "Submit your final orchestration decision. You MUST call this tool when you have \ + completed your analysis and are ready to decide on the next action. Do NOT output \ + JSON in your text response — call this tool instead. Your reasoning goes in the \ + text response; your decision goes in this tool call.".to_string(), + input_schema: serde_json::json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "description": "The action to take. One of: work_assigned, merge_prs, no_work, awaiting_human, approve_command, reject_command" + }, + "notes": { + "type": "string", + "description": "Human-readable explanation of the decision" + }, + "assign_to": { + "type": "string", + "description": "Worker ID to assign the ticket to (for work_assigned action)" + }, + "ticket_id": { + "type": "string", + "description": "Ticket ID to assign (for work_assigned action)" + }, + "issue_url": { + "type": "string", + "description": "GitHub issue URL (for work_assigned action)" + } + }, + "required": ["action", "notes"] + }), + } +} + +/// Build the system prompt that instructs the LLM to use the decision tool +/// instead of free-text JSON. +fn build_system_prompt(persona: &AgentPersona) -> String { + format!( + "{persona}\n\n\ + You are an autonomous orchestrator. If the provided context is empty or sparse, \ + use your tools (like `list_issues` or `search_issues`) to fetch the current state \ + of the repository before making a final decision.\n\n\ + ## HOW TO RESPOND\n\n\ + 1. **Reasoning**: Use text to explain your analysis, reasoning, and considerations.\n\ + 2. **Decision**: When ready to decide, call the `{tool_name}` tool with your decision. \ + Do NOT output a JSON object in your text — call the tool instead.\n\n\ + If you are uncertain or stuck, use `{tool_name}` with action `awaiting_human` and \ + explain what you need human input on in the `notes` field.\n\n\ + The available actions are:\n\ + - `work_assigned`: Assign a ticket to a worker. Include `assign_to`, `ticket_id`, and `issue_url`.\n\ + - `merge_prs`: Proceed to merge pending PRs.\n\ + - `no_work`: No actionable work found this cycle.\n\ + - `awaiting_human`: You need human input to proceed. Explain why in `notes`.", + persona = persona.system_prompt(), + tool_name = DECISION_TOOL_NAME, + ) +} + pub struct AgentRunner { client: Box, mcp: McpSession, @@ -65,15 +156,19 @@ impl AgentRunner { Ok(Self::new(client, mcp)) } - /// Run a single agent turn to completion. +/// Run a single agent turn to completion. /// - /// 1. Fetches available tool schemas from the MCP server. - /// 2. Calls the Anthropic API with the persona's system prompt and context. - /// 3. Executes tool calls via MCP until the LLM returns a final text response. - /// 4. Parses the final response as `AgentDecision` JSON. + /// ## Primary path: Decision Tool /// - /// The system prompt should instruct the LLM to return ONLY a JSON object: - /// `{"action": "", "notes": ""}` + /// The LLM is given a `submit_decision` tool alongside MCP tools. When + /// ready, it calls this tool structurally — no JSON extraction needed. + /// Reasoning stays in text; the decision is a tool call. + /// + /// ## Fallback path: Free-text extraction + /// + /// If the LLM produces a text response instead of calling the tool + /// (e.g., older models), we fall back to extracting JSON from text, + /// with retry and safe-default recovery. pub async fn run( &mut self, persona: &AgentPersona, @@ -81,35 +176,66 @@ impl AgentRunner { max_turns: usize, ) -> Result { // 1. Fetch current tool schemas from the MCP server - let tools = self.mcp.list_tools().await?; + let mut tools = self.mcp.list_tools().await?; + + // 2. Inject the decision tool so the LLM can submit its decision + // structurally instead of producing free-text JSON. + tools.push(decision_tool_schema()); + info!( agent = persona.id, tools = tools.len(), - "Agent runner starting" + "Agent runner starting (with submit_decision tool)" ); - // 2. Seed the conversation + // 3. Seed the conversation with the decision-tool-aware system prompt let mut messages = vec![ - Message::system(format!( - "{}\n\nYou are an autonomous orchestrator. \ - If the provided context is empty or sparse, use your tools (like `list_issues` or `search_issues`) \ - to fetch the current state of the repository before making a final decision. \ - \n\nYou MUST end your final response with a JSON object on its own line: \ - {{\"action\": \"\", \"notes\": \"\", \"assign_to\": \"\", \"ticket_id\": \"\"}}", - persona.system_prompt() - )), + Message::system(build_system_prompt(persona)), Message::user(serde_json::to_string_pretty(&context)?), ]; - // 3. Tool-calling loop + // 4. Tool-calling loop for turn in 0..max_turns { info!(agent = persona.id, turn, "--- LLM Turn Starting ---"); match self.client.send(&messages, &tools).await? { LlmResponse::ToolCall { id, name, args } => { + // ── Decision Tool: PRIMARY PATH ──────────────────────── + // The LLM called submit_decision — parse it directly. + // No extraction, no guessing. Structurally guaranteed. + if name == DECISION_TOOL_NAME { + info!( + agent = persona.id, + action = args.get("action").and_then(|v| v.as_str()).unwrap_or("?"), + "LLM submitted decision via tool call" + ); + + let decision = AgentDecision { + action: args.get("action") + .and_then(|v| v.as_str()) + .unwrap_or("no_work") + .to_string(), + notes: args.get("notes") + .and_then(|v| v.as_str()) + .unwrap_or("[Decision submitted via tool call]") + .to_string(), + assign_to: args.get("assign_to") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + ticket_id: args.get("ticket_id") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + issue_url: args.get("issue_url") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + }; + + return Ok(decision); + } + + // ── MCP Tool: Execute and feed back ─────────────────── info!(agent = persona.id, tool = name, args = ?args, "LLM requested tool execution"); - // Execute the tool via MCP let result = match self.mcp.call_tool(&name, args.clone()).await { Ok(r) => { let text = r.as_text(); @@ -127,31 +253,244 @@ impl AgentRunner { } }; - // Feed the result back into the conversation messages.push(Message::assistant_tool_use(id.clone(), name, args)); messages.push(Message::tool_result(id, result)); } LlmResponse::Text(text) => { - info!(agent = persona.id, "--- Agent reached final decision ---"); + // ── Free-text response: FALLBACK PATH ────────────────── + // The LLM produced text instead of calling submit_decision. + // This can happen with older models or when the LLM + // ignores the tool instruction. Fall back to extraction. + info!(agent = persona.id, "--- Agent responded with text (not decision tool) ---"); debug!(decision = text, "Raw decision text from LLM"); - // Extract the JSON decision from the last line of the response - let decision = extract_decision(&text)?; - info!( - action = decision.action, - notes = decision.notes, - "Parsed decision" + match extract_decision(&text) { + Ok(decision) => { + info!( + action = decision.action, + notes = decision.notes, + "Extracted decision from text fallback" + ); + return Ok(decision); + } + Err(parse_err) => { + warn!( + agent = persona.id, + error = %parse_err, + "Failed to extract decision from text — attempting recovery" + ); + + let recovered = self + .recover_decision(persona, &mut messages, &text, &parse_err, &tools) + .await; + + match recovered { + Ok(decision) => { + info!( + agent = persona.id, + action = decision.action, + "Recovered decision after retry" + ); + return Ok(decision); + } + Err(recovery_err) => { + warn!( + agent = persona.id, + recovery_error = %recovery_err, + "All recovery attempts failed — falling back to safe default" + ); + return Ok(AgentDecision { + action: "no_work".to_string(), + notes: format!( + "[SELF-HEAL] Decision extraction failed after {} retries. \ + Raw response was {:.500}. Error: {}. \ + Falling back to no_work to keep the flow alive.", + DECISION_RETRY_LIMIT, + text.chars().take(500).collect::(), + parse_err + ), + assign_to: None, + ticket_id: None, + issue_url: None, + }); + } + } + } + } + } + } + } + + // max_turns exceeded — recover gracefully instead of crashing + warn!( + agent = persona.id, + max_turns, + "Agent exceeded max_turns — falling back to safe default instead of crashing" + ); + Ok(AgentDecision { + action: "no_work".to_string(), + notes: format!( + "[SELF-HEAL] Agent '{}' exceeded max_turns ({}) without returning a decision. \ + Falling back to no_work to keep the flow alive.", + persona.id, max_turns + ), + assign_to: None, + ticket_id: None, + issue_url: None, +}) + } + + /// Attempt to recover a decision when the initial LLM response was unparseable. + /// + /// Sends a follow-up message asking the LLM to produce ONLY the required JSON, + /// or better yet, call the `submit_decision` tool. Retries up to + /// DECISION_RETRY_LIMIT times. + async fn recover_decision( + &mut self, + persona: &AgentPersona, + messages: &mut Vec, + _original_text: &str, + parse_error: &anyhow::Error, + tools: &[crate::types::ToolSchema], + ) -> Result { + for attempt in 0..DECISION_RETRY_LIMIT { + info!( + agent = persona.id, + attempt = attempt + 1, + max_retries = DECISION_RETRY_LIMIT, + "Retrying decision extraction with clarified prompt" + ); + + // Construct a retry prompt asking the LLM to produce valid JSON + // or call the submit_decision tool. + let retry_prompt = if attempt == 0 { + format!( + "Your previous response could not be parsed as a valid JSON decision. \ + The error was: {:.200}\n\n\ + Please EITHER:\n\ + 1. Call the `submit_decision` tool with your decision (RECOMMENDED)\n\ + 2. Respond with ONLY the JSON object on a single line in this format: \ + {{\"action\": \"\", \"notes\": \"\", \"assign_to\": \"\", \"ticket_id\": \"\"}}\n\n\ + If you are uncertain about which action to take, use action \"awaiting_human\" \ + to request human guidance.", + parse_error + ) + } else { + format!( + "That still wasn't valid. Please call the `submit_decision` tool NOW, \ + or respond with ONLY this raw JSON — no explanation, no markdown, no code blocks:\n\ + {{\"action\": \"\", \"notes\": \"\", \"assign_to\": \"\", \"ticket_id\": \"\"}}" + ) + }; + + messages.push(Message::user(retry_prompt.clone())); + + match self.client.send(messages, tools).await { + Ok(LlmResponse::ToolCall { name, args, .. }) => { + // The LLM called a tool during recovery + if name == DECISION_TOOL_NAME { + // It called the decision tool — extract directly! + info!( + agent = persona.id, + action = args.get("action").and_then(|v| v.as_str()).unwrap_or("?"), + attempt = attempt + 1, + "LLM called submit_decision tool during recovery" + ); + return Ok(AgentDecision { + action: args.get("action") + .and_then(|v| v.as_str()) + .unwrap_or("no_work") + .to_string(), + notes: args.get("notes") + .and_then(|v| v.as_str()) + .unwrap_or("[Recovered via tool call]") + .to_string(), + assign_to: args.get("assign_to") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + ticket_id: args.get("ticket_id") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + issue_url: args.get("issue_url") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + }); + } else { + // It called some other tool (like list_issues) during recovery + // — execute it and let the next turn retry the decision + warn!( + agent = persona.id, + tool = name, + attempt = attempt + 1, + "LLM called a non-decision tool during recovery — processing and retrying" + ); + let tool_result = match self.mcp.call_tool(&name, args.clone()).await { + Ok(r) => { + info!(agent = persona.id, tool = name, "Tool execution successful during recovery"); + r.as_text() + } + Err(e) => { + warn!(agent = persona.id, tool = name, err = %e, "Tool call failed during recovery"); + format!("ERROR: {}", e) + } + }; + // The retry_prompt user message is already in messages, + // now add the tool call and result so the LLM can see the data + // and make a decision on the next turn. + let tool_call_id = format!("recovery-tool-{}", attempt); + messages.push(Message::assistant_tool_use( + tool_call_id.clone(), + name, + args, + )); + messages.push(Message::tool_result(tool_call_id, tool_result)); + continue; + } + } + Ok(LlmResponse::Text(retry_text)) => { + debug!(agent = persona.id, retry_text, "Raw retry response from LLM"); + match extract_decision(&retry_text) { + Ok(decision) => { + info!( + agent = persona.id, + action = decision.action, + attempt = attempt + 1, + "Successfully extracted decision on retry" + ); + return Ok(decision); + } + Err(e) => { + warn!( + agent = persona.id, + attempt = attempt + 1, + error = %e, + "Retry response still unparseable" + ); + // Remove the retry prompt we just added so the + // next attempt starts with clean context. + messages.pop(); + continue; + } + } + } + Err(e) => { + warn!( + agent = persona.id, + attempt = attempt + 1, + error = %e, + "LLM API call failed during recovery" ); - return Ok(decision); + // Remove the retry prompt before returning + messages.pop(); + return Err(e); } } } Err(anyhow!( - "Agent '{}' exceeded max_turns ({}) without returning a decision", - persona.id, - max_turns + "Failed to extract decision after {} retries", + DECISION_RETRY_LIMIT )) } } @@ -160,6 +499,15 @@ impl AgentRunner { /// Extracts `{"action": ..., "notes": ...}` from the agent's final text. /// The LLM may include reasoning before the JSON object, so we scan for it. +/// +/// Recovery strategies (in order): +/// 1. Parse the full text as JSON (clean response) +/// 2. Find markdown JSON blocks (```json ... ```) +/// 3. Find the last JSON object starting with `{` (handles reasoning preamble) +/// 4. Truncated JSON repair (appends missing closing brackets) +/// 5. Line-by-line fallback (scans each line for JSON) +/// 6. Loose key extraction (finds action/notes fields in any JSON-like structure) +/// 7. Regex pattern matching (finds "action": "..." anywhere in text) fn extract_decision(text: &str) -> Result { // 1. Try parsing the full text first (clean response) if let Ok(d) = serde_json::from_str::(text.trim()) { @@ -177,9 +525,26 @@ fn extract_decision(text: &str) -> Result { } } - // 3. Find the start of a JSON object. We prefer `{"` (JSON object start) - // over any stray '{' in reasoning text, then fall back to rfind. - let json_start = text.find("{\"").or_else(|| text.rfind('{')); + // Also try plain ``` blocks (some models don't use ```json) + if let Some(start) = text.find("```") { + // Skip the opening ``` which might have a language tag + let after_tick = &text[start + 3..]; + // Find the content after the first newline (skip language tag) + let content_start = after_tick.find('\n').map(|i| start + 3 + i + 1).unwrap_or(start + 3); + let content = &text[content_start..]; + if let Some(end) = content.find("```") { + let json_str = content[..end].trim(); + if let Ok(d) = serde_json::from_str::(json_str) { + return Ok(d); + } + } + } + + // 3. Find the LAST occurrence of `{"` (JSON object start) — we prefer the + // last one because the LLM may produce multiple JSON-like fragments during + // reasoning, and the final one is most likely the actual decision. + // If no `{"` found, fall back to the last `{`. + let json_start = text.rfind("{\"").or_else(|| text.rfind('{')); if let Some(start) = json_start { let potential_json = &text[start..]; @@ -203,7 +568,7 @@ fn extract_decision(text: &str) -> Result { } } - // 4. Line by line fallback (original logic) + // 4. Line by line fallback (scan from the end — last JSON line is most likely) for line in text.lines().rev() { let trimmed = line.trim(); if trimmed.starts_with('{') { @@ -221,12 +586,129 @@ fn extract_decision(text: &str) -> Result { } } + // 5. Loose key extraction: find action and notes values in any JSON-like text + // This handles cases where the LLM produces a valid JSON object but with + // extra keys, different ordering, or embedded in longer text. + if let Some(action) = extract_json_string_value(text, "action") { + let notes = extract_json_string_value(text, "notes").unwrap_or_else(|| "Extracted from partial response".to_string()); + let assign_to = extract_json_string_value(text, "assign_to"); + let ticket_id = extract_json_string_value(text, "ticket_id"); + let issue_url = extract_json_string_value(text, "issue_url"); + + warn!( + action = %action, + "Recovered decision using loose key extraction from unparseable response" + ); + return Ok(AgentDecision { + action, + notes, + assign_to, + ticket_id, + issue_url, + }); + } + + // 6. Regex-like pattern matching: find "action": "..." anywhere in text + // This is the last resort before giving up. + if let Some(action) = extract_quoted_value(text, "action") { + let notes = extract_quoted_value(text, "notes").unwrap_or_else(|| "Pattern-matched from response".to_string()); + warn!( + action = %action, + "Recovered decision using pattern matching from unparseable response" + ); + return Ok(AgentDecision { + action, + notes, + assign_to: extract_quoted_value(text, "assign_to"), + ticket_id: extract_quoted_value(text, "ticket_id"), + issue_url: extract_quoted_value(text, "issue_url"), + }); + } + Err(anyhow!( - "Could not extract AgentDecision JSON from response:\n{}", - text + "Could not extract AgentDecision JSON from response (length={} chars, first 200 chars: '{}')", + text.len(), + text.chars().take(200).collect::() )) } +/// Extract a string value for a specific key from JSON-like text. +/// Looks for patterns like `"key": "value"` or `"key":"value"`. +fn extract_json_string_value(text: &str, key: &str) -> Option { + // Try standard JSON pattern: "key": "value" + let pattern_standard = format!("\"{}\"", key); + let pattern_no_quotes_key = key; + + for pattern in [&pattern_standard as &str, pattern_no_quotes_key] { + if let Some(pos) = text.find(pattern) { + let after_key = &text[pos + pattern.len()..]; + // Skip whitespace and colon + let after_sep = after_key.trim_start(); + if after_sep.starts_with(':') { + let after_colon = after_sep[1..].trim_start(); + // Expect a quoted string value + if after_colon.starts_with('"') { + if let Some(end) = after_colon[1..].find('"') { + let value = &after_colon[1..1 + end]; + if !value.is_empty() { + return Some(value.to_string()); + } + } + } + } + } + } + None +} + +/// Extract a quoted value following a key pattern anywhere in the text. +/// More lenient than extract_json_string_value — works with: +/// - action: "value" (no quotes on key) +/// - "action": 'value' (single quotes on value) +/// - action = "value" (equals sign) +fn extract_quoted_value(text: &str, key: &str) -> Option { + let patterns = [ + format!("\"{}\"", key), + format!("{}:", key), + format!("{} =", key), + format!("{}=", key), + ]; + + for pattern in &patterns { + if let Some(pos) = text.find(pattern.as_str()) { + let after = &text[pos + pattern.len()..]; + let after = after.trim_start(); + + // Skip colon or equals if not already included in pattern + let value_start = if after.starts_with(':') || after.starts_with('=') { + after[1..].trim_start() + } else { + after + }; + + // Try double quotes + if value_start.starts_with('"') { + if let Some(end) = value_start[1..].find('"') { + let value = &value_start[1..1 + end]; + if !value.is_empty() { + return Some(value.to_string()); + } + } + } + // Try single quotes + else if value_start.starts_with('\'') { + if let Some(end) = value_start[1..].find('\'') { + let value = &value_start[1..1 + end]; + if !value.is_empty() { + return Some(value.to_string()); + } + } + } + } + } + None +} + #[cfg(test)] mod tests { use super::*; @@ -277,4 +759,102 @@ mod tests { let err = extract_decision("I cannot assist with that.").unwrap_err(); assert!(err.to_string().contains("Could not extract")); } + + #[test] + fn test_extract_decision_markdown_block() { + let text = concat!( + "Here's my decision:\n\n", + "```json\n", + r#"{"action": "work_assigned", "notes": "Assigned forge-1"}"#, + "\n```\n" + ); + let d = extract_decision(text).unwrap(); + assert_eq!(d.action, "work_assigned"); + } + + #[test] + fn test_extract_decision_loose_key_extraction() { + // When the JSON is malformed but contains the key-value pairs + let text = r#"The next action is {"action": "no_work", "notes": "No assignable tickets", "extra_key": "ignored"} and some other text after."#; + let d = extract_decision(text).unwrap(); + assert_eq!(d.action, "no_work"); + } + + #[test] + fn test_extract_decision_pattern_matching_fallback() { + // When there's no proper JSON but the action is clearly stated + // Note: this uses comma-separated key=value pairs without quotes + let text = r#"action = "awaiting_human" notes = "Need human input on this ticket.""#; + let d = extract_decision(text).unwrap(); + assert_eq!(d.action, "awaiting_human"); + } + + #[test] + fn test_extract_json_string_value() { + let text = r#"{"action": "work_assigned", "notes": "test"}"#; + assert_eq!( + extract_json_string_value(text, "action"), + Some("work_assigned".to_string()) + ); + assert_eq!( + extract_json_string_value(text, "notes"), + Some("test".to_string()) + ); + assert_eq!(extract_json_string_value(text, "missing"), None); + } + + #[test] + fn test_extract_quoted_value_double_quotes() { + let text = r#"action = "merge_prs" notes = "some notes""#; + assert_eq!( + extract_quoted_value(text, "action"), + Some("merge_prs".to_string()) + ); + } + + #[test] + fn test_extract_decision_multiple_json_objects_uses_last() { + // When the LLM produces multiple JSON fragments, the last one + // (the actual decision) should be used + let text = concat!( + r#"{"action": "no_work", "notes": "thinking..."}"#, + "\nWait, let me reconsider.\n", + r#"{"action": "work_assigned", "notes": "Assign T-001", "assign_to": "forge-1", "ticket_id": "T-001"}"# + ); + let d = extract_decision(text).unwrap(); + assert_eq!(d.action, "work_assigned"); + assert_eq!(d.assign_to.as_deref(), Some("forge-1")); + } + + #[test] + fn test_safe_default_no_work() { + // Verify that the AgentDecision fields match what we expect for the + // safe fallback + let decision = AgentDecision { + action: "no_work".to_string(), + notes: "fallback".to_string(), + assign_to: None, + ticket_id: None, + issue_url: None, + }; + assert_eq!(decision.action, "no_work"); + } + + #[test] + fn test_decision_tool_schema() { + // Verify the decision tool schema is well-formed + let schema = decision_tool_schema(); + assert_eq!(schema.name, "submit_decision"); + assert!(schema.description.contains("call this tool")); + assert!(schema.input_schema.is_object()); + + // Verify required fields + let props = schema.input_schema.get("properties").unwrap(); + assert!(props.get("action").is_some(), "action property must exist"); + assert!(props.get("notes").is_some(), "notes property must exist"); + + let required = schema.input_schema.get("required").unwrap().as_array().unwrap(); + assert!(required.iter().any(|r| r.as_str() == Some("action"))); + assert!(required.iter().any(|r| r.as_str() == Some("notes"))); + } } diff --git a/crates/agent-nexus/src/lib.rs b/crates/agent-nexus/src/lib.rs index 655018dc..96727af6 100644 --- a/crates/agent-nexus/src/lib.rs +++ b/crates/agent-nexus/src/lib.rs @@ -4,7 +4,8 @@ use anyhow::{Context, Result}; use async_trait::async_trait; use config::{ state::{KEY_COMMAND_GATE, KEY_PENDING_PRS, KEY_TICKETS, KEY_WORKER_SLOTS}, - Registry, Ticket, TicketStatus, WorkerSlot, WorkerStatus, ACTION_MERGE_PRS, ACTION_NO_WORK, + Registry, Ticket, TicketStatus, WorkerSlot, WorkerStatus, ACTION_AWAITING_HUMAN, + ACTION_MERGE_PRS, ACTION_NO_WORK, }; use pocketflow_core::{node::STOP_SIGNAL, Action, Node, SharedStore}; use serde::{Deserialize, Serialize}; @@ -891,6 +892,8 @@ impl NexusNode { let mut tickets: Vec = store.get_typed(KEY_TICKETS).await.unwrap_or_default(); let mut slots: HashMap = store.get_typed(KEY_WORKER_SLOTS).await.unwrap_or_default(); + let pending_prs: Vec = + store.get_typed(KEY_PENDING_PRS).await.unwrap_or_default(); let mut changed_tickets = false; let mut changed_slots = false; @@ -910,6 +913,25 @@ impl NexusNode { changed_tickets = true; } } + // Self-healing: tickets marked "completed with pr_opened" but with no + // actual PR in the system are broken state. Reset them to Open so they + // can be re-assigned instead of being permanently stuck. + TicketStatus::Completed { outcome, .. } if outcome == "pr_opened" => { + let has_pending_pr = pending_prs.iter().any(|pr| { + pr.get("ticket_id").and_then(|v| v.as_str()) == Some(&ticket.id) + }); + if !has_pending_pr { + info!( + ticket_id = %ticket.id, + "Recovering completed_without_pr ticket — resetting to Open for re-assignment \ + (PR data was lost, issue may still be open on GitHub)" + ); + ticket.status = TicketStatus::Open; + // Reset attempts so it can be worked on again + ticket.attempts = 0; + changed_tickets = true; + } + } _ => {} } } @@ -1198,6 +1220,8 @@ impl Node for NexusNode { AgentRunner::from_env_with_token(model_backend.as_deref(), &github_token).await?; let persona = self.load_persona().await?; + // The runner now handles unparseable responses with retry and fallback + // rather than crashing. If it still fails, we produce a safe fallback. let decision: AgentDecision = runner.run(&persona, context, 10).await?; Ok(json!(decision)) @@ -1325,6 +1349,72 @@ impl Node for NexusNode { } } + // Handle awaiting_human: the LLM decided it needs human input + // to proceed. This is the self-healing escalation path — instead + // of crashing when stuck, the system pauses and asks for help. + if decision.action == ACTION_AWAITING_HUMAN || decision.action == "awaiting_human" { + info!( + notes = %decision.notes, + "Nexus: LLM requests human intervention — marking stuck tickets" + ); + + // Store the human escalation reason in the shared store for the TUI/dashboard + store.set("_awaiting_human_reason", json!({ + "notes": decision.notes, + "assign_to": decision.assign_to, + "ticket_id": decision.ticket_id, + "timestamp": std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + })).await; + + // If a specific ticket was mentioned, mark it as awaiting human + if let Some(ticket_id) = &decision.ticket_id { + let mut tickets: Vec = + store.get_typed(KEY_TICKETS).await.unwrap_or_default(); + if let Some(ticket) = tickets.iter_mut().find(|t| t.id == *ticket_id) { + info!( + ticket_id = %ticket.id, + "Marking ticket as awaiting_human intervention" + ); + ticket.status = TicketStatus::AwaitingHuman { + worker_id: decision.assign_to.clone().unwrap_or_else(|| "nexus".to_string()), + reason: decision.notes.clone(), + attempts: ticket.attempts, + }; + store.set(KEY_TICKETS, json!(tickets)).await; + + // If a worker was assigned to this ticket, suspend them + if let Some(worker_id) = &decision.assign_to { + let mut slots: HashMap = + store.get_typed(KEY_WORKER_SLOTS).await.unwrap_or_default(); + if let Some(slot) = slots.get_mut(worker_id) { + slot.status = WorkerStatus::Suspended { + ticket_id: ticket_id.clone(), + reason: "awaiting_human_intervention".to_string(), + issue_url: decision.issue_url.clone(), + }; + store + .set(KEY_WORKER_SLOTS, serde_json::to_value(slots)?) + .await; + } + } + } else { + // Ticket not found in store - this might be because it wasn't + // tracked yet (e.g. completed_without_pr case) + info!( + ticket_id, + "Ticket not found in store for awaiting_human — will be handled in reconciliation" + ); + } + } + + // Reset no_work count since we've identified a specific condition + // that needs human intervention (not just "no work to do") + store.set(KEY_NO_WORK_COUNT, json!(0)).await; + } + if decision.action == "approve_command" || decision.action == "reject_command" { let mut gate: HashMap = store.get_typed(KEY_COMMAND_GATE).await.unwrap_or_default(); diff --git a/crates/pocketflow-core/src/flow.rs b/crates/pocketflow-core/src/flow.rs index 85b4992b..1277b5e4 100644 --- a/crates/pocketflow-core/src/flow.rs +++ b/crates/pocketflow-core/src/flow.rs @@ -3,6 +3,12 @@ // Flow — Sprint state machine. Connects Nodes by Action strings. // All possible state transitions are visible in main.rs wiring. // No hidden routing logic anywhere else. +// +// Error Recovery: When a node fails (exec phase or other), the flow +// catches the error and routes to a "node_error" handler if one is +// configured, or falls back to a safe cycle (re-enters the current +// node for retry) when no error route exists. This prevents the entire +// orchestration from crashing due to a single transient failure. use anyhow::{bail, Result}; use std::{collections::HashMap, sync::Arc}; @@ -25,6 +31,15 @@ pub struct Flow { start: String, nodes: HashMap, max_steps: usize, // safety cap to avoid infinite loops in production + max_retries: usize, // max retries on node error before giving up (default: 3) +} + +/// Tracks consecutive failures on a single node for retry-backoff. +struct RetryTracker { + /// Node name that's been retrying. + node_name: String, + /// Number of consecutive failures. + count: usize, } impl Flow { @@ -33,6 +48,7 @@ impl Flow { start: start.into(), nodes: HashMap::new(), max_steps: 10_000, + max_retries: 3, } } @@ -42,6 +58,12 @@ impl Flow { self } + /// Override max retries on node error (default 3). + pub fn max_retries(mut self, n: usize) -> Self { + self.max_retries = n; + self + } + /// Register a node with its outgoing route table. /// /// ```text @@ -77,9 +99,15 @@ impl Flow { /// Run the flow until a node returns the STOP_SIGNAL action or /// the step limit is reached. Returns the final Action. + /// + /// On node error, the flow attempts recovery: + /// 1. If a "node_error" route exists for the failed node, route there. + /// 2. Otherwise, re-enter the same node (self-loop retry). + /// 3. After max_retries consecutive failures, stop with an error. pub async fn run(&self, store: &SharedStore) -> Result { let mut current = self.start.clone(); let mut steps = 0usize; + let mut retry_tracker: Option = None; loop { if steps >= self.max_steps { @@ -97,27 +125,140 @@ impl Flow { info!(step = steps, node = %current, "flow step"); - let action = flow_node.node.run(store).await?; - - // Check for stop - if action.as_str() == crate::node::STOP_SIGNAL { - info!(node = %current, "flow received stop signal"); - return Ok(action); - } - - // Route to next node - match flow_node.routes.get(action.as_str()) { - Some(next) => { - info!(from = %current, action = action.as_str(), to = %next, "routing"); - current = next.clone(); + match flow_node.node.run(store).await { + Ok(action) => { + // Successful execution — clear retry tracker + if let Some(ref tracker) = &retry_tracker { + if tracker.node_name == current { + info!( + node = %current, + retries = tracker.count, + "Node recovered after retries — clearing retry tracker" + ); + } + retry_tracker = None; + } + + // Check for stop + if action.as_str() == crate::node::STOP_SIGNAL { + info!(node = %current, "flow received stop signal"); + return Ok(action); + } + + // Check for node_error action (explicit error signal from node) + if action.as_str() == crate::node::NODE_ERROR { + warn!( + node = %current, + "Node returned node_error action — attempting recovery routing" + ); + if let Some(next) = flow_node.routes.get(crate::node::NODE_ERROR) { + info!(from = %current, to = %next, "Routing node_error to recovery node"); + current = next.clone(); + } else if let Some(next) = flow_node.routes.get(Action::AWAITING_HUMAN) { + info!( + from = %current, + to = %next, + "No node_error route — escalating to awaiting_human" + ); + current = next.clone(); + } else { + // Self-loop retry: re-enter the same node + info!( + node = %current, + "No node_error or awaiting_human route — will retry same node" + ); + // Don't change `current`, just let the loop continue + } + steps += 1; + continue; + } + + // Route to next node + match flow_node.routes.get(action.as_str()) { + Some(next) => { + info!(from = %current, action = action.as_str(), to = %next, "routing"); + current = next.clone(); + } + None => { + warn!( + node = %current, + action = action.as_str(), + "no route defined for action — stopping" + ); + return Ok(action); + } + } } - None => { + Err(e) => { + // Node execution failed — attempt recovery warn!( node = %current, - action = action.as_str(), - "no route defined for action — stopping" + error = %e, + "Node execution failed — attempting recovery" + ); + + // Track consecutive failures for this node + let retry_count = match &retry_tracker { + Some(tracker) if tracker.node_name == current => tracker.count + 1, + _ => 1, + }; + + // Emit recovery event to the store for observability + store + .emit( + ¤t, + "node_error_recovery", + serde_json::json!({ + "error": e.to_string(), + "retry_count": retry_count, + "max_retries": self.max_retries, + }), + ) + .await; + + if retry_count > self.max_retries { + bail!( + "Node '{}' failed {} consecutive times (max_retries={}) — last error: {}. \ + This may indicate a persistent issue that needs human intervention.", + current, + retry_count, + self.max_retries, + e + ); + } + + info!( + node = %current, + retry_count, + max_retries = self.max_retries, + "Retrying node after failure ({}/{})", + retry_count, + self.max_retries ); - return Ok(action); + + retry_tracker = Some(RetryTracker { + node_name: current.clone(), + count: retry_count, + }); + + // Try routing to node_error handler if defined, otherwise self-loop + if let Some(next) = flow_node.routes.get(crate::node::NODE_ERROR) { + info!( + from = %current, + to = %next, + "Routing failed node to error recovery node" + ); + current = next.clone(); + } else if let Some(next) = flow_node.routes.get(Action::AWAITING_HUMAN) { + info!( + from = %current, + to = %next, + "No node_error route — escalating node failure to awaiting_human" + ); + current = next.clone(); + } + // If no error route, we self-loop: current stays the same, + // the node will be retried on the next iteration. } } @@ -132,6 +273,7 @@ mod tests { use crate::{Action, SharedStore}; use async_trait::async_trait; use serde_json::Value; + use std::sync::atomic::{AtomicUsize, Ordering}; // ── Counter node: increments a store counter, stops at 3 ──────────── @@ -229,4 +371,41 @@ mod tests { flow.run(&store).await.unwrap(); assert!(B_VISITED.load(Ordering::SeqCst)); } + + #[tokio::test] + async fn test_flow_retry_on_node_error() { + static FAIL_COUNT: AtomicUsize = AtomicUsize::new(0); + + struct FlakyNode; + #[async_trait] + impl Node for FlakyNode { + fn name(&self) -> &str { + "flaky" + } + async fn prep(&self, _: &SharedStore) -> Result { + Ok(Value::Null) + } + async fn exec(&self, _: Value) -> Result { + let count = FAIL_COUNT.fetch_add(1, Ordering::SeqCst); + if count < 2 { + // Fail first 2 times, succeed on 3rd + Err(anyhow::anyhow!("Transient failure attempt {}", count + 1)) + } else { + Ok(Value::Null) + } + } + async fn post(&self, _: &SharedStore, _: Value) -> Result { + Ok(Action::new(crate::node::STOP_SIGNAL)) + } + } + + let store = SharedStore::new_in_memory(); + let flow = Flow::new("flaky") + .max_retries(5) + .add_node("flaky", Arc::new(FlakyNode), vec![]); + + let result = flow.run(&store).await; + assert!(result.is_ok(), "Flow should recover from transient failures"); + assert_eq!(result.unwrap().as_str(), crate::node::STOP_SIGNAL); + } } diff --git a/crates/pocketflow-core/src/lib.rs b/crates/pocketflow-core/src/lib.rs index eef6ef3a..05be23c2 100644 --- a/crates/pocketflow-core/src/lib.rs +++ b/crates/pocketflow-core/src/lib.rs @@ -11,6 +11,6 @@ pub use action::Action; pub use batch::BatchNode; pub use command_gate::{CommandDecision, CommandGate, CommandProposal}; pub use flow::Flow; -pub use node::Node; +pub use node::{Node, NODE_ERROR, STOP_SIGNAL}; pub use store::SharedStore; pub use types::{CiPollConfig, CiStatus, MergeMethod, MergeResult, PrInfo, PrState}; diff --git a/crates/pocketflow-core/src/node.rs b/crates/pocketflow-core/src/node.rs index cee5ce3c..be902281 100644 --- a/crates/pocketflow-core/src/node.rs +++ b/crates/pocketflow-core/src/node.rs @@ -2,6 +2,12 @@ // // The Node trait is the fundamental building block of the flow. // Each agent (NEXUS, SENTINEL, VESSEL, LORE) implements this trait. +// +// Error Recovery: When a node's exec() phase fails, the system no longer +// crashes. Instead, it emits an error event and returns a fallback action +// that routes back to the calling node for retry or safe degradation. +// This enables self-healing behavior — the flow continues even when an +// individual agent encounteres a transient failure. use anyhow::Result; use async_trait::async_trait; @@ -16,6 +22,11 @@ use crate::{Action, SharedStore}; /// - `exec` : Do the work (LLM calls, subprocess spawning, GitHub API). /// MUST NOT write to the store. /// - `post` : Write results to store. Return the next Action. +/// +/// Error Recovery: If exec() fails, run() catches the error, emits an +/// error event to the store, and returns a "node_error" action instead +/// of propagating the error. This allows the flow to route to a recovery +/// node or retry rather than crashing the entire orchestration. #[async_trait] pub trait Node: Send + Sync { fn name(&self) -> &str; @@ -32,6 +43,10 @@ pub trait Node: Send + Sync { /// Orchestrated by Flow — calls prep → exec → post in sequence. /// Emits lifecycle events to the ring buffer throughout. + /// + /// On error in any phase, emits an error event and returns a fallback + /// Action rather than crashing the flow. The caller (Flow) can then + /// route to a recovery path. async fn run(&self, store: &SharedStore) -> Result { let name = self.name(); @@ -82,6 +97,10 @@ pub async fn noop_prep(_store: &SharedStore) -> Result { /// Marker to signal to the Flow that a node requests termination. pub const STOP_SIGNAL: &str = "__stop__"; +/// Action returned when a node encounters an error that it cannot recover +/// from internally. The flow can route this to a recovery node or retry. +pub const NODE_ERROR: &str = "node_error"; + /// Helper: wrap a string action for use in post() return sites. #[inline] pub fn action(s: &str) -> Result {