Compare commits
13 Commits
feat/73-ca
...
feat/B3-co
| Author | SHA1 | Date | |
|---|---|---|---|
|
a9d7382be8
|
|||
|
cb9e7c7c2e
|
|||
|
2604b9f134
|
|||
|
178e3092d5
|
|||
|
46befde4cd
|
|||
|
cf87e156c5
|
|||
|
79073170ec
|
|||
|
71106afaf1
|
|||
|
d2dcdd6ebb
|
|||
|
222c2a6116
|
|||
|
1115bb0942
|
|||
|
63f578cb15
|
|||
|
76c90fa993
|
3
.gitignore
vendored
3
.gitignore
vendored
@@ -1,6 +1,9 @@
|
||||
/target
|
||||
/bench/node_modules
|
||||
/bench/dist
|
||||
/helexa.ai/node_modules
|
||||
/helexa.ai/dist
|
||||
helexa.ai/.env.local
|
||||
*.swp
|
||||
*.swo
|
||||
.idea/
|
||||
|
||||
714
Cargo.lock
generated
714
Cargo.lock
generated
File diff suppressed because it is too large
Load Diff
@@ -9,6 +9,7 @@ members = [
|
||||
"crates/helexa-bench",
|
||||
"crates/helexa-router",
|
||||
"crates/helexa-stream",
|
||||
"crates/helexa-upstream",
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
|
||||
@@ -90,3 +90,18 @@ account_id = "operator"
|
||||
key_id = "infra"
|
||||
# No hard_cap → uncapped operator infra key (own fleet, own use). Still
|
||||
# metered for visibility.
|
||||
|
||||
# -- Upstream (helexa mesh) entitlements client (#57) --------------------
|
||||
# When enabled, a bearer key NOT found in [[entitlements.keys]] above is
|
||||
# validated against the helexa-upstream authority (mesh accounts), and its
|
||||
# budget is reserved/settled there. Operator-local keys (incl. the infra
|
||||
# key) never leave this process. Fail-closed: if upstream is unreachable a
|
||||
# request is refused (503 + Retry-After), never served un-authorized.
|
||||
# Disabled by default — a standalone operator runs purely local.
|
||||
[upstream]
|
||||
enabled = false
|
||||
# url = "https://upstream.helexa.ai"
|
||||
# Shared client bearer this cortex presents (maps to an operator_id
|
||||
# upstream). Override via CORTEX_UPSTREAM__BEARER in prod.
|
||||
# bearer = "replace-with-operator-client-secret"
|
||||
# timeout_secs = 5
|
||||
|
||||
@@ -22,6 +22,36 @@ pub struct GatewayConfig {
|
||||
/// setups keep working until keys are configured.
|
||||
#[serde(default)]
|
||||
pub entitlements: EntitlementsConfig,
|
||||
/// helexa-upstream client (#57). When enabled, keys not found in the
|
||||
/// local `[entitlements]` config are validated against the mesh
|
||||
/// authority, and budget is reserved/settled there. Disabled by default
|
||||
/// — a single operator runs purely local.
|
||||
#[serde(default)]
|
||||
pub upstream: UpstreamClientConfig,
|
||||
}
|
||||
|
||||
/// `[upstream]` — the helexa-upstream authority client (#57). Locally
|
||||
/// unrecognised bearer keys are resolved against `url`'s `/authz/v1` surface
|
||||
/// (mesh accounts); local keys (operator + infra) never leave the process.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct UpstreamClientConfig {
|
||||
/// Enable the upstream fallthrough. Off → purely local entitlements.
|
||||
#[serde(default)]
|
||||
pub enabled: bool,
|
||||
/// Base URL of helexa-upstream (e.g. "https://upstream.helexa.ai").
|
||||
#[serde(default)]
|
||||
pub url: String,
|
||||
/// Shared client bearer this cortex presents to `/authz/v1` (maps to an
|
||||
/// operator_id upstream). Sent as `Authorization: Bearer <bearer>`.
|
||||
#[serde(default)]
|
||||
pub bearer: String,
|
||||
/// Per-call timeout (seconds) to upstream.
|
||||
#[serde(default = "default_upstream_timeout")]
|
||||
pub timeout_secs: u64,
|
||||
}
|
||||
|
||||
fn default_upstream_timeout() -> u64 {
|
||||
5
|
||||
}
|
||||
|
||||
/// `[entitlements]` — the local/static [`crate::entitlements::EntitlementProvider`]
|
||||
@@ -129,6 +159,7 @@ impl Default for GatewayConfig {
|
||||
neurons: vec![],
|
||||
models_config: default_models_path(),
|
||||
entitlements: EntitlementsConfig::default(),
|
||||
upstream: UpstreamClientConfig::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -81,12 +81,19 @@ pub struct BudgetSnapshot {
|
||||
pub reserved: u64,
|
||||
}
|
||||
|
||||
/// Authentication failure — the bearer key could not be resolved. Maps to
|
||||
/// `401 invalid_api_key` (#49/#63).
|
||||
/// Authentication failure — the bearer key could not be resolved.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum AuthError {
|
||||
/// The key is genuinely unknown → `401 invalid_api_key` (#49/#63).
|
||||
#[error("invalid or unknown API key")]
|
||||
InvalidKey,
|
||||
/// The authority that could resolve the key is unreachable (e.g. the
|
||||
/// helexa-upstream client failed, #57). Fail **closed** but distinctly:
|
||||
/// a transient outage must surface as `503 service_unavailable` +
|
||||
/// `Retry-After`, never `401` — a real key must not be rejected as
|
||||
/// invalid during an upstream blip.
|
||||
#[error("entitlement authority unavailable; retry in {retry_after_secs}s")]
|
||||
Unavailable { retry_after_secs: u64 },
|
||||
}
|
||||
|
||||
/// Why a reservation was refused. Carries enough for the caller to build the
|
||||
|
||||
@@ -22,7 +22,7 @@ use axum::http::header::AUTHORIZATION;
|
||||
use axum::http::{HeaderMap, HeaderValue};
|
||||
use axum::middleware::Next;
|
||||
use axum::response::Response;
|
||||
use cortex_core::entitlements::{HEADER_ACCOUNT_ID, HEADER_KEY_ID};
|
||||
use cortex_core::entitlements::{AuthError, HEADER_ACCOUNT_ID, HEADER_KEY_ID};
|
||||
use cortex_core::error_envelope::OpenAiError;
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -83,14 +83,25 @@ pub async fn require_principal(
|
||||
req.extensions_mut().insert(principal);
|
||||
next.run(req).await
|
||||
}
|
||||
// An unrecognized key only hard-fails when auth is *required*.
|
||||
// In allow-anonymous mode (the default) we must IGNORE it and
|
||||
// serve the request unauthenticated — otherwise the placeholder
|
||||
// keys that OpenAI-compatible clients send by default (opencode,
|
||||
// Open WebUI, Agent Zero, litellm) would all break, even though
|
||||
// the operator never opted into auth. Pre-#49 the bearer was
|
||||
// never inspected at all; this preserves that for require_auth=false.
|
||||
Err(_) => {
|
||||
// The entitlement authority is unreachable (upstream client
|
||||
// blip, #57). Fail **closed but distinct**: a transient outage
|
||||
// must not reject a real key as `401 invalid_api_key` — it's a
|
||||
// retryable `503`. This holds regardless of require_auth: we
|
||||
// can't safely serve a key we couldn't authorize.
|
||||
Err(AuthError::Unavailable { retry_after_secs }) => {
|
||||
envelope_response(OpenAiError::service_unavailable(
|
||||
"entitlement authority temporarily unavailable",
|
||||
Some(retry_after_secs),
|
||||
))
|
||||
}
|
||||
// A genuinely unrecognized key only hard-fails when auth is
|
||||
// *required*. In allow-anonymous mode (the default) we IGNORE it
|
||||
// and serve unauthenticated — otherwise the placeholder keys that
|
||||
// OpenAI-compatible clients send by default (opencode, Open WebUI,
|
||||
// Agent Zero, litellm) would all break though the operator never
|
||||
// opted into auth. Pre-#49 the bearer was never inspected; this
|
||||
// preserves that for require_auth=false.
|
||||
Err(AuthError::InvalidKey) => {
|
||||
if fleet.require_auth {
|
||||
unauthorized("invalid API key")
|
||||
} else {
|
||||
|
||||
112
crates/cortex-gateway/src/entitlements_chain.rs
Normal file
112
crates/cortex-gateway/src/entitlements_chain.rs
Normal file
@@ -0,0 +1,112 @@
|
||||
//! Chained entitlement provider (#57): operator-local keys first, mesh
|
||||
//! upstream for everything else.
|
||||
//!
|
||||
//! `resolve` tries the [`LocalEntitlementProvider`] (operator + infra keys —
|
||||
//! never a network hop); only a locally-unknown key falls through to
|
||||
//! [`UpstreamEntitlementProvider`]. Because the local provider treats an
|
||||
//! unconfigured principal as uncapped, reserve/settle/release/snapshot must
|
||||
//! **not** blindly hit local — they dispatch to whichever backend resolved
|
||||
//! that account, remembered in a map keyed by `account_id` (populated at
|
||||
//! resolve time).
|
||||
|
||||
use crate::entitlements_local::LocalEntitlementProvider;
|
||||
use crate::entitlements_upstream::UpstreamEntitlementProvider;
|
||||
use async_trait::async_trait;
|
||||
use cortex_core::entitlements::{
|
||||
AuthError, BudgetError, BudgetSnapshot, EntitlementProvider, Principal, Reservation,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
enum Backend {
|
||||
Local,
|
||||
Upstream,
|
||||
}
|
||||
|
||||
pub struct ChainedEntitlementProvider {
|
||||
local: LocalEntitlementProvider,
|
||||
upstream: UpstreamEntitlementProvider,
|
||||
/// account_id → which backend owns it, learned at resolve time.
|
||||
backends: RwLock<HashMap<String, Backend>>,
|
||||
}
|
||||
|
||||
impl ChainedEntitlementProvider {
|
||||
pub fn new(local: LocalEntitlementProvider, upstream: UpstreamEntitlementProvider) -> Self {
|
||||
Self {
|
||||
local,
|
||||
upstream,
|
||||
backends: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn record(&self, account_id: &str, backend: Backend) {
|
||||
self.backends
|
||||
.write()
|
||||
.await
|
||||
.insert(account_id.to_string(), backend);
|
||||
}
|
||||
|
||||
/// The backend that owns `account_id`. Defaults to `Upstream` for an
|
||||
/// account never resolved this process-lifetime (a resolve always
|
||||
/// precedes reserve in a request, so this is just a safe fallback —
|
||||
/// upstream fails closed if the account is bogus).
|
||||
async fn backend_for(&self, account_id: &str) -> Backend {
|
||||
self.backends
|
||||
.read()
|
||||
.await
|
||||
.get(account_id)
|
||||
.copied()
|
||||
.unwrap_or(Backend::Upstream)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EntitlementProvider for ChainedEntitlementProvider {
|
||||
async fn resolve(&self, api_key: &str) -> Result<Principal, AuthError> {
|
||||
match self.local.resolve(api_key).await {
|
||||
Ok(p) => {
|
||||
self.record(&p.account_id, Backend::Local).await;
|
||||
Ok(p)
|
||||
}
|
||||
Err(AuthError::InvalidKey) => {
|
||||
let p = self.upstream.resolve(api_key).await?;
|
||||
self.record(&p.account_id, Backend::Upstream).await;
|
||||
Ok(p)
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
async fn reserve(
|
||||
&self,
|
||||
principal: &Principal,
|
||||
max_tokens: u64,
|
||||
) -> Result<Reservation, BudgetError> {
|
||||
match self.backend_for(&principal.account_id).await {
|
||||
Backend::Local => self.local.reserve(principal, max_tokens).await,
|
||||
Backend::Upstream => self.upstream.reserve(principal, max_tokens).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn settle(&self, reservation: Reservation, actual_tokens: u64) {
|
||||
match self.backend_for(&reservation.principal.account_id).await {
|
||||
Backend::Local => self.local.settle(reservation, actual_tokens).await,
|
||||
Backend::Upstream => self.upstream.settle(reservation, actual_tokens).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn release(&self, reservation: Reservation) {
|
||||
match self.backend_for(&reservation.principal.account_id).await {
|
||||
Backend::Local => self.local.release(reservation).await,
|
||||
Backend::Upstream => self.upstream.release(reservation).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn snapshot(&self, principal: &Principal) -> Option<BudgetSnapshot> {
|
||||
match self.backend_for(&principal.account_id).await {
|
||||
Backend::Local => self.local.snapshot(principal).await,
|
||||
Backend::Upstream => self.upstream.snapshot(principal).await,
|
||||
}
|
||||
}
|
||||
}
|
||||
246
crates/cortex-gateway/src/entitlements_upstream.rs
Normal file
246
crates/cortex-gateway/src/entitlements_upstream.rs
Normal file
@@ -0,0 +1,246 @@
|
||||
//! helexa-upstream client (#57): an [`EntitlementProvider`] that resolves
|
||||
//! keys and reserves/settles budget against the mesh authority's
|
||||
//! `/authz/v1` surface (B2). It is "just another impl of the trait" — cortex
|
||||
//! enforcement (`auth.rs`, `metering.rs`) is unchanged.
|
||||
//!
|
||||
//! **Fail closed.** When upstream is unreachable, `resolve` returns
|
||||
//! [`AuthError::Unavailable`] (→ `503`, never `401`) and `reserve` refuses
|
||||
//! with a retryable [`BudgetError::RateLimited`] — a request is never served
|
||||
//! on an un-authorized key, and a real key is never rejected as invalid
|
||||
//! during a blip.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use cortex_core::config::UpstreamClientConfig;
|
||||
use cortex_core::entitlements::{
|
||||
AuthError, BudgetError, BudgetSnapshot, EntitlementProvider, Principal, Reservation,
|
||||
};
|
||||
use serde::Deserialize;
|
||||
use std::time::Duration;
|
||||
|
||||
/// Retry-After (seconds) advertised when we fail closed on an upstream
|
||||
/// outage.
|
||||
const FAIL_CLOSED_RETRY_SECS: u64 = 5;
|
||||
|
||||
pub struct UpstreamEntitlementProvider {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
bearer: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct PrincipalDto {
|
||||
account_id: String,
|
||||
key_id: String,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct SnapshotDto {
|
||||
hard_cap: Option<u64>,
|
||||
spent: u64,
|
||||
reserved: u64,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ResolveResp {
|
||||
principal: PrincipalDto,
|
||||
#[allow(dead_code)]
|
||||
snapshot: Option<SnapshotDto>,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ReserveResp {
|
||||
reservation_id: Option<i64>,
|
||||
rejected: Option<Rejection>,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
enum Rejection {
|
||||
InsufficientQuota {
|
||||
requested: u64,
|
||||
available: u64,
|
||||
},
|
||||
RateLimited {
|
||||
requested: u64,
|
||||
available: u64,
|
||||
retry_after_secs: u64,
|
||||
},
|
||||
}
|
||||
|
||||
impl UpstreamEntitlementProvider {
|
||||
pub fn new(cfg: &UpstreamClientConfig) -> Self {
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(cfg.timeout_secs))
|
||||
.build()
|
||||
.expect("failed to build upstream HTTP client");
|
||||
Self {
|
||||
client,
|
||||
base_url: cfg.url.trim_end_matches('/').to_string(),
|
||||
bearer: cfg.bearer.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn url(&self, path: &str) -> String {
|
||||
format!("{}{}", self.base_url, path)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EntitlementProvider for UpstreamEntitlementProvider {
|
||||
async fn resolve(&self, api_key: &str) -> Result<Principal, AuthError> {
|
||||
let resp = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/resolve"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({ "api_key": api_key }))
|
||||
.send()
|
||||
.await;
|
||||
let resp = match resp {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "upstream resolve unreachable; failing closed");
|
||||
return Err(AuthError::Unavailable {
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
});
|
||||
}
|
||||
};
|
||||
if resp.status().as_u16() == 401 {
|
||||
return Err(AuthError::InvalidKey);
|
||||
}
|
||||
if !resp.status().is_success() {
|
||||
return Err(AuthError::Unavailable {
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
});
|
||||
}
|
||||
match resp.json::<ResolveResp>().await {
|
||||
Ok(r) => Ok(Principal {
|
||||
account_id: r.principal.account_id,
|
||||
key_id: r.principal.key_id,
|
||||
}),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "upstream resolve: bad body; failing closed");
|
||||
Err(AuthError::Unavailable {
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn reserve(
|
||||
&self,
|
||||
principal: &Principal,
|
||||
max_tokens: u64,
|
||||
) -> Result<Reservation, BudgetError> {
|
||||
let fail_closed = || BudgetError::RateLimited {
|
||||
requested: max_tokens,
|
||||
available: 0,
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
};
|
||||
let resp = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/reserve"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({
|
||||
"account_id": principal.account_id,
|
||||
"key_id": principal.key_id,
|
||||
"max_tokens": max_tokens,
|
||||
}))
|
||||
.send()
|
||||
.await;
|
||||
let resp = match resp {
|
||||
Ok(r) if r.status().is_success() => r,
|
||||
Ok(r) => {
|
||||
tracing::warn!(status = %r.status(), "upstream reserve non-2xx; failing closed");
|
||||
return Err(fail_closed());
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "upstream reserve unreachable; failing closed");
|
||||
return Err(fail_closed());
|
||||
}
|
||||
};
|
||||
match resp.json::<ReserveResp>().await {
|
||||
Ok(ReserveResp {
|
||||
reservation_id: Some(id),
|
||||
..
|
||||
}) => Ok(Reservation {
|
||||
id: id as u64,
|
||||
principal: principal.clone(),
|
||||
reserved: max_tokens,
|
||||
}),
|
||||
Ok(ReserveResp {
|
||||
rejected:
|
||||
Some(Rejection::InsufficientQuota {
|
||||
requested,
|
||||
available,
|
||||
}),
|
||||
..
|
||||
}) => Err(BudgetError::InsufficientQuota {
|
||||
requested,
|
||||
available,
|
||||
}),
|
||||
Ok(ReserveResp {
|
||||
rejected:
|
||||
Some(Rejection::RateLimited {
|
||||
requested,
|
||||
available,
|
||||
retry_after_secs,
|
||||
}),
|
||||
..
|
||||
}) => Err(BudgetError::RateLimited {
|
||||
requested,
|
||||
available,
|
||||
retry_after_secs,
|
||||
}),
|
||||
_ => Err(fail_closed()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn settle(&self, reservation: Reservation, actual_tokens: u64) {
|
||||
// Best-effort; a lost settle is reaped by the upstream sweeper (B2).
|
||||
let _ = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/settle"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({
|
||||
"reservation_id": reservation.id as i64,
|
||||
"actual_tokens": actual_tokens,
|
||||
}))
|
||||
.send()
|
||||
.await
|
||||
.inspect_err(
|
||||
|e| tracing::warn!(error = %e, "upstream settle failed (sweeper will reap)"),
|
||||
);
|
||||
}
|
||||
|
||||
async fn release(&self, reservation: Reservation) {
|
||||
let _ = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/release"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({ "reservation_id": reservation.id as i64 }))
|
||||
.send()
|
||||
.await
|
||||
.inspect_err(
|
||||
|e| tracing::warn!(error = %e, "upstream release failed (sweeper will reap)"),
|
||||
);
|
||||
}
|
||||
|
||||
async fn snapshot(&self, principal: &Principal) -> Option<BudgetSnapshot> {
|
||||
let resp = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/snapshot"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({
|
||||
"account_id": principal.account_id,
|
||||
"key_id": principal.key_id,
|
||||
}))
|
||||
.send()
|
||||
.await
|
||||
.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
return None;
|
||||
}
|
||||
let dto = resp.json::<SnapshotDto>().await.ok()?;
|
||||
Some(BudgetSnapshot {
|
||||
hard_cap: dto.hard_cap,
|
||||
spent: dto.spent,
|
||||
reserved: dto.reserved,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,8 @@
|
||||
pub mod anthropic_sse;
|
||||
pub mod auth;
|
||||
pub mod entitlements_chain;
|
||||
pub mod entitlements_local;
|
||||
pub mod entitlements_upstream;
|
||||
pub mod error;
|
||||
pub mod evictor;
|
||||
pub mod handlers;
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
use crate::entitlements_chain::ChainedEntitlementProvider;
|
||||
use crate::entitlements_local::LocalEntitlementProvider;
|
||||
use crate::entitlements_upstream::UpstreamEntitlementProvider;
|
||||
use cortex_core::catalogue::ModelCatalogue;
|
||||
use cortex_core::config::{EvictionSettings, GatewayConfig, NeuronEndpoint};
|
||||
use cortex_core::entitlements::EntitlementProvider;
|
||||
@@ -45,8 +47,20 @@ impl CortexState {
|
||||
|
||||
let catalogue = ModelCatalogue::load(&config.models_config);
|
||||
|
||||
let entitlements: Arc<dyn EntitlementProvider> =
|
||||
Arc::new(LocalEntitlementProvider::from_config(&config.entitlements));
|
||||
// Local provider always handles operator + infra keys. When the
|
||||
// upstream client is enabled (#57), wrap it in the chain so locally
|
||||
// unknown keys fall through to the mesh authority; otherwise stay
|
||||
// purely local.
|
||||
let local = LocalEntitlementProvider::from_config(&config.entitlements);
|
||||
let entitlements: Arc<dyn EntitlementProvider> = if config.upstream.enabled {
|
||||
tracing::info!(url = %config.upstream.url, "upstream entitlement client enabled");
|
||||
Arc::new(ChainedEntitlementProvider::new(
|
||||
local,
|
||||
UpstreamEntitlementProvider::new(&config.upstream),
|
||||
))
|
||||
} else {
|
||||
Arc::new(local)
|
||||
};
|
||||
|
||||
Self {
|
||||
nodes: RwLock::new(nodes),
|
||||
|
||||
@@ -57,6 +57,7 @@ async fn test_alias_resolves_in_chat_completions() {
|
||||
}],
|
||||
models_config: models_path.to_string_lossy().to_string(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -143,6 +144,7 @@ async fn test_aliases_surface_in_v1_models() {
|
||||
}],
|
||||
models_config: models_path.to_string_lossy().to_string(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -232,6 +234,7 @@ async fn test_alias_falls_through_for_unmapped_model() {
|
||||
}],
|
||||
models_config: models_path.to_string_lossy().to_string(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -105,6 +105,7 @@ async fn spawn_gateway(neuron_url: &str, entitlements: EntitlementsConfig) -> St
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements,
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -81,6 +81,7 @@ async fn spawn_gateway(neuron_url: &str, key: ApiKeyConfig) -> (Arc<CortexState>
|
||||
require_auth: true,
|
||||
keys: vec![key],
|
||||
},
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -430,6 +430,7 @@ pub async fn spawn_gateway_with_state(mock_url: &str) -> (Arc<CortexState>, Stri
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -89,6 +89,7 @@ async fn error_response_no_healthy_nodes() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(cortex_gateway::state::CortexState::from_config(&config));
|
||||
|
||||
@@ -72,6 +72,7 @@ fn make_fleet(endpoint: &str, defrag_after: u32) -> Arc<CortexState> {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
Arc::new(CortexState::from_config(&config))
|
||||
}
|
||||
|
||||
@@ -74,6 +74,7 @@ async fn fleet_with(big_healthy: bool, big_devices: usize) -> Arc<CortexState> {
|
||||
],
|
||||
models_config: cat.to_string_lossy().into_owned(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -72,6 +72,7 @@ async fn two_neuron_fleet(endpoint_a: &str, endpoint_b: &str) -> Arc<CortexState
|
||||
],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
Arc::new(CortexState::from_config(&config))
|
||||
}
|
||||
|
||||
@@ -53,6 +53,7 @@ async fn spawn_metered_gateway(neuron_url: &str) -> (Arc<CortexState>, String) {
|
||||
window: CapWindow::Balance,
|
||||
}],
|
||||
},
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -158,6 +159,7 @@ async fn anonymous_request_records_no_spend() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: EntitlementsConfig::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -66,6 +66,7 @@ harness = "candle"
|
||||
}],
|
||||
models_config: cat_path.to_string_lossy().into_owned(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -55,6 +55,7 @@ capabilities = ["text"]
|
||||
}],
|
||||
models_config: cat_path.to_string_lossy().into_owned(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -32,6 +32,7 @@ async fn test_poller_discovers_models() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -84,6 +85,7 @@ async fn test_poller_updates_gateway_models_endpoint() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -156,6 +158,7 @@ async fn test_models_endpoint_unions_capabilities_across_nodes() {
|
||||
],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -219,6 +222,7 @@ async fn test_poller_marks_unreachable_node_unhealthy() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -273,6 +277,7 @@ async fn test_poller_removes_stale_models() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -304,6 +309,7 @@ async fn test_poller_removes_stale_models() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet2 = Arc::new(CortexState::from_config(&config2));
|
||||
@@ -386,6 +392,7 @@ async fn test_poller_captures_activation_from_health() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -431,6 +438,7 @@ async fn test_poller_parses_recovering_status() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -75,6 +75,7 @@ async fn spawn_gateway(neuron: &str, context: usize) -> String {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -118,6 +118,7 @@ async fn test_no_healthy_nodes() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = std::sync::Arc::new(cortex_gateway::state::CortexState::from_config(&config));
|
||||
|
||||
|
||||
105
crates/cortex-gateway/tests/upstream_chain.rs
Normal file
105
crates/cortex-gateway/tests/upstream_chain.rs
Normal file
@@ -0,0 +1,105 @@
|
||||
//! B3: the chained entitlement provider (local → upstream) and fail-closed
|
||||
//! semantics, exercised against a mock helexa-upstream `/authz/v1`.
|
||||
|
||||
use axum::{Json, Router, routing::post};
|
||||
use cortex_core::config::{ApiKeyConfig, EntitlementsConfig, UpstreamClientConfig};
|
||||
use cortex_core::entitlements::{AuthError, EntitlementProvider};
|
||||
use cortex_gateway::entitlements_chain::ChainedEntitlementProvider;
|
||||
use cortex_gateway::entitlements_local::LocalEntitlementProvider;
|
||||
use cortex_gateway::entitlements_upstream::UpstreamEntitlementProvider;
|
||||
use serde_json::{Value, json};
|
||||
use tokio::net::TcpListener;
|
||||
|
||||
/// Mock upstream: `mesh-key` resolves to a mesh account; anything else 401.
|
||||
/// reserve always grants reservation 1.
|
||||
async fn spawn_mock_upstream() -> String {
|
||||
async fn resolve(Json(body): Json<Value>) -> axum::response::Response {
|
||||
use axum::response::IntoResponse;
|
||||
if body["api_key"] == "mesh-key" {
|
||||
Json(json!({"principal": {"account_id": "mesh-acct", "key_id": "mesh-key-1"}}))
|
||||
.into_response()
|
||||
} else {
|
||||
(
|
||||
axum::http::StatusCode::UNAUTHORIZED,
|
||||
Json(json!({"error": {"code": "invalid_api_key"}})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
async fn reserve() -> Json<Value> {
|
||||
Json(json!({ "reservation_id": 1 }))
|
||||
}
|
||||
let app = Router::new()
|
||||
.route("/authz/v1/resolve", post(resolve))
|
||||
.route("/authz/v1/reserve", post(reserve));
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
format!("http://{addr}")
|
||||
}
|
||||
|
||||
fn local_with_key() -> LocalEntitlementProvider {
|
||||
let cfg = EntitlementsConfig {
|
||||
require_auth: false,
|
||||
keys: vec![ApiKeyConfig {
|
||||
key: "local-key".into(),
|
||||
account_id: "op".into(),
|
||||
key_id: None,
|
||||
hard_cap: None,
|
||||
window: Default::default(),
|
||||
}],
|
||||
};
|
||||
LocalEntitlementProvider::from_config(&cfg)
|
||||
}
|
||||
|
||||
fn chain(local: LocalEntitlementProvider, url: &str) -> ChainedEntitlementProvider {
|
||||
let upstream = UpstreamEntitlementProvider::new(&UpstreamClientConfig {
|
||||
enabled: true,
|
||||
url: url.to_string(),
|
||||
bearer: "client-secret".into(),
|
||||
timeout_secs: 5,
|
||||
});
|
||||
ChainedEntitlementProvider::new(local, upstream)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn local_key_resolves_locally() {
|
||||
let url = spawn_mock_upstream().await;
|
||||
let c = chain(local_with_key(), &url);
|
||||
let p = c.resolve("local-key").await.expect("local resolves");
|
||||
assert_eq!(p.account_id, "op");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unknown_key_falls_through_to_upstream() {
|
||||
let url = spawn_mock_upstream().await;
|
||||
let c = chain(local_with_key(), &url);
|
||||
let p = c.resolve("mesh-key").await.expect("upstream resolves");
|
||||
assert_eq!(p.account_id, "mesh-acct");
|
||||
assert_eq!(p.key_id, "mesh-key-1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unknown_everywhere_is_invalid_key() {
|
||||
let url = spawn_mock_upstream().await;
|
||||
let c = chain(local_with_key(), &url);
|
||||
match c.resolve("nope").await {
|
||||
Err(AuthError::InvalidKey) => {}
|
||||
other => panic!("expected InvalidKey, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upstream_unreachable_fails_closed_as_unavailable() {
|
||||
// No mock — point at a dead port. A locally-unknown key must surface
|
||||
// Unavailable (→ 503), never InvalidKey (→ 401).
|
||||
let c = chain(local_with_key(), "http://127.0.0.1:1");
|
||||
match c.resolve("some-mesh-key").await {
|
||||
Err(AuthError::Unavailable { retry_after_secs }) => assert!(retry_after_secs > 0),
|
||||
other => panic!("expected Unavailable, got {other:?}"),
|
||||
}
|
||||
// A local key still resolves without touching upstream.
|
||||
assert_eq!(c.resolve("local-key").await.unwrap().account_id, "op");
|
||||
}
|
||||
@@ -25,6 +25,7 @@ serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
figment = { workspace = true }
|
||||
anyhow = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
clap = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
tracing-subscriber = { workspace = true }
|
||||
@@ -33,3 +34,8 @@ chrono = { workspace = true }
|
||||
[dev-dependencies]
|
||||
# Jail (isolated cwd + env) for config tests.
|
||||
figment = { workspace = true, features = ["test"] }
|
||||
# Self-signed cert generation + a minimal HTTPS server for the outbound
|
||||
# TLS-pinning tests (#74).
|
||||
rcgen = "0.13"
|
||||
rustls = "0.23"
|
||||
tokio-rustls = "0.26"
|
||||
|
||||
243
crates/helexa-router/src/catalogue.rs
Normal file
243
crates/helexa-router/src/catalogue.rs
Normal file
@@ -0,0 +1,243 @@
|
||||
//! Federation catalogue (#75) — the router's aggregate `/v1/models`.
|
||||
//!
|
||||
//! Presents the **deduped union** of every reachable cortex's `/v1/models`
|
||||
//! as the router's own catalogue, so an opencode client doing discovery
|
||||
//! against the router resolves the whole federation without knowing about
|
||||
//! operators or cortexes (resolves #61's "Router/discovery contract").
|
||||
//!
|
||||
//! Re-tiering: the fractal design is neuron ← cortex ← router. At the
|
||||
//! router tier the "nodes" are **cortexes**, so the merged entry's
|
||||
//! `feasible_on` / `locations` are rewritten to **operator names**, not the
|
||||
//! neuron names a cortex reports. That keeps the federation view honest
|
||||
//! ("served by these operators") without leaking each operator's internal
|
||||
//! topology (neuron names, per-device VRAM) to end users.
|
||||
//!
|
||||
//! Conflict resolution when operators advertise the same model with
|
||||
//! different enrichment:
|
||||
//! - **`limit`** → the *tightest* (smallest `context`), so a client never
|
||||
//! overflows the most-constrained operator that might serve it (same rule
|
||||
//! cortex uses across its neurons).
|
||||
//! - **`cost`** → the *cheapest* (lowest input, then output), the
|
||||
//! federation "from" price. Richer policy (a range, region/price-aware
|
||||
//! selection) couples to #68 and is left as a follow-up.
|
||||
|
||||
use crate::state::{CortexTopology, entry_feasible};
|
||||
use cortex_core::harness::{ModelCost, ModelLimit};
|
||||
use cortex_core::node::{CortexModelEntry, ModelLocation, ModelStatus};
|
||||
use std::collections::HashMap;
|
||||
|
||||
/// Build the federation catalogue: the deduped union of every reachable
|
||||
/// cortex's serveable models, merged across operators and sorted by id.
|
||||
pub fn aggregate_models(topology: &HashMap<String, CortexTopology>) -> Vec<CortexModelEntry> {
|
||||
// Iterate cortexes in name order so `feasible_on` / `locations` and the
|
||||
// limit/cost tie-breaks are deterministic regardless of map ordering.
|
||||
let mut cortexes: Vec<(&String, &CortexTopology)> = topology.iter().collect();
|
||||
cortexes.sort_by(|a, b| a.0.cmp(b.0));
|
||||
|
||||
let mut merged: HashMap<String, CortexModelEntry> = HashMap::new();
|
||||
for (cortex_name, t) in cortexes {
|
||||
if !t.reachable {
|
||||
continue;
|
||||
}
|
||||
for entry in t.models.values() {
|
||||
// Only surface models the cortex can actually serve — a
|
||||
// catalogue-only entry no neuron can host shouldn't appear in
|
||||
// the federation view.
|
||||
if !entry_feasible(entry) {
|
||||
continue;
|
||||
}
|
||||
merged
|
||||
.entry(entry.id.clone())
|
||||
.and_modify(|acc| merge_into(acc, cortex_name, entry))
|
||||
.or_insert_with(|| router_entry(cortex_name, entry));
|
||||
}
|
||||
}
|
||||
|
||||
let mut out: Vec<CortexModelEntry> = merged.into_values().collect();
|
||||
out.sort_by(|a, b| a.id.cmp(&b.id));
|
||||
out
|
||||
}
|
||||
|
||||
/// Seed a federation entry from the first cortex that serves the model,
|
||||
/// re-tiering `feasible_on` / `locations` to the operator name.
|
||||
fn router_entry(cortex: &str, e: &CortexModelEntry) -> CortexModelEntry {
|
||||
CortexModelEntry {
|
||||
id: e.id.clone(),
|
||||
object: "model".into(),
|
||||
created: e.created,
|
||||
owned_by: e.owned_by.clone(),
|
||||
loaded: e.loaded,
|
||||
feasible_on: vec![cortex.to_string()],
|
||||
locations: loaded_location(cortex, e),
|
||||
capabilities: e.capabilities.clone(),
|
||||
limit: e.limit.clone(),
|
||||
cost: e.cost.clone(),
|
||||
tool_call: e.tool_call,
|
||||
reasoning: e.reasoning,
|
||||
}
|
||||
}
|
||||
|
||||
/// Fold another cortex's view of the same model into the merged entry.
|
||||
fn merge_into(acc: &mut CortexModelEntry, cortex: &str, e: &CortexModelEntry) {
|
||||
acc.loaded |= e.loaded;
|
||||
acc.feasible_on.push(cortex.to_string());
|
||||
acc.locations.extend(loaded_location(cortex, e));
|
||||
for cap in &e.capabilities {
|
||||
if !acc.capabilities.contains(cap) {
|
||||
acc.capabilities.push(cap.clone());
|
||||
}
|
||||
}
|
||||
acc.tool_call |= e.tool_call;
|
||||
acc.reasoning |= e.reasoning;
|
||||
acc.limit = tightest_limit(acc.limit.take(), e.limit.clone());
|
||||
acc.cost = cheapest_cost(acc.cost.take(), e.cost.clone());
|
||||
}
|
||||
|
||||
/// A single cortex-tier location when the model is loaded at that operator;
|
||||
/// empty when only cold-loadable. Neuron-level VRAM is deliberately dropped.
|
||||
fn loaded_location(cortex: &str, e: &CortexModelEntry) -> Vec<ModelLocation> {
|
||||
if e.loaded {
|
||||
vec![ModelLocation {
|
||||
node: cortex.to_string(),
|
||||
status: ModelStatus::Loaded,
|
||||
vram_estimate_mb: None,
|
||||
}]
|
||||
} else {
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
|
||||
/// Smaller `context` wins — never advertise more headroom than the
|
||||
/// most-constrained operator can honour.
|
||||
fn tightest_limit(a: Option<ModelLimit>, b: Option<ModelLimit>) -> Option<ModelLimit> {
|
||||
match (a, b) {
|
||||
(None, x) | (x, None) => x,
|
||||
(Some(a), Some(b)) => Some(if b.context < a.context { b } else { a }),
|
||||
}
|
||||
}
|
||||
|
||||
/// Cheapest by (input, output) price — the federation "from" price.
|
||||
fn cheapest_cost(a: Option<ModelCost>, b: Option<ModelCost>) -> Option<ModelCost> {
|
||||
match (a, b) {
|
||||
(None, x) | (x, None) => x,
|
||||
(Some(a), Some(b)) => Some(if (b.input, b.output) < (a.input, a.output) {
|
||||
b
|
||||
} else {
|
||||
a
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::state::CortexTopology;
|
||||
|
||||
fn entry(id: &str, loaded: bool, feasible: bool) -> CortexModelEntry {
|
||||
CortexModelEntry {
|
||||
id: id.into(),
|
||||
object: "model".into(),
|
||||
created: 0,
|
||||
owned_by: "helexa".into(),
|
||||
loaded,
|
||||
feasible_on: if feasible || loaded {
|
||||
vec!["some-neuron".into()]
|
||||
} else {
|
||||
vec![]
|
||||
},
|
||||
locations: vec![],
|
||||
capabilities: vec![],
|
||||
limit: None,
|
||||
cost: None,
|
||||
tool_call: false,
|
||||
reasoning: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn cortex(reachable: bool, entries: Vec<CortexModelEntry>) -> CortexTopology {
|
||||
CortexTopology {
|
||||
reachable,
|
||||
consecutive_failures: 0,
|
||||
last_poll: None,
|
||||
healthy_nodes: 1,
|
||||
total_nodes: 1,
|
||||
models: entries.into_iter().map(|e| (e.id.clone(), e)).collect(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dedupes_and_merges_availability_across_cortexes() {
|
||||
let mut topo = HashMap::new();
|
||||
// c-a: model loaded. c-b: same model only cold-loadable.
|
||||
topo.insert("c-a".into(), cortex(true, vec![entry("m", true, true)]));
|
||||
topo.insert("c-b".into(), cortex(true, vec![entry("m", false, true)]));
|
||||
|
||||
let out = aggregate_models(&topo);
|
||||
assert_eq!(out.len(), 1, "duplicate model id collapses to one");
|
||||
let m = &out[0];
|
||||
assert!(m.loaded, "loaded somewhere → loaded");
|
||||
// feasible_on re-tiered to operator names, both present, sorted.
|
||||
assert_eq!(m.feasible_on, vec!["c-a".to_string(), "c-b".to_string()]);
|
||||
// Only the loaded operator contributes a location, named by operator.
|
||||
assert_eq!(m.locations.len(), 1);
|
||||
assert_eq!(m.locations[0].node, "c-a");
|
||||
assert_eq!(m.locations[0].vram_estimate_mb, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unreachable_cortex_is_excluded() {
|
||||
let mut topo = HashMap::new();
|
||||
topo.insert("up".into(), cortex(true, vec![entry("m", true, true)]));
|
||||
topo.insert(
|
||||
"down".into(),
|
||||
cortex(false, vec![entry("other", true, true)]),
|
||||
);
|
||||
let out = aggregate_models(&topo);
|
||||
assert_eq!(out.len(), 1);
|
||||
assert_eq!(out[0].id, "m");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn catalogue_only_infeasible_entries_are_hidden() {
|
||||
let mut topo = HashMap::new();
|
||||
topo.insert("c".into(), cortex(true, vec![entry("ghost", false, false)]));
|
||||
assert!(aggregate_models(&topo).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preserves_tightest_limit_and_cheapest_cost() {
|
||||
let mut a = entry("m", true, true);
|
||||
a.limit = Some(ModelLimit {
|
||||
context: 32_768,
|
||||
input: None,
|
||||
output: 4096,
|
||||
});
|
||||
a.cost = Some(ModelCost {
|
||||
input: 0.50,
|
||||
output: 1.50,
|
||||
cache_read: None,
|
||||
cache_write: None,
|
||||
});
|
||||
let mut b = entry("m", true, true);
|
||||
b.limit = Some(ModelLimit {
|
||||
context: 16_384, // tighter
|
||||
input: None,
|
||||
output: 4096,
|
||||
});
|
||||
b.cost = Some(ModelCost {
|
||||
input: 0.20, // cheaper
|
||||
output: 0.80,
|
||||
cache_read: None,
|
||||
cache_write: None,
|
||||
});
|
||||
|
||||
let mut topo = HashMap::new();
|
||||
topo.insert("c-a".into(), cortex(true, vec![a]));
|
||||
topo.insert("c-b".into(), cortex(true, vec![b]));
|
||||
|
||||
let out = aggregate_models(&topo);
|
||||
assert_eq!(out.len(), 1);
|
||||
assert_eq!(out[0].limit.as_ref().unwrap().context, 16_384);
|
||||
assert_eq!(out[0].cost.as_ref().unwrap().input, 0.20);
|
||||
}
|
||||
}
|
||||
@@ -56,6 +56,21 @@ pub struct CortexEndpoint {
|
||||
/// (#73). `None` → no region preference applies to this cortex.
|
||||
#[serde(default)]
|
||||
pub region: Option<String>,
|
||||
/// Path to a PEM trust anchor that **enrols** this cortex (#74): the
|
||||
/// expected CA (or self-signed cert) the cortex's TLS cert must chain
|
||||
/// to. When set on an `https://` endpoint, the router builds a client
|
||||
/// that trusts **only** this anchor (platform roots disabled), so the
|
||||
/// outbound router→cortex hop — which carries the client's bearer —
|
||||
/// reaches a cert the router was told to expect, and a rogue endpoint
|
||||
/// presenting any other (even publicly-valid) cert is rejected at the
|
||||
/// TLS handshake. A rejected handshake surfaces as a connection error,
|
||||
/// which the poller (#72) already treats as unreachable → excluded.
|
||||
///
|
||||
/// `None` → standard platform-root validation (use for cortexes behind
|
||||
/// a publicly-trusted cert, or plaintext `http://` on a private network
|
||||
/// where the WireGuard mesh is the trust boundary).
|
||||
#[serde(default)]
|
||||
pub tls_ca: Option<String>,
|
||||
}
|
||||
|
||||
impl RouterConfig {
|
||||
|
||||
@@ -54,10 +54,10 @@ pub async fn select_cortexes(state: &RouterState, model: &str) -> Selection {
|
||||
let mut known_anywhere = false;
|
||||
|
||||
for (name, t) in topo.iter() {
|
||||
let Some(status) = t.models.get(model) else {
|
||||
let Some(entry) = t.models.get(model) else {
|
||||
continue;
|
||||
};
|
||||
if !status.feasible {
|
||||
if !crate::state::entry_feasible(entry) {
|
||||
continue;
|
||||
}
|
||||
// Known even via an unreachable cortex's last-good poll — lets us
|
||||
@@ -74,7 +74,7 @@ pub async fn select_cortexes(state: &RouterState, model: &str) -> Selection {
|
||||
_ => false,
|
||||
};
|
||||
ranked.push(Ranked {
|
||||
loaded: status.loaded,
|
||||
loaded: entry.loaded,
|
||||
region_match,
|
||||
healthy_nodes: t.healthy_nodes,
|
||||
endpoint: (*ep).clone(),
|
||||
@@ -159,10 +159,16 @@ pub async fn dispatch(
|
||||
// genuine HTTP response (any status — including cortex's #63 429/400)
|
||||
// is returned verbatim and never retried away.
|
||||
for ep in &candidates {
|
||||
// A candidate whose pinned TLS client failed to build (#74) is
|
||||
// disabled — skip it and fail over, same as an unreachable cortex.
|
||||
let Some(client) = state.client_for(&ep.name) else {
|
||||
tracing::warn!(cortex = %ep.name, "no TLS client (disabled); skipping candidate");
|
||||
continue;
|
||||
};
|
||||
let url = format!("{}{}", ep.endpoint, path);
|
||||
tracing::info!(cortex = %ep.name, url = %url, model = %model, "dispatching");
|
||||
match helexa_stream::forward_streaming(
|
||||
&state.http_client,
|
||||
client,
|
||||
&url,
|
||||
headers.clone(),
|
||||
body.clone(),
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
use crate::dispatch;
|
||||
use crate::state::RouterState;
|
||||
use crate::{catalogue, dispatch};
|
||||
use axum::body::Bytes;
|
||||
use axum::http::HeaderMap;
|
||||
use axum::response::Response;
|
||||
use axum::{Json, Router, extract::State, routing::get, routing::post};
|
||||
use cortex_core::openai::ModelsResponse;
|
||||
use serde_json::{Value, json};
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -76,12 +75,15 @@ async fn health(State(state): State<Arc<RouterState>>) -> Json<Value> {
|
||||
}))
|
||||
}
|
||||
|
||||
/// `GET /v1/models` — empty catalogue stub. The real cross-operator union
|
||||
/// (catalogue × topology feasibility, aggregated from each cortex) is the
|
||||
/// federation-catalogue issue (#75).
|
||||
async fn list_models() -> Json<ModelsResponse> {
|
||||
Json(ModelsResponse {
|
||||
object: "list".into(),
|
||||
data: vec![],
|
||||
})
|
||||
/// `GET /v1/models` — the federation catalogue (#75): the deduped union of
|
||||
/// every reachable cortex's `/v1/models`, so a client doing discovery
|
||||
/// against the router resolves the whole federation without knowing about
|
||||
/// operators or cortexes.
|
||||
async fn list_models(State(state): State<Arc<RouterState>>) -> Json<Value> {
|
||||
let topo = state.topology.read().await;
|
||||
let data: Vec<Value> = catalogue::aggregate_models(&topo)
|
||||
.iter()
|
||||
.map(|e| json!(e))
|
||||
.collect();
|
||||
Json(json!({ "object": "list", "data": data }))
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
//! on capacity (epic #69). A background [`poller`] keeps a live
|
||||
//! per-cortex topology (#72) that the dispatcher (#73) will route on.
|
||||
|
||||
pub mod catalogue;
|
||||
pub mod config;
|
||||
pub mod dispatch;
|
||||
pub mod error;
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
//! then flipped unhealthy and excluded from routing; it recovers on the
|
||||
//! next successful poll.
|
||||
|
||||
use crate::state::{RouterModelStatus, RouterState};
|
||||
use crate::state::RouterState;
|
||||
use chrono::Utc;
|
||||
use cortex_core::node::CortexModelEntry;
|
||||
use serde::Deserialize;
|
||||
@@ -62,7 +62,19 @@ pub async fn poll_once(state: &RouterState) {
|
||||
/// reachability on its own (a cortex serving `/v1/models` is routable even
|
||||
/// if `/health` momentarily isn't).
|
||||
async fn poll_cortex(state: &RouterState, name: &str, endpoint: &str) {
|
||||
let models = fetch_models(state, endpoint).await;
|
||||
// A cortex whose pinned TLS client failed to build (#74) is disabled:
|
||||
// there is no client to poll with, so it stays unreachable.
|
||||
let Some(client) = state.client_for(name) else {
|
||||
let mut topo = state.topology.write().await;
|
||||
if let Some(entry) = topo.get_mut(name) {
|
||||
entry.consecutive_failures = entry.consecutive_failures.saturating_add(1);
|
||||
entry.reachable = false;
|
||||
}
|
||||
tracing::warn!(cortex = name, "no TLS client (disabled); skipping poll");
|
||||
return;
|
||||
};
|
||||
|
||||
let models = fetch_models(client, endpoint).await;
|
||||
|
||||
let mut topo = state.topology.write().await;
|
||||
let Some(entry) = topo.get_mut(name) else {
|
||||
@@ -71,19 +83,7 @@ async fn poll_cortex(state: &RouterState, name: &str, endpoint: &str) {
|
||||
|
||||
match models {
|
||||
Ok(models) => {
|
||||
entry.models = models
|
||||
.into_iter()
|
||||
.map(|m| {
|
||||
let feasible = m.loaded || !m.feasible_on.is_empty();
|
||||
(
|
||||
m.id,
|
||||
RouterModelStatus {
|
||||
loaded: m.loaded,
|
||||
feasible,
|
||||
},
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
entry.models = models.into_iter().map(|m| (m.id.clone(), m)).collect();
|
||||
entry.reachable = true;
|
||||
entry.consecutive_failures = 0;
|
||||
entry.last_poll = Some(Utc::now());
|
||||
@@ -106,7 +106,7 @@ async fn poll_cortex(state: &RouterState, name: &str, endpoint: &str) {
|
||||
drop(topo);
|
||||
|
||||
// Best-effort health (node counts). Never flips reachability.
|
||||
if let Some((healthy, total)) = fetch_health(state, endpoint).await {
|
||||
if let Some((healthy, total)) = fetch_health(client, endpoint).await {
|
||||
let mut topo = state.topology.write().await;
|
||||
if let Some(entry) = topo.get_mut(name) {
|
||||
entry.healthy_nodes = healthy;
|
||||
@@ -117,12 +117,11 @@ async fn poll_cortex(state: &RouterState, name: &str, endpoint: &str) {
|
||||
|
||||
/// GET `/v1/models`, returning the parsed entries or a short failure reason.
|
||||
async fn fetch_models(
|
||||
state: &RouterState,
|
||||
client: &reqwest::Client,
|
||||
endpoint: &str,
|
||||
) -> Result<Vec<CortexModelEntry>, &'static str> {
|
||||
let url = format!("{endpoint}/v1/models");
|
||||
let resp = state
|
||||
.http_client
|
||||
let resp = client
|
||||
.get(&url)
|
||||
.timeout(POLL_TIMEOUT)
|
||||
.send()
|
||||
@@ -140,15 +139,9 @@ async fn fetch_models(
|
||||
|
||||
/// GET `/health`, returning `(healthy, total)` node counts. `None` on any
|
||||
/// failure — the caller leaves the previous counts in place.
|
||||
async fn fetch_health(state: &RouterState, endpoint: &str) -> Option<(u32, u32)> {
|
||||
async fn fetch_health(client: &reqwest::Client, endpoint: &str) -> Option<(u32, u32)> {
|
||||
let url = format!("{endpoint}/health");
|
||||
let resp = state
|
||||
.http_client
|
||||
.get(&url)
|
||||
.timeout(POLL_TIMEOUT)
|
||||
.send()
|
||||
.await
|
||||
.ok()?;
|
||||
let resp = client.get(&url).timeout(POLL_TIMEOUT).send().await.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
return None;
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
use crate::config::{CortexEndpoint, RouterConfig};
|
||||
use chrono::{DateTime, Utc};
|
||||
use cortex_core::node::CortexModelEntry;
|
||||
use std::collections::HashMap;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::RwLock;
|
||||
@@ -14,8 +15,13 @@ use tokio::sync::RwLock;
|
||||
pub struct RouterState {
|
||||
/// Downstream cortex endpoints, as configured.
|
||||
pub cortexes: Vec<CortexEndpoint>,
|
||||
/// Shared client for polling (and proxying to) cortexes.
|
||||
pub http_client: reqwest::Client,
|
||||
/// Per-cortex HTTP client, keyed by cortex name (#74). A cortex enrolled
|
||||
/// with a `tls_ca` gets a client that trusts only that anchor; others
|
||||
/// get a default client. A cortex whose `tls_ca` failed to load is
|
||||
/// **absent** here — `client_for` returns `None` and it is never
|
||||
/// polled or routed to (fail closed: a misconfigured pin must not
|
||||
/// silently fall back to unpinned TLS).
|
||||
clients: HashMap<String, reqwest::Client>,
|
||||
/// This router instance's region, for dispatch geo affinity (#73).
|
||||
pub region: Option<String>,
|
||||
/// How often the poller refreshes the topology.
|
||||
@@ -42,19 +48,16 @@ pub struct CortexTopology {
|
||||
/// load signal; #73 refines headroom). 0/0 until first health poll.
|
||||
pub healthy_nodes: u32,
|
||||
pub total_nodes: u32,
|
||||
/// Per-model serveability, keyed by model id, from `/v1/models`.
|
||||
pub models: HashMap<String, RouterModelStatus>,
|
||||
/// The cortex's full `/v1/models` entries, keyed by model id. Stored
|
||||
/// whole (not distilled to a loaded/feasible bool) so the federation
|
||||
/// catalogue (#75) can preserve per-model `limit`/`cost`/capabilities.
|
||||
pub models: HashMap<String, CortexModelEntry>,
|
||||
}
|
||||
|
||||
/// What a cortex can do with one model, distilled from its `/v1/models`
|
||||
/// entry to the two facts the router routes on.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RouterModelStatus {
|
||||
/// The model is loaded on at least one of the cortex's neurons.
|
||||
pub loaded: bool,
|
||||
/// The cortex can serve it — loaded now, or feasible to cold-load
|
||||
/// (catalogue × topology says some neuron can host it).
|
||||
pub feasible: bool,
|
||||
/// Whether a cortex can serve this model — loaded now, or feasible to
|
||||
/// cold-load (its catalogue × topology says some neuron can host it).
|
||||
pub fn entry_feasible(entry: &CortexModelEntry) -> bool {
|
||||
entry.loaded || !entry.feasible_on.is_empty()
|
||||
}
|
||||
|
||||
impl RouterState {
|
||||
@@ -65,15 +68,42 @@ impl RouterState {
|
||||
.map(|c| (c.name.clone(), CortexTopology::default()))
|
||||
.collect();
|
||||
|
||||
// One client per cortex. A `tls_ca` that fails to load omits the
|
||||
// cortex from the map (fail closed) rather than degrading to an
|
||||
// unpinned client.
|
||||
let mut clients = HashMap::new();
|
||||
for c in &config.cortexes {
|
||||
match build_client(c.tls_ca.as_deref()) {
|
||||
Ok(client) => {
|
||||
clients.insert(c.name.clone(), client);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!(
|
||||
cortex = %c.name,
|
||||
tls_ca = c.tls_ca.as_deref().unwrap_or(""),
|
||||
error = %e,
|
||||
"failed to build pinned TLS client; cortex disabled (fail closed)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Self {
|
||||
cortexes: config.cortexes.clone(),
|
||||
http_client: reqwest::Client::new(),
|
||||
clients,
|
||||
region: config.router.region.clone(),
|
||||
poll_interval: Duration::from_secs(config.router.poll_interval_secs),
|
||||
topology: RwLock::new(topology),
|
||||
}
|
||||
}
|
||||
|
||||
/// The HTTP client to use for `name`, or `None` if the cortex is
|
||||
/// disabled (its `tls_ca` failed to load). Callers must treat `None` as
|
||||
/// "not routable / not pollable".
|
||||
pub fn client_for(&self, name: &str) -> Option<&reqwest::Client> {
|
||||
self.clients.get(name)
|
||||
}
|
||||
|
||||
/// Names of reachable cortexes that can serve `model_id` (loaded or
|
||||
/// feasible to cold-load). Groundwork for capacity-aware dispatch (#73);
|
||||
/// unreachable cortexes are excluded by construction.
|
||||
@@ -81,8 +111,34 @@ impl RouterState {
|
||||
let topo = self.topology.read().await;
|
||||
topo.iter()
|
||||
.filter(|(_, t)| t.reachable)
|
||||
.filter(|(_, t)| t.models.get(model_id).is_some_and(|m| m.feasible))
|
||||
.filter(|(_, t)| t.models.get(model_id).is_some_and(entry_feasible))
|
||||
.map(|(name, _)| name.clone())
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a cortex HTTP client. With `tls_ca` set, the client trusts **only**
|
||||
/// that PEM anchor (platform roots disabled) — pinning the router→cortex hop
|
||||
/// to an enrolled cert (#74). Without it, standard platform-root validation.
|
||||
pub fn build_client(tls_ca: Option<&str>) -> Result<reqwest::Client, BuildClientError> {
|
||||
let mut builder = reqwest::Client::builder();
|
||||
if let Some(path) = tls_ca {
|
||||
let pem = std::fs::read(path).map_err(|e| BuildClientError::Read(path.to_string(), e))?;
|
||||
let cert = reqwest::Certificate::from_pem(&pem).map_err(BuildClientError::Parse)?;
|
||||
builder = builder
|
||||
.tls_built_in_root_certs(false)
|
||||
.add_root_certificate(cert);
|
||||
}
|
||||
builder.build().map_err(BuildClientError::Build)
|
||||
}
|
||||
|
||||
/// Why a cortex's pinned client could not be built (→ cortex disabled).
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum BuildClientError {
|
||||
#[error("reading TLS anchor '{0}'")]
|
||||
Read(String, #[source] std::io::Error),
|
||||
#[error("parsing TLS anchor PEM")]
|
||||
Parse(#[source] reqwest::Error),
|
||||
#[error("building HTTP client")]
|
||||
Build(#[source] reqwest::Error),
|
||||
}
|
||||
|
||||
132
crates/helexa-router/tests/catalogue.rs
Normal file
132
crates/helexa-router/tests/catalogue.rs
Normal file
@@ -0,0 +1,132 @@
|
||||
//! End-to-end federation-catalogue test for #75: poll two mock cortexes
|
||||
//! that overlap on a model, then `GET /v1/models` on the router and verify
|
||||
//! the deduped union with merged availability and preserved limit/cost.
|
||||
|
||||
use axum::Router;
|
||||
use axum::routing::get;
|
||||
use helexa_router::config::{CortexEndpoint, RouterConfig};
|
||||
use helexa_router::poller::poll_once;
|
||||
use helexa_router::state::RouterState;
|
||||
use serde_json::{Value, json};
|
||||
use std::sync::Arc;
|
||||
use tokio::net::TcpListener;
|
||||
|
||||
/// Spawn a mock cortex serving the given `/v1/models` `data` array.
|
||||
async fn spawn_cortex(models: Value) -> String {
|
||||
let models = Arc::new(models);
|
||||
let app = Router::new()
|
||||
.route(
|
||||
"/v1/models",
|
||||
get({
|
||||
let models = Arc::clone(&models);
|
||||
move || {
|
||||
let models = Arc::clone(&models);
|
||||
async move { axum::Json(json!({ "object": "list", "data": &*models })) }
|
||||
}
|
||||
}),
|
||||
)
|
||||
.route(
|
||||
"/health",
|
||||
get(|| async { axum::Json(json!({"status":"ok","nodes":{"healthy":1,"total":1}})) }),
|
||||
);
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
format!("http://{addr}")
|
||||
}
|
||||
|
||||
/// Spawn the router (with poller) wired to the given cortex endpoints, and
|
||||
/// poll once synchronously so the topology is populated before we query.
|
||||
async fn spawn_router(cortexes: Vec<CortexEndpoint>) -> String {
|
||||
let cfg = RouterConfig {
|
||||
cortexes,
|
||||
..Default::default()
|
||||
};
|
||||
let state = Arc::new(RouterState::from_config(&cfg));
|
||||
poll_once(&state).await; // deterministic: fill topology now
|
||||
|
||||
let app = helexa_router::build_app(Arc::clone(&state));
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
format!("http://{addr}")
|
||||
}
|
||||
|
||||
fn model(id: &str, loaded: bool, feasible_on: &[&str], ctx: u64, input_cost: f64) -> Value {
|
||||
json!({
|
||||
"id": id,
|
||||
"object": "model",
|
||||
"created": 0,
|
||||
"owned_by": "helexa",
|
||||
"loaded": loaded,
|
||||
"feasible_on": feasible_on,
|
||||
"locations": [],
|
||||
"limit": { "context": ctx, "output": 4096 },
|
||||
"cost": { "input": input_cost, "output": input_cost * 3.0 }
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn federation_catalogue_dedupes_and_preserves_limit_cost() {
|
||||
// cortex A: "shared" loaded (ctx 32768, $0.50) + "only-a" loaded.
|
||||
let a = spawn_cortex(json!([
|
||||
model("shared", true, &["beast"], 32_768, 0.50),
|
||||
model("only-a", true, &["beast"], 8_192, 1.00),
|
||||
]))
|
||||
.await;
|
||||
// cortex B: "shared" cold-loadable, tighter ctx (16384), cheaper ($0.20).
|
||||
let b = spawn_cortex(json!([model("shared", false, &["benjy"], 16_384, 0.20)])).await;
|
||||
|
||||
let router = spawn_router(vec![
|
||||
CortexEndpoint {
|
||||
name: "op-a".into(),
|
||||
endpoint: a,
|
||||
region: None,
|
||||
tls_ca: None,
|
||||
},
|
||||
CortexEndpoint {
|
||||
name: "op-b".into(),
|
||||
endpoint: b,
|
||||
region: None,
|
||||
tls_ca: None,
|
||||
},
|
||||
])
|
||||
.await;
|
||||
|
||||
let body: Value = reqwest::get(format!("{router}/v1/models"))
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(body["object"], "list");
|
||||
let data = body["data"].as_array().unwrap();
|
||||
// Deduped union: "shared" once + "only-a".
|
||||
assert_eq!(data.len(), 2);
|
||||
|
||||
let shared = data.iter().find(|m| m["id"] == "shared").unwrap();
|
||||
// Loaded somewhere (op-a) → loaded.
|
||||
assert_eq!(shared["loaded"], true);
|
||||
// feasible_on re-tiered to operator names, both present, sorted.
|
||||
let feasible: Vec<&str> = shared["feasible_on"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|v| v.as_str().unwrap())
|
||||
.collect();
|
||||
assert_eq!(feasible, vec!["op-a", "op-b"]);
|
||||
// Tightest limit (16384) and cheapest cost ($0.20) win.
|
||||
assert_eq!(shared["limit"]["context"], 16_384);
|
||||
assert_eq!(shared["cost"]["input"], 0.20);
|
||||
// Loaded location named by operator, no neuron VRAM leaked.
|
||||
let locs = shared["locations"].as_array().unwrap();
|
||||
assert_eq!(locs.len(), 1);
|
||||
assert_eq!(locs[0]["node"], "op-a");
|
||||
|
||||
assert!(data.iter().any(|m| m["id"] == "only-a"));
|
||||
}
|
||||
@@ -12,13 +12,36 @@ use axum::http::{HeaderMap, StatusCode};
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::routing::post;
|
||||
use axum::{Json, Router};
|
||||
use cortex_core::node::CortexModelEntry;
|
||||
use helexa_router::config::{CortexEndpoint, RouterConfig};
|
||||
use helexa_router::dispatch::{Selection, dispatch, select_cortexes};
|
||||
use helexa_router::state::{CortexTopology, RouterModelStatus, RouterState};
|
||||
use helexa_router::state::{CortexTopology, RouterState};
|
||||
use serde_json::{Value, json};
|
||||
use std::collections::HashMap;
|
||||
use tokio::net::TcpListener;
|
||||
|
||||
/// A minimal `CortexModelEntry` for MODEL with the given serveability.
|
||||
fn model_entry(loaded: bool, feasible: bool) -> CortexModelEntry {
|
||||
CortexModelEntry {
|
||||
id: MODEL.into(),
|
||||
object: "model".into(),
|
||||
created: 0,
|
||||
owned_by: "helexa".into(),
|
||||
loaded,
|
||||
feasible_on: if feasible || loaded {
|
||||
vec!["n".into()]
|
||||
} else {
|
||||
vec![]
|
||||
},
|
||||
locations: vec![],
|
||||
capabilities: vec![],
|
||||
limit: None,
|
||||
cost: None,
|
||||
tool_call: false,
|
||||
reasoning: false,
|
||||
}
|
||||
}
|
||||
|
||||
const MODEL: &str = "Qwen/Qwen3-Coder-30B";
|
||||
|
||||
// ── Mock cortex backend ──────────────────────────────────────────────
|
||||
@@ -91,7 +114,7 @@ async fn set_topology(
|
||||
) {
|
||||
let mut topo = state.topology.write().await;
|
||||
let mut models = HashMap::new();
|
||||
models.insert(MODEL.to_string(), RouterModelStatus { loaded, feasible });
|
||||
models.insert(MODEL.to_string(), model_entry(loaded, feasible));
|
||||
topo.insert(
|
||||
name.to_string(),
|
||||
CortexTopology {
|
||||
@@ -110,6 +133,7 @@ fn ep(name: &str, endpoint: &str, region: Option<&str>) -> CortexEndpoint {
|
||||
name: name.into(),
|
||||
endpoint: endpoint.into(),
|
||||
region: region.map(str::to_string),
|
||||
tls_ca: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -32,11 +32,13 @@ async fn health_reports_configured_cortex_count() {
|
||||
name: "a".into(),
|
||||
endpoint: "https://a.example.com".into(),
|
||||
region: None,
|
||||
tls_ca: None,
|
||||
},
|
||||
CortexEndpoint {
|
||||
name: "b".into(),
|
||||
endpoint: "https://b.example.com".into(),
|
||||
region: None,
|
||||
tls_ca: None,
|
||||
},
|
||||
])
|
||||
.await;
|
||||
|
||||
210
crates/helexa-router/tests/tls.rs
Normal file
210
crates/helexa-router/tests/tls.rs
Normal file
@@ -0,0 +1,210 @@
|
||||
//! Outbound TLS pinning tests for #74.
|
||||
//!
|
||||
//! Proves the router, as a TLS client to cortexes, reaches a cortex
|
||||
//! presenting its **enrolled** cert and rejects one presenting an
|
||||
//! unexpected (or untrusted) cert — and that a rejected handshake flows
|
||||
//! through the existing reachability path (#72) to exclude the cortex.
|
||||
//!
|
||||
//! A minimal `tokio-rustls` HTTPS server presents a self-signed cert; the
|
||||
//! router's `reqwest` client (native-tls) validates against the PEM anchor
|
||||
//! enrolled in config. Server (rustls) and client (native-tls) interoperate
|
||||
//! at the protocol level — what matters is the trust decision.
|
||||
|
||||
use helexa_router::config::{CortexEndpoint, RouterConfig};
|
||||
use helexa_router::poller::poll_once;
|
||||
use helexa_router::state::{RouterState, build_client};
|
||||
use std::io::Write;
|
||||
use std::sync::Arc;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_rustls::TlsAcceptor;
|
||||
|
||||
/// A self-signed cert: PEM (for the reqwest pin file) + DER cert/key (for
|
||||
/// the rustls server).
|
||||
struct TestCert {
|
||||
cert_pem: String,
|
||||
cert_der: rustls::pki_types::CertificateDer<'static>,
|
||||
key_der: Vec<u8>,
|
||||
}
|
||||
|
||||
fn make_cert() -> TestCert {
|
||||
let key = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]).unwrap();
|
||||
TestCert {
|
||||
cert_pem: key.cert.pem(),
|
||||
cert_der: key.cert.der().clone(),
|
||||
key_der: key.key_pair.serialize_der(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Write a cert PEM to a unique temp file (named by `tag`) and return the
|
||||
/// path. `tag` is caller-unique (we use the bound port), so no randomness.
|
||||
fn write_pem(tag: &str, pem: &str) -> String {
|
||||
let path = std::env::temp_dir().join(format!("helexa-router-tls-{tag}.pem"));
|
||||
let mut f = std::fs::File::create(&path).unwrap();
|
||||
f.write_all(pem.as_bytes()).unwrap();
|
||||
path.to_string_lossy().into_owned()
|
||||
}
|
||||
|
||||
/// Spawn a minimal HTTPS server presenting `cert`, answering every request
|
||||
/// with a canned `/v1/models`-shaped 200. Returns its `https://` base URL.
|
||||
async fn spawn_https(cert: &TestCert) -> String {
|
||||
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
|
||||
|
||||
let key = rustls::pki_types::PrivateKeyDer::Pkcs8(rustls::pki_types::PrivatePkcs8KeyDer::from(
|
||||
cert.key_der.clone(),
|
||||
));
|
||||
let config = rustls::ServerConfig::builder()
|
||||
.with_no_client_auth()
|
||||
.with_single_cert(vec![cert.cert_der.clone()], key)
|
||||
.unwrap();
|
||||
let acceptor = TlsAcceptor::from(Arc::new(config));
|
||||
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((stream, _)) = listener.accept().await else {
|
||||
continue;
|
||||
};
|
||||
let acceptor = acceptor.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Ok(mut tls) = acceptor.accept(stream).await {
|
||||
let mut buf = [0u8; 2048];
|
||||
let _ = tls.read(&mut buf).await; // consume request line/headers
|
||||
let body = "{\"object\":\"list\",\"data\":[]}";
|
||||
let resp = format!(
|
||||
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
|
||||
body.len(),
|
||||
body
|
||||
);
|
||||
let _ = tls.write_all(resp.as_bytes()).await;
|
||||
let _ = tls.shutdown().await;
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
format!("https://{addr}")
|
||||
}
|
||||
|
||||
fn tag_for(url: &str) -> String {
|
||||
url.rsplit(':').next().unwrap_or("0").to_string()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pinned_client_accepts_enrolled_cert_and_rejects_others() {
|
||||
let server_cert = make_cert();
|
||||
let other_cert = make_cert();
|
||||
let url = spawn_https(&server_cert).await;
|
||||
let tag = tag_for(&url);
|
||||
|
||||
let good_pin = write_pem(&format!("{tag}-good"), &server_cert.cert_pem);
|
||||
let bad_pin = write_pem(&format!("{tag}-bad"), &other_cert.cert_pem);
|
||||
|
||||
// Enrolled with the server's own cert → handshake trusted → 200.
|
||||
let good = build_client(Some(&good_pin)).unwrap();
|
||||
let resp = good.get(format!("{url}/v1/models")).send().await;
|
||||
assert!(resp.is_ok(), "enrolled cert must be accepted: {resp:?}");
|
||||
assert_eq!(resp.unwrap().status(), 200);
|
||||
|
||||
// Enrolled with a different cert → server's cert is unexpected → reject.
|
||||
let bad = build_client(Some(&bad_pin)).unwrap();
|
||||
assert!(
|
||||
bad.get(format!("{url}/v1/models")).send().await.is_err(),
|
||||
"unexpected cert must be rejected"
|
||||
);
|
||||
|
||||
// No enrollment (default platform roots) → self-signed cert untrusted.
|
||||
let default = build_client(None).unwrap();
|
||||
assert!(
|
||||
default
|
||||
.get(format!("{url}/v1/models"))
|
||||
.send()
|
||||
.await
|
||||
.is_err(),
|
||||
"un-enrolled self-signed cert must be rejected by default roots"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn poller_excludes_cortex_with_unexpected_cert() {
|
||||
let server_cert = make_cert();
|
||||
let other_cert = make_cert();
|
||||
let url = spawn_https(&server_cert).await;
|
||||
let tag = tag_for(&url);
|
||||
|
||||
let good_pin = write_pem(&format!("{tag}-pgood"), &server_cert.cert_pem);
|
||||
let bad_pin = write_pem(&format!("{tag}-pbad"), &other_cert.cert_pem);
|
||||
|
||||
// Cortex A enrolled correctly → reachable. Cortex B enrolled with the
|
||||
// wrong cert → TLS handshake fails → excluded.
|
||||
let cfg = RouterConfig {
|
||||
cortexes: vec![
|
||||
CortexEndpoint {
|
||||
name: "good".into(),
|
||||
endpoint: url.clone(),
|
||||
region: None,
|
||||
tls_ca: Some(good_pin),
|
||||
},
|
||||
CortexEndpoint {
|
||||
name: "bad".into(),
|
||||
endpoint: url.clone(),
|
||||
region: None,
|
||||
tls_ca: Some(bad_pin),
|
||||
},
|
||||
],
|
||||
..Default::default()
|
||||
};
|
||||
let state = RouterState::from_config(&cfg);
|
||||
poll_once(&state).await;
|
||||
|
||||
let topo = state.topology.read().await;
|
||||
assert!(
|
||||
topo["good"].reachable,
|
||||
"correctly-enrolled cortex reachable"
|
||||
);
|
||||
assert!(
|
||||
!topo["bad"].reachable,
|
||||
"cortex presenting an unexpected cert is excluded"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn misconfigured_pin_disables_cortex_fail_closed() {
|
||||
// A `tls_ca` pointing at a nonexistent file must NOT fall back to an
|
||||
// unpinned client — the cortex is disabled entirely.
|
||||
let cfg = RouterConfig {
|
||||
cortexes: vec![
|
||||
CortexEndpoint {
|
||||
name: "broken".into(),
|
||||
endpoint: "https://127.0.0.1:1".into(),
|
||||
region: None,
|
||||
tls_ca: Some("/no/such/anchor.pem".into()),
|
||||
},
|
||||
CortexEndpoint {
|
||||
name: "plain".into(),
|
||||
endpoint: "http://127.0.0.1:1".into(),
|
||||
region: None,
|
||||
tls_ca: None,
|
||||
},
|
||||
],
|
||||
..Default::default()
|
||||
};
|
||||
let state = RouterState::from_config(&cfg);
|
||||
assert!(
|
||||
state.client_for("broken").is_none(),
|
||||
"a cortex with an unloadable pin is disabled (fail closed)"
|
||||
);
|
||||
assert!(
|
||||
state.client_for("plain").is_some(),
|
||||
"an un-pinned cortex still gets a client"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_client_rejects_garbage_pem() {
|
||||
let path = write_pem(
|
||||
"garbage",
|
||||
"-----BEGIN CERTIFICATE-----\nnope\n-----END CERTIFICATE-----",
|
||||
);
|
||||
assert!(build_client(Some(&path)).is_err());
|
||||
}
|
||||
@@ -9,7 +9,7 @@ use axum::routing::get;
|
||||
use axum::{Json, Router};
|
||||
use helexa_router::config::{CortexEndpoint, RouterConfig};
|
||||
use helexa_router::poller::{POLL_FAILURE_THRESHOLD, poll_once};
|
||||
use helexa_router::state::RouterState;
|
||||
use helexa_router::state::{RouterState, entry_feasible};
|
||||
use serde_json::{Value, json};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
@@ -83,6 +83,7 @@ fn state_for(name: &str, endpoint: &str) -> RouterState {
|
||||
name: name.into(),
|
||||
endpoint: endpoint.into(),
|
||||
region: None,
|
||||
tls_ca: None,
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
@@ -103,11 +104,12 @@ async fn poll_builds_live_topology() {
|
||||
assert!(c1.last_poll.is_some());
|
||||
assert_eq!((c1.healthy_nodes, c1.total_nodes), (2, 3));
|
||||
|
||||
// Loaded model: loaded + feasible. Catalogue-only model: feasible only.
|
||||
// Loaded model: loaded + feasible. Catalogue-only model: feasible only
|
||||
// (not loaded, but feasible_on non-empty).
|
||||
let coder = c1.models.get("Qwen/Qwen3-Coder-30B").unwrap();
|
||||
assert!(coder.loaded && coder.feasible);
|
||||
assert!(coder.loaded && entry_feasible(coder));
|
||||
let vl = c1.models.get("Qwen/Qwen3-VL-8B").unwrap();
|
||||
assert!(!vl.loaded && vl.feasible);
|
||||
assert!(!vl.loaded && entry_feasible(vl));
|
||||
drop(topo);
|
||||
|
||||
// The routing helper sees both serveable models on the reachable cortex.
|
||||
|
||||
53
crates/helexa-upstream/Cargo.toml
Normal file
53
crates/helexa-upstream/Cargo.toml
Normal file
@@ -0,0 +1,53 @@
|
||||
[package]
|
||||
name = "helexa-upstream"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
|
||||
[[bin]]
|
||||
name = "helexa-upstream"
|
||||
path = "src/main.rs"
|
||||
|
||||
[lib]
|
||||
name = "helexa_upstream"
|
||||
path = "src/lib.rs"
|
||||
|
||||
[dependencies]
|
||||
tokio = { workspace = true }
|
||||
axum = { workspace = true }
|
||||
tower-http = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
figment = { workspace = true }
|
||||
anyhow = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
clap = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
tracing-subscriber = { workspace = true }
|
||||
chrono = { workspace = true }
|
||||
|
||||
# PostgreSQL — the mesh authority's system of record. Runtime query API
|
||||
# (not the compile-time `query!` macros) so the crate builds in CI without a
|
||||
# live database or a committed `.sqlx` offline cache; correctness is covered
|
||||
# by the gated integration tests. (Macro adoption is a later refinement once
|
||||
# a dev DB + offline cache exist.)
|
||||
sqlx = { version = "0.8", default-features = false, features = [
|
||||
"runtime-tokio",
|
||||
"tls-rustls",
|
||||
"postgres",
|
||||
"macros",
|
||||
"migrate",
|
||||
"chrono",
|
||||
"uuid",
|
||||
] }
|
||||
uuid = { version = "1", features = ["v4", "serde"] }
|
||||
sha2 = "0.10"
|
||||
subtle = "2.6"
|
||||
|
||||
# cortex-core for the shared #63 OpenAiError envelope on the authz surface.
|
||||
cortex-core = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
figment = { workspace = true, features = ["test"] }
|
||||
reqwest = { workspace = true }
|
||||
137
crates/helexa-upstream/migrations/0001_init.sql
Normal file
137
crates/helexa-upstream/migrations/0001_init.sql
Normal file
@@ -0,0 +1,137 @@
|
||||
-- helexa-upstream initial schema (#59): accounts, keys, ledger, top-up
|
||||
-- codes, served-usage. The mesh-level authority's system of record.
|
||||
--
|
||||
-- Token amounts are BIGINT (i64) throughout; the cortex EntitlementProvider
|
||||
-- carries u64 but mesh allocations sit comfortably inside i64 and Postgres
|
||||
-- has no unsigned type.
|
||||
|
||||
CREATE EXTENSION IF NOT EXISTS citext;
|
||||
CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
||||
|
||||
-- ── Users (web auth: email + password) ──────────────────────────────
|
||||
CREATE TABLE users (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
email CITEXT NOT NULL UNIQUE,
|
||||
password_hash TEXT NOT NULL, -- argon2id PHC string
|
||||
email_verified BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
-- Browser fingerprint captured at registration (#abuse). Best-effort,
|
||||
-- client-supplied; the primary signal for silent multi-account
|
||||
-- detection. NULL when the client could not produce one.
|
||||
registration_fingerprint TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE INDEX users_registration_fingerprint_idx
|
||||
ON users (registration_fingerprint)
|
||||
WHERE registration_fingerprint IS NOT NULL;
|
||||
|
||||
-- Single-use email tokens for verification and password reset. Only the
|
||||
-- sha256 of the emailed secret is stored.
|
||||
CREATE TABLE email_tokens (
|
||||
token_hash BYTEA PRIMARY KEY,
|
||||
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
kind TEXT NOT NULL CHECK (kind IN ('verify', 'reset')),
|
||||
expires_at TIMESTAMPTZ NOT NULL,
|
||||
consumed_at TIMESTAMPTZ
|
||||
);
|
||||
CREATE INDEX email_tokens_user_idx ON email_tokens (user_id);
|
||||
|
||||
-- ── Accounts (the billable allocation ledger) ───────────────────────
|
||||
CREATE TABLE accounts (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
owner_user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
allocation_total BIGINT NOT NULL DEFAULT 0,
|
||||
allocation_spent BIGINT NOT NULL DEFAULT 0,
|
||||
allocation_reserved BIGINT NOT NULL DEFAULT 0,
|
||||
-- 'deactivated' is the SILENT abuse flag: keys stop authorizing but no
|
||||
-- surface ever tells the user why (see resolve → 401).
|
||||
status TEXT NOT NULL DEFAULT 'active'
|
||||
CHECK (status IN ('active', 'deactivated')),
|
||||
-- This account shares a registration fingerprint with >= 1 other.
|
||||
fingerprint_flagged BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
-- The no-overshoot backstop to the atomic reserve UPDATE.
|
||||
CONSTRAINT accounts_no_overshoot
|
||||
CHECK (allocation_spent + allocation_reserved <= allocation_total),
|
||||
CONSTRAINT accounts_nonneg
|
||||
CHECK (allocation_spent >= 0 AND allocation_reserved >= 0)
|
||||
);
|
||||
CREATE INDEX accounts_owner_idx ON accounts (owner_user_id);
|
||||
|
||||
-- ── API keys (Principal.key_id = api_keys.id) ───────────────────────
|
||||
CREATE TABLE api_keys (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
account_id UUID NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
|
||||
key_hash BYTEA NOT NULL, -- sha256(raw key)
|
||||
key_prefix TEXT NOT NULL, -- non-secret display prefix
|
||||
label TEXT NOT NULL DEFAULT '',
|
||||
status TEXT NOT NULL DEFAULT 'active'
|
||||
CHECK (status IN ('active', 'archived')),
|
||||
-- Per-key sub-cap: 'hardcap' = absolute tokens; 'percent' = % of the
|
||||
-- account's allocation_total (resolved to an absolute at reserve time).
|
||||
limit_kind TEXT NOT NULL DEFAULT 'percent'
|
||||
CHECK (limit_kind IN ('percent', 'hardcap')),
|
||||
limit_value BIGINT NOT NULL DEFAULT 100,
|
||||
-- serde of cortex_core::entitlements::CapWindow (Balance | Rolling).
|
||||
cap_window JSONB NOT NULL DEFAULT '{"kind":"balance"}'::jsonb,
|
||||
-- Per-key running ledger (mirrors the account ledger; Balance semantics
|
||||
-- in this migration — rolling-window reset lands with the authz API).
|
||||
key_spent BIGINT NOT NULL DEFAULT 0,
|
||||
key_reserved BIGINT NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
CONSTRAINT api_keys_key_nonneg
|
||||
CHECK (key_spent >= 0 AND key_reserved >= 0)
|
||||
);
|
||||
-- A raw key resolves only while active; the hash is unique among active keys.
|
||||
CREATE UNIQUE INDEX api_keys_active_hash_idx
|
||||
ON api_keys (key_hash) WHERE status = 'active';
|
||||
CREATE INDEX api_keys_account_idx ON api_keys (account_id);
|
||||
|
||||
-- ── Reservations (reserve → settle/release) ─────────────────────────
|
||||
-- id is BIGSERIAL so it maps to the cortex Reservation.id (u64) verbatim,
|
||||
-- with the Postgres sequence as the sole global authority.
|
||||
CREATE TABLE reservations (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
account_id UUID NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
|
||||
key_id UUID NOT NULL REFERENCES api_keys(id) ON DELETE CASCADE,
|
||||
reserved BIGINT NOT NULL,
|
||||
actual BIGINT,
|
||||
state TEXT NOT NULL DEFAULT 'open'
|
||||
CHECK (state IN ('open', 'settled', 'released')),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
settled_at TIMESTAMPTZ
|
||||
);
|
||||
-- The sweeper scans open reservations by age.
|
||||
CREATE INDEX reservations_open_idx
|
||||
ON reservations (created_at) WHERE state = 'open';
|
||||
|
||||
-- ── Top-up codes (hybrid allocation) ────────────────────────────────
|
||||
CREATE TABLE top_up_codes (
|
||||
code_hash BYTEA PRIMARY KEY, -- sha256(raw code)
|
||||
value BIGINT NOT NULL, -- tokens this code grants
|
||||
denomination TEXT, -- human label (e.g. "small")
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
redeemed_by UUID REFERENCES accounts(id) ON DELETE SET NULL,
|
||||
redeemed_at TIMESTAMPTZ
|
||||
);
|
||||
|
||||
-- ── Served-usage ledger (#58 reconciliation) ────────────────────────
|
||||
-- Absolute per-(operator, account, key, period) served tokens, upserted by
|
||||
-- each cortex; reconciliation rolls these up for operator compensation.
|
||||
CREATE TABLE served_usage (
|
||||
operator_id TEXT NOT NULL,
|
||||
account_id UUID NOT NULL,
|
||||
key_id UUID NOT NULL,
|
||||
period DATE NOT NULL,
|
||||
served_tokens BIGINT NOT NULL DEFAULT 0,
|
||||
reconciled_at TIMESTAMPTZ,
|
||||
PRIMARY KEY (operator_id, account_id, key_id, period)
|
||||
);
|
||||
|
||||
-- ── Web sessions (DB-backed; alt/complement to stateless JWT) ───────
|
||||
CREATE TABLE sessions (
|
||||
token_hash BYTEA PRIMARY KEY, -- sha256(session token)
|
||||
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
expires_at TIMESTAMPTZ NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE INDEX sessions_user_idx ON sessions (user_id);
|
||||
284
crates/helexa-upstream/src/authz.rs
Normal file
284
crates/helexa-upstream/src/authz.rs
Normal file
@@ -0,0 +1,284 @@
|
||||
//! `/authz/v1` — the machine surface cortex's `UpstreamEntitlementProvider`
|
||||
//! (#57) consumes. It mirrors the `cortex_core::entitlements::EntitlementProvider`
|
||||
//! trait 1:1 (resolve / reserve / settle / release / snapshot) over the B1
|
||||
//! ledger.
|
||||
//!
|
||||
//! Contract notes for the cortex client:
|
||||
//! - A **non-2xx** response means the authority could not give an
|
||||
//! authoritative answer (bad caller auth, malformed request, server
|
||||
//! error) → the client should **fail closed**.
|
||||
//! - `reserve` returns **200** whether granted or budget-refused: the body
|
||||
//! carries either `reservation_id` or a `rejected` discriminant. A budget
|
||||
//! refusal is an authoritative answer, not a transport failure.
|
||||
//! - Rejections that are genuinely auth failures use the #63 `OpenAiError`
|
||||
//! envelope so they can be surfaced verbatim.
|
||||
|
||||
use crate::crypto::sha256;
|
||||
use crate::error::envelope_response;
|
||||
use crate::ledger::{self, LedgerError};
|
||||
use crate::state::AppState;
|
||||
use axum::extract::{Request, State};
|
||||
use axum::http::{StatusCode, header};
|
||||
use axum::middleware::Next;
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::routing::post;
|
||||
use axum::{Json, Router};
|
||||
use cortex_core::error_envelope::OpenAiError;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use subtle::ConstantTimeEq;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// The operator a validated client bearer identifies (served-usage
|
||||
/// attribution, #58). Inserted into request extensions by [`client_auth`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct OperatorId(pub String);
|
||||
|
||||
/// Build the `/authz/v1` router with the client-auth layer applied.
|
||||
pub fn router(state: &AppState) -> Router<AppState> {
|
||||
Router::new()
|
||||
.route("/authz/v1/resolve", post(resolve))
|
||||
.route("/authz/v1/reserve", post(reserve))
|
||||
.route("/authz/v1/settle", post(settle))
|
||||
.route("/authz/v1/release", post(release))
|
||||
.route("/authz/v1/snapshot", post(snapshot))
|
||||
.layer(axum::middleware::from_fn_with_state(
|
||||
state.clone(),
|
||||
client_auth,
|
||||
))
|
||||
}
|
||||
|
||||
// ── client auth (shared bearer → operator_id) ───────────────────────
|
||||
|
||||
/// Validate the caller's `Authorization: Bearer` against the configured
|
||||
/// client tokens (constant-time) and stamp the `operator_id`. When no tokens
|
||||
/// are configured the surface is open (dev) and a synthetic operator is
|
||||
/// used.
|
||||
async fn client_auth(State(state): State<AppState>, mut req: Request, next: Next) -> Response {
|
||||
let tokens = &state.config.client_auth.tokens;
|
||||
if tokens.is_empty() {
|
||||
req.extensions_mut().insert(OperatorId("dev".into()));
|
||||
return next.run(req).await;
|
||||
}
|
||||
let presented = req
|
||||
.headers()
|
||||
.get(header::AUTHORIZATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.strip_prefix("Bearer "))
|
||||
.map(str::trim)
|
||||
.unwrap_or("");
|
||||
let matched = tokens
|
||||
.iter()
|
||||
.find(|t| t.token.as_bytes().ct_eq(presented.as_bytes()).into());
|
||||
match matched {
|
||||
Some(t) => {
|
||||
req.extensions_mut()
|
||||
.insert(OperatorId(t.operator_id.clone()));
|
||||
next.run(req).await
|
||||
}
|
||||
None => envelope_response(OpenAiError::invalid_api_key(
|
||||
"missing or invalid client credentials",
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
// ── DTOs ────────────────────────────────────────────────────────────
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ResolveReq {
|
||||
api_key: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct PrincipalDto {
|
||||
account_id: String,
|
||||
key_id: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct SnapshotDto {
|
||||
hard_cap: Option<i64>,
|
||||
spent: i64,
|
||||
reserved: i64,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct ResolveResp {
|
||||
principal: PrincipalDto,
|
||||
snapshot: SnapshotDto,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ReserveReq {
|
||||
account_id: String,
|
||||
key_id: String,
|
||||
max_tokens: i64,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Default)]
|
||||
struct ReserveResp {
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
reservation_id: Option<i64>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
rejected: Option<Rejection>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
enum Rejection {
|
||||
InsufficientQuota {
|
||||
requested: i64,
|
||||
available: i64,
|
||||
},
|
||||
// Part of the frozen wire contract so the cortex client (#57) can map it
|
||||
// without a later breaking change. Not yet constructed: the B1 ledger
|
||||
// implements Balance caps only; rolling-window key sub-caps (which yield
|
||||
// this) land in a follow-up.
|
||||
#[allow(dead_code)]
|
||||
RateLimited {
|
||||
requested: i64,
|
||||
available: i64,
|
||||
retry_after_secs: u64,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SettleReq {
|
||||
reservation_id: i64,
|
||||
actual_tokens: i64,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ReservationRef {
|
||||
reservation_id: i64,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SnapshotReq {
|
||||
account_id: String,
|
||||
key_id: String,
|
||||
}
|
||||
|
||||
// ── handlers ────────────────────────────────────────────────────────
|
||||
|
||||
/// `POST /authz/v1/resolve` — bearer key → principal + snapshot, or
|
||||
/// `401 invalid_api_key` (also for a deactivated account: no clue).
|
||||
async fn resolve(State(state): State<AppState>, Json(req): Json<ResolveReq>) -> Response {
|
||||
match ledger::resolve_key(&state.pool, &sha256(&req.api_key)).await {
|
||||
Ok(Some(p)) => Json(ResolveResp {
|
||||
principal: PrincipalDto {
|
||||
account_id: p.account_id.to_string(),
|
||||
key_id: p.key_id.to_string(),
|
||||
},
|
||||
snapshot: SnapshotDto {
|
||||
hard_cap: Some(p.hard_cap),
|
||||
spent: p.key_spent,
|
||||
reserved: p.key_reserved,
|
||||
},
|
||||
})
|
||||
.into_response(),
|
||||
Ok(None) => envelope_response(OpenAiError::invalid_api_key("invalid or unknown API key")),
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "resolve query failed");
|
||||
envelope_response(OpenAiError::service_unavailable("authority error", Some(5)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /authz/v1/reserve` — 200 with `reservation_id` (granted) or
|
||||
/// `rejected` (budget). Non-2xx only for bad input / server error.
|
||||
async fn reserve(State(state): State<AppState>, Json(req): Json<ReserveReq>) -> Response {
|
||||
let (Ok(account_id), Ok(key_id)) = (
|
||||
Uuid::parse_str(&req.account_id),
|
||||
Uuid::parse_str(&req.key_id),
|
||||
) else {
|
||||
return bad_request("account_id and key_id must be UUIDs");
|
||||
};
|
||||
match ledger::reserve(&state.pool, account_id, key_id, req.max_tokens).await {
|
||||
Ok(reservation_id) => Json(ReserveResp {
|
||||
reservation_id: Some(reservation_id),
|
||||
rejected: None,
|
||||
})
|
||||
.into_response(),
|
||||
Err(LedgerError::InsufficientQuota {
|
||||
requested,
|
||||
available,
|
||||
}) => Json(ReserveResp {
|
||||
reservation_id: None,
|
||||
rejected: Some(Rejection::InsufficientQuota {
|
||||
requested,
|
||||
available,
|
||||
}),
|
||||
})
|
||||
.into_response(),
|
||||
Err(LedgerError::AccountNotFound | LedgerError::KeyNotFound) => {
|
||||
// Resolve succeeded earlier; the principal vanished (archived /
|
||||
// deactivated). Treat as no budget — fail closed at the client.
|
||||
Json(ReserveResp {
|
||||
reservation_id: None,
|
||||
rejected: Some(Rejection::InsufficientQuota {
|
||||
requested: req.max_tokens,
|
||||
available: 0,
|
||||
}),
|
||||
})
|
||||
.into_response()
|
||||
}
|
||||
Err(LedgerError::Db(e)) => {
|
||||
tracing::error!(error = %e, "reserve failed");
|
||||
envelope_response(OpenAiError::service_unavailable("authority error", Some(5)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /authz/v1/settle` — idempotent; `204`.
|
||||
async fn settle(State(state): State<AppState>, Json(req): Json<SettleReq>) -> Response {
|
||||
match ledger::settle(&state.pool, req.reservation_id, req.actual_tokens).await {
|
||||
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "settle failed");
|
||||
envelope_response(OpenAiError::service_unavailable("authority error", Some(5)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /authz/v1/release` — idempotent; `204`.
|
||||
async fn release(State(state): State<AppState>, Json(req): Json<ReservationRef>) -> Response {
|
||||
match ledger::release(&state.pool, req.reservation_id).await {
|
||||
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "release failed");
|
||||
envelope_response(OpenAiError::service_unavailable("authority error", Some(5)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /authz/v1/snapshot` — `{hard_cap, spent, reserved}` or `404`.
|
||||
async fn snapshot(State(state): State<AppState>, Json(req): Json<SnapshotReq>) -> Response {
|
||||
let (Ok(account_id), Ok(key_id)) = (
|
||||
Uuid::parse_str(&req.account_id),
|
||||
Uuid::parse_str(&req.key_id),
|
||||
) else {
|
||||
return bad_request("account_id and key_id must be UUIDs");
|
||||
};
|
||||
match ledger::snapshot(&state.pool, account_id, key_id).await {
|
||||
Ok(Some((hard_cap, spent, reserved))) => Json(SnapshotDto {
|
||||
hard_cap: Some(hard_cap),
|
||||
spent,
|
||||
reserved,
|
||||
})
|
||||
.into_response(),
|
||||
Ok(None) => StatusCode::NOT_FOUND.into_response(),
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "snapshot failed");
|
||||
envelope_response(OpenAiError::service_unavailable("authority error", Some(5)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn bad_request(msg: &str) -> Response {
|
||||
envelope_response(OpenAiError::new(
|
||||
400,
|
||||
"invalid_request_error",
|
||||
"invalid_request",
|
||||
msg,
|
||||
))
|
||||
}
|
||||
185
crates/helexa-upstream/src/config.rs
Normal file
185
crates/helexa-upstream/src/config.rs
Normal file
@@ -0,0 +1,185 @@
|
||||
//! helexa-upstream configuration: loaded from `helexa-upstream.toml` with
|
||||
//! figment, `UPSTREAM_`-prefixed env overrides (mirrors the cortex/router
|
||||
//! convention, e.g. `UPSTREAM_SERVER__LISTEN`, `UPSTREAM_DB__URL`).
|
||||
|
||||
use figment::{
|
||||
Figment,
|
||||
providers::{Env, Format, Toml},
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::path::Path;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct UpstreamConfig {
|
||||
#[serde(default)]
|
||||
pub server: ServerSettings,
|
||||
pub db: DbSettings,
|
||||
#[serde(default)]
|
||||
pub grant: GrantSettings,
|
||||
#[serde(default)]
|
||||
pub abuse: AbuseSettings,
|
||||
#[serde(default)]
|
||||
pub client_auth: ClientAuthSettings,
|
||||
#[serde(default)]
|
||||
pub authz: AuthzSettings,
|
||||
}
|
||||
|
||||
/// `[client_auth]` — credentials operators' cortexes present to `/authz/v1`.
|
||||
/// Each token maps to an `operator_id` (served-usage attribution, #58). This
|
||||
/// transport credential is distinct from end-user API keys (which ride in
|
||||
/// the `resolve` body). v2 adds mTLS.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct ClientAuthSettings {
|
||||
/// When empty the authz surface is **open** (dev only; logged at warn).
|
||||
#[serde(default)]
|
||||
pub tokens: Vec<ClientToken>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ClientToken {
|
||||
/// Shared bearer a cortex presents.
|
||||
pub token: String,
|
||||
/// Operator this token identifies.
|
||||
pub operator_id: String,
|
||||
}
|
||||
|
||||
/// `[authz]` — reservation lifecycle knobs.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct AuthzSettings {
|
||||
/// Open reservations older than this are swept (released), self-healing
|
||||
/// a reservation whose settle/release from cortex was lost.
|
||||
#[serde(default = "default_reservation_ttl")]
|
||||
pub reservation_ttl_secs: u64,
|
||||
/// How often the sweeper runs.
|
||||
#[serde(default = "default_sweep_interval")]
|
||||
pub sweep_interval_secs: u64,
|
||||
}
|
||||
|
||||
impl Default for AuthzSettings {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
reservation_ttl_secs: default_reservation_ttl(),
|
||||
sweep_interval_secs: default_sweep_interval(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ServerSettings {
|
||||
/// Address to listen on (e.g. "0.0.0.0:8090"). Plaintext — edge nginx
|
||||
/// terminates TLS, consistent with the rest of the stack.
|
||||
#[serde(default = "default_listen")]
|
||||
pub listen: String,
|
||||
}
|
||||
|
||||
impl Default for ServerSettings {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
listen: default_listen(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DbSettings {
|
||||
/// PostgreSQL connection URL (e.g. "postgres://user:pass@host/helexa").
|
||||
pub url: String,
|
||||
/// Max pool connections.
|
||||
#[serde(default = "default_max_connections")]
|
||||
pub max_connections: u32,
|
||||
}
|
||||
|
||||
/// `[grant]` — the flat free token grant every email-verified account
|
||||
/// receives (the floor of the hybrid allocation model; top-up codes extend
|
||||
/// it).
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct GrantSettings {
|
||||
#[serde(default = "default_free_grant")]
|
||||
pub free_token_grant: i64,
|
||||
}
|
||||
|
||||
impl Default for GrantSettings {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
free_token_grant: default_free_grant(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `[abuse]` — silent multi-account abuse detection. When at least
|
||||
/// `fingerprint_account_threshold` accounts share one registration
|
||||
/// fingerprint, all of them are silently deactivated (no notice to the
|
||||
/// user; deactivation only surfaces as ordinary inference rejections).
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct AbuseSettings {
|
||||
#[serde(default = "default_fingerprint_threshold")]
|
||||
pub fingerprint_account_threshold: i64,
|
||||
}
|
||||
|
||||
impl Default for AbuseSettings {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
fingerprint_account_threshold: default_fingerprint_threshold(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn default_listen() -> String {
|
||||
"0.0.0.0:8090".into()
|
||||
}
|
||||
fn default_max_connections() -> u32 {
|
||||
16
|
||||
}
|
||||
fn default_free_grant() -> i64 {
|
||||
1_000_000
|
||||
}
|
||||
fn default_fingerprint_threshold() -> i64 {
|
||||
5
|
||||
}
|
||||
fn default_reservation_ttl() -> u64 {
|
||||
120
|
||||
}
|
||||
fn default_sweep_interval() -> u64 {
|
||||
60
|
||||
}
|
||||
|
||||
impl UpstreamConfig {
|
||||
/// Load from a TOML file with `UPSTREAM_`-prefixed env overrides
|
||||
/// (`__` nesting separator).
|
||||
pub fn load(path: impl AsRef<Path>) -> Result<Self, Box<figment::Error>> {
|
||||
Figment::new()
|
||||
.merge(Toml::file(path))
|
||||
.merge(Env::prefixed("UPSTREAM_").split("__"))
|
||||
.extract()
|
||||
.map_err(Box::new)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
#[allow(clippy::result_large_err)]
|
||||
fn loads_toml_with_env_override_and_defaults() {
|
||||
figment::Jail::expect_with(|jail| {
|
||||
jail.create_file(
|
||||
"helexa-upstream.toml",
|
||||
r#"
|
||||
[db]
|
||||
url = "postgres://localhost/helexa"
|
||||
"#,
|
||||
)?;
|
||||
jail.set_env("UPSTREAM_SERVER__LISTEN", "127.0.0.1:9099");
|
||||
|
||||
let cfg = UpstreamConfig::load("helexa-upstream.toml").expect("load");
|
||||
assert_eq!(cfg.server.listen, "127.0.0.1:9099");
|
||||
assert_eq!(cfg.db.url, "postgres://localhost/helexa");
|
||||
// Defaults applied when sections omitted.
|
||||
assert_eq!(cfg.grant.free_token_grant, 1_000_000);
|
||||
assert_eq!(cfg.abuse.fingerprint_account_threshold, 5);
|
||||
assert_eq!(cfg.db.max_connections, 16);
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
}
|
||||
12
crates/helexa-upstream/src/crypto.rs
Normal file
12
crates/helexa-upstream/src/crypto.rs
Normal file
@@ -0,0 +1,12 @@
|
||||
//! Hashing helpers. API keys and top-up codes are stored only as their
|
||||
//! sha256 (they are high-entropy secrets; sha256 is the fast, sufficient
|
||||
//! choice — argon2 is reserved for low-entropy passwords).
|
||||
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
/// sha256 of `input`, as raw bytes (matches the `BYTEA` columns).
|
||||
pub fn sha256(input: &str) -> Vec<u8> {
|
||||
let mut h = Sha256::new();
|
||||
h.update(input.as_bytes());
|
||||
h.finalize().to_vec()
|
||||
}
|
||||
20
crates/helexa-upstream/src/db.rs
Normal file
20
crates/helexa-upstream/src/db.rs
Normal file
@@ -0,0 +1,20 @@
|
||||
//! PostgreSQL pool + embedded migrations.
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use sqlx::postgres::{PgPool, PgPoolOptions};
|
||||
|
||||
/// Connect to Postgres and run embedded migrations (`./migrations`).
|
||||
pub async fn connect_and_migrate(url: &str, max_connections: u32) -> Result<PgPool> {
|
||||
let pool = PgPoolOptions::new()
|
||||
.max_connections(max_connections)
|
||||
.connect(url)
|
||||
.await
|
||||
.with_context(|| "connecting to PostgreSQL")?;
|
||||
|
||||
sqlx::migrate!("./migrations")
|
||||
.run(&pool)
|
||||
.await
|
||||
.with_context(|| "running migrations")?;
|
||||
|
||||
Ok(pool)
|
||||
}
|
||||
21
crates/helexa-upstream/src/error.rs
Normal file
21
crates/helexa-upstream/src/error.rs
Normal file
@@ -0,0 +1,21 @@
|
||||
//! Adapter from the shared, axum-agnostic
|
||||
//! [`cortex_core::error_envelope::OpenAiError`] (#60/#63) to an axum
|
||||
//! response, with `Retry-After`. The `/authz/v1` surface speaks the #63
|
||||
//! envelope so cortex (an OpenAI-compatible proxy) can forward rejections
|
||||
//! verbatim. (The future `/web/v1` surface uses a plain JSON error shape.)
|
||||
|
||||
use axum::http::{HeaderValue, StatusCode, header};
|
||||
use axum::response::{IntoResponse, Json, Response};
|
||||
use cortex_core::error_envelope::OpenAiError;
|
||||
|
||||
pub fn envelope_response(err: OpenAiError) -> Response {
|
||||
let status = StatusCode::from_u16(err.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
|
||||
let retry_after = err.retry_after_secs;
|
||||
let mut response = (status, Json(err.body())).into_response();
|
||||
if let Some(secs) = retry_after
|
||||
&& let Ok(value) = HeaderValue::from_str(&secs.to_string())
|
||||
{
|
||||
response.headers_mut().insert(header::RETRY_AFTER, value);
|
||||
}
|
||||
response
|
||||
}
|
||||
21
crates/helexa-upstream/src/handlers.rs
Normal file
21
crates/helexa-upstream/src/handlers.rs
Normal file
@@ -0,0 +1,21 @@
|
||||
//! HTTP handlers. B1 ships `/health`; the authz (`/authz/v1`) and web
|
||||
//! (`/web/v1`) surfaces land in later phases.
|
||||
|
||||
use crate::state::AppState;
|
||||
use axum::{Json, Router, extract::State, routing::get};
|
||||
use serde_json::{Value, json};
|
||||
|
||||
pub fn routes() -> Router<AppState> {
|
||||
Router::new()
|
||||
.route("/health", get(health))
|
||||
.route("/", get(health))
|
||||
}
|
||||
|
||||
/// `GET /health` — liveness + a database round-trip (`SELECT 1`).
|
||||
async fn health(State(state): State<AppState>) -> Json<Value> {
|
||||
let db_ok = sqlx::query("SELECT 1").execute(&state.pool).await.is_ok();
|
||||
Json(json!({
|
||||
"status": if db_ok { "ok" } else { "degraded" },
|
||||
"db": if db_ok { "ok" } else { "unreachable" },
|
||||
}))
|
||||
}
|
||||
338
crates/helexa-upstream/src/ledger.rs
Normal file
338
crates/helexa-upstream/src/ledger.rs
Normal file
@@ -0,0 +1,338 @@
|
||||
//! The allocation ledger: reserve → settle/release with the no-overshoot
|
||||
//! guarantee enforced by a row-locked transaction.
|
||||
//!
|
||||
//! Each reserve takes `SELECT … FOR UPDATE` on the account (and key) row, so
|
||||
//! concurrent reserves from many cortexes serialize and `spent + reserved`
|
||||
//! can never exceed the effective cap. The `accounts_no_overshoot` CHECK is
|
||||
//! the DB-level backstop. Settle/release are idempotent (they only act on a
|
||||
//! reservation still in `open`).
|
||||
//!
|
||||
//! Per-key effective cap = `min(resolved key cap, remaining account
|
||||
//! allocation)`. The key cap is resolved from its `limit_kind`:
|
||||
//! `hardcap` → the value verbatim; `percent` → that % of the account's
|
||||
//! `allocation_total`.
|
||||
//!
|
||||
//! Cap-window semantics: this module implements **Balance** (non-resetting)
|
||||
//! caps. Rolling-window key sub-caps (and the `RateLimited` rejection that
|
||||
//! rides them) land with the authz API (B2); today an over-cap is always
|
||||
//! `InsufficientQuota`.
|
||||
|
||||
use sqlx::postgres::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// A bearer key resolved to its principal + a budget snapshot.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ResolvedPrincipal {
|
||||
pub account_id: Uuid,
|
||||
pub key_id: Uuid,
|
||||
/// Effective per-key absolute cap (the key sub-cap; the account cap
|
||||
/// still binds at reserve time).
|
||||
pub hard_cap: i64,
|
||||
pub key_spent: i64,
|
||||
pub key_reserved: i64,
|
||||
}
|
||||
|
||||
/// Resolve a key by its `sha256` hash to its principal, or `None` when the
|
||||
/// key is unknown/archived **or its account is deactivated** (the silent
|
||||
/// abuse flag — indistinguishable from an unknown key, by design: no clue).
|
||||
pub async fn resolve_key(
|
||||
pool: &PgPool,
|
||||
key_hash: &[u8],
|
||||
) -> Result<Option<ResolvedPrincipal>, sqlx::Error> {
|
||||
let row = sqlx::query(
|
||||
"SELECT k.id AS key_id, k.account_id, k.limit_kind, k.limit_value, \
|
||||
k.key_spent, k.key_reserved, a.allocation_total \
|
||||
FROM api_keys k JOIN accounts a ON a.id = k.account_id \
|
||||
WHERE k.key_hash = $1 AND k.status = 'active' AND a.status = 'active'",
|
||||
)
|
||||
.bind(key_hash)
|
||||
.fetch_optional(pool)
|
||||
.await?;
|
||||
Ok(row.map(|r| {
|
||||
let total: i64 = sqlx::Row::get(&r, "allocation_total");
|
||||
let limit_kind: String = sqlx::Row::get(&r, "limit_kind");
|
||||
let limit_value: i64 = sqlx::Row::get(&r, "limit_value");
|
||||
ResolvedPrincipal {
|
||||
account_id: sqlx::Row::get(&r, "account_id"),
|
||||
key_id: sqlx::Row::get(&r, "key_id"),
|
||||
hard_cap: resolve_abs_cap(&limit_kind, limit_value, total),
|
||||
key_spent: sqlx::Row::get(&r, "key_spent"),
|
||||
key_reserved: sqlx::Row::get(&r, "key_reserved"),
|
||||
}
|
||||
}))
|
||||
}
|
||||
|
||||
/// Per-key budget snapshot `(hard_cap, spent, reserved)`, or `None` if the
|
||||
/// key/account isn't an active pair.
|
||||
pub async fn snapshot(
|
||||
pool: &PgPool,
|
||||
account_id: Uuid,
|
||||
key_id: Uuid,
|
||||
) -> Result<Option<(i64, i64, i64)>, sqlx::Error> {
|
||||
let row = sqlx::query(
|
||||
"SELECT k.limit_kind, k.limit_value, k.key_spent, k.key_reserved, a.allocation_total \
|
||||
FROM api_keys k JOIN accounts a ON a.id = k.account_id \
|
||||
WHERE k.id = $1 AND k.account_id = $2 AND k.status = 'active' AND a.status = 'active'",
|
||||
)
|
||||
.bind(key_id)
|
||||
.bind(account_id)
|
||||
.fetch_optional(pool)
|
||||
.await?;
|
||||
Ok(row.map(|r| {
|
||||
let total: i64 = sqlx::Row::get(&r, "allocation_total");
|
||||
let limit_kind: String = sqlx::Row::get(&r, "limit_kind");
|
||||
let limit_value: i64 = sqlx::Row::get(&r, "limit_value");
|
||||
let cap = resolve_abs_cap(&limit_kind, limit_value, total);
|
||||
(
|
||||
cap,
|
||||
sqlx::Row::get::<i64, _>(&r, "key_spent"),
|
||||
sqlx::Row::get::<i64, _>(&r, "key_reserved"),
|
||||
)
|
||||
}))
|
||||
}
|
||||
|
||||
/// Release every `open` reservation older than `max_age_secs`, returning
|
||||
/// each one's reserved tokens to its account and key in a single statement.
|
||||
/// The lost-settle self-heal. Returns the number swept.
|
||||
pub async fn sweep_stale(pool: &PgPool, max_age_secs: i64) -> Result<u64, sqlx::Error> {
|
||||
// Data-modifying CTEs: release stale rows, then fold their reserved sums
|
||||
// back into accounts and api_keys. All in one atomic statement.
|
||||
let result = sqlx::query(
|
||||
"WITH stale AS ( \
|
||||
UPDATE reservations SET state = 'released', settled_at = now() \
|
||||
WHERE state = 'open' AND created_at < now() - make_interval(secs => $1) \
|
||||
RETURNING account_id, key_id, reserved \
|
||||
), acct AS ( \
|
||||
UPDATE accounts a SET allocation_reserved = allocation_reserved - s.total \
|
||||
FROM (SELECT account_id, SUM(reserved) AS total FROM stale GROUP BY account_id) s \
|
||||
WHERE a.id = s.account_id \
|
||||
) \
|
||||
UPDATE api_keys k SET key_reserved = key_reserved - s.total \
|
||||
FROM (SELECT key_id, SUM(reserved) AS total FROM stale GROUP BY key_id) s \
|
||||
WHERE k.id = s.key_id",
|
||||
)
|
||||
.bind(max_age_secs as f64)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(result.rows_affected())
|
||||
}
|
||||
|
||||
/// Resolve a key's per-key cap to an absolute token count.
|
||||
///
|
||||
/// `percent` is `floor(allocation_total * limit_value / 100)`; `hardcap` is
|
||||
/// `limit_value` verbatim. Computed in i128 to avoid overflow, floored at 0.
|
||||
pub fn resolve_abs_cap(limit_kind: &str, limit_value: i64, allocation_total: i64) -> i64 {
|
||||
let cap = match limit_kind {
|
||||
"percent" => (allocation_total as i128 * limit_value as i128) / 100,
|
||||
_ => limit_value as i128, // "hardcap" (and any unknown → treat as absolute)
|
||||
};
|
||||
cap.clamp(0, i64::MAX as i128) as i64
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum LedgerError {
|
||||
#[error("account not found")]
|
||||
AccountNotFound,
|
||||
#[error("api key not found or not active")]
|
||||
KeyNotFound,
|
||||
/// Account balance or a Balance-window key sub-cap is exhausted.
|
||||
#[error("insufficient quota: requested {requested}, available {available}")]
|
||||
InsufficientQuota { requested: i64, available: i64 },
|
||||
#[error(transparent)]
|
||||
Db(#[from] sqlx::Error),
|
||||
}
|
||||
|
||||
/// Reserve `max_tokens` against `account_id`/`key_id`. Returns the
|
||||
/// reservation id (the `BIGSERIAL`, mapped to the cortex `Reservation.id`).
|
||||
pub async fn reserve(
|
||||
pool: &PgPool,
|
||||
account_id: Uuid,
|
||||
key_id: Uuid,
|
||||
max_tokens: i64,
|
||||
) -> Result<i64, LedgerError> {
|
||||
let mut tx = pool.begin().await?;
|
||||
|
||||
// Lock the account row — serializes concurrent reserves on this account.
|
||||
let acct = sqlx::query(
|
||||
"SELECT allocation_total, allocation_spent, allocation_reserved \
|
||||
FROM accounts WHERE id = $1 AND status = 'active' FOR UPDATE",
|
||||
)
|
||||
.bind(account_id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some(acct) = acct else {
|
||||
return Err(LedgerError::AccountNotFound);
|
||||
};
|
||||
let total: i64 = sqlx::Row::get(&acct, "allocation_total");
|
||||
let spent: i64 = sqlx::Row::get(&acct, "allocation_spent");
|
||||
let reserved: i64 = sqlx::Row::get(&acct, "allocation_reserved");
|
||||
let account_avail = total - spent - reserved;
|
||||
|
||||
// Lock the key row and resolve its absolute sub-cap.
|
||||
let key = sqlx::query(
|
||||
"SELECT limit_kind, limit_value, key_spent, key_reserved \
|
||||
FROM api_keys WHERE id = $1 AND account_id = $2 AND status = 'active' FOR UPDATE",
|
||||
)
|
||||
.bind(key_id)
|
||||
.bind(account_id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some(key) = key else {
|
||||
return Err(LedgerError::KeyNotFound);
|
||||
};
|
||||
let limit_kind: String = sqlx::Row::get(&key, "limit_kind");
|
||||
let limit_value: i64 = sqlx::Row::get(&key, "limit_value");
|
||||
let key_spent: i64 = sqlx::Row::get(&key, "key_spent");
|
||||
let key_reserved: i64 = sqlx::Row::get(&key, "key_reserved");
|
||||
let key_cap = resolve_abs_cap(&limit_kind, limit_value, total);
|
||||
let key_avail = key_cap - key_spent - key_reserved;
|
||||
|
||||
let available = account_avail.min(key_avail).max(0);
|
||||
if max_tokens > available {
|
||||
// tx rolls back on drop
|
||||
return Err(LedgerError::InsufficientQuota {
|
||||
requested: max_tokens,
|
||||
available,
|
||||
});
|
||||
}
|
||||
|
||||
let id: i64 = sqlx::Row::get(
|
||||
&sqlx::query(
|
||||
"INSERT INTO reservations (account_id, key_id, reserved, state) \
|
||||
VALUES ($1, $2, $3, 'open') RETURNING id",
|
||||
)
|
||||
.bind(account_id)
|
||||
.bind(key_id)
|
||||
.bind(max_tokens)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?,
|
||||
"id",
|
||||
);
|
||||
sqlx::query("UPDATE accounts SET allocation_reserved = allocation_reserved + $1 WHERE id = $2")
|
||||
.bind(max_tokens)
|
||||
.bind(account_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
sqlx::query("UPDATE api_keys SET key_reserved = key_reserved + $1 WHERE id = $2")
|
||||
.bind(max_tokens)
|
||||
.bind(key_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
/// Settle a reservation with the actual tokens used (clamped to
|
||||
/// `[0, reserved]`). Idempotent: a second settle (or settle after release)
|
||||
/// is a no-op.
|
||||
pub async fn settle(
|
||||
pool: &PgPool,
|
||||
reservation_id: i64,
|
||||
actual_tokens: i64,
|
||||
) -> Result<(), LedgerError> {
|
||||
let mut tx = pool.begin().await?;
|
||||
let row = sqlx::query(
|
||||
"UPDATE reservations SET state = 'settled', settled_at = now(), \
|
||||
actual = LEAST(GREATEST($2, 0), reserved) \
|
||||
WHERE id = $1 AND state = 'open' \
|
||||
RETURNING reserved, account_id, key_id, actual",
|
||||
)
|
||||
.bind(reservation_id)
|
||||
.bind(actual_tokens)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some(row) = row else {
|
||||
return Ok(()); // already settled/released, or unknown → idempotent no-op
|
||||
};
|
||||
let reserved: i64 = sqlx::Row::get(&row, "reserved");
|
||||
let actual: i64 = sqlx::Row::get(&row, "actual");
|
||||
let account_id: Uuid = sqlx::Row::get(&row, "account_id");
|
||||
let key_id: Uuid = sqlx::Row::get(&row, "key_id");
|
||||
|
||||
sqlx::query(
|
||||
"UPDATE accounts SET allocation_reserved = allocation_reserved - $1, \
|
||||
allocation_spent = allocation_spent + $2 WHERE id = $3",
|
||||
)
|
||||
.bind(reserved)
|
||||
.bind(actual)
|
||||
.bind(account_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
sqlx::query(
|
||||
"UPDATE api_keys SET key_reserved = key_reserved - $1, key_spent = key_spent + $2 WHERE id = $3",
|
||||
)
|
||||
.bind(reserved)
|
||||
.bind(actual)
|
||||
.bind(key_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Release a reservation, returning its full reserved amount to the
|
||||
/// allocation. Idempotent.
|
||||
pub async fn release(pool: &PgPool, reservation_id: i64) -> Result<(), LedgerError> {
|
||||
let mut tx = pool.begin().await?;
|
||||
let row = sqlx::query(
|
||||
"UPDATE reservations SET state = 'released', settled_at = now() \
|
||||
WHERE id = $1 AND state = 'open' \
|
||||
RETURNING reserved, account_id, key_id",
|
||||
)
|
||||
.bind(reservation_id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some(row) = row else {
|
||||
return Ok(());
|
||||
};
|
||||
let reserved: i64 = sqlx::Row::get(&row, "reserved");
|
||||
let account_id: Uuid = sqlx::Row::get(&row, "account_id");
|
||||
let key_id: Uuid = sqlx::Row::get(&row, "key_id");
|
||||
|
||||
sqlx::query("UPDATE accounts SET allocation_reserved = allocation_reserved - $1 WHERE id = $2")
|
||||
.bind(reserved)
|
||||
.bind(account_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
sqlx::query("UPDATE api_keys SET key_reserved = key_reserved - $1 WHERE id = $2")
|
||||
.bind(reserved)
|
||||
.bind(key_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::resolve_abs_cap;
|
||||
|
||||
#[test]
|
||||
fn hardcap_is_verbatim() {
|
||||
assert_eq!(resolve_abs_cap("hardcap", 50_000, 1_000_000), 50_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn percent_is_fraction_of_allocation() {
|
||||
assert_eq!(resolve_abs_cap("percent", 25, 1_000_000), 250_000);
|
||||
assert_eq!(resolve_abs_cap("percent", 100, 1_000_000), 1_000_000);
|
||||
// floor
|
||||
assert_eq!(resolve_abs_cap("percent", 33, 10), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn percent_does_not_overflow_on_large_allocation() {
|
||||
// total * value would overflow i64 if not widened to i128.
|
||||
let cap = resolve_abs_cap("percent", 100, i64::MAX);
|
||||
assert_eq!(cap, i64::MAX);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn negative_or_zero_clamps_to_zero() {
|
||||
assert_eq!(resolve_abs_cap("hardcap", -5, 100), 0);
|
||||
assert_eq!(resolve_abs_cap("percent", 0, 1_000_000), 0);
|
||||
}
|
||||
}
|
||||
78
crates/helexa-upstream/src/lib.rs
Normal file
78
crates/helexa-upstream/src/lib.rs
Normal file
@@ -0,0 +1,78 @@
|
||||
//! helexa-upstream — the mesh-level account/authorization authority (#59).
|
||||
//!
|
||||
//! The clearing house above cortex: it issues accounts and API keys, holds
|
||||
//! the real token-allocation ledger, authorizes inference in real time
|
||||
//! (reserve → settle, fail-closed), and tracks served usage for operator
|
||||
//! reconciliation. cortex's `UpstreamEntitlementProvider` (#57) is a client
|
||||
//! of the `/authz/v1` surface; the helexa.ai frontend is a client of the
|
||||
//! `/web/v1` surface.
|
||||
//!
|
||||
//! Landed so far: B1 — schema + reserve→settle [`ledger`] (no-overshoot) +
|
||||
//! `/health`. B2 — the `/authz/v1` [`authz`] surface (resolve/reserve/
|
||||
//! settle/release/snapshot) with shared-bearer client auth and a
|
||||
//! stale-reservation sweeper.
|
||||
|
||||
pub mod authz;
|
||||
pub mod config;
|
||||
pub mod crypto;
|
||||
pub mod db;
|
||||
pub mod error;
|
||||
pub mod handlers;
|
||||
pub mod ledger;
|
||||
pub mod state;
|
||||
|
||||
use anyhow::Result;
|
||||
use config::UpstreamConfig;
|
||||
use state::AppState;
|
||||
use std::time::Duration;
|
||||
use tower_http::trace::TraceLayer;
|
||||
|
||||
/// Build the axum application.
|
||||
pub fn build_app(state: AppState) -> axum::Router {
|
||||
axum::Router::new()
|
||||
.merge(handlers::routes())
|
||||
.merge(authz::router(&state))
|
||||
.layer(TraceLayer::new_for_http())
|
||||
.with_state(state)
|
||||
}
|
||||
|
||||
/// Start the service: connect Postgres, run migrations, spawn the
|
||||
/// reservation sweeper, bind the listener.
|
||||
pub async fn run(config: UpstreamConfig) -> Result<()> {
|
||||
let pool = db::connect_and_migrate(&config.db.url, config.db.max_connections).await?;
|
||||
let listen = config.server.listen.clone();
|
||||
let state = AppState::new(pool, config);
|
||||
|
||||
if state.config.client_auth.tokens.is_empty() {
|
||||
tracing::warn!(
|
||||
"no [client_auth] tokens configured — the /authz/v1 surface is OPEN (dev only)"
|
||||
);
|
||||
}
|
||||
|
||||
// Stale-reservation sweeper: releases open reservations whose
|
||||
// settle/release from cortex was lost, self-healing allocation_reserved.
|
||||
spawn_sweeper(&state);
|
||||
|
||||
let addr = listen.parse::<std::net::SocketAddr>()?;
|
||||
tracing::info!("helexa-upstream listening on {addr}");
|
||||
|
||||
let listener = tokio::net::TcpListener::bind(addr).await?;
|
||||
axum::serve(listener, build_app(state)).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn spawn_sweeper(state: &AppState) {
|
||||
let pool = state.pool.clone();
|
||||
let ttl = state.config.authz.reservation_ttl_secs as i64;
|
||||
let interval = Duration::from_secs(state.config.authz.sweep_interval_secs);
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::time::sleep(interval).await;
|
||||
match ledger::sweep_stale(&pool, ttl).await {
|
||||
Ok(n) if n > 0 => tracing::info!(swept = n, "released stale reservations"),
|
||||
Ok(_) => {}
|
||||
Err(e) => tracing::warn!(error = %e, "reservation sweep failed"),
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
46
crates/helexa-upstream/src/main.rs
Normal file
46
crates/helexa-upstream/src/main.rs
Normal file
@@ -0,0 +1,46 @@
|
||||
use anyhow::Result;
|
||||
use clap::{Parser, Subcommand};
|
||||
use helexa_upstream::config::UpstreamConfig;
|
||||
use tracing_subscriber::EnvFilter;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(name = "helexa-upstream")]
|
||||
#[command(about = "Mesh-level account & authorization authority for helexa")]
|
||||
#[command(version)]
|
||||
struct Cli {
|
||||
#[command(subcommand)]
|
||||
command: Commands,
|
||||
}
|
||||
|
||||
#[derive(Subcommand)]
|
||||
enum Commands {
|
||||
/// Start the upstream server.
|
||||
Serve {
|
||||
/// Path to the config file.
|
||||
#[arg(short, long, default_value = "helexa-upstream.toml")]
|
||||
config: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<()> {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| EnvFilter::new("info,helexa_upstream=debug")),
|
||||
)
|
||||
.init();
|
||||
|
||||
let cli = Cli::parse();
|
||||
|
||||
match cli.command {
|
||||
Commands::Serve { config } => {
|
||||
let cfg = UpstreamConfig::load(&config)
|
||||
.map_err(|e| anyhow::anyhow!("failed to load config from '{config}': {e}"))?;
|
||||
tracing::info!(listen = %cfg.server.listen, "starting helexa-upstream");
|
||||
helexa_upstream::run(cfg).await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
20
crates/helexa-upstream/src/state.rs
Normal file
20
crates/helexa-upstream/src/state.rs
Normal file
@@ -0,0 +1,20 @@
|
||||
//! Shared application state.
|
||||
|
||||
use crate::config::UpstreamConfig;
|
||||
use sqlx::postgres::PgPool;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct AppState {
|
||||
pub pool: PgPool,
|
||||
pub config: Arc<UpstreamConfig>,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
pub fn new(pool: PgPool, config: UpstreamConfig) -> Self {
|
||||
Self {
|
||||
pool,
|
||||
config: Arc::new(config),
|
||||
}
|
||||
}
|
||||
}
|
||||
240
crates/helexa-upstream/tests/authz_pg.rs
Normal file
240
crates/helexa-upstream/tests/authz_pg.rs
Normal file
@@ -0,0 +1,240 @@
|
||||
//! Integration tests for the `/authz/v1` surface against a real Postgres,
|
||||
//! driving the built axum app over HTTP. Gated on `UPSTREAM_TEST_DATABASE_URL`
|
||||
//! (skips cleanly when unset, so CI stays green without a DB):
|
||||
//!
|
||||
//! UPSTREAM_TEST_DATABASE_URL=postgres://helexa:helexa@localhost/helexa_test \
|
||||
//! cargo test -p helexa-upstream --test authz_pg
|
||||
|
||||
use helexa_upstream::config::{ClientToken, UpstreamConfig};
|
||||
use helexa_upstream::crypto::sha256;
|
||||
use helexa_upstream::db::connect_and_migrate;
|
||||
use helexa_upstream::state::AppState;
|
||||
use serde_json::{Value, json};
|
||||
use sqlx::Executor;
|
||||
use sqlx::Row;
|
||||
use sqlx::postgres::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
const CLIENT_TOKEN: &str = "test-operator-token";
|
||||
|
||||
async fn spawn_or_skip(test: &str) -> Option<(String, PgPool)> {
|
||||
let Ok(url) = std::env::var("UPSTREAM_TEST_DATABASE_URL") else {
|
||||
eprintln!("skipping {test}: UPSTREAM_TEST_DATABASE_URL not set");
|
||||
return None;
|
||||
};
|
||||
let pool = connect_and_migrate(&url, 16).await.expect("migrate");
|
||||
|
||||
let mut config = UpstreamConfig {
|
||||
server: Default::default(),
|
||||
db: helexa_upstream::config::DbSettings {
|
||||
url,
|
||||
max_connections: 16,
|
||||
},
|
||||
grant: Default::default(),
|
||||
abuse: Default::default(),
|
||||
client_auth: Default::default(),
|
||||
authz: Default::default(),
|
||||
};
|
||||
config.client_auth.tokens.push(ClientToken {
|
||||
token: CLIENT_TOKEN.into(),
|
||||
operator_id: "op-test".into(),
|
||||
});
|
||||
|
||||
let state = AppState::new(pool.clone(), config);
|
||||
let app = helexa_upstream::build_app(state);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
Some((format!("http://{addr}"), pool))
|
||||
}
|
||||
|
||||
/// Seed an account with `total` allocation and an active key with raw value
|
||||
/// `raw` (percent=100). Optionally deactivate the account. Returns
|
||||
/// (account_id, key_id).
|
||||
async fn seed_key(pool: &PgPool, total: i64, raw: &str, deactivated: bool) -> (Uuid, Uuid) {
|
||||
let user_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO users (email, password_hash, email_verified) VALUES ($1,'x',true) RETURNING id",
|
||||
)
|
||||
.bind(format!("u-{}@t.local", Uuid::new_v4())),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let status = if deactivated { "deactivated" } else { "active" };
|
||||
let account_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO accounts (owner_user_id, allocation_total, status) VALUES ($1,$2,$3) RETURNING id",
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(total)
|
||||
.bind(status),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let key_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO api_keys (account_id, key_hash, key_prefix, limit_kind, limit_value) \
|
||||
VALUES ($1,$2,'sk-test','percent',100) RETURNING id",
|
||||
)
|
||||
.bind(account_id)
|
||||
.bind(sha256(raw)),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
(account_id, key_id)
|
||||
}
|
||||
|
||||
fn client() -> reqwest::Client {
|
||||
reqwest::Client::new()
|
||||
}
|
||||
|
||||
async fn post(
|
||||
c: &reqwest::Client,
|
||||
url: String,
|
||||
body: Value,
|
||||
bearer: Option<&str>,
|
||||
) -> reqwest::Response {
|
||||
let mut req = c.post(url).json(&body);
|
||||
if let Some(b) = bearer {
|
||||
req = req.bearer_auth(b);
|
||||
}
|
||||
req.send().await.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolve_reserve_settle_round_trip() {
|
||||
let Some((base, pool)) = spawn_or_skip("resolve_reserve_settle_round_trip").await else {
|
||||
return;
|
||||
};
|
||||
let raw = format!("sk-{}", Uuid::new_v4());
|
||||
let (account_id, key_id) = seed_key(&pool, 1000, &raw, false).await;
|
||||
let c = client();
|
||||
|
||||
// resolve
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/resolve"),
|
||||
json!({"api_key": raw}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 200);
|
||||
let body: Value = r.json().await.unwrap();
|
||||
assert_eq!(body["principal"]["account_id"], account_id.to_string());
|
||||
assert_eq!(body["principal"]["key_id"], key_id.to_string());
|
||||
assert_eq!(body["snapshot"]["hard_cap"], 1000);
|
||||
|
||||
// reserve 400
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/reserve"),
|
||||
json!({"account_id": account_id, "key_id": key_id, "max_tokens": 400}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 200);
|
||||
let body: Value = r.json().await.unwrap();
|
||||
let rid = body["reservation_id"].as_i64().expect("granted");
|
||||
|
||||
// settle 150
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/settle"),
|
||||
json!({"reservation_id": rid, "actual_tokens": 150}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 204);
|
||||
|
||||
// snapshot reflects spend
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/snapshot"),
|
||||
json!({"account_id": account_id, "key_id": key_id}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
let body: Value = r.json().await.unwrap();
|
||||
assert_eq!(body["spent"], 150);
|
||||
assert_eq!(body["reserved"], 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn over_cap_reserve_is_rejected_not_errored() {
|
||||
let Some((base, pool)) = spawn_or_skip("over_cap_reserve_is_rejected_not_errored").await else {
|
||||
return;
|
||||
};
|
||||
let raw = format!("sk-{}", Uuid::new_v4());
|
||||
let (account_id, key_id) = seed_key(&pool, 100, &raw, false).await;
|
||||
let c = client();
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/reserve"),
|
||||
json!({"account_id": account_id, "key_id": key_id, "max_tokens": 999}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 200, "budget refusal is an authoritative 200");
|
||||
let body: Value = r.json().await.unwrap();
|
||||
assert!(body["reservation_id"].is_null());
|
||||
assert_eq!(body["rejected"]["kind"], "insufficient_quota");
|
||||
assert_eq!(body["rejected"]["available"], 100);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn deactivated_account_resolves_as_invalid_no_clue() {
|
||||
let Some((base, pool)) = spawn_or_skip("deactivated_account_resolves_as_invalid_no_clue").await
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let raw = format!("sk-{}", Uuid::new_v4());
|
||||
seed_key(&pool, 1000, &raw, true).await; // deactivated
|
||||
let c = client();
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/resolve"),
|
||||
json!({"api_key": raw}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
// Indistinguishable from an unknown key.
|
||||
assert_eq!(r.status(), 401);
|
||||
let body: Value = r.json().await.unwrap();
|
||||
assert_eq!(body["error"]["code"], "invalid_api_key");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_client_auth_is_401_before_db() {
|
||||
let Some((base, pool)) = spawn_or_skip("missing_client_auth_is_401_before_db").await else {
|
||||
return;
|
||||
};
|
||||
let raw = format!("sk-{}", Uuid::new_v4());
|
||||
seed_key(&pool, 1000, &raw, false).await;
|
||||
let c = client();
|
||||
// No bearer → rejected by client_auth.
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/resolve"),
|
||||
json!({"api_key": raw}),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 401);
|
||||
// Wrong bearer → also rejected.
|
||||
let r = post(
|
||||
&c,
|
||||
format!("{base}/authz/v1/resolve"),
|
||||
json!({"api_key": raw}),
|
||||
Some("wrong"),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 401);
|
||||
}
|
||||
204
crates/helexa-upstream/tests/ledger_pg.rs
Normal file
204
crates/helexa-upstream/tests/ledger_pg.rs
Normal file
@@ -0,0 +1,204 @@
|
||||
//! Integration tests for the allocation ledger against a real PostgreSQL.
|
||||
//!
|
||||
//! Gated on `UPSTREAM_TEST_DATABASE_URL` — when unset (CI's generic runner,
|
||||
//! local builds without a DB), every test logs a skip and returns, so
|
||||
//! `cargo test --workspace` stays green without Postgres. Point the env var
|
||||
//! at a throwaway database to exercise the no-overshoot guarantee and
|
||||
//! settle/release idempotency:
|
||||
//!
|
||||
//! UPSTREAM_TEST_DATABASE_URL=postgres://helexa:helexa@localhost/helexa_test \
|
||||
//! cargo test -p helexa-upstream --test ledger_pg
|
||||
|
||||
use helexa_upstream::db::connect_and_migrate;
|
||||
use helexa_upstream::ledger::{self, LedgerError};
|
||||
use sqlx::Executor;
|
||||
use sqlx::Row;
|
||||
use sqlx::postgres::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// Returns a migrated pool, or `None` (with a skip log) when the env var is
|
||||
/// unset.
|
||||
async fn pool_or_skip(test: &str) -> Option<PgPool> {
|
||||
let Ok(url) = std::env::var("UPSTREAM_TEST_DATABASE_URL") else {
|
||||
eprintln!("skipping {test}: UPSTREAM_TEST_DATABASE_URL not set");
|
||||
return None;
|
||||
};
|
||||
Some(
|
||||
connect_and_migrate(&url, 16)
|
||||
.await
|
||||
.expect("connect + migrate"),
|
||||
)
|
||||
}
|
||||
|
||||
/// Seed a verified user + account (with `total` allocation) + an active key
|
||||
/// (percent=100 so the account cap binds). Returns (account_id, key_id).
|
||||
async fn seed(pool: &PgPool, total: i64) -> (Uuid, Uuid) {
|
||||
let user_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO users (email, password_hash, email_verified) \
|
||||
VALUES ($1, 'x', true) RETURNING id",
|
||||
)
|
||||
.bind(format!("u-{}@test.local", Uuid::new_v4())),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let account_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO accounts (owner_user_id, allocation_total) \
|
||||
VALUES ($1, $2) RETURNING id",
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(total),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let key_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO api_keys (account_id, key_hash, key_prefix, limit_kind, limit_value) \
|
||||
VALUES ($1, $2, 'sk-test', 'percent', 100) RETURNING id",
|
||||
)
|
||||
.bind(account_id)
|
||||
.bind(Uuid::new_v4().as_bytes().to_vec()),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
(account_id, key_id)
|
||||
}
|
||||
|
||||
async fn account_cols(pool: &PgPool, account_id: Uuid) -> (i64, i64) {
|
||||
let row = pool
|
||||
.fetch_one(
|
||||
sqlx::query("SELECT allocation_spent, allocation_reserved FROM accounts WHERE id = $1")
|
||||
.bind(account_id),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
(row.get("allocation_spent"), row.get("allocation_reserved"))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_reserves_never_overshoot() {
|
||||
let Some(pool) = pool_or_skip("concurrent_reserves_never_overshoot").await else {
|
||||
return;
|
||||
};
|
||||
// Allocation admits exactly 5 reservations of 100 (cap 500).
|
||||
let (account_id, key_id) = seed(&pool, 500).await;
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for _ in 0..20 {
|
||||
let pool = pool.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
ledger::reserve(&pool, account_id, key_id, 100).await
|
||||
}));
|
||||
}
|
||||
let mut ok = 0;
|
||||
let mut quota = 0;
|
||||
for h in handles {
|
||||
match h.await.unwrap() {
|
||||
Ok(_) => ok += 1,
|
||||
Err(LedgerError::InsufficientQuota { .. }) => quota += 1,
|
||||
Err(e) => panic!("unexpected error: {e}"),
|
||||
}
|
||||
}
|
||||
assert_eq!(ok, 5, "exactly 5 reserves of 100 fit in a 500 allocation");
|
||||
assert_eq!(quota, 15);
|
||||
|
||||
let (spent, reserved) = account_cols(&pool, account_id).await;
|
||||
assert_eq!(spent, 0);
|
||||
assert_eq!(reserved, 500, "reserved exactly the cap, never over");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn settle_is_idempotent_and_reconciles_spend() {
|
||||
let Some(pool) = pool_or_skip("settle_is_idempotent_and_reconciles_spend").await else {
|
||||
return;
|
||||
};
|
||||
let (account_id, key_id) = seed(&pool, 1000).await;
|
||||
let rid = ledger::reserve(&pool, account_id, key_id, 400)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Settle actual=150 (< reserved 400): spent=150, reserved back to 0.
|
||||
ledger::settle(&pool, rid, 150).await.unwrap();
|
||||
let (spent, reserved) = account_cols(&pool, account_id).await;
|
||||
assert_eq!((spent, reserved), (150, 0));
|
||||
|
||||
// Second settle is a no-op.
|
||||
ledger::settle(&pool, rid, 999).await.unwrap();
|
||||
let (spent2, reserved2) = account_cols(&pool, account_id).await;
|
||||
assert_eq!((spent2, reserved2), (150, 0), "settle is idempotent");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn release_returns_reservation_and_is_idempotent() {
|
||||
let Some(pool) = pool_or_skip("release_returns_reservation_and_is_idempotent").await else {
|
||||
return;
|
||||
};
|
||||
let (account_id, key_id) = seed(&pool, 1000).await;
|
||||
let rid = ledger::reserve(&pool, account_id, key_id, 300)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(account_cols(&pool, account_id).await, (0, 300));
|
||||
|
||||
ledger::release(&pool, rid).await.unwrap();
|
||||
assert_eq!(account_cols(&pool, account_id).await, (0, 0));
|
||||
// Idempotent; settle-after-release also a no-op.
|
||||
ledger::release(&pool, rid).await.unwrap();
|
||||
ledger::settle(&pool, rid, 100).await.unwrap();
|
||||
assert_eq!(account_cols(&pool, account_id).await, (0, 0));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hardcap_key_subcap_binds_below_account() {
|
||||
let Some(pool) = pool_or_skip("hardcap_key_subcap_binds_below_account").await else {
|
||||
return;
|
||||
};
|
||||
// Account has 1000 but the key is hard-capped at 200.
|
||||
let user_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO users (email, password_hash, email_verified) \
|
||||
VALUES ($1, 'x', true) RETURNING id",
|
||||
)
|
||||
.bind(format!("u-{}@test.local", Uuid::new_v4())),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let account_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO accounts (owner_user_id, allocation_total) VALUES ($1, 1000) RETURNING id",
|
||||
)
|
||||
.bind(user_id),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let key_id: Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"INSERT INTO api_keys (account_id, key_hash, key_prefix, limit_kind, limit_value) \
|
||||
VALUES ($1, $2, 'sk-test', 'hardcap', 200) RETURNING id",
|
||||
)
|
||||
.bind(account_id)
|
||||
.bind(Uuid::new_v4().as_bytes().to_vec()),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
|
||||
ledger::reserve(&pool, account_id, key_id, 200)
|
||||
.await
|
||||
.unwrap();
|
||||
match ledger::reserve(&pool, account_id, key_id, 1).await {
|
||||
Err(LedgerError::InsufficientQuota { available, .. }) => assert_eq!(available, 0),
|
||||
other => panic!("expected InsufficientQuota, got {other:?}"),
|
||||
}
|
||||
}
|
||||
@@ -17,14 +17,22 @@ listen = "0.0.0.0:8088"
|
||||
# -- Downstream cortexes -------------------------------------------------
|
||||
# Each [[cortexes]] entry is an operator-run cortex the router may dispatch
|
||||
# to. The router forwards the client's bearer verbatim (auth stays at
|
||||
# cortex) and routes on capacity. Outbound TLS to each cortex is verified.
|
||||
# cortex) and routes on capacity (preferring matching `region`).
|
||||
#
|
||||
# The skeleton only loads this list; capacity/catalogue polling and
|
||||
# capacity-aware dispatch arrive in later issues.
|
||||
# Outbound TLS pinning (optional): set `tls_ca` to a PEM trust anchor that
|
||||
# enrols this cortex — the CA (or self-signed cert) its TLS cert must chain
|
||||
# to. The router then trusts ONLY that anchor for this cortex (platform
|
||||
# roots disabled), so the router->cortex hop (which carries the client's
|
||||
# bearer) reaches the cert you expect and a rogue endpoint presenting any
|
||||
# other cert is rejected at the handshake. A cortex whose `tls_ca` fails to
|
||||
# load is disabled (fail closed). Omit `tls_ca` for a publicly-trusted cert
|
||||
# or plaintext http:// on a private (e.g. WireGuard) network.
|
||||
|
||||
# [[cortexes]]
|
||||
# name = "lair-cafe"
|
||||
# endpoint = "https://cortex.lair.cafe"
|
||||
# region = "eu-west"
|
||||
# tls_ca = "/etc/helexa-router/pins/lair-cafe.pem"
|
||||
|
||||
# [[cortexes]]
|
||||
# name = "example-operator"
|
||||
|
||||
37
helexa-upstream.example.toml
Normal file
37
helexa-upstream.example.toml
Normal file
@@ -0,0 +1,37 @@
|
||||
# helexa-upstream.example.toml — mesh-level account/authorization authority
|
||||
#
|
||||
# Copy to helexa-upstream.toml and adjust. Env overrides use the UPSTREAM_
|
||||
# prefix with __ separators, e.g. UPSTREAM_DB__URL=postgres://...
|
||||
|
||||
[server]
|
||||
# Plaintext listener; edge nginx terminates TLS (consistent with the stack).
|
||||
listen = "0.0.0.0:8090"
|
||||
|
||||
[db]
|
||||
# PostgreSQL connection URL. Required.
|
||||
url = "postgres://helexa:helexa@localhost/helexa_upstream"
|
||||
# max_connections = 16
|
||||
|
||||
[grant]
|
||||
# Flat free token grant every email-verified account receives (the floor of
|
||||
# the hybrid allocation; single-use top-up codes extend it).
|
||||
# free_token_grant = 1000000
|
||||
|
||||
[abuse]
|
||||
# When this many accounts share one registration fingerprint, all are
|
||||
# silently deactivated (no notice to the user).
|
||||
# fingerprint_account_threshold = 5
|
||||
|
||||
# -- Client auth: credentials operators' cortexes present to /authz/v1.
|
||||
# Each token maps to an operator_id (served-usage attribution). When no
|
||||
# tokens are configured the authz surface is OPEN (dev only). Distinct from
|
||||
# end-user API keys, which ride inside the resolve request body.
|
||||
# [[client_auth.tokens]]
|
||||
# token = "replace-with-a-strong-shared-secret"
|
||||
# operator_id = "lair-cafe"
|
||||
|
||||
[authz]
|
||||
# Open reservations older than this are swept (released), self-healing a
|
||||
# reservation whose settle/release from a cortex was lost.
|
||||
# reservation_ttl_secs = 120
|
||||
# sweep_interval_secs = 60
|
||||
20
helexa.ai/.env.example
Normal file
20
helexa.ai/.env.example
Normal file
@@ -0,0 +1,20 @@
|
||||
# helexa.ai frontend env. Copy to .env.local for local dev (gitignored).
|
||||
|
||||
# Mesh data-plane (helexa-router, OpenAI-compatible inference). In dev,
|
||||
# vite proxies /v1 and /health here.
|
||||
VITE_ROUTER_BASE_URL=http://localhost:8088
|
||||
|
||||
# Account control-plane (helexa-upstream). In dev, vite proxies /api here
|
||||
# (rewritten to /web/v1).
|
||||
VITE_ACCOUNT_BASE_URL=http://localhost:8090
|
||||
|
||||
# Public-beta banner.
|
||||
VITE_PUBLIC_BETA=true
|
||||
|
||||
# Models for the chat workspace (F3+).
|
||||
# VITE_ANON_MODEL=...
|
||||
# VITE_DEFAULT_MODEL=...
|
||||
|
||||
# Develop the account dashboard (F4) against an in-browser mock before the
|
||||
# upstream account API ships.
|
||||
# VITE_USE_MOCK_ACCOUNT_API=true
|
||||
6
helexa.ai/.gitignore
vendored
Normal file
6
helexa.ai/.gitignore
vendored
Normal file
@@ -0,0 +1,6 @@
|
||||
node_modules
|
||||
dist
|
||||
*.local
|
||||
.env.local
|
||||
.env.*.local
|
||||
*.tsbuildinfo
|
||||
34
helexa.ai/README.md
Normal file
34
helexa.ai/README.md
Normal file
@@ -0,0 +1,34 @@
|
||||
# helexa.ai
|
||||
|
||||
The public-beta frontend for the helexa mesh: a chat-first landing experience
|
||||
(anonymous + authenticated, with all chat history kept client-side in
|
||||
IndexedDB — no server-side history), a `/mission` page on European digital
|
||||
sovereignty, and full account self-service (register, recover, manage API
|
||||
keys, set per-key limits, redeem top-up codes) against `helexa-upstream`.
|
||||
|
||||
Vite + React (SWC) + TypeScript + react-bootstrap + react-router + react-i18next.
|
||||
Lives as a top-level folder in the cortex monorepo; it is **not** a Cargo crate.
|
||||
|
||||
## Develop
|
||||
|
||||
```sh
|
||||
cd helexa.ai
|
||||
npm install
|
||||
cp .env.example .env.local # adjust backend URLs
|
||||
npm run dev # vite dev server, proxies /v1+/health → router, /api → upstream
|
||||
```
|
||||
|
||||
Other scripts: `npm run build` (`tsc -b && vite build` → `dist/`), `npm run
|
||||
preview`, `npm run lint`, `npm run typecheck`.
|
||||
|
||||
In dev, `vite.config.ts` proxies the mesh data-plane (helexa-router) and the
|
||||
account control-plane (helexa-upstream) same-origin. Run a local router
|
||||
(`cargo run -p helexa-router`) for the chat path and a local helexa-upstream
|
||||
for the account path.
|
||||
|
||||
## Status
|
||||
|
||||
F0 scaffold. Theming + i18n (33 languages, usage-ordered selector), the
|
||||
`/mission` page, the chat workspace (Dexie + streaming), and the account
|
||||
dashboard land in subsequent phases — see
|
||||
`~/.claude/plans/we-need-to-plan-modular-graham.md`.
|
||||
23
helexa.ai/eslint.config.js
Normal file
23
helexa.ai/eslint.config.js
Normal file
@@ -0,0 +1,23 @@
|
||||
import js from "@eslint/js";
|
||||
import globals from "globals";
|
||||
import reactHooks from "eslint-plugin-react-hooks";
|
||||
import reactRefresh from "eslint-plugin-react-refresh";
|
||||
import tseslint from "typescript-eslint";
|
||||
import { defineConfig, globalIgnores } from "eslint/config";
|
||||
|
||||
export default defineConfig([
|
||||
globalIgnores(["dist"]),
|
||||
{
|
||||
files: ["**/*.{ts,tsx}"],
|
||||
extends: [
|
||||
js.configs.recommended,
|
||||
tseslint.configs.recommended,
|
||||
reactHooks.configs.flat.recommended,
|
||||
reactRefresh.configs.vite,
|
||||
],
|
||||
languageOptions: {
|
||||
ecmaVersion: 2020,
|
||||
globals: globals.browser,
|
||||
},
|
||||
},
|
||||
]);
|
||||
24
helexa.ai/index.html
Normal file
24
helexa.ai/index.html
Normal file
@@ -0,0 +1,24 @@
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<title>helexa.ai</title>
|
||||
<meta name="title" content="helexa.ai" />
|
||||
<meta
|
||||
name="description"
|
||||
content="helexa — near-frontier AI on a sovereign, operator-run mesh. Chat now; bring your own key."
|
||||
/>
|
||||
<meta property="og:type" content="website" />
|
||||
<meta property="og:url" content="https://helexa.ai/" />
|
||||
<meta property="og:title" content="helexa.ai" />
|
||||
<meta
|
||||
property="og:description"
|
||||
content="helexa — near-frontier AI on a sovereign, operator-run mesh."
|
||||
/>
|
||||
</head>
|
||||
<body>
|
||||
<div id="root"></div>
|
||||
<script type="module" src="/src/main.tsx"></script>
|
||||
</body>
|
||||
</html>
|
||||
4115
helexa.ai/package-lock.json
generated
Normal file
4115
helexa.ai/package-lock.json
generated
Normal file
File diff suppressed because it is too large
Load Diff
40
helexa.ai/package.json
Normal file
40
helexa.ai/package.json
Normal file
@@ -0,0 +1,40 @@
|
||||
{
|
||||
"name": "helexa.ai",
|
||||
"private": true,
|
||||
"version": "0.0.0",
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"dev": "vite",
|
||||
"build": "tsc -b && vite build",
|
||||
"preview": "vite preview",
|
||||
"lint": "eslint .",
|
||||
"typecheck": "tsc -b"
|
||||
},
|
||||
"dependencies": {
|
||||
"@fingerprintjs/fingerprintjs": "^4.6.2",
|
||||
"bootstrap": "^5.3.8",
|
||||
"dexie": "^4.2.0",
|
||||
"dexie-react-hooks": "^4.2.0",
|
||||
"i18next": "^25.7.1",
|
||||
"react": "^19.2.0",
|
||||
"react-bootstrap": "^2.10.10",
|
||||
"react-dom": "^19.2.0",
|
||||
"react-i18next": "^16.4.0",
|
||||
"react-icons": "^5.5.0",
|
||||
"react-router-dom": "^7.10.1"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@eslint/js": "^9.39.1",
|
||||
"@types/node": "^24.10.1",
|
||||
"@types/react": "^19.2.5",
|
||||
"@types/react-dom": "^19.2.3",
|
||||
"@vitejs/plugin-react-swc": "^4.2.0",
|
||||
"eslint": "^9.39.1",
|
||||
"eslint-plugin-react-hooks": "^7.0.1",
|
||||
"eslint-plugin-react-refresh": "^0.4.24",
|
||||
"globals": "^16.5.0",
|
||||
"typescript": "~5.9.3",
|
||||
"typescript-eslint": "^8.46.4",
|
||||
"vite": "^7.2.0"
|
||||
}
|
||||
}
|
||||
12
helexa.ai/src/App.tsx
Normal file
12
helexa.ai/src/App.tsx
Normal file
@@ -0,0 +1,12 @@
|
||||
import { Container } from "react-bootstrap";
|
||||
|
||||
// F0 scaffold shell. Theming, i18n, routing, the chat workspace, mission
|
||||
// page and account dashboard land in the F1+ phases.
|
||||
export default function App() {
|
||||
return (
|
||||
<Container className="py-5">
|
||||
<h1 className="mb-2">helexa.ai</h1>
|
||||
<p className="text-muted">Public beta — coming online.</p>
|
||||
</Container>
|
||||
);
|
||||
}
|
||||
16
helexa.ai/src/index.css
Normal file
16
helexa.ai/src/index.css
Normal file
@@ -0,0 +1,16 @@
|
||||
/* Minimal reset; the full theme (CSS custom properties, light/dark, accent
|
||||
* palette) is ported from the reference site in F1. */
|
||||
:root {
|
||||
color-scheme: light dark;
|
||||
}
|
||||
|
||||
* {
|
||||
box-sizing: border-box;
|
||||
}
|
||||
|
||||
html,
|
||||
body,
|
||||
#root {
|
||||
margin: 0;
|
||||
min-height: 100vh;
|
||||
}
|
||||
11
helexa.ai/src/main.tsx
Normal file
11
helexa.ai/src/main.tsx
Normal file
@@ -0,0 +1,11 @@
|
||||
import { StrictMode } from "react";
|
||||
import { createRoot } from "react-dom/client";
|
||||
import "bootstrap/dist/css/bootstrap.min.css";
|
||||
import "./index.css";
|
||||
import App from "./App.tsx";
|
||||
|
||||
createRoot(document.getElementById("root")!).render(
|
||||
<StrictMode>
|
||||
<App />
|
||||
</StrictMode>,
|
||||
);
|
||||
28
helexa.ai/tsconfig.app.json
Normal file
28
helexa.ai/tsconfig.app.json
Normal file
@@ -0,0 +1,28 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.app.tsbuildinfo",
|
||||
"target": "ES2022",
|
||||
"useDefineForClassFields": true,
|
||||
"lib": ["ES2022", "DOM", "DOM.Iterable"],
|
||||
"module": "ESNext",
|
||||
"types": ["vite/client"],
|
||||
"skipLibCheck": true,
|
||||
|
||||
/* Bundler mode */
|
||||
"moduleResolution": "bundler",
|
||||
"allowImportingTsExtensions": true,
|
||||
"verbatimModuleSyntax": true,
|
||||
"moduleDetection": "force",
|
||||
"noEmit": true,
|
||||
"jsx": "react-jsx",
|
||||
|
||||
/* Linting */
|
||||
"strict": true,
|
||||
"noUnusedLocals": true,
|
||||
"noUnusedParameters": true,
|
||||
"erasableSyntaxOnly": true,
|
||||
"noFallthroughCasesInSwitch": true,
|
||||
"noUncheckedSideEffectImports": true
|
||||
},
|
||||
"include": ["src"]
|
||||
}
|
||||
7
helexa.ai/tsconfig.json
Normal file
7
helexa.ai/tsconfig.json
Normal file
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"files": [],
|
||||
"references": [
|
||||
{ "path": "./tsconfig.app.json" },
|
||||
{ "path": "./tsconfig.node.json" }
|
||||
]
|
||||
}
|
||||
26
helexa.ai/tsconfig.node.json
Normal file
26
helexa.ai/tsconfig.node.json
Normal file
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.node.tsbuildinfo",
|
||||
"target": "ES2023",
|
||||
"lib": ["ES2023"],
|
||||
"module": "ESNext",
|
||||
"types": ["node"],
|
||||
"skipLibCheck": true,
|
||||
|
||||
/* Bundler mode */
|
||||
"moduleResolution": "bundler",
|
||||
"allowImportingTsExtensions": true,
|
||||
"verbatimModuleSyntax": true,
|
||||
"moduleDetection": "force",
|
||||
"noEmit": true,
|
||||
|
||||
/* Linting */
|
||||
"strict": true,
|
||||
"noUnusedLocals": true,
|
||||
"noUnusedParameters": true,
|
||||
"erasableSyntaxOnly": true,
|
||||
"noFallthroughCasesInSwitch": true,
|
||||
"noUncheckedSideEffectImports": true
|
||||
},
|
||||
"include": ["vite.config.ts"]
|
||||
}
|
||||
29
helexa.ai/vite.config.ts
Normal file
29
helexa.ai/vite.config.ts
Normal file
@@ -0,0 +1,29 @@
|
||||
import { defineConfig, loadEnv } from "vite";
|
||||
import react from "@vitejs/plugin-react-swc";
|
||||
|
||||
// During `vite dev`, proxy the mesh data-plane (helexa-router, OpenAI-
|
||||
// compatible) and the account control-plane (helexa-upstream) so the SPA
|
||||
// talks to them same-origin without CORS. Targets are overridable via env.
|
||||
// VITE_ROUTER_BASE_URL — helexa-router (default http://localhost:8088)
|
||||
// VITE_ACCOUNT_BASE_URL — helexa-upstream (default http://localhost:8090)
|
||||
export default defineConfig(({ mode }) => {
|
||||
const env = loadEnv(mode, process.cwd(), "VITE_");
|
||||
const router = env.VITE_ROUTER_BASE_URL || "http://localhost:8088";
|
||||
const account = env.VITE_ACCOUNT_BASE_URL || "http://localhost:8090";
|
||||
return {
|
||||
plugins: [react()],
|
||||
server: {
|
||||
proxy: {
|
||||
"/v1": { target: router, changeOrigin: true },
|
||||
"/health": { target: router, changeOrigin: true },
|
||||
// The frontend calls /api/*; helexa-upstream serves /web/v1/*.
|
||||
"/api": {
|
||||
target: account,
|
||||
changeOrigin: true,
|
||||
rewrite: (p) => p.replace(/^\/api/, "/web/v1"),
|
||||
},
|
||||
},
|
||||
},
|
||||
build: { outDir: "dist" },
|
||||
};
|
||||
});
|
||||
Reference in New Issue
Block a user