From 624224f08998c44fc3ff7505e3821056771f57f4 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Sun, 12 Apr 2026 21:57:09 +0000 Subject: [PATCH] registry: finish sqlx verification for operation reads --- ...8b68c15efa72029740e28bb881f54f73a6b9b.json | 23 ++ ...ff21a2f73a4416d0cf20be418bba36def9eb4.json | 40 ++++ TASKS.md | 10 +- crates/crank-registry/src/postgres/mod.rs | 206 ++++++++++++++++-- .../crank-registry/src/postgres/operation.rs | 23 +- 5 files changed, 267 insertions(+), 35 deletions(-) create mode 100644 .sqlx/query-6eed48468f97843bbcd6a2f90448b68c15efa72029740e28bb881f54f73a6b9b.json create mode 100644 .sqlx/query-fdd05b145021a19dda9a9f44c31ff21a2f73a4416d0cf20be418bba36def9eb4.json diff --git a/.sqlx/query-6eed48468f97843bbcd6a2f90448b68c15efa72029740e28bb881f54f73a6b9b.json b/.sqlx/query-6eed48468f97843bbcd6a2f90448b68c15efa72029740e28bb881f54f73a6b9b.json new file mode 100644 index 0000000..c66b1b0 --- /dev/null +++ b/.sqlx/query-6eed48468f97843bbcd6a2f90448b68c15efa72029740e28bb881f54f73a6b9b.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "select 1 as \"present!\"\n from agents a\n join published_agents pa on pa.agent_id = a.id\n join agent_operation_bindings b\n on b.agent_id = a.id and b.agent_version = pa.version\n where a.workspace_id = $1\n and b.operation_id = $2\n limit 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "present!", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "6eed48468f97843bbcd6a2f90448b68c15efa72029740e28bb881f54f73a6b9b" +} diff --git a/.sqlx/query-fdd05b145021a19dda9a9f44c31ff21a2f73a4416d0cf20be418bba36def9eb4.json b/.sqlx/query-fdd05b145021a19dda9a9f44c31ff21a2f73a4416d0cf20be418bba36def9eb4.json new file mode 100644 index 0000000..8896b79 --- /dev/null +++ b/.sqlx/query-fdd05b145021a19dda9a9f44c31ff21a2f73a4416d0cf20be418bba36def9eb4.json @@ -0,0 +1,40 @@ +{ + "db_name": "PostgreSQL", + "query": "select\n b.operation_id,\n a.id as agent_id,\n a.slug as agent_slug,\n a.display_name\n from agents a\n join published_agents pa on pa.agent_id = a.id\n join agent_operation_bindings b\n on b.agent_id = a.id and b.agent_version = pa.version\n where a.workspace_id = $1\n order by a.display_name asc, b.tool_name asc", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "operation_id", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "agent_id", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "agent_slug", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "display_name", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false + ] + }, + "hash": "fdd05b145021a19dda9a9f44c31ff21a2f73a4416d0cf20be418bba36def9eb4" +} diff --git a/TASKS.md b/TASKS.md index 6bb1fb4..6420b69 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,18 +2,18 @@ ## Current -### `feat/sqlx-compile-time-verification` +### `feat/typed-timestamps` Status: in_progress DoD: -- the highest-risk registry queries use `sqlx::query!` / `query_as!` -- compile-time SQL verification is enabled in normal backend workflow -- runtime-only queries remain only where dynamic SQL is truly required +- domain and registry models use typed timestamps instead of `String` +- SQL row decoding uses timestamp types directly where possible +- API serialization stays RFC 3339 compatible ## Next -- `feat/typed-timestamps` +- `feat/error-structure-normalization` ## Backlog diff --git a/crates/crank-registry/src/postgres/mod.rs b/crates/crank-registry/src/postgres/mod.rs index 3df1e8b..d83cff1 100644 --- a/crates/crank-registry/src/postgres/mod.rs +++ b/crates/crank-registry/src/postgres/mod.rs @@ -601,12 +601,12 @@ fn map_invocation_log_record(row: &PgRow) -> Result Result { - Ok(OperationUsageSummary { - operation_id: OperationId::new(row.try_get::("operation_id")?), - calls_today: to_u64(row.try_get::("calls_today")?, "calls_today")?, - error_rate_pct: row.try_get("error_rate_pct")?, - avg_latency_ms: to_u64(row.try_get::("avg_latency_ms")?, "avg_latency_ms")?, - }) + build_operation_usage_summary( + row.try_get("operation_id")?, + row.try_get("calls_today")?, + row.try_get("error_rate_pct")?, + row.try_get("avg_latency_ms")?, + ) } fn target_summary(target: &Target) -> (String, String) { @@ -662,15 +662,6 @@ fn join_target_url(base: &str, path: &str) -> String { } } -fn map_operation_agent_ref(row: &PgRow) -> Result { - Ok(OperationAgentRef { - operation_id: OperationId::new(row.try_get::("operation_id")?), - agent_id: AgentId::new(row.try_get::("agent_id")?), - agent_slug: row.try_get("agent_slug")?, - display_name: row.try_get("display_name")?, - }) -} - fn map_agent_binding(row: &PgRow) -> Result { Ok(AgentOperationBinding { operation_id: OperationId::new(row.try_get::("operation_id")?), @@ -772,6 +763,34 @@ fn build_auth_profile( }) } +fn build_operation_usage_summary( + operation_id: String, + calls_today: i64, + error_rate_pct: f64, + avg_latency_ms: i64, +) -> Result { + Ok(OperationUsageSummary { + operation_id: OperationId::new(operation_id), + calls_today: to_u64(calls_today, "calls_today")?, + error_rate_pct, + avg_latency_ms: to_u64(avg_latency_ms, "avg_latency_ms")?, + }) +} + +fn build_operation_agent_ref( + operation_id: String, + agent_id: String, + agent_slug: String, + display_name: String, +) -> Result { + Ok(OperationAgentRef { + operation_id: OperationId::new(operation_id), + agent_id: AgentId::new(agent_id), + agent_slug, + display_name, + }) +} + fn build_agent_version_record( summary: &AgentSummary, version: i32, @@ -1108,11 +1127,11 @@ mod tests { use crank_core::{ AgentId, AgentOperationBinding, AgentStatus, AgentVersion, ApiKeyHeaderAuthConfig, AsyncJobHandle, AuthConfig, AuthKind, AuthProfile, ConfigExport, ExecutionConfig, - ExportMode, GeneratedDraft, GeneratedDraftStatus, HttpMethod, JobStatus, MembershipRole, - OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, - PlatformApiKeyStatus, Protocol, RestTarget, RetryPolicy, Samples, SecretId, StreamSession, - StreamSessionId, StreamStatus, Target, ToolDescription, ToolExample, User, UserId, - Workspace, WorkspaceId, + ExportMode, GeneratedDraft, GeneratedDraftStatus, HttpMethod, InvocationLog, JobStatus, + MembershipRole, OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, + PlatformApiKeyScope, PlatformApiKeyStatus, Protocol, RestTarget, RetryPolicy, Samples, + SecretId, StreamSession, StreamSessionId, StreamStatus, Target, ToolDescription, + ToolExample, User, UserId, Workspace, WorkspaceId, }; use crank_mapping::{MappingRule, MappingSet}; use crank_schema::{Schema, SchemaKind}; @@ -1122,9 +1141,9 @@ mod tests { use crate::{ PostgresRegistry, RegistryError, model::{ - AsyncJobFilter, CreateAgentRequest, CreateAsyncJobRequest, CreatePlatformApiKeyRequest, - CreateStreamSessionRequest, CreateVersionRequest, CreateWorkspaceRequest, - CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata, + AsyncJobFilter, CreateAgentRequest, CreateAsyncJobRequest, CreateInvocationLogRequest, + CreatePlatformApiKeyRequest, CreateStreamSessionRequest, CreateVersionRequest, + CreateWorkspaceRequest, CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata, OperationSampleMetadata, PlatformApiKeyRecord, PublishAgentRequest, PublishRequest, RegistryOperation, SampleKind, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, StreamSessionFilter, UpdateAsyncJobStatusRequest, @@ -1788,6 +1807,119 @@ mod tests { database.cleanup().await; } + #[tokio::test] + async fn manages_operation_usage_and_agent_ref_reads() { + let database = TestDatabase::new().await; + let registry = database.registry().await; + let operation = test_operation("op_usage_01", 1, OperationStatus::Draft); + let operation_v2 = test_operation("op_usage_01", 2, OperationStatus::Draft); + let agent = test_agent("agent_usage_01", AgentStatus::Draft); + let version = test_agent_version(&agent.id, 1, AgentStatus::Draft); + let bindings = vec![AgentOperationBinding { + operation_id: operation.id.clone(), + operation_version: operation_v2.version, + tool_name: "create_lead".to_owned(), + tool_title: "Create lead".to_owned(), + tool_description_override: None, + enabled: true, + }]; + + registry + .create_operation(&test_workspace_id(), &operation, None) + .await + .unwrap(); + registry + .create_version(CreateVersionRequest { + workspace_id: &test_workspace_id(), + snapshot: &operation_v2, + change_note: Some("publishable"), + created_by: Some("alice"), + }) + .await + .unwrap(); + registry + .publish_operation(PublishRequest { + workspace_id: &test_workspace_id(), + operation_id: &operation.id, + version: operation_v2.version, + published_at: "2026-03-25T12:10:00Z", + published_by: Some("alice"), + }) + .await + .unwrap(); + registry + .create_agent(CreateAgentRequest { + agent: &agent, + version: &version, + bindings: &bindings, + }) + .await + .unwrap(); + registry + .publish_agent(PublishAgentRequest { + workspace_id: &test_workspace_id(), + agent_id: &agent.id, + version: version.version, + published_at: "2026-03-25T12:11:00Z", + published_by: Some("alice"), + }) + .await + .unwrap(); + registry + .create_invocation_log(CreateInvocationLogRequest { + log: &test_invocation_log( + "log_usage_ok", + &operation.id, + Some(agent.id.clone()), + crank_core::InvocationStatus::Ok, + 120, + "2026-03-25T12:20:00Z", + ), + }) + .await + .unwrap(); + registry + .create_invocation_log(CreateInvocationLogRequest { + log: &test_invocation_log( + "log_usage_err", + &operation.id, + Some(agent.id.clone()), + crank_core::InvocationStatus::Error, + 240, + "2026-03-25T12:21:00Z", + ), + }) + .await + .unwrap(); + + let has_bindings = registry + .has_published_agent_bindings_for_operation(&test_workspace_id(), &operation.id) + .await + .unwrap(); + let agent_refs = registry + .list_operation_agent_refs(&test_workspace_id()) + .await + .unwrap(); + let usage = registry + .list_operation_usage_summaries(&test_workspace_id(), "2026-03-25T12:00:00Z") + .await + .unwrap(); + + assert!(has_bindings); + assert_eq!(agent_refs.len(), 1); + assert_eq!(agent_refs[0].operation_id, operation.id); + assert_eq!(agent_refs[0].agent_id, agent.id); + assert_eq!(agent_refs[0].agent_slug, agent.slug); + assert_eq!(agent_refs[0].display_name, agent.display_name); + assert_eq!(usage.len(), 1); + assert_eq!(usage[0].operation_id, operation.id); + assert_eq!(usage[0].calls_today, 2); + assert_eq!(usage[0].error_rate_pct, 50.0); + assert_eq!(usage[0].avg_latency_ms, 180); + + database.cleanup().await; + } + #[tokio::test] async fn manages_stream_sessions_with_transitions_and_cleanup() { let database = TestDatabase::new().await; @@ -2278,6 +2410,34 @@ mod tests { } } + fn test_invocation_log( + id: &str, + operation_id: &OperationId, + agent_id: Option, + status: crank_core::InvocationStatus, + duration_ms: u64, + created_at: &str, + ) -> InvocationLog { + InvocationLog { + id: crank_core::InvocationLogId::new(id), + workspace_id: test_workspace_id(), + agent_id, + operation_id: operation_id.clone(), + source: crank_core::InvocationSource::AgentToolCall, + level: crank_core::InvocationLevel::Info, + status, + tool_name: "create_lead".to_owned(), + message: "invocation".to_owned(), + request_id: Some(format!("req_{id}")), + status_code: Some(200), + duration_ms, + error_kind: None, + request_preview: json!({"input":"value"}), + response_preview: json!({"ok":true}), + created_at: created_at.to_owned(), + } + } + fn test_async_job(id: &str, operation_id: &str, status: JobStatus) -> AsyncJobHandle { AsyncJobHandle { id: crank_core::AsyncJobId::new(id), diff --git a/crates/crank-registry/src/postgres/operation.rs b/crates/crank-registry/src/postgres/operation.rs index b06ab0e..7c79f6f 100644 --- a/crates/crank-registry/src/postgres/operation.rs +++ b/crates/crank-registry/src/postgres/operation.rs @@ -286,8 +286,8 @@ impl PostgresRegistry { workspace_id: &WorkspaceId, operation_id: &OperationId, ) -> Result { - let row = sqlx::query( - "select 1 + let row = sqlx::query!( + "select 1 as \"present!\" from agents a join published_agents pa on pa.agent_id = a.id join agent_operation_bindings b @@ -295,9 +295,9 @@ impl PostgresRegistry { where a.workspace_id = $1 and b.operation_id = $2 limit 1", + workspace_id.as_str(), + operation_id.as_str(), ) - .bind(workspace_id.as_str()) - .bind(operation_id.as_str()) .fetch_optional(&self.pool) .await?; @@ -386,7 +386,7 @@ impl PostgresRegistry { &self, workspace_id: &WorkspaceId, ) -> Result, RegistryError> { - let rows = sqlx::query( + let rows = sqlx::query!( "select b.operation_id, a.id as agent_id, @@ -398,12 +398,21 @@ impl PostgresRegistry { on b.agent_id = a.id and b.agent_version = pa.version where a.workspace_id = $1 order by a.display_name asc, b.tool_name asc", + workspace_id.as_str(), ) - .bind(workspace_id.as_str()) .fetch_all(&self.pool) .await?; - rows.iter().map(map_operation_agent_ref).collect() + rows.into_iter() + .map(|row| { + build_operation_agent_ref( + row.operation_id, + row.agent_id, + row.agent_slug, + row.display_name, + ) + }) + .collect() } pub async fn get_operation_summary(