use super::{
Cell, Finding, Health, Interrupted, Priority, ReadError, Report, Source, Status, Table, iso,
};
use crate::json::OValue;
pub const LOG_PAGE_SIZE: u32 = 50;
pub const LIMITATIONS: [&str; 7] = [
"Read-only operational evidence, not a security certification or availability SLA. No inference or configuration changes are performed.",
"All telemetry uses the displayed fixed window. Configuration and organisation/catalog reads describe current state, not daily changes.",
"Requests, errors, feedback and log totals are independent server aggregates, not an atomic snapshot. An AIRS block is not necessarily an outage.",
"Cost values are cents; latency values are milliseconds. Feedback scores may be negative. Null means unknown, not zero.",
"Only aggregate numbers and HTTP status codes are retained. Keys, user identities, workspace names/IDs, metadata, URLs, prompts, raw logs and configuration bodies are excluded.",
"Catalog entries describe available capabilities, not enabled protection. Organisation limit sentinel values are not interpreted as quota exhaustion.",
"Incomplete sources and page budgets are explicit. Current inventory lists have no verified continuation contract; total/has_more mismatches are partial.",
];
const GENERIC_NOTE: &str =
"Request, scope or response validation failed; upstream details omitted.";
const QUOTA_NOTE: &str = "Quota exceeded or quota state unknown; aggregates may be incomplete.";
const LIST_NOTE: &str = "List is truncated or total differs; continuation is not documented.";
const BUDGET_NOTE: &str =
"Transaction page budget reached. Increase --max-pages to collect the remaining rows.";
const TELEMETRY_WINDOW: &str = "Displayed telemetry window";
const WORKSPACE_CONFIGURATION: &str = "Current selected workspace configuration";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Chart {
Requests,
Errors,
Users,
Cost,
Latency,
Tokens,
CacheSummary,
CacheHitTrend,
UserTrends,
ErrorTrends,
ErrorCategoryTrends,
GroupedErrors,
RescuedRetries,
FeedbackTrend,
FeedbackWeighted,
FeedbackScoreDistribution,
FeedbackModels,
}
impl Chart {
pub const ALL: [Chart; 17] = [
Chart::Requests,
Chart::Errors,
Chart::Users,
Chart::Cost,
Chart::Latency,
Chart::Tokens,
Chart::CacheSummary,
Chart::CacheHitTrend,
Chart::UserTrends,
Chart::ErrorTrends,
Chart::ErrorCategoryTrends,
Chart::GroupedErrors,
Chart::RescuedRetries,
Chart::FeedbackTrend,
Chart::FeedbackWeighted,
Chart::FeedbackScoreDistribution,
Chart::FeedbackModels,
];
pub fn method(self) -> &'static str {
self.spec().0
}
pub fn source(self) -> &'static str {
self.spec().1
}
fn fields(self) -> &'static [&'static str] {
self.spec().2
}
fn spec(self) -> (&'static str, &'static str, &'static [&'static str]) {
match self {
Chart::Requests => ("requests", "telemetry.requests", &["total"]),
Chart::Errors => ("errors", "telemetry.errors", &["total"]),
Chart::Users => ("users", "telemetry.users", &["total"]),
Chart::Cost => ("cost", "telemetry.cost", &["total", "avg"]),
Chart::Latency => (
"latency",
"telemetry.latency",
&["total", "p50", "p90", "p99"],
),
Chart::Tokens => (
"tokens",
"telemetry.tokens",
&["total", "total_request_units", "total_response_units"],
),
Chart::CacheSummary => (
"cacheSummary",
"telemetry.cacheSummary",
&[
"summary.cacheHits",
"summary.avgCacheLatency",
"summary.totalRequests",
"summary.cacheSpeedup",
],
),
Chart::CacheHitTrend => (
"cacheHitTrend",
"telemetry.cacheHitTrend",
&["summary.totalCacheHits", "summary.hitRate"],
),
Chart::UserTrends => (
"userTrends",
"telemetry.userTrends",
&["summary.total", "summary.unique", "summary.avg"],
),
Chart::ErrorTrends => (
"errorTrends",
"telemetry.errorTrends",
&["summary.errorPercent"],
),
Chart::ErrorCategoryTrends => (
"errorCategoryTrends",
"telemetry.errorCategoryTrends",
&["summary.totalErrors", "summary.totalUniqueErrorCodes"],
),
Chart::GroupedErrors => ("groupedErrors", "telemetry.groupedErrors", &["total"]),
Chart::RescuedRetries => (
"rescuedRetries",
"telemetry.rescuedRetries",
&["total", "retryTotal", "fallbackTotal"],
),
Chart::FeedbackTrend => ("feedbackTrend", "telemetry.feedbackTrend", &["total"]),
Chart::FeedbackWeighted => {
("feedbackWeighted", "telemetry.feedbackWeighted", &["total"])
}
Chart::FeedbackScoreDistribution => (
"feedbackScoreDistribution",
"telemetry.feedbackScoreDistribution",
&["total"],
),
Chart::FeedbackModels => ("feedbackModels", "telemetry.feedbackModels", &[]),
}
}
}
pub(crate) trait Reader {
async fn workspaces(&self) -> Result<OValue, ReadError>;
async fn chart(&self, chart: Chart, slug: &str) -> Result<OValue, ReadError>;
async fn filter_boundaries(&self, slug: &str) -> Result<OValue, ReadError>;
async fn logs(&self, slug: &str, page: u32) -> Result<OValue, ReadError>;
async fn configs(&self, workspace_id: &str) -> Result<OValue, ReadError>;
async fn service_keys(&self, workspace_id: &str) -> Result<OValue, ReadError>;
async fn user_keys(&self, workspace_id: &str) -> Result<OValue, ReadError>;
async fn organisation_info(&self) -> Result<OValue, ReadError>;
async fn catalog(&self) -> Result<OValue, ReadError>;
}
pub struct Options<'a> {
pub title: &'a str,
pub max_pages: u32,
pub workspace: &'a str,
pub start_ms: i64,
pub end_ms: i64,
pub now: &'a dyn Fn() -> i64,
}
type Step<T> = Result<T, ReadError>;
const INVALID: ReadError = ReadError::Failed;
fn note(error: &ReadError) -> String {
match error {
ReadError::Status(code) if (400..=599).contains(code) => {
format!("HTTP {code}: request failed; check access and service availability.")
}
_ => GENERIC_NOTE.to_owned(),
}
}
fn record(value: Option<&OValue>) -> Step<&OValue> {
match value {
Some(object @ OValue::Object(_)) => Ok(object),
_ => Err(INVALID),
}
}
fn numeric(value: Option<&OValue>) -> Step<Option<f64>> {
match value {
Some(OValue::Null) => Ok(None),
Some(OValue::Number(number)) => number
.as_f64()
.filter(|value| value.is_finite())
.map(Some)
.ok_or(INVALID),
_ => Err(INVALID),
}
}
fn safe_integer(value: f64) -> bool {
value.fract() == 0.0 && value.abs() <= 9_007_199_254_740_991.0
}
fn count(value: Option<&OValue>) -> Step<u64> {
match numeric(value)? {
Some(value) if safe_integer(value) && value >= 0.0 => {
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
let integer = value as u64;
Ok(integer)
}
_ => Err(INVALID),
}
}
fn date(value: Option<&OValue>) -> Option<i64> {
super::runtime::parse_date(value)
}
struct Collector {
report: Report,
metrics: Vec<Vec<Cell>>,
statuses: Vec<Vec<Cell>>,
trends: Vec<Vec<Cell>>,
inventory: Vec<Vec<Cell>>,
coverage: Vec<Vec<Cell>>,
}
impl Collector {
fn finding(&mut self, priority: Priority, title: &str, evidence: &str, recommendation: &str) {
self.finding_owned(
priority,
title.to_owned(),
evidence.to_owned(),
recommendation,
"telemetry.errorCategoryTrends",
);
}
fn finding_owned(
&mut self,
priority: Priority,
title: String,
evidence: String,
recommendation: &str,
source: &str,
) {
self.report.findings.push(Finding {
priority,
title,
evidence,
recommendation: recommendation.to_owned(),
source: source.to_owned(),
});
}
fn finish(&mut self, mut source: Source, result: Step<()>) -> Result<(), Interrupted> {
if let Err(error) = result {
if error == ReadError::Cancelled {
return Err(Interrupted);
}
source.status = if source.records > 0 {
Status::Partial
} else {
Status::Unavailable
};
source.notes.push(note(&error));
}
self.report.sources.push(source);
Ok(())
}
}
fn open(name: &'static str, window: &'static str) -> Source {
Source {
name,
method: name,
window: window.to_owned(),
status: Status::Complete,
records: 0,
pages: 0,
notes: Vec::new(),
}
}
fn envelope<'a>(value: &'a OValue, source: &mut Source) -> Step<&'a OValue> {
let value = record(Some(value))?;
if !matches!(value.get("success"), Some(OValue::Bool(true))) {
return Err(INVALID);
}
let data = record(value.get("data"))?;
if !matches!(data.get("isQuotaExceeded"), Some(OValue::Bool(false))) {
source.status = Status::Partial;
source.notes.push(QUOTA_NOTE.to_owned());
}
Ok(data)
}
fn list<'a>(value: &'a OValue, source: &mut Source) -> Step<&'a [OValue]> {
let value = record(Some(value))?;
if matches!(value.get("success"), Some(OValue::Bool(false))) {
return Err(INVALID);
}
let Some(rows) = value.get("data").and_then(OValue::as_array) else {
return Err(INVALID);
};
for row in rows {
record(Some(row))?;
}
let total = count(value.get("total"))?;
if matches!(value.get("has_more"), Some(OValue::Bool(true))) || total != rows.len() as u64 {
source.status = Status::Partial;
source.notes.push(LIST_NOTE.to_owned());
}
source.records = rows.len() as u64;
source.pages = 1;
Ok(rows)
}
fn field<'a>(data: &'a OValue, field: &str) -> Step<Option<&'a OValue>> {
let mut node = data;
let mut parts = field.split('.').peekable();
while let Some(part) = parts.next() {
let object = record(Some(node))?;
match object.get(part) {
Some(child) if parts.peek().is_some() => node = child,
found if parts.peek().is_none() => return Ok(found),
_ => return Err(INVALID),
}
}
Ok(None)
}
fn metric_rows(chart: Chart, data: &OValue) -> Step<Vec<Vec<Cell>>> {
let method = chart.method();
let signed = method.starts_with("feedback");
let mut rows = Vec::new();
for name in chart.fields() {
let value = numeric(field(data, name)?)?;
let fractional = signed
|| chart == Chart::Cost
|| chart == Chart::Latency
|| ["avg", "Latency", "Speedup", "Rate", "Percent"]
.iter()
.any(|suffix| name.ends_with(suffix));
if let Some(value) = value
&& ((!signed && value < 0.0) || (!fractional && !safe_integer(value)))
{
return Err(INVALID);
}
rows.push(vec![
Cell::from(method),
Cell::from(*name),
value.map_or(Cell::Null, Cell::Number),
]);
}
Ok(rows)
}
pub(crate) async fn collect(
reader: &impl Reader,
options: &Options<'_>,
) -> Result<Report, Interrupted> {
let mut c = Collector {
report: Report {
title: options.title.to_owned(),
collection_started_at: iso((options.now)()),
generated_at: String::new(),
window_start: iso(options.start_ms),
window_end: iso(options.end_ms),
health: Health::Unknown,
sources: Vec::new(),
findings: Vec::new(),
tables: Vec::new(),
},
metrics: Vec::new(),
statuses: Vec::new(),
trends: Vec::new(),
inventory: Vec::new(),
coverage: Vec::new(),
};
let mut s = open(
"workspaces.list",
"Current active tenant workspaces (admin)",
);
let mut workspace: Option<(String, String)> = None;
let result = async {
let answer = reader.workspaces().await?;
let rows = list(&answer, &mut s)?;
c.inventory.push(vec![
Cell::from("workspaces.list"),
Cell::from("Active workspaces returned"),
Cell::Count(rows.len() as u64),
]);
let text =
|row: &OValue, key: &str| row.get(key).and_then(OValue::as_str).map(str::to_owned);
let by = |keys: &[&str]| -> Vec<&OValue> {
rows.iter()
.filter(|row| {
keys.iter()
.any(|key| text(row, key).as_deref() == Some(options.workspace))
})
.collect()
};
let exact = by(&["id", "slug"]);
let matches = if !exact.is_empty() {
exact
} else if s.status == Status::Partial {
Vec::new()
} else {
by(&["name"])
};
match matches.as_slice() {
[only] => match (text(only, "id"), text(only, "slug")) {
(Some(id), Some(slug)) => {
workspace = Some((id, slug));
Ok(())
}
_ => Err(INVALID),
},
_ => Err(INVALID),
}
}
.await;
c.finish(s, result)?;
let slug = workspace.as_ref().map(|(_, slug)| slug.as_str());
let workspace_id = workspace.as_ref().map(|(id, _)| id.as_str());
for chart in Chart::ALL {
let mut s = open(chart.source(), TELEMETRY_WINDOW);
let result = async {
let slug = slug.ok_or(INVALID)?;
let answer = reader.chart(chart, slug).await?;
let data = envelope(&answer, &mut s)?;
let mut rows = metric_rows(chart, data)?;
let buckets = match data.get("records") {
None | Some(OValue::Null) => data.get("trend"),
found => found,
};
s.records = match buckets {
None => 1,
Some(OValue::Array(items)) => items.len() as u64,
Some(_) => return Err(INVALID),
};
s.pages = 1;
if chart == Chart::FeedbackModels {
rows.push(vec![
Cell::from("feedbackModels"),
Cell::from("Models with feedback"),
Cell::Count(s.records),
]);
}
if chart == Chart::ErrorCategoryTrends {
let Some(trend) = data.get("trend").and_then(OValue::as_array) else {
return Err(INVALID);
};
let mut entries = Vec::new();
for entry in trend {
let entry = record(Some(entry))?;
entries.push((
count(entry.get("response_status_code"))?,
count(entry.get("count"))?,
));
}
c.statuses.extend(
entries
.iter()
.map(|(code, n)| vec![Cell::Count(*code), Cell::Count(*n)]),
);
if entries
.iter()
.any(|(code, n)| (200..300).contains(code) && *n > 0)
{
c.finding(
Priority::Review,
"Success-status responses appear in error analytics",
"The server classifies some HTTP 2xx responses as errors. Error analytics are not identical to non-2xx transaction counts.",
"Inspect the underlying application/provider error classifications before calculating an HTTP failure rate.",
);
}
if entries
.iter()
.any(|(code, n)| *n > 0 && (*code == 429 || *code >= 500))
{
c.finding(
Priority::Attention,
"Rate limiting or server errors observed",
"Error categories include HTTP 429 or 5xx responses in the selected window.",
"Review provider availability, quotas and retry policies; correlate with transaction logs.",
);
} else if entries.iter().any(|(_, n)| *n > 0) {
c.finding(
Priority::Review,
"Gateway errors or policy blocks observed",
"Nonzero error-category counts were returned.",
"Review status categories; distinguish expected AIRS blocks from integration failures.",
);
}
}
if chart == Chart::GroupedErrors {
let Some(trend) = data.get("trend").and_then(OValue::as_array) else {
return Err(INVALID);
};
let mut entries = Vec::new();
for bucket in trend {
let bucket = record(Some(bucket))?;
let (Some(time), Some(OValue::Array(items))) =
(date(bucket.get("x")), bucket.get("y"))
else {
return Err(INVALID);
};
for item in items {
let item = record(Some(item))?;
entries.push(vec![
Cell::Text(iso(time)),
Cell::Count(count(item.get("response_status_code"))?),
Cell::Count(count(item.get("count"))?),
]);
}
}
c.trends.extend(entries);
}
c.metrics.extend(rows);
Ok(())
}
.await;
c.finish(s, result)?;
}
let mut s = open("telemetry.filterBoundaries", TELEMETRY_WINDOW);
let result = async {
let slug = slug.ok_or(INVALID)?;
let answer = reader.filter_boundaries(slug).await?;
let data = envelope(&answer, &mut s)?;
for key in ["unique_ai_models", "unique_status_codes"] {
let Some(values) = data.get(key).and_then(OValue::as_array) else {
return Err(INVALID);
};
c.metrics.push(vec![
Cell::from("filterBoundaries"),
Cell::Text(format!("{key} count")),
Cell::Count(values.len() as u64),
]);
}
s.records = 1;
s.pages = 1;
Ok(())
}
.await;
c.finish(s, result)?;
let mut s = open("telemetry.logs", TELEMETRY_WINDOW);
let mut ids = std::collections::BTreeSet::new();
let mut codes: std::collections::BTreeMap<u64, u64> = std::collections::BTreeMap::new();
let mut expected: Option<u64> = None;
let result = async {
for page in 0..options.max_pages {
let slug = slug.ok_or(INVALID)?;
let answer = reader.logs(slug, page).await?;
let data = envelope(&answer, &mut s)?;
s.pages += 1;
let total = count(data.get("total"))?;
let wanted = *expected.get_or_insert(total);
let Some(records) = data.get("records").and_then(OValue::as_array) else {
return Err(INVALID);
};
if total != wanted || records.len() > LOG_PAGE_SIZE as usize {
return Err(INVALID);
}
for row in records {
let row = record(Some(row))?;
let id = match row.get("id") {
Some(OValue::String(id)) if !id.is_empty() && !ids.contains(id) => id.clone(),
_ => return Err(INVALID),
};
if row.get("workspace_slug").and_then(OValue::as_str) != Some(slug) {
return Err(INVALID);
}
match date(row.get("created_at")) {
Some(time) if time >= options.start_ms && time <= options.end_ms => {}
_ => return Err(INVALID),
}
let code = count(row.get("response_status_code"))?;
ids.insert(id);
*codes.entry(code).or_default() += 1;
s.records = ids.len() as u64;
}
let collected = ids.len() as u64;
if collected > wanted {
return Err(INVALID);
}
if collected == wanted {
return Ok(());
}
if records.is_empty() {
return Err(INVALID);
}
}
s.status = Status::Partial;
s.notes.push(BUDGET_NOTE.to_owned());
Ok(())
}
.await;
c.coverage.push(vec![
Cell::from("Server total"),
expected.map_or(Cell::Null, Cell::Count),
]);
c.coverage.push(vec![
Cell::from("Unique collected"),
Cell::Count(ids.len() as u64),
]);
for (code, n) in &codes {
c.coverage
.push(vec![Cell::Text(format!("HTTP {code}")), Cell::Count(*n)]);
}
c.finish(s, result)?;
for name in ["configs.list", "apiKeys.listService", "apiKeys.listUser"] {
let mut s = open(name, WORKSPACE_CONFIGURATION);
let result = async {
let id = workspace_id.ok_or(INVALID)?;
let answer = match name {
"configs.list" => reader.configs(id).await?,
"apiKeys.listService" => reader.service_keys(id).await?,
_ => reader.user_keys(id).await?,
};
let rows = list(&answer, &mut s)?;
c.inventory.push(vec![
Cell::from(name),
Cell::from("Rows returned"),
Cell::Count(rows.len() as u64),
]);
Ok(())
}
.await;
c.finish(s, result)?;
}
let mut s = open("organisations.getInfo", "Current tenant configuration");
let result = async {
let answer = reader.organisation_info().await?;
let limits = record(Some(&answer))
.and_then(|info| record(info.get("benefits")))
.and_then(|benefits| record(benefits.get("limits")))?;
c.inventory.push(vec![
Cell::from("organisations.getInfo"),
Cell::from("Configured limit categories"),
Cell::Count(limits.as_object().map_or(0, <[(String, OValue)]>::len) as u64),
]);
s.records = 1;
s.pages = 1;
Ok(())
}
.await;
c.finish(s, result)?;
let mut s = open(
"guardrails.getCatalog",
"Current tenant capability catalog (not enabled guardrails)",
);
let result = async {
let answer = reader.catalog().await?;
let Some(evals) = record(Some(&answer))?
.get("evals")
.and_then(OValue::as_array)
else {
return Err(INVALID);
};
c.inventory.push(vec![
Cell::from("guardrails.getCatalog"),
Cell::from("Available evaluators"),
Cell::Count(evals.len() as u64),
]);
s.records = evals.len() as u64;
s.pages = 1;
Ok(())
}
.await;
c.finish(s, result)?;
let incomplete: Vec<(&'static str, String)> = c
.report
.sources
.iter()
.filter(|source| source.status != Status::Complete)
.map(|source| (source.name, source.notes.join(" ")))
.collect();
for (name, evidence) in incomplete {
c.finding_owned(
Priority::Review,
format!("Incomplete evidence: {name}"),
evidence,
"Resolve source access or validation problems and rerun. Do not interpret missing metrics as zero.",
name,
);
}
let table = |title: &str, note: &str, headers: &[&str], rows: Vec<Vec<Cell>>| Table {
title: title.to_owned(),
note: note.to_owned(),
headers: headers.iter().map(|header| (*header).to_owned()).collect(),
rows,
};
c.report.tables = vec![
table(
"Telemetry metrics",
"Server field names are preserved. cost: cents; latency and avgCacheLatency: ms; errorPercent: percent. Feedback is not a request count.",
&["Source", "Metric", "Value"],
std::mem::take(&mut c.metrics),
),
table(
"Error categories",
"Server-reported errors by HTTP status; 446 denotes an AIRS block.",
&["HTTP status", "Errors"],
std::mem::take(&mut c.statuses),
),
table(
"Error timeline",
"Server UTC buckets; counts by HTTP status. Not a count of unique violating sessions.",
&["UTC bucket", "HTTP status", "Errors"],
std::mem::take(&mut c.trends),
),
table(
"Current configuration and capabilities",
"Aggregate counts only. Active tenant workspaces are tenant-wide; keys and configs use the selected workspace.",
&["Source", "Metric", "Value"],
std::mem::take(&mut c.inventory),
),
table(
"Transaction coverage",
"Unique collected transactions by HTTP status; only complete pagination represents the entire window. Raw rows are excluded.",
&["Metric", "Value"],
std::mem::take(&mut c.coverage),
),
];
let report = &mut c.report;
let telemetry = report
.sources
.iter()
.any(|source| source.name.starts_with("telemetry.") && source.status == Status::Complete);
report.health = if !telemetry {
Health::Unknown
} else if report
.findings
.iter()
.any(|finding| finding.priority == Priority::Attention)
{
Health::Attention
} else if !report.findings.is_empty() {
Health::Review
} else {
Health::NoFindings
};
report.generated_at = iso((options.now)());
Ok(c.report)
}
#[cfg(test)]
#[path = "gateway_tests.rs"]
mod tests;