--- a/crates/airouter-agent-client/src/session_service.rs +++ b/crates/airouter-agent-client/src/session_service.rs @@ -18,6 +18,17 @@ /// Transitional trusted adapter input. The daemon pins the executable and /// executor to its registration; model-facing requests cannot supply this. pub input: JsonlProcessInput, + /// App-authorized immutable image sources, in admitted attachment order. + /// The daemon copies and hashes bytes before acknowledging admission. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub attachments: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AttachmentSource { + pub reference: String, + pub path: String, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] --- a/crates/airouter-agentd/src/session_service/mod.rs +++ b/crates/airouter-agentd/src/session_service/mod.rs @@ -1,5 +1,6 @@ //! Independent native session ownership. Socket observers never own job futures. pub mod api; +mod attachments; mod driver; pub mod native; mod store; @@ -83,6 +84,9 @@ return Err("session lease is not available for execution".into()); } self.backend.prepare(request, &prepared.run)?; + let workspace = Path::new(&request.input.plan.cwd) + .canonicalize().map_err(|_| "native working directory unavailable")?; + airouter_agent_client::journal::Journal::for_workspace(self.store.root(), &workspace, None)?; Ok(()) } pub async fn submit(self: &Arc, request: Submit) -> Result { @@ -102,6 +106,7 @@ } let prepared = self.client.read_host(id).await?; self.validate(&request, &prepared)?; + attachments::stage(&request, &prepared.run, self.store.root())?; let record = Record { request, attempted: false, --- a/crates/airouter-agentd/src/session_service/driver.rs +++ b/crates/airouter-agentd/src/session_service/driver.rs @@ -80,7 +80,8 @@ } let prepared = self.client.read_host(id).await?; self.validate(&saved.request, &prepared)?; - let host = self.backend.prepare(&saved.request, &prepared.run)?; + let pinned = attachments::pinned(&saved.request, &prepared.run, self.store.root())?; + let host = self.backend.prepare(&pinned, &prepared.run)?; let journal = Journal::new(self.store.root())?; let binding = &saved.request.binding; let renewal = task::Action::Renew { --- a/crates/airouter-agentd/src/session_service/native.rs +++ b/crates/airouter-agentd/src/session_service/native.rs @@ -73,11 +73,7 @@ { return Err("native executable or executor differs from its registration".into()); } - if !session.input.attachments.is_empty() - || session.input.modalities.iter().any(|m| m != "text") - { - return Err("durable attachment staging is not yet supported by this service".into()); - } + super::attachments::validate(request, run)?; if !std::path::Path::new(&request.input.plan.cwd).is_absolute() || !std::path::Path::new(&request.input.plan.cwd).is_dir() { --- /dev/null +++ b/crates/airouter-agentd/src/session_service/attachments.rs @@ -0,0 +1,230 @@ +//! Copy admitted bytes before ACK. Original app staging paths are never needed +//! by an accepted execution, including after a restart before its first attempt. +use airouter_agent_client::session_service::Submit; +use airouter_agent_protocol::{ + Run, + session::{Attachment, Executor}, +}; +use sha2::{Digest, Sha256}; +use std::{ + fs::{self, File, OpenOptions}, + io::{Read, Write}, + os::unix::fs::{DirBuilderExt, MetadataExt, OpenOptionsExt}, + path::{Path, PathBuf}, +}; +use uuid::Uuid; + +const MAX_ATTACHMENTS: usize = 16; +const MAX_BYTES: u64 = 64 * 1024 * 1024; + +pub(super) fn validate(request: &Submit, run: &Run) -> Result<(), String> { + let session = run.session.as_ref().ok_or("session checkpoint missing")?; + let expected = &session.input.attachments; + if request.attachments.len() != expected.len() || expected.len() > MAX_ATTACHMENTS { + return Err("attachment sources differ from the admitted input".into()); + } + let mut total = 0u64; + for (source, admitted) in request.attachments.iter().zip(expected) { + total = total + .checked_add(admitted.bytes) + .ok_or("attachment size overflow")?; + if source.reference != admitted.reference + || !Path::new(&source.path).is_absolute() + || admitted.bytes == 0 + || total > MAX_BYTES + || !["image/png", "image/jpeg", "image/gif", "image/webp"] + .contains(&admitted.media_type.as_str()) + { + return Err("invalid admitted attachment source, encoding or size".into()); + } + } + if !expected.is_empty() + && (session.profile.executor != Executor::NativeCodex + || !session.input.modalities.iter().any(|m| m == "image")) + { + return Err("registered session does not support staged images".into()); + } + if session + .input + .modalities + .iter() + .any(|m| m != "text" && m != "image") + { + return Err("unsupported native session modality".into()); + } + if session.profile.executor == Executor::NativeCodex { + let args = &request.input.plan.args; + // The shared Codex builder emits one --image/path pair per source. + // Other spellings cannot bypass the admitted attachment set. + if args + .iter() + .any(|s| s == "-i" || s.starts_with("-i=") || s.starts_with("--image=")) + { + return Err("use canonical --image attachment arguments".into()); + } + let mut paths = Vec::new(); + for (index, arg) in args.iter().enumerate() { + if arg == "--image" { + paths.push( + args.get(index + 1) + .ok_or("image argument missing")? + .as_str(), + ); + } + } + if paths + != request + .attachments + .iter() + .map(|a| a.path.as_str()) + .collect::>() + { + return Err("native image arguments differ from admitted sources".into()); + } + } + Ok(()) +} + +fn directory(root: &Path, id: Uuid) -> PathBuf { + root.join(format!("{id}.assets")) +} +fn asset(directory: &Path, index: usize, attachment: &Attachment) -> PathBuf { + let extension = match attachment.media_type.as_str() { + "image/png" => "png", + "image/jpeg" => "jpg", + "image/gif" => "gif", + _ => "webp", + }; + directory.join(format!("{index}.{extension}")) +} +fn private_directory(path: &Path) -> Result<(), String> { + let meta = fs::symlink_metadata(path).map_err(|_| "staged attachments unavailable")?; + if !meta.is_dir() + || meta.uid() != nix::unistd::geteuid().as_raw() + || meta.mode() & 0o7777 != 0o700 + || fs::canonicalize(path).map_err(|_| "staged attachment path unavailable")? != path + { + return Err("staged attachment directory is not private and canonical".into()); + } + Ok(()) +} +fn bytes(mut file: File, expected: &Attachment) -> Result, String> { + let meta = file + .metadata() + .map_err(|_| "attachment metadata unavailable")?; + if !meta.is_file() || meta.len() != expected.bytes { + return Err("attachment size/type differs from admitted input".into()); + } + let mut bytes = Vec::new(); + (&mut file) + .take(expected.bytes + 1) + .read_to_end(&mut bytes) + .map_err(|_| "attachment bytes unavailable")?; + if bytes.len() as u64 != expected.bytes + || format!("{:x}", Sha256::digest(&bytes)) != expected.sha256 + { + return Err("attachment digest differs from admitted input".into()); + } + Ok(bytes) +} +fn verify(request: &Submit, run: &Run, root: &Path) -> Result<(), String> { + let directory = directory(root, run.id); + private_directory(&directory)?; + for (index, expected) in run + .session + .as_ref() + .ok_or("session checkpoint missing")? + .input + .attachments + .iter() + .enumerate() + { + bytes( + super::store::private_file(&asset(&directory, index, expected), false, false)?, + expected, + )?; + } + validate(request, run) +} + +pub(super) fn stage(request: &Submit, run: &Run, root: &Path) -> Result<(), String> { + validate(request, run)?; + if request.attachments.is_empty() { + return Ok(()); + } + let target = directory(root, run.id); + match fs::symlink_metadata(&target) { + Ok(_) => return verify(request, run, root), // Crash after copying, before ACK. + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(_) => return Err("staged attachment directory unavailable".into()), + } + let temporary = root.join(format!(".assets-{}", Uuid::new_v4())); + fs::DirBuilder::new() + .mode(0o700) + .create(&temporary) + .map_err(|_| "attachment staging unavailable")?; + let result = (|| { + for (index, (source, expected)) in request + .attachments + .iter() + .zip(&run.session.as_ref().unwrap().input.attachments) + .enumerate() + { + let path = Path::new(&source.path); + if fs::canonicalize(path).map_err(|_| "attachment source unavailable")? != path { + return Err("attachment source aliases are not accepted".into()); + } + let file = OpenOptions::new() + .read(true) + .custom_flags(nix::libc::O_NOFOLLOW | nix::libc::O_NONBLOCK | nix::libc::O_CLOEXEC) + .open(path) + .map_err(|_| "attachment source unavailable")?; + let contents = bytes(file, expected)?; + let mut copy = OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(asset(&temporary, index, expected)) + .map_err(|_| "private attachment copy unavailable")?; + copy.write_all(&contents) + .and_then(|_| copy.sync_all()) + .map_err(|_| "attachment copy could not be saved")?; + } + File::open(&temporary) + .and_then(|f| f.sync_all()) + .map_err(|_| "attachment staging could not be synced")?; + fs::rename(&temporary, &target).map_err(|_| "attachment staging could not be committed")?; + File::open(root) + .and_then(|f| f.sync_all()) + .map_err(|_| "attachment spool could not be synced")?; + Ok(()) + })(); + if temporary.exists() { + let _ = fs::remove_dir_all(temporary); + } + result +} + +pub(super) fn pinned(request: &Submit, run: &Run, root: &Path) -> Result { + validate(request, run)?; + if request.attachments.is_empty() { + return Ok(request.clone()); + } + verify(request, run, root)?; + let mut pinned = request.clone(); + let expected = &run.session.as_ref().unwrap().input.attachments; + let mut index = 0; + for arg in 0..pinned.input.plan.args.len().saturating_sub(1) { + if pinned.input.plan.args[arg] == "--image" { + let path = asset(&directory(root, run.id), index, &expected[index]); + let path = path + .to_str() + .ok_or("non-UTF8 staged attachment path")? + .to_owned(); + pinned.input.plan.args[arg + 1] = path.clone(); + pinned.attachments[index].path = path; + index += 1; + } + } + Ok(pinned) +} --- a/crates/airouter-agentd/tests/session_service/mod.rs +++ b/crates/airouter-agentd/tests/session_service/mod.rs @@ -9,6 +9,8 @@ }; use airouter_agentd::session_service::{Backend, Options, Service, api}; use std::{os::unix::fs::PermissionsExt, path::Path, time::Duration}; +#[path = "attachments.rs"] +mod attachments; struct Effect { calls: AtomicUsize, @@ -92,13 +94,16 @@ } } fn submit(prepared: &controller::Prepared, root: &Path) -> wire::Submit { + let workspace = root.join("workspace"); + std::fs::create_dir_all(&workspace).unwrap(); wire::Submit { version: wire::VERSION.into(), binding: prepared.binding.clone(), + attachments: vec![], input: JsonlProcessInput::new( ProcessLaunchPlan::new( vec!["/fixture/registered-program".into()], - root.to_str().unwrap(), + workspace.to_str().unwrap(), ), "admitted input", ), --- /dev/null +++ b/crates/airouter-agentd/tests/session_service/attachments.rs @@ -0,0 +1,227 @@ +use super::*; +use sha2::{Digest, Sha256}; +use std::sync::{Mutex, atomic::AtomicBool}; + +const IMAGE: &[u8] = b"\x89PNG\r\n\x1a\nfixture image"; + +struct ImageState { + unavailable: Arc, + interrupt_admission: AtomicBool, + calls: AtomicUsize, + started: Notify, + release: Notify, + path: Mutex>, +} +#[derive(Clone)] +struct ImageBackend(Arc); +struct ImageHost { + state: Arc, + request: wire::Submit, +} +impl Backend for ImageBackend { + type Host = ImageHost; + fn prepare(&self, request: &wire::Submit, _: &Run) -> Result { + if self.0.interrupt_admission.swap(false, Ordering::SeqCst) { + self.0.unavailable.store(1, Ordering::SeqCst); + } + Ok(ImageHost { + state: self.0.clone(), + request: request.clone(), + }) + } +} +impl managed_session::Host for ImageHost { + async fn execute( + &self, + run: &Run, + frames: tokio::sync::mpsc::Sender, + ) -> Result { + self.execute_cancellable(run, frames, CancellationToken::new()) + .await + } + async fn execute_cancellable( + &self, + _: &Run, + frames: tokio::sync::mpsc::Sender, + cancellation: CancellationToken, + ) -> Result { + self.state.calls.fetch_add(1, Ordering::SeqCst); + let source = &self.request.attachments[0].path; + assert_eq!(self.request.input.plan.args.last(), Some(source)); + *self.state.path.lock().unwrap() = Some(source.into()); + self.state.started.notify_one(); + tokio::select! { _ = self.state.release.notified() => {}, _ = cancellation.cancelled() => return Err("fixture cancelled".into()) } + assert_eq!( + std::fs::read(source).unwrap(), + IMAGE, + "execution did not retain admitted bytes" + ); + frames + .send(Frame::ResponseStarted { + id: "image-answer".into(), + }) + .await + .unwrap(); + frames + .send(Frame::ResponseCompleted { + id: "image-answer".into(), + text: "pinned image read once".into(), + }) + .await + .unwrap(); + Ok(session::Outcome { + final_response_id: Some("image-answer".into()), + error: None, + usage: Usage { + usage_unknown: true, + coverage: Some(Default::default()), + ..Default::default() + }, + }) + } +} +fn backend(f: &Fixture, interrupt: bool) -> ImageBackend { + ImageBackend(Arc::new(ImageState { + unavailable: f.unavailable.clone(), + interrupt_admission: AtomicBool::new(interrupt), + calls: 0.into(), + started: Notify::new(), + release: Notify::new(), + path: Mutex::new(None), + })) +} +async fn image_request(f: &Fixture, root: &Path) -> (controller::Prepared, wire::Submit) { + let mut admission = f.managed_request().await; + let resolved = f + .client + .resolve_profile(&execution::ResolveProfile { + provider: "openai".into(), + model: "exact-native".into(), + modalities: vec!["text".into(), "image".into()], + }) + .await + .unwrap(); + admission.request.profile = resolved.profile; + admission.expected_profile_digest = Some(resolved.profile_digest); + admission.session_input = Some(session::Input { + modalities: vec!["text".into(), "image".into()], + attachments: vec![session::Attachment { + reference: "fixture/image".into(), + media_type: "image/png".into(), + sha256: format!("{:x}", Sha256::digest(IMAGE)), + bytes: IMAGE.len() as u64, + }], + }); + let prepared = f.client.prepare_host(&admission).await.unwrap(); + let mut request = submit(&prepared, root); + let app = root.join("app-staging"); + std::fs::create_dir(&app).unwrap(); + let image = app.join("image.png"); + std::fs::write(&image, IMAGE).unwrap(); + let path = image.to_str().unwrap().to_owned(); + request.attachments = vec![wire::AttachmentSource { + reference: "fixture/image".into(), + path: path.clone(), + }]; + request.input.plan.args.extend(["--image".into(), path]); + (prepared, request) +} +fn private_dir() -> tempfile::TempDir { + tempfile::Builder::new() + .permissions(std::fs::Permissions::from_mode(0o700)) + .tempdir() + .unwrap() +} + +#[tokio::test] +async fn accepted_images_survive_source_deletion_and_restart_before_first_attempt() { + let f = Fixture::new().await; + let root = private_dir(); + let (prepared, request) = image_request(&f, root.path()).await; + let image = backend(&f, true); + let service = Service::open(f.client.clone(), root.path(), image.clone(), options()).unwrap(); + service.submit(request.clone()).await.unwrap(); + wait_for(async || { + service.status(prepared.run.id).unwrap().state == State::ReconciliationRequired + }) + .await; + assert_eq!(image.0.calls.load(Ordering::SeqCst), 0); + std::fs::remove_dir_all(root.path().join("app-staging")).unwrap(); + service.shutdown().await; + drop(service); + f.unavailable.store(0, Ordering::SeqCst); + let reopened = Service::open(f.client.clone(), root.path(), image.clone(), options()).unwrap(); + reopened.recover().await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), image.0.started.notified()) + .await + .unwrap(); + let copied = image.0.path.lock().unwrap().clone().unwrap(); + assert!(copied.starts_with(root.path().join(format!("{}.assets", prepared.run.id)))); + assert_eq!( + std::fs::metadata(&copied).unwrap().permissions().mode() & 0o777, + 0o600 + ); + assert!(!Path::new(&request.attachments[0].path).exists()); + image.0.release.notify_one(); + wait_for(async || reopened.status(prepared.run.id).unwrap().state == State::Finished).await; + assert_eq!( + reopened.status(prepared.run.id).unwrap().phase, + Phase::Complete + ); + reopened.submit(request).await.unwrap(); + reopened.reconcile(prepared.run.id).await.unwrap(); + assert_eq!(image.0.calls.load(Ordering::SeqCst), 1); + reopened.shutdown().await; +} + +#[tokio::test] +async fn admission_rejects_changed_bytes_aliases_and_unadmitted_image_arguments() { + let f = Fixture::new().await; + let root = private_dir(); + let (prepared, request) = image_request(&f, root.path()).await; + let image = backend(&f, false); + let service = Service::open(f.client.clone(), root.path(), image.clone(), options()).unwrap(); + std::fs::write(&request.attachments[0].path, vec![b'x'; IMAGE.len()]).unwrap(); + assert!( + service + .submit(request.clone()) + .await + .unwrap_err() + .contains("digest") + ); + std::fs::write(&request.attachments[0].path, IMAGE).unwrap(); + let alias = root.path().join("alias.png"); + std::os::unix::fs::symlink(&request.attachments[0].path, &alias).unwrap(); + let mut changed = request.clone(); + changed.attachments[0].path = alias.to_str().unwrap().into(); + *changed.input.plan.args.last_mut().unwrap() = changed.attachments[0].path.clone(); + assert!( + service + .submit(changed) + .await + .unwrap_err() + .contains("aliases") + ); + let mut changed = request.clone(); + changed.attachments[0].reference = "different-reference".into(); + assert!(service.submit(changed).await.is_err()); + let mut changed = request.clone(); + changed + .input + .plan + .args + .extend(["--image".into(), "/unadmitted/path".into()]); + assert!(service.submit(changed).await.is_err()); + let mut changed = request.clone(); + changed.input.plan.cwd = root.path().to_str().unwrap().into(); + assert!( + service + .submit(changed) + .await + .unwrap_err() + .contains("outside project") + ); + assert!(service.status(prepared.run.id).is_err()); + assert_eq!(image.0.calls.load(Ordering::SeqCst), 0); + service.shutdown().await; +} --- a/crates/airouter-agentd/SESSION_SERVICE.md +++ b/crates/airouter-agentd/SESSION_SERVICE.md @@ -6,7 +6,8 @@ modes 0700 and 0600. There is no TCP listener or shared-group admission. The framework application credential stays in the daemon's protected token file. -This draft supports registered text Codex and Claude adapters. The registered +This draft supports registered Codex sessions with text/images and text Claude +adapters. The registered executable, native executor and cgroup placement come from daemon configuration. Prompt/argument/environment assembly remains a trusted application adapter input during migration. This is not a model-facing arbitrary command API. @@ -26,7 +27,8 @@ for the original host's cleanup and journal delivery. Cancellation acceptance is not itself proof of cleanup; observe the resulting status and framework run. -The spool is a private canonical directory outside the model workspace. A +The spool is a private canonical directory outside the model workspace; the +service rejects a spool beneath the submitted working directory. A process lock prevents another socket/daemon from owning it simultaneously. Atomic fsync'd records mark an attempt before polling its execution future; the existing SDK journal also claims the exact session operation. After a @@ -47,6 +49,19 @@ controller `binding`, and shared `JsonlProcessInput` as `input`. Execution placement is never accepted from that wire input. New requests are limited to 8 MiB and the configured active-job capacity (1–64). + +For admitted images, also supply `attachments = [{ reference, path }]` in the +same order as the framework session input. Each source is an app-authorized +regular file at a canonical absolute path. The shared Codex builder's matching +`--image PATH` pairs must appear in the launch arguments; alternative spellings +or extra image paths are rejected. The service copies at most 16 images / 64 MiB +total to private files, verifies every admitted size/SHA-256, and fsyncs the +copies before acknowledging admission. It rewrites only those image arguments +to the verified copies before execution. Input references and the controller +binding remain unchanged. Retrying the exact request works even after its app +staging directory is deleted. Startup can also execute a never-attempted admitted +request from these copies while its original lease is still valid. The private +copies are checked again before execution and retained with the job evidence. - `GET /v1/sessions/RUN` returns state, canonical phase, cancellation request, `activity_complete` and an optional error. It never returns the stored prompt, @@ -92,18 +107,20 @@ ## Remaining production integration -Before Workbench cutover, implement durable attachment staging and preserve -exact admitted attachment hashes, move its permission/tool bridge into an +Before Workbench cutover, move its permission/tool bridge into an independently available application resource service, and replace its direct native driver with submit/observe/project calls. Transfer heartbeat and explicit cancellation together. Do not leave observer-drop cancellation enabled there. +The app adapter must now supply the admitted image reference/path mapping and +retain the original submission for exact retries after its temporary files end. Grok and HTTP adapters are not registered by this draft. Their existing paths continue until separately migrated. Prompt/persona/model/context policy still follows the main AI ownership completion sequence. Focused qualification uses the existing private Unix framework fixture for observer loss/renewal, duplicate binding checks, cancellation/cleanup, saved -outcome recovery and unfinished claims. Standalone qualification uses a private +outcome recovery, unfinished claims, source deletion across restart and altered +attachment rejection. Standalone qualification uses a private non-root cgroup and synthetic JSONL child; it needs no production model login and makes no model calls. See project handoffs for exact source hashes and commands; do not repeat already passing checks absent changed behavior.