scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-16 04:33:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Restore run-scoped model session headers

Browse source
This commit is contained in:
Scott Werner 2026-10-14 11:27:26 -04:00
parent 7c020a0d92
commit 331a936f56
22 changed files with 779 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs/public/core-concepts/models.mdx
View file

@ -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 `<PROVIDER>_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: <run-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: <run-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.

2
docs/public/integrations/openrouter.mdx
View file

@ -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: <run-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: <run-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]

17
lib/apps/fabro-cli/src/commands/run/petri_worker.rs
View file

@ -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<FabroRunToolServices> {
/// home the server named, and the run tools when the run has them.
async fn runtime_spec(
catalog: Catalog,
run_id: RunId,
vault: &Arc<AsyncRwLock<Vault>>,
run_state: &RunProjection,
fabro_home: Option<PathBuf>,
@ -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.

50
lib/apps/fabro-cli/src/commands/run/publish.rs
View file

@ -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-<n>` for its n-th mint, valid for an hour.
#[derive(Default)]
struct CountingMinter(AtomicUsize);

2
lib/apps/fabro-server/src/diagnostics.rs
View file

@ -208,7 +208,7 @@ pub(crate) async fn test_llm_providers(state: &AppState) -> anyhow::Result<Provi
// issues only arise for those providers, so this list is the complete
// population to probe.
let configured_providers = state.configured_llm_provider_ids().await;
let result = state.resolve_llm_client().await?;
let result = state.resolve_llm_client(None).await?;
let catalog = state.catalog();
let client = Arc::new(result.client);

6
lib/apps/fabro-server/src/run_manifest.rs
View file

@ -245,6 +245,7 @@ pub(crate) async fn check_prepared_manifest(
ready_providers,
dry_run,
state.sandbox_provider_config(None),
None,
);
let has_ready_provider = !ready_providers.is_empty();
let prepared = prepared.clone();
@ -1547,6 +1548,7 @@ mod tests {
ready_providers,
false,
state.sandbox_provider_config(None),
None,
);
validate_prepared_manifest(
prepared,
@ -1713,7 +1715,7 @@ enabled = true
state: &Arc<crate::server::AppState>,
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

26
lib/apps/fabro-server/src/server.rs
View file

@ -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<FabroClient> {
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<RunId>,
) -> anyhow::Result<FabroClient> {
let source = match run_id {
Some(run_id) => Arc::new(ExtraHeadersCredentialSource::for_run(
Arc::clone(&self.llm_source),
run_id,
)) as Arc<dyn CredentialProvider>,
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<ProviderId> {
@ -1499,8 +1504,9 @@ impl AppState {
/// resolved twice.
pub(crate) async fn resolve_llm_client_with_ready_ids(
&self,
run_id: Option<RunId>,
) -> (anyhow::Result<FabroClient>, Vec<ProviderId>) {
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<ProviderId> {
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 {

2
lib/apps/fabro-server/src/server/handler/completions.rs
View file

@ -21,7 +21,7 @@ async fn create_completion(
Json(req): Json<CreateCompletionRequest>,
) -> 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");

2
lib/apps/fabro-server/src/server/handler/graph.rs
View file

@ -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,

2
lib/apps/fabro-server/src/server/handler/models.rs
View file

@ -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");

10
lib/apps/fabro-server/src/server/handler/runs.rs
View file

@ -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<AppState>,
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,

13
lib/apps/fabro-server/src/server/handler/sessions.rs
View file

@ -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<CodingAgent, AskFabroBuildError> {
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");
}

6
lib/apps/fabro-server/src/server/petri_runs.rs
View file

@ -88,6 +88,7 @@ pub(crate) fn runtime_spec(
eligible: &[ProviderId],
dry_run: bool,
sandbox: SandboxProviderConfig,
run_id: Option<RunId>,
) -> 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<AppState>, 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<AppState>, 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(),

33
lib/apps/fabro-server/src/server/tests.rs
View file

@ -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"));

1
lib/components/fabro-petri/Cargo.toml
View file

@ -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

10
lib/components/fabro-petri/src/runtime.rs
View file

@ -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<dyn CredentialProvider>,
http: Option<HttpClient>,
eligible: &[ProviderId],
run_id: Option<RunId>,
) -> Result<Option<Client>, 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<dyn CredentialProvider>,
None => credentials,
};
let mut builder = Client::builder()
.catalog(catalog)
.credentials_arc(credentials)

10
lib/components/fabro-petri/tests/check.rs
View file

@ -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 {

1
lib/components/fabro-petri/tests/hooks.rs
View file

@ -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");

12
lib/components/fabro-petri/tests/model.rs
View file

@ -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()),

111
lib/components/fabro-petri/tests/resume.rs
View file

@ -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<MemoryRunStore>,
log: LogId,
log: Option<LogId>,
appended: Arc<AtomicUsize>,
armed: Arc<AtomicBool>,
}
#[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<dyn RunLogs>,
log: LogId,
log: Option<LogId>,
appended: Arc<AtomicUsize>,
armed: Arc<AtomicBool>,
}
#[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"
);
}

429
lib/components/fabro-petri/tests/session_headers.rs Normal file
View file

@ -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<VaultCredentialSource> {
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<RunId>,
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);
}

88
lib/foundation/fabro-auth/src/extra_headers_source.rs
View file

@ -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<dyn CredentialProvider>,
headers: HashMap<String, String>,
}
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<dyn CredentialProvider>, 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<dyn CredentialProvider>, headers: HashMap<String, String>) -> 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<dyn CredentialProvider> =
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);
}
}