Expire stale approval requests
This commit is contained in:
@@ -685,6 +685,191 @@ async fn approval_key_lists_and_decides_pending_requests() {
|
||||
assert_eq!(logs.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn approval_key_denies_without_executing_upstream() {
|
||||
let registry = test_registry().await;
|
||||
let upstream_base_url = spawn_upstream_server().await;
|
||||
let operation = test_operation(&upstream_base_url, "crm_human_deny");
|
||||
registry
|
||||
.create_operation(&test_workspace_id(), &operation, Some("alice"))
|
||||
.await
|
||||
.unwrap();
|
||||
publish_agent_with_bindings(
|
||||
®istry,
|
||||
"sales-human-deny",
|
||||
vec![binding_for_operation(&operation)],
|
||||
)
|
||||
.await;
|
||||
let approval = ApprovalRequest {
|
||||
id: ApprovalRequestId::new("approval_mcp_deny_01"),
|
||||
workspace_id: test_workspace_id(),
|
||||
agent_id: test_agent_id("sales-human-deny"),
|
||||
operation_id: operation.id.clone(),
|
||||
operation_version: 1,
|
||||
status: ApprovalRequestStatus::Pending,
|
||||
risk_level: OperationApprovalRiskLevel::Dangerous,
|
||||
confirmation_title: "Подтвердите создание лида".to_owned(),
|
||||
confirmation_body: "Проверьте email перед отправкой в CRM.".to_owned(),
|
||||
request_payload: json!({"email": "deny@example.com"}),
|
||||
response_payload: None,
|
||||
created_at: OffsetDateTime::now_utc(),
|
||||
expires_at: OffsetDateTime::now_utc() + time::Duration::minutes(5),
|
||||
decided_at: None,
|
||||
decided_by_key_id: None,
|
||||
decision_note: None,
|
||||
};
|
||||
registry
|
||||
.create_approval_request(CreateApprovalRequest {
|
||||
approval: &approval,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let approval_key =
|
||||
create_approval_platform_api_key(®istry, "sales-human-deny", "approval-deny-http").await;
|
||||
let base_url = spawn_mcp_server(build_test_app(
|
||||
registry.clone(),
|
||||
Duration::from_millis(0),
|
||||
Some("https://crank.example.com".to_owned()),
|
||||
))
|
||||
.await;
|
||||
let client = reqwest::Client::new();
|
||||
let deny_url = format!(
|
||||
"{}/approvals/{}/deny",
|
||||
agent_mcp_url(&base_url, "sales-human-deny"),
|
||||
approval.id
|
||||
);
|
||||
|
||||
let denied = client
|
||||
.post(&deny_url)
|
||||
.header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
|
||||
.json(&json!({ "approve": "no", "note": "rejected by test" }))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(denied.status(), reqwest::StatusCode::OK);
|
||||
let denied_body = denied.json::<Value>().await.unwrap();
|
||||
assert_eq!(
|
||||
denied_body["approval"]["status"],
|
||||
Value::String("denied".to_owned())
|
||||
);
|
||||
|
||||
let repeated_deny = client
|
||||
.post(&deny_url)
|
||||
.header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
|
||||
.json(&json!({ "approve": "no", "note": "duplicate rejection" }))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(repeated_deny.status(), reqwest::StatusCode::OK);
|
||||
let repeated_body = repeated_deny.json::<Value>().await.unwrap();
|
||||
assert_eq!(
|
||||
repeated_body["approval"]["status"],
|
||||
Value::String("denied".to_owned())
|
||||
);
|
||||
|
||||
let logs = registry
|
||||
.list_invocation_logs(ListInvocationLogsQuery {
|
||||
workspace_id: &test_workspace_id(),
|
||||
level: None,
|
||||
search_text: None,
|
||||
source: Some(InvocationSource::AgentToolCall),
|
||||
operation_id: Some(&operation.id),
|
||||
agent_id: Some(&test_agent_id("sales-human-deny")),
|
||||
created_after: None,
|
||||
limit: 10,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(logs.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn approval_key_expires_without_executing_upstream() {
|
||||
let registry = test_registry().await;
|
||||
let upstream_base_url = spawn_upstream_server().await;
|
||||
let operation = test_operation(&upstream_base_url, "crm_human_expired");
|
||||
registry
|
||||
.create_operation(&test_workspace_id(), &operation, Some("alice"))
|
||||
.await
|
||||
.unwrap();
|
||||
publish_agent_with_bindings(
|
||||
®istry,
|
||||
"sales-human-expired",
|
||||
vec![binding_for_operation(&operation)],
|
||||
)
|
||||
.await;
|
||||
let approval = ApprovalRequest {
|
||||
id: ApprovalRequestId::new("approval_mcp_expired_01"),
|
||||
workspace_id: test_workspace_id(),
|
||||
agent_id: test_agent_id("sales-human-expired"),
|
||||
operation_id: operation.id.clone(),
|
||||
operation_version: 1,
|
||||
status: ApprovalRequestStatus::Pending,
|
||||
risk_level: OperationApprovalRiskLevel::Dangerous,
|
||||
confirmation_title: "Подтвердите создание лида".to_owned(),
|
||||
confirmation_body: "Проверьте email перед отправкой в CRM.".to_owned(),
|
||||
request_payload: json!({"email": "expired@example.com"}),
|
||||
response_payload: None,
|
||||
created_at: OffsetDateTime::now_utc() - time::Duration::minutes(10),
|
||||
expires_at: OffsetDateTime::now_utc() - time::Duration::minutes(5),
|
||||
decided_at: None,
|
||||
decided_by_key_id: None,
|
||||
decision_note: None,
|
||||
};
|
||||
registry
|
||||
.create_approval_request(CreateApprovalRequest {
|
||||
approval: &approval,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let approval_key =
|
||||
create_approval_platform_api_key(®istry, "sales-human-expired", "approval-expired-http")
|
||||
.await;
|
||||
let base_url = spawn_mcp_server(build_test_app(
|
||||
registry.clone(),
|
||||
Duration::from_millis(0),
|
||||
Some("https://crank.example.com".to_owned()),
|
||||
))
|
||||
.await;
|
||||
let client = reqwest::Client::new();
|
||||
let approve_url = format!(
|
||||
"{}/approvals/{}/approve",
|
||||
agent_mcp_url(&base_url, "sales-human-expired"),
|
||||
approval.id
|
||||
);
|
||||
|
||||
let expired = client
|
||||
.post(&approve_url)
|
||||
.header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
|
||||
.json(&json!({ "approve": "yes", "note": "too late" }))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(expired.status(), reqwest::StatusCode::OK);
|
||||
let expired_body = expired.json::<Value>().await.unwrap();
|
||||
assert_eq!(
|
||||
expired_body["approval"]["status"],
|
||||
Value::String("expired".to_owned())
|
||||
);
|
||||
|
||||
let logs = registry
|
||||
.list_invocation_logs(ListInvocationLogsQuery {
|
||||
workspace_id: &test_workspace_id(),
|
||||
level: None,
|
||||
search_text: None,
|
||||
source: Some(InvocationSource::AgentToolCall),
|
||||
operation_id: Some(&operation.id),
|
||||
agent_id: Some(&test_agent_id("sales-human-expired")),
|
||||
created_after: None,
|
||||
limit: 10,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(logs.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_call_with_approval_policy_creates_pending_request() {
|
||||
let registry = test_registry().await;
|
||||
|
||||
@@ -19,7 +19,8 @@ use crank_core::{
|
||||
};
|
||||
use crank_registry::{
|
||||
ApprovalRequestRecord, CreateApprovalRequest, CreateInvocationLogRequest,
|
||||
DecideApprovalRequest, FinishApprovalRequest, PostgresRegistry, PublishedAgentTool,
|
||||
DecideApprovalRequest, ExpireApprovalRequest, FinishApprovalRequest, PostgresRegistry,
|
||||
PublishedAgentTool,
|
||||
};
|
||||
use crank_runtime::{
|
||||
RequestRateLimiter, ResolvedAuth, RuntimeError, RuntimeExecutionRequest, RuntimeExecutor,
|
||||
@@ -242,15 +243,7 @@ async fn get_approval_request(
|
||||
};
|
||||
let approval_id = ApprovalRequestId::new(path.approval_id);
|
||||
|
||||
match state
|
||||
.registry
|
||||
.get_approval_request_for_agent(&key.api_key.workspace_id, agent_id, &approval_id)
|
||||
.await
|
||||
{
|
||||
Ok(Some(record)) => Json(json!(record)).into_response(),
|
||||
Ok(None) => StatusCode::NOT_FOUND.into_response(),
|
||||
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
}
|
||||
approval_record_response(&state, &key.api_key.workspace_id, agent_id, &approval_id).await
|
||||
}
|
||||
|
||||
async fn deny_request(
|
||||
@@ -327,14 +320,14 @@ async fn decide_approval_request(
|
||||
}
|
||||
Ok(Some(record)) => Json(json!(record)).into_response(),
|
||||
Ok(None) => {
|
||||
existing_decision_response(&state, &key.api_key.workspace_id, agent_id, &approval_id)
|
||||
terminal_decision_response(&state, &key.api_key.workspace_id, agent_id, &approval_id)
|
||||
.await
|
||||
}
|
||||
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn existing_decision_response(
|
||||
async fn approval_record_response(
|
||||
state: &Arc<AppState>,
|
||||
workspace_id: &crank_core::WorkspaceId,
|
||||
agent_id: &crank_core::AgentId,
|
||||
@@ -345,6 +338,35 @@ async fn existing_decision_response(
|
||||
.get_approval_request_for_agent(workspace_id, agent_id, approval_id)
|
||||
.await
|
||||
{
|
||||
Ok(Some(record))
|
||||
if record.approval.status == ApprovalRequestStatus::Pending
|
||||
&& record.approval.expires_at <= OffsetDateTime::now_utc() =>
|
||||
{
|
||||
expire_approval_response(state, workspace_id, agent_id, approval_id).await
|
||||
}
|
||||
Ok(Some(record)) => Json(json!(record)).into_response(),
|
||||
Ok(None) => StatusCode::NOT_FOUND.into_response(),
|
||||
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn terminal_decision_response(
|
||||
state: &Arc<AppState>,
|
||||
workspace_id: &crank_core::WorkspaceId,
|
||||
agent_id: &crank_core::AgentId,
|
||||
approval_id: &ApprovalRequestId,
|
||||
) -> Response {
|
||||
match state
|
||||
.registry
|
||||
.get_approval_request_for_agent(workspace_id, agent_id, approval_id)
|
||||
.await
|
||||
{
|
||||
Ok(Some(record))
|
||||
if record.approval.status == ApprovalRequestStatus::Pending
|
||||
&& record.approval.expires_at <= OffsetDateTime::now_utc() =>
|
||||
{
|
||||
expire_approval_response(state, workspace_id, agent_id, approval_id).await
|
||||
}
|
||||
Ok(Some(record))
|
||||
if matches!(
|
||||
record.approval.status,
|
||||
@@ -362,6 +384,28 @@ async fn existing_decision_response(
|
||||
}
|
||||
}
|
||||
|
||||
async fn expire_approval_response(
|
||||
state: &Arc<AppState>,
|
||||
workspace_id: &crank_core::WorkspaceId,
|
||||
agent_id: &crank_core::AgentId,
|
||||
approval_id: &ApprovalRequestId,
|
||||
) -> Response {
|
||||
match state
|
||||
.registry
|
||||
.expire_approval_request(ExpireApprovalRequest {
|
||||
workspace_id,
|
||||
agent_id,
|
||||
approval_id,
|
||||
expired_at: OffsetDateTime::now_utc(),
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(Some(record)) => Json(json!(record)).into_response(),
|
||||
Ok(None) => StatusCode::CONFLICT.into_response(),
|
||||
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_approved_request(
|
||||
state: &Arc<AppState>,
|
||||
path: &AgentRoutePath,
|
||||
|
||||
@@ -27,10 +27,11 @@ pub mod requests {
|
||||
CreateImportJobRequest, CreateInvitationRequest, CreateInvocationLogRequest,
|
||||
CreatePlatformApiKeyRequest, CreateSecretRequest, CreateVersionRequest,
|
||||
CreateWorkspaceRequest, CreateYamlImportJobRequest, DecideApprovalRequest,
|
||||
FinishApprovalRequest, FinishImportJobRequest, ListInvocationLogsQuery,
|
||||
PublishAgentRequest, PublishRequest, RotateSecretRequest, SaveAgentBindingsRequest,
|
||||
SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest,
|
||||
SaveWorkspaceUpstreamRequest, UpdateWorkspaceRequest, UsageQuery,
|
||||
ExpireApprovalRequest, FinishApprovalRequest, FinishImportJobRequest,
|
||||
ListInvocationLogsQuery, PublishAgentRequest, PublishRequest, RotateSecretRequest,
|
||||
SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
|
||||
SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, UpdateWorkspaceRequest,
|
||||
UsageQuery,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -45,8 +46,8 @@ pub use model::{
|
||||
CreateImportJobRequest, CreateInvitationRequest, CreateInvocationLogRequest,
|
||||
CreatePlatformApiKeyRequest, CreateSecretRequest, CreateVersionRequest, CreateWorkspaceRequest,
|
||||
CreateYamlImportJobRequest, DecideApprovalRequest, DescriptorKind, DescriptorMetadata,
|
||||
FinishApprovalRequest, FinishImportJobRequest, ImportJob, ImportJobId, ImportJobKind,
|
||||
ImportJobStatus, InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery,
|
||||
ExpireApprovalRequest, FinishApprovalRequest, FinishImportJobRequest, ImportJob, ImportJobId,
|
||||
ImportJobKind, ImportJobStatus, InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery,
|
||||
MembershipRecord, OperationAgentRef, OperationSampleMetadata, OperationSummary,
|
||||
OperationUsageSummary, OperationVersionRecord, Page, PlatformApiKeyRecord, PublishAgentRequest,
|
||||
PublishRequest, PublishedAgentTool, RegistryOperation, RotateSecretRequest, SampleKind,
|
||||
|
||||
@@ -546,6 +546,14 @@ pub struct FinishApprovalRequest<'a> {
|
||||
pub decision_note: Option<&'a str>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ExpireApprovalRequest<'a> {
|
||||
pub workspace_id: &'a WorkspaceId,
|
||||
pub agent_id: &'a AgentId,
|
||||
pub approval_id: &'a ApprovalRequestId,
|
||||
pub expired_at: OffsetDateTime,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct CreateSecretRequest<'a> {
|
||||
pub secret: &'a Secret,
|
||||
|
||||
@@ -230,6 +230,46 @@ impl PostgresRegistry {
|
||||
|
||||
row.map(map_approval_request_row).transpose()
|
||||
}
|
||||
|
||||
pub async fn expire_approval_request(
|
||||
&self,
|
||||
request: ExpireApprovalRequest<'_>,
|
||||
) -> Result<Option<ApprovalRequestRecord>, RegistryError> {
|
||||
let row = sqlx::query(
|
||||
"update approval_requests
|
||||
set status = 'expired'
|
||||
where workspace_id = $1
|
||||
and agent_id = $2
|
||||
and id = $3
|
||||
and status = 'pending'
|
||||
and expires_at <= $4::timestamptz
|
||||
returning
|
||||
id,
|
||||
workspace_id,
|
||||
agent_id,
|
||||
operation_id,
|
||||
operation_version,
|
||||
status,
|
||||
risk_level,
|
||||
confirmation_title,
|
||||
confirmation_body,
|
||||
request_payload_json,
|
||||
response_payload_json,
|
||||
created_at,
|
||||
expires_at,
|
||||
decided_at,
|
||||
decided_by_key_id,
|
||||
decision_note",
|
||||
)
|
||||
.bind(request.workspace_id.as_str())
|
||||
.bind(request.agent_id.as_str())
|
||||
.bind(request.approval_id.as_str())
|
||||
.bind(request.expired_at)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
|
||||
row.map(map_approval_request_row).transpose()
|
||||
}
|
||||
}
|
||||
|
||||
fn map_approval_request_row(row: PgRow) -> Result<ApprovalRequestRecord, RegistryError> {
|
||||
|
||||
@@ -35,15 +35,15 @@ use crate::{
|
||||
CreateImportJobRequest, CreateInvitationRequest, CreateInvocationLogRequest,
|
||||
CreatePlatformApiKeyRequest, CreateSecretRequest, CreateVersionRequest,
|
||||
CreateWorkspaceRequest, CreateYamlImportJobRequest, DecideApprovalRequest,
|
||||
DescriptorMetadata, FinishApprovalRequest, FinishImportJobRequest, ImportJob, ImportJobId,
|
||||
InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord,
|
||||
OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary,
|
||||
OperationVersionRecord, PlatformApiKeyRecord, PublishAgentRequest, PublishRequest,
|
||||
PublishedAgentTool, RegistryOperation, RotateSecretRequest, SaveAgentBindingsRequest,
|
||||
SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest,
|
||||
SaveWorkspaceUpstreamRequest, SecretRecord, SecretVersionRecord, SessionRecord,
|
||||
UpdateWorkspaceRequest, UsageAgentBreakdown, UsageOperationBreakdown, UsageQuery,
|
||||
UsageRollupRecord, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord,
|
||||
DescriptorMetadata, ExpireApprovalRequest, FinishApprovalRequest, FinishImportJobRequest,
|
||||
ImportJob, ImportJobId, InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery,
|
||||
MembershipRecord, OperationAgentRef, OperationSampleMetadata, OperationSummary,
|
||||
OperationUsageSummary, OperationVersionRecord, PlatformApiKeyRecord, PublishAgentRequest,
|
||||
PublishRequest, PublishedAgentTool, RegistryOperation, RotateSecretRequest,
|
||||
SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
|
||||
SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, SecretRecord, SecretVersionRecord,
|
||||
SessionRecord, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageOperationBreakdown,
|
||||
UsageQuery, UsageRollupRecord, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord,
|
||||
WorkspaceRecord, WorkspaceUpstream, YamlImportJob, YamlImportJobCompletion,
|
||||
YamlImportJobId, YamlImportJobStatus,
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user