registry: extract agent and operation modules

TASKS.md

@@ -2,18 +2,18 @@
 
 ## Current
 
-### `feat/postgres-registry-modularization`
+### `feat/sqlx-compile-time-verification`
 
 Status: in_progress
 
 DoD:
-- `crates/crank-registry/src/postgres.rs` is split into domain modules under `postgres/`
-- `PostgresRegistry` keeps only constructors and high-level surface wiring
-- existing registry tests still pass after module split
+- 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
 
 ## Next
 
-- `feat/sqlx-compile-time-verification`
+- `feat/typed-timestamps`
 
 ## Backlog
 

crates/crank-registry/src/postgres/agent.rs

@@ -0,0 +1,471 @@
+use super::*;
+
+impl PostgresRegistry {
+    pub async fn list_agents(
+        &self,
+        workspace_id: &WorkspaceId,
+    ) -> Result<Vec<AgentSummary>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                id,
+                workspace_id,
+                slug,
+                display_name,
+                description,
+                status,
+                current_draft_version,
+                latest_published_version,
+                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
+                to_char(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
+             from agents
+             where workspace_id = $1
+             order by slug asc",
+        )
+        .bind(workspace_id.as_str())
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter().map(map_agent_summary).collect()
+    }
+
+    pub async fn get_agent_summary(
+        &self,
+        workspace_id: &WorkspaceId,
+        agent_id: &AgentId,
+    ) -> Result<Option<AgentSummary>, RegistryError> {
+        let row = sqlx::query(
+            "select
+                id,
+                workspace_id,
+                slug,
+                display_name,
+                description,
+                status,
+                current_draft_version,
+                latest_published_version,
+                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
+                to_char(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
+             from agents
+             where workspace_id = $1 and id = $2",
+        )
+        .bind(workspace_id.as_str())
+        .bind(agent_id.as_str())
+        .fetch_optional(&self.pool)
+        .await?;
+
+        row.as_ref().map(map_agent_summary).transpose()
+    }
+
+    pub async fn create_agent(&self, request: CreateAgentRequest<'_>) -> Result<(), RegistryError> {
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query(
+            "insert into agents (
+                id,
+                workspace_id,
+                slug,
+                display_name,
+                description,
+                status,
+                current_draft_version,
+                latest_published_version,
+                created_at,
+                updated_at,
+                published_at
+            ) values (
+                $1, $2, $3, $4, $5, $6, $7, $8, $9::timestamptz, $10::timestamptz, $11::timestamptz
+            )",
+        )
+        .bind(request.agent.id.as_str())
+        .bind(request.agent.workspace_id.as_str())
+        .bind(&request.agent.slug)
+        .bind(&request.agent.display_name)
+        .bind(&request.agent.description)
+        .bind(serialize_enum_text(&request.agent.status, "status")?)
+        .bind(to_db_version(request.agent.current_draft_version))
+        .bind(request.agent.latest_published_version.map(to_db_version))
+        .bind(&request.agent.created_at)
+        .bind(&request.agent.updated_at)
+        .bind(request.agent.published_at.as_deref())
+        .execute(&mut *tx)
+        .await?;
+
+        insert_agent_version_row(&mut tx, request.version).await?;
+        replace_agent_bindings_rows(
+            &mut tx,
+            &request.agent.id,
+            request.version.version,
+            request.bindings,
+        )
+        .await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn get_agent_version(
+        &self,
+        workspace_id: &WorkspaceId,
+        agent_id: &AgentId,
+        version: u32,
+    ) -> Result<Option<AgentVersionRecord>, RegistryError> {
+        let summary = match self.get_agent_summary(workspace_id, agent_id).await? {
+            Some(value) => value,
+            None => return Ok(None),
+        };
+
+        let row = sqlx::query(
+            "select
+                agent_id,
+                version,
+                status,
+                instructions_json,
+                tool_selection_policy_json,
+                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
+             from agent_versions
+             where agent_id = $1 and version = $2",
+        )
+        .bind(agent_id.as_str())
+        .bind(to_db_version(version))
+        .fetch_optional(&self.pool)
+        .await?;
+
+        let Some(row) = row else {
+            return Ok(None);
+        };
+
+        let bindings = self.list_agent_bindings(agent_id, version).await?;
+        Ok(Some(map_agent_version_record(&summary, &row, bindings)?))
+    }
+
+    pub async fn save_agent_bindings(
+        &self,
+        request: SaveAgentBindingsRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        if self
+            .get_agent_summary(request.workspace_id, request.agent_id)
+            .await?
+            .is_none()
+        {
+            return Err(RegistryError::AgentNotFound {
+                agent_id: request.agent_id.as_str().to_owned(),
+            });
+        }
+
+        let mut tx = self.pool.begin().await?;
+        replace_agent_bindings_rows(
+            &mut tx,
+            request.agent_id,
+            request.agent_version,
+            request.bindings,
+        )
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn update_agent_summary(
+        &self,
+        workspace_id: &WorkspaceId,
+        agent_id: &AgentId,
+        slug: &str,
+        display_name: &str,
+        description: &str,
+        updated_at: &str,
+    ) -> Result<(), RegistryError> {
+        let result = sqlx::query(
+            "update agents
+                set slug = $3,
+                    display_name = $4,
+                    description = $5,
+                    updated_at = $6::timestamptz
+             where workspace_id = $1 and id = $2",
+        )
+        .bind(workspace_id.as_str())
+        .bind(agent_id.as_str())
+        .bind(slug)
+        .bind(display_name)
+        .bind(description)
+        .bind(updated_at)
+        .execute(&self.pool)
+        .await?;
+
+        if result.rows_affected() == 0 {
+            return Err(RegistryError::AgentNotFound {
+                agent_id: agent_id.as_str().to_owned(),
+            });
+        }
+
+        Ok(())
+    }
+
+    pub async fn delete_agent(
+        &self,
+        workspace_id: &WorkspaceId,
+        agent_id: &AgentId,
+    ) -> Result<(), RegistryError> {
+        let result = sqlx::query("delete from agents where workspace_id = $1 and id = $2")
+            .bind(workspace_id.as_str())
+            .bind(agent_id.as_str())
+            .execute(&self.pool)
+            .await?;
+
+        if result.rows_affected() == 0 {
+            return Err(RegistryError::AgentNotFound {
+                agent_id: agent_id.as_str().to_owned(),
+            });
+        }
+
+        Ok(())
+    }
+
+    pub async fn publish_agent(
+        &self,
+        request: PublishAgentRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        if self
+            .get_agent_version(request.workspace_id, request.agent_id, request.version)
+            .await?
+            .is_none()
+        {
+            return Err(RegistryError::AgentNotFound {
+                agent_id: request.agent_id.as_str().to_owned(),
+            });
+        }
+
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query(
+            "insert into published_agents (
+                agent_id,
+                version,
+                published_at,
+                published_by
+             ) values ($1, $2, $3::timestamptz, $4)
+             on conflict(agent_id) do update set
+                version = excluded.version,
+                published_at = excluded.published_at,
+                published_by = excluded.published_by",
+        )
+        .bind(request.agent_id.as_str())
+        .bind(to_db_version(request.version))
+        .bind(request.published_at)
+        .bind(request.published_by)
+        .execute(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            "update agent_versions
+                set status = $1
+              where agent_id = $2 and version = $3",
+        )
+        .bind(serialize_enum_text(&AgentStatus::Published, "status")?)
+        .bind(request.agent_id.as_str())
+        .bind(to_db_version(request.version))
+        .execute(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            "update agents
+                set status = $1,
+                    latest_published_version = $2,
+                    published_at = $3::timestamptz,
+                    updated_at = $4::timestamptz
+              where id = $5 and workspace_id = $6",
+        )
+        .bind(serialize_enum_text(&AgentStatus::Published, "status")?)
+        .bind(to_db_version(request.version))
+        .bind(request.published_at)
+        .bind(request.published_at)
+        .bind(request.agent_id.as_str())
+        .bind(request.workspace_id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn unpublish_agent(
+        &self,
+        workspace_id: &WorkspaceId,
+        agent_id: &AgentId,
+        updated_at: &str,
+    ) -> Result<(), RegistryError> {
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query("delete from published_agents where agent_id = $1")
+            .bind(agent_id.as_str())
+            .execute(&mut *tx)
+            .await?;
+
+        sqlx::query(
+            "update agent_versions
+                set status = $1
+              where agent_id = $2
+                and version = (
+                    select current_draft_version
+                    from agents
+                    where id = $2 and workspace_id = $3
+                )",
+        )
+        .bind(serialize_enum_text(&AgentStatus::Draft, "status")?)
+        .bind(agent_id.as_str())
+        .bind(workspace_id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        let result = sqlx::query(
+            "update agents
+                set status = $1,
+                    published_at = null,
+                    updated_at = $2::timestamptz
+              where id = $3 and workspace_id = $4",
+        )
+        .bind(serialize_enum_text(&AgentStatus::Draft, "status")?)
+        .bind(updated_at)
+        .bind(agent_id.as_str())
+        .bind(workspace_id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        if result.rows_affected() == 0 {
+            return Err(RegistryError::AgentNotFound {
+                agent_id: agent_id.as_str().to_owned(),
+            });
+        }
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn archive_agent(
+        &self,
+        workspace_id: &WorkspaceId,
+        agent_id: &AgentId,
+        updated_at: &str,
+    ) -> Result<(), RegistryError> {
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query("delete from published_agents where agent_id = $1")
+            .bind(agent_id.as_str())
+            .execute(&mut *tx)
+            .await?;
+
+        sqlx::query(
+            "update agent_versions
+                set status = $1
+              where agent_id = $2
+                and version = (
+                    select current_draft_version
+                    from agents
+                    where id = $2 and workspace_id = $3
+                )",
+        )
+        .bind(serialize_enum_text(&AgentStatus::Archived, "status")?)
+        .bind(agent_id.as_str())
+        .bind(workspace_id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        let result = sqlx::query(
+            "update agents
+                set status = $1,
+                    published_at = null,
+                    updated_at = $2::timestamptz
+              where id = $3 and workspace_id = $4",
+        )
+        .bind(serialize_enum_text(&AgentStatus::Archived, "status")?)
+        .bind(updated_at)
+        .bind(agent_id.as_str())
+        .bind(workspace_id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        if result.rows_affected() == 0 {
+            return Err(RegistryError::AgentNotFound {
+                agent_id: agent_id.as_str().to_owned(),
+            });
+        }
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn get_published_agent_tools_by_slug(
+        &self,
+        workspace_slug: &str,
+        agent_slug: &str,
+    ) -> Result<Vec<PublishedAgentTool>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                w.id as workspace_id,
+                w.slug as workspace_slug,
+                a.id as agent_id,
+                a.slug as agent_slug,
+                b.tool_name,
+                b.tool_title,
+                coalesce(b.tool_description_override, ov.tool_description_json->>'description') as tool_description,
+                o.id,
+                o.name,
+                o.display_name,
+                o.category,
+                o.protocol,
+                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
+                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
+                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
+                ov.version,
+                ov.status,
+                ov.target_json,
+                ov.input_schema_json,
+                ov.output_schema_json,
+                ov.input_mapping_json,
+                ov.output_mapping_json,
+                ov.execution_config_json,
+                ov.tool_description_json,
+                ov.samples_json,
+                ov.generated_draft_json,
+                ov.config_export_json,
+                ov.change_note,
+                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                ov.created_by
+             from workspaces w
+             join agents a on a.workspace_id = w.id
+             join published_agents pa on pa.agent_id = a.id
+             join agent_operation_bindings b on b.agent_id = a.id and b.agent_version = pa.version
+             join operation_versions ov on ov.operation_id = b.operation_id and ov.version = b.operation_version
+             join operations o on o.id = ov.operation_id and o.workspace_id = w.id
+             where w.slug = $1 and a.slug = $2 and b.enabled = true
+             order by b.tool_name asc",
+        )
+        .bind(workspace_slug)
+        .bind(agent_slug)
+        .fetch_all(&self.pool)
+        .await?;
+
+        if rows.is_empty() {
+            let exists = sqlx::query(
+                "select 1
+                 from workspaces w
+                 join agents a on a.workspace_id = w.id
+                 join published_agents pa on pa.agent_id = a.id
+                 where w.slug = $1 and a.slug = $2",
+            )
+            .bind(workspace_slug)
+            .bind(agent_slug)
+            .fetch_optional(&self.pool)
+            .await?;
+
+            if exists.is_none() {
+                return Err(RegistryError::PublishedAgentNotFound {
+                    workspace_slug: workspace_slug.to_owned(),
+                    agent_slug: agent_slug.to_owned(),
+                });
+            }
+        }
+
+        rows.iter().map(map_published_agent_tool).collect()
+    }
+}

crates/crank-registry/src/postgres/operation.rs

@@ -0,0 +1,808 @@
+use super::*;
+
+impl PostgresRegistry {
+    pub async fn create_operation(
+        &self,
+        workspace_id: &WorkspaceId,
+        snapshot: &RegistryOperation,
+        created_by: Option<&str>,
+    ) -> Result<(), RegistryError> {
+        if snapshot.version != 1 {
+            return Err(RegistryError::InvalidInitialVersion {
+                operation_id: snapshot.id.as_str().to_owned(),
+                version: snapshot.version,
+            });
+        }
+
+        if self
+            .get_operation_summary(workspace_id, &snapshot.id)
+            .await?
+            .is_some()
+        {
+            return Err(RegistryError::OperationAlreadyExists {
+                operation_id: snapshot.id.as_str().to_owned(),
+            });
+        }
+
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query(
+            "insert into operations (
+                id,
+                workspace_id,
+                name,
+                display_name,
+                category,
+                protocol,
+                status,
+                current_draft_version,
+                latest_published_version,
+                created_at,
+                updated_at,
+                published_at
+            ) values (
+                $1, $2, $3, $4, $5, $6, $7, $8, $9,
+                $10::timestamptz,
+                $11::timestamptz,
+                $12::timestamptz
+            )",
+        )
+        .bind(snapshot.id.as_str())
+        .bind(workspace_id.as_str())
+        .bind(&snapshot.name)
+        .bind(&snapshot.display_name)
+        .bind(&snapshot.category)
+        .bind(serialize_enum_text(&snapshot.protocol, "protocol")?)
+        .bind(serialize_enum_text(&snapshot.status, "status")?)
+        .bind(to_db_version(snapshot.version))
+        .bind(
+            snapshot
+                .published_at
+                .as_ref()
+                .map(|_| to_db_version(snapshot.version)),
+        )
+        .bind(&snapshot.created_at)
+        .bind(&snapshot.updated_at)
+        .bind(snapshot.published_at.as_deref())
+        .execute(&mut *tx)
+        .await?;
+
+        insert_version_row(&mut tx, snapshot, None, created_by).await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn create_version(
+        &self,
+        request: CreateVersionRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        let Some(summary) = self
+            .get_operation_summary(request.workspace_id, &request.snapshot.id)
+            .await?
+        else {
+            return Err(RegistryError::OperationNotFound {
+                operation_id: request.snapshot.id.as_str().to_owned(),
+            });
+        };
+
+        assert_immutable_fields(&summary, request.snapshot)?;
+
+        let expected = summary.current_draft_version + 1;
+        if request.snapshot.version != expected {
+            return Err(RegistryError::InvalidVersionSequence {
+                operation_id: request.snapshot.id.as_str().to_owned(),
+                expected,
+                actual: request.snapshot.version,
+            });
+        }
+
+        let mut tx = self.pool.begin().await?;
+
+        insert_version_row(
+            &mut tx,
+            request.snapshot,
+            request.change_note,
+            request.created_by,
+        )
+        .await?;
+
+        sqlx::query(
+            "update operations
+                set status = $1,
+                    current_draft_version = $2,
+                    updated_at = $3::timestamptz
+              where id = $4",
+        )
+        .bind(serialize_enum_text(&request.snapshot.status, "status")?)
+        .bind(to_db_version(request.snapshot.version))
+        .bind(&request.snapshot.updated_at)
+        .bind(request.snapshot.id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn update_operation_draft(
+        &self,
+        workspace_id: &WorkspaceId,
+        snapshot: &RegistryOperation,
+    ) -> Result<(), RegistryError> {
+        let Some(summary) = self
+            .get_operation_summary(workspace_id, &snapshot.id)
+            .await?
+        else {
+            return Err(RegistryError::OperationNotFound {
+                operation_id: snapshot.id.as_str().to_owned(),
+            });
+        };
+
+        if snapshot.version != summary.current_draft_version {
+            return Err(RegistryError::InvalidVersionSequence {
+                operation_id: snapshot.id.as_str().to_owned(),
+                expected: summary.current_draft_version,
+                actual: snapshot.version,
+            });
+        }
+
+        assert_immutable_fields(&summary, snapshot)?;
+
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query(
+            "update operations
+                set display_name = $1,
+                    category = $2,
+                    status = $3,
+                    updated_at = $4::timestamptz
+              where workspace_id = $5 and id = $6",
+        )
+        .bind(&snapshot.display_name)
+        .bind(&snapshot.category)
+        .bind(serialize_enum_text(&snapshot.status, "status")?)
+        .bind(&snapshot.updated_at)
+        .bind(workspace_id.as_str())
+        .bind(snapshot.id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            "update operation_versions
+                set status = $1,
+                    target_json = $2,
+                    input_schema_json = $3,
+                    output_schema_json = $4,
+                    input_mapping_json = $5,
+                    output_mapping_json = $6,
+                    execution_config_json = $7,
+                    tool_description_json = $8,
+                    samples_json = $9,
+                    generated_draft_json = $10,
+                    config_export_json = $11
+              where operation_id = $12 and version = $13",
+        )
+        .bind(serialize_enum_text(&snapshot.status, "status")?)
+        .bind(Json(serialize_json_value(&snapshot.target)?))
+        .bind(Json(serialize_json_value(&snapshot.input_schema)?))
+        .bind(Json(serialize_json_value(&snapshot.output_schema)?))
+        .bind(Json(serialize_json_value(&snapshot.input_mapping)?))
+        .bind(Json(serialize_json_value(&snapshot.output_mapping)?))
+        .bind(Json(serialize_json_value(&snapshot.execution_config)?))
+        .bind(Json(serialize_json_value(&snapshot.tool_description)?))
+        .bind(serialize_option_json_value(&snapshot.samples)?.map(Json))
+        .bind(serialize_option_json_value(&snapshot.generated_draft)?.map(Json))
+        .bind(serialize_option_json_value(&snapshot.config_export)?.map(Json))
+        .bind(snapshot.id.as_str())
+        .bind(to_db_version(snapshot.version))
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn archive_operation(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &OperationId,
+        archived_at: &str,
+    ) -> Result<(), RegistryError> {
+        let Some(summary) = self
+            .get_operation_summary(workspace_id, operation_id)
+            .await?
+        else {
+            return Err(RegistryError::OperationNotFound {
+                operation_id: operation_id.as_str().to_owned(),
+            });
+        };
+
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query(
+            "update operations
+                set status = $1,
+                    updated_at = $2::timestamptz
+              where workspace_id = $3 and id = $4",
+        )
+        .bind(serialize_enum_text(&OperationStatus::Archived, "status")?)
+        .bind(archived_at)
+        .bind(workspace_id.as_str())
+        .bind(operation_id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            "update operation_versions
+                set status = $1
+              where operation_id = $2 and version = $3",
+        )
+        .bind(serialize_enum_text(&OperationStatus::Archived, "status")?)
+        .bind(operation_id.as_str())
+        .bind(to_db_version(summary.current_draft_version))
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn delete_operation(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &OperationId,
+    ) -> Result<(), RegistryError> {
+        if self
+            .has_published_agent_bindings_for_operation(workspace_id, operation_id)
+            .await?
+        {
+            return Err(RegistryError::OperationHasPublishedAgentBindings {
+                operation_id: operation_id.as_str().to_owned(),
+            });
+        }
+
+        let deleted = sqlx::query(
+            "delete from operations
+              where workspace_id = $1 and id = $2",
+        )
+        .bind(workspace_id.as_str())
+        .bind(operation_id.as_str())
+        .execute(&self.pool)
+        .await?
+        .rows_affected();
+
+        if deleted == 0 {
+            return Err(RegistryError::OperationNotFound {
+                operation_id: operation_id.as_str().to_owned(),
+            });
+        }
+
+        Ok(())
+    }
+
+    pub async fn has_published_agent_bindings_for_operation(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &OperationId,
+    ) -> Result<bool, RegistryError> {
+        let row = sqlx::query(
+            "select 1
+             from agents a
+             join published_agents pa on pa.agent_id = a.id
+             join agent_operation_bindings b
+               on b.agent_id = a.id and b.agent_version = pa.version
+             where a.workspace_id = $1
+               and b.operation_id = $2
+             limit 1",
+        )
+        .bind(workspace_id.as_str())
+        .bind(operation_id.as_str())
+        .fetch_optional(&self.pool)
+        .await?;
+
+        Ok(row.is_some())
+    }
+
+    pub async fn list_operations(
+        &self,
+        workspace_id: &WorkspaceId,
+    ) -> Result<Vec<OperationSummary>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                operations.id,
+                operations.workspace_id,
+                operations.name,
+                operations.display_name,
+                operations.category,
+                operations.protocol,
+                ov.target_json,
+                operations.status,
+                operations.current_draft_version,
+                operations.latest_published_version,
+                to_char(operations.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                to_char(operations.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
+                to_char(operations.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
+             from operations
+             join operation_versions ov
+               on ov.operation_id = operations.id
+              and ov.version = operations.current_draft_version
+             where operations.workspace_id = $1
+             order by operations.name asc",
+        )
+        .bind(workspace_id.as_str())
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter().map(map_operation_summary).collect()
+    }
+
+    pub async fn list_operation_usage_summaries(
+        &self,
+        workspace_id: &WorkspaceId,
+        created_after: &str,
+    ) -> Result<Vec<OperationUsageSummary>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                operation_id,
+                count(*)::bigint as calls_today,
+                coalesce(
+                    round((sum(case when status = 'error' then 1 else 0 end)::numeric / nullif(count(*), 0)) * 100, 2),
+                    0
+                )::float8 as error_rate_pct,
+                coalesce(round(avg(duration_ms))::bigint, 0) as avg_latency_ms
+             from invocation_logs
+             where workspace_id = $1
+               and created_at >= $2::timestamptz
+             group by operation_id",
+        )
+        .bind(workspace_id.as_str())
+        .bind(created_after)
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter().map(map_operation_usage_summary).collect()
+    }
+
+    pub async fn list_operation_agent_refs(
+        &self,
+        workspace_id: &WorkspaceId,
+    ) -> Result<Vec<OperationAgentRef>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                b.operation_id,
+                a.id as agent_id,
+                a.slug as agent_slug,
+                a.display_name
+             from agents a
+             join published_agents pa on pa.agent_id = a.id
+             join agent_operation_bindings b
+               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",
+        )
+        .bind(workspace_id.as_str())
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter().map(map_operation_agent_ref).collect()
+    }
+
+    pub async fn get_operation_summary(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &OperationId,
+    ) -> Result<Option<OperationSummary>, RegistryError> {
+        let row = sqlx::query(
+            "select
+                operations.id,
+                operations.workspace_id,
+                operations.name,
+                operations.display_name,
+                operations.category,
+                operations.protocol,
+                ov.target_json,
+                operations.status,
+                operations.current_draft_version,
+                operations.latest_published_version,
+                to_char(operations.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                to_char(operations.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
+                to_char(operations.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
+             from operations
+             join operation_versions ov
+               on ov.operation_id = operations.id
+              and ov.version = operations.current_draft_version
+             where operations.workspace_id = $1 and operations.id = $2",
+        )
+        .bind(workspace_id.as_str())
+        .bind(operation_id.as_str())
+        .fetch_optional(&self.pool)
+        .await?;
+
+        row.as_ref().map(map_operation_summary).transpose()
+    }
+
+    pub async fn get_operation_version(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &OperationId,
+        version: u32,
+    ) -> Result<Option<OperationVersionRecord>, RegistryError> {
+        let row = sqlx::query(
+            "select
+                o.id,
+                o.workspace_id,
+                o.name,
+                o.display_name,
+                o.category,
+                o.protocol,
+                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
+                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
+                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
+                ov.version,
+                ov.status,
+                ov.target_json,
+                ov.input_schema_json,
+                ov.output_schema_json,
+                ov.input_mapping_json,
+                ov.output_mapping_json,
+                ov.execution_config_json,
+                ov.tool_description_json,
+                ov.samples_json,
+                ov.generated_draft_json,
+                ov.config_export_json,
+                ov.change_note,
+                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                ov.created_by
+             from operation_versions ov
+             join operations o on o.id = ov.operation_id
+             where o.workspace_id = $1 and ov.operation_id = $2 and ov.version = $3",
+        )
+        .bind(workspace_id.as_str())
+        .bind(operation_id.as_str())
+        .bind(to_db_version(version))
+        .fetch_optional(&self.pool)
+        .await?;
+
+        row.as_ref().map(map_operation_version_record).transpose()
+    }
+
+    pub async fn list_operation_versions(
+        &self,
+        workspace_id: &WorkspaceId,
+        operation_id: &OperationId,
+    ) -> Result<Vec<OperationVersionRecord>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                o.id,
+                o.workspace_id,
+                o.name,
+                o.display_name,
+                o.category,
+                o.protocol,
+                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
+                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
+                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
+                ov.version,
+                ov.status,
+                ov.target_json,
+                ov.input_schema_json,
+                ov.output_schema_json,
+                ov.input_mapping_json,
+                ov.output_mapping_json,
+                ov.execution_config_json,
+                ov.tool_description_json,
+                ov.samples_json,
+                ov.generated_draft_json,
+                ov.config_export_json,
+                ov.change_note,
+                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                ov.created_by
+             from operation_versions ov
+             join operations o on o.id = ov.operation_id
+             where o.workspace_id = $1 and ov.operation_id = $2
+             order by ov.version asc",
+        )
+        .bind(workspace_id.as_str())
+        .bind(operation_id.as_str())
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter().map(map_operation_version_record).collect()
+    }
+
+    pub async fn publish_operation(
+        &self,
+        request: PublishRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        if self
+            .get_operation_version(request.workspace_id, request.operation_id, request.version)
+            .await?
+            .is_none()
+        {
+            return Err(RegistryError::OperationVersionNotFound {
+                operation_id: request.operation_id.as_str().to_owned(),
+                version: request.version,
+            });
+        }
+
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query(
+            "insert into published_operations (
+                operation_id,
+                version,
+                published_at,
+                published_by
+             ) values ($1, $2, $3::timestamptz, $4)
+             on conflict(operation_id) do update set
+                version = excluded.version,
+                published_at = excluded.published_at,
+                published_by = excluded.published_by",
+        )
+        .bind(request.operation_id.as_str())
+        .bind(to_db_version(request.version))
+        .bind(request.published_at)
+        .bind(request.published_by)
+        .execute(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            "update operation_versions
+                set status = $1
+              where operation_id = $2 and version = $3",
+        )
+        .bind(serialize_enum_text(&OperationStatus::Published, "status")?)
+        .bind(request.operation_id.as_str())
+        .bind(to_db_version(request.version))
+        .execute(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            "update operations
+                set status = $1,
+                    latest_published_version = $2,
+                    published_at = $3::timestamptz,
+                    updated_at = $4::timestamptz
+              where id = $5",
+        )
+        .bind(serialize_enum_text(&OperationStatus::Published, "status")?)
+        .bind(to_db_version(request.version))
+        .bind(request.published_at)
+        .bind(request.published_at)
+        .bind(request.operation_id.as_str())
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn get_published_operation(
+        &self,
+        operation_id: &OperationId,
+    ) -> Result<Option<RegistryOperation>, RegistryError> {
+        let row = sqlx::query(
+            "select
+                o.id,
+                o.workspace_id,
+                o.name,
+                o.display_name,
+                o.category,
+                o.protocol,
+                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
+                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
+                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
+                ov.version,
+                ov.status,
+                ov.target_json,
+                ov.input_schema_json,
+                ov.output_schema_json,
+                ov.input_mapping_json,
+                ov.output_mapping_json,
+                ov.execution_config_json,
+                ov.tool_description_json,
+                ov.samples_json,
+                ov.generated_draft_json,
+                ov.config_export_json,
+                ov.change_note,
+                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                ov.created_by
+             from published_operations po
+             join operation_versions ov
+               on ov.operation_id = po.operation_id and ov.version = po.version
+             join operations o on o.id = po.operation_id
+             where po.operation_id = $1",
+        )
+        .bind(operation_id.as_str())
+        .fetch_optional(&self.pool)
+        .await?;
+
+        row.as_ref()
+            .map(|value| map_operation_version_record(value).map(|record| record.snapshot))
+            .transpose()
+    }
+
+    pub async fn list_published_operations(&self) -> Result<Vec<RegistryOperation>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                o.id,
+                o.workspace_id,
+                o.name,
+                o.display_name,
+                o.category,
+                o.protocol,
+                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
+                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
+                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
+                ov.version,
+                ov.status,
+                ov.target_json,
+                ov.input_schema_json,
+                ov.output_schema_json,
+                ov.input_mapping_json,
+                ov.output_mapping_json,
+                ov.execution_config_json,
+                ov.tool_description_json,
+                ov.samples_json,
+                ov.generated_draft_json,
+                ov.config_export_json,
+                ov.change_note,
+                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
+                ov.created_by
+             from published_operations po
+             join operation_versions ov
+               on ov.operation_id = po.operation_id and ov.version = po.version
+             join operations o on o.id = po.operation_id
+             order by o.name asc",
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter()
+            .map(|row| map_operation_version_record(row).map(|record| record.snapshot))
+            .collect()
+    }
+
+    pub async fn save_sample_metadata(
+        &self,
+        request: SaveSampleMetadataRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        sqlx::query(
+            "insert into operation_samples (
+                id,
+                operation_id,
+                version,
+                sample_kind,
+                storage_ref,
+                content_type,
+                file_name,
+                created_at
+             ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
+             on conflict(id) do update set
+                operation_id = excluded.operation_id,
+                version = excluded.version,
+                sample_kind = excluded.sample_kind,
+                storage_ref = excluded.storage_ref,
+                content_type = excluded.content_type,
+                file_name = excluded.file_name,
+                created_at = excluded.created_at",
+        )
+        .bind(request.sample.id.as_str())
+        .bind(request.sample.operation_id.as_str())
+        .bind(to_db_version(request.sample.version))
+        .bind(serialize_enum_text(
+            &request.sample.sample_kind,
+            "sample_kind",
+        )?)
+        .bind(&request.sample.storage_ref)
+        .bind(&request.sample.content_type)
+        .bind(request.sample.file_name.as_deref())
+        .bind(&request.sample.created_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn list_sample_metadata(
+        &self,
+        operation_id: &OperationId,
+        version: u32,
+    ) -> Result<Vec<OperationSampleMetadata>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                id,
+                operation_id,
+                version,
+                sample_kind,
+                storage_ref,
+                content_type,
+                file_name,
+                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
+             from operation_samples
+             where operation_id = $1 and version = $2
+             order by created_at asc",
+        )
+        .bind(operation_id.as_str())
+        .bind(to_db_version(version))
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter().map(map_sample_metadata).collect()
+    }
+
+    pub async fn save_descriptor_metadata(
+        &self,
+        request: SaveDescriptorMetadataRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        sqlx::query(
+            "insert into descriptors (
+                id,
+                operation_id,
+                version,
+                descriptor_kind,
+                storage_ref,
+                source_name,
+                package_index_json,
+                created_at
+             ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
+             on conflict(id) do update set
+                operation_id = excluded.operation_id,
+                version = excluded.version,
+                descriptor_kind = excluded.descriptor_kind,
+                storage_ref = excluded.storage_ref,
+                source_name = excluded.source_name,
+                package_index_json = excluded.package_index_json,
+                created_at = excluded.created_at",
+        )
+        .bind(request.descriptor.id.as_str())
+        .bind(
+            request
+                .descriptor
+                .operation_id
+                .as_ref()
+                .map(|value| value.as_str()),
+        )
+        .bind(request.descriptor.version.map(to_db_version))
+        .bind(serialize_enum_text(
+            &request.descriptor.descriptor_kind,
+            "descriptor_kind",
+        )?)
+        .bind(&request.descriptor.storage_ref)
+        .bind(request.descriptor.source_name.as_deref())
+        .bind(serialize_option_json_value(&request.descriptor.package_index)?.map(Json))
+        .bind(&request.descriptor.created_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn list_descriptor_metadata(
+        &self,
+        operation_id: &OperationId,
+        version: u32,
+    ) -> Result<Vec<DescriptorMetadata>, RegistryError> {
+        let rows = sqlx::query(
+            "select
+                id,
+                operation_id,
+                version,
+                descriptor_kind,
+                storage_ref,
+                source_name,
+                package_index_json,
+                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
+             from descriptors
+             where operation_id = $1 and version = $2
+             order by created_at asc",
+        )
+        .bind(operation_id.as_str())
+        .bind(to_db_version(version))
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.iter().map(map_descriptor_metadata).collect()
+    }
+}