feat: add downloadable agent document outputs

This commit is contained in:
boos
2026-07-14 07:45:30 +08:00
parent 9fe770ad26
commit 5dde5679e1
16 changed files with 695 additions and 10 deletions
Generated
+1
View File
@@ -527,6 +527,7 @@ dependencies = [
"jsonwebtoken",
"lettre",
"libc",
"lopdf",
"md-5",
"mime",
"mime_guess",
+1
View File
@@ -96,6 +96,7 @@ include_dir = "0.7"
mime_guess = "2"
office_oxide = "=0.1.5"
pdf-extract = "=0.12.0"
lopdf = { version = "0.42", default-features = false }
# Embedded AI agent engine (MIT, vendored as a path dependency). Provides the
# agentic loop, built-in tools, permission callback and cost tracking that the
+8
View File
@@ -69,6 +69,7 @@
- **Workflow UI 更新**:`bun run workflow-ui:install:cloud` 会安装当前 Comfy Cloud 前端资产并写入构建元数据,Rust 服务会对 JS/CSS 走 no-cache,避免继续显示旧版本。
- **本地化时间**:前端展示时间优先使用浏览器/UA 时区,后端无时区时间按 UTC 解析后再转成本地显示。
- **图片助手模板**:模板接口按请求读取远程图库数据,不在仓库中捆绑大体积 JSON。
- **Agent 文档能力**:可读取常见 Office、PDF 和文本附件,并生成可下载的 DOCX、XLSX、PPTX、PDF、Markdown、TXT、CSV、JSON 文件。
- **部署方式统一**:前端统一使用 Bun,生产构建统一使用 `./build-and-run.sh --production`。
## ✨ 核心功能
@@ -87,6 +88,13 @@
- **默认走本地模式**:复用本服务自身的 `/v1/*` 路由 + 启动时引导出来的内置 downstream key(`comfyui-agent`),开箱即用,无需额外配置
- **回退远端**:设置 `BOOSAPI_UPSTREAM_BASE`(可选 `BOOSAPI_UPSTREAM_KEY`)即切换为外部上游;`/api/comfyui-agent/status` 的 `mode` 字段会反映当前模式
### 🤖 Agent Workbench
- 支持上传并读取 PDF、DOC/DOCX、XLS/XLSX、PPT/PPTX、TXT、Markdown、CSV 和 JSON 附件
- 支持生成 DOCX、XLSX、PPTX、PDF、Markdown、TXT、CSV 和 JSON,完成后在对话中提供鉴权下载
- 生成文件归档到当前用户的工作空间,并随会话恢复;单个附件或输出文件上限为 64 MB
- 工具写入遵循 Agent 权限模式,默认模式下生成文件需要用户确认
### 🌐 统一代理网关
- 兼容 **OpenAI** 与 **Claude** 下游格式
+8 -2
View File
@@ -17,6 +17,8 @@ pub enum ToolCapability {
AskUserQuestion,
#[serde(rename = "Bash")]
Bash,
#[serde(rename = "CreateDocument")]
CreateDocument,
#[serde(rename = "Skill")]
Skill,
}
@@ -31,6 +33,7 @@ impl ToolCapability {
Self::Grep => "Grep",
Self::AskUserQuestion => "AskUserQuestion",
Self::Bash => "Bash",
Self::CreateDocument => "CreateDocument",
Self::Skill => "Skill",
}
}
@@ -44,6 +47,7 @@ impl ToolCapability {
"Grep" => Some(Self::Grep),
"AskUserQuestion" => Some(Self::AskUserQuestion),
"Bash" => Some(Self::Bash),
"CreateDocument" => Some(Self::CreateDocument),
"Skill" => Some(Self::Skill),
_ => None,
}
@@ -64,7 +68,7 @@ pub struct ToolSpec {
pub approval_can_be_remembered: bool,
}
const SERVER_TOOLS: [ToolSpec; 7] = [
const SERVER_TOOLS: [ToolSpec; 8] = [
ToolSpec::policy(ToolCapability::Read),
ToolSpec::policy(ToolCapability::Write),
ToolSpec::policy(ToolCapability::Edit),
@@ -72,6 +76,7 @@ const SERVER_TOOLS: [ToolSpec; 7] = [
ToolSpec::policy(ToolCapability::Grep),
ToolSpec::policy(ToolCapability::AskUserQuestion),
ToolSpec::policy(ToolCapability::Bash),
ToolSpec::policy(ToolCapability::CreateDocument),
];
const DESKTOP_TOOLS: [ToolSpec; 8] = [
@@ -207,7 +212,8 @@ mod tests {
"Glob",
"Grep",
"AskUserQuestion",
"Bash"
"Bash",
"CreateDocument"
]
);
assert!(!policy.allows("Skill"));
@@ -0,0 +1 @@
ALTER TABLE agent_runs ADD COLUMN artifacts_json TEXT;
@@ -0,0 +1 @@
ALTER TABLE `agent_runs` ADD COLUMN `artifacts_json` LONGTEXT NULL;
@@ -0,0 +1 @@
ALTER TABLE agent_runs ADD COLUMN IF NOT EXISTS artifacts_json TEXT;
+155 -2
View File
@@ -24,6 +24,7 @@
use crate::auth::jwt::validate_jwt;
use crate::services::agent_workbench::{
bash_sandbox::SandboxedBashTool,
document::{output_mime, CreateDocumentTool},
permission::{classify_tool_risk, decide_permission, PermissionAction, PermissionMode},
run_registry, sandbox,
session_store::{self, CancelRunOutcome, ClaimRunError, RunTerminal, UsageDelta},
@@ -68,8 +69,9 @@ const SYSTEM_PROMPT: &str =
"You are BoosAPI's built-in engineering agent. You can read and write files \
and run shell commands, but only inside the current working directory, which is an \
isolated sandbox for this session. Be concise. Explain what you are about to do before \
using a tool that writes files or runs commands. Never attempt to access paths outside \
the working directory.";
using a tool that writes files or runs commands. Use CreateDocument when the user asks \
for a downloadable document, spreadsheet, presentation, PDF, or structured text file. \
Never attempt to access paths outside the working directory.";
static INTERACTIONS: Lazy<InteractionBroker> =
Lazy::new(|| InteractionBroker::new(Duration::from_secs(300)));
@@ -1126,6 +1128,15 @@ async fn chat_handler(
return Err(fail_claim(format!("failed to create session sandbox: {error}")).await)
}
};
let outputs_dir = cwd.join("outputs");
if let Err(error) = tokio::fs::remove_dir_all(&outputs_dir).await {
if error.kind() != std::io::ErrorKind::NotFound {
return Err(fail_claim(format!("failed to reset Agent outputs: {error}")).await);
}
}
if let Err(error) = tokio::fs::create_dir_all(&outputs_dir).await {
return Err(fail_claim(format!("failed to create Agent outputs: {error}")).await);
}
let attachment_paths =
match materialize_attachments(&state.db, auth.user_id, &cwd, &body.attachments).await {
@@ -1189,6 +1200,7 @@ async fn chat_handler(
};
let db = state.db.clone();
let app_state = state.clone();
let user_id = auth.user_id;
let actor_key = auth.actor_key;
let agent_cancel = cancel.clone();
@@ -1257,6 +1269,9 @@ async fn chat_handler(
if tool_policy.allows("Bash") {
custom_tools.push(Arc::new(SandboxedBashTool::new(cwd.clone(), mode)));
}
if tool_policy.allows("CreateDocument") {
custom_tools.push(Arc::new(CreateDocumentTool));
}
let agent = Agent::new(AgentOptions {
provider: Some(provider),
@@ -1362,6 +1377,36 @@ async fn chat_handler(
}
}
if failure.is_none() && !agent_cancel.is_cancelled() && latest_transcript.is_some() {
match archive_agent_outputs(
&app_state,
&cwd,
user_id,
&session_id,
&run_id,
&model,
)
.await
{
Ok(artifacts) if !artifacts.is_empty() => {
match serde_json::to_string(&artifacts) {
Ok(raw) => {
if let Err(error) =
session_store::save_run_artifacts(&db, &run_id, &raw).await
{
failure = Some(error.to_string());
} else {
result_data["artifacts"] = Value::Array(artifacts);
}
}
Err(error) => failure = Some(error.to_string()),
}
}
Ok(_) => {}
Err(error) => failure = Some(error),
}
}
if failure.is_none() && !agent_cancel.is_cancelled() {
if let Some(messages) = latest_transcript.as_ref() {
match context_codec.encode_compacted(messages) {
@@ -1628,6 +1673,102 @@ async fn forward_sdk_event(
Ok(())
}
async fn archive_agent_outputs(
state: &AppState,
cwd: &std::path::Path,
user_id: Option<i64>,
session_id: &str,
run_id: &str,
model: &str,
) -> Result<Vec<Value>, String> {
const MAX_OUTPUT_FILES: usize = 24;
let output_dir = cwd.join("outputs");
let mut entries = tokio::fs::read_dir(&output_dir)
.await
.map_err(|error| format!("failed to read Agent outputs: {error}"))?;
let mut files = Vec::new();
while let Some(entry) = entries
.next_entry()
.await
.map_err(|error| format!("failed to enumerate Agent outputs: {error}"))?
{
let metadata = tokio::fs::symlink_metadata(entry.path())
.await
.map_err(|error| format!("failed to inspect Agent output: {error}"))?;
if metadata.file_type().is_symlink() || !metadata.is_file() {
continue;
}
if output_mime(&entry.path()).is_some() {
files.push(entry.path());
}
}
files.sort();
if files.len() > MAX_OUTPUT_FILES {
return Err(format!("Agent produced more than {MAX_OUTPUT_FILES} output files"));
}
let mut artifacts = Vec::with_capacity(files.len());
for path in files {
let mime_type = output_mime(&path)
.ok_or_else(|| "unsupported Agent output format".to_string())?;
let file_name = path
.file_name()
.and_then(|value| value.to_str())
.ok_or_else(|| "Agent output has an invalid file name".to_string())?
.to_string();
let bytes = tokio::fs::read(&path)
.await
.map_err(|error| format!("failed to read Agent output '{file_name}': {error}"))?;
if bytes.len() > agent_attachments::MAX_ATTACHMENT_BYTES {
return Err(format!("Agent output '{file_name}' exceeds 64 MB"));
}
let asset = workspace::record_bytes_asset(
state,
workspace::WorkspaceAssetInput {
user_id,
source: "agent-output".to_string(),
kind: "document".to_string(),
title: Some(file_name.clone()),
prompt: None,
model: Some(model.to_string()),
remote_url: None,
metadata: json!({
"workspaceAsset": { "role": "output" },
"agentSessionId": session_id,
"agentRunId": run_id,
"originalFileName": file_name,
}),
},
&bytes,
mime_type,
)
.await
.map_err(|error| format!("failed to archive Agent output '{file_name}': {error}"))?;
crate::db::sql::query(
"UPDATE workspace_assets SET file_name = ?, title = ? WHERE public_id = ?",
)
.bind(&file_name)
.bind(&file_name)
.bind(&asset.public_id)
.execute(&*state.db)
.await
.map_err(|error| format!("failed to name Agent output asset: {error}"))?;
let file_url = if user_id.is_some() {
format!("/api/user/workspace/assets/{}/file", asset.public_id)
} else {
format!("/api/workspace/assets/{}/file", asset.public_id)
};
artifacts.push(json!({
"publicId": asset.public_id,
"fileName": file_name,
"mimeType": mime_type,
"sizeBytes": bytes.len(),
"fileUrl": file_url,
}));
}
Ok(artifacts)
}
async fn agent_run_ledger_cost(db: &sqlx::AnyPool, run_id: &str) -> Result<Option<f64>, String> {
crate::db::sql::query_scalar::<_, Option<f64>>(
"SELECT SUM(COALESCE(estimated_cost, 0))
@@ -1787,12 +1928,18 @@ async fn get_run_handler(
Json(json!({ "ok": false, "error": "Agent run not found" })),
)
})?;
let artifacts = run
.artifacts_json
.as_deref()
.and_then(|raw| serde_json::from_str::<Value>(raw).ok())
.unwrap_or_else(|| json!([]));
let binding = InteractionBinding::new(&auth.actor_key, &run.session_id, &run.id);
let interactions = INTERACTIONS.pending_for_binding(&binding).await;
Ok(Json(json!({
"ok": true,
"run": run,
"interactions": interactions,
"artifacts": artifacts,
})))
}
@@ -1843,10 +1990,16 @@ async fn get_session_handler(
})?
.and_then(|raw| serde_json::from_str::<Value>(&raw).ok())
.unwrap_or_else(|| json!([]));
let artifacts = session_store::latest_session_artifacts(&state.db, &id, auth.user_id)
.await
.map_err(internal_db_error)?
.and_then(|raw| serde_json::from_str::<Value>(&raw).ok())
.unwrap_or_else(|| json!([]));
Ok(Json(json!({
"ok": true,
"session": session,
"transcript": transcript,
"artifacts": artifacts,
})))
}
+283
View File
@@ -0,0 +1,283 @@
use async_trait::async_trait;
use lopdf::{
content::{Content, Operation},
dictionary, Document, Object, Stream, StringFormat,
};
use office_oxide::DocumentFormat;
use open_agent_sdk::types::{Tool, ToolError, ToolInputSchema, ToolResult, ToolUseContext};
use serde_json::{json, Value};
use std::{collections::HashMap, io::Cursor, path::Path};
const MAX_DOCUMENT_CHARS: usize = 2_000_000;
pub struct CreateDocumentTool;
#[async_trait]
impl Tool for CreateDocumentTool {
fn name(&self) -> &str {
"CreateDocument"
}
fn description(&self) -> &str {
"Creates a downloadable document in the Agent outputs directory from Markdown content. Supports docx, xlsx, pptx, pdf, md, txt, csv, and json."
}
fn input_schema(&self) -> ToolInputSchema {
ToolInputSchema {
schema_type: "object".to_string(),
properties: HashMap::from([
("file_name".to_string(), json!({
"type": "string",
"description": "Output file name without a directory path"
})),
("format".to_string(), json!({
"type": "string",
"enum": ["docx", "xlsx", "pptx", "pdf", "md", "txt", "csv", "json"]
})),
("content".to_string(), json!({
"type": "string",
"description": "Markdown or plain-text document content"
})),
]),
required: vec!["file_name".to_string(), "format".to_string(), "content".to_string()],
additional_properties: Some(false),
}
}
async fn call(&self, input: Value, context: &ToolUseContext) -> Result<ToolResult, ToolError> {
let requested_name = input
.get("file_name")
.and_then(Value::as_str)
.ok_or_else(|| ToolError::InvalidInput("Missing 'file_name' field".to_string()))?;
let format = input
.get("format")
.and_then(Value::as_str)
.unwrap_or_default()
.to_ascii_lowercase();
let content = input
.get("content")
.and_then(Value::as_str)
.ok_or_else(|| ToolError::InvalidInput("Missing 'content' field".to_string()))?;
if content.chars().count() > MAX_DOCUMENT_CHARS {
return Err(ToolError::InvalidInput(format!(
"Document exceeds {MAX_DOCUMENT_CHARS} characters"
)));
}
let file_name = safe_file_name(requested_name, &format)?;
let output_dir = Path::new(&context.working_dir).join("outputs");
tokio::fs::create_dir_all(&output_dir)
.await
.map_err(ToolError::IoError)?;
let path = output_dir.join(&file_name);
let bytes = match format.as_str() {
"docx" | "xlsx" | "pptx" => create_office(content, &format).await?,
"pdf" => create_pdf(content).map_err(ToolError::IoError)?,
"md" | "txt" | "csv" | "json" => content.as_bytes().to_vec(),
_ => {
return Err(ToolError::InvalidInput(
"Unsupported document format".to_string(),
))
}
};
tokio::fs::write(&path, &bytes)
.await
.map_err(ToolError::IoError)?;
Ok(ToolResult::text(
json!({
"created": true,
"path": format!("outputs/{file_name}"),
"file_name": file_name,
"format": format,
"size_bytes": bytes.len(),
})
.to_string(),
))
}
}
fn safe_file_name(requested: &str, format: &str) -> Result<String, ToolError> {
let base = Path::new(requested)
.file_name()
.and_then(|value| value.to_str())
.unwrap_or_default();
let stem = Path::new(base)
.file_stem()
.and_then(|value| value.to_str())
.unwrap_or("document");
let stem = stem
.chars()
.map(|ch| {
if ch.is_alphanumeric() || matches!(ch, '-' | '_' | ' ') {
ch
} else {
'_'
}
})
.collect::<String>();
let stem = stem.trim().chars().take(96).collect::<String>();
if stem.is_empty()
|| !matches!(
format,
"docx" | "xlsx" | "pptx" | "pdf" | "md" | "txt" | "csv" | "json"
)
{
return Err(ToolError::InvalidInput(
"Invalid document file name or format".to_string(),
));
}
Ok(format!("{stem}.{format}"))
}
async fn create_office(content: &str, format: &str) -> Result<Vec<u8>, ToolError> {
let format = match format {
"docx" => DocumentFormat::Docx,
"xlsx" => DocumentFormat::Xlsx,
"pptx" => DocumentFormat::Pptx,
_ => {
return Err(ToolError::InvalidInput(
"Unsupported Office format".to_string(),
))
}
};
let content = content.to_string();
tokio::task::spawn_blocking(move || {
let mut writer = Cursor::new(Vec::new());
office_oxide::create::create_from_markdown_to_writer(&content, format, &mut writer)
.map_err(|error| ToolError::ExecutionError(error.to_string()))?;
Ok(writer.into_inner())
})
.await
.map_err(|error| ToolError::ExecutionError(error.to_string()))?
}
fn create_pdf(markdown: &str) -> std::io::Result<Vec<u8>> {
let lines = wrap_text(&plain_text(markdown), 72);
let mut document = Document::with_version("1.5");
let cid_font = document.add_object(dictionary! {
"Type" => "Font", "Subtype" => "CIDFontType0", "BaseFont" => "STSong-Light",
"CIDSystemInfo" => dictionary! { "Registry" => "Adobe", "Ordering" => "GB1", "Supplement" => 5 },
});
let font = document.add_object(dictionary! {
"Type" => "Font", "Subtype" => "Type0", "BaseFont" => "STSong-Light",
"Encoding" => "UniGB-UCS2-H", "DescendantFonts" => vec![Object::Reference(cid_font)],
});
let resources = document.add_object(dictionary! { "Font" => dictionary! { "F1" => font } });
let pages = document.new_object_id();
let mut page_ids = Vec::new();
for page_lines in lines.chunks(46) {
let mut operations = vec![
Operation::new("BT", vec![]),
Operation::new("Tf", vec!["F1".into(), 11.into()]),
Operation::new("Td", vec![48.into(), 790.into()]),
Operation::new("TL", vec![16.into()]),
];
for (index, line) in page_lines.iter().enumerate() {
if index > 0 {
operations.push(Operation::new("T*", vec![]));
}
operations.push(Operation::new(
"Tj",
vec![Object::String(
utf16be(line),
StringFormat::Hexadecimal,
)],
));
}
operations.push(Operation::new("ET", vec![]));
let content = Content { operations }.encode().map_err(std::io::Error::other)?;
let content_id = document.add_object(Stream::new(dictionary! {}, content));
let page_id = document.add_object(dictionary! {
"Type" => "Page", "Parent" => pages, "Resources" => resources,
"MediaBox" => vec![0.into(), 0.into(), 595.into(), 842.into()], "Contents" => content_id,
});
page_ids.push(page_id);
}
document.objects.insert(
pages,
Object::Dictionary(dictionary! {
"Type" => "Pages", "Kids" => page_ids.iter().copied().map(Object::Reference).collect::<Vec<_>>(),
"Count" => page_ids.len() as i64,
}),
);
let catalog = document.add_object(dictionary! { "Type" => "Catalog", "Pages" => pages });
document.trailer.set("Root", catalog);
let mut bytes = Vec::new();
document.save_to(&mut bytes).map_err(std::io::Error::other)?;
Ok(bytes)
}
fn plain_text(markdown: &str) -> String {
markdown
.lines()
.map(|line| line.trim_start_matches(['#', '>', '-', '*', ' ']))
.collect::<Vec<_>>()
.join("\n")
}
fn wrap_text(text: &str, width: usize) -> Vec<String> {
let mut lines = Vec::new();
for source in text.lines() {
if source.is_empty() {
lines.push(String::new());
continue;
}
let chars = source.chars().collect::<Vec<_>>();
for chunk in chars.chunks(width) {
lines.push(chunk.iter().collect());
}
}
if lines.is_empty() {
lines.push(String::new());
}
lines
}
fn utf16be(value: &str) -> Vec<u8> {
value.encode_utf16().flat_map(u16::to_be_bytes).collect()
}
pub fn output_mime(path: &Path) -> Option<&'static str> {
match path.extension()?.to_str()?.to_ascii_lowercase().as_str() {
"docx" => Some("application/vnd.openxmlformats-officedocument.wordprocessingml.document"),
"xlsx" => Some("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"),
"pptx" => Some("application/vnd.openxmlformats-officedocument.presentationml.presentation"),
"pdf" => Some("application/pdf"),
"md" => Some("text/markdown"),
"txt" => Some("text/plain"),
"csv" => Some("text/csv"),
"json" => Some("application/json"),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn pdf_is_valid_and_names_are_sanitized() {
let pdf = create_pdf("# 报告\n\n收入 42").unwrap();
assert!(pdf.starts_with(b"%PDF-"));
Document::load_mem(&pdf).unwrap();
assert_eq!(
safe_file_name("../季度/报告.docx", "docx").unwrap(),
"报告.docx"
);
}
#[tokio::test]
async fn office_outputs_can_be_read_back() {
let markdown = "# Quarterly report\n\n| Item | Value |\n| --- | --- |\n| Revenue | 42 |";
for (extension, format) in [
("docx", DocumentFormat::Docx),
("xlsx", DocumentFormat::Xlsx),
("pptx", DocumentFormat::Pptx),
] {
let bytes = create_office(markdown, extension).await.unwrap();
assert!(bytes.starts_with(b"PK"));
let document = office_oxide::Document::from_reader(Cursor::new(bytes), format).unwrap();
assert!(!document.to_markdown().trim().is_empty());
}
}
}
+1
View File
@@ -16,6 +16,7 @@
pub mod approval;
pub mod ask;
pub mod bash_sandbox;
pub mod document;
pub mod permission;
pub mod run_registry;
pub mod sandbox;
+2 -1
View File
@@ -74,7 +74,7 @@ pub fn classify_tool_risk(tool_name: &str) -> RiskLevel {
if matches!(
n.as_str(),
"write" | "edit" | "create" | "delete" | "rename" | "patch" | "move" | "mkdir" | "remove"
"write" | "edit" | "create" | "createdocument" | "delete" | "rename" | "patch" | "move" | "mkdir" | "remove"
) || n.contains("write")
|| n.contains("edit")
|| n.contains("create")
@@ -127,6 +127,7 @@ mod tests {
"Write",
"Edit",
"create_file",
"CreateDocument",
"delete_file",
"patch",
"ApplyPatch",
+86 -1
View File
@@ -86,6 +86,8 @@ pub struct AgentRunSummary {
pub finished_at: Option<String>,
pub created_at: String,
pub updated_at: String,
#[serde(skip_serializing)]
pub artifacts_json: Option<String>,
}
#[derive(Debug, thiserror::Error)]
@@ -408,7 +410,7 @@ pub async fn get_run(
) -> AppResult<Option<AgentRunSummary>> {
crate::db::sql::query_as::<_, AgentRunSummary>(
"SELECT id, session_id, user_id, status, model, revision, error_message,
cancel_requested_at, started_at, finished_at, created_at, updated_at
cancel_requested_at, started_at, finished_at, created_at, updated_at, artifacts_json
FROM agent_runs WHERE id = ? AND actor_key = ?",
)
.bind(run_id)
@@ -418,6 +420,42 @@ pub async fn get_run(
.map_err(AppError::Database)
}
pub async fn save_run_artifacts(
db: &AnyPool,
run_id: &str,
artifacts_json: &str,
) -> AppResult<()> {
crate::db::sql::query(
"UPDATE agent_runs SET artifacts_json = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
)
.bind(artifacts_json)
.bind(run_id)
.execute(db)
.await
.map_err(AppError::Database)?;
Ok(())
}
pub async fn latest_session_artifacts(
db: &AnyPool,
session_id: &str,
user_id: Option<i64>,
) -> AppResult<Option<String>> {
crate::db::sql::query_scalar::<_, Option<String>>(
"SELECT artifacts_json FROM agent_runs
WHERE session_id = ? AND ((? IS NULL AND user_id IS NULL) OR user_id = ?)
AND artifacts_json IS NOT NULL
ORDER BY datetime(created_at) DESC, id DESC LIMIT 1",
)
.bind(session_id)
.bind(user_id)
.bind(user_id)
.fetch_optional(db)
.await
.map(|value| value.flatten())
.map_err(AppError::Database)
}
/// Settle a run exactly once and release its session lease only after the
/// worker has stopped. The returned boolean is the canonical terminal-event
/// ownership flag.
@@ -845,6 +883,12 @@ mod tests {
.execute(&pool)
.await
.unwrap();
crate::db::sql::query(include_str!(
"../../../migrations/20260714000001_agent_run_artifacts.sql"
))
.execute(&pool)
.await
.unwrap();
pool
}
@@ -1077,6 +1121,47 @@ mod tests {
assert_eq!(run.status, "starting");
}
#[tokio::test]
async fn run_artifacts_round_trip_for_session_owner() {
let db = pool().await;
let session_id = Uuid::new_v4().to_string();
let claim = claim_run(
&db,
&session_id,
&Uuid::new_v4().to_string(),
&Uuid::new_v4().to_string(),
Some(6),
"user:6",
"m",
"default",
2,
)
.await
.unwrap();
let artifacts = r#"[{"publicId":"wa_1","fileName":"report.pdf"}]"#;
save_run_artifacts(&db, &claim.run_id, artifacts)
.await
.unwrap();
let run = get_run(&db, &claim.run_id, "user:6")
.await
.unwrap()
.unwrap();
assert_eq!(run.artifacts_json.as_deref(), Some(artifacts));
assert_eq!(
latest_session_artifacts(&db, &session_id, Some(6))
.await
.unwrap()
.as_deref(),
Some(artifacts)
);
assert!(latest_session_artifacts(&db, &session_id, Some(7))
.await
.unwrap()
.is_none());
}
#[tokio::test]
async fn cancelling_run_keeps_lease_until_worker_finishes() {
let db = pool().await;
+40
View File
@@ -59,6 +59,14 @@ export type AudioResult = {
url: string;
};
export type AgentArtifact = {
publicId: string;
fileName: string;
fileUrl: string;
mimeType?: string;
sizeBytes?: number;
};
export type ChatMessage = {
role: 'user' | 'assistant';
text: string;
@@ -66,6 +74,7 @@ export type ChatMessage = {
tools: ToolCall[];
creativeJob?: CreativeJob;
audioResult?: AudioResult;
artifacts?: AgentArtifact[];
};
const TERMINAL_JOB_STATUSES = new Set([
@@ -247,12 +256,43 @@ export function AgentMessage({
/>
)}
{message.audioResult && <AudioResultCard result={message.audioResult} audience={audience} />}
{message.artifacts && message.artifacts.length > 0 && (
<div className="agent-artifacts" aria-label="Agent 生成文件">
{message.artifacts.map((artifact) => (
<ArtifactDownload key={artifact.publicId} artifact={artifact} audience={audience} />
))}
</div>
)}
{message.tools.map((tool) => <ToolCard key={`${tool.id}-${tool.name}`} tool={tool} />)}
</div>
</article>
);
}
function ArtifactDownload({
artifact,
audience,
}: {
artifact: AgentArtifact;
audience: 'admin' | 'user';
}) {
const href = useAuthedMediaUrl(audience, artifact.fileUrl);
const size = artifact.sizeBytes && artifact.sizeBytes > 0
? `${Math.max(1, Math.round(artifact.sizeBytes / 1024))} KB`
: '文档';
return (
<div className="agent-artifact">
<div>
<strong>{artifact.fileName}</strong>
<span>{size}</span>
</div>
{href
? <a href={href} download={artifact.fileName}>下载</a>
: <span>准备下载...</span>}
</div>
);
}
function AudioResultCard({ result, audience }: { result: AudioResult; audience: 'admin' | 'user' }) {
const src = useAuthedMediaUrl(audience, result.url);
return (
+48
View File
@@ -2839,6 +2839,54 @@
min-width: 0;
}
.agent-artifacts {
display: grid;
gap: 8px;
margin-top: 12px;
}
.agent-artifact {
display: flex;
align-items: center;
justify-content: space-between;
gap: 12px;
min-height: 52px;
padding: 10px 12px;
border: 1px solid var(--color-border, #dce7e2);
border-radius: 8px;
background: var(--color-bg-card, #fff);
}
.agent-artifact > div {
display: grid;
min-width: 0;
gap: 2px;
}
.agent-artifact strong {
overflow: hidden;
font-size: 13px;
text-overflow: ellipsis;
white-space: nowrap;
}
.agent-artifact span {
color: var(--color-text-muted, #718078);
font-size: 12px;
}
.agent-artifact a {
flex: 0 0 auto;
color: var(--color-primary, #167c68);
font-size: 13px;
font-weight: 700;
text-decoration: none;
}
.agent-artifact a:hover {
text-decoration: underline;
}
.agent-workbench-composer {
position: relative;
z-index: 10;
+22 -1
View File
@@ -5,7 +5,7 @@ import {
parseAgentSseFrame,
readAgentSse,
} from '../agentStream.js';
import { normalizeJimengErrorMessage } from './AgentWorkbench.js';
import { normalizeAgentArtifacts, normalizeJimengErrorMessage } from './AgentWorkbench.js';
const agentWorkbenchSource = readFileSync(new URL('./AgentWorkbench.tsx', import.meta.url), 'utf8');
@@ -83,6 +83,26 @@ describe('AgentWorkbench session restoration', () => {
expect(normalizePermissionMode('full-access')).toBe('full_access');
expect(normalizePermissionMode('unexpected')).toBe('default');
});
it('normalizes downloadable artifacts and rejects incomplete rows', () => {
expect(normalizeAgentArtifacts([
{
public_id: 'wa_1',
file_name: 'report.pdf',
file_url: '/api/user/workspace/assets/wa_1/file',
size_bytes: 1024,
},
{ publicId: 'incomplete' },
])).toEqual([{
publicId: 'wa_1',
fileName: 'report.pdf',
fileUrl: '/api/user/workspace/assets/wa_1/file',
mimeType: undefined,
sizeBytes: 1024,
}]);
expect(agentWorkbenchSource).toContain('normalizeAgentArtifacts(payload?.artifacts)');
expect(agentWorkbenchSource).toContain('normalizeAgentArtifacts(d.artifacts)');
});
});
describe('AgentWorkbench detached run recovery', () => {
@@ -91,6 +111,7 @@ describe('AgentWorkbench detached run recovery', () => {
expect(agentWorkbenchSource).toContain('ACTIVE_AGENT_RUN_STATUSES.has(recovered.status)');
expect(agentWorkbenchSource).toContain("pending?.kind === 'approval'");
expect(agentWorkbenchSource).toContain("pending?.kind === 'input'");
expect(agentWorkbenchSource).toContain('normalizeAgentArtifacts(payload?.artifacts)');
expect(agentWorkbenchSource).toContain('window.setTimeout(poll, 2000)');
});
+35 -1
View File
@@ -19,6 +19,7 @@ import {
import {
AgentMessage,
isTerminalJobStatus,
type AgentArtifact,
type ChatMessage,
type CreativeJob,
type CreativeJobRequestSnapshot,
@@ -635,6 +636,25 @@ function normalizeCreativeJob(raw: unknown, kind: 'image' | 'video'): CreativeJo
};
}
export function normalizeAgentArtifacts(raw: unknown): AgentArtifact[] {
if (!Array.isArray(raw)) return [];
return raw.flatMap((item) => {
if (!item || typeof item !== 'object') return [];
const row = item as Record<string, unknown>;
const publicId = String(row.publicId ?? row.public_id ?? '').trim();
const fileName = String(row.fileName ?? row.file_name ?? '').trim();
const fileUrl = String(row.fileUrl ?? row.file_url ?? '').trim();
if (!publicId || !fileName || !fileUrl) return [];
return [{
publicId,
fileName,
fileUrl,
mimeType: typeof (row.mimeType ?? row.mime_type) === 'string' ? String(row.mimeType ?? row.mime_type) : undefined,
sizeBytes: Number.isFinite(Number(row.sizeBytes ?? row.size_bytes)) ? Number(row.sizeBytes ?? row.size_bytes) : undefined,
}];
});
}
function seedanceCreativeJob(job: SeedanceJob): CreativeJob {
return {
id: job.id,
@@ -1694,6 +1714,10 @@ export default function AgentWorkbench() {
costUsd: normalizeNumber(eventUsage.cost_usd, eventUsage.costUsd),
numTurns: normalizeNumber(eventUsage.num_turns, eventUsage.numTurns),
});
const artifacts = normalizeAgentArtifacts(d.artifacts);
if (artifacts.length > 0) {
updateAssistant((m) => { m.artifacts = artifacts; });
}
}
break;
case 'error':
@@ -1750,6 +1774,11 @@ export default function AgentWorkbench() {
setSessionId(run.session_id);
setCancellingRun(run.status === 'cancelling');
const artifacts = normalizeAgentArtifacts(payload?.artifacts);
if (artifacts.length > 0) {
updateAssistant((message) => { message.artifacts = artifacts; });
}
const interactions = Array.isArray(payload?.interactions)
? payload.interactions as PendingAgentInteraction[]
: [];
@@ -1778,7 +1807,7 @@ export default function AgentWorkbench() {
setAsk(null);
}
return run;
}, [fetchJson]);
}, [fetchJson, updateAssistant]);
const send = useCallback(async () => {
const rawPrompt = input.trim();
@@ -2060,6 +2089,11 @@ export default function AgentWorkbench() {
});
if (controller.signal.aborted || restoreAbortRef.current !== controller) return;
const restored = transcriptToMessages(payload?.transcript);
const artifacts = normalizeAgentArtifacts(payload?.artifacts);
if (artifacts.length > 0) {
const assistant = [...restored].reverse().find((message) => message.role === 'assistant');
if (assistant) assistant.artifacts = artifacts;
}
setSessionId(id);
activeRunIdRef.current = null;
setActiveRunId(null);