feat(agent): add session list + transcript restore endpoints

- session_store::list_sessions(按 user_id 隔离,updated_at 倒序,限流)
- session_store::set_title_if_empty(用首条 prompt 命名会话,仅首轮生效)
- GET /api/agent/sessions 列出调用方自己的会话摘要
- GET /api/agent/sessions/{id} 返回会话账目 + 重建成前端聊天结构的 transcript
  (tool_result 回填到对应 tool_use,纯工具结果消息不单独成气泡)
- chat handler 每轮调用 set_title_if_empty
- 补 list/title 单测
This commit is contained in:
boos
2026-06-23 20:47:28 +08:00
parent 99f9d5995d
commit b5df7caad8
2 changed files with 297 additions and 2 deletions
+189 -2
View File
@@ -40,9 +40,9 @@ use axum::{
}; };
use futures::stream::Stream; use futures::stream::Stream;
use open_agent_sdk::api::openai::OpenAIProvider; use open_agent_sdk::api::openai::OpenAIProvider;
use open_agent_sdk::types::{ContentBlock, Message, PermissionDecision}; use open_agent_sdk::types::{ContentBlock, Message, MessageRole, PermissionDecision};
use open_agent_sdk::{Agent, AgentOptions, CancellationToken, LLMProvider, SDKMessage}; use open_agent_sdk::{Agent, AgentOptions, CancellationToken, LLMProvider, SDKMessage};
use serde::Deserialize; use serde::{Deserialize, Serialize};
use serde_json::{json, Value}; use serde_json::{json, Value};
use std::convert::Infallible; use std::convert::Infallible;
use std::pin::Pin; use std::pin::Pin;
@@ -74,6 +74,8 @@ pub fn routes() -> Router<AppState> {
.route("/api/agent/approve", post(approve_handler)) .route("/api/agent/approve", post(approve_handler))
.route("/api/agent/answer", post(answer_handler)) .route("/api/agent/answer", post(answer_handler))
.route("/api/agent/status", get(status_handler)) .route("/api/agent/status", get(status_handler))
.route("/api/agent/sessions", get(list_sessions_handler))
.route("/api/agent/sessions/{id}", get(get_session_handler))
.layer(DefaultBodyLimit::max(AGENT_CHAT_BODY_LIMIT_BYTES)) .layer(DefaultBodyLimit::max(AGENT_CHAT_BODY_LIMIT_BYTES))
} }
@@ -364,6 +366,12 @@ async fn chat_handler(
tracing::warn!(%e, "failed to ensure agent session row"); tracing::warn!(%e, "failed to ensure agent session row");
} }
// Title the session by its opening prompt (no-op on later turns) so the
// history list shows something meaningful instead of a bare uuid.
if let Err(e) = session_store::set_title_if_empty(&state.db, &session_id, &body.message).await {
tracing::warn!(%e, "failed to set agent session title");
}
// Load any prior conversation for this session so the agent sees the full // Load any prior conversation for this session so the agent sees the full
// multi-turn history (scoped to the caller — never another tenant's). // multi-turn history (scoped to the caller — never another tenant's).
let history: Vec<Message> = match session_store::load_transcript( let history: Vec<Message> = match session_store::load_transcript(
@@ -654,3 +662,182 @@ struct StatusQuery {
#[serde(default)] #[serde(default)]
session_id: Option<String>, session_id: Option<String>,
} }
#[derive(Debug, Deserialize)]
struct ListSessionsQuery {
/// Max sessions to return (most-recent first); clamped to a sane ceiling.
#[serde(default)]
limit: Option<i64>,
}
/// GET /api/agent/sessions — the caller's own session history (summaries only).
async fn list_sessions_handler(
State(state): State<AppState>,
headers: HeaderMap,
axum::extract::Query(q): axum::extract::Query<ListSessionsQuery>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let auth = resolve_auth(&headers, &state).await?;
let limit = q.limit.unwrap_or(50).clamp(1, 200);
let sessions = session_store::list_sessions(&state.db, auth.user_id, limit)
.await
.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({ "error": e.to_string() })),
)
})?;
Ok(Json(json!({ "ok": true, "sessions": sessions })))
}
/// GET /api/agent/sessions/{id} — a session's accounting plus its conversation
/// transcript rebuilt into the frontend chat shape. Scoped to the caller, so a
/// user can never read another tenant's history.
async fn get_session_handler(
State(state): State<AppState>,
headers: HeaderMap,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let auth = resolve_auth(&headers, &state).await?;
let session = session_store::get_session(&state.db, &id, auth.user_id)
.await
.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({ "error": e.to_string() })),
)
})?
.ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(json!({ "error": "session not found" })),
)
})?;
let messages = match session_store::load_transcript(&state.db, &id, auth.user_id).await {
Ok(Some(raw)) => match serde_json::from_str::<Vec<Message>>(&raw) {
Ok(msgs) => transcript_to_chat(&msgs),
Err(e) => {
tracing::warn!(%e, "failed to parse transcript for restore");
Vec::new()
}
},
Ok(None) => Vec::new(),
Err(e) => {
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({ "error": e.to_string() })),
));
}
};
Ok(Json(json!({
"ok": true,
"session": session,
"messages": messages,
})))
}
/// A restored tool call, shaped to match the frontend `ToolCall` type.
#[derive(Debug, Serialize)]
struct ChatToolView {
id: String,
name: String,
input: Value,
#[serde(skip_serializing_if = "Option::is_none")]
result: Option<String>,
#[serde(rename = "isError", skip_serializing_if = "std::ops::Not::not")]
is_error: bool,
}
/// A restored chat message, shaped to match the frontend `ChatMessage` type.
#[derive(Debug, Serialize)]
struct ChatMessageView {
role: &'static str,
text: String,
tools: Vec<ChatToolView>,
}
/// Rebuild the SDK transcript (`Vec<Message>`) into the frontend's flat chat
/// shape: user/assistant bubbles, each assistant carrying its tool calls with
/// the results stitched back in. Tool results live in the *following* user
/// message (Anthropic format), so we index tool calls by id and attach results
/// as we encounter them. Pure tool-result carrier messages (no text) don't
/// become their own bubble; thinking blocks are omitted from the restore view.
fn transcript_to_chat(messages: &[Message]) -> Vec<ChatMessageView> {
use std::collections::HashMap;
let mut chat: Vec<ChatMessageView> = Vec::new();
// tool_use_id -> (chat message index, tool index within that message)
let mut tool_loc: HashMap<String, (usize, usize)> = HashMap::new();
let push_text = |buf: &mut String, t: &str| {
if !t.is_empty() {
if !buf.is_empty() {
buf.push('\n');
}
buf.push_str(t);
}
};
for msg in messages {
match msg.role {
MessageRole::Assistant => {
let mut text = String::new();
let mut tools: Vec<ChatToolView> = Vec::new();
for block in &msg.content {
match block {
ContentBlock::Text { text: t } => push_text(&mut text, t),
ContentBlock::ToolUse { id, name, input } => tools.push(ChatToolView {
id: id.clone(),
name: name.clone(),
input: input.clone(),
result: None,
is_error: false,
}),
_ => {}
}
}
let msg_idx = chat.len();
for (ti, tool) in tools.iter().enumerate() {
tool_loc.insert(tool.id.clone(), (msg_idx, ti));
}
chat.push(ChatMessageView { role: "assistant", text, tools });
}
MessageRole::User => {
let mut text = String::new();
for block in &msg.content {
match block {
ContentBlock::Text { text: t } => push_text(&mut text, t),
ContentBlock::ToolResult { tool_use_id, content, is_error } => {
let result_text = content
.iter()
.map(|c| match c {
open_agent_sdk::types::ToolResultContentBlock::Text { text } => {
text.clone()
}
open_agent_sdk::types::ToolResultContentBlock::Image { .. } => {
"[image]".to_string()
}
})
.collect::<Vec<_>>()
.join("\n");
if let Some((mi, ti)) = tool_loc.get(tool_use_id) {
if let Some(tool) = chat.get_mut(*mi).and_then(|m| m.tools.get_mut(*ti))
{
tool.result = Some(result_text);
tool.is_error = *is_error;
}
}
}
_ => {}
}
}
// A user message that only carried tool results is not its own bubble.
if !text.is_empty() {
chat.push(ChatMessageView { role: "user", text, tools: Vec::new() });
}
}
}
}
chat
}
@@ -185,6 +185,82 @@ pub async fn get_session(
Ok(row) Ok(row)
} }
/// A session summary row for the history list (no transcript blob — that is
/// fetched separately when a session is opened).
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct AgentSessionSummary {
pub id: String,
pub title: Option<String>,
pub model: String,
pub permission_mode: String,
pub total_tokens: i64,
pub cost_usd: f64,
pub num_turns: i64,
pub message_count: i64,
pub created_at: String,
pub updated_at: String,
}
/// List a caller's sessions, most-recently-updated first, capped at `limit`.
/// Scoped by `user_id` exactly like the other accessors: a user sees only their
/// own rows, an admin (`None`) sees the NULL-`user_id` rows.
pub async fn list_sessions(
db: &SqlitePool,
user_id: Option<i64>,
limit: i64,
) -> AppResult<Vec<AgentSessionSummary>> {
let rows = match user_id {
Some(uid) => {
sqlx::query_as::<_, AgentSessionSummary>(
"SELECT id, title, model, permission_mode, total_tokens, cost_usd, \
num_turns, message_count, created_at, updated_at \
FROM agent_sessions WHERE user_id = ? \
ORDER BY updated_at DESC LIMIT ?",
)
.bind(uid)
.bind(limit)
.fetch_all(db)
.await
}
None => {
sqlx::query_as::<_, AgentSessionSummary>(
"SELECT id, title, model, permission_mode, total_tokens, cost_usd, \
num_turns, message_count, created_at, updated_at \
FROM agent_sessions WHERE user_id IS NULL \
ORDER BY updated_at DESC LIMIT ?",
)
.bind(limit)
.fetch_all(db)
.await
}
}
.map_err(AppError::Database)?;
Ok(rows)
}
/// Set a session's title from its first prompt, but only if it has none yet.
/// Called once per turn; the `title IS NULL` guard makes later turns no-ops, so
/// the title stays pinned to whatever opened the conversation.
pub async fn set_title_if_empty(
db: &SqlitePool,
session_id: &str,
title: &str,
) -> AppResult<()> {
let trimmed = title.trim();
if trimmed.is_empty() {
return Ok(());
}
// Keep titles short for the list rail.
let title: String = trimmed.chars().take(60).collect();
sqlx::query("UPDATE agent_sessions SET title = ? WHERE id = ? AND title IS NULL")
.bind(title)
.bind(session_id)
.execute(db)
.await
.map_err(AppError::Database)?;
Ok(())
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@@ -318,4 +394,36 @@ mod tests {
assert!(session_is_claimable(&db, "c2", None).await.unwrap()); assert!(session_is_claimable(&db, "c2", None).await.unwrap());
assert!(!session_is_claimable(&db, "c2", Some(9)).await.unwrap()); assert!(!session_is_claimable(&db, "c2", Some(9)).await.unwrap());
} }
#[tokio::test]
async fn list_sessions_is_user_scoped_and_ordered() {
let db = pool().await;
ensure_session(&db, "a", Some(1), "m", "default").await.unwrap();
ensure_session(&db, "b", Some(1), "m", "default").await.unwrap();
ensure_session(&db, "c", Some(2), "m", "default").await.unwrap();
ensure_session(&db, "admin", None, "m", "default").await.unwrap();
// Bump "a" so it sorts ahead of "b" (most-recent first).
record_turn(&db, "a", UsageDelta::default(), 2).await.unwrap();
let u1 = list_sessions(&db, Some(1), 50).await.unwrap();
let ids: Vec<&str> = u1.iter().map(|s| s.id.as_str()).collect();
assert_eq!(ids, vec!["a", "b"]); // user 1 only, a before b
assert!(u1.iter().all(|s| s.id != "c" && s.id != "admin"));
// Admin sees only the NULL-user row.
let admin = list_sessions(&db, None, 50).await.unwrap();
assert_eq!(admin.len(), 1);
assert_eq!(admin[0].id, "admin");
}
#[tokio::test]
async fn set_title_only_fills_when_empty() {
let db = pool().await;
ensure_session(&db, "ti", Some(1), "m", "default").await.unwrap();
set_title_if_empty(&db, "ti", "first prompt").await.unwrap();
set_title_if_empty(&db, "ti", "later prompt").await.unwrap();
let s = list_sessions(&db, Some(1), 1).await.unwrap();
assert_eq!(s[0].title.as_deref(), Some("first prompt"));
}
} }