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

registry: finish sqlx verification for operation reads

Browse Source
This commit is contained in:
a.tolmachev
2026-04-12 21:57:09 +00:00
parent 075c1762e2
commit 624224f089
5 changed files with 267 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.sqlx/query-6eed48468f97843bbcd6a2f90448b68c15efa72029740e28bb881f54f73a6b9b.json
Generated
+23
View File
@@ -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"
}
.sqlx/query-fdd05b145021a19dda9a9f44c31ff21a2f73a4416d0cf20be418bba36def9eb4.json
Generated
+40
View File
@@ -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"
}
TASKS.md
+5 -5
View File
@@ -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
crates/crank-registry/src/postgres/mod.rs
+183 -23
View File
@@ -601,12 +601,12 @@ fn map_invocation_log_record(row: &PgRow) -> Result<InvocationLogRecord, Registr
}
fn map_operation_usage_summary(row: &PgRow) -> Result<OperationUsageSummary, RegistryError> {
Ok(OperationUsageSummary {
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
calls_today: to_u64(row.try_get::<i64, _>("calls_today")?, "calls_today")?,
error_rate_pct: row.try_get("error_rate_pct")?,
avg_latency_ms: to_u64(row.try_get::<i64, _>("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<OperationAgentRef, RegistryError> {
Ok(OperationAgentRef {
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
agent_id: AgentId::new(row.try_get::<String, _>("agent_id")?),
agent_slug: row.try_get("agent_slug")?,
display_name: row.try_get("display_name")?,
})
}
fn map_agent_binding(row: &PgRow) -> Result<AgentOperationBinding, RegistryError> {
Ok(AgentOperationBinding {
operation_id: OperationId::new(row.try_get::<String, _>("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<OperationUsageSummary, RegistryError> {
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<OperationAgentRef, RegistryError> {
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<AgentId>,
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),
crates/crank-registry/src/postgres/operation.rs
+16 -7
View File
@@ -286,8 +286,8 @@ impl PostgresRegistry {
workspace_id: &WorkspaceId,
operation_id: &OperationId,
) -> Result<bool, RegistryError> {
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<Vec<OperationAgentRef>, 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(