diff --git a/docs/public/core-concepts/models.mdx b/docs/public/core-concepts/models.mdx index ae0c60fcf..64fb23c4e 100644 --- a/docs/public/core-concepts/models.mdx +++ b/docs/public/core-concepts/models.mdx @@ -137,7 +137,9 @@ Model roles are separate: the provider's `default_model` controls normal model s Provider auth has two parts. The lithos `auth` scheme says how a credential is sent: `{ type = "bearer" }`, `{ type = "header", name = "x-api-key" }`, `{ type = "headers" }` for providers that take several secret headers, `{ type = "none" }`, or `{ type = "aws" }`. lithos also says which secret names a provider reads: `OPENAI_API_KEY` for `openai`, `GEMINI_API_KEY` then `GOOGLE_API_KEY` for `gemini`, `MODAL_TOKEN_ID` and `MODAL_TOKEN_SECRET` for `modal`, and `_API_KEY` for a provider you define. Fabro looks each name up in the process environment first and the server vault second. Custom headers for any provider go in `default_headers` as literal text or `{{ secrets.NAME }}` tokens; put credentials in secrets and reference them with `{{ secrets.NAME }}` instead of a bare literal. -Workflow runs also add `x-session-id: ` to every LLM request so compatible gateways can group requests from the same run. An explicitly configured `x-session-id` in provider `default_headers` takes precedence. +Workflow runs also add `x-session-id: ` to native HTTP model requests (including model-based hooks, run titles, Ask Fabro sessions, and pull request generation) so compatible gateways can group requests from the same run. An explicitly configured `x-session-id` in provider `default_headers` takes precedence, regardless of header-name casing. + +The ID stays the same when the run resumes or a request retries or falls back to another provider. A new run gets its own ID. AWS-signed requests and native Bedrock bearer requests and external ACP agents do not receive this header automatically. Provider `metadata.agent.profile` defaults from `adapter` and controls profile-specific behavior such as which tools the agent registers, project-memory filenames, CLI/ACP command selection, and native session routing. Valid values are `anthropic`, `claude-5`, `openai`, `gemini`, `kimi`, `gpt56`, and `gpt6`; model-level values override provider-level values. diff --git a/docs/public/integrations/openrouter.mdx b/docs/public/integrations/openrouter.mdx index 1fb8c6f1c..445667355 100644 --- a/docs/public/integrations/openrouter.mdx +++ b/docs/public/integrations/openrouter.mdx @@ -127,7 +127,7 @@ OpenRouter's [provider routing preferences](https://openrouter.ai/docs/guides/ro ## Attribution headers -Fabro does not send OpenRouter's optional attribution headers (`HTTP-Referer`, `X-Title`) by default, so self-hosted installations stay anonymous on OpenRouter's public app leaderboard. Workflow runs do send `x-session-id: ` for request grouping; an explicit provider `default_headers` value for that header takes precedence. To opt in to attribution: +Fabro does not send OpenRouter's optional attribution headers (`HTTP-Referer`, `X-Title`) by default, so self-hosted installations stay anonymous on OpenRouter's public app leaderboard. Native HTTP model requests in workflow runs do send `x-session-id: ` for request grouping; an explicit provider `default_headers` value for that header takes precedence. To opt in to attribution: ```toml title="settings.toml" [llm.providers.openrouter.default_headers] diff --git a/lib/apps/fabro-cli/src/commands/run/petri_worker.rs b/lib/apps/fabro-cli/src/commands/run/petri_worker.rs index 267c18070..4704a5032 100644 --- a/lib/apps/fabro-cli/src/commands/run/petri_worker.rs +++ b/lib/apps/fabro-cli/src/commands/run/petri_worker.rs @@ -181,6 +181,7 @@ pub(super) async fn execute(worker: PetriWorker<'_>) -> Result<()> { command_context::load_cli_catalog().context("failed to build worker LLM catalog")?; let mut runtime = runtime_spec( catalog.clone(), + run_id, &vault, &worker.run_state, worker.fabro_home.clone(), @@ -652,6 +653,7 @@ fn run_tool_services(worker: &PetriWorker<'_>) -> Option { /// home the server named, and the run tools when the run has them. async fn runtime_spec( catalog: Catalog, + run_id: RunId, vault: &Arc>, run_state: &RunProjection, fabro_home: Option, @@ -663,13 +665,14 @@ async fn runtime_spec( for (provider, issue) in &ready.issues { warn!(provider = %provider, error = %issue, "model provider credentials unusable"); } - let model_client = match runtime::model_client(catalog, credentials, None, &ready.ready) { - Ok(client) => client, - Err(err) => { - warn!(error = %err, "Petri model client unavailable; LLM nodes run without one"); - None - } - }; + let model_client = + match runtime::model_client(catalog, credentials, None, &ready.ready, Some(run_id)) { + Ok(client) => client, + Err(err) => { + warn!(error = %err, "Petri model client unavailable; LLM nodes run without one"); + None + } + }; let daytona = vault.read().await.get(EnvVars::DAYTONA_API_KEY).map(|key| { // The same shared client the server attaches, so the worker's // Daytona calls take the server's proxy and CA policy. diff --git a/lib/apps/fabro-cli/src/commands/run/publish.rs b/lib/apps/fabro-cli/src/commands/run/publish.rs index 9832bb55b..c140787a8 100644 --- a/lib/apps/fabro-cli/src/commands/run/publish.rs +++ b/lib/apps/fabro-cli/src/commands/run/publish.rs @@ -26,6 +26,7 @@ use std::time::Duration; use anyhow::{Context, Result}; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; +use fabro_auth::ExtraHeadersCredentialSource; use fabro_config::ServerSettingsBuilder; use fabro_github::token_source::{InstallationTokenSource, SecretString}; use fabro_github::{GitHubContext, GitHubCredentials}; @@ -210,7 +211,7 @@ impl GitHubPublisher { .pull_request .clone() .filter(|pull_request| pull_request.enabled), - llm_source, + llm_source: Arc::new(ExtraHeadersCredentialSource::for_run(llm_source, run_id)), catalog, records, client, @@ -368,6 +369,7 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; use chrono::Utc; + use fabro_auth::test_support as auth_test_support; use fabro_github::InstallationToken; use fabro_github::test_support::{InstallationTokenMinter, installation_token_source}; use fabro_llm::credentials::NoCredentials; @@ -420,6 +422,52 @@ mod tests { assert!(!publishes(&empty)); } + #[tokio::test] + async fn publication_model_calls_carry_the_run_id() { + let server = httpmock::MockServer::start_async().await; + let run_id = RunId::new(); + let request = server.mock(|when, then| { + when.method(httpmock::Method::POST) + .path("/v1/responses") + .header("authorization", "Bearer fixture-key") + .header("x-session-id", run_id.to_string()); + then.status(200).json_body(serde_json::json!({ + "id": "resp_1", "status": "completed", "model": "gpt-5.4", + "output": [{"type":"message", "id":"msg_1", "role":"assistant", "status":"completed", + "content":[{"type":"output_text", "text": r#"{"title":"Update widgets","body":"Updated widgets."}"#, "annotations":[]}]}], + "usage":{"input_tokens":10,"output_tokens":2,"total_tokens":12} + })); + }); + let publisher = GitHubPublisher::for_run( + run_id, + &spec(), + None, + auth_test_support::env_credential_source(|name| { + (name == "OPENAI_API_KEY").then(|| "fixture-key".to_string()) + }), + Arc::new(test_support::test_catalog_with_provider_base_url( + "openai", + &server.url("/v1"), + )), + Arc::new(MemoryPlatformRecords::new()), + fabro_client::Client::new_no_proxy("http://127.0.0.1:9").unwrap(), + ) + .unwrap(); + let content = pull_request::build_pr_content( + "+change", + "Update widgets", + "openai/gpt-5.4", + publisher.llm_source, + publisher.catalog, + None, + None, + ) + .await + .unwrap(); + assert_eq!(content.title, "Update widgets"); + request.assert_calls(1); + } + /// Mints `token-` for its n-th mint, valid for an hour. #[derive(Default)] struct CountingMinter(AtomicUsize); diff --git a/lib/apps/fabro-server/src/diagnostics.rs b/lib/apps/fabro-server/src/diagnostics.rs index d9f5111a2..bf1049c80 100644 --- a/lib/apps/fabro-server/src/diagnostics.rs +++ b/lib/apps/fabro-server/src/diagnostics.rs @@ -208,7 +208,7 @@ pub(crate) async fn test_llm_providers(state: &AppState) -> anyhow::Result, model: &str, ) -> (types::PreflightResponse, bool) { - let (_, ready_providers) = state.resolve_llm_client_with_ready_ids().await; + let (_, ready_providers) = state.resolve_llm_client_with_ready_ids(None).await; let mut manifest = minimal_manifest(); manifest.workflows.get_mut("workflow.fabro").unwrap().source = format!( r#" @@ -2868,7 +2870,7 @@ digraph Demo { &manifest, ) .unwrap(); - let (_, ready_providers) = state.resolve_llm_client_with_ready_ids().await; + let (_, ready_providers) = state.resolve_llm_client_with_ready_ids(None).await; assert!(ready_providers.is_empty()); // With no provider ready there is no model client to resolve the diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 27164f760..735f05403 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -47,7 +47,7 @@ pub use fabro_api::types::{ UpdateVariableRequest, UsageByModel, UsageStageRef, VariableListResponse, VncPreviewResponse, WriteBlobResponse, }; -use fabro_auth::SqlVaultCredentialSource; +use fabro_auth::{ExtraHeadersCredentialSource, SqlVaultCredentialSource}; use fabro_automation::{self, AutomationStore}; use fabro_config::daemon::ServerDaemon; use fabro_config::{LlmLayer, RunLayer, Storage, WorkflowSettingsBuilder}; @@ -1479,13 +1479,18 @@ impl AppState { Some(format!("{}/runs/{run_id}", base.trim_end_matches('/'))) } - pub(crate) async fn resolve_llm_client(&self) -> anyhow::Result { - resolve_llm_client_from_source( - Arc::clone(&self.llm_source), - self.catalog(), - self.http_client.clone(), - ) - .await + pub(crate) async fn resolve_llm_client( + &self, + run_id: Option, + ) -> anyhow::Result { + let source = match run_id { + Some(run_id) => Arc::new(ExtraHeadersCredentialSource::for_run( + Arc::clone(&self.llm_source), + run_id, + )) as Arc, + None => Arc::clone(&self.llm_source), + }; + resolve_llm_client_from_source(source, self.catalog(), self.http_client.clone()).await } pub(crate) async fn configured_llm_provider_ids(&self) -> Vec { @@ -1499,8 +1504,9 @@ impl AppState { /// resolved twice. pub(crate) async fn resolve_llm_client_with_ready_ids( &self, + run_id: Option, ) -> (anyhow::Result, Vec) { - let llm_result = self.resolve_llm_client().await; + let llm_result = self.resolve_llm_client(run_id).await; if let Err(err) = &llm_result { warn!(error = ?err, "Failed to resolve LLM client while checking ready providers"); } @@ -1512,7 +1518,7 @@ impl AppState { } pub(crate) async fn ready_llm_provider_ids(&self) -> Vec { - self.resolve_llm_client_with_ready_ids().await.1 + self.resolve_llm_client_with_ready_ids(None).await.1 } pub(crate) async fn decorate_run_summary(&self, run: fabro_types::Run) -> fabro_types::Run { diff --git a/lib/apps/fabro-server/src/server/handler/completions.rs b/lib/apps/fabro-server/src/server/handler/completions.rs index eb701d215..244697e1e 100644 --- a/lib/apps/fabro-server/src/server/handler/completions.rs +++ b/lib/apps/fabro-server/src/server/handler/completions.rs @@ -21,7 +21,7 @@ async fn create_completion( Json(req): Json, ) -> Response { let catalog = state.catalog(); - let llm_result = match state.resolve_llm_client().await { + let llm_result = match state.resolve_llm_client(None).await { Ok(result) => result, Err(err) => { error!(error = ?err, "Failed to create LLM client"); diff --git a/lib/apps/fabro-server/src/server/handler/graph.rs b/lib/apps/fabro-server/src/server/handler/graph.rs index 9208ec7b2..5fbb03601 100644 --- a/lib/apps/fabro-server/src/server/handler/graph.rs +++ b/lib/apps/fabro-server/src/server/handler/graph.rs @@ -59,7 +59,7 @@ async fn render_graph_from_manifest( .into_response(); } }; - let (_, ready_providers) = state.resolve_llm_client_with_ready_ids().await; + let (_, ready_providers) = state.resolve_llm_client_with_ready_ids(None).await; let check = match run_manifest::check_prepared_manifest( &state, &prepared, diff --git a/lib/apps/fabro-server/src/server/handler/models.rs b/lib/apps/fabro-server/src/server/handler/models.rs index 02c5f92b8..845f20328 100644 --- a/lib/apps/fabro-server/src/server/handler/models.rs +++ b/lib/apps/fabro-server/src/server/handler/models.rs @@ -217,7 +217,7 @@ async fn test_model( }, None => None, }; - let llm_result = match state.resolve_llm_client().await { + let llm_result = match state.resolve_llm_client(None).await { Ok(result) => result, Err(err) => { error!(error = ?err, "Failed to resolve LLM client"); diff --git a/lib/apps/fabro-server/src/server/handler/runs.rs b/lib/apps/fabro-server/src/server/handler/runs.rs index 696c8af1e..9d6f818da 100644 --- a/lib/apps/fabro-server/src/server/handler/runs.rs +++ b/lib/apps/fabro-server/src/server/handler/runs.rs @@ -699,7 +699,7 @@ pub(crate) async fn create_run_from_intent( return response; } let prepared = prepared.with_web_url(state.run_web_url(&run_id)); - finalize_created_run(state, prepared, explicit_title_supplied, entrypoint).await + finalize_created_run(state, prepared, run_id, explicit_title_supplied, entrypoint).await } /// A run must not be its own parent; an explicit parent must pass @@ -724,6 +724,7 @@ async fn validate_optional_parent( async fn finalize_created_run( state: Arc, prepared: run_compiler::PreparedRun, + run_id: RunId, explicit_title_supplied: bool, title_generation_target: ManifestPath, ) -> Response { @@ -731,7 +732,8 @@ async fn finalize_created_run( // and ask-fabro-readiness) and the LLM client itself (for the spawned // title-generation task). `ready_llm_provider_ids` would otherwise call // `resolve_llm_client` a second time and discard the client. - let (llm_result, ready_provider_ids) = state.resolve_llm_client_with_ready_ids().await; + let (llm_result, ready_provider_ids) = + state.resolve_llm_client_with_ready_ids(Some(run_id)).await; let llm_client_for_title = llm_result.ok(); let run_materialization_provider_ids = { #[cfg(any(test, feature = "test-support"))] @@ -1211,7 +1213,7 @@ async fn run_preflight( return ApiError::bad_request(format!("Run config variable interpolation failed: {err}")) .into_response(); } - let (_, ready_providers) = state.resolve_llm_client_with_ready_ids().await; + let (_, ready_providers) = state.resolve_llm_client_with_ready_ids(None).await; let check = match run_manifest::check_prepared_manifest( &state, &prepared, @@ -1264,7 +1266,7 @@ async fn validate_run_manifest( return ApiError::bad_request(format!("Run config variable interpolation failed: {err}")) .into_response(); } - let (_, ready_providers) = state.resolve_llm_client_with_ready_ids().await; + let (_, ready_providers) = state.resolve_llm_client_with_ready_ids(None).await; let check = match run_manifest::check_prepared_manifest( &state, &prepared, diff --git a/lib/apps/fabro-server/src/server/handler/sessions.rs b/lib/apps/fabro-server/src/server/handler/sessions.rs index 6383cbc0b..bd1055b46 100644 --- a/lib/apps/fabro-server/src/server/handler/sessions.rs +++ b/lib/apps/fabro-server/src/server/handler/sessions.rs @@ -157,7 +157,7 @@ async fn create_run_session( if let Err(response) = ensure_run(&state, run_id).await { return response; } - let llm_result = match state.resolve_llm_client().await { + let llm_result = match state.resolve_llm_client(Some(run_id)).await { Ok(result) => result, Err(err) => { return ApiError::new( @@ -697,9 +697,14 @@ async fn build_agent( session: &ProjectedRunSession, ) -> Result { let catalog = state.catalog(); - let llm_result = state.resolve_llm_client().await.map_err(|err| { - AskFabroBuildError::LlmUnconfigured(format!("LLM credentials are not configured: {err}")) - })?; + let llm_result = state + .resolve_llm_client(Some(run_id)) + .await + .map_err(|err| { + AskFabroBuildError::LlmUnconfigured(format!( + "LLM credentials are not configured: {err}" + )) + })?; for (provider, issue) in &llm_result.auth_issues { warn!(provider = %provider, error = %issue, "LLM provider unavailable due to auth issue"); } diff --git a/lib/apps/fabro-server/src/server/petri_runs.rs b/lib/apps/fabro-server/src/server/petri_runs.rs index 7a09f4743..4d43a7a3e 100644 --- a/lib/apps/fabro-server/src/server/petri_runs.rs +++ b/lib/apps/fabro-server/src/server/petri_runs.rs @@ -88,6 +88,7 @@ pub(crate) fn runtime_spec( eligible: &[ProviderId], dry_run: bool, sandbox: SandboxProviderConfig, + run_id: Option, ) -> RuntimeSpec { let settings_toml = settings_layer_toml(state); let mcp_catalog_toml = mcp_catalog_toml(&state.mcp_server_store().catalog_settings()); @@ -97,6 +98,7 @@ pub(crate) fn runtime_spec( Arc::clone(&state.llm_source), state.http_client.clone(), eligible, + run_id, ) { Ok(client) => client, Err(err) => { @@ -294,6 +296,7 @@ pub(crate) async fn admit( eligible, dry_run, state.sandbox_provider_config(None), + None, ), false, ) @@ -469,7 +472,7 @@ pub(crate) async fn execute(state: Arc, run_id: RunId) { }; let petri_interviewer = FabroInterviewer::new(interviewer, approval); let observers = vec![petri_interviewer.observer()]; - let (_, eligible) = state.resolve_llm_client_with_ready_ids().await; + let (_, eligible) = state.resolve_llm_client_with_ready_ids(None).await; let dry_run = run_state.spec.settings.run.execution.mode == RunMode::DryRun; // The in-process path serves the server's tests: a Git target is // fetched without a credential, and nothing is published. @@ -491,6 +494,7 @@ pub(crate) async fn execute(state: Arc, run_id: RunId) { &eligible, dry_run, state.sandbox_provider_config(vault.get(EnvVars::DAYTONA_API_KEY).map(str::to_owned)), + Some(run_id), ); let request = RunRequest { run_id: run_id.to_string(), diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index a1f978590..3f17e627b 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -1721,7 +1721,7 @@ async fn resolve_llm_client_reads_openai_token_from_vault() { .await .unwrap(); - let llm_result = state.resolve_llm_client().await.unwrap(); + let llm_result = state.resolve_llm_client(None).await.unwrap(); assert_eq!(llm_result.provider_ids(), vec![ lithos_llm::catalog::builtin::openai() @@ -1738,7 +1738,7 @@ async fn resolve_llm_client_ignores_env_lookup_provider_tokens() { |name| (name == EnvVars::OPENAI_API_KEY).then(|| "env-openai-key".to_string()), ); - let llm_result = state.resolve_llm_client().await.unwrap(); + let llm_result = state.resolve_llm_client(None).await.unwrap(); assert!( llm_result.provider_ids().is_empty(), @@ -1809,7 +1809,8 @@ async fn resolve_llm_client_uses_vault_key_without_env_lookup_openai_settings() .mock_async(|when, then| { when.method(POST) .path("/v1/responses") - .header("authorization", "Bearer vault-openai-key"); + .header("authorization", "Bearer vault-openai-key") + .header_missing("x-session-id"); then.status(200) .header("content-type", "application/json") .json_body(openai_responses_payload("hello from vault key")); @@ -1836,7 +1837,7 @@ async fn resolve_llm_client_uses_vault_key_without_env_lookup_openai_settings() .await .unwrap(); - let llm_result = state.resolve_llm_client().await.unwrap(); + let llm_result = state.resolve_llm_client(None).await.unwrap(); let response = llm_result .client .complete( @@ -3091,7 +3092,19 @@ url = "http://127.0.0.1:32276" #[tokio::test] async fn create_run_without_explicit_title_returns_deterministic_then_updates_generated_title() { let llm = MockServer::start_async().await; - let title_mock = mock_openai_title_response(&llm, "Generated deploy title", None).await; + let captured = Arc::new(std::sync::Mutex::new(Vec::new())); + let headers = Arc::clone(&captured); + let title_mock = llm.mock(|when, then| { + when.method(POST).path("/v1/responses"); + then.respond_with(move |request| { + *headers.lock().unwrap() = request.headers_vec().clone(); + httpmock::HttpMockResponse::builder() + .status(200) + .header("content-type", "application/json") + .body(openai_responses_payload(r#"{"title":"Generated deploy title"}"#).to_string()) + .build() + }); + }); let state = TestAppStateBuilder::new() .provider_base_url("openai", llm.url("/v1")) .env_lookup(|_| None) @@ -3111,6 +3124,13 @@ async fn create_run_without_explicit_title_returns_deterministic_then_updates_ge wait_for_run_title(&state, run_id, "Generated deploy title").await; assert_eq!(title_update_event_count(&state, run_id).await, 1); title_mock.assert_async().await; + let headers = captured.lock().unwrap(); + let sessions: Vec<_> = headers + .iter() + .filter(|(name, _)| name.eq_ignore_ascii_case("x-session-id")) + .map(|(_, value)| value.as_str()) + .collect(); + assert_eq!(sessions, [run_id.to_string()]); } #[tokio::test] @@ -5844,7 +5864,8 @@ async fn test_providers_successful_probe_returns_probe_model() { .mock_async(|when, then| { when.method(POST) .path("/v1/responses") - .header("authorization", "Bearer vault-openai-key"); + .header("authorization", "Bearer vault-openai-key") + .header_missing("x-session-id"); then.status(200) .header("content-type", "application/json") .json_body(openai_responses_payload("OK")); diff --git a/lib/components/fabro-petri/Cargo.toml b/lib/components/fabro-petri/Cargo.toml index 628b5503a..1e8c7048d 100644 --- a/lib/components/fabro-petri/Cargo.toml +++ b/lib/components/fabro-petri/Cargo.toml @@ -19,6 +19,7 @@ workspace = true test-support = ["dep:petri_testkit", "dep:tempfile"] [dependencies] +fabro-auth = { path = "../../foundation/fabro-auth" } fabro-github = { path = "../fabro-github" } percent-encoding.workspace = true smol_str.workspace = true diff --git a/lib/components/fabro-petri/src/runtime.rs b/lib/components/fabro-petri/src/runtime.rs index 0cb377727..aa6c9b126 100644 --- a/lib/components/fabro-petri/src/runtime.rs +++ b/lib/components/fabro-petri/src/runtime.rs @@ -16,7 +16,9 @@ use std::path::PathBuf; use std::sync::Arc; +use fabro_auth::ExtraHeadersCredentialSource; use fabro_http::HttpClient; +use fabro_types::RunId; use fabro_workflow::services::FabroRunToolServices; use lithos_llm::Client; use lithos_llm::catalog::{Catalog, ProviderId}; @@ -127,16 +129,24 @@ impl AdmissionPass for LocalDryRun { /// ready, the same eligible set the legacy compiler pinned models against. /// `None` when no provider is eligible, so Petri's admission pass leaves the /// graph alone rather than refusing every model. +/// Execution supplies the durable run ID for HTTP session grouping; admission +/// supplies `None` because it has no run-owned model calls. pub fn model_client( catalog: Catalog, credentials: Arc, http: Option, eligible: &[ProviderId], + run_id: Option, ) -> Result, ClientBuildError> { if eligible.is_empty() { debug!("no eligible model provider; the Petri runtime gets no model client"); return Ok(None); } + let credentials = match run_id { + Some(run_id) => Arc::new(ExtraHeadersCredentialSource::for_run(credentials, run_id)) + as Arc, + None => credentials, + }; let mut builder = Client::builder() .catalog(catalog) .credentials_arc(credentials) diff --git a/lib/components/fabro-petri/tests/check.rs b/lib/components/fabro-petri/tests/check.rs index f990bb57a..ddd1cec80 100644 --- a/lib/components/fabro-petri/tests/check.rs +++ b/lib/components/fabro-petri/tests/check.rs @@ -83,9 +83,13 @@ fn runtime_with_openai() -> RuntimeSpec { "OPENAI_API_KEY" => Some("test-key".to_string()), _ => None, }); - let client = runtime::model_client(test_catalog(), credentials, None, &[ProviderId::new( - "openai", - )]) + let client = runtime::model_client( + test_catalog(), + credentials, + None, + &[ProviderId::new("openai")], + None, + ) .expect("the model client builds") .expect("openai is eligible"); RuntimeSpec { diff --git a/lib/components/fabro-petri/tests/hooks.rs b/lib/components/fabro-petri/tests/hooks.rs index 50c758f6d..59e55b1e4 100644 --- a/lib/components/fabro-petri/tests/hooks.rs +++ b/lib/components/fabro-petri/tests/hooks.rs @@ -830,6 +830,7 @@ async fn a_run_hook_blocks_a_tool_effect_through_the_forwarded_service() { credentials, None, &[ProviderId::new("openai")], + None, ) .expect("the model client builds") .expect("openai is eligible"); diff --git a/lib/components/fabro-petri/tests/model.rs b/lib/components/fabro-petri/tests/model.rs index c4dc064f1..790786fd8 100644 --- a/lib/components/fabro-petri/tests/model.rs +++ b/lib/components/fabro-petri/tests/model.rs @@ -49,9 +49,15 @@ async fn a_model_call_authenticates_through_the_vault_and_skills_read_the_home() AsyncRwLock::new(vault), ))); let catalog = test_catalog_with_provider_base_url("openai", &twin.base_url); - let client = runtime::model_client(catalog, credentials, None, &[ProviderId::new("openai")]) - .expect("the model client builds") - .expect("openai is eligible"); + let client = runtime::model_client( + catalog, + credentials, + None, + &[ProviderId::new("openai")], + None, + ) + .expect("the model client builds") + .expect("openai is eligible"); let runtime = RuntimeSpec { model_client: Some(client), fabro_home: Some(home.clone()), diff --git a/lib/components/fabro-petri/tests/resume.rs b/lib/components/fabro-petri/tests/resume.rs index bf607dfed..abcccd0bf 100644 --- a/lib/components/fabro-petri/tests/resume.rs +++ b/lib/components/fabro-petri/tests/resume.rs @@ -5,7 +5,7 @@ mod support; use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use fabro_petri::admission::AdmittedGraphs; use fabro_petri::check::Launch; @@ -69,8 +69,9 @@ async fn a_failed_store_write_interrupts_the_run_and_a_resume_finishes_it() { let memory = Arc::new(MemoryRunStore::new()); let failing = Arc::new(FailingStore { inner: Arc::clone(&memory), - log: LogId::Resources, + log: Some(LogId::Resources), appended: Arc::new(AtomicUsize::new(0)), + armed: Arc::new(AtomicBool::new(true)), }); let runtime = RuntimeSpec::default(); let graphs = || { @@ -152,8 +153,9 @@ fn resume_request( /// would, before anything is stored. struct FailingStore { inner: Arc, - log: LogId, + log: Option, appended: Arc, + armed: Arc, } #[async_trait::async_trait] @@ -163,14 +165,16 @@ impl RunStore for FailingStore { inner: self.inner.open(key, access).await?, log: self.log, appended: Arc::clone(&self.appended), + armed: Arc::clone(&self.armed), })) } } struct FailingLogs { inner: Arc, - log: LogId, + log: Option, appended: Arc, + armed: Arc, } #[async_trait::async_trait] @@ -180,7 +184,10 @@ impl RunLogs for FailingLogs { } async fn append(&self, log: &LogId, records: &[Record]) -> Result<(), StoreError> { - if *log == self.log && self.appended.fetch_add(1, Ordering::SeqCst) == 0 { + if self.log.is_none_or(|target| *log == target) + && self.armed.load(Ordering::SeqCst) + && self.appended.fetch_add(1, Ordering::SeqCst) == 0 + { return Err(StoreError::backend( self.locator(), "append", @@ -202,3 +209,97 @@ impl RunLogs for FailingLogs { self.inner.get_blob(digest).await } } + +/// Fail a durable write after a model request, then resume with a rebuilt +/// runtime. Every request on either side of that interruption belongs to +/// the same run, even if Petri must repeat an uncommitted model call. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn resumed_model_calls_keep_the_original_run_session_header() { + use fabro_auth::test_support::env_credential_source; + use fabro_llm::test_support::test_catalog_with_provider_base_url; + use fabro_petri::runtime; + use fabro_types::RunId; + use httpmock::{HttpMockResponse, MockServer}; + use lithos_llm::catalog::ProviderId; + use serde_json::json; + + let root = tempfile::tempdir().unwrap(); + let server = MockServer::start_async().await; + let run_id = RunId::new(); + let armed = Arc::new(AtomicBool::new(false)); + let arm_after_request = Arc::clone(&armed); + let mock = server.mock(|when, then| { + when.path("/v1/responses") + .header("x-session-id", run_id.to_string()); + then.respond_with(move |_| { + arm_after_request.store(true, Ordering::SeqCst); + HttpMockResponse::builder().status(200).body(json!({ + "id":"resp_1", "status":"completed", "model":"gpt-5.4", + "output":[{"type":"message", "id":"msg_1", "role":"assistant", "status":"completed", + "content":[{"type":"output_text", "text":"Done", "annotations":[]}]}], + "usage":{"input_tokens":10,"output_tokens":2,"total_tokens":12} + }).to_string()).build() + }); + }); + let runtime = || RuntimeSpec { + model_client: runtime::model_client( + test_catalog_with_provider_base_url("openai", &server.url("/v1")), + env_credential_source(|name| { + (name == "OPENAI_API_KEY").then(|| "fixture-key".to_string()) + }), + Some(fabro_test::test_http_client()), + &[ProviderId::new("openai")], + Some(run_id), + ) + .unwrap(), + ..RuntimeSpec::default() + }; + let workflow = r#"digraph Resume { + graph [backend="api", default_max_retries=0] + start [shape=Mdiamond] + first [shape=tab, prompt="Say Done", model="gpt-5.4", provider="openai"] + second [shape=tab, prompt="Say Done again", model="gpt-5.4", provider="openai"] + exit [shape=Msquare] + start -> first -> second -> exit + }"#; + let graphs = || { + admit( + &[("workflow.fabro", workflow), ("workflow.toml", SETTINGS)], + Launch::default(), + &runtime(), + ) + }; + let memory = Arc::new(MemoryRunStore::new()); + let failing = Arc::new(FailingStore { + inner: Arc::clone(&memory), + log: None, + appended: Arc::new(AtomicUsize::new(0)), + armed, + }); + let first = engine::run(run_request( + &run_id.to_string(), + root.path(), + graphs(), + failing, + runtime(), + no_questions(Arc::new(Silent)), + )) + .await; + assert!(matches!(first, Err(RunError::StoreFailed(_))), "{first:?}"); + let before_resume = mock.calls(); + assert!(before_resume > 0); + let outcome = engine::run(resume_request( + &run_id.to_string(), + root.path(), + graphs(), + memory, + runtime(), + )) + .await + .unwrap(); + assert_eq!(outcome.status, RunStatus::Success, "{outcome:?}"); + assert!( + mock.calls() > before_resume, + "the rebuilt runtime made a model request" + ); +} diff --git a/lib/components/fabro-petri/tests/session_headers.rs b/lib/components/fabro-petri/tests/session_headers.rs new file mode 100644 index 000000000..4aced6775 --- /dev/null +++ b/lib/components/fabro-petri/tests/session_headers.rs @@ -0,0 +1,429 @@ +//! Recording HTTP providers exercise the actual run client and native steps. +mod support; + +use std::collections::HashMap; +use std::fmt::Write as _; +use std::sync::Arc; + +use fabro_auth::VaultCredentialSource; +use fabro_llm::test_support; +use fabro_petri::check::Launch; +use fabro_petri::engine::{self, RunStatus}; +use fabro_petri::runtime::{self, RuntimeSpec}; +use fabro_types::{RunId, SecretType}; +use fabro_vault::Vault; +use httpmock::{Method, MockServer}; +use lithos_llm::catalog::{Catalog, ProviderId}; +use lithos_llm::types::Request; +use petri_store::MemoryRunStore; +use serde_json::{Value, json}; +use tokio::sync::RwLock; + +fn credentials() -> Arc { + let mut vault = Vault::from_entries(HashMap::new()); + for name in [ + "OPENAI_API_KEY", + "OPENROUTER_API_KEY", + "ANTHROPIC_API_KEY", + "CUSTOM_API_KEY", + "SESSION", + ] { + vault + .set(name, "fixture-value", SecretType::Token, None) + .expect("the detached vault takes fixture secrets"); + } + Arc::new(VaultCredentialSource::vault_only(Arc::new(RwLock::new( + vault, + )))) +} + +fn catalog(server: &MockServer, provider: &str, header: Option<&str>) -> Catalog { + let custom = if provider == "custom" { + "display_name = 'Custom'\ncodecs = ['openai-chat']\nauth = { type = 'bearer' }\nallow_passthrough = true\nmetadata = { agent = { profile = 'openai' } }\n" + } else { + "" + }; + let header = header + .map(|value| format!("default_headers = {{ 'X-SeSsIoN-Id' = '{value}' }}\n")) + .unwrap_or_default(); + test_support::test_catalog_with_overlay(&format!( + "[providers.{provider}]\nbase_url = '{}'\n{custom}{header}", + server.url("/v1") + )) +} + +fn model(provider: &str) -> &'static str { + match provider { + "openai" => "gpt-5.4", + "anthropic" => "claude-sonnet-4-6", + "openrouter" => "openai/gpt-5.4", + _ => "fixture-model", + } +} + +fn path(provider: &str) -> &'static str { + match provider { + "openai" => "/v1/responses", + "anthropic" => "/v1/messages", + _ => "/v1/chat/completions", + } +} + +fn response(provider: &str) -> Value { + match provider { + "openai" => { + json!({"id":"resp_1","object":"response","status":"completed","model":model(provider), + "output":[{"type":"message","id":"msg_1","role":"assistant","status":"completed", + "content":[{"type":"output_text","text":"Done","annotations":[]}]}], + "usage":{"input_tokens":10,"output_tokens":2,"total_tokens":12}}) + } + "anthropic" => { + json!({"id":"msg_1","type":"message","role":"assistant","model":model(provider), + "content":[{"type":"text","text":"Done"}],"stop_reason":"end_turn","stop_sequence":null, + "usage":{"input_tokens":10,"output_tokens":2}}) + } + _ => json!({"id":"chat_1","object":"chat.completion","created":1,"model":model(provider), + "choices":[{"index":0,"message":{"role":"assistant","content":"Done"},"finish_reason":"stop"}], + "usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}), + } +} + +fn stream(provider: &str) -> String { + match provider { + "openai" => format!("event: response.completed\ndata: {}\n\n", json!({"type":"response.completed","response":response(provider)})), + "anthropic" => [ + json!({"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":model(provider),"content":[],"usage":{"input_tokens":10,"output_tokens":0}}}), + json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}), + json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Done"}}), + json!({"type":"content_block_stop","index":0}), + json!({"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":2}}), + json!({"type":"message_stop"}), + ].iter().fold(String::new(), |mut output, event| { + write!(output, "event: {}\ndata: {event}\n\n", event["type"].as_str().expect("fixture event type")).expect("writing to a string"); + output + }), + _ => format!("data: {}\n\ndata: [DONE]\n\n", json!({"id":"chat_1","object":"chat.completion.chunk","created":1,"model":model(provider),"choices":[{"index":0,"delta":{"role":"assistant","content":"Done"},"finish_reason":"stop"}]})), + } +} + +fn client( + server: &MockServer, + provider: &str, + run_id: Option, + header: Option<&str>, +) -> lithos_llm::Client { + runtime::model_client( + catalog(server, provider, header), + credentials(), + Some(fabro_test::test_http_client()), + &[ProviderId::new(provider)], + run_id, + ) + .expect("the model client builds") + .expect("the provider is eligible") +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn native_agents_and_prompts_send_the_run_id_across_http_providers() { + for provider in ["openai", "openrouter", "anthropic", "custom"] { + let server = MockServer::start_async().await; + let run_id = RunId::new(); + let session = run_id.to_string(); + let prompt = server.mock(|when, then| { + when.method(Method::POST) + .path(path(provider)) + .header("x-session-id", &session) + .header( + if provider == "anthropic" { + "x-api-key" + } else { + "authorization" + }, + if provider == "anthropic" { + "fixture-value" + } else { + "Bearer fixture-value" + }, + ) + .json_body_excludes(json!({"stream":true}).to_string()); + then.status(200).json_body(response(provider)); + }); + let agent = server.mock(|when, then| { + when.method(Method::POST) + .path(path(provider)) + .header("x-session-id", &session) + .header( + if provider == "anthropic" { + "x-api-key" + } else { + "authorization" + }, + if provider == "anthropic" { + "fixture-value" + } else { + "Bearer fixture-value" + }, + ) + .json_body_includes(json!({"stream":true}).to_string()); + then.status(200) + .header("content-type", "text/event-stream") + .body(stream(provider)); + }); + let runtime = RuntimeSpec { + model_client: Some(client(&server, provider, Some(run_id), None)), + ..RuntimeSpec::default() + }; + let workflow = format!( + r#"digraph Headers {{ + graph [backend="api", default_max_retries=0] + start [shape=Mdiamond] + agent [shape=box, prompt="Say Done", model="{}", provider="{provider}"] + prompt [shape=tab, prompt="Say Done", model="{}", provider="{provider}"] + exit [shape=Msquare] + start -> agent -> prompt -> exit + }}"#, + model(provider), + model(provider) + ); + let graphs = support::admit( + &[ + ("workflow.fabro", &workflow), + ("workflow.toml", support::SETTINGS), + ], + Launch::default(), + &runtime, + ); + let root = tempfile::tempdir().unwrap(); + let request = support::run_request( + &session, + root.path(), + graphs, + Arc::new(MemoryRunStore::new()), + runtime, + support::no_questions(Arc::new(support::Silent)), + ); + let outcome = engine::run(request).await.unwrap(); + assert_eq!( + outcome.status, + RunStatus::Success, + "{provider}: {outcome:?}" + ); + prompt.assert_calls(1); + agent.assert_calls(1); + } +} + +#[tokio::test] +async fn explicit_headers_win_and_unscoped_clients_do_not_inherit_a_run() { + for provider in ["openai", "openrouter", "anthropic", "custom"] { + for configured in [ + None, + Some("operator-session"), + Some("{{ secrets.SESSION }}"), + ] { + let server = MockServer::start_async().await; + let run_id = RunId::new(); + let session = run_id.to_string(); + let expected = match configured { + None => session.as_str(), + Some("operator-session") => "operator-session", + _ => "fixture-value", + }; + let scoped = server.mock(|when, then| { + when.method(Method::POST) + .path(path(provider)) + .header("x-session-id", expected); + then.status(200).json_body(response(provider)); + }); + let request = || { + Request::builder() + .model(format!("{provider}/{}", model(provider))) + .user("Say Done") + .build() + .unwrap() + }; + client(&server, provider, Some(run_id), configured) + .complete(request()) + .await + .unwrap(); + scoped.assert_calls(1); + } + let server = MockServer::start_async().await; + let unscoped = server.mock(|when, then| { + when.method(Method::POST) + .path(path(provider)) + .header_missing("x-session-id"); + then.status(200).json_body(response(provider)); + }); + client(&server, provider, None, None) + .complete( + Request::builder() + .model(format!("{provider}/{}", model(provider))) + .user("Say Done") + .build() + .unwrap(), + ) + .await + .unwrap(); + unscoped.assert_calls(1); + } +} + +#[tokio::test] +async fn concurrent_runs_and_rebuilt_clients_keep_their_own_identity() { + let server = MockServer::start_async().await; + let first = RunId::new(); + let second = RunId::new(); + let one = server.mock(|when, then| { + when.header("x-session-id", first.to_string()); + then.status(200).json_body(response("openai")); + }); + let two = server.mock(|when, then| { + when.header("x-session-id", second.to_string()); + then.status(200).json_body(response("openai")); + }); + let request = || { + Request::builder() + .model("openai/gpt-5.4") + .user("Say Done") + .build() + .unwrap() + }; + let a = client(&server, "openai", Some(first), None); + let b = client(&server, "openai", Some(second), None); + let (a, b) = tokio::join!(a.complete(request()), b.complete(request())); + a.unwrap(); + b.unwrap(); + // Workers reconstruct their clients from the persisted run ID on resume. + client(&server, "openai", Some(first), None) + .complete(request()) + .await + .unwrap(); + one.assert_calls(2); + two.assert_calls(1); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn prompt_fallback_keeps_the_run_id_on_both_providers() { + let server = MockServer::start_async().await; + let run_id = RunId::new(); + let session = run_id.to_string(); + let primary = server.mock(|when, then| { + when.method(Method::POST) + .path(path("openai")) + .header("x-session-id", &session); + then.status(401).json_body( + json!({"error":{"type":"authentication_error","message":"fixture rejection"}}), + ); + }); + let fallback = server.mock(|when, then| { + when.method(Method::POST) + .path(path("anthropic")) + .header("x-session-id", &session); + then.status(200).json_body(response("anthropic")); + }); + let catalog = test_support::test_catalog_with_overlay(&format!( + "[providers.openai]\nbase_url = '{}'\n[providers.anthropic]\nbase_url = '{}'\n", + server.url("/v1"), + server.url("/v1") + )); + let client = runtime::model_client( + catalog, + credentials(), + Some(fabro_test::test_http_client()), + &[ProviderId::new("openai"), ProviderId::new("anthropic")], + Some(run_id), + ) + .unwrap() + .unwrap(); + let runtime = RuntimeSpec { + model_client: Some(client), + ..RuntimeSpec::default() + }; + let workflow = r#"digraph Fallback { + graph [backend="api", default_max_retries=0] + start [shape=Mdiamond] + prompt [shape=tab, prompt="Say Done", model="gpt-5.4", provider="openai"] + exit [shape=Msquare] + start -> prompt -> exit + }"#; + let settings = format!( + "{}\n[run.model.fallbacks]\n'gpt-5.4' = ['anthropic:claude-sonnet-4-6']\n", + support::SETTINGS + ); + let graphs = support::admit( + &[("workflow.fabro", workflow), ("workflow.toml", &settings)], + Launch::default(), + &runtime, + ); + let root = tempfile::tempdir().unwrap(); + let request = support::run_request( + &session, + root.path(), + graphs, + Arc::new(MemoryRunStore::new()), + runtime, + support::no_questions(Arc::new(support::Silent)), + ); + let outcome = engine::run(request).await.unwrap(); + assert_eq!(outcome.status, RunStatus::Success, "{outcome:?}"); + primary.assert_calls(1); + fallback.assert_calls(1); +} + +#[tokio::test] +async fn retried_http_calls_keep_the_run_id() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use fabro_auth::ExtraHeadersCredentialSource; + use fabro_llm::ClientOptions; + use httpmock::HttpMockResponse; + + let server = MockServer::start_async().await; + let run_id = RunId::new(); + let calls = Arc::new(AtomicUsize::new(0)); + let count = Arc::clone(&calls); + let mock = server.mock(|when, then| { + when.method(Method::POST) + .path(path("openai")) + .header("x-session-id", run_id.to_string()); + then.respond_with(move |_| { + if count.fetch_add(1, Ordering::SeqCst) == 0 { + HttpMockResponse::builder() + .status(503) + .body(r#"{"error":{"message":"try again","type":"server_error"}}"#) + .build() + } else { + HttpMockResponse::builder() + .status(200) + .body(response("openai").to_string()) + .build() + } + }); + }); + let credentials = Arc::new(ExtraHeadersCredentialSource::for_run(credentials(), run_id)); + let built = fabro_llm::build_client( + catalog(&server, "openai", None), + credentials, + ClientOptions { + http: Some(fabro_test::test_http_client()), + retry: Some(test_support::test_retry_policy()), + ..ClientOptions::default() + }, + ) + .await + .unwrap(); + let reply = built + .client + .complete( + Request::builder() + .model("openai/gpt-5.4") + .user("Say Done") + .build() + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(reply.text(), "Done"); + mock.assert_calls(2); +} diff --git a/lib/foundation/fabro-auth/src/extra_headers_source.rs b/lib/foundation/fabro-auth/src/extra_headers_source.rs index 8b358bdde..87cb7c18a 100644 --- a/lib/foundation/fabro-auth/src/extra_headers_source.rs +++ b/lib/foundation/fabro-auth/src/extra_headers_source.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use std::sync::Arc; use async_trait::async_trait; +use fabro_types::RunId; use lithos_llm::catalog::CatalogProvider; use lithos_llm::credentials::{ CredentialError, CredentialHeader, CredentialProvider, Credentials, SecretValue, @@ -11,14 +12,25 @@ use lithos_llm::credentials::{ /// to every HTTP credential it resolves. /// /// Headers already present on a credential (for example from explicit -/// provider configuration) are left untouched. AWS-signed credentials carry -/// no header list and pass through unchanged. +/// provider configuration) or on the provider catalog are left untouched. +/// AWS-signed credentials carry no header list and pass through unchanged. pub struct ExtraHeadersCredentialSource { inner: Arc, headers: HashMap, } impl ExtraHeadersCredentialSource { + /// Tag HTTP model calls with their owning run, including retries and + /// resume. Explicit provider headers still take precedence; AWS + /// credentials are unchanged. + #[must_use] + pub fn for_run(inner: Arc, run_id: RunId) -> Self { + Self::new( + inner, + HashMap::from([("x-session-id".to_string(), run_id.to_string())]), + ) + } + #[must_use] pub fn new(inner: Arc, headers: HashMap) -> Self { Self { inner, headers } @@ -34,10 +46,14 @@ impl CredentialProvider for ExtraHeadersCredentialSource { let mut credentials = self.inner.credentials(provider).await?; if let Credentials::Http(http) = &mut credentials { for (name, value) in &self.headers { - if http - .extra_headers - .iter() - .any(|existing| existing.name.eq_ignore_ascii_case(name)) + if provider + .default_headers() + .keys() + .any(|existing| existing.eq_ignore_ascii_case(name)) + || http + .extra_headers + .iter() + .any(|existing| existing.name.eq_ignore_ascii_case(name)) { continue; } @@ -151,4 +167,64 @@ mod tests { .await ); } + #[tokio::test] + async fn run_headers_preserve_aws_and_bedrock_credentials() { + use lithos_llm::credentials::StaticCredentials; + + let catalog = test_catalog(); + let run_id = RunId::new(); + for credentials in [ + Credentials::AwsDefaultChain { + region: Some("us-west-2".to_string()), + }, + Credentials::BedrockBearer(SecretValue::new("bedrock-key")), + ] { + let inner = StaticCredentials::new().with("bedrock", credentials.clone()); + let source = ExtraHeadersCredentialSource::for_run(Arc::new(inner), run_id); + assert_eq!( + source + .credentials(catalog.provider("bedrock").unwrap()) + .await + .unwrap(), + credentials + ); + } + } + #[tokio::test] + async fn run_headers_do_not_cache_credentials_or_hide_resolution_errors() { + use fabro_types::SecretType; + + use crate::VaultCredentialSource; + use crate::test_support::empty_vault; + + let catalog = test_catalog(); + let provider = catalog.provider("openai").unwrap(); + let vault = empty_vault(); + let inner: Arc = + Arc::new(VaultCredentialSource::vault_only(Arc::clone(&vault))); + let source = ExtraHeadersCredentialSource::for_run(Arc::clone(&inner), RunId::new()); + assert!(!source.is_configured(provider).await); + let expected = inner.credentials(provider).await.unwrap_err(); + let actual = source.credentials(provider).await.unwrap_err(); + assert_eq!( + std::mem::discriminant(&actual), + std::mem::discriminant(&expected) + ); + assert_eq!(actual.to_string(), expected.to_string()); + for key in ["initial-key", "refreshed-key"] { + vault + .write() + .await + .set("OPENAI_API_KEY", key, SecretType::Token, None) + .unwrap(); + let Credentials::Http(http) = source.credentials(provider).await.unwrap() else { + panic!("HTTP credentials") + }; + let HttpAuthentication::Bearer(actual) = http.auth else { + panic!("bearer authentication") + }; + assert_eq!(actual.expose_secret(), key); + } + assert!(source.is_configured(provider).await); + } }