Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(sagent): add agent messages, events, watch and streaming archive

ZhenYi committed
parent d57efed commit 0079d8d Browse files

21 files changed

+779 −31

File tree 21
Modified lib/models/src/sagent/agent_event_line.rs View lib/models/src/sagent/agent_event_line.rs
@@ -46,6 +46,8 @@
4646 RepoRelationAdded,
4747 #[sea_orm(string_value = "repo_relation_removed")]
4848 RepoRelationRemoved,
49 #[sea_orm(string_value = "run_event")]
50 RunEvent,
4951 #[sea_orm(string_value = "subscriber_lost")]
5052 SubscriberLost,
5153 #[sea_orm(string_value = "closed")]
Modified lib/models/src/sagent/agent_watch.rs View lib/models/src/sagent/agent_watch.rs
@@ -22,4 +22,4 @@
2222 pub user: BelongsTo<super::super::users::user::Entity>,
2323}
2424
25impl ActiveModelBehavior for ActiveModel {}
No newline at end of file
25impl ActiveModelBehavior for ActiveModel {}
Added lib/sagent/src/handlers/events/list_events.rs View lib/sagent/src/handlers/events/list_events.rs
@@ -0,0 +1,41 @@
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::events::{AgentEventPage, ListAgentEventQuery};
8use crate::services::service::SagentService;
9use crate::services::tasks::AgentPath;
10
11#[utoipa::path(
12 get,
13 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/events",
14 tag = "sagent",
15 description = "读取任务活动流，含状态迁移、run 事件与订阅中断记录，按时间倒序分页。",
16 params(AgentPath, ListAgentEventQuery),
17 responses(
18 (status = 200, description = "list events", body = AppResponseData<AgentEventPage>),
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 )
23)]
24pub async fn list_events(
25 State(service): State<SagentService>,
26 session: Session,
27 Path(path): Path<AgentPath>,
28 Query(query): Query<ListAgentEventQuery>,
29) -> Result<AppResponseData<AgentEventPage>, SagentApiError> {
30 let uid = require_uid(&session)?;
31 let page = service
32 .list_events(
33 uid,
34 &path.space_name,
35 path.collection_id,
36 &path.task_id,
37 query,
38 )
39 .await?;
40 Ok(AppResponseData::ok(page))
41}
Added lib/sagent/src/handlers/events/mod.rs View lib/sagent/src/handlers/events/mod.rs
@@ -0,0 +1 @@
1pub mod list_events;
Added lib/sagent/src/handlers/messages/list_messages.rs View lib/sagent/src/handlers/messages/list_messages.rs
@@ -0,0 +1,41 @@
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::messages::{AgentMessagePage, ListMessageQuery};
8use crate::services::service::SagentService;
9use crate::services::tasks::AgentPath;
10
11#[utoipa::path(
12 get,
13 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/messages",
14 tag = "sagent",
15 description = "读取任务的对话镜像（本地落库），按网关序号升序分页。用户回复由浏览器直连网关发出，再由落库订阅者写回此表。",
16 params(AgentPath, ListMessageQuery),
17 responses(
18 (status = 200, description = "list messages", body = AppResponseData<AgentMessagePage>),
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 )
23)]
24pub async fn list_messages(
25 State(service): State<SagentService>,
26 session: Session,
27 Path(path): Path<AgentPath>,
28 Query(query): Query<ListMessageQuery>,
29) -> Result<AppResponseData<AgentMessagePage>, SagentApiError> {
30 let uid = require_uid(&session)?;
31 let page = service
32 .list_messages(
33 uid,
34 &path.space_name,
35 path.collection_id,
36 &path.task_id,
37 query,
38 )
39 .await?;
40 Ok(AppResponseData::ok(page))
41}
Added lib/sagent/src/handlers/messages/mod.rs View lib/sagent/src/handlers/messages/mod.rs
@@ -0,0 +1 @@
1pub mod list_messages;
Added lib/sagent/src/handlers/tokens/get_task_token.rs View lib/sagent/src/handlers/tokens/get_task_token.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::AgentPath;
9use crate::services::tokens::TaskTokenDto;
10
11#[utoipa::path(
12 get,
13 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/token",
14 tag = "sagent",
15 description = "为当前用户签发该任务会话的短期令牌，浏览器据此直连沙箱智能体网关订阅事件与发送消息。",
16 params(AgentPath),
17 responses(
18 (status = 200, description = "issue session token", body = AppResponseData<TaskTokenDto>),
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 has not been started", body = AppResponseData<NoData>),
23 (status = 429, description = "rate limited", body = AppResponseData<NoData>),
24 (status = 502, description = "agent gateway rejected the request", body = AppResponseData<NoData>),
25 (status = 503, description = "agent gateway unavailable", body = AppResponseData<NoData>)
26 )
27)]
28pub async fn get_task_token(
29 State(service): State<SagentService>,
30 session: Session,
31 Path(path): Path<AgentPath>,
32) -> Result<AppResponseData<TaskTokenDto>, SagentApiError> {
33 let uid = require_uid(&session)?;
34 let dto = service
35 .sign_task_token(uid, &path.space_name, path.collection_id, &path.task_id)
36 .await?;
37 Ok(AppResponseData::ok(dto))
38}
Added lib/sagent/src/handlers/tokens/mod.rs View lib/sagent/src/handlers/tokens/mod.rs
@@ -0,0 +1 @@
1pub mod get_task_token;
Added lib/sagent/src/handlers/watches/mod.rs View lib/sagent/src/handlers/watches/mod.rs
@@ -0,0 +1 @@
1pub mod upsert_watch;
Added lib/sagent/src/handlers/watches/upsert_watch.rs View lib/sagent/src/handlers/watches/upsert_watch.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::AgentPath;
9use crate::services::watches::{AgentWatchDto, UpsertWatchCmd};
10
11#[utoipa::path(
12 put,
13 path = "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/watch",
14 tag = "sagent",
15 description = "设置当前用户对任务的关注状态。",
16 params(AgentPath),
17 request_body = UpsertWatchCmd,
18 responses(
19 (status = 200, description = "upsert watch", body = AppResponseData<AgentWatchDto>),
20 (status = 400, description = "bad request", body = AppResponseData<NoData>),
21 (status = 401, description = "authentication required", body = AppResponseData<NoData>),
22 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
23 (status = 404, description = "not found", body = AppResponseData<NoData>)
24 )
25)]
26pub async fn upsert_watch(
27 State(service): State<SagentService>,
28 session: Session,
29 Path(path): Path<AgentPath>,
30 Json(cmd): Json<UpsertWatchCmd>,
31) -> Result<AppResponseData<AgentWatchDto>, SagentApiError> {
32 let uid = require_uid(&session)?;
33 let dto = service
34 .upsert_watch(
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,2 +1,6 @@
11pub mod collections;
2pub mod events;
3pub mod messages;
24pub mod tasks;
5pub mod tokens;
6pub mod watches;
Added lib/sagent/src/services/events.rs View lib/sagent/src/services/events.rs
@@ -0,0 +1,75 @@
1use models::sagent::agent_event_line::{self, AgentEventType};
2use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect};
3use serde::{Deserialize, Serialize};
4use time::OffsetDateTime;
5use utoipa::{IntoParams, ToSchema};
6use uuid::Uuid;
7
8use crate::errors::SagentError;
9use crate::services::service::SagentService;
10use crate::share::lookup::task_context;
11
12#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
13pub struct AgentEventDto {
14 pub id: Uuid,
15 pub agent_id: Option<Uuid>,
16 pub collection_id: Option<Uuid>,
17 pub user_id: Option<Uuid>,
18 pub event_type: AgentEventType,
19 pub run_id: Option<Uuid>,
20 pub content: Option<serde_json::Value>,
21 pub created_at: OffsetDateTime,
22}
23
24pub type AgentEventPage = export::pagination::Page<AgentEventDto>;
25
26#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
27pub struct ListAgentEventQuery {
28 pub limit: Option<u64>,
29 pub offset: Option<u64>,
30}
31
32impl SagentService {
33 pub(crate) fn event_dto(event: agent_event_line::Model) -> AgentEventDto {
34 AgentEventDto {
35 id: event.id,
36 agent_id: event.agent_id,
37 collection_id: event.collection_id,
38 user_id: event.user_id,
39 event_type: event.event_type,
40 run_id: event.run_id,
41 content: event.content,
42 created_at: event.created_at,
43 }
44 }
45
46 pub async fn list_events(
47 &self,
48 caller: Uuid,
49 space_name: &str,
50 collection_id: Uuid,
51 task_ref: &str,
52 query: ListAgentEventQuery,
53 ) -> Result<AgentEventPage, SagentError> {
54 let (_, _, task) = task_context(self, caller, space_name, collection_id, task_ref).await?;
55 let (limit, offset) = self.paginate(query.limit, query.offset);
56
57 let select = agent_event_line::Entity::find()
58 .filter(agent_event_line::Column::AgentId.eq(task.id))
59 .order_by_desc(agent_event_line::Column::CreatedAt);
60
61 let total = select.clone().count(self.db().read()).await? as i64;
62 let rows = select
63 .limit(limit)
64 .offset(offset)
65 .all(self.db().read())
66 .await?;
67
68 Ok(AgentEventPage {
69 items: rows.into_iter().map(Self::event_dto).collect(),
70 total,
71 limit,
72 offset,
73 })
74 }
75}
Added lib/sagent/src/services/messages.rs View lib/sagent/src/services/messages.rs
@@ -0,0 +1,79 @@
1use models::sagent::agent_message::{self, AgentRole};
2use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect};
3use serde::{Deserialize, Serialize};
4use time::OffsetDateTime;
5use utoipa::{IntoParams, ToSchema};
6use uuid::Uuid;
7
8use crate::errors::SagentError;
9use crate::services::service::SagentService;
10use crate::share::lookup::task_context;
11
12#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
13pub struct AgentMessageDto {
14 pub id: Uuid,
15 pub agent_id: Uuid,
16 pub seq: i64,
17 pub gateway_message_id: Option<Uuid>,
18 pub author_id: Option<Uuid>,
19 pub role: AgentRole,
20 pub content: String,
21 pub created_at: OffsetDateTime,
22}
23
24pub type AgentMessagePage = export::pagination::Page<AgentMessageDto>;
25
26#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
27pub struct ListMessageQuery {
28 pub after_seq: Option<i64>,
29 pub limit: Option<u64>,
30 pub offset: Option<u64>,
31}
32
33impl SagentService {
34 pub(crate) fn message_dto(message: agent_message::Model) -> AgentMessageDto {
35 AgentMessageDto {
36 id: message.id,
37 agent_id: message.agent_id,
38 seq: message.seq,
39 gateway_message_id: message.gateway_message_id,
40 author_id: message.author_id,
41 role: message.role,
42 content: message.content,
43 created_at: message.created_at,
44 }
45 }
46
47 pub async fn list_messages(
48 &self,
49 caller: Uuid,
50 space_name: &str,
51 collection_id: Uuid,
52 task_ref: &str,
53 query: ListMessageQuery,
54 ) -> Result<AgentMessagePage, SagentError> {
55 let (_, _, task) = task_context(self, caller, space_name, collection_id, task_ref).await?;
56 let (limit, offset) = self.paginate(query.limit, query.offset);
57
58 let mut select =
59 agent_message::Entity::find().filter(agent_message::Column::AgentId.eq(task.id));
60 if let Some(after_seq) = query.after_seq {
61 select = select.filter(agent_message::Column::Seq.gt(after_seq));
62 }
63 select = select.order_by_asc(agent_message::Column::Seq);
64
65 let total = select.clone().count(self.db().read()).await? as i64;
66 let rows = select
67 .limit(limit)
68 .offset(offset)
69 .all(self.db().read())
70 .await?;
71
72 Ok(AgentMessagePage {
73 items: rows.into_iter().map(Self::message_dto).collect(),
74 total,
75 limit,
76 offset,
77 })
78 }
79}
Modified lib/sagent/src/services/mod.rs View lib/sagent/src/services/mod.rs
@@ -1,3 +1,8 @@
11pub mod collections;
2pub mod events;
3pub mod messages;
24pub mod service;
5pub mod subscribe;
36pub mod tasks;
7pub mod tokens;
8pub mod watches;
Modified lib/sagent/src/services/service.rs View lib/sagent/src/services/service.rs
@@ -5,7 +5,6 @@
55use export::AppState;
66use export::database::AppDatabase;
77use models::sagent::agent_event_line::{self, AgentEventType};
8use models::users::user_notifications::{self, NotificationType};
98use sea_orm::{ActiveModelTrait, ConnectionTrait, Set};
109use session::SessionError;
1110use session::share::guard::{SessionGuard, SessionWithUser};
@@ -78,34 +77,6 @@
7877 }
7978 .insert(txn)
8079 .await?;
81 Ok(())
82 }
83
84 pub(crate) async fn notify_user(
85 &self,
86 txn: &impl ConnectionTrait,
87 pending: &export::notify::PendingNotifications,
88 recipient: Uuid,
89 ntype: NotificationType,
90 title: String,
91 body: Option<String>,
92 data: Option<serde_json::Value>,
93 actor_uid: Option<Uuid>,
94 ) -> Result<(), SagentError> {
95 let row = user_notifications::ActiveModel {
96 id: Set(Uuid::now_v7()),
97 user_id: Set(recipient),
98 notification_type: Set(ntype),
99 title: Set(title),
100 body: Set(body),
101 data_json: Set(data),
102 actor_uid: Set(actor_uid),
103 read_at: Set(None),
104 created_at: Set(OffsetDateTime::now_utc()),
105 }
106 .insert(txn)
107 .await?;
108 pending.push(recipient, row.to_event_value());
10980 Ok(())
11081 }
11182}
Added lib/sagent/src/services/subscribe.rs View lib/sagent/src/services/subscribe.rs
@@ -0,0 +1,284 @@
1use std::time::Duration;
2
3use agent_sdk::SseEvent;
4use agent_types::session::{MessageDto, MessageRole};
5use models::sagent::agent;
6use models::sagent::agent_event_line::AgentEventType;
7use models::sagent::agent_message::{self, AgentRole};
8use sea_orm::sea_query::{Expr, OnConflict};
9use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, Set, TransactionTrait};
10use time::OffsetDateTime;
11use uuid::Uuid;
12
13use crate::errors::SagentError;
14use crate::gateway::{INVOKE_SCOPES, map_run_state};
15use crate::services::collections::ACTIVE_STATES;
16use crate::services::service::SagentService;
17
18const RECONNECT_BACKOFF_MS: u64 = 500;
19const RECONNECT_BACKOFF_MAX_MS: u64 = 8_000;
20
21fn backoff(attempt: u64) -> Duration {
22 let factor = 1u64 << attempt.min(4);
23 Duration::from_millis((RECONNECT_BACKOFF_MS * factor).min(RECONNECT_BACKOFF_MAX_MS))
24}
25
26fn millis_to_time(millis: i64) -> OffsetDateTime {
27 OffsetDateTime::from_unix_timestamp_nanos(i128::from(millis) * 1_000_000)
28 .unwrap_or_else(|_| OffsetDateTime::now_utc())
29}
30
31impl SagentService {
32 pub(crate) async fn load_task(
33 &self,
34 task_id: Uuid,
35 ) -> Result<Option<agent::Model>, SagentError> {
36 Ok(agent::Entity::find_by_id(task_id)
37 .one(self.db().read())
38 .await?)
39 }
40
41 pub(crate) fn spawn_archive_subscriber(&self, task_id: Uuid) {
42 let service = self.clone();
43 tokio::spawn(async move {
44 if let Err(error) = service.run_archive_subscriber(task_id).await {
45 tracing::error!(?error, task_id = %task_id, "sagent archive subscriber failed");
46 }
47 });
48 }
49
50 pub async fn run_reconcile_loop(self) {
51 let interval = self.config().sagent.reconcile_interval_secs.max(10);
52 loop {
53 tokio::time::sleep(Duration::from_secs(interval)).await;
54 if let Err(error) = self.reconcile_active_tasks().await {
55 tracing::warn!(?error, "sagent reconcile pass failed");
56 }
57 }
58 }
59
60 async fn run_archive_subscriber(&self, task_id: Uuid) -> Result<(), SagentError> {
61 let reconnect_max = u64::from(self.config().sagent.subscribe_reconnect_max);
62 let mut attempt: u64 = 0;
63 loop {
64 let Some(task) = self.load_task(task_id).await? else {
65 return Ok(());
66 };
67 if task.state.is_terminal() {
68 return Ok(());
69 }
70 let Some(session_id) = task.session_id else {
71 return Ok(());
72 };
73
74 match self.consume_stream(&task, session_id).await {
75 Ok(true) => return Ok(()),
76 Ok(false) => {}
77 Err(error) => {
78 tracing::warn!(?error, task_id = %task_id, "sagent subscriber disconnected");
79 if matches!(
80 error,
81 SagentError::Unauthorized | SagentError::Forbidden | SagentError::NotFound
82 ) {
83 self.mark_subscriber_lost(task_id, &error.to_string())
84 .await?;
85 return Ok(());
86 }
87 }
88 }
89
90 attempt += 1;
91 if attempt > reconnect_max {
92 self.mark_subscriber_lost(task_id, "reconnect budget exhausted")
93 .await?;
94 return Ok(());
95 }
96 tokio::time::sleep(backoff(attempt)).await;
97 }
98 }
99
100 async fn consume_stream(
101 &self,
102 task: &agent::Model,
103 session_id: Uuid,
104 ) -> Result<bool, SagentError> {
105 let gateway = self.gateway()?;
106 let handle = gateway
107 .open_session(session_id, task.author_id, &INVOKE_SCOPES)
108 .await?;
109 let from = u64::try_from(task.last_event_seq).unwrap_or(0);
110 let mut stream = handle.stream_events(Some(from)).await?;
111
112 while let Some(frame) = stream.next().await {
113 if self.apply_frame(task, frame?).await? {
114 return Ok(true);
115 }
116 }
117 Ok(false)
118 }
119
120 async fn apply_frame(&self, task: &agent::Model, event: SseEvent) -> Result<bool, SagentError> {
121 match event {
122 SseEvent::MessageAdded { message, .. } => {
123 self.record_message(task, &message).await?;
124 Ok(false)
125 }
126 SseEvent::RunEvent { run_id, seq, event } => {
127 let txn = self.db().write().begin().await?;
128 self.write_event(
129 &txn,
130 task.id,
131 task.collection_id,
132 None,
133 AgentEventType::RunEvent,
134 Some(run_id.as_uuid()),
135 Some(serde_json::json!({ "seq": seq, "event": event })),
136 )
137 .await?;
138 txn.commit().await?;
139 Ok(false)
140 }
141 SseEvent::RunState { run_id, state } => {
142 self.apply_run_state(task, run_id.as_uuid(), state).await
143 }
144 SseEvent::Other { .. } => Ok(false),
145 }
146 }
147
148 async fn record_message(
149 &self,
150 task: &agent::Model,
151 message: &MessageDto,
152 ) -> Result<(), SagentError> {
153 let seq = message.seq.min(i64::MAX as u64) as i64;
154 if seq <= task.last_event_seq {
155 return Ok(());
156 }
157 let message_id = message.message_id.as_uuid();
158 let role = match message.role {
159 MessageRole::User => AgentRole::User,
160 MessageRole::Assistant => AgentRole::Agent,
161 };
162
163 let mut conflict =
164 OnConflict::columns([agent_message::Column::AgentId, agent_message::Column::Seq]);
165 conflict.do_nothing();
166
167 let txn = self.db().write().begin().await?;
168 agent_message::Entity::insert(agent_message::ActiveModel {
169 id: Set(message_id),
170 agent_id: Set(task.id),
171 seq: Set(seq),
172 gateway_message_id: Set(Some(message_id)),
173 author_id: Set(None),
174 role: Set(role),
175 content: Set(message.content.to_string()),
176 created_at: Set(millis_to_time(message.created_at_millis)),
177 updated_at: Set(None),
178 })
179 .on_conflict(conflict)
180 .exec(&txn)
181 .await?;
182
183 agent::Entity::update_many()
184 .col_expr(agent::Column::LastEventSeq, Expr::value(seq))
185 .col_expr(
186 agent::Column::UpdatedAt,
187 Expr::value(OffsetDateTime::now_utc()),
188 )
189 .filter(agent::Column::Id.eq(task.id))
190 .filter(agent::Column::LastEventSeq.lt(seq))
191 .exec(&txn)
192 .await?;
193
194 txn.commit().await?;
195 Ok(())
196 }
197
198 async fn apply_run_state(
199 &self,
200 task: &agent::Model,
201 run_id: Uuid,
202 state: agent_types::state::RunState,
203 ) -> Result<bool, SagentError> {
204 let mapped = map_run_state(state);
205 let terminal = mapped.is_terminal();
206 let now = OffsetDateTime::now_utc();
207
208 let txn = self.db().write().begin().await?;
209 let mut update = agent::Entity::update_many()
210 .col_expr(agent::Column::State, Expr::value(mapped))
211 .col_expr(agent::Column::UpdatedAt, Expr::value(now))
212 .filter(agent::Column::Id.eq(task.id))
213 .filter(agent::Column::State.ne(mapped));
214 if terminal {
215 update = update.col_expr(agent::Column::FinishedAt, Expr::value(Some(now)));
216 }
217 update.exec(&txn).await?;
218
219 self.write_event(
220 &txn,
221 task.id,
222 task.collection_id,
223 None,
224 AgentEventType::StateChanged,
225 Some(run_id),
226 Some(serde_json::json!({ "state": mapped, "terminal": terminal })),
227 )
228 .await?;
229 txn.commit().await?;
230
231 Ok(terminal)
232 }
233
234 async fn mark_subscriber_lost(&self, task_id: Uuid, reason: &str) -> Result<(), SagentError> {
235 let Some(task) = self.load_task(task_id).await? else {
236 return Ok(());
237 };
238 let txn = self.db().write().begin().await?;
239 self.write_event(
240 &txn,
241 task.id,
242 task.collection_id,
243 None,
244 AgentEventType::SubscriberLost,
245 task.last_run_id,
246 Some(serde_json::json!({ "reason": reason })),
247 )
248 .await?;
249 txn.commit().await?;
250 tracing::error!(task_id = %task_id, reason, "sagent archive subscriber lost");
251 Ok(())
252 }
253
254 async fn reconcile_active_tasks(&self) -> Result<(), SagentError> {
255 let tasks = agent::Entity::find()
256 .filter(agent::Column::State.is_in(ACTIVE_STATES))
257 .filter(agent::Column::SessionId.is_not_null())
258 .all(self.db().read())
259 .await?;
260
261 for task in tasks {
262 let Some(session_id) = task.session_id else {
263 continue;
264 };
265 let gateway = self.gateway()?;
266 let handle = gateway
267 .open_session(session_id, task.author_id, &INVOKE_SCOPES)
268 .await?;
269
270 let from = u64::try_from(task.last_event_seq).unwrap_or(0);
271 let cursor = if from == 0 { None } else { Some(from) };
272 let page = handle.list_messages(Some(200), cursor).await?;
273 for message in page.items {
274 self.record_message(&task, &message).await?;
275 }
276
277 if let Some(run_id) = task.last_run_id {
278 let run = handle.get_run(run_id).await?;
279 self.apply_run_state(&task, run_id, run.state).await?;
280 }
281 }
282 Ok(())
283 }
284}
Modified lib/sagent/src/services/tasks.rs View lib/sagent/src/services/tasks.rs
@@ -561,6 +561,7 @@
561561 updated
562562 };
563563
564 self.spawn_archive_subscriber(updated.id);
564565 Ok(Self::agent_dto(updated))
565566 }
566567
Added lib/sagent/src/services/tokens.rs View lib/sagent/src/services/tokens.rs
@@ -0,0 +1,44 @@
1use serde::{Deserialize, Serialize};
2use time::OffsetDateTime;
3use utoipa::ToSchema;
4use uuid::Uuid;
5
6use crate::errors::SagentError;
7use crate::gateway::INVOKE_SCOPES;
8use crate::services::service::SagentService;
9use crate::share::lookup::task_context;
10
11#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
12pub struct TaskTokenDto {
13 pub session_id: Uuid,
14 pub token: String,
15 pub expires_at: OffsetDateTime,
16 pub base_url: String,
17}
18
19impl SagentService {
20 pub async fn sign_task_token(
21 &self,
22 caller: Uuid,
23 space_name: &str,
24 collection_id: Uuid,
25 task_ref: &str,
26 ) -> Result<TaskTokenDto, SagentError> {
27 let (_, _, task) = task_context(self, caller, space_name, collection_id, task_ref).await?;
28 let session_id = task
29 .session_id
30 .ok_or_else(|| SagentError::Conflict("task has not been started".to_string()))?;
31
32 let gateway = self.gateway()?;
33 let handle = gateway
34 .open_session(session_id, caller, &INVOKE_SCOPES)
35 .await?;
36
37 Ok(TaskTokenDto {
38 session_id,
39 token: handle.token().to_string(),
40 expires_at: handle.expires_at(),
41 base_url: gateway.base().to_string(),
42 })
43 }
44}
Added lib/sagent/src/services/watches.rs View lib/sagent/src/services/watches.rs
@@ -0,0 +1,76 @@
1use models::common::watch_mode::WatchMode;
2use models::sagent::agent_watch;
3use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
4use serde::{Deserialize, Serialize};
5use time::OffsetDateTime;
6use utoipa::ToSchema;
7use uuid::Uuid;
8
9use crate::errors::SagentError;
10use crate::services::service::SagentService;
11use crate::share::lookup::task_context;
12
13#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
14pub struct AgentWatchDto {
15 pub id: Uuid,
16 pub agent_id: Uuid,
17 pub user_id: Uuid,
18 pub mode: WatchMode,
19 pub created_at: OffsetDateTime,
20 pub updated_at: Option<OffsetDateTime>,
21}
22
23#[derive(Debug, Clone, Deserialize, ToSchema)]
24pub struct UpsertWatchCmd {
25 pub mode: WatchMode,
26}
27
28impl SagentService {
29 pub async fn upsert_watch(
30 &self,
31 caller: Uuid,
32 space_name: &str,
33 collection_id: Uuid,
34 task_ref: &str,
35 cmd: UpsertWatchCmd,
36 ) -> Result<AgentWatchDto, SagentError> {
37 let (_, _, task) = task_context(self, caller, space_name, collection_id, task_ref).await?;
38 let now = OffsetDateTime::now_utc();
39
40 let existing = agent_watch::Entity::find()
41 .filter(agent_watch::Column::AgentId.eq(task.id))
42 .filter(agent_watch::Column::UserId.eq(caller))
43 .one(self.db().read())
44 .await?;
45
46 let row = match existing {
47 Some(row) => {
48 let mut update: agent_watch::ActiveModel = row.into();
49 update.mode = Set(cmd.mode);
50 update.updated_at = Set(Some(now));
51 update.update(self.db().write()).await?
52 }
53 None => {
54 agent_watch::ActiveModel {
55 id: Set(Uuid::now_v7()),
56 agent_id: Set(task.id),
57 user_id: Set(caller),
58 mode: Set(cmd.mode),
59 created_at: Set(now),
60 updated_at: Set(None),
61 }
62 .insert(self.db().write())
63 .await?
64 }
65 };
66
67 Ok(AgentWatchDto {
68 id: row.id,
69 agent_id: row.agent_id,
70 user_id: row.user_id,
71 mode: row.mode,
72 created_at: row.created_at,
73 updated_at: row.updated_at,
74 })
75 }
76}
Modified lib/sagent/src/openapi.rs View lib/sagent/src/openapi.rs
@@ -18,6 +18,10 @@
1818 crate::handlers::tasks::cancel_task::cancel_task,
1919 crate::handlers::tasks::close_task::close_task,
2020 crate::handlers::tasks::reopen_task::reopen_task,
21 crate::handlers::messages::list_messages::list_messages,
22 crate::handlers::events::list_events::list_events,
23 crate::handlers::tokens::get_task_token::get_task_token,
24 crate::handlers::watches::upsert_watch::upsert_watch,
2125 ),
2226 components(schemas(
2327 AppResponseData<export::rest::NoData>,
@@ -42,6 +46,22 @@
4246 crate::services::tasks::AgentPath,
4347 models::sagent::agent::AgentState,
4448 models::sagent::agent::AgentPriority,
49 AppResponseData<crate::services::messages::AgentMessagePage>,
50 AppResponseData<crate::services::events::AgentEventPage>,
51 AppResponseData<crate::services::watches::AgentWatchDto>,
52 AppResponseData<crate::services::tokens::TaskTokenDto>,
53 crate::services::messages::AgentMessageDto,
54 crate::services::messages::AgentMessagePage,
55 crate::services::messages::ListMessageQuery,
56 crate::services::events::AgentEventDto,
57 crate::services::events::AgentEventPage,
58 crate::services::events::ListAgentEventQuery,
59 crate::services::watches::AgentWatchDto,
60 crate::services::watches::UpsertWatchCmd,
61 crate::services::tokens::TaskTokenDto,
62 models::sagent::agent_message::AgentRole,
63 models::sagent::agent_event_line::AgentEventType,
64 models::common::watch_mode::WatchMode,
4565 )),
4666 tags(
4767 (name = "sagent", description = "Sandbox agent API")
Modified lib/sagent/src/routes.rs View lib/sagent/src/routes.rs
@@ -1,11 +1,13 @@
11use axum::Router;
2use axum::routing::{get, post};
2use axum::routing::{get, post, put};
33
44use crate::handlers::collections::create_collection::create_collection;
55use crate::handlers::collections::delete_collection::delete_collection;
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::events::list_events::list_events;
10use crate::handlers::messages::list_messages::list_messages;
911use crate::handlers::tasks::cancel_task::cancel_task;
1012use crate::handlers::tasks::close_task::close_task;
1113use crate::handlers::tasks::create_task::create_task;
@@ -15,6 +17,8 @@
1517use crate::handlers::tasks::reopen_task::reopen_task;
1618use crate::handlers::tasks::start_task::start_task;
1719use crate::handlers::tasks::update_task::update_task;
20use crate::handlers::tokens::get_task_token::get_task_token;
21use crate::handlers::watches::upsert_watch::upsert_watch;
1822use crate::services::service::SagentService;
1923use crate::share::guard::sagent_guard;
2024
@@ -54,6 +58,22 @@
5458 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/reopen",
5559 post(reopen_task),
5660 )
61 .route(
62 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/messages",
63 get(list_messages),
64 )
65 .route(
66 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/events",
67 get(list_events),
68 )
69 .route(
70 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/token",
71 get(get_task_token),
72 )
73 .route(
74 "/agents/{space_name}/collections/{collection_id}/tasks/{task_id}/watch",
75 put(upsert_watch),
76 )
5777 .route_layer(axum::middleware::from_fn_with_state(
5878 service.clone(),
5979 sagent_guard,