firstpass_proxy/
run.rs

1//! Server bootstrap shared by the `firstpass` and `firstpass-proxy` binaries: build state from
2//! config, open the trace store, and serve until Ctrl-C. Keeping this in the lib means both the
3//! unified CLI (`firstpass up`) and the bare proxy binary start the server the exact same way.
4
5use std::sync::Arc;
6
7use crate::gate::GateHealthRegistry;
8use crate::provider::ProviderRegistry;
9use crate::{AppState, ProxyConfig, app, store};
10
11/// Initialize the global tracing subscriber from `RUST_LOG` (default `info`). Called by the
12/// binaries, not the library internals.
13pub fn init_tracing() {
14    tracing_subscriber::fmt()
15        .with_env_filter(
16            tracing_subscriber::EnvFilter::try_from_default_env()
17                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
18        )
19        .init();
20}
21
22/// Register a default error budget for every gate named across enforce routes: auto-disable a gate
23/// whose abstain rate exceeds 25% over its last 50 runs (SPEC §7.2).
24#[must_use]
25pub fn build_gate_health(config: &ProxyConfig) -> GateHealthRegistry {
26    let mut gate_health = GateHealthRegistry::new();
27    if let Some(routing) = config.routing.as_ref() {
28        let mut seen = std::collections::HashSet::new();
29        for route in &routing.routes {
30            for gate in route.gates.iter().chain(&route.deferred_gates) {
31                if seen.insert(gate.clone()) {
32                    gate_health = gate_health.with_budget(gate.clone(), 50, 0.25);
33                }
34            }
35        }
36    }
37    gate_health
38}
39
40/// Open the trace store, build [`AppState`], and serve the HTTP proxy until Ctrl-C.
41///
42/// # Errors
43/// Returns any error from opening the store, building the HTTP client, binding the listener, or
44/// serving.
45pub async fn serve(config: ProxyConfig) -> Result<(), Box<dyn std::error::Error>> {
46    let (traces, writer) = store::open(&config.db_path)?;
47    let bind = config.bind.clone();
48    // Build the provider registry from any `[[provider]]` entries in the routing config (Groq,
49    // Together, Ollama, …), on top of the built-in anthropic/openai defaults. No config => defaults.
50    let provider_defs = config
51        .routing
52        .as_ref()
53        .map(|r| r.providers.as_slice())
54        .unwrap_or_default();
55    let providers = ProviderRegistry::from_config(
56        provider_defs,
57        &config.upstream_anthropic,
58        &config.upstream_openai,
59    );
60    let gate_health = build_gate_health(&config);
61    // Online adaptive conformal (opt-in): seed the live threshold from the fixed one (or 0.5) and
62    // let /v1/feedback track it. Absent config => None => fixed-threshold behavior, byte-identical.
63    let adaptive = config
64        .routing
65        .as_ref()
66        .and_then(|r| r.escalation.adaptive.as_ref())
67        .map(|a| {
68            let init = config
69                .routing
70                .as_ref()
71                .and_then(|r| r.escalation.serve_threshold)
72                .unwrap_or(0.5);
73            Arc::new(std::sync::Mutex::new(
74                firstpass_core::conformal::AdaptiveConformal::new(a.alpha, a.gamma, init),
75            ))
76        });
77
78    let state = AppState {
79        config: Arc::new(config),
80        // Observe passthrough may stream SSE, so only bound the CONNECT phase here — a total or
81        // read timeout would sever a long-lived stream. (The enforce providers, which never stream
82        // through the adapter, carry a full request timeout — see `ProviderRegistry::new`.)
83        http: reqwest::Client::builder()
84            .connect_timeout(std::time::Duration::from_secs(10))
85            .build()?,
86        providers,
87        gate_health: Arc::new(gate_health),
88        traces,
89        adaptive,
90    };
91
92    let listener = tokio::net::TcpListener::bind(&bind).await?;
93    tracing::info!(%bind, "firstpass listening");
94    tracing::info!("offboard: unset ANTHROPIC_BASE_URL");
95
96    axum::serve(listener, app(state)?)
97        .with_graceful_shutdown(shutdown_signal())
98        .await?;
99
100    // Dropping `traces` (via `state`, already out of scope) closes the channel; wait for the
101    // writer to flush and exit before the process ends.
102    drop(writer);
103    Ok(())
104}
105
106async fn shutdown_signal() {
107    if let Err(err) = tokio::signal::ctrl_c().await {
108        tracing::warn!(%err, "failed to install Ctrl-C handler; shutting down anyway");
109    }
110}
firstpass_proxy/run.rs

firstpass_proxy/
run.rs
