bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #5 from bsodfather/feat/observability-api

Browse Source
This commit is contained in:
a.tolmachev
2026-03-30 00:01:01 +03:00
parent 0a1680f24e be9ee95cbe
commit f33aa29c88
21 changed files with 1535 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -1068,6 +1068,7 @@ dependencies = [
"serde",
"serde_json",
"sqlx",
"time",
"tokio",
"tracing",
"tracing-subscriber",
TASKS.md
+6 -6
View File
@@ -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
apps/admin-api/src/app.rs
+87 -1
View File
@@ -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::<Value>()
.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::<Value>()
.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::<Value>()
.await
.unwrap();
let usage = client
.get(format!("{base_url}/usage?period=7d"))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let operation_usage = client
.get(format!(
"{base_url}/usage/operations/{operation_id}?period=7d"
))
.send()
.await
.unwrap()
.json::<Value>()
.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;
apps/admin-api/src/error.rs
+3
View File
@@ -108,6 +108,9 @@ impl From<RegistryError> 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,
apps/admin-api/src/routes.rs
+1
View File
@@ -1,6 +1,7 @@
pub mod access;
pub mod agents;
pub mod auth_profiles;
pub mod observability;
pub mod operations;
pub mod workspaces;
apps/admin-api/src/routes/observability.rs
+100
View File
@@ -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<WorkspacePath>,
Query(query): Query<LogsQuery>,
State(state): State<AppState>,
) -> Result<Json<Value>, 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<WorkspaceLogPath>,
State(state): State<AppState>,
) -> Result<Json<Value>, 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<WorkspacePath>,
Query(query): Query<UsageRequestQuery>,
State(state): State<AppState>,
) -> Result<Json<Value>, 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<WorkspaceOperationUsagePath>,
Query(query): Query<UsageRequestQuery>,
State(state): State<AppState>,
) -> Result<Json<Value>, 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<WorkspaceAgentUsagePath>,
Query(query): Query<UsageRequestQuery>,
State(state): State<AppState>,
) -> Result<Json<Value>, 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)))
}
apps/admin-api/src/service.rs
+347 -17
View File
@@ -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<InvocationLevel>,
pub search: Option<String>,
pub source: Option<InvocationSource>,
pub operation_id: Option<String>,
pub agent_id: Option<String>,
pub period: Option<UsagePeriod>,
pub limit: Option<u32>,
}
#[derive(Clone, Debug, Deserialize)]
pub struct UsageRequestQuery {
pub period: Option<UsagePeriod>,
pub source: Option<InvocationSource>,
}
#[derive(Clone, Debug, Serialize)]
pub struct UsageOverviewResponse {
pub summary: UsageSummary,
pub timeline: Vec<UsageTimelinePoint>,
pub operations: Vec<UsageOperationBreakdown>,
pub agents: Vec<UsageAgentBreakdown>,
}
#[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<Vec<InvocationLogRecord>, 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<InvocationLogRecord, ApiError> {
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<UsageOverviewResponse, ApiError> {
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<crank_registry::UsageRollupRecord, ApiError> {
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<crank_registry::UsageRollupRecord, ApiError> {
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<u16>,
error_kind: Option<String>,
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<String, ApiError> {
.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<GrpcServiceSummary, ApiError> {
let methods = service
.unary_methods()
apps/mcp-server/Cargo.toml
+2 -1
View File
@@ -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
apps/mcp-server/src/app.rs
+147 -48
View File
@@ -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<String>,
) -> 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<AppState>,
tool: &PublishedAgentTool,
status: InvocationStatus,
level: InvocationLevel,
message: &str,
status_code: Option<u16>,
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
}
apps/mcp-server/src/main.rs
+25 -2
View File
@@ -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(&registry, &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]
crates/crank-core/src/ids.rs
+1
View File
@@ -45,3 +45,4 @@ define_id!(UserId);
define_id!(AgentId);
define_id!(InvitationId);
define_id!(PlatformApiKeyId);
define_id!(InvocationLogId);
crates/crank-core/src/lib.rs
+6 -2
View File
@@ -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,
crates/crank-core/src/observability.rs
+81
View File
@@ -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<AgentId>,
pub operation_id: OperationId,
pub source: InvocationSource,
pub level: InvocationLevel,
pub status: InvocationStatus,
pub tool_name: String,
pub message: String,
pub request_id: Option<String>,
pub status_code: Option<u16>,
pub duration_ms: u64,
pub error_kind: Option<String>,
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<AgentId>,
pub operation_id: Option<OperationId>,
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,
}
crates/crank-registry/src/error.rs
+2
View File
@@ -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")]
crates/crank-registry/src/lib.rs
+8 -6
View File
@@ -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;
crates/crank-registry/src/migrations.rs
+59
View File
@@ -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(())
}
crates/crank-registry/src/model.rs
+89 -2
View File
@@ -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<String>,
pub agent_display_name: Option<String>,
}
#[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<InvocationLevel>,
pub search_text: Option<&'a str>,
pub source: Option<InvocationSource>,
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<InvocationSource>,
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,
crates/crank-registry/src/postgres.rs
+532 -8
View File
@@ -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<Vec<InvocationLogRecord>, 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<Option<InvocationLogRecord>, 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<UsageSummary, 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 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::<i64, _>("calls_total")?;
let calls_ok = row.try_get::<i64, _>("calls_ok")?;
let calls_error = row.try_get::<i64, _>("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::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("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<Vec<UsageTimelinePoint>, 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::<i64, _>("calls_ok")?, "calls_ok")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
})
})
.collect()
}
pub async fn list_usage_by_operation(
&self,
query: UsageQuery<'_>,
) -> Result<Vec<UsageOperationBreakdown>, 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<Option<UsageRollupRecord>, 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::<i64, _>("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::<i64, _>("calls_ok")?, "calls_ok")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("p99_ms")?, "p99_ms")?,
},
}))
}
pub async fn list_usage_by_agent(
&self,
query: UsageQuery<'_>,
) -> Result<Vec<UsageAgentBreakdown>, 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<Option<UsageRollupRecord>, 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::<i64, _>("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::<i64, _>("calls_ok")?, "calls_ok")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("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<PlatformApiKeyRecord, Regi
})
}
fn map_invocation_log_record(row: &PgRow) -> Result<InvocationLogRecord, RegistryError> {
Ok(InvocationLogRecord {
log: InvocationLog {
id: InvocationLogId::new(row.try_get::<String, _>("id")?),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
agent_id: row
.try_get::<Option<String>, _>("agent_id")?
.map(AgentId::new),
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
source: deserialize_enum_text(&row.try_get::<String, _>("source")?, "source")?,
level: deserialize_enum_text(&row.try_get::<String, _>("level")?, "level")?,
status: deserialize_enum_text(&row.try_get::<String, _>("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::<Option<i32>, _>("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::<i64, _>("duration_ms")?, "duration_ms")?,
error_kind: row.try_get("error_kind")?,
request_preview: row.try_get::<Json<Value>, _>("request_preview_json")?.0,
response_preview: row.try_get::<Json<Value>, _>("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<AgentSummary, RegistryError> {
Ok(AgentSummary {
id: AgentId::new(row.try_get::<String, _>("id")?),
@@ -1918,6 +2402,33 @@ fn map_yaml_import_job(row: &PgRow) -> Result<YamlImportJob, RegistryError> {
})
}
fn map_usage_operation_breakdown(row: &PgRow) -> Result<UsageOperationBreakdown, RegistryError> {
Ok(UsageOperationBreakdown {
operation_id: OperationId::new(row.try_get::<String, _>("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::<String, _>("protocol")?, "protocol")?,
calls_total: to_u64(row.try_get::<i64, _>("calls_total")?, "calls_total")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("p99_ms")?, "p99_ms")?,
})
}
fn map_usage_agent_breakdown(row: &PgRow) -> Result<UsageAgentBreakdown, RegistryError> {
Ok(UsageAgentBreakdown {
agent_id: AgentId::new(row.try_get::<String, _>("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::<i64, _>("calls_total")?, "calls_total")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("p99_ms")?, "p99_ms")?,
})
}
fn serialize_json_value<T: Serialize>(value: &T) -> Result<Value, RegistryError> {
Ok(serde_json::to_value(value)?)
}
@@ -1961,6 +2472,19 @@ fn from_db_version(value: i32, field: &'static str) -> Result<u32, RegistryError
})
}
fn to_u64(value: i64, field: &'static str) -> Result<u64, RegistryError> {
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::{
crates/crank-runtime/src/executor.rs
+16 -1
View File
@@ -35,11 +35,26 @@ impl RuntimeExecutor {
operation: &RuntimeOperation,
input: &Value,
) -> Result<Value, RuntimeError> {
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<PreparedRequest, RuntimeError> {
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<Value, RuntimeError> {
let adapter_response = match &operation.target {
Target::Grpc(target) => {
let request = GrpcRequest {
docs/admin-api.md
+13
View File
@@ -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` до начала реализации.
docs/database-schema.md
+9
View File
@@ -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
Переход от текущей схемы к целевой идет так: