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

registry: add sqlx checks for published operation reads

Browse Source
This commit is contained in:
a.tolmachev
2026-04-12 21:17:56 +00:00
parent 934c872753
commit c940b9b97a
9 changed files with 966 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/crank-registry/src/postgres/mod.rs
+141 -74
View File
@@ -665,38 +665,6 @@ fn map_operation_agent_ref(row: &PgRow) -> Result<OperationAgentRef, RegistryErr
})
}
fn map_operation_version_record(row: &PgRow) -> Result<OperationVersionRecord, RegistryError> {
build_operation_version_record(
row.try_get("id")?,
row.try_get("workspace_id")?,
row.try_get("name")?,
row.try_get("display_name")?,
row.try_get("category")?,
row.try_get("protocol")?,
row.try_get("operation_created_at")?,
row.try_get("operation_updated_at")?,
row.try_get("operation_published_at")?,
row.try_get("version")?,
row.try_get("status")?,
row.try_get::<Json<Value>, _>("target_json")?.0,
row.try_get::<Json<Value>, _>("input_schema_json")?.0,
row.try_get::<Json<Value>, _>("output_schema_json")?.0,
row.try_get::<Json<Value>, _>("input_mapping_json")?.0,
row.try_get::<Json<Value>, _>("output_mapping_json")?.0,
row.try_get::<Json<Value>, _>("execution_config_json")?.0,
row.try_get::<Json<Value>, _>("tool_description_json")?.0,
row.try_get::<Option<Json<Value>>, _>("samples_json")?
.map(|value| value.0),
row.try_get::<Option<Json<Value>>, _>("generated_draft_json")?
.map(|value| value.0),
row.try_get::<Option<Json<Value>>, _>("config_export_json")?
.map(|value| value.0),
row.try_get("change_note")?,
row.try_get("created_at")?,
row.try_get("created_by")?,
)
}
fn map_auth_profile(row: &PgRow) -> Result<AuthProfile, RegistryError> {
Ok(AuthProfile {
id: crank_core::AuthProfileId::new(row.try_get::<String, _>("id")?),
@@ -886,57 +854,74 @@ fn build_operation_version_record(
})
}
fn map_published_agent_tool(row: &PgRow) -> Result<PublishedAgentTool, RegistryError> {
let record = map_operation_version_record(row)?;
#[allow(clippy::too_many_arguments)]
fn build_published_agent_tool(
workspace_id: String,
workspace_slug: String,
agent_id: String,
agent_slug: String,
tool_name: String,
tool_title: String,
tool_description: String,
operation: RegistryOperation,
) -> Result<PublishedAgentTool, RegistryError> {
Ok(PublishedAgentTool {
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
workspace_slug: row.try_get("workspace_slug")?,
agent_id: AgentId::new(row.try_get::<String, _>("agent_id")?),
agent_slug: row.try_get("agent_slug")?,
tool_name: row.try_get("tool_name")?,
tool_title: row.try_get("tool_title")?,
tool_description: row.try_get("tool_description")?,
operation: record.snapshot,
workspace_id: WorkspaceId::new(workspace_id),
workspace_slug,
agent_id: AgentId::new(agent_id),
agent_slug,
tool_name,
tool_title,
tool_description,
operation,
})
}
fn map_sample_metadata(row: &PgRow) -> Result<OperationSampleMetadata, RegistryError> {
#[allow(clippy::too_many_arguments)]
fn build_sample_metadata(
id: String,
operation_id: String,
version: i32,
sample_kind: String,
storage_ref: String,
content_type: String,
file_name: Option<String>,
created_at: String,
) -> Result<OperationSampleMetadata, RegistryError> {
Ok(OperationSampleMetadata {
id: crank_core::SampleId::new(row.try_get::<String, _>("id")?),
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
version: from_db_version(row.try_get("version")?, "version")?,
sample_kind: deserialize_enum_text(
&row.try_get::<String, _>("sample_kind")?,
"sample_kind",
)?,
storage_ref: row.try_get("storage_ref")?,
content_type: row.try_get("content_type")?,
file_name: row.try_get("file_name")?,
created_at: row.try_get("created_at")?,
id: crank_core::SampleId::new(id),
operation_id: OperationId::new(operation_id),
version: from_db_version(version, "version")?,
sample_kind: deserialize_enum_text(&sample_kind, "sample_kind")?,
storage_ref,
content_type,
file_name,
created_at,
})
}
fn map_descriptor_metadata(row: &PgRow) -> Result<DescriptorMetadata, RegistryError> {
#[allow(clippy::too_many_arguments)]
fn build_descriptor_metadata(
id: String,
operation_id: Option<String>,
version: Option<i32>,
descriptor_kind: String,
storage_ref: String,
source_name: Option<String>,
package_index: Option<Value>,
created_at: String,
) -> Result<DescriptorMetadata, RegistryError> {
Ok(DescriptorMetadata {
id: crank_core::DescriptorId::new(row.try_get::<String, _>("id")?),
operation_id: row
.try_get::<Option<String>, _>("operation_id")?
.map(OperationId::new),
version: row
.try_get::<Option<i32>, _>("version")?
id: crank_core::DescriptorId::new(id),
operation_id: operation_id.map(OperationId::new),
version: version
.map(|value| from_db_version(value, "version"))
.transpose()?,
descriptor_kind: deserialize_enum_text(
&row.try_get::<String, _>("descriptor_kind")?,
"descriptor_kind",
)?,
storage_ref: row.try_get("storage_ref")?,
source_name: row.try_get("source_name")?,
package_index: row
.try_get::<Option<Json<Value>>, _>("package_index_json")?
.map(|value| value.0),
created_at: row.try_get("created_at")?,
descriptor_kind: deserialize_enum_text(&descriptor_kind, "descriptor_kind")?,
storage_ref,
source_name,
package_index,
created_at,
})
}
@@ -1126,8 +1111,8 @@ mod tests {
AsyncJobFilter, CreateAgentRequest, CreateAsyncJobRequest, CreatePlatformApiKeyRequest,
CreateStreamSessionRequest, CreateVersionRequest, CreateWorkspaceRequest,
CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata,
OperationSampleMetadata, PlatformApiKeyRecord, PublishRequest, RegistryOperation,
SampleKind, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
OperationSampleMetadata, PlatformApiKeyRecord, PublishAgentRequest, PublishRequest,
RegistryOperation, SampleKind, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
SaveSampleMetadataRequest, StreamSessionFilter, UpdateAsyncJobStatusRequest,
UpdateStreamSessionStateRequest, WorkspaceRecord, YamlImportJobCompletion,
YamlImportJobId, YamlImportJobStatus,
@@ -1186,6 +1171,7 @@ mod tests {
.await
.unwrap()
.unwrap();
let published_list = registry.list_published_operations().await.unwrap();
assert_eq!(summary.current_draft_version, 2);
assert_eq!(summary.latest_published_version, Some(2));
@@ -1197,6 +1183,7 @@ mod tests {
);
assert_eq!(published.version, 2);
assert!(published.is_published());
assert_eq!(published_list, vec![published.clone()]);
database.cleanup().await;
}
@@ -1641,6 +1628,86 @@ mod tests {
database.cleanup().await;
}
#[tokio::test]
async fn manages_published_agent_tool_reads() {
let database = TestDatabase::new().await;
let registry = database.registry().await;
let operation = test_operation("op_agent_pub_01", 1, OperationStatus::Draft);
let operation_v2 = test_operation("op_agent_pub_01", 2, OperationStatus::Draft);
let agent = test_agent("agent_pub_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: Some("Creates CRM lead".to_owned()),
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();
let tools = registry
.get_published_agent_tools_by_slug("default", &agent.slug)
.await
.unwrap();
assert_eq!(tools.len(), 1);
assert_eq!(tools[0].workspace_id, test_workspace_id());
assert_eq!(tools[0].workspace_slug, "default");
assert_eq!(tools[0].agent_id, agent.id);
assert_eq!(tools[0].agent_slug, agent.slug);
assert_eq!(tools[0].tool_name, bindings[0].tool_name);
assert_eq!(tools[0].tool_title, bindings[0].tool_title);
assert_eq!(tools[0].tool_description, "Creates CRM lead");
assert_eq!(tools[0].operation.id, operation_v2.id);
assert_eq!(tools[0].operation.version, operation_v2.version);
assert_eq!(tools[0].operation.protocol, operation_v2.protocol);
assert!(tools[0].operation.is_published());
database.cleanup().await;
}
#[tokio::test]
async fn manages_stream_sessions_with_transitions_and_cleanup() {
let database = TestDatabase::new().await;