From 2b35312abdb406bf8ebee51420cd13357853d605 Mon Sep 17 00:00:00 2001 From: xavix-yo Date: Sat, 18 Jul 2026 18:02:45 +0100 Subject: [PATCH] =?UTF-8?q?feat(media):=20multimodal=20attachments=20?= =?UTF-8?q?=E2=80=94=20image/video=20inlining=20per=20model=20capabilities?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds media.rs for per-turn attachment routing, message_builder partitioning by resolved model capabilities, controller endpoints for uploads, and server routing for /data/* behind session auth. See CLAUDE.md §Multimodal attachments for full design. --- CLAUDE.md | 9 +- crates/llm-client/src/anthropic.rs | 74 ++++- crates/skald-core/src/llm/manager.rs | 1 + crates/skald-core/src/llm/mod.rs | 3 + .../skald-core/src/llm/providers/declared.rs | 22 +- .../src/session/handler/llm_call.rs | 6 +- .../src/session/handler/llm_loop.rs | 2 +- .../skald-core/src/session/handler/media.rs | 308 ++++++++++++++++++ .../src/session/handler/message_builder.rs | 170 +++++++++- .../src/session/handler/messages.rs | 3 +- crates/skald-core/src/session/handler/mod.rs | 1 + providers.yaml | 4 +- src/frontend/api/mod.rs | 4 + src/frontend/api/uploads.rs | 43 ++- src/frontend/server.rs | 16 +- 15 files changed, 631 insertions(+), 35 deletions(-) create mode 100644 crates/skald-core/src/session/handler/media.rs diff --git a/CLAUDE.md b/CLAUDE.md index 90655d6..04c0424 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -64,7 +64,7 @@ Two rules keep the boundary real, and both are enforced by the compiler: | `src/main.rs` | Thin entry point: tracing → `Skald::new` → `WebFrontend::start` → shutdown. Branches on the `desktop` feature: under `--features desktop` enters `desktop::run()` (Tauri event loop) instead of blocking on a tokio runtime. Exposes `run_backend()` / `shutdown_backend()` shared by both entry points | | `src/desktop/mod.rs` | Tauri shell — **only compiled under `--features desktop`**. Builds the system-tray icon + menu (`Open` / `Quit`), creates the main `WebviewWindow` (URL = `http://127.0.0.1:{config.port}`), spawns the backend on Tauri's shared tokio runtime, handles graceful shutdown. Holds the `OnceLock`, and installs the core's restart handler. See [docs/desktop.md](docs/desktop.md) | | `crates/skald-core/src/skald/` | `Skald` — headless application core. `mod.rs` (struct + staged `new()` / `shutdown()`), `runtime.rs` (cross-cutting `Runtime` context), `bundles.rs` (8 domain bundles + `build()`), `wiring.rs` (`wire()` + `spawn_background()`), `supervisor.rs` (`TaskSupervisor`), `accessors.rs` (per-manager accessor facade — the API surface the frontend uses) | -| `crates/skald-core/src/session/handler/` | Core LLM loop — `mod.rs`, `llm_loop.rs` (`run_agent_turn`), `agent_dispatch.rs`, `dispatcher.rs`, `approval.rs`, `resume.rs`, `messages.rs`, `config.rs`, `interface_tools.rs` | +| `crates/skald-core/src/session/handler/` | Core LLM loop — `mod.rs`, `llm_loop.rs` (`run_agent_turn`), `agent_dispatch.rs`, `dispatcher.rs`, `approval.rs`, `resume.rs`, `messages.rs`, `config.rs`, `interface_tools.rs`, `media.rs` (multimodal attachments — see below) | | `crates/skald-core/src/session/manager.rs` | Creates/retrieves `ChatSessionHandler` per session | | `crates/skald-core/src/chat_hub/` | `ChatHub`: broadcast events to all connected WS clients | | `crates/skald-core/src/chat_event_bus.rs` | Global async bus for cross-session events | @@ -162,8 +162,13 @@ OAuth2 authorization-code + PKCE is wired for per-user connectors (Gmail is the **Deferred:** the other §15 interactive kinds (QR / SSH via elicitation) — `deliver.as=file` and non-Google providers are unimplemented paths that error clearly rather than half-work. No boot seed of catalog presets; the admin populates the catalog from the Marketplace. -## Sub-agent system +## Multimodal attachments +Uploads (`POST /api/{source}/uploads`) are saved per-user under `data/uploads/{userid}/{session_id}/` (older rows may still reference the pre-namespacing `data/uploads/{session_id}/` layout — both stay readable), streamed to disk with a 256 MiB cap, with the sniffed magic-byte MIME preferred over the client claim; `/data/*` is served behind the same session-cookie gate as `/api`. Attachment metadata travels as structured JSON in `chat_history.metadata` — never as persisted text. + +At context-build time (`MessageBuilder`), attachments of the **current turn** (the user/agent rows following the last completed assistant reply, including across in-flight tool rounds) are partitioned by `session/handler/media.rs`: when the resolved model's `LlmEntry.capabilities` include the modality (`vision` → `image_url` parts, `video` → `video_url` parts), the file is inlined as a base64 data-URL content part — but only if it canonicalizes under `data/uploads/`, its sniffed MIME is in the allowlist, and it fits the budgets (4 files / 10 MiB image / 32 MiB video / 48 MiB total per turn). Everything else — older turns, other kinds, any failed check — keeps the textual `[SYSTEM INFO]` path block, so a non-vision model produces a byte-identical payload to before. `OpenAiClient` forwards parts verbatim; `AnthropicClient` translates `image_url` data URLs to `image` blocks (video unsupported; Anthropic models get `vision` by editing the model row's capabilities — no catalog refresh writes them). On LLM fallback mid-round, messages are rebuilt with the replacement model's capabilities. + +## Sub-agent system - Synchronous sub-agents (`execute_task` mode=sync / `execute_subtask`) are **not** plain `Tool`s — they are intercepted in `run_agent_turn` before registry dispatch. - `dispatch_sub_agent` (in `agent_dispatch.rs`) creates a child `chat_sessions_stack` row and runs `run_agent_turn` **recursively in the same task**, holding the same `processing` lock and sharing the same cancellation token. The child's result string becomes the parent tool call's result (completion lives in one place — the `run_agent_turn` tool-result match); then it terminates the child frame. There is no task-spawn / `WaitingChild` / resume cascade for the sync path. - Max recursion depth: `MAX_AGENT_DEPTH = 5`. diff --git a/crates/llm-client/src/anthropic.rs b/crates/llm-client/src/anthropic.rs index 2779462..34a981c 100644 --- a/crates/llm-client/src/anthropic.rs +++ b/crates/llm-client/src/anthropic.rs @@ -97,7 +97,7 @@ impl AnthropicClient { "user" => { out.push(json!({ "role": "user", - "content": msg["content"].as_str().unwrap_or(""), + "content": convert_user_content(&msg["content"]), })); i += 1; } @@ -159,6 +159,45 @@ impl AnthropicClient { } } +/// User content arrives either as a plain string or as an OpenAI-style parts +/// array (text + `image_url` data URLs, produced when the resolved model has +/// the `vision` capability). Strings pass through; parts become Anthropic +/// blocks. Video and unknown parts are dropped with a warning — providers +/// gate capabilities upstream, so this should only indicate a misconfigured +/// model row. +fn convert_user_content(content: &Value) -> Value { + let Some(parts) = content.as_array() else { + return Value::String(content.as_str().unwrap_or("").to_string()); + }; + let mut blocks = Vec::new(); + for p in parts { + match p["type"].as_str().unwrap_or("") { + "text" => blocks.push(json!({ + "type": "text", + "text": p["text"].as_str().unwrap_or(""), + })), + "image_url" => { + if let Some(block) = parse_data_image(&p["image_url"]) { + blocks.push(block); + } + } + other => tracing::warn!(part_type = other, "dropping content part unsupported by Anthropic"), + } + } + Value::Array(blocks) +} + +/// `{"url": "data:;base64,"}` (or the bare-string shorthand) → an +/// Anthropic base64 image block. Only data URLs are supported. +fn parse_data_image(image_url: &Value) -> Option { + let url = image_url["url"].as_str().or_else(|| image_url.as_str())?; + let (mime, data) = url.strip_prefix("data:")?.split_once(";base64,")?; + Some(json!({ + "type": "image", + "source": { "type": "base64", "media_type": mime, "data": data }, + })) +} + #[async_trait] impl ChatbotClient for AnthropicClient { async fn chat( @@ -364,3 +403,36 @@ impl ChatbotClient for AnthropicClient { Ok((turn, Some(raw_meta))) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn user_content_string_passthrough() { + let v = convert_user_content(&json!("hello")); + assert_eq!(v, json!("hello")); + } + + #[test] + fn user_content_parts_become_anthropic_blocks() { + let v = convert_user_content(&json!([ + { "type": "text", "text": "what is this?" }, + { "type": "image_url", "image_url": { "url": "data:image/png;base64,QUJD" } }, + ])); + assert_eq!(v, json!([ + { "type": "text", "text": "what is this?" }, + { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": "QUJD" } }, + ])); + } + + #[test] + fn user_content_drops_video_and_non_data_urls() { + let v = convert_user_content(&json!([ + { "type": "text", "text": "t" }, + { "type": "video_url", "video_url": { "url": "data:video/mp4;base64,QUJD" } }, + { "type": "image_url", "image_url": { "url": "https://example.com/x.png" } }, + ])); + assert_eq!(v, json!([{ "type": "text", "text": "t" }])); + } +} diff --git a/crates/skald-core/src/llm/manager.rs b/crates/skald-core/src/llm/manager.rs index ec6d880..8ae55af 100644 --- a/crates/skald-core/src/llm/manager.rs +++ b/crates/skald-core/src/llm/manager.rs @@ -530,6 +530,7 @@ fn build_entry( extra_params: extra, context_length: model.context_length, prompt_cache, + capabilities: model.capabilities.clone(), }) } diff --git a/crates/skald-core/src/llm/mod.rs b/crates/skald-core/src/llm/mod.rs index 006af25..b89d4f5 100644 --- a/crates/skald-core/src/llm/mod.rs +++ b/crates/skald-core/src/llm/mod.rs @@ -23,6 +23,9 @@ pub struct LlmEntry { pub context_length: Option, /// When true, prompt-caching hints are injected into requests. pub prompt_cache: bool, + /// Input capabilities of the resolved model (`vision`, `video`, …), from + /// `llm_models.capabilities`. Drives multimodal attachment inlining. + pub capabilities: Vec, } // ── Provider ────────────────────────────────────────────────────────────────── diff --git a/crates/skald-core/src/llm/providers/declared.rs b/crates/skald-core/src/llm/providers/declared.rs index c11388f..462b539 100644 --- a/crates/skald-core/src/llm/providers/declared.rs +++ b/crates/skald-core/src/llm/providers/declared.rs @@ -429,13 +429,25 @@ fn apply_enrich(rules: &[EnrichRule], info: &mut RemoteLlmModelInfo) { set(&mut info.context_length, rule.context_length); set(&mut info.max_completion_tokens, rule.max_completion_tokens); if let Some(v) = rule.vision { - match rule.mode { + let applied = match rule.mode { EnrichMode::Fill => { if info.vision.is_none() { info.vision = Some(v); + true + } else { + false } } - EnrichMode::Override => info.vision = Some(v), + EnrichMode::Override => { + info.vision = Some(v); + true + } + }; + // Keep the capability in sync with the flag — the DB metadata writer + // stores the capabilities vec, so an enrich-set `vision: true` must + // also unlock the `vision` capability for multimodal inlining. + if applied && v && !info.capabilities.iter().any(|c| c == "vision") { + info.capabilities.push("vision".to_string()); } } for cap in &rule.add_capabilities { @@ -777,7 +789,7 @@ mod tests { let rules: Vec = serde_yaml::from_str( r#" - { match: "*coder*", context_length: 16384, mode: override } - - { match: "k3*", context_length: 1048576, vision: true } + - { match: "k3*", context_length: 1048576, vision: true, add_capabilities: [video] } "#, ) .unwrap(); @@ -801,6 +813,10 @@ mod tests { apply_enrich(&rules, &mut info); assert_eq!(info.context_length, Some(999)); // fill keeps the endpoint value assert_eq!(info.vision, Some(true)); + // An enrich-set `vision: true` also unlocks the capability, and + // add_capabilities are unioned in. + assert!(info.capabilities.iter().any(|c| c == "vision")); + assert!(info.capabilities.iter().any(|c| c == "video")); } /// The catalog shipped at the repository root must always parse: the file diff --git a/crates/skald-core/src/session/handler/llm_call.rs b/crates/skald-core/src/session/handler/llm_call.rs index 399def8..2895f35 100644 --- a/crates/skald-core/src/session/handler/llm_call.rs +++ b/crates/skald-core/src/session/handler/llm_call.rs @@ -122,12 +122,14 @@ impl ChatSessionHandler { *cur_name = next_name; *cur_llm = next_llm; // Rebuild messages if the new model uses different prompt_cache - // settings (e.g. switching from OpenRouter/Anthropic to DeepSeek). + // settings (e.g. switching from OpenRouter/Anthropic to DeepSeek) + // or different input capabilities (a non-vision fallback drops + // inline media back to the textual path block). match self.build_openai_messages( &self.db, stack_id, &config.agent_id, config.extra_system.as_deref(), config.extra_system_dynamic.as_deref(), config.tail_reminder.as_deref(), active_grants, - &config.system_substitutions, cur_llm.prompt_cache, + &config.system_substitutions, cur_llm.prompt_cache, &cur_llm.capabilities, ).await { Ok(m) => *messages = m, Err(e) => return RoundLlm::Failed(e), diff --git a/crates/skald-core/src/session/handler/llm_loop.rs b/crates/skald-core/src/session/handler/llm_loop.rs index acdb9d1..e7432dc 100644 --- a/crates/skald-core/src/session/handler/llm_loop.rs +++ b/crates/skald-core/src/session/handler/llm_loop.rs @@ -118,7 +118,7 @@ impl ChatSessionHandler { // Messages are (re)built with the current model's prompt_cache flag. // On fallback within the same round `call_llm_round` rebuilds them again // if the replacement model has a different prompt_cache setting. - let mut messages = self.build_openai_messages(pool, stack_id, &config.agent_id, config.extra_system.as_deref(), config.extra_system_dynamic.as_deref(), config.tail_reminder.as_deref(), &active_grants_snapshot, &config.system_substitutions, cur_llm.prompt_cache).await?; + let mut messages = self.build_openai_messages(pool, stack_id, &config.agent_id, config.extra_system.as_deref(), config.extra_system_dynamic.as_deref(), config.tail_reminder.as_deref(), &active_grants_snapshot, &config.system_substitutions, cur_llm.prompt_cache, &cur_llm.capabilities).await?; let tool_defs = config.all_tool_defs(); // Record every tool actually offered to the LLM so the Security-groups diff --git a/crates/skald-core/src/session/handler/media.rs b/crates/skald-core/src/session/handler/media.rs new file mode 100644 index 0000000..0dec385 --- /dev/null +++ b/crates/skald-core/src/session/handler/media.rs @@ -0,0 +1,308 @@ +//! Inline multimodal media for chat attachments. +//! +//! Attachments normally reach the model as a textual list of paths (see +//! `attachments_block`) and the agent decides whether to read them. When the +//! resolved model declares a matching capability (`vision`, `video`), media +//! attachments of the **current turn** are instead sent as native content +//! parts — `image_url` / `video_url` data URLs, the OpenAI wire shape, which +//! non-OpenAI clients translate — so the model actually sees the bytes. +//! +//! Promotion is deliberately strict: an attachment is inlined only when ALL of +//! these hold — +//! - the model has the modality's capability; +//! - the file lives under `data/uploads/`, canonicalized (attachments saved +//! anywhere else, e.g. by the Telegram plugin, stay textual); +//! - the sniffed magic bytes match an allowed MIME — the client-supplied +//! `mimetype` is never trusted; +//! - the per-file and per-turn byte/count budgets are not exhausted. +//! +//! Anything failing a check silently stays on the textual path. + +use std::path::Path; + +use base64::Engine as _; +use serde_json::{json, Value}; +use tracing::debug; + +use core_api::message_meta::Attachment; + +/// Max media parts inlined per turn. +const MAX_MEDIA_PER_TURN: usize = 4; +/// Max bytes for one inlined image. +const MAX_IMAGE_BYTES: u64 = 10 * 1024 * 1024; +/// Max bytes for one inlined video. +const MAX_VIDEO_BYTES: u64 = 32 * 1024 * 1024; +/// Max combined media bytes inlined per turn. +const MAX_TOTAL_MEDIA_BYTES: u64 = 48 * 1024 * 1024; + +/// A model-input modality: the capability that unlocks it, the content-part +/// type it maps to, its byte cap and the sniffed MIME types accepted. +struct Modality { + capability: &'static str, + part_type: &'static str, + max_bytes: u64, + mimes: &'static [&'static str], +} + +const MODALITIES: &[Modality] = &[ + Modality { + capability: "vision", + part_type: "image_url", + max_bytes: MAX_IMAGE_BYTES, + mimes: &["image/png", "image/jpeg", "image/gif", "image/webp"], + }, + Modality { + capability: "video", + part_type: "video_url", + max_bytes: MAX_VIDEO_BYTES, + mimes: &[ + "video/mp4", + "video/mpeg", + "video/quicktime", + "video/webm", + "video/x-msvideo", + "video/x-flv", + "video/3gpp", + ], + }, +]; + +/// The result of partitioning a message's attachments. +pub struct MediaPartition { + /// OpenAI-style content parts, ready to append after the text part. + pub parts: Vec, + /// Attachments that stay on the textual path block. + pub rest: Vec, +} + +/// Splits a message's attachments into inline media parts and leftovers. +/// Files are resolved against the process working directory. +pub async fn partition(attachments: &[Attachment], capabilities: &[String]) -> MediaPartition { + let base = std::env::current_dir().unwrap_or_default(); + partition_under(attachments, capabilities, &base).await +} + +/// [`partition`] with an explicit base directory (tests). +pub async fn partition_under( + attachments: &[Attachment], + capabilities: &[String], + base: &Path, +) -> MediaPartition { + let capable = MODALITIES + .iter() + .any(|m| capabilities.iter().any(|c| c == m.capability)); + let root = std::fs::canonicalize(base.join("data").join("uploads")).ok(); + if !capable || root.is_none() { + return MediaPartition { parts: Vec::new(), rest: attachments.to_vec() }; + } + let root = root.unwrap(); + + let mut parts: Vec = Vec::new(); + let mut rest: Vec = Vec::new(); + let mut total: u64 = 0; + for a in attachments { + if parts.len() >= MAX_MEDIA_PER_TURN { + debug!(path = %a.path, "media not inlined: per-turn count budget exhausted"); + rest.push(a.clone()); + continue; + } + match try_inline(a, capabilities, base, &root, total).await { + Some((part, bytes)) => { + total += bytes; + parts.push(part); + } + None => rest.push(a.clone()), + } + } + MediaPartition { parts, rest } +} + +/// Promotes one attachment to a content part, or `None` when any check fails +/// (logged at debug level; the caller keeps it on the textual path). +async fn try_inline( + a: &Attachment, + capabilities: &[String], + base: &Path, + root: &Path, + used_total: u64, +) -> Option<(Value, u64)> { + let abs = tokio::fs::canonicalize(base.join(&a.path)).await.ok()?; + if !abs.starts_with(root) { + debug!(path = %a.path, "media not inlined: outside the uploads root"); + return None; + } + + let mut file = tokio::fs::File::open(&abs).await.ok()?; + let mut head = [0u8; 16]; + let n = tokio::io::AsyncReadExt::read(&mut file, &mut head).await.ok()?; + let mime = sniff_mime(&head[..n])?; + let modality = MODALITIES.iter().find(|m| m.mimes.contains(&mime))?; + if !capabilities.iter().any(|c| c == modality.capability) { + debug!(path = %a.path, mime, "media not inlined: model lacks the capability"); + return None; + } + + let size = file.metadata().await.ok()?.len(); + if size > modality.max_bytes { + debug!(path = %a.path, size, "media not inlined: file too large"); + return None; + } + if used_total + size > MAX_TOTAL_MEDIA_BYTES { + debug!(path = %a.path, "media not inlined: per-turn byte budget exhausted"); + return None; + } + + let bytes = tokio::fs::read(&abs).await.ok()?; + let b64 = base64::engine::general_purpose::STANDARD.encode(&bytes); + let url = format!("data:{mime};base64,{b64}"); + let t = modality.part_type; + Some((json!({ "type": t, t: { "url": url } }), size)) +} + +/// Sniffs the magic bytes of a medium we know how to inline, returning its +/// canonical MIME type. `None` = not a recognized medium (not an error — +/// ordinary files simply stay on the textual path). +pub fn sniff_mime(head: &[u8]) -> Option<&'static str> { + if head.starts_with(b"\x89PNG\r\n\x1a\n") { + return Some("image/png"); + } + if head.starts_with(b"\xff\xd8\xff") { + return Some("image/jpeg"); + } + if head.starts_with(b"GIF87a") || head.starts_with(b"GIF89a") { + return Some("image/gif"); + } + if head.len() >= 12 && &head[0..4] == b"RIFF" && &head[8..12] == b"WEBP" { + return Some("image/webp"); + } + if head.len() >= 12 && &head[4..8] == b"ftyp" { + let brand = &head[8..12]; + if brand.starts_with(b"3gp") || brand.starts_with(b"3g2") { + return Some("video/3gpp"); + } + if brand == b"qt " { + return Some("video/quicktime"); + } + // isom / mp41 / mp42 / avc1 / M4V … + return Some("video/mp4"); + } + // EBML header — WebM (and Matroska, close enough for the video models). + if head.starts_with(&[0x1A, 0x45, 0xDF, 0xA3]) { + return Some("video/webm"); + } + if head.len() >= 12 && &head[0..4] == b"RIFF" && &head[8..12] == b"AVI " { + return Some("video/x-msvideo"); + } + if head.starts_with(b"FLV\x01") { + return Some("video/x-flv"); + } + if head.starts_with(&[0x00, 0x00, 0x01, 0xBA]) || head.starts_with(&[0x00, 0x00, 0x01, 0xB3]) { + return Some("video/mpeg"); + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + fn att(path: &str) -> Attachment { + Attachment { + path: path.to_string(), + name: path.rsplit('/').next().unwrap().to_string(), + mimetype: None, + filesize: None, + } + } + + fn png_bytes() -> Vec { + let mut v = b"\x89PNG\r\n\x1a\n".to_vec(); + v.extend_from_slice(&[0xAA; 64]); + v + } + + fn caps(xs: &[&str]) -> Vec { + xs.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn sniff_known_signatures() { + assert_eq!(sniff_mime(b"\x89PNG\r\n\x1a\n...."), Some("image/png")); + assert_eq!(sniff_mime(b"\xff\xd8\xff\xe0...."), Some("image/jpeg")); + assert_eq!(sniff_mime(b"GIF89a...."), Some("image/gif")); + assert_eq!(sniff_mime(b"RIFF\x00\x00\x00\x00WEBP"), Some("image/webp")); + assert_eq!(sniff_mime(b"\x00\x00\x00\x18ftypisom"), Some("video/mp4")); + assert_eq!(sniff_mime(b"\x00\x00\x00\x18ftypqt "), Some("video/quicktime")); + assert_eq!(sniff_mime(b"\x00\x00\x00\x18ftyp3gp4"), Some("video/3gpp")); + assert_eq!(sniff_mime(&[0x1A, 0x45, 0xDF, 0xA3, 0, 0]), Some("video/webm")); + assert_eq!(sniff_mime(b"RIFF\x00\x00\x00\x00AVI "), Some("video/x-msvideo")); + assert_eq!(sniff_mime(b"FLV\x01\x05"), Some("video/x-flv")); + assert_eq!(sniff_mime(&[0x00, 0x00, 0x01, 0xBA]), Some("video/mpeg")); + assert_eq!(sniff_mime(b"%PDF-1.7"), None); + assert_eq!(sniff_mime(b""), None); + } + + #[tokio::test] + async fn partition_inlines_png_for_vision_model() { + let tmp = std::env::temp_dir().join(format!("skald-media-{}", uuid::Uuid::new_v4())); + let dir = tmp.join("data/uploads/u/1"); + tokio::fs::create_dir_all(&dir).await.unwrap(); + tokio::fs::write(dir.join("a.png"), png_bytes()).await.unwrap(); + + let p = partition_under(&[att("data/uploads/u/1/a.png")], &caps(&["vision"]), &tmp).await; + assert!(p.rest.is_empty()); + assert_eq!(p.parts.len(), 1); + let url = p.parts[0]["image_url"]["url"].as_str().unwrap(); + assert!(url.starts_with("data:image/png;base64,")); + + let _ = tokio::fs::remove_dir_all(&tmp).await; + } + + #[tokio::test] + async fn partition_gates_on_capability_and_containment() { + let tmp = std::env::temp_dir().join(format!("skald-media-{}", uuid::Uuid::new_v4())); + let dir = tmp.join("data/uploads/u/1"); + tokio::fs::create_dir_all(&dir).await.unwrap(); + tokio::fs::write(dir.join("a.png"), png_bytes()).await.unwrap(); + tokio::fs::write(tmp.join("secret.png"), png_bytes()).await.unwrap(); + + // No capability → everything stays textual. + let p = partition_under(&[att("data/uploads/u/1/a.png")], &caps(&[]), &tmp).await; + assert_eq!(p.rest.len(), 1); + assert!(p.parts.is_empty()); + + // vision capability does not unlock video parts. + let p = partition_under(&[att("data/uploads/u/1/a.png")], &caps(&["video"]), &tmp).await; + assert_eq!(p.rest.len(), 1); + + // A real image outside the uploads root is never read inline. + let p = partition_under(&[att("secret.png")], &caps(&["vision"]), &tmp).await; + assert_eq!(p.rest.len(), 1); + assert!(p.parts.is_empty()); + + // Traversal out of the root is rejected. + let p = partition_under(&[att("data/uploads/../../secret.png")], &caps(&["vision"]), &tmp).await; + assert_eq!(p.rest.len(), 1); + assert!(p.parts.is_empty()); + + let _ = tokio::fs::remove_dir_all(&tmp).await; + } + + #[tokio::test] + async fn partition_enforces_count_budget() { + let tmp = std::env::temp_dir().join(format!("skald-media-{}", uuid::Uuid::new_v4())); + let dir = tmp.join("data/uploads/u/1"); + tokio::fs::create_dir_all(&dir).await.unwrap(); + let mut atts = Vec::new(); + for i in 0..(MAX_MEDIA_PER_TURN + 2) { + let rel = format!("data/uploads/u/1/{i}.png"); + tokio::fs::write(dir.join(format!("{i}.png")), png_bytes()).await.unwrap(); + atts.push(att(&rel)); + } + let p = partition_under(&atts, &caps(&["vision"]), &tmp).await; + assert_eq!(p.parts.len(), MAX_MEDIA_PER_TURN); + assert_eq!(p.rest.len(), 2); + + let _ = tokio::fs::remove_dir_all(&tmp).await; + } +} diff --git a/crates/skald-core/src/session/handler/message_builder.rs b/crates/skald-core/src/session/handler/message_builder.rs index 36c621d..fd08d4e 100644 --- a/crates/skald-core/src/session/handler/message_builder.rs +++ b/crates/skald-core/src/session/handler/message_builder.rs @@ -83,6 +83,9 @@ impl MessageBuilder { active_mcp_grants: &HashSet, system_substitutions: &HashMap, cache_hints: bool, + // Input capabilities of the resolved model (`vision`, `video`, …) — + // drives inline media for current-turn attachments. + capabilities: &[String], ) -> anyhow::Result> { let pool = &*self.pool; @@ -196,20 +199,59 @@ impl MessageBuilder { .iter() .rposition(|e| matches!(e.role, chat_history::Role::User | chat_history::Role::Agent)); + // Inline-media turn group. Trailing assistant rows are the in-flight + // turn's own rounds (their tool calls are already persisted), so the + // current turn's user messages sit just before them; a coalesced run of + // user/agent rows ahead of those belongs to the same turn. Media from + // earlier turns degrades to the textual path block — re-sending images + // on every round would re-bill them each time. + let mut media_turn_start = history.len(); + while media_turn_start > 0 + && matches!(history[media_turn_start - 1].role, chat_history::Role::Assistant) + { + media_turn_start -= 1; + } + while media_turn_start > 0 + && matches!( + history[media_turn_start - 1].role, + chat_history::Role::User | chat_history::Role::Agent + ) + { + media_turn_start -= 1; + } + for (idx, entry) in history.iter().enumerate() { - let is_previous_turn = current_turn_boundary.map_or(false, |b| idx < b); + let is_previous_turn = current_turn_boundary.is_some_and(|b| idx < b); match entry.role { chat_history::Role::User | chat_history::Role::Agent => { - // Render attachments (if any) as a textual block appended to the - // user turn, generated on the fly — never persisted as content. - let content = match &entry.metadata { - Some(meta) if !meta.attachments.is_empty() => format!( - "{}{}", - entry.content, - core_api::message_meta::attachments_block(&meta.attachments), + // Attachments reach the model two ways: media of the current + // turn is inlined as native content parts when the resolved + // model declares the capability (media::partition); everything + // else — and every attachment of older turns — keeps the + // textual path block, generated on the fly and never + // persisted as content. + let (text, media) = match &entry.metadata { + Some(meta) if !meta.attachments.is_empty() && idx >= media_turn_start => { + let partition = super::media::partition(&meta.attachments, capabilities).await; + ( + format!( + "{}{}", + entry.content, + core_api::message_meta::attachments_block(&partition.rest), + ), + partition.parts, + ) + } + Some(meta) if !meta.attachments.is_empty() => ( + format!( + "{}{}", + entry.content, + core_api::message_meta::attachments_block(&meta.attachments), + ), + Vec::new(), ), - _ => entry.content.clone(), + _ => (entry.content.clone(), Vec::new()), }; // Coalesce consecutive user/agent rows into a single `role:user` // turn. The DB keeps each message as its own row (distinct bubbles, @@ -217,13 +259,7 @@ impl MessageBuilder { // turn — e.g. when several messages were injected back-to-back at a // round boundary, or queued together while idle. `for_stack` already // excludes `failed` rows, so only non-failed messages merge here. - match out.last_mut() { - Some(last) if last["role"] == "user" => { - let prev = last["content"].as_str().unwrap_or("").to_string(); - last["content"] = Value::String(format!("{prev}\n\n{content}")); - } - _ => out.push(json!({ "role": "user", "content": content })), - } + push_user_chunk(&mut out, text, media); } chat_history::Role::Assistant => { let tool_calls = chat_llm_tools::for_message(pool, entry.id).await?; @@ -485,6 +521,48 @@ impl MessageBuilder { // ── Free helpers ────────────────────────────────────────────────────────────── +/// Appends one user/agent chunk — text plus any inline media parts — to the +/// message stream, coalescing with a preceding `user` message. Plain-text +/// chunks merge exactly as before (one string); when either side carries +/// parts, the merged content is normalized to a parts array, with the new +/// text folded into the LAST text part so media parts keep their position. +fn push_user_chunk(out: &mut Vec, text: String, media: Vec) { + fn text_part(t: &str) -> Value { + json!({ "type": "text", "text": t }) + } + + if let Some(last) = out.last_mut() + && last["role"] == "user" + { + if !last["content"].is_array() && media.is_empty() { + let prev = last["content"].as_str().unwrap_or("").to_string(); + last["content"] = Value::String(format!("{prev}\n\n{text}")); + return; + } + let mut parts = match last["content"].take() { + Value::Array(a) => a, + Value::String(s) => vec![text_part(&s)], + _ => Vec::new(), + }; + if let Some(tp) = parts.iter_mut().rev().find(|p| p["type"] == "text") { + let prev = tp["text"].as_str().unwrap_or("").to_string(); + tp["text"] = Value::String(format!("{prev}\n\n{text}")); + } else { + parts.insert(0, text_part(&text)); + } + parts.extend(media); + last["content"] = Value::Array(parts); + return; + } + if media.is_empty() { + out.push(json!({ "role": "user", "content": text })); + } else { + let mut parts = vec![text_part(&text)]; + parts.extend(media); + out.push(json!({ "role": "user", "content": parts })); + } +} + /// Creates an informative 1-line summary of a tool call result. /// /// Produces human-readable descriptions like: @@ -587,3 +665,63 @@ fn summarize_tool_result(tool_name: &str, arguments: Option<&str>, result: &str) } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn img() -> Value { + json!({ "type": "image_url", "image_url": { "url": "data:image/png;base64,QUJD" } }) + } + + #[test] + fn plain_text_chunks_merge_as_string() { + let mut out = vec![]; + push_user_chunk(&mut out, "one".into(), vec![]); + push_user_chunk(&mut out, "two".into(), vec![]); + assert_eq!(out, vec![json!({ "role": "user", "content": "one\n\ntwo" })]); + } + + #[test] + fn media_chunk_normalizes_to_parts() { + let mut out = vec![]; + push_user_chunk(&mut out, "look".into(), vec![img()]); + assert_eq!(out, vec![json!({ + "role": "user", + "content": [ + { "type": "text", "text": "look" }, + { "type": "image_url", "image_url": { "url": "data:image/png;base64,QUJD" } }, + ] + })]); + } + + #[test] + fn text_after_media_folds_into_last_text_part() { + let mut out = vec![]; + push_user_chunk(&mut out, "look".into(), vec![img()]); + push_user_chunk(&mut out, "and this".into(), vec![]); + let content = out[0]["content"].as_array().unwrap(); + assert_eq!(content.len(), 2); + assert_eq!(content[0]["text"], json!("look\n\nand this")); + assert_eq!(content[1]["type"], json!("image_url")); + } + + #[test] + fn media_merges_after_plain_text() { + let mut out = vec![]; + push_user_chunk(&mut out, "one".into(), vec![]); + push_user_chunk(&mut out, "two".into(), vec![img()]); + let content = out[0]["content"].as_array().unwrap(); + assert_eq!(content[0]["text"], json!("one\n\ntwo")); + assert_eq!(content[1]["type"], json!("image_url")); + } + + #[test] + fn chunk_after_assistant_starts_new_message() { + let mut out = vec![json!({ "role": "assistant", "content": "hi" })]; + push_user_chunk(&mut out, "one".into(), vec![img()]); + assert_eq!(out.len(), 2); + assert_eq!(out[1]["role"], json!("user")); + assert!(out[1]["content"].is_array()); + } +} diff --git a/crates/skald-core/src/session/handler/messages.rs b/crates/skald-core/src/session/handler/messages.rs index ed5104f..6ea8cb9 100644 --- a/crates/skald-core/src/session/handler/messages.rs +++ b/crates/skald-core/src/session/handler/messages.rs @@ -22,6 +22,7 @@ impl ChatSessionHandler { active_mcp_grants: &HashSet, system_substitutions: &HashMap, cache_hints: bool, + capabilities: &[String], ) -> anyhow::Result> { let effective_wd = self.run_context.read().await .as_ref() @@ -40,6 +41,6 @@ impl ChatSessionHandler { // `pool` is passed in from the caller (always `&self.db`) but we take // ownership via Arc::clone above so the signature stays backward-compatible. let _ = pool; // suppress unused-variable warning; MessageBuilder uses its own Arc - builder.build(stack_id, agent_id, extra_system_static, extra_system_dynamic, tail_reminder, active_mcp_grants, system_substitutions, cache_hints).await + builder.build(stack_id, agent_id, extra_system_static, extra_system_dynamic, tail_reminder, active_mcp_grants, system_substitutions, cache_hints, capabilities).await } } diff --git a/crates/skald-core/src/session/handler/mod.rs b/crates/skald-core/src/session/handler/mod.rs index 8b865ca..7881ba5 100644 --- a/crates/skald-core/src/session/handler/mod.rs +++ b/crates/skald-core/src/session/handler/mod.rs @@ -37,6 +37,7 @@ mod gate; mod interface_tools; mod llm_call; mod llm_loop; +pub mod media; pub mod message_builder; mod messages; mod outcome; diff --git a/providers.yaml b/providers.yaml index 6dc0264..ad248d7 100644 --- a/providers.yaml +++ b/providers.yaml @@ -79,9 +79,9 @@ providers: reasoning: supports_reasoning base_capabilities: [function_calling] # Fills the metadata the endpoint may omit (endpoint values always win): - # k3 → 1M context + native vision; kimi-for-coding → 256k. + # k3 → 1M context + native vision and video input; kimi-for-coding → 256k. enrich: - - { match: "k3*", context_length: 1048576, vision: true } + - { match: "k3*", context_length: 1048576, vision: true, add_capabilities: [video] } - { match: "kimi-for-coding*", context_length: 262144 } reasoning: # k3 exposes a graded reasoning_effort ("disabled" routes to K2.6); diff --git a/src/frontend/api/mod.rs b/src/frontend/api/mod.rs index cee4562..485beb0 100644 --- a/src/frontend/api/mod.rs +++ b/src/frontend/api/mod.rs @@ -210,6 +210,10 @@ impl ApiError { pub fn forbidden(msg: impl Into) -> Self { Self { status: StatusCode::FORBIDDEN, message: msg.into() } } + + pub fn payload_too_large(msg: impl Into) -> Self { + Self { status: StatusCode::PAYLOAD_TOO_LARGE, message: msg.into() } + } } /// Resolves the authenticated caller's per-user runtime context, or `401` when the diff --git a/src/frontend/api/uploads.rs b/src/frontend/api/uploads.rs index cce257d..fced831 100644 --- a/src/frontend/api/uploads.rs +++ b/src/frontend/api/uploads.rs @@ -9,17 +9,25 @@ use tokio::io::AsyncWriteExt; use core_api::message_meta::Attachment; +use skald_core::session::handler::media::sniff_mime; use skald_core::skald::Skald; use skald_core::tools::fs as fs_tools; use super::{ApiError, guard::AuthUser, require_context}; use super::sessions::SourcePath; +/// Max bytes accepted for a single uploaded file; anything larger is cut off +/// mid-stream, the partial file removed, and the request answered 413. +const MAX_UPLOAD_BYTES: u64 = 256 * 1024 * 1024; + /// `POST /api/{source}/uploads` /// /// Accepts a `multipart/form-data` body with one or more file fields and saves -/// each under `data/uploads/{session_id}/`. Bytes are streamed straight to disk -/// (`field.chunk()` → file), never buffered whole in RAM, so arbitrarily large -/// files are fine — the route disables the default body-size limit (see router). +/// each under `data/uploads/{user_id}/{session_id}/` (per-user namespaced, so +/// colliding session ids across users never share a directory). Bytes are +/// streamed straight to disk (`field.chunk()` → file), never buffered whole in +/// RAM — the route disables the default body-size limit (see router) and +/// enforces [`MAX_UPLOAD_BYTES`] itself. When the magic bytes are recognized, +/// the sniffed MIME wins over the client-supplied `Content-Type`. /// /// Returns the saved [`Attachment`]s (project-root-relative path, name, MIME, /// size) so the client can show chips and echo them back when sending the message. @@ -34,7 +42,7 @@ pub async fn upload( // directory the message will reference. let session_id = ctx.chat_hub.session_handler(&p.source).await?.session_id; - let dir_rel = format!("data/uploads/{session_id}"); + let dir_rel = format!("data/uploads/{}/{session_id}", auth.user_id); let dir_abs = fs_tools::resolve(&dir_rel)?; tokio::fs::create_dir_all(&dir_abs).await?; @@ -54,13 +62,30 @@ pub async fn upload( .map_err(|e| ApiError::from(anyhow::anyhow!("cannot create {}: {e}", abs_path.display())))?; let mut size: u64 = 0; + let mut too_large = false; while let Some(chunk) = field.chunk().await .map_err(|e| ApiError::bad_request(format!("upload read error: {e}")))? { - file.write_all(&chunk).await?; size += chunk.len() as u64; + if size > MAX_UPLOAD_BYTES { + too_large = true; + break; + } + file.write_all(&chunk).await?; } file.flush().await?; + drop(file); + + if too_large { + let _ = tokio::fs::remove_file(&abs_path).await; + return Err(ApiError::payload_too_large(format!( + "'{final_name}' exceeds the {} MiB upload limit", + MAX_UPLOAD_BYTES / 1024 / 1024 + ))); + } + + // The sniffed type wins over the client claim when we recognize the bytes. + let mimetype = sniff_head(&abs_path).await.map(String::from).or(mimetype); saved.push(Attachment { path: format!("{dir_rel}/{final_name}"), @@ -73,6 +98,14 @@ pub async fn upload( Ok(Json(saved)) } +/// Reads the first bytes of a saved upload and sniffs its real media type. +async fn sniff_head(path: &StdPath) -> Option<&'static str> { + let mut file = tokio::fs::File::open(path).await.ok()?; + let mut head = [0u8; 16]; + let n = tokio::io::AsyncReadExt::read(&mut file, &mut head).await.ok()?; + sniff_mime(&head[..n]) +} + /// Reduces an arbitrary client filename to a safe basename: directory components /// are dropped and an empty/`.`/`..` result falls back to `"file"`. fn sanitize_filename(raw: &str) -> String { diff --git a/src/frontend/server.rs b/src/frontend/server.rs index ae1c260..fe3a385 100644 --- a/src/frontend/server.rs +++ b/src/frontend/server.rs @@ -78,6 +78,7 @@ impl WebServer { Arc::clone(&skald), api::guard::require_auth, )); + let skald_for_data = Arc::clone(&skald); // Resolve the app state first so the resulting `Router<()>` can host the // stateless plugin routers via `nest`. @@ -87,7 +88,8 @@ impl WebServer { for (id, plugin_router) in plugin_routers { router = router.nest(&format!("/api/plugin/{id}"), plugin_router); } - // Serve the data/ directory under /data/ (accessible via URL). + // Serve the data/ directory under /data/ (accessible via URL), behind the + // same session-cookie gate as /api — uploads are private user content. let data_dir = Path::new(static_dir).parent().unwrap_or(Path::new(".")).join("data"); // Static responses (SPA assets + /data) get `Cache-Control: no-cache`: // the browser may store them but MUST revalidate before use, so after a @@ -98,7 +100,17 @@ impl WebServer { header::CACHE_CONTROL, HeaderValue::from_static("no-cache"), )); - router = router.nest_service("/data", static_assets().service(ServeDir::new(&data_dir))); + let data_service = ServiceBuilder::new() + .layer(axum::middleware::from_fn_with_state( + skald_for_data, + api::guard::require_auth, + )) + .layer(SetResponseHeaderLayer::overriding( + header::CACHE_CONTROL, + HeaderValue::from_static("no-cache"), + )) + .service(ServeDir::new(&data_dir)); + router = router.nest_service("/data", data_service); router = router.fallback_service(static_assets().service(ServeDir::new(static_dir))); // Negotiated gzip/brotli compression (Accept-Encoding). Matters most for // the mobile WebView, whose HTTP traffic is reverse-proxied byte-for-byte