test: production ingest E2E test suite with enhanced logging #55

Open
poimen wants to merge 34 commits from feat/production-ingest-test-logging into main
3 changed files with 197 additions and 35 deletions
Showing only changes of commit 40371fd99c - Show all commits
+6 -5
View File
@@ -6,6 +6,7 @@
use crate::queue_adapter::{QueueAdapter, QueueMessage, QueueStats};
use anyhow::{anyhow, Result};
use async_trait::async_trait;
use base64::Engine;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use std::sync::Arc;
@@ -234,7 +235,7 @@ impl QueueAdapter for GatewayQueueAdapter {
let token = self.token_source.token().await?;
// Base64 encode body
let encoded_body = base64::encode(body.as_bytes());
let encoded_body = base64::engine::general_purpose::STANDARD.encode(body.as_bytes());
// Build request
let mut attrs = attributes;
@@ -311,7 +312,7 @@ impl QueueAdapter for GatewayQueueAdapter {
if let Some(sqs_msgs) = sqs_resp.messages {
for msg in sqs_msgs {
// Decode body from base64
let body_bytes = base64::decode(msg.body.as_bytes())?;
let body_bytes = base64::engine::general_purpose::STANDARD.decode(msg.body.as_bytes())?;
let body = String::from_utf8(body_bytes)?;
let chunk_id = msg
@@ -404,7 +405,7 @@ impl QueueAdapter for GatewayQueueAdapter {
})
.to_string();
let encoded_body = base64::encode(dlq_body.as_bytes());
let encoded_body = base64::engine::general_purpose::STANDARD.encode(dlq_body.as_bytes());
let req = SendMessageRequest {
message_body: encoded_body,
@@ -511,8 +512,8 @@ mod tests {
#[test]
fn test_base64_roundtrip() {
let original = "hello world";
let encoded = base64::encode(original.as_bytes());
let decoded = String::from_utf8(base64::decode(encoded.as_bytes()).unwrap()).unwrap();
let encoded = base64::engine::general_purpose::STANDARD.encode(original.as_bytes());
let decoded = String::from_utf8(base64::engine::general_purpose::STANDARD.decode(encoded.as_bytes()).unwrap()).unwrap();
assert_eq!(decoded, original);
}
+177 -30
View File
@@ -12,6 +12,7 @@ use crate::agent::{Agent, AgentConfig, AgentCapability, DefaultAgent};
use crate::agent::client_sdk::SynthesisClient;
use crate::handlers::response_builder;
use mem_store::agent_repo::{AgentRepository, AgentPrompt, AgentSkill, AgentDecision, RolePromptMapping};
use crate::metrics::{ERROR_AUTH_FAILURE_AGENT, ERROR_BAD_REQUEST_AGENT, ERROR_NOT_FOUND_AGENT, ERROR_UNEXPECTED_AGENT, ERROR_UNEXPECTED_TOTAL};
use tracing::{debug, info, error, warn};
/// Register agent request
@@ -51,10 +52,14 @@ pub async fn register_agent_handler(
}
if body.agent_id.is_empty() || body.project_id.is_empty() {
ERROR_BAD_REQUEST_AGENT.inc();
warn!(agent_id = %body.agent_id, "Expected error: missing agent_id or project_id");
return response_builder::bad_request("agent_id and project_id required");
}
if body.capabilities.is_empty() {
ERROR_BAD_REQUEST_AGENT.inc();
warn!(agent_id = %body.agent_id, "Expected error: no capabilities provided");
return response_builder::bad_request("At least one capability required");
}
@@ -74,6 +79,8 @@ pub async fn register_agent_handler(
.collect();
if caps.is_empty() {
ERROR_BAD_REQUEST_AGENT.inc();
warn!(agent_id = %body.agent_id, "Expected error: invalid capability names");
return response_builder::bad_request("Invalid capabilities");
}
@@ -96,11 +103,15 @@ pub async fn register_agent_handler(
.await;
if let Err(e) = project_exists {
error!("Failed to verify project: {}", e);
ERROR_UNEXPECTED_AGENT.inc();
ERROR_UNEXPECTED_TOTAL.inc();
error!(agent_id = %body.agent_id, error = %e, "Unexpected error: DB failure verifying project");
return response_builder::internal_error("Database error during project verification");
}
if project_exists.unwrap().is_none() {
ERROR_NOT_FOUND_AGENT.inc();
info!(agent_id = %body.agent_id, project_id = %body.project_id, "Expected error: project not found");
return response_builder::bad_request(&format!("Project not found: {}", body.project_id));
}
@@ -126,7 +137,9 @@ pub async fn register_agent_handler(
.await;
if let Err(e) = agent_insert {
error!("Failed to insert agent registry: {}", e);
ERROR_UNEXPECTED_AGENT.inc();
ERROR_UNEXPECTED_TOTAL.inc();
error!(agent_id = %body.agent_id, error = %e, "Unexpected error: DB failure inserting agent");
return response_builder::internal_error("Failed to register agent");
}
@@ -203,7 +216,46 @@ pub async fn register_agent_handler(
})
}
/// GET /agents/{id} - Get agent status
/// Full agent progress response
#[derive(Debug, Serialize)]
pub struct AgentProgressResponse {
pub agent_id: String,
pub project_id: String,
pub capabilities: Vec<String>,
pub status: String,
pub prompts: Vec<PromptResponse>,
pub skills: Vec<SkillSummary>,
pub decisions: Vec<DecisionSummary>,
pub metrics: Option<MetricsSummary>,
pub created_at: String,
pub updated_at: String,
}
#[derive(Debug, Serialize)]
pub struct SkillSummary {
pub name: String,
pub success_rate: f32,
pub invocation_count: i64,
pub enabled: bool,
}
#[derive(Debug, Serialize)]
pub struct DecisionSummary {
pub action: String,
pub confidence: f32,
pub outcome_success: Option<bool>,
pub created_at: String,
}
#[derive(Debug, Serialize)]
pub struct MetricsSummary {
pub requests_total: i64,
pub requests_success: i64,
pub error_rate: f32,
pub average_latency_ms: f32,
}
/// GET /agents/{id} - Get agent progress
pub async fn get_agent_handler(
req: HttpRequest,
path: web::Path<String>,
@@ -217,33 +269,110 @@ pub async fn get_agent_handler(
return response;
}
debug!("Getting agent: {}", agent_id);
debug!("Getting agent progress: {}", agent_id);
// Extract JWT for agent operations
let jwt = crate::handlers::extract_jwt_token(&req)
.unwrap_or_else(|| {
warn!("No JWT token in get_agent request");
"invalid".to_string()
});
// Fetch agent registry
let agent_row = sqlx::query_as::<_, (String, Vec<String>, Option<String>, i32, String, String, String)>(
r#"SELECT project_id, capabilities, webhook_url, rate_limit, status,
created_at::text, updated_at::text
FROM agent_registry WHERE agent_id = $1"#
)
.bind(&agent_id)
.fetch_optional(&state.pool)
.await;
// Stub: would fetch from DB
let config = AgentConfig {
agent_id: agent_id.clone(),
project_id: "poimen".to_string(),
capabilities: vec![AgentCapability::Summarization],
webhook_url: None,
rate_limit: 1000,
metadata: std::collections::HashMap::new(),
let (project_id, capabilities, _webhook, _rate_limit, status, created_at, updated_at) = match agent_row {
Ok(Some(row)) => row,
Ok(None) => {
ERROR_NOT_FOUND_AGENT.inc();
info!(agent_id = %agent_id, "Expected error: agent not found");
return response_builder::not_found(&format!("Agent not found: {}", agent_id));
}
Err(e) => {
ERROR_UNEXPECTED_AGENT.inc();
ERROR_UNEXPECTED_TOTAL.inc();
error!(agent_id = %agent_id, error = %e, "Unexpected error: DB failure fetching agent");
return response_builder::internal_error("Database error");
}
};
let agent = DefaultAgent::new(config);
// Fetch prompts
let prompts: Vec<PromptResponse> = sqlx::query_as::<_, (String, String, String, Option<String>, String, Vec<String>, i64, f32, i32, String)>(
r#"SELECT id::text, name, template, target_model, task_category,
tags, usage_count, avg_quality, version, created_at::text
FROM agent_prompt WHERE project_id = $1 ORDER BY created_at DESC"#
)
.bind(&project_id)
.fetch_all(&state.pool)
.await
.unwrap_or_default()
.into_iter()
.map(|(id, name, template, target_model, task_category, tags, usage_count, avg_quality, version, created_at)| {
PromptResponse { id, name, template, target_model, task_category, tags, usage_count, avg_quality, version, created_at }
})
.collect();
match futures::executor::block_on(agent.status()) {
status => {
info!("Agent status: {} with JWT auth", agent_id);
response_builder::success_response(status)
}
}
// Fetch skills
let skills: Vec<SkillSummary> = sqlx::query_as::<_, (String, f32, i64, bool)>(
r#"SELECT name, success_rate, invocation_count, enabled
FROM agent_skill WHERE agent_id = $1 ORDER BY created_at DESC"#
)
.bind(&agent_id)
.fetch_all(&state.pool)
.await
.unwrap_or_default()
.into_iter()
.map(|(name, success_rate, invocation_count, enabled)| {
SkillSummary { name, success_rate, invocation_count, enabled }
})
.collect();
// Fetch recent decisions
let decisions: Vec<DecisionSummary> = sqlx::query_as::<_, (String, f32, Option<bool>, String)>(
r#"SELECT action, confidence, outcome_success, created_at::text
FROM agent_decision WHERE agent_id = $1
ORDER BY created_at DESC LIMIT 20"#
)
.bind(&agent_id)
.fetch_all(&state.pool)
.await
.unwrap_or_default()
.into_iter()
.map(|(action, confidence, outcome_success, created_at)| {
DecisionSummary { action, confidence, outcome_success, created_at }
})
.collect();
// Fetch latest metrics
let metrics = sqlx::query_as::<_, (i64, i64, f32, f32)>(
r#"SELECT requests_total, requests_success, error_rate, average_latency_ms
FROM agent_metrics WHERE agent_id = $1
ORDER BY recorded_at DESC LIMIT 1"#
)
.bind(&agent_id)
.fetch_optional(&state.pool)
.await
.ok()
.flatten()
.map(|(requests_total, requests_success, error_rate, average_latency_ms)| {
MetricsSummary { requests_total, requests_success, error_rate, average_latency_ms }
});
info!("Agent progress: {} ({} prompts, {} skills, {} decisions)",
agent_id, prompts.len(), skills.len(), decisions.len());
response_builder::success_response(AgentProgressResponse {
agent_id,
project_id,
capabilities,
status,
prompts,
skills,
decisions,
metrics,
created_at,
updated_at,
})
}
/// Metrics response
@@ -406,6 +535,8 @@ pub async fn create_prompt_handler(
}
if body.name.is_empty() || body.template.is_empty() {
ERROR_BAD_REQUEST_AGENT.inc();
warn!("Expected error: missing prompt name or template");
return response_builder::bad_request("name and template required");
}
@@ -449,7 +580,9 @@ pub async fn create_prompt_handler(
})
}
Err(e) => {
error!("Failed to create prompt: {}", e);
ERROR_UNEXPECTED_AGENT.inc();
ERROR_UNEXPECTED_TOTAL.inc();
error!(error = %e, "Unexpected error: DB failure creating prompt");
response_builder::internal_error("Failed to create prompt")
}
}
@@ -478,6 +611,8 @@ pub async fn map_role_to_prompt_handler(
}
if body.role_name.is_empty() || body.prompt_id.is_empty() {
ERROR_BAD_REQUEST_AGENT.inc();
warn!("Expected error: missing role_name or prompt_id");
return response_builder::bad_request("role_name and prompt_id required");
}
@@ -485,7 +620,11 @@ pub async fn map_role_to_prompt_handler(
let prompt_uuid = match Uuid::parse_str(&body.prompt_id) {
Ok(id) => id,
Err(_) => return response_builder::bad_request("Invalid prompt_id UUID format"),
Err(_) => {
ERROR_BAD_REQUEST_AGENT.inc();
warn!(prompt_id = %body.prompt_id, "Expected error: invalid UUID format");
return response_builder::bad_request("Invalid prompt_id UUID format");
}
};
let priority = body.priority.unwrap_or(0);
@@ -527,16 +666,22 @@ pub async fn map_role_to_prompt_handler(
}))
}
Err(e) => {
error!("Failed to create role mapping: {}", e);
ERROR_UNEXPECTED_AGENT.inc();
ERROR_UNEXPECTED_TOTAL.inc();
error!(error = %e, "Unexpected error: DB failure creating role mapping");
response_builder::internal_error("Failed to map role to prompt")
}
}
}
Ok(None) => {
ERROR_NOT_FOUND_AGENT.inc();
info!(prompt_id = %body.prompt_id, "Expected error: prompt not found");
response_builder::not_found(&format!("Prompt not found: {}", body.prompt_id))
}
Err(e) => {
error!("Database error checking prompt: {}", e);
ERROR_UNEXPECTED_AGENT.inc();
ERROR_UNEXPECTED_TOTAL.inc();
error!(error = %e, "Unexpected error: DB failure checking prompt");
response_builder::internal_error("Database error")
}
}
@@ -603,7 +748,9 @@ pub async fn get_role_prompts_handler(
})
}
Err(e) => {
error!("Failed to fetch role prompts: {}", e);
ERROR_UNEXPECTED_AGENT.inc();
ERROR_UNEXPECTED_TOTAL.inc();
error!(role_name = %role_name, error = %e, "Unexpected error: DB failure fetching role prompts");
response_builder::internal_error("Failed to fetch role prompts")
}
}
+14
View File
@@ -381,6 +381,16 @@ pub static ERROR_UNEXPECTED_QUERY: Counter = Counter::new(
pub static ERROR_UNEXPECTED_CONTEXT: Counter = Counter::new(
"memory_error_unexpected_context_total", "Unexpected errors during context");
// Agent endpoint error counters
pub static ERROR_AUTH_FAILURE_AGENT: Counter = Counter::new(
"memory_error_auth_failure_agent_total", "Auth failures on agent endpoints");
pub static ERROR_BAD_REQUEST_AGENT: Counter = Counter::new(
"memory_error_bad_request_agent_total", "Bad request errors on agent endpoints (expected)");
pub static ERROR_NOT_FOUND_AGENT: Counter = Counter::new(
"memory_error_not_found_agent_total", "Not found errors on agent endpoints (expected)");
pub static ERROR_UNEXPECTED_AGENT: Counter = Counter::new(
"memory_error_unexpected_agent_total", "Unexpected errors on agent endpoints (DB failures, 500s)");
// Last error info (most recent error for debugging)
pub static LAST_ERROR_TIMESTAMP: Gauge = Gauge::new(
"memory_last_error_timestamp_seconds", "Unix timestamp of most recent error");
@@ -593,6 +603,10 @@ pub fn render_metrics() -> String {
counter!(ERROR_UNEXPECTED_INGEST);
counter!(ERROR_UNEXPECTED_QUERY);
counter!(ERROR_UNEXPECTED_CONTEXT);
counter!(ERROR_AUTH_FAILURE_AGENT);
counter!(ERROR_BAD_REQUEST_AGENT);
counter!(ERROR_NOT_FOUND_AGENT);
counter!(ERROR_UNEXPECTED_AGENT);
gauge!(LAST_ERROR_TIMESTAMP);
out