From be9ee95cbeaa1e3e5f839084e8adab4ad417c1f8 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Mon, 30 Mar 2026 00:00:13 +0300 Subject: [PATCH] feat: add observability api foundation --- Cargo.lock | 1 + TASKS.md | 12 +- apps/admin-api/src/app.rs | 88 +++- apps/admin-api/src/error.rs | 3 + apps/admin-api/src/routes.rs | 1 + apps/admin-api/src/routes/observability.rs | 100 ++++ apps/admin-api/src/service.rs | 364 +++++++++++++- apps/mcp-server/Cargo.toml | 3 +- apps/mcp-server/src/app.rs | 195 ++++++-- apps/mcp-server/src/main.rs | 27 +- crates/crank-core/src/ids.rs | 1 + crates/crank-core/src/lib.rs | 8 +- crates/crank-core/src/observability.rs | 81 ++++ crates/crank-registry/src/error.rs | 2 + crates/crank-registry/src/lib.rs | 14 +- crates/crank-registry/src/migrations.rs | 59 +++ crates/crank-registry/src/model.rs | 91 +++- crates/crank-registry/src/postgres.rs | 540 ++++++++++++++++++++- crates/crank-runtime/src/executor.rs | 17 +- docs/admin-api.md | 13 + docs/database-schema.md | 9 + 21 files changed, 1535 insertions(+), 94 deletions(-) create mode 100644 apps/admin-api/src/routes/observability.rs create mode 100644 crates/crank-core/src/observability.rs diff --git a/Cargo.lock b/Cargo.lock index 297159f..1ecccfd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1068,6 +1068,7 @@ dependencies = [ "serde", "serde_json", "sqlx", + "time", "tokio", "tracing", "tracing-subscriber", diff --git a/TASKS.md b/TASKS.md index 642ff47..38b6cfb 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,19 +2,19 @@ ## Current -### `feat/platform-access` +### `feat/observability-api` Status: completed DoD: -- реализованы `platform-api-keys` как отдельная сущность уровня платформы -- `memberships` и `invitations` имеют workspace-scoped backend-контракт -- `platform-api-keys` не смешиваются с upstream `auth_profiles` -- `Settings` и `Workspace` экраны имеют достаточный backend foundation +- `Logs` page и `Usage` page имеют backend-контракт на реальных данных +- `test-runs` и `MCP tools/call` пишут продуктовые invocation logs +- usage endpoints отдают summary, timeline, breakdown по operation и agent +- observability не смешана с application logs ## Next -- `feat/observability-api` +- `feat/alpine-ui` ## Backlog diff --git a/apps/admin-api/src/app.rs b/apps/admin-api/src/app.rs index f72095e..ebf8a7f 100644 --- a/apps/admin-api/src/app.rs +++ b/apps/admin-api/src/app.rs @@ -14,6 +14,7 @@ use crate::{ save_agent_bindings, }, auth_profiles::{create_auth_profile, get_auth_profile, list_auth_profiles}, + observability::{get_agent_usage, get_log, get_operation_usage, get_usage, list_logs}, operations::{ create_operation, create_version, export_operation, generate_draft, get_operation, get_operation_version, list_grpc_services, list_operations, publish_operation, @@ -98,7 +99,12 @@ pub fn build_app(state: AppState) -> Router { .route( "/platform-api-keys/{key_id}", axum::routing::delete(delete_platform_api_key), - ); + ) + .route("/logs", get(list_logs)) + .route("/logs/{log_id}", get(get_log)) + .route("/usage", get(get_usage)) + .route("/usage/operations/{operation_id}", get(get_operation_usage)) + .route("/usage/agents/{agent_id}", get(get_agent_usage)); let admin_router = Router::new() .route("/workspaces", get(list_workspaces).post(create_workspace)) @@ -394,6 +400,86 @@ mod tests { assert_eq!(delete_key_status, reqwest::StatusCode::NO_CONTENT); } + #[tokio::test] + async fn exposes_logs_and_usage_from_real_test_runs() { + let registry = test_registry().await; + let storage_root = test_storage_root("observability"); + let upstream_base_url = spawn_upstream_server().await; + let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await; + let client = reqwest::Client::new(); + + let created = client + .post(format!("{base_url}/operations")) + .json(&test_operation_payload( + &upstream_base_url, + "crm_observability", + )) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let operation_id = created["operation_id"].as_str().unwrap().to_owned(); + + client + .post(format!("{base_url}/operations/{operation_id}/test-runs")) + .json(&json!({ + "version": 1, + "input": { "email": "user@example.com" } + })) + .send() + .await + .unwrap(); + + let logs = client + .get(format!("{base_url}/logs?period=7d")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let log_id = logs["items"][0]["log"]["id"].as_str().unwrap().to_owned(); + let log_detail = client + .get(format!("{base_url}/logs/{log_id}")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let usage = client + .get(format!("{base_url}/usage?period=7d")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let operation_usage = client + .get(format!( + "{base_url}/usage/operations/{operation_id}?period=7d" + )) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + + assert_eq!(logs["items"][0]["log"]["source"], "admin_test_run"); + assert_eq!(logs["items"][0]["operation_name"], "crm_observability"); + assert_eq!(log_detail["log"]["status"], "ok"); + assert_eq!(usage["summary"]["rollup"]["calls_total"], 1); + assert_eq!(usage["summary"]["rollup"]["calls_ok"], 1); + assert_eq!( + usage["operations"][0]["operation_name"], + "crm_observability" + ); + assert_eq!(operation_usage["rollup"]["calls_total"], 1); + } + #[tokio::test] async fn creates_publishes_and_tests_graphql_operation() { let registry = test_registry().await; diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index 92ca27f..b962ea4 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -108,6 +108,9 @@ impl From for ApiError { RegistryError::PlatformApiKeyNotFound { key_id } => { Self::not_found(format!("platform api key {key_id} was not found")) } + RegistryError::InvocationLogNotFound { log_id } => { + Self::not_found(format!("invocation log {log_id} was not found")) + } RegistryError::PublishedAgentNotFound { workspace_slug, agent_slug, diff --git a/apps/admin-api/src/routes.rs b/apps/admin-api/src/routes.rs index 2fceb42..428b67d 100644 --- a/apps/admin-api/src/routes.rs +++ b/apps/admin-api/src/routes.rs @@ -1,6 +1,7 @@ pub mod access; pub mod agents; pub mod auth_profiles; +pub mod observability; pub mod operations; pub mod workspaces; diff --git a/apps/admin-api/src/routes/observability.rs b/apps/admin-api/src/routes/observability.rs new file mode 100644 index 0000000..dfa6329 --- /dev/null +++ b/apps/admin-api/src/routes/observability.rs @@ -0,0 +1,100 @@ +use axum::{ + Json, + extract::{Path, Query, State}, +}; +use serde_json::{Value, json}; + +use crate::{ + error::ApiError, + routes::access::WorkspacePath, + service::{LogsQuery, UsageRequestQuery}, + state::AppState, +}; + +#[derive(serde::Deserialize)] +pub struct WorkspaceLogPath { + pub workspace_id: String, + pub log_id: String, +} + +#[derive(serde::Deserialize)] +pub struct WorkspaceOperationUsagePath { + pub workspace_id: String, + pub operation_id: String, +} + +#[derive(serde::Deserialize)] +pub struct WorkspaceAgentUsagePath { + pub workspace_id: String, + pub agent_id: String, +} + +pub async fn list_logs( + Path(path): Path, + Query(query): Query, + State(state): State, +) -> Result, ApiError> { + let items = state + .service + .list_logs(&path.workspace_id.as_str().into(), query) + .await?; + Ok(Json(json!({ "items": items }))) +} + +pub async fn get_log( + Path(path): Path, + State(state): State, +) -> Result, ApiError> { + let item = state + .service + .get_log( + &path.workspace_id.as_str().into(), + &path.log_id.as_str().into(), + ) + .await?; + Ok(Json(json!(item))) +} + +pub async fn get_usage( + Path(path): Path, + Query(query): Query, + State(state): State, +) -> Result, ApiError> { + let usage = state + .service + .get_usage_overview(&path.workspace_id.as_str().into(), query) + .await?; + Ok(Json(json!(usage))) +} + +pub async fn get_operation_usage( + Path(path): Path, + Query(query): Query, + State(state): State, +) -> Result, ApiError> { + let usage = state + .service + .get_operation_usage( + &path.workspace_id.as_str().into(), + &path.operation_id.as_str().into(), + query, + ) + .await?; + Ok(Json(json!(usage))) +} + +pub async fn get_agent_usage( + Path(path): Path, + Query(query): Query, + State(state): State, +) -> Result, ApiError> { + let usage = state + .service + .get_agent_usage( + &path.workspace_id.as_str().into(), + &path.agent_id.as_str().into(), + query, + ) + .await?; + Ok(Json(json!(usage))) +} diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 50db63c..ecdbe27 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -4,19 +4,22 @@ use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use crank_core::{ Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AuthConfig, AuthKind, AuthProfile, AuthProfileId, ConfigExport, ExportMode, GeneratedDraft, GeneratedDraftStatus, - InvitationId, InvitationStatus, InvitationToken, MembershipRole, OperationId, OperationStatus, - PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, Protocol, - SampleId, Samples, Target, Workspace, WorkspaceId, WorkspaceStatus, + InvitationId, InvitationStatus, InvitationToken, InvocationLevel, InvocationLog, + InvocationLogId, InvocationSource, InvocationStatus, MembershipRole, OperationId, + OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, + Protocol, SampleId, Samples, Target, UsagePeriod, Workspace, WorkspaceId, WorkspaceStatus, }; use crank_mapping::{JsonPathRoot, MappingSet, infer_mapping_from_samples}; use crank_proto::{ProtoService, services_from_descriptor_set_bytes}; use crank_registry::{ AgentSummary, AgentVersionRecord, CreateAgentRequest, CreateInvitationRequest, - CreatePlatformApiKeyRequest, CreateVersionRequest, CreateWorkspaceRequest, InvitationRecord, + CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest, + CreateWorkspaceRequest, InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, OperationSampleMetadata, OperationSummary, OperationVersionRecord, PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest, PublishRequest, RegistryOperation, SampleKind, SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, - SaveSampleMetadataRequest, UpdateWorkspaceRequest, WorkspaceRecord, + SaveSampleMetadataRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, + UsageOperationBreakdown, UsageQuery, UsageSummary, UsageTimelinePoint, WorkspaceRecord, }; use crank_runtime::{PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeOperation}; use crank_schema::Schema; @@ -163,6 +166,31 @@ pub struct CreatedPlatformApiKeyResponse { pub secret: String, } +#[derive(Clone, Debug, Deserialize)] +pub struct LogsQuery { + pub level: Option, + pub search: Option, + pub source: Option, + pub operation_id: Option, + pub agent_id: Option, + pub period: Option, + pub limit: Option, +} + +#[derive(Clone, Debug, Deserialize)] +pub struct UsageRequestQuery { + pub period: Option, + pub source: Option, +} + +#[derive(Clone, Debug, Serialize)] +pub struct UsageOverviewResponse { + pub summary: UsageSummary, + pub timeline: Vec, + pub operations: Vec, + pub agents: Vec, +} + #[derive(Clone, Debug, Deserialize)] pub struct GenerateDraftPayload { #[serde(default)] @@ -461,6 +489,144 @@ impl AdminService { Ok(()) } + #[instrument(skip(self))] + pub async fn list_logs( + &self, + workspace_id: &WorkspaceId, + query: LogsQuery, + ) -> Result, ApiError> { + self.ensure_workspace_exists(workspace_id).await?; + let operation_id = query.operation_id.as_deref().map(OperationId::new); + let agent_id = query.agent_id.as_deref().map(AgentId::new); + let (_, created_after, _) = usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?; + + self.registry + .list_invocation_logs(ListInvocationLogsQuery { + workspace_id, + level: query.level, + search_text: query.search.as_deref(), + source: query.source, + operation_id: operation_id.as_ref(), + agent_id: agent_id.as_ref(), + created_after: Some(&created_after), + limit: query.limit.unwrap_or(100), + }) + .await + .map_err(ApiError::from) + } + + #[instrument(skip(self))] + pub async fn get_log( + &self, + workspace_id: &WorkspaceId, + log_id: &InvocationLogId, + ) -> Result { + self.registry + .get_invocation_log(workspace_id, log_id) + .await? + .ok_or_else(|| { + ApiError::not_found(format!("invocation log {} was not found", log_id.as_str())) + }) + } + + #[instrument(skip(self))] + pub async fn get_usage_overview( + &self, + workspace_id: &WorkspaceId, + query: UsageRequestQuery, + ) -> Result { + self.ensure_workspace_exists(workspace_id).await?; + let (period, created_after, bucket) = + usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?; + let usage_query = UsageQuery { + workspace_id, + period, + source: query.source, + created_after: &created_after, + bucket, + }; + + let summary = self.registry.summarize_usage(usage_query.clone()).await?; + let timeline = self + .registry + .list_usage_timeline(usage_query.clone()) + .await?; + let operations = self + .registry + .list_usage_by_operation(usage_query.clone()) + .await?; + let agents = self.registry.list_usage_by_agent(usage_query).await?; + + Ok(UsageOverviewResponse { + summary, + timeline, + operations, + agents, + }) + } + + #[instrument(skip(self))] + pub async fn get_operation_usage( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + query: UsageRequestQuery, + ) -> Result { + self.get_operation(workspace_id, operation_id).await?; + let (period, created_after, bucket) = + usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?; + + self.registry + .get_usage_for_operation( + UsageQuery { + workspace_id, + period, + source: query.source, + created_after: &created_after, + bucket, + }, + operation_id, + ) + .await? + .ok_or_else(|| { + ApiError::not_found(format!( + "usage for operation {} was not found", + operation_id.as_str() + )) + }) + } + + #[instrument(skip(self))] + pub async fn get_agent_usage( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + query: UsageRequestQuery, + ) -> Result { + self.get_agent(workspace_id, agent_id).await?; + let (period, created_after, bucket) = + usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?; + + self.registry + .get_usage_for_agent( + UsageQuery { + workspace_id, + period, + source: query.source, + created_after: &created_after, + bucket, + }, + agent_id, + ) + .await? + .ok_or_else(|| { + ApiError::not_found(format!( + "usage for agent {} was not found", + agent_id.as_str() + )) + }) + } + pub async fn list_operations( &self, workspace_id: &WorkspaceId, @@ -658,6 +824,21 @@ impl AdminService { match build_request_preview(&record.snapshot.input_mapping, &payload.input) { Ok(preview) => preview, Err(error) => { + self.record_invocation( + workspace_id, + None, + &record.snapshot, + InvocationSource::AdminTestRun, + InvocationLevel::Error, + InvocationStatus::Error, + "mapping preview failed".to_owned(), + None, + Some("mapping".to_owned()), + 0, + Value::Null, + Value::Null, + ) + .await?; return Ok(TestRunResult { ok: false, request_preview: Value::Null, @@ -669,19 +850,58 @@ impl AdminService { } }; + let started_at = std::time::Instant::now(); match self.runtime.execute(&runtime, &payload.input).await { - Ok(output) => Ok(TestRunResult { - ok: true, - request_preview, - response_preview: output, - errors: Vec::new(), - }), - Err(error) => Ok(TestRunResult { - ok: false, - request_preview, - response_preview: Value::Null, - errors: vec![crate::error::runtime_test_failure(&error)], - }), + Ok(output) => { + let duration_ms = + u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + self.record_invocation( + workspace_id, + None, + &record.snapshot, + InvocationSource::AdminTestRun, + InvocationLevel::Info, + InvocationStatus::Ok, + "admin test run completed".to_owned(), + None, + None, + duration_ms, + request_preview.clone(), + output.clone(), + ) + .await?; + Ok(TestRunResult { + ok: true, + request_preview, + response_preview: output, + errors: Vec::new(), + }) + } + Err(error) => { + let duration_ms = + u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + self.record_invocation( + workspace_id, + None, + &record.snapshot, + InvocationSource::AdminTestRun, + InvocationLevel::Error, + InvocationStatus::Error, + error.to_string(), + None, + Some(runtime_error_code(&error).to_owned()), + duration_ms, + request_preview.clone(), + Value::Null, + ) + .await?; + Ok(TestRunResult { + ok: false, + request_preview, + response_preview: Value::Null, + errors: vec![crate::error::runtime_test_failure(&error)], + }) + } } } @@ -1324,6 +1544,47 @@ impl AdminService { async fn ensure_workspace_exists(&self, workspace_id: &WorkspaceId) -> Result<(), ApiError> { self.get_workspace(workspace_id).await.map(|_| ()) } + + async fn record_invocation( + &self, + workspace_id: &WorkspaceId, + agent_id: Option<&AgentId>, + operation: &RegistryOperation, + source: InvocationSource, + level: InvocationLevel, + status: InvocationStatus, + message: String, + status_code: Option, + error_kind: Option, + duration_ms: u64, + request_preview: Value, + response_preview: Value, + ) -> Result<(), ApiError> { + let log = InvocationLog { + id: InvocationLogId::new(new_prefixed_id("log")), + workspace_id: workspace_id.clone(), + agent_id: agent_id.cloned(), + operation_id: operation.id.clone(), + source, + level, + status, + tool_name: operation.name.clone(), + message, + request_id: None, + status_code, + duration_ms, + error_kind, + request_preview, + response_preview, + created_at: now_string()?, + }; + + self.registry + .create_invocation_log(CreateInvocationLogRequest { log: &log }) + .await?; + + Ok(()) + } } fn build_request_preview( @@ -1425,6 +1686,75 @@ fn default_invitation_expiry() -> Result { .map_err(|error| ApiError::internal(error.to_string())) } +fn runtime_error_code(error: &RuntimeError) -> &'static str { + match error { + RuntimeError::Schema(_) => "schema_error", + RuntimeError::Mapping(_) => "mapping_error", + RuntimeError::InvalidPreparedRequest { .. } => "invalid_request", + RuntimeError::GraphqlAdapter(_) => "graphql_error", + RuntimeError::GrpcAdapter(_) => "grpc_error", + RuntimeError::RestAdapter(_) => "rest_error", + RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", + } +} + +fn usage_window(period: UsagePeriod) -> Result<(UsagePeriod, String, UsageBucket), ApiError> { + let now = OffsetDateTime::now_utc(); + let (start, bucket) = match period { + UsagePeriod::Last30Minutes => ( + now.checked_sub(time::Duration::minutes(30)) + .ok_or_else(|| ApiError::internal("failed to compute usage period"))?, + UsageBucket::Hour, + ), + UsagePeriod::LastHour => ( + now.checked_sub(time::Duration::hours(1)) + .ok_or_else(|| ApiError::internal("failed to compute usage period"))?, + UsageBucket::Hour, + ), + UsagePeriod::Last6Hours => ( + now.checked_sub(time::Duration::hours(6)) + .ok_or_else(|| ApiError::internal("failed to compute usage period"))?, + UsageBucket::Hour, + ), + UsagePeriod::Last24Hours => ( + now.checked_sub(time::Duration::hours(24)) + .ok_or_else(|| ApiError::internal("failed to compute usage period"))?, + UsageBucket::Hour, + ), + UsagePeriod::Last7Days => ( + now.checked_sub(time::Duration::days(7)) + .ok_or_else(|| ApiError::internal("failed to compute usage period"))?, + UsageBucket::Day, + ), + UsagePeriod::Last30Days => ( + now.checked_sub(time::Duration::days(30)) + .ok_or_else(|| ApiError::internal("failed to compute usage period"))?, + UsageBucket::Week, + ), + UsagePeriod::Last90Days => ( + now.checked_sub(time::Duration::days(90)) + .ok_or_else(|| ApiError::internal("failed to compute usage period"))?, + UsageBucket::Month, + ), + UsagePeriod::ThisMonth => ( + OffsetDateTime::from_unix_timestamp( + now.unix_timestamp() - i64::from(now.day() - 1) * 24 * 60 * 60, + ) + .map_err(|error| ApiError::internal(error.to_string()))? + .replace_time(time::Time::MIDNIGHT), + UsageBucket::Week, + ), + }; + + Ok(( + period, + start + .format(&Rfc3339) + .map_err(|error| ApiError::internal(error.to_string()))?, + bucket, + )) +} + fn proto_service_summary(service: ProtoService) -> Result { let methods = service .unary_methods() diff --git a/apps/mcp-server/Cargo.toml b/apps/mcp-server/Cargo.toml index 847aa90..8d031f9 100644 --- a/apps/mcp-server/Cargo.toml +++ b/apps/mcp-server/Cargo.toml @@ -7,11 +7,13 @@ version.workspace = true [dependencies] axum.workspace = true +crank-core = { path = "../../crates/crank-core" } crank-registry = { path = "../../crates/crank-registry" } crank-runtime = { path = "../../crates/crank-runtime" } crank-schema = { path = "../../crates/crank-schema" } serde.workspace = true serde_json.workspace = true +time.workspace = true tokio = { workspace = true, features = ["sync"] } tracing.workspace = true tracing-subscriber.workspace = true @@ -19,7 +21,6 @@ uuid.workspace = true [dev-dependencies] crank-adapter-grpc = { path = "../../crates/crank-adapter-grpc", features = ["test-support"] } -crank-core = { path = "../../crates/crank-core" } crank-mapping = { path = "../../crates/crank-mapping" } crank-schema = { path = "../../crates/crank-schema" } reqwest.workspace = true diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs index 478f1c2..8fcd434 100644 --- a/apps/mcp-server/src/app.rs +++ b/apps/mcp-server/src/app.rs @@ -1,4 +1,7 @@ -use std::{sync::Arc, time::Duration}; +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; use axum::{ Json, Router, @@ -10,10 +13,14 @@ use axum::{ response::{IntoResponse, Response}, routing::get, }; -use crank_registry::{PostgresRegistry, PublishedAgentTool}; +use crank_core::{ + InvocationLevel, InvocationLog, InvocationLogId, InvocationSource, InvocationStatus, +}; +use crank_registry::{CreateInvocationLogRequest, PostgresRegistry, PublishedAgentTool}; use crank_runtime::{RuntimeError, RuntimeExecutor, RuntimeOperation}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use crate::{ catalog::PublishedToolCatalog, @@ -30,6 +37,7 @@ const HEADER_MCP_PROTOCOL_VERSION: &str = "MCP-Protocol-Version"; #[derive(Clone)] pub struct AppState { + registry: PostgresRegistry, catalog: PublishedToolCatalog, runtime: RuntimeExecutor, sessions: SessionStore, @@ -66,6 +74,7 @@ pub fn build_app( public_base_url: Option, ) -> Router { let state = Arc::new(AppState { + registry: registry.clone(), catalog: PublishedToolCatalog::new(registry, refresh_interval), runtime: RuntimeExecutor::new(), sessions: SessionStore::new(), @@ -223,52 +232,83 @@ async fn mcp_post( .await { Ok(Some(tool)) => { - match state - .runtime - .execute(&runtime_operation(tool), &arguments) - .await - { + let runtime_operation = runtime_operation(&tool); + let request_preview = + build_request_preview(&state.runtime, &runtime_operation, &arguments); + let started_at = Instant::now(); + + match state.runtime.execute(&runtime_operation, &arguments).await { Ok(output) => json_response( StatusCode::OK, - jsonrpc_result( - request_id(&message), - json!({ - "content": [ - { - "type": "text", - "text": serde_json::to_string_pretty(&output).unwrap_or_else(|_| "{}".to_owned()) - } - ], - "structuredContent": output, - "isError": false - }), - ), - None, - Some(&session.protocol_version), - ), - Err(error) => json_response( - StatusCode::OK, - jsonrpc_result( - request_id(&message), - json!({ - "content": [ - { - "type": "text", - "text": error.to_string() - } - ], - "structuredContent": { - "error": { - "code": runtime_error_code(&error), - "message": error.to_string() - } - }, - "isError": true - }), - ), + { + let _ = persist_invocation( + &state, + &tool, + InvocationStatus::Ok, + InvocationLevel::Info, + "agent tool call completed", + None, + None, + started_at.elapsed(), + request_preview, + output.clone(), + ) + .await; + jsonrpc_result( + request_id(&message), + json!({ + "content": [ + { + "type": "text", + "text": serde_json::to_string_pretty(&output).unwrap_or_else(|_| "{}".to_owned()) + } + ], + "structuredContent": output, + "isError": false + }), + ) + }, None, Some(&session.protocol_version), ), + Err(error) => { + let _ = persist_invocation( + &state, + &tool, + InvocationStatus::Error, + InvocationLevel::Error, + &error.to_string(), + None, + Some(runtime_error_code(&error)), + started_at.elapsed(), + request_preview, + Value::Null, + ) + .await; + json_response( + StatusCode::OK, + jsonrpc_result( + request_id(&message), + json!({ + "content": [ + { + "type": "text", + "text": error.to_string() + } + ], + "structuredContent": { + "error": { + "code": runtime_error_code(&error), + "message": error.to_string() + } + }, + "isError": true + }), + ), + None, + Some(&session.protocol_version), + ) + } } } Ok(None) => json_response( @@ -492,6 +532,65 @@ fn internal_jsonrpc_error(message: &Value, error: impl std::fmt::Display) -> Res ) } +fn build_request_preview( + runtime: &RuntimeExecutor, + operation: &RuntimeOperation, + arguments: &Value, +) -> Value { + match runtime.prepare_request(operation, arguments) { + Ok(prepared) => json!({ + "path": prepared.path_params, + "query": prepared.query_params, + "headers": prepared.headers, + "grpc": prepared.grpc.unwrap_or(Value::Null), + "variables": prepared.variables.unwrap_or(Value::Null), + "body": prepared.body.unwrap_or(Value::Null) + }), + Err(_) => Value::Null, + } +} + +async fn persist_invocation( + state: &Arc, + tool: &PublishedAgentTool, + status: InvocationStatus, + level: InvocationLevel, + message: &str, + status_code: Option, + error_kind: Option<&str>, + duration: Duration, + request_preview: Value, + response_preview: Value, +) -> Result<(), crank_registry::RegistryError> { + let created_at = OffsetDateTime::now_utc() + .format(&Rfc3339) + .unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_owned()); + let duration_ms = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX); + let log = InvocationLog { + id: InvocationLogId::new(format!("log_{}", uuid::Uuid::now_v7().simple())), + workspace_id: tool.workspace_id.clone(), + agent_id: Some(tool.agent_id.clone()), + operation_id: tool.operation.id.clone(), + source: InvocationSource::AgentToolCall, + level, + status, + tool_name: tool.tool_name.clone(), + message: message.to_owned(), + request_id: None, + status_code, + duration_ms, + error_kind: error_kind.map(ToOwned::to_owned), + request_preview, + response_preview, + created_at, + }; + + state + .registry + .create_invocation_log(CreateInvocationLogRequest { log: &log }) + .await +} + fn runtime_error_code(error: &RuntimeError) -> &'static str { match error { RuntimeError::Schema(_) => "schema_validation_error", @@ -540,11 +639,11 @@ fn tool_definition(tool: &PublishedAgentTool) -> Value { }) } -fn runtime_operation(tool: PublishedAgentTool) -> RuntimeOperation { - let mut operation = RuntimeOperation::from(tool.operation); - operation.tool_name = tool.tool_name; - operation.tool_description.title = tool.tool_title; - operation.tool_description.description = tool.tool_description; +fn runtime_operation(tool: &PublishedAgentTool) -> RuntimeOperation { + let mut operation = RuntimeOperation::from(tool.operation.clone()); + operation.tool_name = tool.tool_name.clone(); + operation.tool_description.title = tool.tool_title.clone(); + operation.tool_description.description = tool.tool_description.clone(); operation } diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs index 8ccaede..91db26e 100644 --- a/apps/mcp-server/src/main.rs +++ b/apps/mcp-server/src/main.rs @@ -57,7 +57,8 @@ mod tests { }; use crank_mapping::{MappingRule, MappingSet}; use crank_registry::{ - CreateAgentRequest, PostgresRegistry, PublishAgentRequest, PublishRequest, + CreateAgentRequest, ListInvocationLogsQuery, PostgresRegistry, PublishAgentRequest, + PublishRequest, }; use crank_schema::{Schema, SchemaKind}; use serde_json::{Value, json}; @@ -97,7 +98,7 @@ mod tests { publish_agent_for_operation(®istry, &operation, "sales-rest").await; let base_url = spawn_mcp_server(build_app( - registry, + registry.clone(), Duration::from_millis(0), Some("https://crank.example.com".to_owned()), )) @@ -142,6 +143,28 @@ mod tests { json!({ "id": "lead_123" }) ); assert_eq!(call_result["result"]["isError"], false); + + let logs = registry + .list_invocation_logs(ListInvocationLogsQuery { + workspace_id: &test_workspace_id(), + level: None, + search_text: None, + source: Some(crank_core::InvocationSource::AgentToolCall), + operation_id: Some(&operation.id), + agent_id: None, + created_after: None, + limit: 10, + }) + .await + .unwrap(); + + assert_eq!(logs.len(), 1); + assert_eq!( + logs[0].log.source, + crank_core::InvocationSource::AgentToolCall + ); + assert_eq!(logs[0].log.status, crank_core::InvocationStatus::Ok); + assert_eq!(logs[0].log.tool_name, "crm_create_lead"); } #[tokio::test] diff --git a/crates/crank-core/src/ids.rs b/crates/crank-core/src/ids.rs index 2f30e0a..ef7e652 100644 --- a/crates/crank-core/src/ids.rs +++ b/crates/crank-core/src/ids.rs @@ -45,3 +45,4 @@ define_id!(UserId); define_id!(AgentId); define_id!(InvitationId); define_id!(PlatformApiKeyId); +define_id!(InvocationLogId); diff --git a/crates/crank-core/src/lib.rs b/crates/crank-core/src/lib.rs index 2b32d14..f61f931 100644 --- a/crates/crank-core/src/lib.rs +++ b/crates/crank-core/src/lib.rs @@ -2,6 +2,7 @@ pub mod access; pub mod agent; pub mod auth; pub mod ids; +pub mod observability; pub mod operation; pub mod protocol; pub mod workspace; @@ -16,8 +17,11 @@ pub use auth::{ BearerAuthConfig, SecretRef, }; pub use ids::{ - AgentId, AuthProfileId, DescriptorId, InvitationId, OperationId, PlatformApiKeyId, SampleId, - ToolId, UserId, WorkspaceId, + AgentId, AuthProfileId, DescriptorId, InvitationId, InvocationLogId, OperationId, + PlatformApiKeyId, SampleId, ToolId, UserId, WorkspaceId, +}; +pub use observability::{ + InvocationLevel, InvocationLog, InvocationSource, InvocationStatus, UsagePeriod, UsageRollup, }; pub use operation::{ ConfigExport, ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, GraphqlTarget, diff --git a/crates/crank-core/src/observability.rs b/crates/crank-core/src/observability.rs new file mode 100644 index 0000000..02a67e4 --- /dev/null +++ b/crates/crank-core/src/observability.rs @@ -0,0 +1,81 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::{AgentId, OperationId, WorkspaceId}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum InvocationSource { + AdminTestRun, + AgentToolCall, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum InvocationLevel { + Debug, + Info, + Warn, + Error, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum InvocationStatus { + Ok, + Error, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum UsagePeriod { + #[serde(rename = "30m")] + Last30Minutes, + #[serde(rename = "1h")] + LastHour, + #[serde(rename = "6h")] + Last6Hours, + #[serde(rename = "24h")] + Last24Hours, + #[serde(rename = "7d")] + Last7Days, + #[serde(rename = "30d")] + Last30Days, + #[serde(rename = "90d")] + Last90Days, + #[serde(rename = "this_month")] + ThisMonth, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct InvocationLog { + pub id: crate::ids::InvocationLogId, + pub workspace_id: WorkspaceId, + pub agent_id: Option, + pub operation_id: OperationId, + pub source: InvocationSource, + pub level: InvocationLevel, + pub status: InvocationStatus, + pub tool_name: String, + pub message: String, + pub request_id: Option, + pub status_code: Option, + pub duration_ms: u64, + pub error_kind: Option, + pub request_preview: Value, + pub response_preview: Value, + pub created_at: String, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct UsageRollup { + pub workspace_id: WorkspaceId, + pub agent_id: Option, + pub operation_id: Option, + pub period: UsagePeriod, + pub calls_total: u64, + pub calls_ok: u64, + pub calls_error: u64, + pub p50_ms: u64, + pub p95_ms: u64, + pub p99_ms: u64, +} diff --git a/crates/crank-registry/src/error.rs b/crates/crank-registry/src/error.rs index 76ef992..cb5d1e6 100644 --- a/crates/crank-registry/src/error.rs +++ b/crates/crank-registry/src/error.rs @@ -14,6 +14,8 @@ pub enum RegistryError { InvitationNotFound { invitation_id: String }, #[error("platform api key {key_id} was not found")] PlatformApiKeyNotFound { key_id: String }, + #[error("invocation log {log_id} was not found")] + InvocationLogNotFound { log_id: String }, #[error("agent {agent_id} was not found")] AgentNotFound { agent_id: String }, #[error("published agent {workspace_slug}/{agent_slug} was not found")] diff --git a/crates/crank-registry/src/lib.rs b/crates/crank-registry/src/lib.rs index 4c952fc..d5e2cfa 100644 --- a/crates/crank-registry/src/lib.rs +++ b/crates/crank-registry/src/lib.rs @@ -6,12 +6,14 @@ mod postgres; pub use error::RegistryError; pub use model::{ AgentSummary, AgentVersionRecord, CreateAgentRequest, CreateInvitationRequest, - CreatePlatformApiKeyRequest, CreateVersionRequest, CreateWorkspaceRequest, - CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata, InvitationRecord, - MembershipRecord, OperationSampleMetadata, OperationSummary, OperationVersionRecord, - PlatformApiKeyRecord, PublishAgentRequest, PublishRequest, PublishedAgentTool, - RegistryOperation, SampleKind, SaveAgentBindingsRequest, SaveAuthProfileRequest, - SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, UpdateWorkspaceRequest, + CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest, + CreateWorkspaceRequest, CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata, + InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, + OperationSampleMetadata, OperationSummary, OperationVersionRecord, PlatformApiKeyRecord, + PublishAgentRequest, PublishRequest, PublishedAgentTool, RegistryOperation, SampleKind, + SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, + SaveSampleMetadataRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, + UsageOperationBreakdown, UsageQuery, UsageRollupRecord, UsageSummary, UsageTimelinePoint, WorkspaceRecord, YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, }; pub use postgres::PostgresRegistry; diff --git a/crates/crank-registry/src/migrations.rs b/crates/crank-registry/src/migrations.rs index 8299308..355beaf 100644 --- a/crates/crank-registry/src/migrations.rs +++ b/crates/crank-registry/src/migrations.rs @@ -385,5 +385,64 @@ pub async fn apply_postgres(pool: &PgPool) -> Result<(), sqlx::Error> { .execute(pool) .await?; + query( + "create table if not exists invocation_logs ( + id text primary key, + workspace_id text not null references workspaces(id) on delete cascade, + agent_id text null references agents(id) on delete set null, + operation_id text not null references operations(id) on delete cascade, + source text not null, + level text not null, + status text not null, + tool_name text not null, + message text not null, + request_id text null, + status_code integer null, + duration_ms bigint not null, + error_kind text null, + request_preview_json jsonb not null, + response_preview_json jsonb not null, + created_at timestamptz not null + )", + ) + .execute(pool) + .await?; + + query( + "create index if not exists invocation_logs_workspace_created_idx on invocation_logs(workspace_id, created_at desc)", + ) + .execute(pool) + .await?; + + query( + "create index if not exists invocation_logs_workspace_operation_created_idx on invocation_logs(workspace_id, operation_id, created_at desc)", + ) + .execute(pool) + .await?; + + query( + "create index if not exists invocation_logs_workspace_agent_created_idx on invocation_logs(workspace_id, agent_id, created_at desc)", + ) + .execute(pool) + .await?; + + query( + "create table if not exists usage_rollups ( + workspace_id text not null references workspaces(id) on delete cascade, + agent_id text null references agents(id) on delete cascade, + operation_id text null references operations(id) on delete cascade, + period text not null, + calls_total bigint not null, + calls_ok bigint not null, + calls_error bigint not null, + p50_ms bigint not null, + p95_ms bigint not null, + p99_ms bigint not null, + updated_at timestamptz not null + )", + ) + .execute(pool) + .await?; + Ok(()) } diff --git a/crates/crank-registry/src/model.rs b/crates/crank-registry/src/model.rs index fcde65a..c159ffd 100644 --- a/crates/crank-registry/src/model.rs +++ b/crates/crank-registry/src/model.rs @@ -1,7 +1,8 @@ use crank_core::{ Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AuthProfile, DescriptorId, - ExportMode, InvitationToken, MembershipRole, Operation, OperationId, OperationStatus, - PlatformApiKey, Protocol, SampleId, User, Workspace, WorkspaceId, + ExportMode, InvitationToken, InvocationLevel, InvocationLog, InvocationSource, MembershipRole, + Operation, OperationId, OperationStatus, PlatformApiKey, Protocol, SampleId, UsagePeriod, + UsageRollup, User, Workspace, WorkspaceId, }; use crank_mapping::MappingSet; use crank_schema::Schema; @@ -64,6 +65,52 @@ pub struct PlatformApiKeyRecord { pub api_key: PlatformApiKey, } +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct InvocationLogRecord { + pub log: InvocationLog, + pub operation_name: String, + pub operation_display_name: String, + pub agent_slug: Option, + pub agent_display_name: Option, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct UsageRollupRecord { + pub rollup: UsageRollup, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct UsageTimelinePoint { + pub bucket_start: String, + pub calls_ok: u64, + pub calls_error: u64, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct UsageOperationBreakdown { + pub operation_id: OperationId, + pub operation_name: String, + pub operation_display_name: String, + pub protocol: Protocol, + pub calls_total: u64, + pub calls_error: u64, + pub p50_ms: u64, + pub p95_ms: u64, + pub p99_ms: u64, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct UsageAgentBreakdown { + pub agent_id: AgentId, + pub agent_slug: String, + pub agent_display_name: String, + pub calls_total: u64, + pub calls_error: u64, + pub p50_ms: u64, + pub p95_ms: u64, + pub p99_ms: u64, +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct AgentSummary { pub id: AgentId, @@ -200,6 +247,46 @@ pub struct YamlImportJobCompletion { pub finished_at: String, } +#[derive(Clone, Debug, PartialEq)] +pub struct CreateInvocationLogRequest<'a> { + pub log: &'a InvocationLog, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ListInvocationLogsQuery<'a> { + pub workspace_id: &'a WorkspaceId, + pub level: Option, + pub search_text: Option<&'a str>, + pub source: Option, + pub operation_id: Option<&'a OperationId>, + pub agent_id: Option<&'a AgentId>, + pub created_after: Option<&'a str>, + pub limit: u32, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum UsageBucket { + Hour, + Day, + Week, + Month, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct UsageQuery<'a> { + pub workspace_id: &'a WorkspaceId, + pub period: UsagePeriod, + pub source: Option, + pub created_after: &'a str, + pub bucket: UsageBucket, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct UsageSummary { + pub rollup: UsageRollup, + pub success_rate: f64, +} + #[derive(Clone, Debug, PartialEq)] pub struct CreateVersionRequest<'a> { pub workspace_id: &'a WorkspaceId, diff --git a/crates/crank-registry/src/postgres.rs b/crates/crank-registry/src/postgres.rs index 62111dd..fe3820b 100644 --- a/crates/crank-registry/src/postgres.rs +++ b/crates/crank-registry/src/postgres.rs @@ -1,7 +1,7 @@ use crank_core::{ AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AuthProfile, InvitationId, - InvitationToken, OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, - PlatformApiKeyStatus, User, UserId, Workspace, WorkspaceId, + InvitationToken, InvocationLog, InvocationLogId, OperationId, OperationStatus, PlatformApiKey, + PlatformApiKeyId, PlatformApiKeyStatus, UsageRollup, User, UserId, Workspace, WorkspaceId, }; use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; @@ -16,12 +16,14 @@ use crate::{ migrations, model::{ AgentSummary, AgentVersionRecord, CreateAgentRequest, CreateInvitationRequest, - CreatePlatformApiKeyRequest, CreateVersionRequest, CreateWorkspaceRequest, - CreateYamlImportJobRequest, DescriptorMetadata, InvitationRecord, MembershipRecord, - OperationSampleMetadata, OperationSummary, OperationVersionRecord, PlatformApiKeyRecord, - PublishAgentRequest, PublishRequest, PublishedAgentTool, RegistryOperation, - SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, - SaveSampleMetadataRequest, UpdateWorkspaceRequest, WorkspaceRecord, YamlImportJob, + CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest, + CreateWorkspaceRequest, CreateYamlImportJobRequest, DescriptorMetadata, InvitationRecord, + InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, OperationSampleMetadata, + OperationSummary, OperationVersionRecord, PlatformApiKeyRecord, PublishAgentRequest, + PublishRequest, PublishedAgentTool, RegistryOperation, SaveAgentBindingsRequest, + SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, + UpdateWorkspaceRequest, UsageAgentBreakdown, UsageOperationBreakdown, UsageQuery, + UsageRollupRecord, UsageSummary, UsageTimelinePoint, WorkspaceRecord, YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, }, }; @@ -289,6 +291,449 @@ impl PostgresRegistry { Ok(()) } + pub async fn create_invocation_log( + &self, + request: CreateInvocationLogRequest<'_>, + ) -> Result<(), RegistryError> { + sqlx::query( + "insert into invocation_logs ( + id, + workspace_id, + agent_id, + operation_id, + source, + level, + status, + tool_name, + message, + request_id, + status_code, + duration_ms, + error_kind, + request_preview_json, + response_preview_json, + created_at + ) values ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::timestamptz + )", + ) + .bind(request.log.id.as_str()) + .bind(request.log.workspace_id.as_str()) + .bind(request.log.agent_id.as_ref().map(|value| value.as_str())) + .bind(request.log.operation_id.as_str()) + .bind(serialize_enum_text(&request.log.source, "source")?) + .bind(serialize_enum_text(&request.log.level, "level")?) + .bind(serialize_enum_text(&request.log.status, "status")?) + .bind(&request.log.tool_name) + .bind(&request.log.message) + .bind(&request.log.request_id) + .bind(request.log.status_code.map(i32::from)) + .bind(i64::try_from(request.log.duration_ms).map_err(|_| { + RegistryError::InvalidNumericValue { + field: "duration_ms", + value: i64::MAX, + } + })?) + .bind(&request.log.error_kind) + .bind(Json(request.log.request_preview.clone())) + .bind(Json(request.log.response_preview.clone())) + .bind(&request.log.created_at) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn list_invocation_logs( + &self, + query: ListInvocationLogsQuery<'_>, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + l.id, + l.workspace_id, + l.agent_id, + l.operation_id, + l.source, + l.level, + l.status, + l.tool_name, + l.message, + l.request_id, + l.status_code, + l.duration_ms, + l.error_kind, + l.request_preview_json, + l.response_preview_json, + to_char(l.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"') as created_at, + o.name as operation_name, + o.display_name as operation_display_name, + a.slug as agent_slug, + a.display_name as agent_display_name + from invocation_logs l + join operations o on o.id = l.operation_id + left join agents a on a.id = l.agent_id + where l.workspace_id = $1 + and ($2::text is null or l.level = $2) + and ($3::text is null or l.source = $3) + and ($4::text is null or l.operation_id = $4) + and ($5::text is null or l.agent_id = $5) + and ($6::timestamptz is null or l.created_at >= $6::timestamptz) + and ( + $7::text is null + or l.tool_name ilike '%' || $7 || '%' + or l.message ilike '%' || $7 || '%' + or o.name ilike '%' || $7 || '%' + or o.display_name ilike '%' || $7 || '%' + ) + order by l.created_at desc + limit $8", + ) + .bind(query.workspace_id.as_str()) + .bind( + query + .level + .as_ref() + .map(|value| serialize_enum_text(value, "level")) + .transpose()?, + ) + .bind( + query + .source + .as_ref() + .map(|value| serialize_enum_text(value, "source")) + .transpose()?, + ) + .bind(query.operation_id.map(|value| value.as_str())) + .bind(query.agent_id.map(|value| value.as_str())) + .bind(query.created_after) + .bind(query.search_text) + .bind(i64::from(query.limit)) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_invocation_log_record).collect() + } + + pub async fn get_invocation_log( + &self, + workspace_id: &WorkspaceId, + log_id: &InvocationLogId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + l.id, + l.workspace_id, + l.agent_id, + l.operation_id, + l.source, + l.level, + l.status, + l.tool_name, + l.message, + l.request_id, + l.status_code, + l.duration_ms, + l.error_kind, + l.request_preview_json, + l.response_preview_json, + to_char(l.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"') as created_at, + o.name as operation_name, + o.display_name as operation_display_name, + a.slug as agent_slug, + a.display_name as agent_display_name + from invocation_logs l + join operations o on o.id = l.operation_id + left join agents a on a.id = l.agent_id + where l.workspace_id = $1 and l.id = $2", + ) + .bind(workspace_id.as_str()) + .bind(log_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_invocation_log_record).transpose() + } + + pub async fn summarize_usage( + &self, + query: UsageQuery<'_>, + ) -> Result { + let row = sqlx::query( + "select + count(*)::bigint as calls_total, + count(*) filter (where status = 'ok')::bigint as calls_ok, + count(*) filter (where status = 'error')::bigint as calls_error, + coalesce(percentile_cont(0.5) within group (order by duration_ms), 0)::bigint as p50_ms, + coalesce(percentile_cont(0.95) within group (order by duration_ms), 0)::bigint as p95_ms, + coalesce(percentile_cont(0.99) within group (order by duration_ms), 0)::bigint as p99_ms + from invocation_logs + where workspace_id = $1 + and created_at >= $2::timestamptz + and ($3::text is null or source = $3)", + ) + .bind(query.workspace_id.as_str()) + .bind(query.created_after) + .bind( + query + .source + .as_ref() + .map(|value| serialize_enum_text(value, "source")) + .transpose()?, + ) + .fetch_one(&self.pool) + .await?; + + let calls_total = row.try_get::("calls_total")?; + let calls_ok = row.try_get::("calls_ok")?; + let calls_error = row.try_get::("calls_error")?; + let rollup = UsageRollup { + workspace_id: query.workspace_id.clone(), + agent_id: None, + operation_id: None, + period: query.period, + calls_total: to_u64(calls_total, "calls_total")?, + calls_ok: to_u64(calls_ok, "calls_ok")?, + calls_error: to_u64(calls_error, "calls_error")?, + p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, + p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, + p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, + }; + + let success_rate = if rollup.calls_total == 0 { + 0.0 + } else { + (rollup.calls_ok as f64 / rollup.calls_total as f64) * 100.0 + }; + + Ok(UsageSummary { + rollup, + success_rate, + }) + } + + pub async fn list_usage_timeline( + &self, + query: UsageQuery<'_>, + ) -> Result, RegistryError> { + let bucket = usage_bucket_sql(query.bucket); + let sql = format!( + "select + to_char(date_trunc('{bucket}', created_at at time zone 'UTC'), 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as bucket_start, + count(*) filter (where status = 'ok')::bigint as calls_ok, + count(*) filter (where status = 'error')::bigint as calls_error + from invocation_logs + where workspace_id = $1 + and created_at >= $2::timestamptz + and ($3::text is null or source = $3) + group by 1 + order by 1 asc" + ); + let rows = sqlx::query(&sql) + .bind(query.workspace_id.as_str()) + .bind(query.created_after) + .bind( + query + .source + .as_ref() + .map(|value| serialize_enum_text(value, "source")) + .transpose()?, + ) + .fetch_all(&self.pool) + .await?; + + rows.iter() + .map(|row| { + Ok(UsageTimelinePoint { + bucket_start: row.try_get("bucket_start")?, + calls_ok: to_u64(row.try_get::("calls_ok")?, "calls_ok")?, + calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, + }) + }) + .collect() + } + + pub async fn list_usage_by_operation( + &self, + query: UsageQuery<'_>, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + o.id as operation_id, + o.name as operation_name, + o.display_name as operation_display_name, + o.protocol, + count(*)::bigint as calls_total, + count(*) filter (where l.status = 'error')::bigint as calls_error, + coalesce(percentile_cont(0.5) within group (order by l.duration_ms), 0)::bigint as p50_ms, + coalesce(percentile_cont(0.95) within group (order by l.duration_ms), 0)::bigint as p95_ms, + coalesce(percentile_cont(0.99) within group (order by l.duration_ms), 0)::bigint as p99_ms + from invocation_logs l + join operations o on o.id = l.operation_id + where l.workspace_id = $1 + and l.created_at >= $2::timestamptz + and ($3::text is null or l.source = $3) + group by o.id, o.name, o.display_name, o.protocol + order by calls_total desc, o.name asc", + ) + .bind(query.workspace_id.as_str()) + .bind(query.created_after) + .bind( + query + .source + .as_ref() + .map(|value| serialize_enum_text(value, "source")) + .transpose()?, + ) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_usage_operation_breakdown).collect() + } + + pub async fn get_usage_for_operation( + &self, + query: UsageQuery<'_>, + operation_id: &OperationId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + count(*)::bigint as calls_total, + count(*) filter (where status = 'ok')::bigint as calls_ok, + count(*) filter (where status = 'error')::bigint as calls_error, + coalesce(percentile_cont(0.5) within group (order by duration_ms), 0)::bigint as p50_ms, + coalesce(percentile_cont(0.95) within group (order by duration_ms), 0)::bigint as p95_ms, + coalesce(percentile_cont(0.99) within group (order by duration_ms), 0)::bigint as p99_ms + from invocation_logs + where workspace_id = $1 + and operation_id = $2 + and created_at >= $3::timestamptz + and ($4::text is null or source = $4)", + ) + .bind(query.workspace_id.as_str()) + .bind(operation_id.as_str()) + .bind(query.created_after) + .bind( + query + .source + .as_ref() + .map(|value| serialize_enum_text(value, "source")) + .transpose()?, + ) + .fetch_one(&self.pool) + .await?; + + let calls_total = to_u64(row.try_get::("calls_total")?, "calls_total")?; + if calls_total == 0 { + return Ok(None); + } + + Ok(Some(UsageRollupRecord { + rollup: UsageRollup { + workspace_id: query.workspace_id.clone(), + agent_id: None, + operation_id: Some(operation_id.clone()), + period: query.period, + calls_total, + calls_ok: to_u64(row.try_get::("calls_ok")?, "calls_ok")?, + calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, + p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, + p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, + p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, + }, + })) + } + + pub async fn list_usage_by_agent( + &self, + query: UsageQuery<'_>, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + a.id as agent_id, + a.slug as agent_slug, + a.display_name as agent_display_name, + count(*)::bigint as calls_total, + count(*) filter (where l.status = 'error')::bigint as calls_error, + coalesce(percentile_cont(0.5) within group (order by l.duration_ms), 0)::bigint as p50_ms, + coalesce(percentile_cont(0.95) within group (order by l.duration_ms), 0)::bigint as p95_ms, + coalesce(percentile_cont(0.99) within group (order by l.duration_ms), 0)::bigint as p99_ms + from invocation_logs l + join agents a on a.id = l.agent_id + where l.workspace_id = $1 + and l.created_at >= $2::timestamptz + and ($3::text is null or l.source = $3) + group by a.id, a.slug, a.display_name + order by calls_total desc, a.slug asc", + ) + .bind(query.workspace_id.as_str()) + .bind(query.created_after) + .bind( + query + .source + .as_ref() + .map(|value| serialize_enum_text(value, "source")) + .transpose()?, + ) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_usage_agent_breakdown).collect() + } + + pub async fn get_usage_for_agent( + &self, + query: UsageQuery<'_>, + agent_id: &AgentId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + count(*)::bigint as calls_total, + count(*) filter (where status = 'ok')::bigint as calls_ok, + count(*) filter (where status = 'error')::bigint as calls_error, + coalesce(percentile_cont(0.5) within group (order by duration_ms), 0)::bigint as p50_ms, + coalesce(percentile_cont(0.95) within group (order by duration_ms), 0)::bigint as p95_ms, + coalesce(percentile_cont(0.99) within group (order by duration_ms), 0)::bigint as p99_ms + from invocation_logs + where workspace_id = $1 + and agent_id = $2 + and created_at >= $3::timestamptz + and ($4::text is null or source = $4)", + ) + .bind(query.workspace_id.as_str()) + .bind(agent_id.as_str()) + .bind(query.created_after) + .bind( + query + .source + .as_ref() + .map(|value| serialize_enum_text(value, "source")) + .transpose()?, + ) + .fetch_one(&self.pool) + .await?; + + let calls_total = to_u64(row.try_get::("calls_total")?, "calls_total")?; + if calls_total == 0 { + return Ok(None); + } + + Ok(Some(UsageRollupRecord { + rollup: UsageRollup { + workspace_id: query.workspace_id.clone(), + agent_id: Some(agent_id.clone()), + operation_id: None, + period: query.period, + calls_total, + calls_ok: to_u64(row.try_get::("calls_ok")?, "calls_ok")?, + calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, + p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, + p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, + p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, + }, + })) + } + pub async fn get_workspace( &self, workspace_id: &WorkspaceId, @@ -1689,6 +2134,45 @@ fn map_platform_api_key_record(row: &PgRow) -> Result Result { + Ok(InvocationLogRecord { + log: InvocationLog { + id: InvocationLogId::new(row.try_get::("id")?), + workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), + agent_id: row + .try_get::, _>("agent_id")? + .map(AgentId::new), + operation_id: OperationId::new(row.try_get::("operation_id")?), + source: deserialize_enum_text(&row.try_get::("source")?, "source")?, + level: deserialize_enum_text(&row.try_get::("level")?, "level")?, + status: deserialize_enum_text(&row.try_get::("status")?, "status")?, + tool_name: row.try_get("tool_name")?, + message: row.try_get("message")?, + request_id: row.try_get("request_id")?, + status_code: match row.try_get::, _>("status_code")? { + Some(value) => { + Some( + u16::try_from(value).map_err(|_| RegistryError::InvalidNumericValue { + field: "status_code", + value: i64::from(value), + })?, + ) + } + None => None, + }, + duration_ms: to_u64(row.try_get::("duration_ms")?, "duration_ms")?, + error_kind: row.try_get("error_kind")?, + request_preview: row.try_get::, _>("request_preview_json")?.0, + response_preview: row.try_get::, _>("response_preview_json")?.0, + created_at: row.try_get("created_at")?, + }, + operation_name: row.try_get("operation_name")?, + operation_display_name: row.try_get("operation_display_name")?, + agent_slug: row.try_get("agent_slug")?, + agent_display_name: row.try_get("agent_display_name")?, + }) +} + fn map_agent_summary(row: &PgRow) -> Result { Ok(AgentSummary { id: AgentId::new(row.try_get::("id")?), @@ -1918,6 +2402,33 @@ fn map_yaml_import_job(row: &PgRow) -> Result { }) } +fn map_usage_operation_breakdown(row: &PgRow) -> Result { + Ok(UsageOperationBreakdown { + operation_id: OperationId::new(row.try_get::("operation_id")?), + operation_name: row.try_get("operation_name")?, + operation_display_name: row.try_get("operation_display_name")?, + protocol: deserialize_enum_text(&row.try_get::("protocol")?, "protocol")?, + calls_total: to_u64(row.try_get::("calls_total")?, "calls_total")?, + calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, + p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, + p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, + p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, + }) +} + +fn map_usage_agent_breakdown(row: &PgRow) -> Result { + Ok(UsageAgentBreakdown { + agent_id: AgentId::new(row.try_get::("agent_id")?), + agent_slug: row.try_get("agent_slug")?, + agent_display_name: row.try_get("agent_display_name")?, + calls_total: to_u64(row.try_get::("calls_total")?, "calls_total")?, + calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, + p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, + p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, + p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, + }) +} + fn serialize_json_value(value: &T) -> Result { Ok(serde_json::to_value(value)?) } @@ -1961,6 +2472,19 @@ fn from_db_version(value: i32, field: &'static str) -> Result Result { + u64::try_from(value).map_err(|_| RegistryError::InvalidNumericValue { field, value }) +} + +fn usage_bucket_sql(bucket: crate::model::UsageBucket) -> &'static str { + match bucket { + crate::model::UsageBucket::Hour => "hour", + crate::model::UsageBucket::Day => "day", + crate::model::UsageBucket::Week => "week", + crate::model::UsageBucket::Month => "month", + } +} + #[cfg(test)] mod tests { use std::{ diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index 3871e10..34642c1 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -35,11 +35,26 @@ impl RuntimeExecutor { operation: &RuntimeOperation, input: &Value, ) -> Result { + let prepared_request = self.prepare_request(operation, input)?; + self.execute_prepared(operation, prepared_request).await + } + + pub fn prepare_request( + &self, + operation: &RuntimeOperation, + input: &Value, + ) -> Result { operation.input_schema.validate_shape(input)?; let mapped_input = operation.input_mapping.apply(&json!({ "mcp": input }))?; - let prepared_request = PreparedRequest::from_mapping_output(&mapped_input)?; + PreparedRequest::from_mapping_output(&mapped_input) + } + async fn execute_prepared( + &self, + operation: &RuntimeOperation, + prepared_request: PreparedRequest, + ) -> Result { let adapter_response = match &operation.target { Target::Grpc(target) => { let request = GrpcRequest { diff --git a/docs/admin-api.md b/docs/admin-api.md index d594197..27c6284 100644 --- a/docs/admin-api.md +++ b/docs/admin-api.md @@ -126,6 +126,14 @@ - `GET /api/admin/workspaces/{workspace_id}/usage/operations/{operation_id}` - `GET /api/admin/workspaces/{workspace_id}/usage/agents/{agent_id}` +Контракт: + +- `GET /logs` поддерживает `level`, `search`, `source`, `operation_id`, `agent_id`, `period`, `limit`; +- `period` использует UI-friendly значения `30m`, `1h`, `6h`, `24h`, `7d`, `30d`, `90d`, `this_month`; +- `source` различает `admin_test_run` и `agent_tool_call`; +- `GET /usage` возвращает `summary`, `timeline`, `operations`, `agents` одним ответом; +- detail endpoints по operation и agent возвращают rollup для выбранного периода. + ## 6. Page-to-endpoint mapping ### Operations catalog @@ -185,6 +193,11 @@ - breakdown по agent; - CSV export. +Текущая реализация: + +- summary и breakdown считаются по `invocation_logs`; +- materialized `usage_rollups` остаются совместимым storage-слоем для дальнейшей оптимизации, но не являются единственным source of truth в MVP. + ## 7. Принцип совместимости Если UI расходится с текущим backend, приоритет отдается целевой продуктовой модели, но конфликт должен быть явно разобран в `docs/as-is-to-be.md` до начала реализации. diff --git a/docs/database-schema.md b/docs/database-schema.md index b425bb0..20d5e51 100644 --- a/docs/database-schema.md +++ b/docs/database-schema.md @@ -254,9 +254,13 @@ - `workspace_id` - `agent_id` - `operation_id` +- `source` - `request_id` - `level` - `status` +- `tool_name` +- `message` +- `status_code` - `duration_ms` - `error_kind` - `request_preview_json` @@ -277,6 +281,11 @@ - `p95_ms` - `p99_ms` +Замечание: + +- в MVP usage read-model может вычисляться напрямую из `invocation_logs`; +- `usage_rollups` сохраняется как совместимая таблица под materialized aggregates и дальнейшую оптимизацию. + ## 10. Migration strategy Переход от текущей схемы к целевой идет так: