diff --git a/Cargo.lock b/Cargo.lock index 11a20dccc..023c37a69 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4355,6 +4355,7 @@ dependencies = [ "bitcoin", "block2 0.6.2", "bs58", + "bytes", "chacha20poly1305", "chrono", "chrono-tz", diff --git a/Cargo.toml b/Cargo.toml index 9dc7211dd..ee2519017 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -103,6 +103,9 @@ serde_yaml = "0.9" # providers/gmail/post_process.rs) and prefer the email's # `text/plain` MIME part when available.) reqwest = { version = "0.12", default-features = false, features = ["json", "blocking", "rustls-tls", "native-tls", "stream", "http2", "multipart", "socks"] } +# Already in-tree via reqwest/hyper; named directly so `IntegrationClient::get_bytes` +# can return `bytes::Bytes` without a copy. +bytes = "1" tokio = { version = "1", features = ["full", "sync"] } once_cell = "1.19" parking_lot = "0.12" diff --git a/docs/TEST-COVERAGE-MATRIX.md b/docs/TEST-COVERAGE-MATRIX.md index 30b5422cd..4df2be339 100644 --- a/docs/TEST-COVERAGE-MATRIX.md +++ b/docs/TEST-COVERAGE-MATRIX.md @@ -292,6 +292,15 @@ End-to-end coverage of the agent harness via the web-chat RPC surface against an | 6.3.8 | Subconscious Triggers debug/manage panel (Brain) | WD | `app/test/playwright/specs/subconscious-triggers.spec.ts` | ✅ | Brain→Subconscious panel: renders disabled baseline + hint + reserved thread ids; enable toggle → Pipeline Enabled + event_driven + orchestrator running; disable; refresh re-fetches. | | 6.3.9 | Vision sub-agent reads attached images | RU | `src/openhuman/agent_registry/agents/loader.rs::vision_agent_loads_on_vision_hint`, `src/openhuman/inference/provider/factory_tests.rs::vision_tier_is_vision_capable`, `src/openhuman/agent/harness/engine/core.rs::gate_tests`, `src/openhuman/agent/multimodal_tests.rs::extract_image_placeholders_pulls_att_tokens_in_order` | ✅ | Orchestrator (non-vision `chat-v1`) keeps the image as a placeholder, delegates to `vision_agent` on the `vision-v1` tier, which rehydrates the on-disk attachment and reads it. Engine gate prefers per-tier `current_model_vision`; turn placeholders forwarded into the sub-agent prompt. | +### 6.4 Managed Cloud File Storage + +| ID | Feature | Layer | Test path(s) | Status | Notes | +| ----- | ---------------------------------------------------------------- | ----- | ------------------------------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| 6.4.1 | `storage_upload_file` (multipart, quota/TTL args, path safety) | RU | `src/openhuman/file_storage/tools_tests.rs` | ✅ | wiremock upload against the backend envelope; rejects workspace-escaping/symlinked paths, bad visibility/ttl; surfaces backend errors (e.g. insufficient balance). | +| 6.4.2 | `storage_download_file` (redirect follow + persist to workspace) | RU | `src/openhuman/file_storage/tools_tests.rs` | ✅ | Follows the 302 to the presigned URL, persists bytes under the action dir, honors explicit filename. | +| 6.4.3 | `storage_list_files` / `storage_get_link` | RU | `src/openhuman/file_storage/tools_tests.rs` | ✅ | List + usage rendering; presigned link generation with expiry arg validation. | +| 6.4.4 | `storage_set_visibility` / `storage_delete_file` | RU | `src/openhuman/file_storage/tools_tests.rs` | ✅ | Public/private toggle surfaces the stable public URL; delete confirms backend `deleted` flag; both are Write-level tools with external effect. | + --- ## 7. Web & Network Capabilities diff --git a/src/openhuman/file_storage/mod.rs b/src/openhuman/file_storage/mod.rs new file mode 100644 index 000000000..0b5f2f2f0 --- /dev/null +++ b/src/openhuman/file_storage/mod.rs @@ -0,0 +1,17 @@ +//! File storage domain — agent tools for managed cloud file storage backed by +//! the OpenHuman backend's `file_storage` provider +//! (`/agent-integrations/file-storage/*`, S3 under the hood). +//! +//! The backend owns the bucket, billing (S3 rates + margin, charged via the +//! standard integration billing flow), per-user quota (1 GiB), and TTL +//! enforcement (7 days free / up to 1 year paid); these tools upload from and +//! download into the agent's action dir and expose list / link / visibility / +//! delete operations. + +pub mod tools; +pub mod types; + +pub use tools::{ + build_file_storage_tools, StorageDeleteFileTool, StorageDownloadFileTool, StorageGetLinkTool, + StorageListFilesTool, StorageSetVisibilityTool, StorageUploadFileTool, +}; diff --git a/src/openhuman/file_storage/tools.rs b/src/openhuman/file_storage/tools.rs new file mode 100644 index 000000000..996c42c22 --- /dev/null +++ b/src/openhuman/file_storage/tools.rs @@ -0,0 +1,837 @@ +//! Agent-facing file-storage tools backed by the OpenHuman backend's +//! `file_storage` provider (S3 under the hood). +//! +//! **Endpoints** (see the file-storage API contract): +//! - `POST /agent-integrations/file-storage/files` (multipart upload) +//! - `GET /agent-integrations/file-storage/files` (list) +//! - `GET /agent-integrations/file-storage/files/{id}/download` (302 → presigned S3) +//! - `POST /agent-integrations/file-storage/files/{id}/link` (presigned link) +//! - `PATCH /agent-integrations/file-storage/files/{id}` (visibility) +//! - `DELETE /agent-integrations/file-storage/files/{id}` +//! +//! Billing: uploads are charged upfront for the whole TTL at S3 rates plus a +//! margin; downloads and link generation are charged as egress. Quota is +//! 1 GiB per user; TTL is 7 days on the free plan / up to 1 year on paid +//! plans. Public files get a stable public URL. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{json, Value}; + +use crate::openhuman::config::Config; +use crate::openhuman::integrations::IntegrationClient; +use crate::openhuman::tools::traits::{ + PermissionLevel, Tool, ToolCallOptions, ToolCategory, ToolResult, +}; +use tinyagents::harness::tool::ToolExecutionContext; + +use super::types::{DeleteResponse, FileMeta, LinkResponse, ListFilesResponse, UploadResponse}; + +const FILES_PATH: &str = "/agent-integrations/file-storage/files"; + +/// Subdirectory (under `action_dir`) where downloaded files are stored. +/// Mirrors `media_generation`'s `generated-media/` root — the action dir is +/// the agent's canonical read/write root, so this stays read-only-container +/// compatible. +const DOWNLOADS_DIR: &str = "storage-downloads"; + +// ── Shared helpers ────────────────────────────────────────────────── + +/// Resolve `raw` (absolute or relative to `action_dir`) to a canonical path +/// and reject anything that escapes the action dir (the agent's workspace). +/// The file must exist — canonicalization also resolves symlinks, so a +/// symlink pointing outside the workspace is rejected too. +fn resolve_upload_path(action_dir: &Path, raw: &str) -> Result { + let candidate = { + let p = Path::new(raw); + if p.is_absolute() { + p.to_path_buf() + } else { + action_dir.join(p) + } + }; + let root = action_dir.canonicalize().map_err(|e| { + format!( + "workspace dir {} is not accessible: {e}", + action_dir.display() + ) + })?; + let resolved = candidate.canonicalize().map_err(|e| { + format!( + "path {} does not exist or is not readable: {e}", + candidate.display() + ) + })?; + if !resolved.starts_with(&root) { + return Err(format!( + "path {} escapes the agent workspace ({}) — only files inside the workspace can be uploaded", + raw, + root.display() + )); + } + if !resolved.is_file() { + return Err(format!("path {} is not a regular file", resolved.display())); + } + Ok(resolved) +} + +/// Validate a caller-supplied file id before interpolating it into a URL +/// path segment. +fn validate_file_id(args: &Value) -> Result { + let id = args + .get("file_id") + .and_then(|v| v.as_str()) + .map(str::trim) + .unwrap_or_default(); + if id.is_empty() { + return Err("file_id is required".to_string()); + } + if !id + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') + { + return Err(format!("file_id '{id}' contains invalid characters")); + } + Ok(id.to_string()) +} + +/// Parse + validate a `visibility` arg value. +fn validate_visibility(raw: &str) -> Result { + match raw.trim().to_ascii_lowercase().as_str() { + v @ ("public" | "private") => Ok(v.to_string()), + other => Err(format!( + "visibility must be 'public' or 'private' (got '{other}')" + )), + } +} + +/// Strip path separators / traversal from a caller- or server-supplied +/// filename so it always lands directly inside the downloads dir. +fn sanitize_filename(name: &str) -> Option { + let base = name + .rsplit(['/', '\\']) + .next() + .unwrap_or(name) + .trim() + .trim_matches('.'); + if base.is_empty() { + return None; + } + let safe: String = base + .chars() + .map(|c| { + if c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_' | ' ') { + c + } else { + '_' + } + }) + .collect(); + let safe = safe.trim().to_string(); + if safe.is_empty() { + None + } else { + Some(safe) + } +} + +/// Pick a file extension from a content type (mirrors +/// `media_generation::download::extension_for`'s content-type branch, plus +/// common document types). +fn extension_for_content_type(content_type: Option<&str>) -> &'static str { + let Some(ct) = content_type else { return "bin" }; + let ct = ct.to_ascii_lowercase(); + for (needle, ext) in [ + ("png", "png"), + ("webp", "webp"), + ("jpeg", "jpg"), + ("jpg", "jpg"), + ("gif", "gif"), + ("mp4", "mp4"), + ("webm", "webm"), + ("pdf", "pdf"), + ("zip", "zip"), + ("json", "json"), + ("csv", "csv"), + ("html", "html"), + ("text/plain", "txt"), + ] { + if ct.contains(needle) { + return ext; + } + } + "bin" +} + +/// Guess a mime type from a filename extension for the multipart upload +/// part. Best-effort — the backend stores whatever we send and S3 doesn't +/// care; unknown extensions fall back to `application/octet-stream`. +fn mime_for_path(path: &Path) -> &'static str { + match path + .extension() + .and_then(|e| e.to_str()) + .map(|e| e.to_ascii_lowercase()) + .as_deref() + { + Some("png") => "image/png", + Some("jpg" | "jpeg") => "image/jpeg", + Some("gif") => "image/gif", + Some("webp") => "image/webp", + Some("mp4") => "video/mp4", + Some("webm") => "video/webm", + Some("pdf") => "application/pdf", + Some("zip") => "application/zip", + Some("json") => "application/json", + Some("csv") => "text/csv", + Some("html" | "htm") => "text/html", + Some("txt" | "md" | "log") => "text/plain", + _ => "application/octet-stream", + } +} + +/// Resolve the effective action dir for a call, preferring the TinyAgents +/// workspace from the execution context (mirrors `media_generation`). +fn action_dir_for_context( + default_action_dir: &Path, + context: Option<&ToolExecutionContext>, + tool_name: &str, +) -> PathBuf { + if let Some(workspace) = context.and_then(|ctx| ctx.workspace.as_ref()) { + tracing::debug!( + tool = tool_name, + workspace_root = %workspace.root.display(), + policy_id = %workspace.policy_id, + "[file_storage] using ToolExecutionContext workspace root" + ); + return workspace.root.clone(); + } + default_action_dir.to_path_buf() +} + +fn file_path(file_id: &str, suffix: &str) -> String { + format!("{FILES_PATH}/{file_id}{suffix}") +} + +// ── StorageUploadFileTool ─────────────────────────────────────────── + +pub struct StorageUploadFileTool { + client: Arc, + action_dir: PathBuf, +} + +impl StorageUploadFileTool { + pub fn new(client: Arc, action_dir: PathBuf) -> Self { + Self { client, action_dir } + } + + async fn run(&self, args: Value, action_dir: &Path) -> anyhow::Result { + let raw_path = match args.get("path").and_then(|v| v.as_str()) { + Some(p) if !p.trim().is_empty() => p.trim(), + _ => return Ok(ToolResult::error("path is required")), + }; + let resolved = match resolve_upload_path(action_dir, raw_path) { + Ok(p) => p, + Err(e) => return Ok(ToolResult::error(e)), + }; + + let visibility = match args.get("visibility").and_then(|v| v.as_str()) { + Some(v) => match validate_visibility(v) { + Ok(v) => Some(v), + Err(e) => return Ok(ToolResult::error(e)), + }, + None => None, + }; + let ttl_days = match args.get("ttl_days") { + Some(v) => match v.as_u64() { + Some(d) if d >= 1 => Some(d), + _ => return Ok(ToolResult::error("ttl_days must be a positive integer")), + }, + None => None, + }; + + let bytes = match tokio::fs::read(&resolved).await { + Ok(b) => b, + Err(e) => { + return Ok(ToolResult::error(format!( + "failed to read {}: {e}", + resolved.display() + ))) + } + }; + let filename = resolved + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("file") + .to_string(); + let mime = mime_for_path(&resolved); + + tracing::debug!( + "[file_storage] uploading {} ({} bytes, mime={}, visibility={:?}, ttl_days={:?})", + resolved.display(), + bytes.len(), + mime, + visibility, + ttl_days + ); + + let part = match reqwest::multipart::Part::bytes(bytes) + .file_name(filename.clone()) + .mime_str(mime) + { + Ok(p) => p, + Err(e) => return Ok(ToolResult::error(format!("invalid mime '{mime}': {e}"))), + }; + let mut form = reqwest::multipart::Form::new().part("file", part); + if let Some(v) = &visibility { + form = form.text("visibility", v.clone()); + } + if let Some(d) = ttl_days { + form = form.text("ttlDays", d.to_string()); + } + + match self + .client + .upload_multipart::(FILES_PATH, form) + .await + { + Ok(resp) => { + let mut lines = vec![format!( + "Uploaded {} ({} bytes) as file_id {} — visibility {}, expires {}.", + resp.filename, + resp.size, + resp.file_id, + resp.visibility, + resp.expires_at.as_deref().unwrap_or("unknown"), + )]; + if let Some(url) = &resp.public_url { + lines.push(format!("Public URL: {url}")); + } + lines.push(format!("Cost: ${:.4}", resp.cost_usd)); + let payload = json!({ + "file_id": resp.file_id, + "filename": resp.filename, + "size": resp.size, + "content_type": resp.content_type, + "visibility": resp.visibility, + "expires_at": resp.expires_at, + "public_url": resp.public_url, + "cost_usd": resp.cost_usd, + }); + Ok(ToolResult::success_with_markdown(payload, lines.join("\n"))) + } + Err(e) => Ok(ToolResult::error(format!("File upload failed: {e}"))), + } + } +} + +#[async_trait] +impl Tool for StorageUploadFileTool { + fn name(&self) -> &str { + "storage_upload_file" + } + + fn description(&self) -> &str { + "Upload a file from the agent workspace to managed cloud file storage and get a \ + file_id (and, for public files, a stable public URL). The path must be inside the \ + agent workspace. Files are billed at S3 rates plus margin (storage for the whole \ + TTL charged upfront on upload). Quota: 1 GiB per user. TTL: 7 days free / up to \ + 1 year on paid plans." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "path": { "type": "string", "description": "File to upload — absolute or relative to the agent workspace; must resolve inside the workspace" }, + "visibility": { "type": "string", "enum": ["public", "private"], "description": "Default private. Public files get a stable public URL anyone can fetch (egress billed to you)." }, + "ttl_days": { "type": "integer", "minimum": 1, "description": "File lifetime in days (clamped to plan max: 7 free / 365 paid). Default: plan max." } + }, + "required": ["path"] + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::Execute + } + + fn category(&self) -> ToolCategory { + ToolCategory::Workflow + } + + fn external_effect(&self) -> bool { + true + } + + async fn execute(&self, args: Value) -> anyhow::Result { + self.run(args, &self.action_dir).await + } + + async fn execute_with_context( + &self, + args: Value, + _options: ToolCallOptions, + context: Option<&ToolExecutionContext>, + ) -> anyhow::Result { + let action_dir = action_dir_for_context(&self.action_dir, context, self.name()); + self.run(args, &action_dir).await + } +} + +// ── StorageDownloadFileTool ───────────────────────────────────────── + +pub struct StorageDownloadFileTool { + client: Arc, + action_dir: PathBuf, +} + +impl StorageDownloadFileTool { + pub fn new(client: Arc, action_dir: PathBuf) -> Self { + Self { client, action_dir } + } + + async fn run(&self, args: Value, action_dir: &Path) -> anyhow::Result { + let file_id = match validate_file_id(&args) { + Ok(id) => id, + Err(e) => return Ok(ToolResult::error(e)), + }; + let requested_name = args + .get("filename") + .and_then(|v| v.as_str()) + .and_then(sanitize_filename); + + tracing::debug!("[file_storage] downloading file_id={file_id}"); + let (body, content_type, server_name) = match self + .client + .get_bytes(&file_path(&file_id, "/download")) + .await + { + Ok(t) => t, + Err(e) => return Ok(ToolResult::error(format!("File download failed: {e}"))), + }; + + // Naming: explicit arg > server Content-Disposition > file_id + a + // content-type-derived extension (mirrors persist_media's scheme). + let filename = requested_name + .or_else(|| server_name.as_deref().and_then(sanitize_filename)) + .unwrap_or_else(|| { + format!( + "{file_id}.{}", + extension_for_content_type(content_type.as_deref()) + ) + }); + + let dir = action_dir.join(DOWNLOADS_DIR); + if let Err(e) = tokio::fs::create_dir_all(&dir).await { + return Ok(ToolResult::error(format!( + "failed to create downloads dir {}: {e}", + dir.display() + ))); + } + let path = dir.join(&filename); + if let Err(e) = tokio::fs::write(&path, &body).await { + return Ok(ToolResult::error(format!( + "failed to write {}: {e}", + path.display() + ))); + } + tracing::debug!( + "[file_storage] saved file_id={} → {} ({} bytes)", + file_id, + path.display(), + body.len() + ); + + let payload = json!({ + "file_id": file_id, + "path": path.display().to_string(), + "size": body.len(), + "content_type": content_type, + }); + Ok(ToolResult::success_with_markdown( + payload, + format!( + "Downloaded file {} ({} bytes) → {}", + file_id, + body.len(), + path.display() + ), + )) + } +} + +#[async_trait] +impl Tool for StorageDownloadFileTool { + fn name(&self) -> &str { + "storage_download_file" + } + + fn description(&self) -> &str { + "Download a file from managed cloud file storage into the agent workspace and \ + return the saved local path. Egress is billed at S3 rates plus margin." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "file_id": { "type": "string", "description": "The stored file's id (from storage_upload_file / storage_list_files)" }, + "filename": { "type": "string", "description": "Optional local filename to save as (defaults to the stored filename)" } + }, + "required": ["file_id"] + }) + } + + fn category(&self) -> ToolCategory { + ToolCategory::Workflow + } + + async fn execute(&self, args: Value) -> anyhow::Result { + self.run(args, &self.action_dir).await + } + + async fn execute_with_context( + &self, + args: Value, + _options: ToolCallOptions, + context: Option<&ToolExecutionContext>, + ) -> anyhow::Result { + let action_dir = action_dir_for_context(&self.action_dir, context, self.name()); + self.run(args, &action_dir).await + } +} + +// ── StorageListFilesTool ──────────────────────────────────────────── + +pub struct StorageListFilesTool { + client: Arc, +} + +impl StorageListFilesTool { + pub fn new(client: Arc) -> Self { + Self { client } + } +} + +#[async_trait] +impl Tool for StorageListFilesTool { + fn name(&self) -> &str { + "storage_list_files" + } + + fn description(&self) -> &str { + "List your files in managed cloud file storage with sizes, visibility, expiry, \ + and current storage usage against the 1 GiB quota. Listing is free." + } + + fn parameters_schema(&self) -> Value { + json!({ "type": "object", "properties": {} }) + } + + fn category(&self) -> ToolCategory { + ToolCategory::Workflow + } + + async fn execute(&self, _args: Value) -> anyhow::Result { + tracing::debug!("[file_storage] listing files"); + match self.client.get::(FILES_PATH).await { + Ok(resp) => { + let mut lines = vec![format!( + "{} file(s), using {} of {} bytes:", + resp.files.len(), + resp.usage.used_bytes, + resp.usage.limit_bytes + )]; + for f in &resp.files { + lines.push(format!( + "- {} — {} ({} bytes, {}, expires {}){}", + f.file_id, + f.filename, + f.size, + f.visibility, + f.expires_at.as_deref().unwrap_or("unknown"), + f.public_url + .as_deref() + .map(|u| format!(" — {u}")) + .unwrap_or_default(), + )); + } + let payload = json!({ + "files": resp.files.iter().map(FileMeta::to_json).collect::>(), + "next_cursor": resp.next_cursor, + "usage": { + "used_bytes": resp.usage.used_bytes, + "limit_bytes": resp.usage.limit_bytes, + }, + }); + Ok(ToolResult::success_with_markdown(payload, lines.join("\n"))) + } + Err(e) => Ok(ToolResult::error(format!("Failed to list files: {e}"))), + } + } +} + +// ── StorageGetLinkTool ────────────────────────────────────────────── + +pub struct StorageGetLinkTool { + client: Arc, +} + +impl StorageGetLinkTool { + pub fn new(client: Arc) -> Self { + Self { client } + } +} + +#[async_trait] +impl Tool for StorageGetLinkTool { + fn name(&self) -> &str { + "storage_get_link" + } + + fn description(&self) -> &str { + "Generate a short-lived presigned download link for a stored file (works for \ + private files; 60s to 7 days, default 1 hour). Link generation is billed as \ + egress at S3 rates plus margin. For a stable permanent URL, set the file's \ + visibility to public instead." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "file_id": { "type": "string", "description": "The stored file's id" }, + "expires_in_seconds": { "type": "integer", "minimum": 60, "maximum": 604800, "description": "Link lifetime in seconds (default 3600)" } + }, + "required": ["file_id"] + }) + } + + fn category(&self) -> ToolCategory { + ToolCategory::Workflow + } + + async fn execute(&self, args: Value) -> anyhow::Result { + let file_id = match validate_file_id(&args) { + Ok(id) => id, + Err(e) => return Ok(ToolResult::error(e)), + }; + let mut body = json!({}); + if let Some(secs) = args.get("expires_in_seconds").and_then(|v| v.as_u64()) { + body["expiresInSeconds"] = json!(secs.clamp(60, 604_800)); + } + tracing::debug!("[file_storage] generating link for file_id={file_id}"); + match self + .client + .post::(&file_path(&file_id, "/link"), &body) + .await + { + Ok(resp) => Ok(ToolResult::success_with_markdown( + json!({ + "file_id": file_id, + "url": resp.url, + "expires_at": resp.expires_at, + "cost_usd": resp.cost_usd, + }), + format!( + "Presigned link for {} (expires {}): {}\nCost: ${:.4}", + file_id, + resp.expires_at.as_deref().unwrap_or("unknown"), + resp.url, + resp.cost_usd + ), + )), + Err(e) => Ok(ToolResult::error(format!( + "Failed to generate download link: {e}" + ))), + } + } +} + +// ── StorageSetVisibilityTool ──────────────────────────────────────── + +pub struct StorageSetVisibilityTool { + client: Arc, +} + +impl StorageSetVisibilityTool { + pub fn new(client: Arc) -> Self { + Self { client } + } +} + +#[async_trait] +impl Tool for StorageSetVisibilityTool { + fn name(&self) -> &str { + "storage_set_visibility" + } + + fn description(&self) -> &str { + "Change a stored file's visibility. Public files get a stable public URL anyone \ + can fetch (egress billed to you); private files are only reachable via \ + authenticated download or presigned links. Visibility changes are free." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "file_id": { "type": "string", "description": "The stored file's id" }, + "visibility": { "type": "string", "enum": ["public", "private"], "description": "New visibility" } + }, + "required": ["file_id", "visibility"] + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::Write + } + + fn category(&self) -> ToolCategory { + ToolCategory::Workflow + } + + fn external_effect(&self) -> bool { + true + } + + async fn execute(&self, args: Value) -> anyhow::Result { + let file_id = match validate_file_id(&args) { + Ok(id) => id, + Err(e) => return Ok(ToolResult::error(e)), + }; + let visibility = match args.get("visibility").and_then(|v| v.as_str()) { + Some(v) => match validate_visibility(v) { + Ok(v) => v, + Err(e) => return Ok(ToolResult::error(e)), + }, + None => return Ok(ToolResult::error("visibility is required")), + }; + tracing::debug!("[file_storage] setting visibility={visibility} for file_id={file_id}"); + match self + .client + .patch::( + &file_path(&file_id, ""), + &json!({ "visibility": visibility }), + ) + .await + { + Ok(meta) => { + let mut md = format!("File {} is now {}.", meta.file_id, meta.visibility); + if let Some(url) = &meta.public_url { + md.push_str(&format!("\nPublic URL: {url}")); + } + Ok(ToolResult::success_with_markdown(meta.to_json(), md)) + } + Err(e) => Ok(ToolResult::error(format!( + "Failed to change file visibility: {e}" + ))), + } + } +} + +// ── StorageDeleteFileTool ─────────────────────────────────────────── + +pub struct StorageDeleteFileTool { + client: Arc, +} + +impl StorageDeleteFileTool { + pub fn new(client: Arc) -> Self { + Self { client } + } +} + +#[async_trait] +impl Tool for StorageDeleteFileTool { + fn name(&self) -> &str { + "storage_delete_file" + } + + fn description(&self) -> &str { + "Permanently delete a file from managed cloud file storage, freeing quota. \ + Deletion is free and cannot be undone." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "file_id": { "type": "string", "description": "The stored file's id" } + }, + "required": ["file_id"] + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::Write + } + + fn category(&self) -> ToolCategory { + ToolCategory::Workflow + } + + fn external_effect(&self) -> bool { + true + } + + async fn execute(&self, args: Value) -> anyhow::Result { + let file_id = match validate_file_id(&args) { + Ok(id) => id, + Err(e) => return Ok(ToolResult::error(e)), + }; + tracing::debug!("[file_storage] deleting file_id={file_id}"); + match self + .client + .delete::(&file_path(&file_id, "")) + .await + { + Ok(resp) if resp.deleted => Ok(ToolResult::success_with_markdown( + json!({ "file_id": file_id, "deleted": true }), + format!("Deleted file {file_id}."), + )), + Ok(_) => Ok(ToolResult::error(format!( + "Backend did not confirm deletion of file {file_id}" + ))), + Err(e) => Ok(ToolResult::error(format!("Failed to delete file: {e}"))), + } + } +} + +// ── Builder ───────────────────────────────────────────────────────── + +/// Build the file-storage tool surface. Returns empty when no integration +/// client is configured (no backend URL / not signed in), mirroring +/// `build_media_tools`. +pub fn build_file_storage_tools(root_config: &Config, action_dir: &Path) -> Vec> { + let Some(client) = crate::openhuman::integrations::build_client(root_config) else { + tracing::debug!("[file_storage] no integration client — file-storage tools skipped"); + return Vec::new(); + }; + + let action_dir = action_dir.to_path_buf(); + let tools: Vec> = vec![ + Box::new(StorageUploadFileTool::new( + Arc::clone(&client), + action_dir.clone(), + )), + Box::new(StorageDownloadFileTool::new( + Arc::clone(&client), + action_dir, + )), + Box::new(StorageListFilesTool::new(Arc::clone(&client))), + Box::new(StorageGetLinkTool::new(Arc::clone(&client))), + Box::new(StorageSetVisibilityTool::new(Arc::clone(&client))), + Box::new(StorageDeleteFileTool::new(Arc::clone(&client))), + ]; + tracing::debug!( + "[file_storage] registered {} file-storage tools", + tools.len() + ); + tools +} + +#[cfg(test)] +#[path = "tools_tests.rs"] +mod tools_tests; diff --git a/src/openhuman/file_storage/tools_tests.rs b/src/openhuman/file_storage/tools_tests.rs new file mode 100644 index 000000000..184c0cae6 --- /dev/null +++ b/src/openhuman/file_storage/tools_tests.rs @@ -0,0 +1,494 @@ +use std::path::PathBuf; +use std::sync::Arc; + +use serde_json::json; + +use super::{ + resolve_upload_path, sanitize_filename, StorageDeleteFileTool, StorageDownloadFileTool, + StorageGetLinkTool, StorageListFilesTool, StorageSetVisibilityTool, StorageUploadFileTool, +}; +use crate::openhuman::integrations::IntegrationClient; +use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolCategory}; + +fn dummy_client() -> Arc { + // No requests are made in these tests; the URL/token are placeholders. + Arc::new(IntegrationClient::new( + "http://127.0.0.1:0".to_string(), + "test-token".to_string(), + )) +} + +// ── Metadata / schema ─────────────────────────────────────────────── + +#[test] +fn upload_tool_schema_and_metadata() { + let tool = StorageUploadFileTool::new(dummy_client(), PathBuf::from("/tmp")); + assert_eq!(tool.name(), "storage_upload_file"); + assert_eq!(tool.permission_level(), PermissionLevel::Execute); + assert_eq!(tool.category(), ToolCategory::Workflow); + assert!(tool.external_effect()); + + let schema = tool.parameters_schema(); + assert_eq!(schema["required"], json!(["path"])); + let props = schema["properties"].as_object().unwrap(); + for key in ["path", "visibility", "ttl_days"] { + assert!(props.contains_key(key), "missing upload property {key}"); + } +} + +#[test] +fn download_tool_schema_and_metadata() { + let tool = StorageDownloadFileTool::new(dummy_client(), PathBuf::from("/tmp")); + assert_eq!(tool.name(), "storage_download_file"); + assert_eq!(tool.permission_level(), PermissionLevel::ReadOnly); + assert_eq!(tool.category(), ToolCategory::Workflow); + assert!(!tool.external_effect()); + + let schema = tool.parameters_schema(); + assert_eq!(schema["required"], json!(["file_id"])); + let props = schema["properties"].as_object().unwrap(); + for key in ["file_id", "filename"] { + assert!(props.contains_key(key), "missing download property {key}"); + } +} + +#[test] +fn list_tool_metadata() { + let tool = StorageListFilesTool::new(dummy_client()); + assert_eq!(tool.name(), "storage_list_files"); + assert_eq!(tool.permission_level(), PermissionLevel::ReadOnly); + assert_eq!(tool.category(), ToolCategory::Workflow); + assert!(!tool.external_effect()); +} + +#[test] +fn get_link_tool_schema_and_metadata() { + let tool = StorageGetLinkTool::new(dummy_client()); + assert_eq!(tool.name(), "storage_get_link"); + assert_eq!(tool.permission_level(), PermissionLevel::ReadOnly); + assert_eq!(tool.category(), ToolCategory::Workflow); + + let schema = tool.parameters_schema(); + assert_eq!(schema["required"], json!(["file_id"])); + let props = schema["properties"].as_object().unwrap(); + for key in ["file_id", "expires_in_seconds"] { + assert!(props.contains_key(key), "missing link property {key}"); + } +} + +#[test] +fn set_visibility_tool_schema_and_metadata() { + let tool = StorageSetVisibilityTool::new(dummy_client()); + assert_eq!(tool.name(), "storage_set_visibility"); + assert_eq!(tool.permission_level(), PermissionLevel::Write); + assert_eq!(tool.category(), ToolCategory::Workflow); + assert!(tool.external_effect()); + + let schema = tool.parameters_schema(); + assert_eq!(schema["required"], json!(["file_id", "visibility"])); +} + +#[test] +fn delete_tool_schema_and_metadata() { + let tool = StorageDeleteFileTool::new(dummy_client()); + assert_eq!(tool.name(), "storage_delete_file"); + assert_eq!(tool.permission_level(), PermissionLevel::Write); + assert_eq!(tool.category(), ToolCategory::Workflow); + assert!(tool.external_effect()); + assert_eq!(tool.parameters_schema()["required"], json!(["file_id"])); +} + +// ── Arg validation (no network) ───────────────────────────────────── + +#[tokio::test] +async fn upload_rejects_missing_path_without_network() { + let tool = StorageUploadFileTool::new(dummy_client(), PathBuf::from("/tmp")); + let res = tool.execute(json!({})).await.unwrap(); + assert!(res.is_error); + assert!(res.output().contains("path is required")); +} + +#[tokio::test] +async fn upload_rejects_path_escaping_workspace() { + let tmp = tempfile::tempdir().unwrap(); + // A real file OUTSIDE the workspace root. + let outside = tempfile::tempdir().unwrap(); + let secret = outside.path().join("secret.txt"); + std::fs::write(&secret, b"nope").unwrap(); + + let tool = StorageUploadFileTool::new(dummy_client(), tmp.path().to_path_buf()); + + // Absolute path outside the workspace. + let res = tool + .execute(json!({ "path": secret.display().to_string() })) + .await + .unwrap(); + assert!(res.is_error, "absolute escape must be rejected: {res:?}"); + assert!(res.output().contains("escapes"), "got: {}", res.output()); + + // Relative traversal out of the workspace. + let rel = format!( + "../{}/secret.txt", + outside.path().file_name().unwrap().to_str().unwrap() + ); + let res = tool.execute(json!({ "path": rel })).await.unwrap(); + assert!(res.is_error, "relative escape must be rejected: {res:?}"); +} + +#[tokio::test] +async fn upload_rejects_nonexistent_file() { + let tmp = tempfile::tempdir().unwrap(); + let tool = StorageUploadFileTool::new(dummy_client(), tmp.path().to_path_buf()); + let res = tool + .execute(json!({ "path": "missing.txt" })) + .await + .unwrap(); + assert!(res.is_error); +} + +#[tokio::test] +async fn upload_rejects_bad_visibility_and_ttl() { + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("a.txt"), b"hi").unwrap(); + let tool = StorageUploadFileTool::new(dummy_client(), tmp.path().to_path_buf()); + + let res = tool + .execute(json!({ "path": "a.txt", "visibility": "everyone" })) + .await + .unwrap(); + assert!(res.is_error); + assert!(res.output().contains("visibility")); + + let res = tool + .execute(json!({ "path": "a.txt", "ttl_days": 0 })) + .await + .unwrap(); + assert!(res.is_error); + assert!(res.output().contains("ttl_days")); +} + +#[tokio::test] +async fn download_rejects_missing_or_invalid_file_id() { + let tool = StorageDownloadFileTool::new(dummy_client(), PathBuf::from("/tmp")); + let res = tool.execute(json!({})).await.unwrap(); + assert!(res.is_error); + let res = tool.execute(json!({ "file_id": "../etc" })).await.unwrap(); + assert!(res.is_error); +} + +#[tokio::test] +async fn get_link_rejects_missing_file_id() { + let tool = StorageGetLinkTool::new(dummy_client()); + let res = tool.execute(json!({})).await.unwrap(); + assert!(res.is_error); +} + +#[tokio::test] +async fn set_visibility_rejects_missing_or_invalid_args() { + let tool = StorageSetVisibilityTool::new(dummy_client()); + let res = tool.execute(json!({ "file_id": "f1" })).await.unwrap(); + assert!(res.is_error); + let res = tool + .execute(json!({ "file_id": "f1", "visibility": "hidden" })) + .await + .unwrap(); + assert!(res.is_error); +} + +#[tokio::test] +async fn delete_rejects_missing_file_id() { + let tool = StorageDeleteFileTool::new(dummy_client()); + let res = tool.execute(json!({})).await.unwrap(); + assert!(res.is_error); +} + +// ── Path / filename helpers ───────────────────────────────────────── + +#[test] +fn resolve_upload_path_accepts_inside_and_rejects_outside() { + let tmp = tempfile::tempdir().unwrap(); + let inner = tmp.path().join("sub"); + std::fs::create_dir_all(&inner).unwrap(); + let file = inner.join("data.bin"); + std::fs::write(&file, b"x").unwrap(); + + // Relative path inside the root resolves. + let ok = resolve_upload_path(tmp.path(), "sub/data.bin").unwrap(); + assert!(ok.ends_with("data.bin")); + + // Traversal escaping the root is rejected. + let err = resolve_upload_path(&inner, "../../etc/hosts").unwrap_err(); + assert!( + err.contains("escapes") || err.contains("not exist"), + "got: {err}" + ); + + // A directory is not uploadable. + let err = resolve_upload_path(tmp.path(), "sub").unwrap_err(); + assert!(err.contains("not a regular file"), "got: {err}"); +} + +#[test] +fn sanitize_filename_strips_separators_and_traversal() { + assert_eq!( + sanitize_filename("report.pdf").as_deref(), + Some("report.pdf") + ); + assert_eq!( + sanitize_filename("../../evil.sh").as_deref(), + Some("evil.sh") + ); + assert_eq!(sanitize_filename("a/b\\c.txt").as_deref(), Some("c.txt")); + assert_eq!(sanitize_filename("..").is_none(), true); + assert_eq!(sanitize_filename(" ").is_none(), true); +} + +// ── End-to-end flows against a mock backend (wiremock) ────────────── + +use wiremock::matchers::{body_string_contains, header, method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +fn client_for(server: &MockServer) -> Arc { + Arc::new(IntegrationClient::new(server.uri(), "tok".to_string())) +} + +#[tokio::test] +async fn upload_tool_posts_multipart_and_reports_file_id() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/agent-integrations/file-storage/files")) + .and(header("authorization", "Bearer tok")) + // Multipart body carries the file part + our extra form fields. + .and(body_string_contains("name=\"file\"")) + .and(body_string_contains("HELLO-BYTES")) + .and(body_string_contains("name=\"visibility\"")) + .and(body_string_contains("name=\"ttlDays\"")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "success": true, + "data": { + "fileId": "file-1", + "filename": "hello.txt", + "size": 11, + "contentType": "text/plain", + "visibility": "public", + "expiresAt": "2026-07-12T00:00:00.000Z", + "publicUrl": "https://api.example/agent-integrations/file-storage/public/file-1", + "costUsd": 0.0001 + } + }))) + .expect(1) + .mount(&server) + .await; + + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("hello.txt"), b"HELLO-BYTES").unwrap(); + + let tool = StorageUploadFileTool::new(client_for(&server), tmp.path().to_path_buf()); + let res = tool + .execute(json!({ "path": "hello.txt", "visibility": "public", "ttl_days": 7 })) + .await + .unwrap(); + + assert!(!res.is_error, "expected success, got {res:?}"); + let out = res.output(); + assert!(out.contains("file-1"), "output should carry file_id: {out}"); + assert!( + out.contains("public"), + "output should carry public url/visibility: {out}" + ); +} + +#[tokio::test] +async fn download_tool_follows_redirect_and_persists_file() { + let server = MockServer::start().await; + // The backend 302s to a presigned URL; reqwest follows it (same host in + // this test, but the redirect-following behavior is what's exercised). + let presigned = format!("{}/s3/blob", server.uri()); + Mock::given(method("GET")) + .and(path( + "/agent-integrations/file-storage/files/file-2/download", + )) + .and(header("authorization", "Bearer tok")) + .respond_with(ResponseTemplate::new(302).insert_header("Location", presigned.as_str())) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/s3/blob")) + .respond_with( + ResponseTemplate::new(200) + .set_body_raw(b"RAW-CONTENT".to_vec(), "text/plain") + .insert_header("Content-Disposition", "attachment; filename=\"notes.txt\""), + ) + .mount(&server) + .await; + + let tmp = tempfile::tempdir().unwrap(); + let tool = StorageDownloadFileTool::new(client_for(&server), tmp.path().to_path_buf()); + let res = tool.execute(json!({ "file_id": "file-2" })).await.unwrap(); + + assert!(!res.is_error, "expected success, got {res:?}"); + let saved = tmp.path().join("storage-downloads").join("notes.txt"); + assert_eq!(std::fs::read(&saved).unwrap(), b"RAW-CONTENT"); + assert!(res.output().contains("notes.txt")); +} + +#[tokio::test] +async fn download_tool_honors_explicit_filename() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path( + "/agent-integrations/file-storage/files/file-3/download", + )) + .respond_with(ResponseTemplate::new(200).set_body_raw(b"BYTES".to_vec(), "application/pdf")) + .mount(&server) + .await; + + let tmp = tempfile::tempdir().unwrap(); + let tool = StorageDownloadFileTool::new(client_for(&server), tmp.path().to_path_buf()); + let res = tool + .execute(json!({ "file_id": "file-3", "filename": "../sneaky/mine.pdf" })) + .await + .unwrap(); + + assert!(!res.is_error, "expected success, got {res:?}"); + // Traversal in the requested name is stripped to the basename. + let saved = tmp.path().join("storage-downloads").join("mine.pdf"); + assert_eq!(std::fs::read(&saved).unwrap(), b"BYTES"); +} + +#[tokio::test] +async fn list_tool_renders_files_and_usage() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/agent-integrations/file-storage/files")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "success": true, + "data": { + "files": [ + { + "fileId": "file-1", + "filename": "hello.txt", + "size": 11, + "contentType": "text/plain", + "visibility": "public", + "expiresAt": "2026-07-12T00:00:00.000Z", + "createdAt": "2026-07-05T00:00:00.000Z", + "publicUrl": "https://api.example/agent-integrations/file-storage/public/file-1" + } + ], + "usage": { "usedBytes": 11, "limitBytes": 1073741824 } + } + }))) + .mount(&server) + .await; + + let tool = StorageListFilesTool::new(client_for(&server)); + let res = tool.execute(json!({})).await.unwrap(); + assert!(!res.is_error, "expected success, got {res:?}"); + let out = res.output(); + assert!(out.contains("file-1")); + assert!(out.contains("hello.txt")); +} + +#[tokio::test] +async fn get_link_tool_posts_expiry_and_returns_url() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/agent-integrations/file-storage/files/file-1/link")) + .and(body_string_contains("expiresInSeconds")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "success": true, + "data": { + "url": "https://s3.example/presigned?sig=abc", + "expiresAt": "2026-07-05T01:00:00.000Z", + "costUsd": 0.0001 + } + }))) + .mount(&server) + .await; + + let tool = StorageGetLinkTool::new(client_for(&server)); + let res = tool + .execute(json!({ "file_id": "file-1", "expires_in_seconds": 120 })) + .await + .unwrap(); + assert!(!res.is_error, "expected success, got {res:?}"); + assert!(res.output().contains("https://s3.example/presigned")); +} + +#[tokio::test] +async fn set_visibility_tool_patches_and_reports_public_url() { + let server = MockServer::start().await; + Mock::given(method("PATCH")) + .and(path("/agent-integrations/file-storage/files/file-1")) + .and(body_string_contains("\"visibility\":\"public\"")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "success": true, + "data": { + "fileId": "file-1", + "filename": "hello.txt", + "size": 11, + "visibility": "public", + "expiresAt": "2026-07-12T00:00:00.000Z", + "publicUrl": "https://api.example/agent-integrations/file-storage/public/file-1" + } + }))) + .mount(&server) + .await; + + let tool = StorageSetVisibilityTool::new(client_for(&server)); + let res = tool + .execute(json!({ "file_id": "file-1", "visibility": "public" })) + .await + .unwrap(); + assert!(!res.is_error, "expected success, got {res:?}"); + assert!(res.output().contains("public")); + assert!(res.output().contains("file-storage/public/file-1")); +} + +#[tokio::test] +async fn delete_tool_deletes_and_confirms() { + let server = MockServer::start().await; + Mock::given(method("DELETE")) + .and(path("/agent-integrations/file-storage/files/file-1")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "success": true, + "data": { "deleted": true } + }))) + .mount(&server) + .await; + + let tool = StorageDeleteFileTool::new(client_for(&server)); + let res = tool.execute(json!({ "file_id": "file-1" })).await.unwrap(); + assert!(!res.is_error, "expected success, got {res:?}"); + assert!( + res.output().contains("\"deleted\": true"), + "got: {}", + res.output() + ); + assert!(res.output_for_llm(true).contains("Deleted")); +} + +#[tokio::test] +async fn tools_surface_backend_envelope_errors() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/agent-integrations/file-storage/files")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "success": false, + "error": "Insufficient balance" + }))) + .mount(&server) + .await; + + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("a.txt"), b"hi").unwrap(); + let tool = StorageUploadFileTool::new(client_for(&server), tmp.path().to_path_buf()); + let res = tool.execute(json!({ "path": "a.txt" })).await.unwrap(); + assert!(res.is_error); + assert!( + res.output().contains("Insufficient balance"), + "got: {}", + res.output() + ); +} diff --git a/src/openhuman/file_storage/types.rs b/src/openhuman/file_storage/types.rs new file mode 100644 index 000000000..b9c8088f3 --- /dev/null +++ b/src/openhuman/file_storage/types.rs @@ -0,0 +1,104 @@ +//! Shared types for the `file_storage` agent tools. +//! +//! These mirror the backend's `file_storage` integration contract +//! (`/agent-integrations/file-storage/*`) — see the file-storage API contract. +//! Every response arrives inside the standard `{ success, data, error }` +//! envelope handled by `IntegrationClient`; these structs are the `data` +//! payloads. + +use serde::Deserialize; +use serde_json::json; + +/// Metadata for a stored file as returned by list / metadata / visibility +/// endpoints. +#[derive(Debug, Clone, Deserialize)] +pub struct FileMeta { + #[serde(rename = "fileId")] + pub file_id: String, + pub filename: String, + #[serde(default)] + pub size: u64, + #[serde(rename = "contentType", default)] + pub content_type: Option, + #[serde(default)] + pub visibility: String, + #[serde(rename = "expiresAt", default)] + pub expires_at: Option, + #[serde(rename = "createdAt", default)] + pub created_at: Option, + /// Stable public URL — present only when `visibility == "public"`. + #[serde(rename = "publicUrl", default)] + pub public_url: Option, +} + +impl FileMeta { + pub fn to_json(&self) -> serde_json::Value { + json!({ + "file_id": self.file_id, + "filename": self.filename, + "size": self.size, + "content_type": self.content_type, + "visibility": self.visibility, + "expires_at": self.expires_at, + "created_at": self.created_at, + "public_url": self.public_url, + }) + } +} + +/// `POST /files` (multipart upload) response. +#[derive(Debug, Clone, Deserialize)] +pub struct UploadResponse { + #[serde(rename = "fileId")] + pub file_id: String, + pub filename: String, + #[serde(default)] + pub size: u64, + #[serde(rename = "contentType", default)] + pub content_type: Option, + #[serde(default)] + pub visibility: String, + #[serde(rename = "expiresAt", default)] + pub expires_at: Option, + #[serde(rename = "publicUrl", default)] + pub public_url: Option, + #[serde(rename = "costUsd", default)] + pub cost_usd: f64, +} + +/// Storage usage summary embedded in the list response. +#[derive(Debug, Clone, Deserialize, Default)] +pub struct StorageUsage { + #[serde(rename = "usedBytes", default)] + pub used_bytes: u64, + #[serde(rename = "limitBytes", default)] + pub limit_bytes: u64, +} + +/// `GET /files` response. +#[derive(Debug, Clone, Deserialize)] +pub struct ListFilesResponse { + #[serde(default)] + pub files: Vec, + #[serde(rename = "nextCursor", default)] + pub next_cursor: Option, + #[serde(default)] + pub usage: StorageUsage, +} + +/// `POST /files/:fileId/link` response — a short-lived presigned GET URL. +#[derive(Debug, Clone, Deserialize)] +pub struct LinkResponse { + pub url: String, + #[serde(rename = "expiresAt", default)] + pub expires_at: Option, + #[serde(rename = "costUsd", default)] + pub cost_usd: f64, +} + +/// `DELETE /files/:fileId` response. +#[derive(Debug, Clone, Deserialize)] +pub struct DeleteResponse { + #[serde(default)] + pub deleted: bool, +} diff --git a/src/openhuman/integrations/client.rs b/src/openhuman/integrations/client.rs index e90b8150d..2c5421c28 100644 --- a/src/openhuman/integrations/client.rs +++ b/src/openhuman/integrations/client.rs @@ -206,6 +206,40 @@ fn sanitize_backend_url(backend_url: &str) -> String { } } +/// Extract the `filename` (or RFC 5987 `filename*`) parameter from a +/// `Content-Disposition` header value, e.g. +/// `attachment; filename="report.pdf"` → `Some("report.pdf")`. +/// Best-effort: unparseable values yield `None` and callers fall back to +/// their own naming scheme. +fn parse_content_disposition_filename(value: &str) -> Option { + for part in value.split(';') { + let part = part.trim(); + let lower = part.to_ascii_lowercase(); + if let Some(rest) = lower + .starts_with("filename*=") + .then(|| &part["filename*=".len()..]) + { + // RFC 5987: filename*=UTF-8''percent-encoded — keep the tail + // after the last `'` and leave percent-decoding to callers who + // care (the raw form is still a usable, safe name). + let tail = rest.rsplit('\'').next().unwrap_or(rest); + let trimmed = tail.trim_matches('"').trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } else if let Some(rest) = lower + .starts_with("filename=") + .then(|| &part["filename=".len()..]) + { + let trimmed = rest.trim().trim_matches('"').trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + None +} + /// Shared client for all integration tools. Holds backend URL, auth token, /// a reusable `reqwest::Client`, and a lazily-fetched pricing cache. pub struct IntegrationClient { @@ -472,6 +506,227 @@ impl IntegrationClient { .ok_or_else(|| anyhow::anyhow!("Backend returned success but no data for GET {}", url)) } + /// Render a reqwest transport error with its full source chain (mirrors + /// the inline closures in [`Self::post`] / [`Self::get`]) and route it + /// through the observability classifier so network-environment failures + /// skip Sentry. + fn report_transport_error( + e: reqwest::Error, + method: &str, + path: &str, + url: &str, + ) -> anyhow::Error { + let mut chain = format!("{e}"); + let mut src: Option<&(dyn std::error::Error + 'static)> = e.source(); + while let Some(s) = src { + chain.push_str(" → "); + chain.push_str(&s.to_string()); + src = s.source(); + } + crate::core::observability::report_error_or_expected( + chain.as_str(), + "integrations", + method, + &[("path", path), ("failure", "transport")], + ); + anyhow::anyhow!("{} {} failed: {}", method.to_uppercase(), url, chain) + } + + /// Shared non-2xx / 401 / envelope handling for the newer HTTP verbs + /// (`patch`, `delete`, `upload_multipart`). Mirrors the [`Self::post`] + /// error classification: 401 → session-expiry recovery flow, other + /// non-2xx → demoted/classified generic error, then `BackendResponse` + /// envelope parsing with `success:false` classification. + async fn parse_json_response( + method: &str, + path: &str, + url: &str, + resp: reqwest::Response, + ) -> anyhow::Result { + let status = resp.status(); + let method_upper = method.to_uppercase(); + if !status.is_success() { + let body_text = resp.text().await.unwrap_or_default(); + let detail = extract_error_detail(&body_text, MAX_ERROR_BODY_LEN); + let status_str = status.as_u16().to_string(); + // Session-JWT rejection — same argument as in post()/get(): the + // backend auth middleware is the only 401 source on these routes. + if status == reqwest::StatusCode::UNAUTHORIZED { + let message = handle_session_jwt_unauthorized(&method_upper, path, url, &detail); + anyhow::bail!("{message}"); + } + crate::core::observability::report_error_or_expected( + format!("Backend returned {status} for {method_upper} {url}: {detail}").as_str(), + "integrations", + method, + &[ + ("path", path), + ("status", status_str.as_str()), + ("failure", "non_2xx"), + ], + ); + anyhow::bail!("Backend returned {status} for {method_upper} {url}: {detail}"); + } + + let envelope: BackendResponse = resp.json().await?; + if !envelope.success { + let msg = envelope + .error + .unwrap_or_else(|| "unknown backend error".into()); + crate::core::observability::report_error_or_expected( + msg.as_str(), + "integrations", + method, + &[("path", path), ("failure", "envelope_error")], + ); + anyhow::bail!("Backend error for {} {}: {}", method_upper, url, msg); + } + envelope.data.ok_or_else(|| { + anyhow::anyhow!( + "Backend returned success but no data for {} {}", + method_upper, + url + ) + }) + } + + /// PATCH JSON to a backend endpoint and parse the response `data` field. + /// Mirrors [`Self::post`] (auth header, 401 → session-expiry, error + /// classification). + pub async fn patch( + &self, + path: &str, + body: &serde_json::Value, + ) -> anyhow::Result { + self.ensure_budget_available(path).await?; + let url = crate::api::config::api_url(&self.backend_url, path); + tracing::debug!("[integrations] PATCH {}", url); + + let resp = self + .http_client + .patch(&url) + .header("Authorization", format!("Bearer {}", self.auth_token)) + .header("Content-Type", "application/json") + .json(body) + .send() + .await + .map_err(|e| Self::report_transport_error(e, "patch", path, &url))?; + + Self::parse_json_response("patch", path, &url, resp).await + } + + /// DELETE a backend resource and parse the response `data` field. + /// Mirrors [`Self::post`] (auth header, 401 → session-expiry, error + /// classification). + pub async fn delete(&self, path: &str) -> anyhow::Result { + self.ensure_budget_available(path).await?; + let url = crate::api::config::api_url(&self.backend_url, path); + tracing::debug!("[integrations] DELETE {}", url); + + let resp = self + .http_client + .delete(&url) + .header("Authorization", format!("Bearer {}", self.auth_token)) + .send() + .await + .map_err(|e| Self::report_transport_error(e, "delete", path, &url))?; + + Self::parse_json_response("delete", path, &url, resp).await + } + + /// POST a `multipart/form-data` body to a backend endpoint and parse the + /// response `data` field. Mirrors [`Self::post`] URL building, Bearer + /// auth, 401 → session-expiry handling and error classification; the + /// content type is set by reqwest from the form boundary. + pub async fn upload_multipart( + &self, + path: &str, + form: reqwest::multipart::Form, + ) -> anyhow::Result { + self.ensure_budget_available(path).await?; + let url = crate::api::config::api_url(&self.backend_url, path); + tracing::debug!("[integrations] POST(multipart) {}", url); + + let resp = self + .http_client + .post(&url) + .header("Authorization", format!("Bearer {}", self.auth_token)) + .multipart(form) + .send() + .await + .map_err(|e| Self::report_transport_error(e, "post_multipart", path, &url))?; + + Self::parse_json_response("post_multipart", path, &url, resp).await + } + + /// Authenticated GET returning the raw response body plus content-type and + /// any `Content-Disposition` filename. Used for backend download routes + /// that `302`-redirect to a presigned S3 URL: reqwest follows redirects by + /// default and its redirect policy strips sensitive headers (including + /// `Authorization`) on cross-host hops, so the bearer token never leaks to + /// S3 while the presigned URL still authorizes the fetch. + pub async fn get_bytes( + &self, + path: &str, + ) -> anyhow::Result<(bytes::Bytes, Option, Option)> { + self.ensure_budget_available(path).await?; + let url = crate::api::config::api_url(&self.backend_url, path); + tracing::debug!("[integrations] GET(bytes) {}", url); + + let resp = self + .http_client + .get(&url) + .header("Authorization", format!("Bearer {}", self.auth_token)) + .send() + .await + .map_err(|e| Self::report_transport_error(e, "get_bytes", path, &url))?; + + let status = resp.status(); + if !status.is_success() { + let body_text = resp.text().await.unwrap_or_default(); + let detail = extract_error_detail(&body_text, MAX_ERROR_BODY_LEN); + let status_str = status.as_u16().to_string(); + if status == reqwest::StatusCode::UNAUTHORIZED { + let message = handle_session_jwt_unauthorized("GET", path, &url, &detail); + anyhow::bail!("{message}"); + } + crate::core::observability::report_error_or_expected( + format!("Backend returned {status} for GET {url}: {detail}").as_str(), + "integrations", + "get_bytes", + &[ + ("path", path), + ("status", status_str.as_str()), + ("failure", "non_2xx"), + ], + ); + anyhow::bail!("Backend returned {status} for GET {url}: {detail}"); + } + + let content_type = resp + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()); + let filename = resp + .headers() + .get(reqwest::header::CONTENT_DISPOSITION) + .and_then(|v| v.to_str().ok()) + .and_then(parse_content_disposition_filename); + + let body = resp + .bytes() + .await + .map_err(|e| anyhow::anyhow!("failed to read response body for GET {}: {}", url, e))?; + tracing::debug!( + "[integrations] GET(bytes) {} → {} bytes (content_type={:?})", + url, + body.len(), + content_type + ); + Ok((body, content_type, filename)) + } + /// Fetch and cache pricing info from the backend. Returns a default /// (empty) pricing struct on network errors so tool registration never fails. pub async fn pricing(&self) -> &IntegrationPricing { diff --git a/src/openhuman/mcp_server/resources.rs b/src/openhuman/mcp_server/resources.rs index a289ad296..fe797e7ad 100644 --- a/src/openhuman/mcp_server/resources.rs +++ b/src/openhuman/mcp_server/resources.rs @@ -247,6 +247,18 @@ const RESOURCE_CATALOG: &[PromptResource] = &[ description: "Specialist worker for screen context and desktop state inspection.", content: include_str!("../agent_registry/agents/screen_awareness_agent/prompt.md"), }, + PromptResource { + uri: "openhuman://prompts/agents/flow_discovery", + name: "flow_discovery", + description: "Flow Scout — read-only workflow discovery agent that suggests automations from memory, threads, and integrations.", + content: include_str!("../agent_registry/agents/flow_discovery/prompt.md"), + }, + PromptResource { + uri: "openhuman://prompts/agents/workflow_builder", + name: "workflow_builder", + description: "Workflow authoring specialist that builds tinyflows automation graphs and returns proposals for review.", + content: include_str!("../agent_registry/agents/workflow_builder/prompt.md"), + }, PromptResource { uri: "openhuman://prompts/agents/agent_memory", name: "agent_memory", diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index b977df0e7..c89f88ff8 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -50,6 +50,7 @@ pub mod doctor; pub mod embeddings; pub mod encryption; pub mod file_state; +pub mod file_storage; pub mod flows; pub mod harness_init; pub mod health; diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 3b6f1aa03..b44b59568 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -814,6 +814,13 @@ pub fn all_tools_with_runtime( action_dir, )); + // Managed cloud file storage (S3 via the backend). Skipped when no + // integration client is configured; downloads land under `action_dir`. + tools.extend(crate::openhuman::file_storage::build_file_storage_tools( + root_config, + action_dir, + )); + // High-level web3 tools (swaps / bridges / dapp calls) built on the wallet. // They call the backend deBridge proxy per-invocation and error gracefully // when the user is not signed in, so they register unconditionally.