Split admin agents service module
This commit is contained in:
github-ops
+16
-485
@@ -4,28 +4,26 @@ use std::sync::Arc;
|
||||
|
||||
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
|
||||
use crank_core::{
|
||||
Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AuditSink, AuthConfig,
|
||||
AuthKind, AuthProfile, AuthProfileId, CapabilityProfile, CommunityCapabilityProfile,
|
||||
ConfigExport, EditionCapabilities, ExecutionMode, ExportMode, GeneratedDraft,
|
||||
GeneratedDraftStatus, IdentityError, IdentityProvider, InvocationLevel, InvocationLog,
|
||||
InvocationLogId, InvocationSource, InvocationStatus, LoginOutcome, MembershipRole,
|
||||
NoopAuditSink, OperationId, OperationSecurityLevel, OperationStatus, OwnerOnlyPolicyEngine,
|
||||
PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, PolicyEngine,
|
||||
ProductEdition, Protocol, ResponseCachePolicy, SampleId, Samples, SecretKind, Target,
|
||||
ToolQualityMappingRule, ToolQualityMappingSet, ToolQualitySchemaKind, ToolQualitySchemaNode,
|
||||
UsagePeriod, UserSessionId, WizardState, Workspace, WorkspaceId, WorkspaceStatus,
|
||||
AgentId, AgentStatus, AuditSink, AuthConfig, AuthKind, AuthProfile, AuthProfileId,
|
||||
CapabilityProfile, CommunityCapabilityProfile, ConfigExport, EditionCapabilities,
|
||||
ExecutionMode, ExportMode, GeneratedDraft, GeneratedDraftStatus, IdentityError,
|
||||
IdentityProvider, InvocationLevel, InvocationLog, InvocationLogId, InvocationSource,
|
||||
InvocationStatus, LoginOutcome, MembershipRole, NoopAuditSink, OperationId,
|
||||
OperationSecurityLevel, OperationStatus, OwnerOnlyPolicyEngine, PlatformApiKey,
|
||||
PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, PolicyEngine, ProductEdition,
|
||||
Protocol, ResponseCachePolicy, SampleId, Samples, SecretKind, Target, ToolQualityMappingRule,
|
||||
ToolQualityMappingSet, ToolQualitySchemaKind, ToolQualitySchemaNode, UsagePeriod,
|
||||
UserSessionId, WizardState, Workspace, WorkspaceId, WorkspaceStatus,
|
||||
};
|
||||
use crank_mapping::{JsonPathRoot, MappingRule, MappingSet, infer_mapping_from_samples};
|
||||
use crank_registry::{
|
||||
AgentSummary, AgentVersionRecord, CreateAgentDraftVersionRequest, CreateAgentRequest,
|
||||
CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest,
|
||||
AgentSummary, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest,
|
||||
CreateWorkspaceRequest, ListInvocationLogsQuery, OperationAgentRef, OperationSampleMetadata,
|
||||
OperationSummary, OperationUsageSummary, OperationVersionRecord, PlatformApiKeyRecord,
|
||||
PostgresRegistry, PublishAgentRequest, PublishRequest, RegistryOperation, SampleKind,
|
||||
SaveAgentBindingsRequest, SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest,
|
||||
UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, UsageQuery,
|
||||
UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord,
|
||||
WorkspaceUpstream, WorkspaceUpstreamId,
|
||||
PostgresRegistry, PublishRequest, RegistryOperation, SampleKind, SaveSampleMetadataRequest,
|
||||
SaveWorkspaceUpstreamRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket,
|
||||
UsageOperationBreakdown, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord,
|
||||
WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId,
|
||||
};
|
||||
use crank_runtime::{
|
||||
PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, RuntimeOperation,
|
||||
@@ -39,6 +37,7 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339};
|
||||
use tracing::{info, instrument};
|
||||
use uuid::Uuid;
|
||||
|
||||
mod agents;
|
||||
mod import_export;
|
||||
mod observability;
|
||||
mod secrets;
|
||||
@@ -1813,474 +1812,6 @@ impl AdminService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub async fn list_agents(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
) -> Result<Vec<AgentSummaryView>, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let workspace = self.get_workspace(workspace_id).await?;
|
||||
let summaries = self.registry.list_agents(workspace_id).await?;
|
||||
let usage = self
|
||||
.registry
|
||||
.list_usage_by_agent(UsageQuery {
|
||||
workspace_id,
|
||||
period: UsagePeriod::Last24Hours,
|
||||
source: None,
|
||||
created_after: &today_start_utc()?,
|
||||
bucket: crank_registry::UsageBucket::Hour,
|
||||
})
|
||||
.await?;
|
||||
let calls_today = usage
|
||||
.into_iter()
|
||||
.map(|item| (item.agent_id.as_str().to_owned(), item.calls_total))
|
||||
.collect::<BTreeMap<_, _>>();
|
||||
let key_counts = self
|
||||
.registry
|
||||
.list_platform_api_keys(workspace_id)
|
||||
.await?
|
||||
.into_iter()
|
||||
.fold(BTreeMap::new(), |mut counts, record| {
|
||||
if let Some(agent_id) = record.api_key.agent_id {
|
||||
*counts.entry(agent_id.as_str().to_owned()).or_insert(0usize) += 1;
|
||||
}
|
||||
counts
|
||||
});
|
||||
|
||||
let mut items = Vec::with_capacity(summaries.len());
|
||||
for summary in summaries {
|
||||
let version = self
|
||||
.get_agent_version(workspace_id, &summary.id, summary.current_draft_version)
|
||||
.await?;
|
||||
let operation_ids = version
|
||||
.bindings
|
||||
.iter()
|
||||
.map(|binding| binding.operation_id.as_str().to_owned())
|
||||
.collect::<Vec<_>>();
|
||||
items.push(AgentSummaryView {
|
||||
operation_count: operation_ids.len(),
|
||||
operation_ids,
|
||||
key_count: key_counts.get(summary.id.as_str()).copied().unwrap_or(0),
|
||||
calls_today: calls_today.get(summary.id.as_str()).copied().unwrap_or(0),
|
||||
mcp_endpoint: agent_mcp_endpoint(
|
||||
workspace.workspace.slug.as_str(),
|
||||
summary.slug.as_str(),
|
||||
),
|
||||
..map_agent_summary_view(summary)
|
||||
});
|
||||
}
|
||||
|
||||
Ok(items)
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentSummaryView, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let workspace = self.get_workspace(workspace_id).await?;
|
||||
let summary = self
|
||||
.registry
|
||||
.get_agent_summary(workspace_id, agent_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!("agent {} was not found", agent_id.as_str()),
|
||||
json!({ "agent_id": agent_id.as_str() }),
|
||||
)
|
||||
})?;
|
||||
let version = self
|
||||
.get_agent_version(workspace_id, agent_id, summary.current_draft_version)
|
||||
.await?;
|
||||
let operation_ids = version
|
||||
.bindings
|
||||
.iter()
|
||||
.map(|binding| binding.operation_id.as_str().to_owned())
|
||||
.collect::<Vec<_>>();
|
||||
let usage = self
|
||||
.registry
|
||||
.get_usage_for_agent(
|
||||
UsageQuery {
|
||||
workspace_id,
|
||||
period: UsagePeriod::Last24Hours,
|
||||
source: None,
|
||||
created_after: &today_start_utc()?,
|
||||
bucket: crank_registry::UsageBucket::Hour,
|
||||
},
|
||||
agent_id,
|
||||
)
|
||||
.await?;
|
||||
let key_count = self
|
||||
.registry
|
||||
.list_platform_api_keys_for_agent(workspace_id, agent_id)
|
||||
.await?
|
||||
.len();
|
||||
|
||||
Ok(AgentSummaryView {
|
||||
operation_count: operation_ids.len(),
|
||||
operation_ids,
|
||||
key_count,
|
||||
calls_today: usage.map(|item| item.rollup.calls_total).unwrap_or(0),
|
||||
mcp_endpoint: agent_mcp_endpoint(
|
||||
workspace.workspace.slug.as_str(),
|
||||
summary.slug.as_str(),
|
||||
),
|
||||
..map_agent_summary_view(summary)
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_agent_version(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
version: u32,
|
||||
) -> Result<AgentVersionRecord, ApiError> {
|
||||
self.registry
|
||||
.get_agent_version(workspace_id, agent_id, version)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!(
|
||||
"agent version {version} for {} was not found",
|
||||
agent_id.as_str()
|
||||
),
|
||||
json!({
|
||||
"agent_id": agent_id.as_str(),
|
||||
"version": version,
|
||||
}),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_slug = %payload.slug))]
|
||||
pub async fn create_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
payload: AgentPayload,
|
||||
) -> Result<CreatedAgentResponse, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
|
||||
if self
|
||||
.find_agent_by_slug(workspace_id, &payload.slug)
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
return Err(ApiError::conflict_with_context(
|
||||
format!("agent with slug {} already exists", payload.slug),
|
||||
json!({ "slug": payload.slug }),
|
||||
));
|
||||
}
|
||||
|
||||
let now = OffsetDateTime::now_utc();
|
||||
let agent_id = AgentId::new(new_prefixed_id("agent"));
|
||||
let agent = Agent {
|
||||
id: agent_id.clone(),
|
||||
workspace_id: workspace_id.clone(),
|
||||
slug: payload.slug,
|
||||
display_name: payload.display_name,
|
||||
description: payload.description,
|
||||
status: AgentStatus::Draft,
|
||||
current_draft_version: 1,
|
||||
latest_published_version: None,
|
||||
created_at: now,
|
||||
updated_at: now,
|
||||
published_at: None,
|
||||
};
|
||||
let version = AgentVersion {
|
||||
agent_id: agent_id.clone(),
|
||||
version: 1,
|
||||
status: AgentStatus::Draft,
|
||||
instructions: payload.instructions,
|
||||
tool_selection_policy: payload.tool_selection_policy,
|
||||
created_at: now,
|
||||
};
|
||||
|
||||
self.registry
|
||||
.create_agent(CreateAgentRequest {
|
||||
agent: &agent,
|
||||
version: &version,
|
||||
bindings: &[],
|
||||
})
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), version = 1, "agent created");
|
||||
|
||||
Ok(CreatedAgentResponse {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
version: 1,
|
||||
status: AgentStatus::Draft,
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn update_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
payload: UpdateAgentPayload,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
let existing = self
|
||||
.registry
|
||||
.get_agent_summary(workspace_id, agent_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!("agent {} was not found", agent_id.as_str()),
|
||||
json!({ "agent_id": agent_id.as_str() }),
|
||||
)
|
||||
})?;
|
||||
|
||||
if payload.slug != existing.slug
|
||||
&& self
|
||||
.find_agent_by_slug(workspace_id, &payload.slug)
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
return Err(ApiError::conflict_with_context(
|
||||
format!("agent with slug {} already exists", payload.slug),
|
||||
json!({ "slug": payload.slug }),
|
||||
));
|
||||
}
|
||||
|
||||
let updated_at = OffsetDateTime::now_utc();
|
||||
self.registry
|
||||
.update_agent_summary(
|
||||
workspace_id,
|
||||
agent_id,
|
||||
&payload.slug,
|
||||
&payload.display_name,
|
||||
&payload.description,
|
||||
&updated_at,
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(updated_at),
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn delete_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
let existing = self
|
||||
.registry
|
||||
.get_agent_summary(workspace_id, agent_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!("agent {} was not found", agent_id.as_str()),
|
||||
json!({ "agent_id": agent_id.as_str() }),
|
||||
)
|
||||
})?;
|
||||
|
||||
self.registry.delete_agent(workspace_id, agent_id).await?;
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(existing.updated_at),
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn save_agent_bindings(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
payload: Vec<AgentBindingPayload>,
|
||||
) -> Result<AgentVersionRecord, ApiError> {
|
||||
let agent = self.get_agent(workspace_id, agent_id).await?;
|
||||
let bindings = payload
|
||||
.into_iter()
|
||||
.map(|binding| AgentOperationBinding {
|
||||
operation_id: OperationId::new(binding.operation_id),
|
||||
operation_version: binding.operation_version,
|
||||
tool_name: binding.tool_name,
|
||||
tool_title: binding.tool_title,
|
||||
tool_description_override: binding.tool_description_override,
|
||||
enabled: binding.enabled,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
self.registry
|
||||
.save_agent_bindings(SaveAgentBindingsRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
agent_version: agent.current_draft_version,
|
||||
bindings: &bindings,
|
||||
})
|
||||
.await?;
|
||||
info!(
|
||||
agent_id = %agent_id.as_str(),
|
||||
version = agent.current_draft_version,
|
||||
binding_count = bindings.len(),
|
||||
"agent bindings saved"
|
||||
);
|
||||
|
||||
self.get_agent_version(workspace_id, agent_id, agent.current_draft_version)
|
||||
.await
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), version))]
|
||||
pub async fn publish_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
version: u32,
|
||||
) -> Result<PublishAgentResponse, ApiError> {
|
||||
let agent_version = self
|
||||
.get_agent_version(workspace_id, agent_id, version)
|
||||
.await?;
|
||||
let published_bindings = self
|
||||
.published_agent_bindings(workspace_id, &agent_version.bindings)
|
||||
.await?;
|
||||
|
||||
if published_bindings.is_empty() {
|
||||
return Err(ApiError::conflict_with_context(
|
||||
"agent cannot be published without published enabled tools",
|
||||
json!({
|
||||
"agent_id": agent_id.as_str(),
|
||||
"version": version,
|
||||
"binding_count": agent_version.bindings.len()
|
||||
}),
|
||||
));
|
||||
}
|
||||
|
||||
let published_at = OffsetDateTime::now_utc();
|
||||
if published_bindings != agent_version.bindings {
|
||||
let draft_version = AgentVersion {
|
||||
agent_id: agent_id.clone(),
|
||||
version: agent_version.version + 1,
|
||||
status: AgentStatus::Draft,
|
||||
instructions: agent_version.snapshot.instructions.clone(),
|
||||
tool_selection_policy: agent_version.snapshot.tool_selection_policy.clone(),
|
||||
created_at: published_at,
|
||||
};
|
||||
|
||||
self.registry
|
||||
.create_agent_draft_version(CreateAgentDraftVersionRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
version: &draft_version,
|
||||
bindings: &agent_version.bindings,
|
||||
updated_at: &published_at,
|
||||
})
|
||||
.await?;
|
||||
|
||||
self.registry
|
||||
.save_agent_bindings(SaveAgentBindingsRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
agent_version: agent_version.version,
|
||||
bindings: &published_bindings,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
self.registry
|
||||
.publish_agent(PublishAgentRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
version,
|
||||
published_at: &published_at,
|
||||
published_by: None,
|
||||
})
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), version, "agent published");
|
||||
|
||||
Ok(PublishAgentResponse {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
published_version: version,
|
||||
published_at: format_timestamp(published_at),
|
||||
})
|
||||
}
|
||||
|
||||
async fn published_agent_bindings(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
bindings: &[AgentOperationBinding],
|
||||
) -> Result<Vec<AgentOperationBinding>, ApiError> {
|
||||
let mut published = Vec::new();
|
||||
|
||||
for binding in bindings {
|
||||
if !binding.enabled {
|
||||
continue;
|
||||
}
|
||||
|
||||
let Some(summary) = self
|
||||
.registry
|
||||
.get_operation_summary(workspace_id, &binding.operation_id)
|
||||
.await?
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let Some(operation_version) = summary.latest_published_version else {
|
||||
continue;
|
||||
};
|
||||
|
||||
published.push(AgentOperationBinding {
|
||||
operation_id: summary.id,
|
||||
operation_version,
|
||||
tool_name: binding.tool_name.clone(),
|
||||
tool_title: binding.tool_title.clone(),
|
||||
tool_description_override: binding.tool_description_override.clone(),
|
||||
enabled: true,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(published)
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn unpublish_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let updated_at = OffsetDateTime::now_utc();
|
||||
self.registry
|
||||
.unpublish_agent(workspace_id, agent_id, &updated_at)
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), "agent moved to draft");
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(updated_at),
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn archive_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let updated_at = OffsetDateTime::now_utc();
|
||||
self.registry
|
||||
.archive_agent(workspace_id, agent_id, &updated_at)
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), "agent archived");
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(updated_at),
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))]
|
||||
pub async fn save_json_sample(
|
||||
&self,
|
||||
|
||||
@@ -0,0 +1,492 @@
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use crank_core::{
|
||||
Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, OperationId, UsagePeriod,
|
||||
WorkspaceId,
|
||||
};
|
||||
use crank_registry::{
|
||||
AgentVersionRecord, CreateAgentDraftVersionRequest, CreateAgentRequest, PublishAgentRequest,
|
||||
SaveAgentBindingsRequest, UsageBucket, UsageQuery,
|
||||
};
|
||||
use serde_json::json;
|
||||
use time::OffsetDateTime;
|
||||
use tracing::{info, instrument};
|
||||
|
||||
use crate::{
|
||||
error::ApiError,
|
||||
service::{
|
||||
AdminService, AgentBindingPayload, AgentMutationResult, AgentPayload, AgentSummaryView,
|
||||
CreatedAgentResponse, PublishAgentResponse, UpdateAgentPayload, agent_mcp_endpoint,
|
||||
format_timestamp, map_agent_summary_view, new_prefixed_id, today_start_utc,
|
||||
},
|
||||
};
|
||||
|
||||
impl AdminService {
|
||||
#[instrument(skip(self))]
|
||||
pub async fn list_agents(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
) -> Result<Vec<AgentSummaryView>, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let workspace = self.get_workspace(workspace_id).await?;
|
||||
let summaries = self.registry.list_agents(workspace_id).await?;
|
||||
let usage = self
|
||||
.registry
|
||||
.list_usage_by_agent(UsageQuery {
|
||||
workspace_id,
|
||||
period: UsagePeriod::Last24Hours,
|
||||
source: None,
|
||||
created_after: &today_start_utc()?,
|
||||
bucket: UsageBucket::Hour,
|
||||
})
|
||||
.await?;
|
||||
let calls_today = usage
|
||||
.into_iter()
|
||||
.map(|item| (item.agent_id.as_str().to_owned(), item.calls_total))
|
||||
.collect::<BTreeMap<_, _>>();
|
||||
let key_counts = self
|
||||
.registry
|
||||
.list_platform_api_keys(workspace_id)
|
||||
.await?
|
||||
.into_iter()
|
||||
.fold(BTreeMap::new(), |mut counts, record| {
|
||||
if let Some(agent_id) = record.api_key.agent_id {
|
||||
*counts.entry(agent_id.as_str().to_owned()).or_insert(0usize) += 1;
|
||||
}
|
||||
counts
|
||||
});
|
||||
|
||||
let mut items = Vec::with_capacity(summaries.len());
|
||||
for summary in summaries {
|
||||
let version = self
|
||||
.get_agent_version(workspace_id, &summary.id, summary.current_draft_version)
|
||||
.await?;
|
||||
let operation_ids = version
|
||||
.bindings
|
||||
.iter()
|
||||
.map(|binding| binding.operation_id.as_str().to_owned())
|
||||
.collect::<Vec<_>>();
|
||||
items.push(AgentSummaryView {
|
||||
operation_count: operation_ids.len(),
|
||||
operation_ids,
|
||||
key_count: key_counts.get(summary.id.as_str()).copied().unwrap_or(0),
|
||||
calls_today: calls_today.get(summary.id.as_str()).copied().unwrap_or(0),
|
||||
mcp_endpoint: agent_mcp_endpoint(
|
||||
workspace.workspace.slug.as_str(),
|
||||
summary.slug.as_str(),
|
||||
),
|
||||
..map_agent_summary_view(summary)
|
||||
});
|
||||
}
|
||||
|
||||
Ok(items)
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentSummaryView, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let workspace = self.get_workspace(workspace_id).await?;
|
||||
let summary = self
|
||||
.registry
|
||||
.get_agent_summary(workspace_id, agent_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!("agent {} was not found", agent_id.as_str()),
|
||||
json!({ "agent_id": agent_id.as_str() }),
|
||||
)
|
||||
})?;
|
||||
let version = self
|
||||
.get_agent_version(workspace_id, agent_id, summary.current_draft_version)
|
||||
.await?;
|
||||
let operation_ids = version
|
||||
.bindings
|
||||
.iter()
|
||||
.map(|binding| binding.operation_id.as_str().to_owned())
|
||||
.collect::<Vec<_>>();
|
||||
let usage = self
|
||||
.registry
|
||||
.get_usage_for_agent(
|
||||
UsageQuery {
|
||||
workspace_id,
|
||||
period: UsagePeriod::Last24Hours,
|
||||
source: None,
|
||||
created_after: &today_start_utc()?,
|
||||
bucket: UsageBucket::Hour,
|
||||
},
|
||||
agent_id,
|
||||
)
|
||||
.await?;
|
||||
let key_count = self
|
||||
.registry
|
||||
.list_platform_api_keys_for_agent(workspace_id, agent_id)
|
||||
.await?
|
||||
.len();
|
||||
|
||||
Ok(AgentSummaryView {
|
||||
operation_count: operation_ids.len(),
|
||||
operation_ids,
|
||||
key_count,
|
||||
calls_today: usage.map(|item| item.rollup.calls_total).unwrap_or(0),
|
||||
mcp_endpoint: agent_mcp_endpoint(
|
||||
workspace.workspace.slug.as_str(),
|
||||
summary.slug.as_str(),
|
||||
),
|
||||
..map_agent_summary_view(summary)
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_agent_version(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
version: u32,
|
||||
) -> Result<AgentVersionRecord, ApiError> {
|
||||
self.registry
|
||||
.get_agent_version(workspace_id, agent_id, version)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!(
|
||||
"agent version {version} for {} was not found",
|
||||
agent_id.as_str()
|
||||
),
|
||||
json!({
|
||||
"agent_id": agent_id.as_str(),
|
||||
"version": version,
|
||||
}),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_slug = %payload.slug))]
|
||||
pub async fn create_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
payload: AgentPayload,
|
||||
) -> Result<CreatedAgentResponse, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
|
||||
if self
|
||||
.find_agent_by_slug(workspace_id, &payload.slug)
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
return Err(ApiError::conflict_with_context(
|
||||
format!("agent with slug {} already exists", payload.slug),
|
||||
json!({ "slug": payload.slug }),
|
||||
));
|
||||
}
|
||||
|
||||
let now = OffsetDateTime::now_utc();
|
||||
let agent_id = AgentId::new(new_prefixed_id("agent"));
|
||||
let agent = Agent {
|
||||
id: agent_id.clone(),
|
||||
workspace_id: workspace_id.clone(),
|
||||
slug: payload.slug,
|
||||
display_name: payload.display_name,
|
||||
description: payload.description,
|
||||
status: AgentStatus::Draft,
|
||||
current_draft_version: 1,
|
||||
latest_published_version: None,
|
||||
created_at: now,
|
||||
updated_at: now,
|
||||
published_at: None,
|
||||
};
|
||||
let version = AgentVersion {
|
||||
agent_id: agent_id.clone(),
|
||||
version: 1,
|
||||
status: AgentStatus::Draft,
|
||||
instructions: payload.instructions,
|
||||
tool_selection_policy: payload.tool_selection_policy,
|
||||
created_at: now,
|
||||
};
|
||||
|
||||
self.registry
|
||||
.create_agent(CreateAgentRequest {
|
||||
agent: &agent,
|
||||
version: &version,
|
||||
bindings: &[],
|
||||
})
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), version = 1, "agent created");
|
||||
|
||||
Ok(CreatedAgentResponse {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
version: 1,
|
||||
status: AgentStatus::Draft,
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn update_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
payload: UpdateAgentPayload,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
let existing = self
|
||||
.registry
|
||||
.get_agent_summary(workspace_id, agent_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!("agent {} was not found", agent_id.as_str()),
|
||||
json!({ "agent_id": agent_id.as_str() }),
|
||||
)
|
||||
})?;
|
||||
|
||||
if payload.slug != existing.slug
|
||||
&& self
|
||||
.find_agent_by_slug(workspace_id, &payload.slug)
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
return Err(ApiError::conflict_with_context(
|
||||
format!("agent with slug {} already exists", payload.slug),
|
||||
json!({ "slug": payload.slug }),
|
||||
));
|
||||
}
|
||||
|
||||
let updated_at = OffsetDateTime::now_utc();
|
||||
self.registry
|
||||
.update_agent_summary(
|
||||
workspace_id,
|
||||
agent_id,
|
||||
&payload.slug,
|
||||
&payload.display_name,
|
||||
&payload.description,
|
||||
&updated_at,
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(updated_at),
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn delete_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
let existing = self
|
||||
.registry
|
||||
.get_agent_summary(workspace_id, agent_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_with_context(
|
||||
format!("agent {} was not found", agent_id.as_str()),
|
||||
json!({ "agent_id": agent_id.as_str() }),
|
||||
)
|
||||
})?;
|
||||
|
||||
self.registry.delete_agent(workspace_id, agent_id).await?;
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(existing.updated_at),
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn save_agent_bindings(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
payload: Vec<AgentBindingPayload>,
|
||||
) -> Result<AgentVersionRecord, ApiError> {
|
||||
let agent = self.get_agent(workspace_id, agent_id).await?;
|
||||
let bindings = payload
|
||||
.into_iter()
|
||||
.map(|binding| AgentOperationBinding {
|
||||
operation_id: OperationId::new(binding.operation_id),
|
||||
operation_version: binding.operation_version,
|
||||
tool_name: binding.tool_name,
|
||||
tool_title: binding.tool_title,
|
||||
tool_description_override: binding.tool_description_override,
|
||||
enabled: binding.enabled,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
self.registry
|
||||
.save_agent_bindings(SaveAgentBindingsRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
agent_version: agent.current_draft_version,
|
||||
bindings: &bindings,
|
||||
})
|
||||
.await?;
|
||||
info!(
|
||||
agent_id = %agent_id.as_str(),
|
||||
version = agent.current_draft_version,
|
||||
binding_count = bindings.len(),
|
||||
"agent bindings saved"
|
||||
);
|
||||
|
||||
self.get_agent_version(workspace_id, agent_id, agent.current_draft_version)
|
||||
.await
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), version))]
|
||||
pub async fn publish_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
version: u32,
|
||||
) -> Result<PublishAgentResponse, ApiError> {
|
||||
let agent_version = self
|
||||
.get_agent_version(workspace_id, agent_id, version)
|
||||
.await?;
|
||||
let published_bindings = self
|
||||
.published_agent_bindings(workspace_id, &agent_version.bindings)
|
||||
.await?;
|
||||
|
||||
if published_bindings.is_empty() {
|
||||
return Err(ApiError::conflict_with_context(
|
||||
"agent cannot be published without published enabled tools",
|
||||
json!({
|
||||
"agent_id": agent_id.as_str(),
|
||||
"version": version,
|
||||
"binding_count": agent_version.bindings.len()
|
||||
}),
|
||||
));
|
||||
}
|
||||
|
||||
let published_at = OffsetDateTime::now_utc();
|
||||
if published_bindings != agent_version.bindings {
|
||||
let draft_version = AgentVersion {
|
||||
agent_id: agent_id.clone(),
|
||||
version: agent_version.version + 1,
|
||||
status: AgentStatus::Draft,
|
||||
instructions: agent_version.snapshot.instructions.clone(),
|
||||
tool_selection_policy: agent_version.snapshot.tool_selection_policy.clone(),
|
||||
created_at: published_at,
|
||||
};
|
||||
|
||||
self.registry
|
||||
.create_agent_draft_version(CreateAgentDraftVersionRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
version: &draft_version,
|
||||
bindings: &agent_version.bindings,
|
||||
updated_at: &published_at,
|
||||
})
|
||||
.await?;
|
||||
|
||||
self.registry
|
||||
.save_agent_bindings(SaveAgentBindingsRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
agent_version: agent_version.version,
|
||||
bindings: &published_bindings,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
self.registry
|
||||
.publish_agent(PublishAgentRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
version,
|
||||
published_at: &published_at,
|
||||
published_by: None,
|
||||
})
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), version, "agent published");
|
||||
|
||||
Ok(PublishAgentResponse {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
published_version: version,
|
||||
published_at: format_timestamp(published_at),
|
||||
})
|
||||
}
|
||||
|
||||
async fn published_agent_bindings(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
bindings: &[AgentOperationBinding],
|
||||
) -> Result<Vec<AgentOperationBinding>, ApiError> {
|
||||
let mut published = Vec::new();
|
||||
|
||||
for binding in bindings {
|
||||
if !binding.enabled {
|
||||
continue;
|
||||
}
|
||||
|
||||
let Some(summary) = self
|
||||
.registry
|
||||
.get_operation_summary(workspace_id, &binding.operation_id)
|
||||
.await?
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let Some(operation_version) = summary.latest_published_version else {
|
||||
continue;
|
||||
};
|
||||
|
||||
published.push(AgentOperationBinding {
|
||||
operation_id: summary.id,
|
||||
operation_version,
|
||||
tool_name: binding.tool_name.clone(),
|
||||
tool_title: binding.tool_title.clone(),
|
||||
tool_description_override: binding.tool_description_override.clone(),
|
||||
enabled: true,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(published)
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn unpublish_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let updated_at = OffsetDateTime::now_utc();
|
||||
self.registry
|
||||
.unpublish_agent(workspace_id, agent_id, &updated_at)
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), "agent moved to draft");
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(updated_at),
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
|
||||
pub async fn archive_agent(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
agent_id: &AgentId,
|
||||
) -> Result<AgentMutationResult, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let updated_at = OffsetDateTime::now_utc();
|
||||
self.registry
|
||||
.archive_agent(workspace_id, agent_id, &updated_at)
|
||||
.await?;
|
||||
info!(agent_id = %agent_id.as_str(), "agent archived");
|
||||
|
||||
Ok(AgentMutationResult {
|
||||
agent_id: agent_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
updated_at: format_timestamp(updated_at),
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user