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

community: remove remaining premium wizard and admin surface

Browse Source
This commit is contained in:
a.tolmachev
2026-05-07 21:16:36 +00:00
parent 29f971c22d
commit ca9b86aada
15 changed files with 48 additions and 1278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/admin-api/Cargo.toml
+5
View File
@@ -5,6 +5,11 @@ license.workspace = true
rust-version.workspace = true
version.workspace = true
[[bin]]
name = "admin-api"
path = "src/main.rs"
test = false
[dependencies]
argon2.workspace = true
axum.workspace = true
apps/admin-api/src/service.rs
+21 -477
View File
@@ -17,15 +17,15 @@ use crank_core::{
};
use crank_mapping::{JsonPathRoot, MappingSet, infer_mapping_from_samples};
use crank_registry::{
AgentSummary, AgentVersionRecord, AsyncJobFilter, CreateAgentRequest, CreateAsyncJobRequest,
AgentSummary, AgentVersionRecord, CreateAgentRequest, CreateAsyncJobRequest,
CreateInvitationRequest, CreateInvocationLogRequest, CreatePlatformApiKeyRequest,
CreateSecretRequest, CreateStreamSessionRequest, CreateVersionRequest, CreateWorkspaceRequest,
InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord,
OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary,
OperationVersionRecord, Page, PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest,
OperationVersionRecord, PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest,
PublishRequest, RegistryError, RegistryOperation, RotateSecretRequest, SampleKind,
SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveSampleMetadataRequest,
StreamSessionFilter, UpdateAsyncJobStatusRequest, UpdateWorkspaceRequest,
UpdateAsyncJobStatusRequest, UpdateWorkspaceRequest,
UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, UsageQuery, UsageSummary,
UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord,
};
@@ -434,74 +434,6 @@ pub struct ProtocolCapabilityView {
pub supports_aggregation_mode: Vec<AggregationMode>,
}
#[derive(Clone, Debug, Deserialize)]
pub struct StreamSessionsQuery {
pub operation_id: Option<String>,
pub agent_id: Option<String>,
pub status: Option<StreamStatus>,
pub mode: Option<ExecutionMode>,
pub page: Option<u32>,
pub page_size: Option<u32>,
}
#[derive(Clone, Debug, Serialize)]
pub struct StreamSessionSummaryView {
pub id: String,
pub operation_id: String,
pub agent_id: Option<String>,
pub protocol: Protocol,
pub mode: ExecutionMode,
pub status: StreamStatus,
#[serde(with = "time::serde::rfc3339")]
pub expires_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339::option")]
pub last_poll_at: Option<OffsetDateTime>,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339::option")]
pub closed_at: Option<OffsetDateTime>,
}
#[derive(Clone, Debug, Serialize)]
pub struct StreamSessionDetailView {
#[serde(flatten)]
pub summary: StreamSessionSummaryView,
pub cursor: Option<Value>,
}
#[derive(Clone, Debug, Deserialize)]
pub struct AsyncJobsQuery {
pub operation_id: Option<String>,
pub agent_id: Option<String>,
pub status: Option<JobStatus>,
pub page: Option<u32>,
pub page_size: Option<u32>,
}
#[derive(Clone, Debug, Serialize)]
pub struct AsyncJobSummaryView {
pub id: String,
pub operation_id: String,
pub agent_id: Option<String>,
pub status: JobStatus,
pub progress: Value,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339")]
pub updated_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339::option")]
pub finished_at: Option<OffsetDateTime>,
#[serde(with = "time::serde::rfc3339::option")]
pub expires_at: Option<OffsetDateTime>,
}
#[derive(Clone, Debug, Serialize)]
pub struct AsyncJobDetailView {
#[serde(flatten)]
pub summary: AsyncJobSummaryView,
pub error: Option<Value>,
}
#[derive(Clone, Debug, Deserialize)]
pub struct GenerateDraftPayload {
#[serde(default)]
@@ -1613,327 +1545,6 @@ impl AdminService {
))
}
#[instrument(skip(self))]
pub async fn list_stream_sessions(
&self,
workspace_id: &WorkspaceId,
query: StreamSessionsQuery,
) -> Result<Page<StreamSessionSummaryView>, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
let operation_id = query.operation_id.as_deref().map(OperationId::new);
let agent_id = query.agent_id.as_deref().map(AgentId::new);
let limit = query.page_size.unwrap_or(20).max(1);
let page = self
.registry
.list_stream_sessions(StreamSessionFilter {
workspace_id,
agent_id: agent_id.as_ref(),
operation_id: operation_id.as_ref(),
status: query.status,
mode: query.mode,
limit,
})
.await?;
Ok(Page {
items: page
.items
.into_iter()
.map(stream_session_summary_view)
.collect(),
total: page.total,
})
}
#[instrument(skip(self))]
pub async fn get_stream_session(
&self,
workspace_id: &WorkspaceId,
session_id: &crank_core::StreamSessionId,
) -> Result<StreamSessionDetailView, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
let session = self
.registry
.get_stream_session(session_id)
.await?
.ok_or_else(|| {
ApiError::not_found_with_context(
format!("stream session {} was not found", session_id.as_str()),
json!({ "session_id": session_id.as_str() }),
)
})?;
ensure_stream_session_workspace(&session, workspace_id)?;
let session = self
.touch_stream_session_poll_if_ready(workspace_id, session)
.await?;
Ok(stream_session_detail_view(session))
}
#[instrument(skip(self))]
pub async fn stop_stream_session(
&self,
workspace_id: &WorkspaceId,
session_id: &crank_core::StreamSessionId,
) -> Result<StreamSessionDetailView, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
let session = self
.registry
.get_stream_session(session_id)
.await?
.ok_or_else(|| {
ApiError::not_found_with_context(
format!("stream session {} was not found", session_id.as_str()),
json!({ "session_id": session_id.as_str() }),
)
})?;
ensure_stream_session_workspace(&session, workspace_id)?;
self.registry
.close_stream_session(session_id, &OffsetDateTime::now_utc())
.await?;
let updated = self
.registry
.get_stream_session(session_id)
.await?
.ok_or_else(|| {
ApiError::not_found_with_context(
format!("stream session {} was not found", session_id.as_str()),
json!({ "session_id": session_id.as_str() }),
)
})?;
Ok(stream_session_detail_view(updated))
}
#[instrument(skip(self))]
pub async fn list_async_jobs(
&self,
workspace_id: &WorkspaceId,
query: AsyncJobsQuery,
) -> Result<Page<AsyncJobSummaryView>, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
let operation_id = query.operation_id.as_deref().map(OperationId::new);
let agent_id = query.agent_id.as_deref().map(AgentId::new);
let limit = query.page_size.unwrap_or(20).max(1);
let page = self
.registry
.list_async_jobs(AsyncJobFilter {
workspace_id,
agent_id: agent_id.as_ref(),
operation_id: operation_id.as_ref(),
status: query.status,
limit,
})
.await?;
Ok(Page {
items: page.items.into_iter().map(async_job_summary_view).collect(),
total: page.total,
})
}
#[instrument(skip(self))]
pub async fn get_async_job(
&self,
workspace_id: &WorkspaceId,
job_id: &crank_core::AsyncJobId,
) -> Result<AsyncJobDetailView, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
let job = self.registry.get_async_job(job_id).await?.ok_or_else(|| {
ApiError::not_found_with_context(
format!("async job {} was not found", job_id.as_str()),
json!({ "job_id": job_id.as_str() }),
)
})?;
ensure_async_job_workspace(&job, workspace_id)?;
let job = if matches!(job.status, JobStatus::Completed) {
job
} else {
self.touch_async_job_poll_if_ready(workspace_id, job)
.await?
};
Ok(async_job_detail_view(job))
}
#[instrument(skip(self))]
pub async fn cancel_async_job(
&self,
workspace_id: &WorkspaceId,
job_id: &crank_core::AsyncJobId,
) -> Result<AsyncJobDetailView, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
let job = self.registry.get_async_job(job_id).await?.ok_or_else(|| {
ApiError::not_found_with_context(
format!("async job {} was not found", job_id.as_str()),
json!({ "job_id": job_id.as_str() }),
)
})?;
ensure_async_job_workspace(&job, workspace_id)?;
self.registry
.cancel_async_job(job_id, &OffsetDateTime::now_utc())
.await?;
let updated = self.registry.get_async_job(job_id).await?.ok_or_else(|| {
ApiError::not_found_with_context(
format!("async job {} was not found", job_id.as_str()),
json!({ "job_id": job_id.as_str() }),
)
})?;
Ok(async_job_detail_view(updated))
}
#[instrument(skip(self))]
pub async fn get_async_job_result(
&self,
workspace_id: &WorkspaceId,
job_id: &crank_core::AsyncJobId,
) -> Result<Value, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
let job = self.registry.get_async_job(job_id).await?.ok_or_else(|| {
ApiError::not_found_with_context(
format!("async job {} was not found", job_id.as_str()),
json!({ "job_id": job_id.as_str() }),
)
})?;
ensure_async_job_workspace(&job, workspace_id)?;
let job = if matches!(job.status, JobStatus::Completed) {
job
} else {
self.touch_async_job_poll_if_ready(workspace_id, job)
.await?
};
match job.status {
JobStatus::Completed => Ok(job.result.unwrap_or(Value::Null)),
JobStatus::Failed => Err(ApiError::conflict_with_context(
"async job failed",
json!({
"job_id": job_id.as_str(),
"status": "failed",
"error": job.error,
}),
)),
JobStatus::Cancelled => Err(ApiError::conflict_with_context(
"async job was cancelled",
json!({
"job_id": job_id.as_str(),
"status": "cancelled",
}),
)),
_ => Err(ApiError::conflict_with_context(
"async job result is not ready",
json!({
"job_id": job_id.as_str(),
"status": match job.status {
JobStatus::Created => "created",
JobStatus::Running => "running",
JobStatus::Completed => "completed",
JobStatus::Failed => "failed",
JobStatus::Cancelled => "cancelled",
JobStatus::Expired => "expired",
},
}),
)),
}
}
async fn touch_stream_session_poll_if_ready(
&self,
workspace_id: &WorkspaceId,
session: StreamSession,
) -> Result<StreamSession, ApiError> {
let poll_interval_ms = self
.streaming_poll_interval_ms(workspace_id, &session.operation_id)
.await?;
let now = OffsetDateTime::now_utc();
let remaining_delay_ms = session.remaining_poll_delay_ms(now, poll_interval_ms);
if remaining_delay_ms > 0 {
return Err(ApiError::rate_limited_with_context(
"stream session poll rate limit exceeded",
json!({
"session_id": session.id.as_str(),
"poll_after_ms": remaining_delay_ms,
}),
));
}
self.registry
.touch_stream_session_poll(&session.id, &now)
.await
.map_err(ApiError::from)
}
async fn touch_async_job_poll_if_ready(
&self,
workspace_id: &WorkspaceId,
job: AsyncJobHandle,
) -> Result<AsyncJobHandle, ApiError> {
let poll_interval_ms = self
.streaming_poll_interval_ms(workspace_id, &job.operation_id)
.await?;
let now = OffsetDateTime::now_utc();
let remaining_delay_ms = job.remaining_poll_delay_ms(now, poll_interval_ms);
if remaining_delay_ms > 0 {
return Err(ApiError::rate_limited_with_context(
"async job poll rate limit exceeded",
json!({
"job_id": job.id.as_str(),
"poll_after_ms": remaining_delay_ms,
}),
));
}
self.registry
.touch_async_job_poll(&job.id, &now)
.await
.map_err(ApiError::from)
}
async fn streaming_poll_interval_ms(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
) -> Result<u64, ApiError> {
let summary = self
.registry
.get_operation_summary(workspace_id, operation_id)
.await?
.ok_or_else(|| {
ApiError::not_found_with_context(
format!("operation {} was not found", operation_id.as_str()),
json!({ "operation_id": operation_id.as_str() }),
)
})?;
let version = self
.registry
.get_operation_version(workspace_id, operation_id, summary.current_draft_version)
.await?
.ok_or_else(|| {
ApiError::not_found_with_context(
format!(
"operation version {} for {} was not found",
summary.current_draft_version,
operation_id.as_str()
),
json!({
"operation_id": operation_id.as_str(),
"version": summary.current_draft_version,
}),
)
})?;
Ok(version
.snapshot
.execution_config
.streaming
.as_ref()
.and_then(|streaming| streaming.poll_interval_ms)
.unwrap_or(1_000))
}
pub async fn list_operations(
&self,
workspace_id: &WorkspaceId,
@@ -3557,6 +3168,7 @@ impl AdminService {
fn validate_operation_payload(&self, payload: &OperationPayload) -> Result<(), ApiError> {
self.validate_operation_capabilities(payload.protocol, payload.security_level)?;
validate_protocol_target(payload.protocol, &payload.target)?;
validate_streaming_policy(&payload.execution_config)?;
validate_response_cache_policy(&payload.target, &payload.execution_config)?;
payload.input_mapping.validate_paths()?;
payload.output_mapping.validate_paths()?;
@@ -3566,6 +3178,7 @@ impl AdminService {
fn validate_registry_operation(&self, operation: &RegistryOperation) -> Result<(), ApiError> {
self.validate_operation_capabilities(operation.protocol, operation.security_level)?;
validate_protocol_target(operation.protocol, &operation.target)?;
validate_streaming_policy(&operation.execution_config)?;
validate_response_cache_policy(&operation.target, &operation.execution_config)?;
operation.input_mapping.validate_paths()?;
operation.output_mapping.validate_paths()?;
@@ -4498,91 +4111,6 @@ fn protocol_capability_view(protocol: Protocol, edition: ProductEdition) -> Prot
}
}
fn stream_session_summary_view(session: StreamSession) -> StreamSessionSummaryView {
StreamSessionSummaryView {
id: session.id.as_str().to_owned(),
operation_id: session.operation_id.as_str().to_owned(),
agent_id: session.agent_id.map(|value| value.as_str().to_owned()),
protocol: session.protocol,
mode: session.mode,
status: session.status,
expires_at: session.expires_at,
last_poll_at: session.last_poll_at,
created_at: session.created_at,
closed_at: session.closed_at,
}
}
fn stream_session_detail_view(session: StreamSession) -> StreamSessionDetailView {
StreamSessionDetailView {
cursor: session.cursor.clone(),
summary: stream_session_summary_view(session),
}
}
fn async_job_summary_view(job: AsyncJobHandle) -> AsyncJobSummaryView {
AsyncJobSummaryView {
id: job.id.as_str().to_owned(),
operation_id: job.operation_id.as_str().to_owned(),
agent_id: job.agent_id.map(|value| value.as_str().to_owned()),
status: job.status,
progress: job.progress,
created_at: job.created_at,
updated_at: job.updated_at,
finished_at: job.finished_at,
expires_at: job.expires_at,
}
}
fn async_job_detail_view(job: AsyncJobHandle) -> AsyncJobDetailView {
AsyncJobDetailView {
error: job.error.clone(),
summary: async_job_summary_view(job),
}
}
fn ensure_stream_session_workspace(
session: &StreamSession,
workspace_id: &WorkspaceId,
) -> Result<(), ApiError> {
if &session.workspace_id == workspace_id {
return Ok(());
}
Err(ApiError::not_found_with_context(
format!(
"stream session {} was not found in workspace {}",
session.id.as_str(),
workspace_id.as_str()
),
json!({
"session_id": session.id.as_str(),
"workspace_id": workspace_id.as_str(),
}),
))
}
fn ensure_async_job_workspace(
job: &AsyncJobHandle,
workspace_id: &WorkspaceId,
) -> Result<(), ApiError> {
if &job.workspace_id == workspace_id {
return Ok(());
}
Err(ApiError::not_found_with_context(
format!(
"async job {} was not found in workspace {}",
job.id.as_str(),
workspace_id.as_str()
),
json!({
"job_id": job.id.as_str(),
"workspace_id": workspace_id.as_str(),
}),
))
}
fn usage_window(period: UsagePeriod) -> Result<(UsagePeriod, String, UsageBucket), ApiError> {
let now = OffsetDateTime::now_utc();
let (start, bucket) = match period {
@@ -4710,6 +4238,22 @@ fn agent_mcp_endpoint(workspace_slug: &str, agent_slug: &str) -> String {
format!("/mcp/v1/{workspace_slug}/{agent_slug}")
}
fn validate_streaming_policy(
execution_config: &crank_core::ExecutionConfig,
) -> Result<(), ApiError> {
if execution_config.streaming.is_some() {
return Err(ApiError::validation_with_context(
"streaming execution is not supported in Community".to_owned(),
json!({
"field": "execution_config.streaming",
"edition": "community",
}),
));
}
Ok(())
}
fn validate_response_cache_policy(
target: &Target,
execution_config: &crank_core::ExecutionConfig,