From 38f1dcdef6cf1ec51dffd2d441c1f921ddc8c3aa Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Fri, 6 Nov 2026 19:49:39 +0000 Subject: [PATCH] Harden provider diagnostics and tool failure handling --- AGENTS.md | 6 + app/src/ai/background_jobs/mod.rs | 7 + app/src/ai/background_jobs/tests.rs | 85 ++++++ .../ai/blocklist/action_model/execute/grep.rs | 36 ++- .../action_model/execute/grep_tests.rs | 42 +++ .../action_model/execute/read_files.rs | 185 ++++++++----- .../action_model/execute/read_files_tests.rs | 143 ++++++++++ app/src/ai/blocklist/controller.rs | 9 + app/src/ai/blocklist/controller_tests.rs | 12 +- app/src/ai/grep_query.rs | 42 +++ app/src/ai/grep_query_tests.rs | 39 +++ app/src/ai/mod.rs | 1 + app/src/ai/prompt_builder/tools.rs | 5 +- app/src/ai/provider/request_translator.rs | 5 +- app/src/ai/provider/response_translator.rs | 8 + .../ai/provider/response_translator_tests.rs | 33 +++ app/src/ai/remote_logging.rs | 114 +++++++- app/src/ai/remote_logging_tests.rs | 103 +++++++ app/src/ai/runtime/mod.rs | 2 + .../ai/runtime/model_failure_diagnostics.rs | 63 +++++ .../model_failure_diagnostics_tests.rs | 58 ++++ .../ai/runtime/provider_run_coordinator.rs | 17 ++ app/src/ai/runtime/rig_tool.rs | 6 +- app/src/ai/runtime/rig_tool_tests.rs | 71 +++++ crates/galaxy_agent_rig/Cargo.toml | 2 +- crates/galaxy_agent_rig/src/anthropic_auth.rs | 13 +- crates/galaxy_agent_rig/src/chatgpt.rs | 2 + crates/galaxy_agent_rig/src/http_client.rs | 224 ++++++++++++++++ .../galaxy_agent_rig/src/http_client_tests.rs | 253 ++++++++++++++++++ crates/galaxy_agent_rig/src/lib.rs | 1 + crates/galaxy_agent_rig/src/native.rs | 12 +- .../galaxy_agent_rig/src/openai_compatible.rs | 2 + crates/galaxy_agent_rig/src/stream.rs | 4 +- crates/galaxy_agent_rig/src/stream_tests.rs | 16 ++ .../galaxy_agent_rig/src/stream_unit_tests.rs | 16 ++ 35 files changed, 1558 insertions(+), 79 deletions(-) create mode 100644 app/src/ai/blocklist/action_model/execute/read_files_tests.rs create mode 100644 app/src/ai/grep_query.rs create mode 100644 app/src/ai/grep_query_tests.rs create mode 100644 app/src/ai/runtime/model_failure_diagnostics.rs create mode 100644 app/src/ai/runtime/model_failure_diagnostics_tests.rs create mode 100644 crates/galaxy_agent_rig/src/http_client.rs create mode 100644 crates/galaxy_agent_rig/src/http_client_tests.rs diff --git a/AGENTS.md b/AGENTS.md index fda133ea..91044010 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -94,6 +94,12 @@ zsh -lc ' conclusions about retries or repeated discovery. - Inspect `context.app.source_commit` and `source_modified` to identify the running build. Remote events may come from another machine or an older build. +- Terminal `provider_run_finished` events include `model_failure_diagnostics`: + the last eight sanitized model failures observed in the current process, with + an observed count and explicit truncation. A recovered run can complete with + retries; inspect this evidence even if individual retry logs were lost during + a connection interruption. This diagnostic history is never model context or + a persisted checkpoint and does not cover failures before a process restart. - `raw_model_request` events can include prompts, history, and tool data in `context.raw_payload.payload`. Check `truncated` and the accompanying size fields before parsing or treating missing content as evidence. These are Galaxy request diff --git a/app/src/ai/background_jobs/mod.rs b/app/src/ai/background_jobs/mod.rs index 3b08858d..5196dfd6 100644 --- a/app/src/ai/background_jobs/mod.rs +++ b/app/src/ai/background_jobs/mod.rs @@ -469,6 +469,13 @@ impl BackgroundJobsModel { id: BackgroundJobId, ctx: &mut ModelContext, ) -> Result<(), String> { + let job = self.jobs.get(&id).ok_or("Background job not found")?; + // Completion can win the race with an already-authorized cancellation. + // Its recorded outcome is final; acknowledge the request without + // signalling a process or publishing another completion. + if job.status.is_terminal() { + return Ok(()); + } if self.unreconciled_remote_jobs.contains(&id) && !self.remote_cancellations.contains_key(&id) { diff --git a/app/src/ai/background_jobs/tests.rs b/app/src/ai/background_jobs/tests.rs index 28a5b255..d86182e7 100644 --- a/app/src/ai/background_jobs/tests.rs +++ b/app/src/ai/background_jobs/tests.rs @@ -144,3 +144,88 @@ fn corrupt_history_is_reported_instead_of_replaced() { b"broken" ); } + +fn cancellation_fixture(status: BackgroundJobStatus) -> BackgroundJob { + BackgroundJob { + id: uuid::Uuid::new_v4(), + conversation_id: crate::ai::agent::conversation::AIConversationId::new(), + action_id: "cancellation-test-action".to_owned().into(), + command: "already completed command".to_owned(), + working_directory: std::path::PathBuf::new(), + project_root: None, + host_id: None, + started_at: Utc::now(), + finished_at: Some(Utc::now()), + status, + exit_code: Some(17), + error: Some("original outcome".to_owned()), + completion_delivered: true, + log_path: std::path::PathBuf::new(), + output: "original output".to_owned(), + output_truncated: false, + log_bytes: 15, + } +} + +#[test] +fn cancelling_a_completed_job_preserves_its_outcome_without_signalling_or_redelivery() { + galaxyui::App::test((), |mut app| async move { + let jobs = app.add_model(super::BackgroundJobsModel::new); + for status in [ + BackgroundJobStatus::Succeeded, + BackgroundJobStatus::Failed, + BackgroundJobStatus::Cancelled, + ] { + jobs.update(&mut app, |jobs, ctx| { + let job = cancellation_fixture(status); + let id = job.id; + let owner = job.conversation_id; + let finished_at = job.finished_at; + let (cancel, cancellations) = async_channel::bounded(1); + jobs.jobs.insert(id, job); + // Even a stale remote cancellation handle must not be used. + jobs.remote_cancellations.insert(id, cancel); + assert_eq!(jobs.cancel(id, ctx), Ok(())); + assert_eq!(jobs.cancel(id, ctx), Ok(())); + assert_eq!( + cancellations.try_recv(), + Err(async_channel::TryRecvError::Empty) + ); + let job = jobs.job(id).unwrap(); + assert_eq!(job.status, status); + assert_eq!(job.exit_code, Some(17)); + assert_eq!(job.finished_at, finished_at); + assert_eq!(job.error.as_deref(), Some("original outcome")); + assert_eq!(job.output, "original output"); + assert_eq!(job.completion_delivered, true); + assert_eq!(jobs.pending_completions(owner).len(), 0); + assert_eq!(jobs.unpublished_completions.contains(&id), false); + }); + } + }); +} + +#[test] +fn cancellation_still_rejects_unknown_and_unowned_running_processes() { + galaxyui::App::test((), |mut app| async move { + let jobs = app.add_model(super::BackgroundJobsModel::new); + jobs.update(&mut app, |jobs, ctx| { + let unknown = uuid::Uuid::new_v4(); + let (cancel, cancellations) = async_channel::bounded(1); + jobs.remote_cancellations.insert(unknown, cancel); + assert_eq!(jobs.cancel(unknown, ctx), Err("Background job not found".to_owned())); + assert_eq!(cancellations.try_recv(), Err(async_channel::TryRecvError::Empty)); + + let mut job = cancellation_fixture(BackgroundJobStatus::Running); + job.finished_at = None; + let id = job.id; + jobs.jobs.insert(id, job); + // A known Running record alone never authorizes signalling a PID. + assert_eq!(jobs.cancel(id, ctx), Err("This background job is no longer running".to_owned())); + + jobs.unreconciled_remote_jobs.insert(id); + assert_eq!(jobs.cancel(id, ctx), Err("Waiting to reconnect and confirm this job's state on its host before cancellation is available".to_owned())); + assert_eq!(jobs.job(id).unwrap().status, BackgroundJobStatus::Running); + }); + }); +} diff --git a/app/src/ai/blocklist/action_model/execute/grep.rs b/app/src/ai/blocklist/action_model/execute/grep.rs index 8e37b2e1..15602c48 100644 --- a/app/src/ai/blocklist/action_model/execute/grep.rs +++ b/app/src/ai/blocklist/action_model/execute/grep.rs @@ -93,7 +93,7 @@ impl GrepError { error: GrepErrorType::NonZeroExitCode, output: Some(output), .. - } => format!("{NON_ZERO_EXIT_CODE_ERROR}, output:\n{output}"), + } => actionable_grep_error(format!("{NON_ZERO_EXIT_CODE_ERROR}, output:\n{output}")), GrepError { error: GrepErrorType::NonZeroExitCode, output: None, @@ -102,11 +102,43 @@ impl GrepError { GrepError { error: GrepErrorType::Other(error), .. - } => error.clone(), + } => actionable_grep_error(error.clone()), } } } +/// Preserve backend regex semantics and report recovery guidance only after +/// that backend has diagnosed invalid syntax. Never retry with different +/// semantics or reject valid backend-specific expressions in a local parser. +fn actionable_grep_error(message: String) -> String { + let lower = message.to_ascii_lowercase(); + let invalid_regex = [ + "invalid regular expression", + "invalid pattern", + "regex parse error", + "unmatched [", + "unmatched (", + "unmatched )", + "unbalanced parenth", + "parentheses not balanced", + "brackets ([ ]) not balanced", + "unclosed group", + "unclosed character class", + "unterminated character class", + "unterminated [", + "missing closing parenthesis", + "not enough )", + "not enough ]", + ] + .iter() + .any(|marker| lower.contains(marker)); + if invalid_regex { + format!("{message}\nGrep uses regex by default. For literal source text, retry the same queries with fixed_strings: true. For an intended regex, correct or escape the invalid syntax; do not repeat the same malformed pattern.") + } else { + message + } +} + #[allow(clippy::too_many_arguments)] fn create_redacted_grep_error_event( should_collect_ugc: bool, diff --git a/app/src/ai/blocklist/action_model/execute/grep_tests.rs b/app/src/ai/blocklist/action_model/execute/grep_tests.rs index 2c96fcb6..3f79b472 100644 --- a/app/src/ai/blocklist/action_model/execute/grep_tests.rs +++ b/app/src/ai/blocklist/action_model/execute/grep_tests.rs @@ -125,3 +125,45 @@ fn parse_grep_output_reports_when_results_reach_the_per_file_limit() { ); assert!(truncated); } + +#[test] +fn backend_regex_syntax_errors_explain_literal_and_regex_recovery() { + for output in [ + "fatal: -e option, 'variantResolutions: []': brackets ([ ]) not balanced", + "fatal: -e option, 'indexAsset(context': parentheses not balanced", + "fatal: command line, 'variantResolutions: []': Unmatched [, [^, [:, [., or [=", + r"fatal: command line, 'indexAsset(context': Unmatched ( or \(", + "regex parse error: unclosed group", + "Select-String: Invalid pattern 'indexAsset(context': Not enough )'s.", + ] { + let error = GrepError::new_for_non_zero_exit_code().with_output(output.to_owned()); + let message = error.error_for_conversation(); + assert!(message.contains(output)); + assert!(message.contains("fixed_strings: true")); + assert!(message.contains("correct or escape")); + assert_eq!(error.error_message(), NON_ZERO_EXIT_CODE_ERROR); + } +} + +#[test] +fn ripgrep_regex_errors_also_include_literal_mode_guidance() { + let error = + GrepError::new("Ripgrep search failed: regex parse error: unclosed group".to_owned()); + assert!(error + .error_for_conversation() + .contains("fixed_strings: true")); +} + +#[test] +fn unrelated_grep_failures_keep_the_original_diagnostic() { + let output = "fatal: cannot access '/project': Permission denied"; + let error = GrepError::new_for_non_zero_exit_code().with_output(output.to_owned()); + assert_eq!( + error.error_for_conversation(), + format!("{NON_ZERO_EXIT_CODE_ERROR}, output:\n{output}") + ); + assert_eq!( + GrepError::new("Grep operation timed out".to_owned()).error_for_conversation(), + "Grep operation timed out" + ); +} diff --git a/app/src/ai/blocklist/action_model/execute/read_files.rs b/app/src/ai/blocklist/action_model/execute/read_files.rs index 4d6d0f26..5ceae9b6 100644 --- a/app/src/ai/blocklist/action_model/execute/read_files.rs +++ b/app/src/ai/blocklist/action_model/execute/read_files.rs @@ -173,57 +173,7 @@ impl ReadFilesExecutor { .await .map_err(|e| anyhow::anyhow!("Remote read failed: {e}"))?; - if !response.failed_files.is_empty() && response.file_contexts.is_empty() { - let failed = response - .failed_files - .iter() - .map(|f| { - let reason = f - .error - .as_ref() - .map(|e| e.message.as_str()) - .unwrap_or("unknown error"); - format!("{}: {reason}", f.path) - }) - .collect::>() - .join(", "); - return Ok(ReadFilesResult::Error(format!( - "Failed to read files: {failed}" - ))); - } - - let file_contexts = response - .file_contexts - .into_iter() - .filter_map(|fc| { - let content = match fc.content? { - remote_server::proto::file_context_proto::Content::TextContent( - text, - ) => crate::ai::agent::AnyFileContent::StringContent(text), - remote_server::proto::file_context_proto::Content::BinaryContent( - bytes, - ) => crate::ai::agent::AnyFileContent::BinaryContent(bytes), - }; - let line_range = match (fc.line_range_start, fc.line_range_end) { - (Some(start), Some(end)) => Some(start as usize..end as usize), - _ => None, - }; - let last_modified = fc.last_modified_epoch_millis.map(|ms| { - std::time::UNIX_EPOCH + std::time::Duration::from_millis(ms) - }); - Some(crate::ai::agent::FileContext { - file_name: fc.file_name, - content, - line_range, - last_modified, - line_count: fc.line_count as usize, - }) - }) - .collect(); - - Ok(ReadFilesResult::Success { - files: file_contexts, - }) + Ok(remote_read_files_result(response)) }), on_complete: Box::new(|res: Result, _ctx| { let action_result = @@ -254,16 +204,7 @@ impl ReadFilesExecutor { result.file_contexts.extend(file_result.file_contexts); result.missing_files.extend(file_result.missing_files); } - if result.missing_files.is_empty() { - Ok(ReadFilesResult::Success { - files: result.file_contexts, - }) - } else { - let missing_files = result.missing_files.join(", "); - Ok(ReadFilesResult::Error(format!( - "These files do not exist: {missing_files}" - ))) - } + Ok(local_read_files_result(result)) }), on_complete: Box::new(|res: Result, _ctx| { let action_result = res.unwrap_or_else(|e| ReadFilesResult::Error(e.to_string())); @@ -284,3 +225,125 @@ impl ReadFilesExecutor { impl Entity for ReadFilesExecutor { type Event = (); } + +/// The shared action/protobuf contract has no partial-success representation. +/// Do not claim a remote batch succeeded while silently losing failed paths. +fn remote_read_files_result( + response: remote_server::proto::ReadFileContextResponse, +) -> ReadFilesResult { + if !response.failed_files.is_empty() + || response + .file_contexts + .iter() + .any(|file| file.content.is_none()) + { + let failed = response.failed_files.iter().map(|failure| { + ( + failure.path.as_str(), + failure.error.as_ref().map(|error| error.message.as_str()), + ) + }); + let missing_content = response + .file_contexts + .iter() + .filter(|file| file.content.is_none()) + .map(|file| { + let name = if file.file_name.is_empty() { + "" + } else { + file.file_name.as_str() + }; + (name, Some("remote host returned no file content")) + }); + return read_files_failure(failed.chain(missing_content)); + } + let file_contexts = response + .file_contexts + .into_iter() + .filter_map(|fc| { + let content = match fc.content? { + remote_server::proto::file_context_proto::Content::TextContent(text) => { + crate::ai::agent::AnyFileContent::StringContent(text) + } + remote_server::proto::file_context_proto::Content::BinaryContent(bytes) => { + crate::ai::agent::AnyFileContent::BinaryContent(bytes) + } + }; + let line_range = match (fc.line_range_start, fc.line_range_end) { + (Some(start), Some(end)) => Some(start as usize..end as usize), + _ => None, + }; + let last_modified = fc + .last_modified_epoch_millis + .map(|ms| std::time::UNIX_EPOCH + std::time::Duration::from_millis(ms)); + Some(crate::ai::agent::FileContext { + file_name: fc.file_name, + content, + line_range, + last_modified, + line_count: fc.line_count as usize, + }) + }) + .collect(); + ReadFilesResult::Success { + files: file_contexts, + } +} + +fn local_read_files_result(result: super::ReadFileContextResult) -> ReadFilesResult { + if result.missing_files.is_empty() { + ReadFilesResult::Success { + files: result.file_contexts, + } + } else { + read_files_failure( + result + .missing_files + .iter() + .map(|path| (path.as_str(), None)), + ) + } +} + +const MAX_REPORTED_READ_FAILURES: usize = 8; +const MAX_READ_FAILURE_FIELD_CHARS: usize = 256; + +fn read_files_failure<'a>( + failures: impl IntoIterator)>, +) -> ReadFilesResult { + let mut message = "Could not read all requested files. No file contents were returned for this batch.\nFailed paths:".to_owned(); + let mut total = 0; + for (path, reason) in failures { + total += 1; + if total > MAX_REPORTED_READ_FAILURES { + continue; + } + message.push_str("\n- "); + append_failure_field(&mut message, path); + message.push_str(": "); + append_failure_field( + &mut message, + reason.unwrap_or("missing, unreadable, or exceeds the read limit"), + ); + } + if total > MAX_REPORTED_READ_FAILURES { + message.push_str(&format!( + "\n{} additional failed paths omitted.", + total - MAX_REPORTED_READ_FAILURES + )); + } + message.push_str("\nUse file discovery to confirm exact paths before retrying. For large files, request a smaller line range or fewer files. A file may exist but be unreadable or exceed the read limit."); + ReadFilesResult::Error(message) +} + +fn append_failure_field(message: &mut String, field: &str) { + let mut characters = field.chars(); + message.extend(characters.by_ref().take(MAX_READ_FAILURE_FIELD_CHARS)); + if characters.next().is_some() { + message.push('…'); + } +} + +#[cfg(test)] +#[path = "read_files_tests.rs"] +mod tests; diff --git a/app/src/ai/blocklist/action_model/execute/read_files_tests.rs b/app/src/ai/blocklist/action_model/execute/read_files_tests.rs new file mode 100644 index 00000000..0ac89962 --- /dev/null +++ b/app/src/ai/blocklist/action_model/execute/read_files_tests.rs @@ -0,0 +1,143 @@ +use super::*; +use crate::ai::agent::{AnyFileContent, FileContext}; + +fn remote_file() -> remote_server::proto::FileContextProto { + remote_server::proto::FileContextProto { + file_name: "/project/present.rs".to_owned(), + content: Some( + remote_server::proto::file_context_proto::Content::TextContent( + "returned file content".to_owned(), + ), + ), + line_range_start: Some(3), + line_range_end: Some(5), + last_modified_epoch_millis: Some(42), + line_count: 20, + } +} + +#[test] +fn partial_remote_reads_report_failed_paths_instead_of_claiming_success() { + let result = remote_read_files_result(remote_server::proto::ReadFileContextResponse { + file_contexts: vec![remote_file()], + failed_files: vec![remote_server::proto::FailedFileRead { + path: "/project/restricted.rs".to_owned(), + error: Some(remote_server::proto::FileOperationError { + message: "Permission denied".to_owned(), + }), + }], + }); + let ReadFilesResult::Error(message) = result else { + panic!("A partial remote read must not claim full success"); + }; + assert!(message.contains("/project/restricted.rs: Permission denied")); + assert!(message.contains("No file contents were returned for this batch")); + assert!(!message.contains("returned file content")); + assert!(message.contains("confirm exact paths")); +} + +#[test] +fn remote_file_context_without_content_is_an_explicit_failure() { + let mut missing = remote_file(); + missing.file_name = "/project/no-content.rs".to_owned(); + missing.content = None; + let result = remote_read_files_result(remote_server::proto::ReadFileContextResponse { + file_contexts: vec![remote_file(), missing.clone()], + failed_files: vec![], + }); + let ReadFilesResult::Error(message) = result else { + panic!("A remote context without content must not disappear into Success"); + }; + assert!(message.contains("/project/no-content.rs: remote host returned no file content")); + assert!(!message.contains("returned file content")); + + let result = remote_read_files_result(remote_server::proto::ReadFileContextResponse { + file_contexts: vec![missing], + failed_files: vec![remote_server::proto::FailedFileRead { + path: "/project/restricted.rs".to_owned(), + error: Some(remote_server::proto::FileOperationError { + message: "Permission denied".to_owned(), + }), + }], + }); + let ReadFilesResult::Error(message) = result else { + panic!("Both remote failure categories must remain visible"); + }; + assert!(message.contains("/project/no-content.rs: remote host returned no file content")); + assert!(message.contains("/project/restricted.rs: Permission denied")); +} + +#[test] +fn successful_remote_reads_preserve_content_ranges_and_file_metadata() { + let result = remote_read_files_result(remote_server::proto::ReadFileContextResponse { + file_contexts: vec![remote_file()], + failed_files: vec![], + }); + let ReadFilesResult::Success { files } = result else { + panic!("A complete remote read must succeed"); + }; + assert_eq!(files.len(), 1); + assert_eq!(files[0].file_name, "/project/present.rs"); + assert_eq!( + files[0].content, + AnyFileContent::StringContent("returned file content".to_owned()) + ); + assert_eq!(files[0].line_range, Some(3..5)); + assert_eq!(files[0].line_count, 20); + assert_eq!( + files[0].last_modified, + Some(std::time::UNIX_EPOCH + std::time::Duration::from_millis(42)) + ); +} + +#[test] +fn local_failures_do_not_assert_that_oversized_or_unreadable_files_are_missing() { + let result = local_read_files_result(super::super::ReadFileContextResult { + file_contexts: vec![], + missing_files: vec!["/project/large.bin".to_owned()], + }); + let ReadFilesResult::Error(message) = result else { + panic!("An unread file must remain an error"); + }; + assert!(message.contains("/project/large.bin")); + assert!(message.contains("exceeds the read limit")); + assert!(message.contains("smaller line range or fewer files")); + assert!(!message.contains("do not exist")); +} + +#[test] +fn successful_local_reads_preserve_their_original_content() { + let file = FileContext { + file_name: "/project/present.rs".to_owned(), + content: AnyFileContent::StringContent("unchanged".to_owned()), + line_range: None, + last_modified: None, + line_count: 1, + }; + assert_eq!( + local_read_files_result(super::super::ReadFileContextResult { + file_contexts: vec![file.clone()], + missing_files: vec![], + }), + ReadFilesResult::Success { files: vec![file] }, + ); +} + +#[test] +fn failure_details_bound_unicode_paths_reasons_and_the_number_of_reported_files() { + let long_field = "界".repeat(10_000); + let result = read_files_failure( + (0..MAX_REPORTED_READ_FAILURES + 3) + .map(|_| (long_field.as_str(), Some(long_field.as_str()))), + ); + let ReadFilesResult::Error(message) = result else { + panic!("Failed paths must produce an error"); + }; + assert_eq!(message.matches("\n- ").count(), MAX_REPORTED_READ_FAILURES); + assert!(message.contains("3 additional failed paths omitted")); + assert!(message.contains('…')); + assert!( + message.chars().count() + < MAX_REPORTED_READ_FAILURES * MAX_READ_FAILURE_FIELD_CHARS * 2 + 1_000 + ); +} diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index 0b260f20..f3c2c4a8 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -2258,7 +2258,12 @@ fn provider_run_terminal_remote_log_record( stream_id: &ResponseStreamId, run: &ProviderRun, outcome: &ProviderRunOutcome, + model_failure_diagnostics: serde_json::Value, ) -> RemoteLogRecord { + let source_error = match outcome { + ProviderRunOutcome::Failed(failure) => failure.source.as_ref(), + ProviderRunOutcome::Completed(_) | ProviderRunOutcome::Cancelled { .. } => None, + }; let (level, outcome_name, llm_finished, stop_reason, failure_kind, error) = match outcome { ProviderRunOutcome::Completed(completion) => ( RemoteLogLevel::Info, @@ -2297,10 +2302,13 @@ fn provider_run_terminal_remote_log_record( "profile": run.profile().as_str(), "model_turn_count": run.model_turns(), "model_retry_count": run.model_retries(), + "model_failure_diagnostics": model_failure_diagnostics, "outcome": outcome_name, "stop_reason": stop_reason, "failure_kind": failure_kind, "error": error, + "error_kind": source_error.map(|error| format!("{:?}", error.kind)), + "error_recoverable": source_error.map(|error| error.recoverable), "llm_finished": llm_finished, "provider_run_finished": true, "response_stream_terminal": true, @@ -9519,6 +9527,7 @@ impl BlocklistAIController { &stream_id, run.coordinator.run(), &outcome, + run.coordinator.model_failure_diagnostics(), ), ); let completed_cleanly = matches!( diff --git a/app/src/ai/blocklist/controller_tests.rs b/app/src/ai/blocklist/controller_tests.rs index f8289a55..af97c000 100644 --- a/app/src/ai/blocklist/controller_tests.rs +++ b/app/src/ai/blocklist/controller_tests.rs @@ -156,6 +156,7 @@ fn provider_terminal_logs_distinguish_clean_completion_from_failure() { &ProviderRunOutcome::Completed(galaxy_agent_core::ProviderRunCompletion { stop_reason: StopReason::Completed, }), + serde_json::json!({"observed_count": 0, "recent": []}), ); let failed = super::provider_run_terminal_remote_log_record( conversation_id, @@ -164,8 +165,14 @@ fn provider_terminal_logs_distinguish_clean_completion_from_failure() { &ProviderRunOutcome::Failed(ProviderRunFailure { kind: ProviderRunFailureKind::RetryLimitExceeded, message: "sk-secret timeout".to_owned(), - source: None, + source: Some(galaxy_agent_core::AgentError { + kind: galaxy_agent_core::AgentErrorKind::Transport, + message: "connection failed".to_owned(), + user_message: None, + recoverable: true, + }), }), + serde_json::json!({"observed_count": 1, "recent": [{"error": "connection failed"}]}), ); assert_eq!(completed.context["llm_finished"], true); @@ -176,6 +183,9 @@ fn provider_terminal_logs_distinguish_clean_completion_from_failure() { assert_eq!(failed.context["llm_finished"], false); assert_eq!(failed.context["outcome"], "failed"); assert_eq!(failed.context["failure_kind"], "RetryLimitExceeded"); + assert_eq!(failed.context["error_kind"], "Transport"); + assert_eq!(failed.context["error_recoverable"], true); + assert_eq!(failed.context["model_failure_diagnostics"]["observed_count"], 1); assert_eq!(failed.context["error"], "[redacted] timeout"); } diff --git a/app/src/ai/grep_query.rs b/app/src/ai/grep_query.rs new file mode 100644 index 00000000..b14c4d13 --- /dev/null +++ b/app/src/ai/grep_query.rs @@ -0,0 +1,42 @@ +//! Match-mode preparation shared by native and compatibility grep tool calls. + +/// Keep regex mode unchanged; literal mode escapes only metacharacters shared +/// by git/grep extended regex, ripgrep, and PowerShell Select-String. This keeps +/// the persisted action and backend execution paths identical in both modes. +pub(crate) fn prepare_grep_queries(queries: Vec, fixed_strings: bool) -> Vec { + if !fixed_strings { + return queries; + } + queries + .into_iter() + .map(|query| { + let mut escaped = String::with_capacity(query.len()); + for character in query.chars() { + if matches!( + character, + '\\' | '.' + | '^' + | '$' + | '|' + | '?' + | '*' + | '+' + | '(' + | ')' + | '[' + | ']' + | '{' + | '}' + ) { + escaped.push('\\'); + } + escaped.push(character); + } + escaped + }) + .collect() +} + +#[cfg(test)] +#[path = "grep_query_tests.rs"] +mod tests; diff --git a/app/src/ai/grep_query_tests.rs b/app/src/ai/grep_query_tests.rs new file mode 100644 index 00000000..b306d91d --- /dev/null +++ b/app/src/ai/grep_query_tests.rs @@ -0,0 +1,39 @@ +use super::prepare_grep_queries; + +#[test] +fn literal_queries_match_source_punctuation_backslashes_and_unicode_exactly() { + let cases = [ + ("variantResolutions: []", r"variantResolutions: \[\]"), + ("indexAsset(context", r"indexAsset\(context"), + ( + r#"describe("duplicate"|managed output|duplicateContentAsset"#, + r#"describe\("duplicate"\|managed output\|duplicateContentAsset"#, + ), + (r"C:\repo\file.rs", r"C:\\repo\\file\.rs"), + ("🚀 café [数据] (名)", r"🚀 café \[数据\] \(名\)"), + ( + r#"owner's $HOME `id` #&-~ name=foo: bar"#, + r#"owner's \$HOME `id` #&-~ name=foo: bar"#, + ), + ( + "^foo.*(bar|baz)?[0-9]+{2}$", + r"\^foo\.\*\(bar\|baz\)\?\[0-9\]\+\{2\}\$", + ), + ]; + for (literal, escaped) in cases { + let queries = prepare_grep_queries(vec![literal.to_string()], true); + assert_eq!(queries, vec![escaped], "literal: {literal}"); + let regex = regex::Regex::new(&queries[0]).unwrap(); + assert_eq!(regex.find(literal).unwrap().as_str(), literal); + } +} + +#[test] +fn regex_queries_stay_byte_for_byte_unchanged_including_backend_specific_syntax() { + let queries = vec![ + r"^(foo|bar)\s+[0-9]+$".to_string(), + r"(?<=prefix)lookbehind".to_string(), + "[invalid".to_string(), + ]; + assert_eq!(prepare_grep_queries(queries.clone(), false), queries); +} diff --git a/app/src/ai/mod.rs b/app/src/ai/mod.rs index 0df3a0df..aa33858b 100644 --- a/app/src/ai/mod.rs +++ b/app/src/ai/mod.rs @@ -81,6 +81,7 @@ pub mod connected_self_hosted_workers; pub mod execution_profiles; pub mod facts; pub(crate) mod generate_block_title; +pub(crate) mod grep_query; pub(crate) mod generate_code_review_content; pub(crate) mod loading; pub mod mcp; diff --git a/app/src/ai/prompt_builder/tools.rs b/app/src/ai/prompt_builder/tools.rs index dc067fc3..01c2ad01 100644 --- a/app/src/ai/prompt_builder/tools.rs +++ b/app/src/ai/prompt_builder/tools.rs @@ -136,11 +136,12 @@ fn apply_file_diffs() -> ToolDefinition { fn grep() -> ToolDefinition { ToolDefinition { name: "grep".to_string(), - description: "Search for up to 3 focused regex patterns per call. Use for exact symbols, strings, imports, and call sites in source code or other text files. Scope to known files or directories when possible; use search_codebase for semantic discovery. Uses git grep in git repos (respects .gitignore) or ripgrep otherwise. Returns up to 1,000 matching line locations and reports when additional matches were omitted. Use read_files afterward to see context around matches.".to_string(), + description: "Search for up to 3 focused patterns per call. Regex is the default; set fixed_strings=true for literal source text containing brackets, parentheses, or other regex metacharacters. Use for exact symbols, strings, imports, and call sites in source code or other text files. Scope to known files or directories when possible; use search_codebase for semantic discovery. Uses git grep in git repos (respects .gitignore) or ripgrep otherwise. Returns up to 1,000 matching line locations and reports when additional matches were omitted. Use read_files afterward to see context around matches.".to_string(), input_schema: serde_json::json!({ "type": "object", "properties": { - "queries": { "type": "array", "maxItems": 3, "items": { "type": "string" }, "description": "Regex patterns to search for (maximum 3 per call)" }, + "queries": { "type": "array", "maxItems": 3, "items": { "type": "string" }, "description": "Patterns (literal when fixed_strings=true) to search for (maximum 3 per call)" }, + "fixed_strings": { "type": "boolean", "default": false, "description": "Treat every query as literal text instead of regex" }, "path": { "type": "string", "description": "Directory to scope the search to" } }, "required": ["queries"] diff --git a/app/src/ai/provider/request_translator.rs b/app/src/ai/provider/request_translator.rs index cbce6660..303a3b0f 100644 --- a/app/src/ai/provider/request_translator.rs +++ b/app/src/ai/provider/request_translator.rs @@ -1868,11 +1868,12 @@ pub fn default_tool_definitions() -> Vec { }, ToolDefinition { name: "grep".to_string(), - description: "Search for regex patterns in files. Use for exact symbols, strings, imports, and call sites in source code or other text files. Scope to known files or directories when possible; use search_codebase for semantic discovery. Uses git grep in git repos (respects .gitignore) or ripgrep otherwise. Returns up to 1,000 matching line locations and reports when additional matches were omitted. Use read_files afterward to see context around matches. Pass ALL patterns you need in one call.".to_string(), + description: "Search for patterns in files. Regex is the default; set fixed_strings=true for literal source text containing brackets, parentheses, or other regex metacharacters. Use for exact symbols, strings, imports, and call sites in source code or other text files. Scope to known files or directories when possible; use search_codebase for semantic discovery. Uses git grep in git repos (respects .gitignore) or ripgrep otherwise. Returns up to 1,000 matching line locations and reports when additional matches were omitted. Use read_files afterward to see context around matches. Pass ALL patterns you need in one call.".to_string(), input_schema: serde_json::json!({ "type": "object", "properties": { - "queries": { "type": "array", "items": { "type": "string" }, "description": "Regex patterns to search for" }, + "queries": { "type": "array", "items": { "type": "string" }, "description": "Patterns (literal when fixed_strings=true) to search for" }, + "fixed_strings": { "type": "boolean", "default": false, "description": "Treat every query as literal text instead of regex" }, "path": { "type": "string", "description": "Directory to scope the search to" } }, "required": ["queries"] diff --git a/app/src/ai/provider/response_translator.rs b/app/src/ai/provider/response_translator.rs index 00aa1d8b..30d20b55 100644 --- a/app/src/ai/provider/response_translator.rs +++ b/app/src/ai/provider/response_translator.rs @@ -16,6 +16,7 @@ use warp_multi_agent_api::{self as api, ClientAction, ResponseEvent}; use super::convert::{ContentPart, ConversationMessage, MessageContent, MessageRole}; use super::diagnostic::BedrockDiagnosticLogger; use crate::ai::agent::api::LegacyEvent; +use crate::ai::grep_query::prepare_grep_queries; use crate::server::server_api::AIApiError; fn json_to_prost_struct(value: &serde_json::Value) -> prost_types::Struct { @@ -1018,6 +1019,13 @@ pub fn build_tool_call_message( .collect() }) .unwrap_or_default(); + let queries = prepare_grep_queries( + queries, + input + .get("fixed_strings") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false), + ); let path = input .get("path") .and_then(|v| v.as_str()) diff --git a/app/src/ai/provider/response_translator_tests.rs b/app/src/ai/provider/response_translator_tests.rs index a8308402..035c334a 100644 --- a/app/src/ai/provider/response_translator_tests.rs +++ b/app/src/ai/provider/response_translator_tests.rs @@ -522,3 +522,36 @@ fn direct_provider_known_tools_exclude_hosted_only_tools() { assert!(is_known_tool("recall_tool_history")); assert!(is_known_tool("interrupt_shell_command")); } + + +#[test] +fn compatibility_grep_literal_mode_uses_the_same_escaped_patterns() { + let tool = tool_from_event(build_tool_call_message( + "task-1", + "grep-1", + "grep", + r#"{"queries":["variantResolutions: []","indexAsset(context"],"fixed_strings":true,"path":"/project/src"}"#, + )); + let api::message::tool_call::Tool::Grep(grep) = tool else { + panic!("expected grep tool"); + }; + assert_eq!( + grep.queries, + vec![r"variantResolutions: \[\]", r"indexAsset\(context"] + ); + assert_eq!(grep.path, "/project/src"); +} + +#[test] +fn compatibility_grep_keeps_default_regex_semantics() { + let tool = tool_from_event(build_tool_call_message( + "task-1", + "grep-1", + "grep", + r#"{"queries":["foo|bar","[unclosed"]}"#, + )); + let api::message::tool_call::Tool::Grep(grep) = tool else { + panic!("expected grep tool"); + }; + assert_eq!(grep.queries, vec!["foo|bar", "[unclosed"]); +} diff --git a/app/src/ai/remote_logging.rs b/app/src/ai/remote_logging.rs index d37a2882..d424458c 100644 --- a/app/src/ai/remote_logging.rs +++ b/app/src/ai/remote_logging.rs @@ -17,6 +17,7 @@ use std::{ use chrono::Utc; use galaxy_core::channel::ChannelState; use galaxyui::{Entity, ModelContext, SingletonEntity, View, ViewContext}; +use regex::Regex; use serde::Serialize; use serde_json::{json, Value}; use settings::Setting; @@ -34,6 +35,15 @@ static REMOTE_LOG_CLIENT: LazyLock> = LazyLock:: .map_err(|error| format!("could not create HTTP client: {error}")) }); const MAX_ERROR_CHARS: usize = 500; +const MAX_ERROR_INPUT_CHARS: usize = 16 * 1024; +static ERROR_URL: LazyLock = + LazyLock::new(|| Regex::new(r#"(?i)https?://[^\s\"'<>]+"#).expect("valid error URL pattern")); +static ERROR_CREDENTIAL: LazyLock = LazyLock::new(|| { + Regex::new(r#"(?i)[\"'\\]*\b(?:authorization|proxy-authorization|x-api-key|api[_-]?key|access[_-]?token|refresh[_-]?token|id[_-]?token|password|secret)\b[\"'\\\s]*[:=]\s*"#) + .expect("valid error credential pattern") +}); +static ERROR_BEARER: LazyLock = + LazyLock::new(|| Regex::new(r#"(?i)\bbearer\s+[^\s,\"';}\]]+"#).expect("valid bearer pattern")); #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum RemoteLogLevel { @@ -210,8 +220,49 @@ fn write_local_log(payload: &RemoteLogPayload) { } pub(crate) fn sanitize_error(error: impl std::fmt::Display) -> String { - let compact = error - .to_string() + // Stop formatting before scanning/redacting an arbitrarily large upstream + // error body. A truncated diagnostic must never traverse the entire body. + struct ErrorPrefix { + text: String, + remaining: usize, + } + impl std::fmt::Write for ErrorPrefix { + fn write_str(&mut self, value: &str) -> std::fmt::Result { + if let Some((boundary, _)) = value.char_indices().nth(self.remaining) { + self.text.push_str(&value[..boundary]); + self.remaining = 0; + Err(std::fmt::Error) + } else { + self.text.push_str(value); + self.remaining -= value.chars().count(); + Ok(()) + } + } + } + let mut prefix = ErrorPrefix { + text: String::new(), + remaining: MAX_ERROR_INPUT_CHARS, + }; + let _ = std::fmt::write(&mut prefix, format_args!("{error}")); + let text = prefix.text; + let safe_urls = ERROR_URL.replace_all(&text, |captures: ®ex::Captures<'_>| { + let matched = &captures[0]; + let url_text = matched.trim_end_matches([')', ']', '}', ',', ';']); + let suffix = &matched[url_text.len()..]; + match reqwest::Url::parse(url_text) { + Ok(mut url) => { + let _ = url.set_username(""); + let _ = url.set_password(None); + url.set_query(None); + url.set_fragment(None); + format!("{url}{suffix}") + } + Err(_) => format!("[redacted URL]{suffix}"), + } + }); + let safe_credentials = redact_error_credentials(&safe_urls); + let safe_bearers = ERROR_BEARER.replace_all(&safe_credentials, "Bearer [redacted]"); + let compact = safe_bearers .split_whitespace() .map(redact_sensitive_token) .collect::>() @@ -219,6 +270,63 @@ pub(crate) fn sanitize_error(error: impl std::fmt::Display) -> String { truncate_chars(&compact, MAX_ERROR_CHARS) } +fn redact_error_credentials(text: &str) -> String { + let mut redacted = String::with_capacity(text.len()); + let mut cursor = 0; + for assignment in ERROR_CREDENTIAL.find_iter(text) { + // A match inside an already-redacted credential is credential content, + // not a new assignment. Do not copy any of that content back out. + if assignment.start() < cursor { + continue; + } + redacted.push_str(&text[cursor..assignment.end()]); + redacted.push_str("[redacted]"); + cursor = assignment.end() + error_credential_value_end(&text[assignment.end()..]); + } + redacted.push_str(&text[cursor..]); + redacted +} + +fn error_credential_value_end(value: &str) -> usize { + let escapes = value.bytes().take_while(|byte| *byte == b'\\').count(); + if let Some(quote @ (b'"' | b'\'')) = value.as_bytes().get(escapes).copied() { + // JSON strings can themselves contain escaped JSON. A delimiter at + // nesting depth N has 2^N-1 preceding slashes; an escaped quote inside + // its value has a different residue. Preserve the original escaping, + // so a quoted password cannot be broken open by global unescaping. + let span = escapes + 1; + if !span.is_power_of_two() { + return value.len(); + } + let mut slashes = 0; + let start = escapes + 1; + for (index, character) in value[start..].char_indices() { + if character == '\\' { + slashes += 1; + continue; + } + if character as u32 == u32::from(quote) && slashes % (2 * span) == escapes { + return start + index + 1; + } + slashes = 0; + } + // An incomplete quoted value may be a truncated provider error body. + return value.len(); + } + if value.starts_with('{') || value.starts_with('[') { + let mut values = serde_json::Deserializer::from_str(value).into_iter::(); + return if values.next().is_some_and(|result| result.is_ok()) { + values.byte_offset() + } else { + value.len() + }; + } + // An unquoted credential/header has no reliable value terminator. It may + // contain spaces, an authentication scheme, or multiple signed fields. + // Redact through the line boundary instead of guessing and exposing a suffix. + value.find(['\r', '\n']).unwrap_or(value.len()) +} + pub(crate) fn raw_model_payload_context( ctx: &ModelContext, build_payload: impl FnOnce(usize) -> Value, @@ -287,7 +395,7 @@ async fn send_remote_log(config: RemoteLogConfig, payload: RemoteLogPayload) -> .json(&payload) .send() .await - .map_err(|error| format!("request failed: {error}"))?; + .map_err(|error| format!("request failed: {}", sanitize_error(error)))?; if !response.status().is_success() { return Err(format!("server returned HTTP {}", response.status())); diff --git a/app/src/ai/remote_logging_tests.rs b/app/src/ai/remote_logging_tests.rs index f1691dde..1007089a 100644 --- a/app/src/ai/remote_logging_tests.rs +++ b/app/src/ai/remote_logging_tests.rs @@ -24,6 +24,109 @@ fn errors_are_compacted_truncated_and_lightly_redacted() { assert!(sanitized.chars().count() <= 501); } +#[test] +fn error_redaction_removes_url_credentials_query_tokens_and_header_values() { + let input = "request failed for https://alice:private-password@example.com/v1/chat?api_key=query-value&code=oauth-code#fragment\nAuthorization: Bearer bearer-value\nx-api-key: header-value\n{\"refresh_token\":\"refresh-value\",\"api_key\":\"json-value\"}\nTLS certificate unknown issuer"; + let actual = sanitize_error(input); + for secret in [ + "alice", + "private-password", + "query-value", + "oauth-code", + "fragment", + "bearer-value", + "header-value", + "refresh-value", + "json-value", + ] { + assert!( + !actual.contains(secret), + "credential escaped redaction: {secret}" + ); + } + assert!(actual.contains("https://example.com/v1/chat")); + assert!(actual.contains("TLS certificate unknown issuer")); +} + +#[test] +fn error_redaction_handles_nested_json_and_preserves_nonsecret_token_counts() { + let actual = sanitize_error( + r#"{\"access_token\":\"private-access\",\"password\":\"private-password\",\"input_tokens\":42} network timed out"#, + ); + assert!(!actual.contains("private-access")); + assert!(!actual.contains("private-password")); + assert!(actual.contains("input_tokens")); + assert!(actual.contains("42")); + assert!(actual.contains("network timed out")); +} + +#[test] +fn escaped_quotes_and_backslashes_do_not_expose_credential_suffixes() { + for input in [ + r#"{"password":"private-\"quoted\"-suffix","input_tokens":42} connection reset"#, + r#"{\"password\":\"private-\\\"quoted\\\"-suffix\",\"input_tokens\":42} connection reset"#, + r#"api_key="private\\path\"quoted-suffix" connection reset"#, + r#"password="private unterminated credential suffix"#, + ] { + let actual = sanitize_error(input); + for secret in ["private", "quoted", "suffix", "path"] { + assert!( + !actual.contains(secret), + "credential fragment escaped redaction: {actual}" + ); + } + if input.contains("connection reset") { + assert!(actual.contains("connection reset")); + } + } +} + +#[test] +fn basic_and_multifield_authorization_values_are_redacted() { + let basic = sanitize_error("Authorization: Basic dXNlcjpwYXNz\nTLS certificate rejected"); + assert!(!basic.contains("dXNlcjpwYXNz")); + assert!(basic.contains("TLS certificate rejected")); + let digest = sanitize_error("Proxy-Authorization: Digest username=private-user, response=private-response\nconnection reset"); + assert!(!digest.contains("private-user")); + assert!(!digest.contains("private-response")); + assert!(digest.contains("connection reset")); + let password = sanitize_error("password: private password with spaces\nnetwork timeout"); + assert!(!password.contains("private password")); + assert!(password.contains("network timeout")); + for prefix in ["Basic ", "Bearer "] { + let password = sanitize_error(format!( + "password: {prefix}private password with spaces\nnetwork timeout" + )); + assert!(!password.contains("private password")); + assert!(!password.contains("with spaces")); + assert!(password.contains("network timeout")); + } + let spaced = sanitize_error("Authorization: Bearer\t private-token\nx-api-key: private key with spaces\nnetwork timeout"); + assert!(!spaced.contains("private-token")); + assert!(!spaced.contains("private key")); + assert!(spaced.contains("network timeout")); +} + +#[test] +fn error_formatting_stops_before_walking_an_unbounded_display() { + use std::cell::Cell; + struct LongError<'a>(&'a Cell); + impl std::fmt::Display for LongError<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + for _ in 0..1_000_000 { + self.0.set(self.0.get() + 1); + f.write_str("é日本🙂 network timed out ")?; + } + Ok(()) + } + } + let written = Cell::new(0); + let output = sanitize_error(LongError(&written)); + assert!(written.get() < 2_000); + assert!(output.chars().count() <= 501); + assert!(output.contains("network timed out")); +} + #[test] fn raw_payload_cap_keeps_tail() { assert_eq!( diff --git a/app/src/ai/runtime/mod.rs b/app/src/ai/runtime/mod.rs index 0c9e0634..9b7e25bb 100644 --- a/app/src/ai/runtime/mod.rs +++ b/app/src/ai/runtime/mod.rs @@ -1,5 +1,7 @@ mod child_context; mod event_translator; +#[cfg(not(target_family = "wasm"))] +mod model_failure_diagnostics; mod provider_run_coordinator; mod rig; mod rig_request; diff --git a/app/src/ai/runtime/model_failure_diagnostics.rs b/app/src/ai/runtime/model_failure_diagnostics.rs new file mode 100644 index 00000000..aacc9ea6 --- /dev/null +++ b/app/src/ai/runtime/model_failure_diagnostics.rs @@ -0,0 +1,63 @@ +//! Bounded operational evidence retained until a provider run finishes. +//! +//! A failed telemetry POST must not be the only copy of a model failure. The +//! terminal event includes these summaries even when the run later recovers. +//! This is process-local diagnostic state, never provider history/checkpoints. + +use std::collections::VecDeque; + +use chrono::Utc; +use galaxy_agent_core::AgentError; +use serde_json::{json, Value}; + +use crate::ai::remote_logging::sanitize_error; + +const MAX_FAILURES: usize = 8; +const MAX_ID_CHARS: usize = 256; + +#[derive(Default)] +pub(crate) struct ModelFailureDiagnostics { + observed_count: u64, + recent: VecDeque, +} + +impl ModelFailureDiagnostics { + pub(crate) fn record(&mut self, identity: Value, error: &AgentError) { + self.observed_count = self.observed_count.saturating_add(1); + // Select operational identifiers explicitly; never retain a request, + // prompt, provider configuration, or arbitrary caller-supplied context. + let mut record = json!({ + "timestamp": Utc::now().to_rfc3339(), + "error_kind": format!("{:?}", error.kind), + "error_recoverable": error.recoverable, + "error": sanitize_error(&error.message), + }); + for key in ["provider_run_id", "profile", "runtime_id", "model_id"] { + if let Some(value) = identity.get(key).and_then(Value::as_str) { + record[key] = json!(value.chars().take(MAX_ID_CHARS).collect::()); + } + } + for key in ["provider_epoch", "retry_attempt", "elapsed_ms"] { + if let Some(value) = identity.get(key).and_then(Value::as_u64) { + record[key] = json!(value); + } + } + if self.recent.len() == MAX_FAILURES { + self.recent.pop_front(); + } + self.recent.push_back(record); + } + + pub(crate) fn snapshot(&self) -> Value { + json!({ + "scope": "current_process", + "observed_count": self.observed_count, + "truncated": self.observed_count > self.recent.len() as u64, + "recent": self.recent, + }) + } +} + +#[cfg(test)] +#[path = "model_failure_diagnostics_tests.rs"] +mod tests; diff --git a/app/src/ai/runtime/model_failure_diagnostics_tests.rs b/app/src/ai/runtime/model_failure_diagnostics_tests.rs new file mode 100644 index 00000000..edc75c6b --- /dev/null +++ b/app/src/ai/runtime/model_failure_diagnostics_tests.rs @@ -0,0 +1,58 @@ +use galaxy_agent_core::AgentErrorKind; + +use super::*; + +#[test] +fn retains_sanitized_causes_after_recovery_without_request_content() { + let mut diagnostics = ModelFailureDiagnostics::default(); + let mut error = AgentError::new(AgentErrorKind::Transport, "sk-private connection failed"); + error.recoverable = true; + diagnostics.record( + json!({ + "provider_run_id": "run-1", "provider_epoch": 6, + "retry_attempt": 0, "elapsed_ms": 100, + "model_id": "test-model", "runtime_id": "rig:openai", "profile": "base", + "prompt": "private prompt", "api_key": "private credential", + }), + &error, + ); + let snapshot = diagnostics.snapshot(); + let serialized = snapshot.to_string(); + assert_eq!(snapshot["observed_count"], 1); + assert_eq!( + snapshot["recent"][0]["error"], + "[redacted] connection failed" + ); + assert_eq!(snapshot["recent"][0]["error_kind"], "Transport"); + assert_eq!(snapshot["recent"][0]["provider_epoch"], 6); + assert_eq!(snapshot["recent"][0]["error_recoverable"], true); + assert!(!serialized.contains("private prompt")); + assert!(!serialized.contains("private credential")); + assert!(!serialized.contains("sk-private")); +} + +#[test] +fn retention_keeps_newest_failures_with_explicit_omission_accounting() { + let mut diagnostics = ModelFailureDiagnostics::default(); + for epoch in 0..MAX_FAILURES + 2 { + diagnostics.record( + json!({"provider_epoch": epoch, "model_id": "é".repeat(10_000)}), + &AgentError::new(AgentErrorKind::Provider, "overloaded"), + ); + } + let snapshot = diagnostics.snapshot(); + assert_eq!(snapshot["scope"], "current_process"); + assert_eq!(snapshot["observed_count"], MAX_FAILURES as u64 + 2); + assert_eq!(snapshot["truncated"], true); + let recent = snapshot["recent"].as_array().unwrap(); + assert_eq!(recent.len(), MAX_FAILURES); + assert_eq!(recent[0]["provider_epoch"], 2); + assert_eq!( + recent[0]["model_id"].as_str().unwrap().chars().count(), + MAX_ID_CHARS + ); + assert_eq!( + ModelFailureDiagnostics::default().snapshot()["observed_count"], + 0 + ); +} diff --git a/app/src/ai/runtime/provider_run_coordinator.rs b/app/src/ai/runtime/provider_run_coordinator.rs index b35b38af..598d1813 100644 --- a/app/src/ai/runtime/provider_run_coordinator.rs +++ b/app/src/ai/runtime/provider_run_coordinator.rs @@ -18,6 +18,8 @@ use galaxy_agent_core::{ use instant::Instant; use warpui::r#async::Timer; +#[cfg(not(target_family = "wasm"))] +use super::model_failure_diagnostics::ModelFailureDiagnostics; use super::task_progress::{add_task_list_context, initial_task_list, task_list_from_plan_batch}; #[cfg(not(target_family = "wasm"))] use crate::ai::activity; @@ -290,6 +292,8 @@ pub(crate) struct ProviderRunCoordinator { max_context_tokens: Option, #[cfg(not(target_family = "wasm"))] activity_session_id: Option, + #[cfg(not(target_family = "wasm"))] + model_failure_diagnostics: ModelFailureDiagnostics, } impl ProviderRunCoordinator { @@ -334,6 +338,8 @@ impl ProviderRunCoordinator { max_context_tokens: None, #[cfg(not(target_family = "wasm"))] activity_session_id: None, + #[cfg(not(target_family = "wasm"))] + model_failure_diagnostics: ModelFailureDiagnostics::default(), }) } @@ -349,6 +355,11 @@ impl ProviderRunCoordinator { self.activity_session_id = Some(conversation_id.to_string()); } + #[cfg(not(target_family = "wasm"))] + pub(crate) fn model_failure_diagnostics(&self) -> serde_json::Value { + self.model_failure_diagnostics.snapshot() + } + #[cfg(not(target_family = "wasm"))] fn activity_context( &self, @@ -1206,6 +1217,12 @@ impl ProviderRunCoordinator { where F: FnMut(ProviderRunProjection) -> BoxFuture<'static, Result<(), String>>, { + #[cfg(not(target_family = "wasm"))] + { + let mut identity = self.activity_context(call, profile); + identity["elapsed_ms"] = serde_json::json!(elapsed_millis(started_at)); + self.model_failure_diagnostics.record(identity, &error); + } #[cfg(not(target_family = "wasm"))] model_output_logging::log(serde_json::json!({ "event": "model_error", diff --git a/app/src/ai/runtime/rig_tool.rs b/app/src/ai/runtime/rig_tool.rs index 6a83c012..908b36f7 100644 --- a/app/src/ai/runtime/rig_tool.rs +++ b/app/src/ai/runtime/rig_tool.rs @@ -16,6 +16,7 @@ use crate::ai::agent::{ RunAgentsRequest, SearchCodebaseRequest, ShellCommandDelay, }; use crate::ai::document::ai_document_model::AIDocumentId; +use crate::ai::grep_query::prepare_grep_queries; const MAX_BATCH_READ_ITEMS: usize = 6; const MAX_BATCH_SEARCH_ITEMS: usize = 3; @@ -83,7 +84,10 @@ pub(crate) fn action_from_tool_call( title: Some(required_string(input, "summary")?), }, "grep" => AIAgentActionType::Grep { - queries: limited_required_strings(input, "queries", MAX_BATCH_SEARCH_ITEMS)?, + queries: prepare_grep_queries( + limited_required_strings(input, "queries", MAX_BATCH_SEARCH_ITEMS)?, + optional_boolean(input, "fixed_strings")?.unwrap_or(false), + ), path: optional_string(input, "path")?.unwrap_or_default(), }, "file_glob" => AIAgentActionType::FileGlob { diff --git a/app/src/ai/runtime/rig_tool_tests.rs b/app/src/ai/runtime/rig_tool_tests.rs index faa76f05..b0dd2971 100644 --- a/app/src/ai/runtime/rig_tool_tests.rs +++ b/app/src/ai/runtime/rig_tool_tests.rs @@ -651,3 +651,74 @@ fn mcp_title_metadata_does_not_change_tool_arguments() { if input == serde_json::json!({"message": "hello", "_galaxy_tool_title": "a real server argument"})) ); } + + +#[test] +fn grep_literal_mode_escapes_queries_without_changing_scope() { + let action = action_from_tool_call( + "task", + &call( + "grep", + serde_json::json!({ + "queries": ["variantResolutions: []", "indexAsset(context"], + "fixed_strings": true, + "path": "/project/src", + }), + ), + &SkillPathOrigin::Local, + &HashMap::new(), + ) + .unwrap(); + let AIAgentActionType::Grep { queries, path } = action.action else { + panic!("expected grep action"); + }; + assert_eq!( + queries, + vec![r"variantResolutions: \[\]", r"indexAsset\(context"] + ); + assert_eq!(path, "/project/src"); +} + +#[test] +fn grep_default_and_explicit_regex_modes_preserve_patterns() { + for fixed_strings in [None, Some(false)] { + let mut input = serde_json::json!({"queries": ["[unclosed", "foo|bar"]}); + if let Some(value) = fixed_strings { + input["fixed_strings"] = value.into(); + } + let action = action_from_tool_call( + "task", + &call("grep", input), + &SkillPathOrigin::Local, + &HashMap::new(), + ) + .unwrap(); + let AIAgentActionType::Grep { queries, .. } = action.action else { + panic!("expected grep action"); + }; + assert_eq!(queries, vec!["[unclosed", "foo|bar"]); + } +} + +#[test] +fn grep_literal_mode_still_rejects_invalid_booleans_and_excessive_batches() { + for (input, expected) in [ + ( + serde_json::json!({"queries": ["needle"], "fixed_strings": "true"}), + "fixed_strings", + ), + ( + serde_json::json!({"queries": ["a", "b", "c", "d"], "fixed_strings": true}), + "queries", + ), + ] { + let error = action_from_tool_call( + "task", + &call("grep", input), + &SkillPathOrigin::Local, + &HashMap::new(), + ) + .unwrap_err(); + assert!(error.contains(expected), "unexpected error: {error}"); + } +} diff --git a/crates/galaxy_agent_rig/Cargo.toml b/crates/galaxy_agent_rig/Cargo.toml index d555eea7..b84784ea 100644 --- a/crates/galaxy_agent_rig/Cargo.toml +++ b/crates/galaxy_agent_rig/Cargo.toml @@ -10,6 +10,7 @@ async-stream.workspace = true async-trait.workspace = true aws-sdk-bedrockruntime.workspace = true base64.workspace = true +bytes.workspace = true futures.workspace = true galaxy_agent_core.workspace = true http.workspace = true @@ -21,6 +22,5 @@ uuid.workspace = true [dev-dependencies] aws-smithy-http-client.workspace = true -bytes.workspace = true rig-core = { workspace = true, features = ["test-utils"] } tokio = { workspace = true, features = ["macros", "rt"] } diff --git a/crates/galaxy_agent_rig/src/anthropic_auth.rs b/crates/galaxy_agent_rig/src/anthropic_auth.rs index 2dc73067..9e79f6eb 100644 --- a/crates/galaxy_agent_rig/src/anthropic_auth.rs +++ b/crates/galaxy_agent_rig/src/anthropic_auth.rs @@ -13,6 +13,8 @@ use rig_core::providers::anthropic; use rig_core::providers::anthropic::client::AnthropicExt; use serde_json::json; +use crate::http_client::DiagnosticHttpClient; + pub fn is_claude_subscription_token(credential: &str) -> bool { credential.trim().contains("sk-ant-oat") } @@ -64,15 +66,22 @@ impl ProviderBuilder for SubscriptionBuilder { } } -pub(crate) fn anthropic_client(credential: &str) -> Result { +pub(crate) fn anthropic_client( + credential: &str, +) -> Result, String> { let credential = credential.trim(); if !is_claude_subscription_token(credential) { - return anthropic::Client::new(credential).map_err(|error| error.to_string()); + return anthropic::Client::builder() + .api_key(credential) + .http_client(DiagnosticHttpClient::new()) + .build() + .map_err(|error| error.to_string()); } let headers = subscription_headers(); Client::::builder() .api_key(SubscriptionToken(credential.to_string())) .http_headers(headers) + .http_client(DiagnosticHttpClient::new()) .build() // Never include a credential or raw header-validation error in diagnostics. .map_err(|_| "Could not configure Claude subscription authentication.".to_string()) diff --git a/crates/galaxy_agent_rig/src/chatgpt.rs b/crates/galaxy_agent_rig/src/chatgpt.rs index 6036a266..6ec86fd6 100644 --- a/crates/galaxy_agent_rig/src/chatgpt.rs +++ b/crates/galaxy_agent_rig/src/chatgpt.rs @@ -11,6 +11,7 @@ use galaxy_agent_core::{ use rig_core::client::CompletionClient; use rig_core::providers::chatgpt::{self, ChatGPTAuth}; +use crate::http_client::DiagnosticHttpClient; use crate::request::build_completion_request; use crate::stream::start_model_turn; @@ -155,6 +156,7 @@ impl AgentRuntime for ChatGPTSubscriptionRuntime { .api_key(auth) .allow_device_flow(false) .originator("galaxy") + .http_client(DiagnosticHttpClient::new()) .build() .map_err(|error| AgentError::new(AgentErrorKind::Configuration, error.to_string()))?; let model = client.completion_model(&self.config.model); diff --git a/crates/galaxy_agent_rig/src/http_client.rs b/crates/galaxy_agent_rig/src/http_client.rs new file mode 100644 index 00000000..c9a973ff --- /dev/null +++ b/crates/galaxy_agent_rig/src/http_client.rs @@ -0,0 +1,224 @@ +//! Keep useful transport causes before Rig reduces SSE errors to their Display text. +use std::error::Error; +use std::fmt; + +use bytes::Bytes; +use futures::StreamExt; +use rig_core::http_client::{ + self, HttpClientExt, LazyBody, MultipartForm, Request, Response, StreamingResponse, +}; +use rig_core::wasm_compat::WasmCompatSend; + +#[derive(Clone, Debug, Default)] +pub(crate) struct DiagnosticHttpClient(H); + +impl DiagnosticHttpClient { + pub(crate) fn new() -> Self { + Self(http_client::ReqwestClient::default()) + } +} + +impl HttpClientExt for DiagnosticHttpClient { + fn send( + &self, + request: Request, + ) -> impl Future>>> + WasmCompatSend + 'static + where + T: Into + WasmCompatSend, + U: From + WasmCompatSend + 'static, + { + let client = self.0.clone(); + let request = request.map(Into::::into); + async move { + let response = client + .send(request) + .await + .map_err(diagnose_transport_error)?; + Ok(response.map(|body| { + Box::pin(async move { body.await.map_err(diagnose_transport_error) }) as LazyBody + })) + } + } + + fn send_multipart( + &self, + request: Request, + ) -> impl Future>>> + WasmCompatSend + 'static + where + U: From + WasmCompatSend + 'static, + { + let client = self.0.clone(); + async move { + let response = client + .send_multipart(request) + .await + .map_err(diagnose_transport_error)?; + Ok(response.map(|body| { + Box::pin(async move { body.await.map_err(diagnose_transport_error) }) as LazyBody + })) + } + } + + fn send_streaming( + &self, + request: Request, + ) -> impl Future> + WasmCompatSend + where + T: Into + WasmCompatSend, + { + let client = self.0.clone(); + let request = request.map(Into::::into); + async move { + let response = client + .send_streaming(request) + .await + .map_err(diagnose_transport_error)?; + Ok(response.map(|body| { + Box::pin(body.map(|chunk| chunk.map_err(diagnose_transport_error))) + as http_client::sse::BoxedStream + })) + } + } +} + +// Only static categories and numeric OS codes leave this module. Transport errors +// can contain proxy passwords, URL userinfo/query keys, or arbitrary source text; +// neither Display nor Debug of the original error is safe to forward verbatim. +// HTTP status/body errors pass through unchanged so Rig's provider classification +// and its body-inspection helpers retain their original semantics. +pub(crate) fn diagnose_transport_error(error: http_client::Error) -> http_client::Error { + match error { + http_client::Error::Instance(source) => http_client::Error::Instance(Box::new( + TransportDiagnostic::from_source(source.as_ref()), + )), + other => other, + } +} + +#[derive(Debug)] +struct TransportDiagnostic(String); + +impl TransportDiagnostic { + fn from_source(source: &(dyn Error + 'static)) -> Self { + const MAX_DEPTH: usize = 12; + const MAX_CAUSES: usize = 4; + let mut causes = Vec::::new(); + let mut source = Some(source); + for _ in 0..MAX_DEPTH { + let Some(error) = source else { break }; + let text = bounded_error_text(error) + .split_whitespace() + .filter(|word| !word.contains("://")) + .collect::>() + .join(" ") + .to_ascii_lowercase(); + if let Some(category) = transport_category(&text) { + let category = category.to_string(); + if !causes.contains(&category) && causes.len() < MAX_CAUSES { + causes.push(category); + } + } + if let Some(error) = error.downcast_ref::() { + let detail = match error.raw_os_error() { + Some(code) => format!("I/O {:?} (OS error {code})", error.kind()), + None => format!("I/O {:?}", error.kind()), + }; + if !causes.contains(&detail) && causes.len() < MAX_CAUSES { + causes.push(detail); + } + } + source = error.source(); + } + if causes.is_empty() { + causes.push("underlying transport cause unavailable".to_string()); + } + Self(format!( + "provider request transport failed: {}", + causes.join("; caused by: ") + )) + } +} + +impl fmt::Display for TransportDiagnostic { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +impl Error for TransportDiagnostic {} + +fn bounded_error_text(error: &(dyn Error + 'static)) -> String { + struct BoundedText(String); + impl fmt::Write for BoundedText { + fn write_str(&mut self, value: &str) -> fmt::Result { + let remaining = 1024usize.saturating_sub(self.0.chars().count()); + match value.char_indices().nth(remaining) { + Some((boundary, _)) => { + self.0.push_str(&value[..boundary]); + Err(fmt::Error) + } + None => { + self.0.push_str(value); + Ok(()) + } + } + } + } + let mut text = BoundedText(String::new()); + let _ = fmt::write(&mut text, format_args!("{error}")); + text.0 +} + +fn transport_category(text: &str) -> Option<&'static str> { + if text.contains("unknownissuer") + || text.contains("unknown issuer") + || text.contains("unknown ca") + { + Some("TLS certificate issuer is not trusted") + } else if text.contains("notvalidforname") + || text.contains("certificate") + && (text.contains("hostname") || text.contains("not valid for name")) + { + Some("TLS certificate hostname mismatch") + } else if text.contains("certificate") + && (text.contains("expired") + || text.contains("notvalidyet") + || text.contains("not valid yet")) + { + Some("TLS certificate is expired or not yet valid") + } else if text.contains("certificate") + && (text.contains("invalid") || text.contains("verify") || text.contains("validation")) + || text.contains("cert verify") + { + Some("TLS certificate validation failed") + } else if text.contains("dns") + || text.contains("failed to lookup address") + || text.contains("name or service not known") + || text.contains("nodename nor servname") + { + Some("DNS resolution failed") + } else if text.contains("proxy") && (text.contains("authentication") || text.contains("407")) { + Some("proxy authentication failed") + } else if text.contains("connection refused") { + Some("connection refused") + } else if text.contains("connection reset") { + Some("connection reset") + } else if text.contains("timed out") || text.contains("timeout") { + Some("network operation timed out") + } else if text.contains("unreachable") { + Some("network or destination unreachable") + } else if text.contains("tls") || text.contains("ssl") || text.contains("handshake") { + Some("TLS handshake failed") + } else if text.contains("unexpected eof") + || text.contains("connection closed") + || text.contains("incomplete message") + { + Some("connection closed before response completed") + } else { + None + } +} + +#[cfg(test)] +#[path = "http_client_tests.rs"] +mod tests; diff --git a/crates/galaxy_agent_rig/src/http_client_tests.rs b/crates/galaxy_agent_rig/src/http_client_tests.rs new file mode 100644 index 00000000..577e7a9a --- /dev/null +++ b/crates/galaxy_agent_rig/src/http_client_tests.rs @@ -0,0 +1,253 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use futures::executor::block_on; +use futures::{StreamExt, future, stream}; +use http::StatusCode; + +use super::*; + +#[derive(Debug)] +struct SourceError { + message: &'static str, + source: Option>, +} + +impl fmt::Display for SourceError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.message) + } +} + +impl Error for SourceError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + self.source.as_deref().map(|source| source as &dyn Error) + } +} + +fn transport_failure() -> http_client::Error { + http_client::Error::Instance(Box::new(SourceError { + message: "error sending request for url (https://user:password@example.test/v1?api_key=secret-certificate)", + source: Some(Box::new(SourceError { + message: "failed to lookup address information: Name or service not known; Authorization: Bearer secret-header", + source: None, + })), + })) +} + +#[test] +fn nested_dns_cause_survives_without_url_credentials_queries_or_headers() { + let message = diagnose_transport_error(transport_failure()).to_string(); + assert_eq!( + message, + "Http client error: provider request transport failed: DNS resolution failed" + ); +} + +#[test] +fn tls_certificate_cause_is_specific_and_contains_only_static_diagnostics() { + let error = http_client::Error::Instance(Box::new(SourceError { + message: "error sending request for url (https://example.test/?token=private)", + source: Some(Box::new(SourceError { + message: "invalid peer certificate: UnknownIssuer (private-certificate-content)", + source: None, + })), + })); + assert_eq!( + diagnose_transport_error(error).to_string(), + "Http client error: provider request transport failed: TLS certificate issuer is not trusted" + ); +} + +#[test] +fn unrecognized_causes_never_forward_arbitrary_source_text() { + let error = http_client::Error::Instance(Box::new(SourceError { + message: "Authorization: Bearer secret; api_key=private; password=hidden", + source: None, + })); + assert_eq!( + diagnose_transport_error(error).to_string(), + "Http client error: provider request transport failed: underlying transport cause unavailable" + ); +} + +#[test] +fn os_errors_retain_numeric_codes_without_the_original_message() { + let message = diagnose_transport_error(http_client::Error::Instance(Box::new( + std::io::Error::from_raw_os_error(111), + ))) + .to_string(); + assert_eq!(message.contains("OS error 111"), true); + assert_eq!(message.len() < 400, true); +} + +#[test] +fn large_unicode_error_display_stops_before_formatting_later_chunks() { + #[derive(Debug)] + struct LargeError { + payload: String, + later_chunks: AtomicUsize, + } + impl fmt::Display for LargeError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.payload)?; + self.later_chunks.fetch_add(1, Ordering::Relaxed); + formatter.write_str("sensitive trailing data") + } + } + impl Error for LargeError {} + + let error = LargeError { + payload: "🌌".repeat(100_000), + later_chunks: AtomicUsize::new(0), + }; + assert_eq!(bounded_error_text(&error), "🌌".repeat(1024)); + assert_eq!(error.later_chunks.load(Ordering::Relaxed), 0); +} + +#[test] +fn http_status_and_provider_body_remain_available_for_classification() { + for status in [ + StatusCode::UNAUTHORIZED, + StatusCode::TOO_MANY_REQUESTS, + StatusCode::SERVICE_UNAVAILABLE, + ] { + let error = diagnose_transport_error(http_client::Error::InvalidStatusCodeWithMessage( + status, + "provider error body".to_owned(), + )); + match error { + http_client::Error::InvalidStatusCodeWithMessage(actual_status, body) => { + assert_eq!(actual_status, status); + assert_eq!(body, "provider error body"); + } + other => panic!("HTTP response changed error variant: {other:?}"), + } + } +} + +#[derive(Clone, Debug)] +struct FakeHttpClient { + calls: Arc, + fail_before_response: bool, +} + +impl HttpClientExt for FakeHttpClient { + fn send( + &self, + _request: Request, + ) -> impl Future>>> + WasmCompatSend + 'static + where + T: Into + WasmCompatSend, + U: From + WasmCompatSend + 'static, + { + self.calls.fetch_add(1, Ordering::Relaxed); + future::ready(if self.fail_before_response { + Err(transport_failure()) + } else { + Ok(Response::new( + Box::pin(async { Err(transport_failure()) }) as LazyBody + )) + }) + } + + fn send_multipart( + &self, + request: Request, + ) -> impl Future>>> + WasmCompatSend + 'static + where + U: From + WasmCompatSend + 'static, + { + self.send(request.map(|_| Bytes::new())) + } + + fn send_streaming( + &self, + request: Request, + ) -> impl Future> + WasmCompatSend + where + T: Into + WasmCompatSend, + { + self.calls.fetch_add(1, Ordering::Relaxed); + // The interceptor must forward the approved payload and credential header unchanged. + assert_eq!(request.headers()["authorization"], "Bearer synthetic-key"); + assert_eq!( + request.into_body().into(), + Bytes::from_static(b"approved request") + ); + future::ready(if self.fail_before_response { + Err(transport_failure()) + } else { + let body = stream::iter([ + Ok(Bytes::from_static(b"original SSE bytes")), + Err(transport_failure()), + ]); + Ok(Response::builder() + .status(StatusCode::OK) + .header("x-request-id", "original-request-id") + .body(Box::pin(body) as http_client::sse::BoxedStream) + .unwrap()) + }) + } +} + +fn request() -> Request { + Request::builder() + .uri("https://example.test/v1?secret=private") + .header("authorization", "Bearer synthetic-key") + .body(Bytes::from_static(b"approved request")) + .unwrap() +} + +#[test] +fn streaming_adapter_preserves_response_and_chunks_without_internal_retries() { + block_on(async { + let calls = Arc::new(AtomicUsize::new(0)); + let client = DiagnosticHttpClient(FakeHttpClient { + calls: calls.clone(), + fail_before_response: false, + }); + let response = client.send_streaming(request()).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["x-request-id"], "original-request-id"); + let mut body = response.into_body(); + assert_eq!( + body.next().await.unwrap().unwrap(), + Bytes::from_static(b"original SSE bytes") + ); + assert_eq!( + body.next() + .await + .unwrap() + .unwrap_err() + .to_string() + .contains("DNS resolution failed"), + true + ); + assert_eq!(body.next().await.is_none(), true); + assert_eq!(calls.load(Ordering::Relaxed), 1); + }); +} + +#[test] +fn adapter_preserves_causes_before_headers_and_during_lazy_body_reads() { + block_on(async { + let calls = Arc::new(AtomicUsize::new(0)); + let client = DiagnosticHttpClient(FakeHttpClient { + calls: calls.clone(), + fail_before_response: true, + }); + let error = client.send_streaming(request()).await.err().unwrap(); + assert_eq!(error.to_string().contains("DNS resolution failed"), true); + assert_eq!(calls.load(Ordering::Relaxed), 1); + + let client = DiagnosticHttpClient(FakeHttpClient { + calls: calls.clone(), + fail_before_response: false, + }); + let response = client.send::<_, Bytes>(request()).await.unwrap(); + let error = response.into_body().await.unwrap_err(); + assert_eq!(error.to_string().contains("DNS resolution failed"), true); + assert_eq!(calls.load(Ordering::Relaxed), 2); + }); +} diff --git a/crates/galaxy_agent_rig/src/lib.rs b/crates/galaxy_agent_rig/src/lib.rs index cb247379..d08cb2e8 100644 --- a/crates/galaxy_agent_rig/src/lib.rs +++ b/crates/galaxy_agent_rig/src/lib.rs @@ -3,6 +3,7 @@ mod anthropic_auth; mod bedrock; mod chatgpt; +mod http_client; mod native; mod openai_compatible; mod request; diff --git a/crates/galaxy_agent_rig/src/native.rs b/crates/galaxy_agent_rig/src/native.rs index 406026e0..7fc1bcf5 100644 --- a/crates/galaxy_agent_rig/src/native.rs +++ b/crates/galaxy_agent_rig/src/native.rs @@ -13,6 +13,7 @@ use rig_core::http_client::{self, HttpClientExt}; use rig_core::providers::gemini; use crate::anthropic_auth::{anthropic_client, prepare_subscription_request, restore_tool_names}; +use crate::http_client::DiagnosticHttpClient; use crate::request::build_completion_request; use crate::stream::{start_model_completion, start_model_turn}; @@ -124,7 +125,11 @@ fn anthropic_reasoning_efforts(entry: &serde_json::Value) -> Vec { } pub async fn discover_gemini_models(api_key: &str) -> Result, String> { - let client = gemini::Client::new(api_key).map_err(|error| error.to_string())?; + let client = gemini::Client::builder() + .api_key(api_key) + .http_client(DiagnosticHttpClient::new()) + .build() + .map_err(|error| error.to_string())?; let models = client .list_models() .await @@ -349,7 +354,10 @@ impl AgentRuntime for GeminiRuntime { "A Gemini API key is required", )); } - let client = gemini::Client::new(&self.config.api_key) + let client = gemini::Client::builder() + .api_key(self.config.api_key.as_str()) + .http_client(DiagnosticHttpClient::new()) + .build() .map_err(|error| AgentError::new(AgentErrorKind::Configuration, error.to_string()))?; let model = client.completion_model(&self.config.model); let max_output_tokens = request.max_output_tokens.or(self.config.max_output_tokens); diff --git a/crates/galaxy_agent_rig/src/openai_compatible.rs b/crates/galaxy_agent_rig/src/openai_compatible.rs index afbc0962..b949e6ad 100644 --- a/crates/galaxy_agent_rig/src/openai_compatible.rs +++ b/crates/galaxy_agent_rig/src/openai_compatible.rs @@ -7,6 +7,7 @@ use rig_core::client::CompletionClient; use rig_core::completion::{CompletionModel, CompletionRequest}; use rig_core::providers::openai; +use crate::http_client::DiagnosticHttpClient; use crate::request::build_completion_request as build_provider_completion_request; use crate::stream::start_model_turn as start_provider_model_turn; @@ -63,6 +64,7 @@ impl AgentRuntime for OpenAICompatibleRuntime { // compatibility with unauthenticated local OpenAI-compatible servers. .api_key(self.config.api_key.as_deref().unwrap_or_default()) .base_url(&self.config.base_url) + .http_client(DiagnosticHttpClient::new()) .build() .map_err(|error| AgentError::new(AgentErrorKind::Configuration, error.to_string()))?; let model = client.completion_model(&self.config.model); diff --git a/crates/galaxy_agent_rig/src/stream.rs b/crates/galaxy_agent_rig/src/stream.rs index c67cb89e..e620abed 100644 --- a/crates/galaxy_agent_rig/src/stream.rs +++ b/crates/galaxy_agent_rig/src/stream.rs @@ -484,9 +484,9 @@ fn map_completion_error(error: CompletionError) -> AgentError { } }; let mut mapped = AgentError::new(kind, error.to_string()); - if is_authentication_failure { + if kind == AgentErrorKind::Authentication { mapped.user_message = Some( - "AWS credentials for Bedrock are missing or expired. Refresh your AWS login and try again." + "Provider authentication failed. Check the credentials and access permissions for this connection, then try again." .to_string(), ); } diff --git a/crates/galaxy_agent_rig/src/stream_tests.rs b/crates/galaxy_agent_rig/src/stream_tests.rs index 61df2a33..b1026f9f 100644 --- a/crates/galaxy_agent_rig/src/stream_tests.rs +++ b/crates/galaxy_agent_rig/src/stream_tests.rs @@ -15,6 +15,22 @@ fn flattened_sse_http_client_error_is_recoverable_transport() { assert!(mapped.recoverable); } +#[test] +fn safe_transport_diagnostics_survive_rig_flattening_and_remain_retryable() { + let error = crate::http_client::diagnose_transport_error( + rig_core::http_client::Error::Instance(Box::new(std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "private-token-not-for-diagnostics", + ))), + ); + // Rig uses Display text for SSE failures without a provider response status. + let mapped = map_completion_error(CompletionError::ProviderError(error.to_string())); + assert_eq!(mapped.kind, AgentErrorKind::Transport); + assert_eq!(mapped.recoverable, true); + assert_eq!(mapped.message.contains("ConnectionRefused"), true); + assert_eq!(mapped.message.contains("private-token"), false); +} + #[test] fn flattened_streamed_provider_overload_is_recoverable() { let error = CompletionError::ProviderError( diff --git a/crates/galaxy_agent_rig/src/stream_unit_tests.rs b/crates/galaxy_agent_rig/src/stream_unit_tests.rs index f103a275..21621243 100644 --- a/crates/galaxy_agent_rig/src/stream_unit_tests.rs +++ b/crates/galaxy_agent_rig/src/stream_unit_tests.rs @@ -117,6 +117,22 @@ fn flattened_bedrock_credential_errors_are_authentication_failures() { } } +#[test] +fn non_aws_authentication_errors_do_not_recommend_an_aws_login() { + for error in [ + CompletionError::from_http_response(http::StatusCode::UNAUTHORIZED, "Invalid API key"), + CompletionError::ProviderError("AccessDenied: model access is not permitted".to_owned()), + ] { + let mapped = map_completion_error(error); + assert_eq!(mapped.kind, AgentErrorKind::Authentication); + assert_eq!(mapped.recoverable, false); + let message = mapped.user_message.unwrap(); + assert_eq!(message.contains("Provider authentication failed"), true); + assert_eq!(message.contains("AWS"), false); + assert_eq!(message.contains("Bedrock"), false); + } +} + #[test] fn unrelated_provider_failures_do_not_become_authentication_failures() { for message in [