Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(sagent): add agent task domain and gateway dispatch

ZhenYi committed
parent 2cfa36b commit d57efed Browse files

17 files changed

+1040 −0

File tree 17
Modified lib/sagent/src/gateway/mod.rs View lib/sagent/src/gateway/mod.rs
@@ -6,10 +6,13 @@
66use agent_types::ids::{RunId, SessionId, TenantId};
77use agent_types::session::{MessageAccepted, MessageDto, SessionDto, SessionTokenDto};
88use config::sagent::SagentConfig;
9use models::sagent::agent::AgentState;
910use smol_str::SmolStr;
1011use time::OffsetDateTime;
1112use uuid::Uuid;
1213
14use agent_types::state::RunState;
15
1316use crate::errors::SagentError;
1417
1518pub const INVOKE_SCOPES: [&str; 6] = [
@@ -215,6 +218,20 @@
215218 }
216219}
217220
221pub fn map_run_state(state: RunState) -> AgentState {
222 match state {
223 RunState::Pending => AgentState::Pending,
224 RunState::Planning => AgentState::Planning,
225 RunState::Executing => AgentState::Executing,
226 RunState::AwaitingApproval | RunState::Verifying | RunState::Deferred => {
227 AgentState::Waiting
228 }
229 RunState::Completed => AgentState::Succeeded,
230 RunState::Failed => AgentState::Failed,
231 RunState::Cancelled => AgentState::Canceled,
232 }
233}
234
218235impl From<SdkError> for SagentError {
219236 fn from(error: SdkError) -> Self {
220237 match error {
Added lib/sagent/src/handlers/tasks/cancel_task.rs View lib/sagent/src/handlers/tasks/cancel_task.rs
@@ -0,0 +1,37 @@
1use axum::extract::{Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentDto, AgentPath};
9
10#[utoipa::path(
11 post,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/cancel",
13 tag = "sagent",
14 description = "取消任务当前 run，并把任务状态置为已取消。",
15 params(AgentPath),
16 responses(
17 (status = 200, description = "cancel task", body = AppResponseData<AgentDto>),
18 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
19 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
20 (status = 404, description = "not found", body = AppResponseData<NoData>),
21 (status = 409, description = "task has no run", body = AppResponseData<NoData>),
22 (status = 429, description = "gateway rate limited", body = AppResponseData<NoData>),
23 (status = 502, description = "gateway rejected the request", body = AppResponseData<NoData>),
24 (status = 503, description = "gateway unavailable", body = AppResponseData<NoData>)
25 )
26)]
27pub async fn cancel_task(
28 State(service): State<SagentService>,
29 session: Session,
30 Path(path): Path<AgentPath>,
31) -> Result<AppResponseData<AgentDto>, SagentApiError> {
32 let uid = require_uid(&session)?;
33 let dto = service
34 .cancel_task(uid, &path.space_name, path.collection_id, &path.task_id)
35 .await?;
36 Ok(AppResponseData::ok(dto))
37}
Added lib/sagent/src/handlers/tasks/close_task.rs View lib/sagent/src/handlers/tasks/close_task.rs
@@ -0,0 +1,34 @@
1use axum::extract::{Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentDto, AgentPath};
9
10#[utoipa::path(
11 post,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/close",
13 tag = "sagent",
14 description = "关闭任务，记录关闭时间。",
15 params(AgentPath),
16 responses(
17 (status = 200, description = "close task", body = AppResponseData<AgentDto>),
18 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
19 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
20 (status = 404, description = "not found", body = AppResponseData<NoData>),
21 (status = 409, description = "task is already closed", body = AppResponseData<NoData>)
22 )
23)]
24pub async fn close_task(
25 State(service): State<SagentService>,
26 session: Session,
27 Path(path): Path<AgentPath>,
28) -> Result<AppResponseData<AgentDto>, SagentApiError> {
29 let uid = require_uid(&session)?;
30 let dto = service
31 .close_task(uid, &path.space_name, path.collection_id, &path.task_id)
32 .await?;
33 Ok(AppResponseData::ok(dto))
34}
Added lib/sagent/src/handlers/tasks/create_task.rs View lib/sagent/src/handlers/tasks/create_task.rs
@@ -0,0 +1,36 @@
1use axum::extract::{Json, Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentDto, CollectionRef, CreateAgentCmd};
9
10#[utoipa::path(
11 post,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks",
13 tag = "sagent",
14 description = "在智能体集下创建任务，自动分配集合内编号，初始状态为草稿。",
15 params(CollectionRef),
16 request_body = CreateAgentCmd,
17 responses(
18 (status = 200, description = "create task", body = AppResponseData<AgentDto>),
19 (status = 400, description = "bad request", body = AppResponseData<NoData>),
20 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
21 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
22 (status = 404, description = "not found", body = AppResponseData<NoData>)
23 )
24)]
25pub async fn create_task(
26 State(service): State<SagentService>,
27 session: Session,
28 Path(path): Path<CollectionRef>,
29 Json(cmd): Json<CreateAgentCmd>,
30) -> Result<AppResponseData<AgentDto>, SagentApiError> {
31 let uid = require_uid(&session)?;
32 let dto = service
33 .create_task(uid, &path.space_name, path.collection_id, cmd)
34 .await?;
35 Ok(AppResponseData::ok(dto))
36}
Added lib/sagent/src/handlers/tasks/delete_task.rs View lib/sagent/src/handlers/tasks/delete_task.rs
@@ -0,0 +1,33 @@
1use axum::extract::{Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::AgentPath;
9
10#[utoipa::path(
11 delete,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}",
13 tag = "sagent",
14 description = "删除任务并级联清除其消息、关注与活动流。",
15 params(AgentPath),
16 responses(
17 (status = 200, description = "delete task", body = AppResponseData<NoData>),
18 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
19 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
20 (status = 404, description = "not found", body = AppResponseData<NoData>)
21 )
22)]
23pub async fn delete_task(
24 State(service): State<SagentService>,
25 session: Session,
26 Path(path): Path<AgentPath>,
27) -> Result<AppResponseData<NoData>, SagentApiError> {
28 let uid = require_uid(&session)?;
29 service
30 .delete_task(uid, &path.space_name, path.collection_id, &path.task_id)
31 .await?;
32 Ok(AppResponseData::ok(NoData))
33}
Added lib/sagent/src/handlers/tasks/get_task.rs View lib/sagent/src/handlers/tasks/get_task.rs
@@ -0,0 +1,33 @@
1use axum::extract::{Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentDto, AgentPath};
9
10#[utoipa::path(
11 get,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}",
13 tag = "sagent",
14 description = "读取任务详情，task_id 支持 UUID 或集合内编号。",
15 params(AgentPath),
16 responses(
17 (status = 200, description = "get task", body = AppResponseData<AgentDto>),
18 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
19 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
20 (status = 404, description = "not found", body = AppResponseData<NoData>)
21 )
22)]
23pub async fn get_task(
24 State(service): State<SagentService>,
25 session: Session,
26 Path(path): Path<AgentPath>,
27) -> Result<AppResponseData<AgentDto>, SagentApiError> {
28 let uid = require_uid(&session)?;
29 let dto = service
30 .get_task(uid, &path.space_name, path.collection_id, &path.task_id)
31 .await?;
32 Ok(AppResponseData::ok(dto))
33}
Added lib/sagent/src/handlers/tasks/list_tasks.rs View lib/sagent/src/handlers/tasks/list_tasks.rs
@@ -0,0 +1,34 @@
1use axum::extract::{Path, Query, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentPage, CollectionRef, ListAgentQuery};
9
10#[utoipa::path(
11 get,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks",
13 tag = "sagent",
14 description = "分页列出集合内任务，可按关键字、状态、优先级、作者过滤并指定排序。",
15 params(CollectionRef, ListAgentQuery),
16 responses(
17 (status = 200, description = "list tasks", body = AppResponseData<AgentPage>),
18 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
19 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
20 (status = 404, description = "not found", body = AppResponseData<NoData>)
21 )
22)]
23pub async fn list_tasks(
24 State(service): State<SagentService>,
25 session: Session,
26 Path(path): Path<CollectionRef>,
27 Query(query): Query<ListAgentQuery>,
28) -> Result<AppResponseData<AgentPage>, SagentApiError> {
29 let uid = require_uid(&session)?;
30 let page = service
31 .list_tasks(uid, &path.space_name, path.collection_id, query)
32 .await?;
33 Ok(AppResponseData::ok(page))
34}
Added lib/sagent/src/handlers/tasks/mod.rs View lib/sagent/src/handlers/tasks/mod.rs
@@ -0,0 +1,9 @@
1pub mod cancel_task;
2pub mod close_task;
3pub mod create_task;
4pub mod delete_task;
5pub mod get_task;
6pub mod list_tasks;
7pub mod reopen_task;
8pub mod start_task;
9pub mod update_task;
Added lib/sagent/src/handlers/tasks/reopen_task.rs View lib/sagent/src/handlers/tasks/reopen_task.rs
@@ -0,0 +1,34 @@
1use axum::extract::{Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentDto, AgentPath};
9
10#[utoipa::path(
11 post,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/reopen",
13 tag = "sagent",
14 description = "重开已关闭的任务，回到草稿状态。",
15 params(AgentPath),
16 responses(
17 (status = 200, description = "reopen task", body = AppResponseData<AgentDto>),
18 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
19 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
20 (status = 404, description = "not found", body = AppResponseData<NoData>),
21 (status = 409, description = "task is not closed", body = AppResponseData<NoData>)
22 )
23)]
24pub async fn reopen_task(
25 State(service): State<SagentService>,
26 session: Session,
27 Path(path): Path<AgentPath>,
28) -> Result<AppResponseData<AgentDto>, SagentApiError> {
29 let uid = require_uid(&session)?;
30 let dto = service
31 .reopen_task(uid, &path.space_name, path.collection_id, &path.task_id)
32 .await?;
33 Ok(AppResponseData::ok(dto))
34}
Added lib/sagent/src/handlers/tasks/start_task.rs View lib/sagent/src/handlers/tasks/start_task.rs
@@ -0,0 +1,38 @@
1use axum::extract::{Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentDto, AgentPath};
9
10#[utoipa::path(
11 post,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/start",
13 tag = "sagent",
14 description = "启动任务：创建或复用网关会话、签发会话内调用、下发正文并记录首个 run。",
15 params(AgentPath),
16 responses(
17 (status = 200, description = "start task", body = AppResponseData<AgentDto>),
18 (status = 400, description = "bad request", body = AppResponseData<NoData>),
19 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
20 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
21 (status = 404, description = "not found", body = AppResponseData<NoData>),
22 (status = 409, description = "task already started", body = AppResponseData<NoData>),
23 (status = 429, description = "gateway rate limited", body = AppResponseData<NoData>),
24 (status = 502, description = "gateway rejected the request", body = AppResponseData<NoData>),
25 (status = 503, description = "gateway unavailable", body = AppResponseData<NoData>)
26 )
27)]
28pub async fn start_task(
29 State(service): State<SagentService>,
30 session: Session,
31 Path(path): Path<AgentPath>,
32) -> Result<AppResponseData<AgentDto>, SagentApiError> {
33 let uid = require_uid(&session)?;
34 let dto = service
35 .start_task(uid, &path.space_name, path.collection_id, &path.task_id)
36 .await?;
37 Ok(AppResponseData::ok(dto))
38}
Added lib/sagent/src/handlers/tasks/update_task.rs View lib/sagent/src/handlers/tasks/update_task.rs
@@ -0,0 +1,43 @@
1use axum::extract::{Json, Path, State};
2use export::rest::{AppResponseData, NoData};
3use session::Session;
4use session::share::session::require_uid;
5
6use crate::errors::SagentApiError;
7use crate::services::service::SagentService;
8use crate::services::tasks::{AgentDto, AgentPath, UpdateAgentCmd};
9
10#[utoipa::path(
11 patch,
12 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}",
13 tag = "sagent",
14 description = "更新任务字段；正文仅在草稿状态可改，首次运行后冻结。",
15 params(AgentPath),
16 request_body = UpdateAgentCmd,
17 responses(
18 (status = 200, description = "update task", body = AppResponseData<AgentDto>),
19 (status = 400, description = "bad request", body = AppResponseData<NoData>),
20 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
21 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
22 (status = 404, description = "not found", body = AppResponseData<NoData>),
23 (status = 409, description = "task body is frozen", body = AppResponseData<NoData>)
24 )
25)]
26pub async fn update_task(
27 State(service): State<SagentService>,
28 session: Session,
29 Path(path): Path<AgentPath>,
30 Json(cmd): Json<UpdateAgentCmd>,
31) -> Result<AppResponseData<AgentDto>, SagentApiError> {
32 let uid = require_uid(&session)?;
33 let dto = service
34 .update_task(
35 uid,
36 &path.space_name,
37 path.collection_id,
38 &path.task_id,
39 cmd,
40 )
41 .await?;
42 Ok(AppResponseData::ok(dto))
43}
Modified lib/sagent/src/handlers/mod.rs View lib/sagent/src/handlers/mod.rs
@@ -1 +1,2 @@
11pub mod collections;
2pub mod tasks;
Modified lib/sagent/src/services/mod.rs View lib/sagent/src/services/mod.rs
@@ -1,2 +1,3 @@
11pub mod collections;
22pub mod service;
3pub mod tasks;
Added lib/sagent/src/services/tasks.rs View lib/sagent/src/services/tasks.rs
@@ -0,0 +1,615 @@
1use models::sagent::agent::{AgentPriority, AgentState};
2use models::sagent::{agent, agent_event_line, agent_message, agent_watch};
3use sea_orm::sea_query::LockType;
4use sea_orm::{
5 ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
6 QuerySelect, Set, TransactionTrait, Unchanged,
7};
8use serde::{Deserialize, Serialize};
9use time::OffsetDateTime;
10use utoipa::{IntoParams, ToSchema};
11use uuid::Uuid;
12
13use crate::errors::SagentError;
14use crate::gateway::{INVOKE_SCOPES, map_run_state};
15use crate::services::service::SagentService;
16use crate::share::authz::{require_task_draft, require_task_write};
17use crate::share::lookup::{space_context, task_context};
18
19#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
20pub struct AgentDto {
21 pub id: Uuid,
22 pub collection_id: Uuid,
23 pub number: i32,
24 pub author_id: Uuid,
25 pub title: String,
26 pub description: Option<String>,
27 pub body: String,
28 pub state: AgentState,
29 pub priority: AgentPriority,
30 pub due_at: Option<OffsetDateTime>,
31 pub session_id: Option<Uuid>,
32 pub last_run_id: Option<Uuid>,
33 pub last_event_seq: i64,
34 pub created_at: OffsetDateTime,
35 pub updated_at: Option<OffsetDateTime>,
36 pub started_at: Option<OffsetDateTime>,
37 pub finished_at: Option<OffsetDateTime>,
38 pub closed_at: Option<OffsetDateTime>,
39}
40
41pub type AgentPage = export::pagination::Page<AgentDto>;
42
43#[derive(Debug, Clone, Deserialize, ToSchema)]
44pub struct CreateAgentCmd {
45 pub title: String,
46 #[serde(default)]
47 pub description: Option<String>,
48 #[serde(default)]
49 pub body: Option<String>,
50 #[serde(default)]
51 pub priority: AgentPriority,
52 #[serde(default)]
53 pub due_at: Option<OffsetDateTime>,
54}
55
56#[derive(Debug, Clone, Deserialize, ToSchema)]
57pub struct UpdateAgentCmd {
58 #[serde(default)]
59 pub title: Option<String>,
60 #[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
61 pub description: Option<Option<String>>,
62 #[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
63 pub body: Option<Option<String>>,
64 #[serde(default)]
65 pub priority: Option<AgentPriority>,
66 #[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
67 pub due_at: Option<Option<OffsetDateTime>>,
68}
69
70#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
71pub struct ListAgentQuery {
72 pub q: Option<String>,
73 pub state: Option<AgentState>,
74 pub priority: Option<AgentPriority>,
75 pub author_id: Option<Uuid>,
76 pub sort: Option<String>,
77 pub limit: Option<u64>,
78 pub offset: Option<u64>,
79}
80
81#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
82pub struct CollectionRef {
83 pub space_name: String,
84 pub collection_id: Uuid,
85}
86
87#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
88pub struct AgentPath {
89 pub space_name: String,
90 pub collection_id: Uuid,
91 pub task_id: String,
92}
93
94impl SagentService {
95 pub(crate) fn agent_dto(task: agent::Model) -> AgentDto {
96 AgentDto {
97 id: task.id,
98 collection_id: task.collection_id,
99 number: task.autoincrement_id,
100 author_id: task.author_id,
101 title: task.title,
102 description: task.description,
103 body: task.body,
104 state: task.state,
105 priority: task.priority,
106 due_at: task.due_at,
107 session_id: task.session_id,
108 last_run_id: task.last_run_id,
109 last_event_seq: task.last_event_seq,
110 created_at: task.created_at,
111 updated_at: task.updated_at,
112 started_at: task.started_at,
113 finished_at: task.finished_at,
114 closed_at: task.closed_at,
115 }
116 }
117
118 async fn set_state(&self, task_id: Uuid, state: AgentState) -> Result<(), SagentError> {
119 let update = agent::ActiveModel {
120 id: Unchanged(task_id),
121 state: Set(state),
122 updated_at: Set(Some(OffsetDateTime::now_utc())),
123 ..Default::default()
124 };
125 update.update(self.db().write()).await?;
126 Ok(())
127 }
128
129 pub async fn create_task(
130 &self,
131 caller: Uuid,
132 space_name: &str,
133 collection_id: Uuid,
134 cmd: CreateAgentCmd,
135 ) -> Result<AgentDto, SagentError> {
136 let space = space_context(self, caller, space_name).await?;
137 let collection =
138 crate::share::lookup::lookup_collection(self, &space, collection_id, caller).await?;
139
140 let title = cmd.title.trim();
141 if title.is_empty() || title.chars().count() > self.config().sagent.max_title_len {
142 return Err(SagentError::BadRequest(
143 "task title is empty or too long".to_string(),
144 ));
145 }
146
147 let body = cmd.body.unwrap_or_default();
148 if body.chars().count() > self.config().sagent.max_body_len {
149 return Err(SagentError::BadRequest("task body is too long".to_string()));
150 }
151
152 let now = OffsetDateTime::now_utc();
153 let task = {
154 let txn = self.db().write().begin().await?;
155 let max_row = agent::Entity::find()
156 .filter(agent::Column::CollectionId.eq(collection.id))
157 .order_by_desc(agent::Column::AutoincrementId)
158 .lock(LockType::Update)
159 .one(&txn)
160 .await?;
161 let number = max_row.map(|row| row.autoincrement_id + 1).unwrap_or(1);
162
163 let row = agent::ActiveModel {
164 id: Set(Uuid::now_v7()),
165 collection_id: Set(collection.id),
166 autoincrement_id: Set(number),
167 author_id: Set(caller),
168 title: Set(title.to_string()),
169 description: Set(cmd.description),
170 body: Set(body),
171 state: Set(AgentState::Draft),
172 priority: Set(cmd.priority),
173 due_at: Set(cmd.due_at),
174 session_id: Set(None),
175 last_run_id: Set(None),
176 last_event_seq: Set(0),
177 created_at: Set(now),
178 updated_at: Set(None),
179 started_at: Set(None),
180 finished_at: Set(None),
181 closed_at: Set(None),
182 }
183 .insert(&txn)
184 .await?;
185
186 self.write_event(
187 &txn,
188 row.id,
189 collection.id,
190 Some(caller),
191 agent_event_line::AgentEventType::Created,
192 None,
193 None,
194 )
195 .await?;
196
197 txn.commit().await?;
198 row
199 };
200
201 Ok(Self::agent_dto(task))
202 }
203
204 pub async fn list_tasks(
205 &self,
206 caller: Uuid,
207 space_name: &str,
208 collection_id: Uuid,
209 query: ListAgentQuery,
210 ) -> Result<AgentPage, SagentError> {
211 let space = space_context(self, caller, space_name).await?;
212 crate::share::lookup::lookup_collection(self, &space, collection_id, caller).await?;
213 let (limit, offset) = self.paginate(query.limit, query.offset);
214
215 let mut select =
216 agent::Entity::find().filter(agent::Column::CollectionId.eq(collection_id));
217 if let Some(search) = query.q {
218 let keyword = search.trim();
219 if !keyword.is_empty() {
220 select = select.filter(agent::Column::Title.contains(keyword));
221 }
222 }
223 if let Some(state) = query.state {
224 select = select.filter(agent::Column::State.eq(state));
225 }
226 if let Some(priority) = query.priority {
227 select = select.filter(agent::Column::Priority.eq(priority));
228 }
229 if let Some(author_id) = query.author_id {
230 select = select.filter(agent::Column::AuthorId.eq(author_id));
231 }
232
233 select = match query.sort.as_deref() {
234 Some("updated") => select.order_by_desc(agent::Column::UpdatedAt),
235 Some("priority") => select
236 .order_by_desc(agent::Column::Priority)
237 .order_by_desc(agent::Column::CreatedAt),
238 Some("number") => select.order_by_asc(agent::Column::AutoincrementId),
239 _ => select.order_by_desc(agent::Column::CreatedAt),
240 };
241
242 let total = select.clone().count(self.db().read()).await? as i64;
243 let rows = select
244 .limit(limit)
245 .offset(offset)
246 .all(self.db().read())
247 .await?;
248
249 Ok(AgentPage {
250 items: rows.into_iter().map(Self::agent_dto).collect(),
251 total,
252 limit,
253 offset,
254 })
255 }
256
257 pub async fn get_task(
258 &self,
259 caller: Uuid,
260 space_name: &str,
261 collection_id: Uuid,
262 task_ref: &str,
263 ) -> Result<AgentDto, SagentError> {
264 let (_, _, task) = task_context(self, caller, space_name, collection_id, task_ref).await?;
265 Ok(Self::agent_dto(task))
266 }
267
268 pub async fn update_task(
269 &self,
270 caller: Uuid,
271 space_name: &str,
272 collection_id: Uuid,
273 task_ref: &str,
274 cmd: UpdateAgentCmd,
275 ) -> Result<AgentDto, SagentError> {
276 let (space, collection, task) =
277 task_context(self, caller, space_name, collection_id, task_ref).await?;
278 require_task_write(self, &space, caller, &task).await?;
279
280 let now = OffsetDateTime::now_utc();
281 let mut update: agent::ActiveModel = task.clone().into();
282
283 if let Some(new_title) = cmd.title {
284 let title = new_title.trim();
285 if title.is_empty() || title.chars().count() > self.config().sagent.max_title_len {
286 return Err(SagentError::BadRequest(
287 "task title is empty or too long".to_string(),
288 ));
289 }
290 update.title = Set(title.to_string());
291 }
292 if let Some(description) = cmd.description {
293 update.description = Set(description);
294 }
295 if let Some(body) = cmd.body {
296 require_task_draft(task.state)?;
297 let body = body.unwrap_or_default();
298 if body.chars().count() > self.config().sagent.max_body_len {
299 return Err(SagentError::BadRequest("task body is too long".to_string()));
300 }
301 update.body = Set(body);
302 }
303 if let Some(priority) = cmd.priority {
304 update.priority = Set(priority);
305 }
306 if let Some(due_at) = cmd.due_at {
307 update.due_at = Set(due_at);
308 }
309 update.updated_at = Set(Some(now));
310
311 let updated = {
312 let txn = self.db().write().begin().await?;
313 let updated = update.update(&txn).await?;
314 self.write_event(
315 &txn,
316 task.id,
317 collection.id,
318 Some(caller),
319 agent_event_line::AgentEventType::BodyUpdated,
320 None,
321 None,
322 )
323 .await?;
324 txn.commit().await?;
325 updated
326 };
327
328 Ok(Self::agent_dto(updated))
329 }
330
331 pub async fn delete_task(
332 &self,
333 caller: Uuid,
334 space_name: &str,
335 collection_id: Uuid,
336 task_ref: &str,
337 ) -> Result<(), SagentError> {
338 let (space, _, task) =
339 task_context(self, caller, space_name, collection_id, task_ref).await?;
340 require_task_write(self, &space, caller, &task).await?;
341
342 self.db()
343 .primary()
344 .transaction::<_, (), SagentError>(|txn| {
345 Box::pin(async move {
346 agent_message::Entity::delete_many()
347 .filter(agent_message::Column::AgentId.eq(task.id))
348 .exec(txn)
349 .await?;
350 agent_watch::Entity::delete_many()
351 .filter(agent_watch::Column::AgentId.eq(task.id))
352 .exec(txn)
353 .await?;
354 agent_event_line::Entity::delete_many()
355 .filter(agent_event_line::Column::AgentId.eq(task.id))
356 .exec(txn)
357 .await?;
358 agent::Entity::delete_by_id(task.id).exec(txn).await?;
359 Ok(())
360 })
361 })
362 .await?;
363
364 Ok(())
365 }
366
367 pub async fn close_task(
368 &self,
369 caller: Uuid,
370 space_name: &str,
371 collection_id: Uuid,
372 task_ref: &str,
373 ) -> Result<AgentDto, SagentError> {
374 let (space, collection, task) =
375 task_context(self, caller, space_name, collection_id, task_ref).await?;
376 require_task_write(self, &space, caller, &task).await?;
377
378 if task.state == AgentState::Closed {
379 return Err(SagentError::Conflict("task is already closed".to_string()));
380 }
381
382 let now = OffsetDateTime::now_utc();
383 let update = agent::ActiveModel {
384 id: Unchanged(task.id),
385 state: Set(AgentState::Closed),
386 closed_at: Set(Some(now)),
387 updated_at: Set(Some(now)),
388 ..Default::default()
389 };
390
391 let updated = {
392 let txn = self.db().write().begin().await?;
393 let updated = update.update(&txn).await?;
394 self.write_event(
395 &txn,
396 task.id,
397 collection.id,
398 Some(caller),
399 agent_event_line::AgentEventType::Closed,
400 None,
401 None,
402 )
403 .await?;
404 txn.commit().await?;
405 updated
406 };
407
408 Ok(Self::agent_dto(updated))
409 }
410
411 pub async fn reopen_task(
412 &self,
413 caller: Uuid,
414 space_name: &str,
415 collection_id: Uuid,
416 task_ref: &str,
417 ) -> Result<AgentDto, SagentError> {
418 let (space, collection, task) =
419 task_context(self, caller, space_name, collection_id, task_ref).await?;
420 require_task_write(self, &space, caller, &task).await?;
421
422 if task.state != AgentState::Closed {
423 return Err(SagentError::Conflict("task is not closed".to_string()));
424 }
425
426 let now = OffsetDateTime::now_utc();
427 let update = agent::ActiveModel {
428 id: Unchanged(task.id),
429 state: Set(AgentState::Draft),
430 closed_at: Set(None),
431 updated_at: Set(Some(now)),
432 ..Default::default()
433 };
434
435 let updated = {
436 let txn = self.db().write().begin().await?;
437 let updated = update.update(&txn).await?;
438 self.write_event(
439 &txn,
440 task.id,
441 collection.id,
442 Some(caller),
443 agent_event_line::AgentEventType::Reopened,
444 None,
445 None,
446 )
447 .await?;
448 txn.commit().await?;
449 updated
450 };
451
452 Ok(Self::agent_dto(updated))
453 }
454
455 pub async fn start_task(
456 &self,
457 caller: Uuid,
458 space_name: &str,
459 collection_id: Uuid,
460 task_ref: &str,
461 ) -> Result<AgentDto, SagentError> {
462 let (space, collection, task) =
463 task_context(self, caller, space_name, collection_id, task_ref).await?;
464 require_task_write(self, &space, caller, &task).await?;
465
466 if !matches!(task.state, AgentState::Draft | AgentState::Pending) {
467 return Err(SagentError::Conflict(
468 "task has already been started".to_string(),
469 ));
470 }
471 if task.body.trim().is_empty() {
472 return Err(SagentError::BadRequest("task body is empty".to_string()));
473 }
474
475 let previous_state = task.state;
476 {
477 let txn = self.db().write().begin().await?;
478 let current = agent::Entity::find_by_id(task.id)
479 .filter(agent::Column::State.is_in([AgentState::Draft, AgentState::Pending]))
480 .lock(LockType::Update)
481 .one(&txn)
482 .await?;
483 let Some(current) = current else {
484 return Err(SagentError::Conflict(
485 "task has already been started".to_string(),
486 ));
487 };
488 let mut update: agent::ActiveModel = current.into();
489 update.state = Set(AgentState::Pending);
490 update.updated_at = Set(Some(OffsetDateTime::now_utc()));
491 update.update(&txn).await?;
492 txn.commit().await?;
493 }
494
495 match self
496 .dispatch_task(caller, &space.id, &collection, &task)
497 .await
498 {
499 Ok(dto) => Ok(dto),
500 Err(error) => {
501 if let Err(restore) = self.set_state(task.id, previous_state).await {
502 tracing::error!(?restore, task_id = %task.id, "failed to restore task state");
503 }
504 Err(error)
505 }
506 }
507 }
508
509 async fn dispatch_task(
510 &self,
511 caller: Uuid,
512 space_id: &Uuid,
513 collection: &models::sagent::agents::Model,
514 task: &agent::Model,
515 ) -> Result<AgentDto, SagentError> {
516 let gateway = self.gateway()?;
517 let title = format!(
518 "{} #{} {}",
519 collection.name, task.autoincrement_id, task.title
520 );
521 let metadata = serde_json::json!({
522 "task_id": task.id,
523 "collection_id": collection.id,
524 "space_id": space_id,
525 });
526
527 let session = gateway
528 .ensure_session(task.session_id, &title, Some(metadata))
529 .await?;
530 let session_id = session.session_id.as_uuid();
531
532 let handle = gateway
533 .open_session(session_id, caller, &INVOKE_SCOPES)
534 .await?;
535 let accepted = handle.send_message(task.body.clone(), None).await?;
536 let run_id = accepted.run_id.as_uuid();
537 let now = OffsetDateTime::now_utc();
538
539 let updated = {
540 let txn = self.db().write().begin().await?;
541 let mut update: agent::ActiveModel = task.clone().into();
542 update.session_id = Set(Some(session_id));
543 update.last_run_id = Set(Some(run_id));
544 update.last_event_seq = Set(0);
545 update.state = Set(map_run_state(accepted.state));
546 update.started_at = Set(Some(task.started_at.unwrap_or(now)));
547 update.finished_at = Set(None);
548 update.updated_at = Set(Some(now));
549 let updated = update.update(&txn).await?;
550 self.write_event(
551 &txn,
552 task.id,
553 collection.id,
554 Some(caller),
555 agent_event_line::AgentEventType::Started,
556 Some(run_id),
557 Some(serde_json::json!({ "state": accepted.state })),
558 )
559 .await?;
560 txn.commit().await?;
561 updated
562 };
563
564 Ok(Self::agent_dto(updated))
565 }
566
567 pub async fn cancel_task(
568 &self,
569 caller: Uuid,
570 space_name: &str,
571 collection_id: Uuid,
572 task_ref: &str,
573 ) -> Result<AgentDto, SagentError> {
574 let (space, collection, task) =
575 task_context(self, caller, space_name, collection_id, task_ref).await?;
576 require_task_write(self, &space, caller, &task).await?;
577
578 let session_id = task
579 .session_id
580 .ok_or_else(|| SagentError::Conflict("task has not been started".to_string()))?;
581 let run_id = task
582 .last_run_id
583 .ok_or_else(|| SagentError::Conflict("task has no run".to_string()))?;
584
585 let gateway = self.gateway()?;
586 let handle = gateway
587 .open_session(session_id, caller, &INVOKE_SCOPES)
588 .await?;
589 let run = handle.cancel_run(run_id).await?;
590
591 let now = OffsetDateTime::now_utc();
592 let updated = {
593 let txn = self.db().write().begin().await?;
594 let mut update: agent::ActiveModel = task.clone().into();
595 update.state = Set(AgentState::Canceled);
596 update.finished_at = Set(Some(now));
597 update.updated_at = Set(Some(now));
598 let updated = update.update(&txn).await?;
599 self.write_event(
600 &txn,
601 task.id,
602 collection.id,
603 Some(caller),
604 agent_event_line::AgentEventType::StateChanged,
605 Some(run_id),
606 Some(serde_json::json!({ "state": run.state })),
607 )
608 .await?;
609 txn.commit().await?;
610 updated
611 };
612
613 Ok(Self::agent_dto(updated))
614 }
615}
Modified lib/sagent/src/share/authz.rs View lib/sagent/src/share/authz.rs
@@ -1,3 +1,4 @@
1use models::sagent::agent::AgentState;
12use models::space::spaces;
23use space::share::authz::{MANAGE_CHANNELS, effective_permissions};
34use uuid::Uuid;
@@ -17,6 +18,29 @@
1718 }
1819}
1920
21pub async fn require_task_write(
22 service: &SagentService,
23 space: &spaces::Model,
24 caller: Uuid,
25 task: &models::sagent::agent::Model,
26) -> Result<(), SagentError> {
27 if task.author_id == caller || is_collection_manager(service, space, caller).await? {
28 Ok(())
29 } else {
30 Err(SagentError::Forbidden)
31 }
32}
33
34pub fn require_task_draft(state: AgentState) -> Result<(), SagentError> {
35 if state == AgentState::Draft {
36 Ok(())
37 } else {
38 Err(SagentError::Conflict(
39 "task body is frozen after the first run".to_string(),
40 ))
41 }
42}
43
2044pub async fn is_collection_manager(
2145 service: &SagentService,
2246 space: &spaces::Model,
Modified lib/sagent/src/openapi.rs View lib/sagent/src/openapi.rs
@@ -9,11 +9,22 @@
99 crate::handlers::collections::get_collection::get_collection,
1010 crate::handlers::collections::update_collection::update_collection,
1111 crate::handlers::collections::delete_collection::delete_collection,
12 crate::handlers::tasks::create_task::create_task,
13 crate::handlers::tasks::list_tasks::list_tasks,
14 crate::handlers::tasks::get_task::get_task,
15 crate::handlers::tasks::update_task::update_task,
16 crate::handlers::tasks::delete_task::delete_task,
17 crate::handlers::tasks::start_task::start_task,
18 crate::handlers::tasks::cancel_task::cancel_task,
19 crate::handlers::tasks::close_task::close_task,
20 crate::handlers::tasks::reopen_task::reopen_task,
1221 ),
1322 components(schemas(
1423 AppResponseData<export::rest::NoData>,
1524 AppResponseData<crate::services::collections::CollectionDto>,
1625 AppResponseData<crate::services::collections::CollectionPage>,
26 AppResponseData<crate::services::tasks::AgentDto>,
27 AppResponseData<crate::services::tasks::AgentPage>,
1728 export::rest::NoData,
1829 space::services::spaces::SpaceNamePath,
1930 crate::services::collections::CollectionDto,
@@ -22,6 +33,13 @@
2233 crate::services::collections::UpdateCollectionCmd,
2334 crate::services::collections::ListCollectionQuery,
2435 crate::services::collections::CollectionPath,
36 crate::services::tasks::AgentDto,
37 crate::services::tasks::AgentPage,
38 crate::services::tasks::CreateAgentCmd,
39 crate::services::tasks::UpdateAgentCmd,
40 crate::services::tasks::ListAgentQuery,
41 crate::services::tasks::CollectionRef,
42 crate::services::tasks::AgentPath,
2543 models::sagent::agent::AgentState,
2644 models::sagent::agent::AgentPriority,
2745 )),
Modified lib/sagent/src/routes.rs View lib/sagent/src/routes.rs
@@ -6,6 +6,15 @@
66use crate::handlers::collections::get_collection::get_collection;
77use crate::handlers::collections::list_collections::list_collections;
88use crate::handlers::collections::update_collection::update_collection;
9use crate::handlers::tasks::cancel_task::cancel_task;
10use crate::handlers::tasks::close_task::close_task;
11use crate::handlers::tasks::create_task::create_task;
12use crate::handlers::tasks::delete_task::delete_task;
13use crate::handlers::tasks::get_task::get_task;
14use crate::handlers::tasks::list_tasks::list_tasks;
15use crate::handlers::tasks::reopen_task::reopen_task;
16use crate::handlers::tasks::start_task::start_task;
17use crate::handlers::tasks::update_task::update_task;
918use crate::services::service::SagentService;
1019use crate::share::guard::sagent_guard;
1120
@@ -21,6 +30,30 @@
2130 .patch(update_collection)
2231 .delete(delete_collection),
2332 )
33 .route(
34 "/agents/{space_name}/collections/{collection_id}/tasks",
35 post(create_task).get(list_tasks),
36 )
37 .route(
38 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}",
39 get(get_task).patch(update_task).delete(delete_task),
40 )
41 .route(
42 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/start",
43 post(start_task),
44 )
45 .route(
46 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/cancel",
47 post(cancel_task),
48 )
49 .route(
50 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/close",
51 post(close_task),
52 )
53 .route(
54 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/reopen",
55 post(reopen_task),
56 )
2457 .route_layer(axum::middleware::from_fn_with_state(
2558 service.clone(),
2659 sagent_guard,