use super::*;
use cap_std::{ambient_authority, fs::Dir};
#[tokio::test]
async fn reading_and_publishing_images_retain_only_their_own_files() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let mut encoded = std::io::Cursor::new(Vec::new());
image::DynamicImage::ImageRgba8(image::RgbaImage::new(2, 2))
.write_to(&mut encoded, image::ImageFormat::Png)
.expect("encode image");
let bytes = encoded.into_inner();
let file = store
.publish_artifact("session", "source.png".into(), "image/png".into(), &bytes)
.await
.expect("stored image");
assert_eq!(
store.read_image("session", &file).await.expect("image"),
bytes
);
assert_eq!(
list_completed(&store.session_dir("session"), &store.blob_dir())
.await
.expect("files")
.len(),
1
);
assert!(
store
.publish_image("session", "wrong.jpg".into(), "image/jpeg", bytes.clone())
.await
.is_err()
);
store
.publish_image("session", "generated.png".into(), "image/png", bytes)
.await
.expect("generated artifact");
assert_eq!(
list_completed(&store.session_dir("session"), &store.blob_dir())
.await
.expect("files")
.len(),
2
);
}
#[tokio::test]
async fn upload_round_trip_is_session_scoped_and_atomic() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let session_id = "thread:not-a-uuid";
let mut pending = store
.begin_upload(session_id, "notes.txt".into(), 5, "text/plain".into())
.await
.expect("begin");
pending.append(0, b"hello").await.expect("append");
let file = pending.finish().await.expect("finish");
store
.verify_upload(session_id, &file)
.await
.expect("verify upload");
let bytes = store
.read_chunk(session_id, &file.id, 0, MAX_READ_CHUNK_BYTES)
.await
.expect("read");
assert_eq!(bytes.data, b"hello");
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
let session = store.session_dir(session_id);
let directory = session.join(&file.id);
for path in [store.root.as_path(), &session, &directory] {
let mode = std::fs::metadata(path)
.expect("directory mode")
.permissions()
.mode();
assert_eq!(mode & 0o777, 0o700);
}
let metadata = load_metadata(&directory.join(METADATA_FILE))
.await
.expect("metadata");
for path in [
store.blob_path(&metadata.content_hash),
directory.join(METADATA_FILE),
] {
let mode = std::fs::metadata(path)
.expect("file mode")
.permissions()
.mode();
assert_eq!(mode & 0o777, 0o600);
}
}
}
#[tokio::test]
async fn file_list_identifies_user_uploads_and_agent_artifacts() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let mut upload = store
.begin_upload("session", "input.txt".into(), 1, "text/plain".into())
.await
.expect("begin upload");
upload.append(0, b"u").await.expect("append upload");
upload.finish().await.expect("finish upload");
store
.publish_artifact("session", "output.txt".into(), "text/plain".into(), b"a")
.await
.expect("publish artifact");
let files = store
.list_files(
"session",
&[
crate::backend::session_files::SessionFileOrigin::Upload,
crate::backend::session_files::SessionFileOrigin::Artifact,
],
)
.await
.expect("list files");
assert!(
files.iter().any(|(origin, file)| {
*origin == SessionFileOrigin::Upload && file.name == "input.txt"
}) && files.iter().any(|(origin, file)| {
*origin == SessionFileOrigin::Artifact && file.name == "output.txt"
})
);
}
#[tokio::test]
async fn chat_file_grants_preserve_origin_identity_and_survive_source_deletion() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let mut pending = store
.begin_upload("source", "input.txt".into(), 5, "text/plain".into())
.await
.expect("upload");
pending.append(0, b"input").await.expect("append");
let upload = pending.finish().await.expect("finish");
let artifact = store
.publish_artifact(
"source",
"result.txt".into(),
"text/plain".into(),
b"result",
)
.await
.expect("artifact");
let observation = store
.grant_file("source", "observer", &upload)
.await
.expect("observation grant");
let mut altered = upload.clone();
altered.name = "different.txt".into();
for (source, file) in [
("source", &altered),
("source", &artifact),
("observer", &observation),
("unrelated", &upload),
] {
assert!(store.grant_upload(source, "execution", file).await.is_err());
}
assert!(
store
.list_files(
"execution",
&[
crate::backend::session_files::SessionFileOrigin::Upload,
crate::backend::session_files::SessionFileOrigin::Artifact
]
)
.await
.expect("rejected grants")
.is_empty()
);
for _ in 0..2 {
assert_eq!(
store
.grant_upload("source", "execution", &upload)
.await
.expect("upload grant"),
upload
);
assert_eq!(
store
.share_artifact("source", "chat", &artifact)
.await
.expect("artifact share"),
artifact
);
}
store.delete_session("source").await.expect("delete source");
let reopened = SessionFileStore::new(state.path(), None);
reopened
.verify_upload("execution", &upload)
.await
.expect("user origin");
assert_eq!(
reopened
.list_files(
"execution",
&[crate::backend::session_files::SessionFileOrigin::Upload]
)
.await
.expect("uploads"),
[(SessionFileOrigin::Upload, upload.clone())]
);
assert_eq!(
reopened
.list_files(
"chat",
&[crate::backend::session_files::SessionFileOrigin::Artifact]
)
.await
.expect("artifacts"),
[(SessionFileOrigin::Artifact, artifact.clone())]
);
assert!(reopened.verify_upload("chat", &artifact).await.is_err());
assert_eq!(
reopened
.read_file("execution", &upload)
.await
.expect("retained upload"),
b"input"
);
assert_eq!(
reopened
.read_file("chat", &artifact)
.await
.expect("retained artifact"),
b"result"
);
assert_eq!(blob_entries(&reopened), 2);
}
#[tokio::test]
async fn delete_session_removes_only_that_sessions_files() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
for session_id in ["deleted", "retained"] {
store
.publish_artifact(
session_id,
"result.txt".into(),
"text/plain".into(),
b"result",
)
.await
.expect("publish artifact");
}
store
.delete_session("deleted")
.await
.expect("delete session files");
assert!(
store
.list_files(
"deleted",
&[crate::backend::session_files::SessionFileOrigin::Artifact]
)
.await
.expect("deleted artifacts")
.is_empty()
);
assert_eq!(
store
.list_files(
"retained",
&[crate::backend::session_files::SessionFileOrigin::Artifact]
)
.await
.expect("retained artifacts")
.len(),
1
);
}
#[tokio::test]
async fn identical_payloads_share_one_content_blob() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let first = store
.publish_artifact("first", "one.txt".into(), "text/plain".into(), b"same")
.await
.expect("first artifact");
let second = store
.publish_artifact("second", "two.txt".into(), "text/plain".into(), b"same")
.await
.expect("second artifact");
assert_ne!(first.id, second.id);
assert_eq!(blob_entries(&store), 1);
}
#[tokio::test]
async fn repeated_chunk_reads_share_one_blob_verification() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let file = store
.publish_artifact("session", "result.txt".into(), "text/plain".into(), b"safe")
.await
.expect("artifact");
let reader = store.clone();
storage::HASH_FILE_CALLS
.scope(std::cell::Cell::new(0), async {
for offset in 0..file.size {
let chunk = reader
.read_chunk("session", &file.id, offset, 1)
.await
.expect("chunk");
assert_eq!(chunk.data, [b"safe"[offset as usize]]);
}
store
.read_file("session", &file)
.await
.expect("read through original store");
assert_eq!(storage::HASH_FILE_CALLS.with(std::cell::Cell::get), 1);
})
.await;
}
#[tokio::test]
async fn changed_blob_is_reverified_and_failed_verification_is_not_cached() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let file = store
.publish_artifact("session", "result.txt".into(), "text/plain".into(), b"safe")
.await
.expect("artifact");
let metadata = load_metadata(
&store
.session_dir("session")
.join(&file.id)
.join(METADATA_FILE),
)
.await
.expect("metadata");
let blob = store.blob_path(&metadata.content_hash);
let modified = std::fs::metadata(&blob)
.expect("blob metadata")
.modified()
.expect("modified time");
storage::HASH_FILE_CALLS
.scope(std::cell::Cell::new(0), async {
store.read_file("session", &file).await.expect("warm cache");
assert_eq!(storage::HASH_FILE_CALLS.with(std::cell::Cell::get), 1);
std::fs::write(&blob, b"evil").expect("tamper");
std::fs::File::options()
.write(true)
.open(&blob)
.expect("blob")
.set_modified(modified + std::time::Duration::from_secs(2))
.expect("change verification stamp");
for expected_hashes in [2, 3] {
let error = store
.read_file("session", &file)
.await
.expect_err("tampered blob");
assert!(error.to_string().contains("content hash does not match"));
assert_eq!(
storage::HASH_FILE_CALLS.with(std::cell::Cell::get),
expected_hashes
);
}
})
.await;
}
#[tokio::test]
async fn tampered_content_blob_is_rejected() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let file = store
.publish_artifact("session", "result.txt".into(), "text/plain".into(), b"safe")
.await
.expect("artifact");
let metadata = load_metadata(
&store
.session_dir("session")
.join(&file.id)
.join(METADATA_FILE),
)
.await
.expect("metadata");
std::fs::write(store.blob_path(&metadata.content_hash), b"evil").expect("tamper");
assert!(store.read_chunk("session", &file.id, 0, 4).await.is_err());
}
#[test]
fn validated_blob_cache_is_bounded() {
let cache = StdMutex::new(BTreeMap::new());
let stamp = BlobValidationStamp {
size: 1,
modified: SystemTime::now(),
};
for index in 0..=MAX_VALIDATED_BLOBS {
remember_validated_blob(&cache, &format!("{index:064x}"), stamp);
}
let cache = cache.into_inner().expect("validation cache");
assert_eq!(cache.len(), MAX_VALIDATED_BLOBS);
assert!(cache.contains_key(&format!("{:064x}", MAX_VALIDATED_BLOBS)));
}
#[tokio::test]
async fn private_content_identity_round_trips_without_wire_changes() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let file = store
.publish_artifact(
"session",
"upload.txt".into(),
"text/plain".into(),
b"hello",
)
.await
.expect("artifact");
assert!(store.upload_content_hash("session", &file).await.is_err());
let mut pending = store
.begin_upload("session", "upload.txt".into(), 5, "text/plain".into())
.await
.expect("upload");
pending.append(0, b"hello").await.expect("append");
let upload = pending.finish().await.expect("finish");
store
.upload_content_hash("session", &upload)
.await
.expect("private hash");
assert_eq!(
store
.read_file("session", &upload)
.await
.expect("private blob"),
b"hello"
);
}
#[tokio::test]
async fn deleting_last_reference_garbage_collects_the_blob() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
store
.publish_artifact("first", "one.txt".into(), "text/plain".into(), b"same")
.await
.expect("first artifact");
store
.publish_artifact("second", "two.txt".into(), "text/plain".into(), b"same")
.await
.expect("second artifact");
store.delete_session("first").await.expect("delete first");
assert_eq!(blob_entries(&store), 1);
store.delete_session("second").await.expect("delete second");
assert_eq!(blob_entries(&store), 0);
}
#[tokio::test]
async fn deleting_an_upload_removes_its_last_content_blob() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let mut upload = store
.begin_upload("session", "input.txt".into(), 1, "text/plain".into())
.await
.expect("begin upload");
upload.append(0, b"x").await.expect("append upload");
let file = upload.finish().await.expect("finish upload");
store
.prepare_delete_sessions(
&["session".into()],
SessionFileSelection::Ids(vec![file.id.clone()]),
)
.await
.expect("prepare")
.delete()
.await
.expect("delete upload");
assert!(
store
.list_files(
"session",
&[crate::backend::session_files::SessionFileOrigin::Upload]
)
.await
.expect("list uploads")
.is_empty()
&& blob_entries(&store) == 0
);
}
#[tokio::test]
async fn deleting_an_upload_reports_blob_cleanup_failure_after_removing_reference() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let mut upload = store
.begin_upload("session", "input.txt".into(), 1, "text/plain".into())
.await
.expect("begin upload");
upload.append(0, b"x").await.expect("append upload");
let file = upload.finish().await.expect("finish upload");
std::fs::create_dir(store.blob_dir().join("invalid-entry")).expect("invalid blob entry");
store
.prepare_delete_sessions(
&["session".into()],
SessionFileSelection::Ids(vec![file.id.clone()]),
)
.await
.expect("prepare")
.delete()
.await
.expect_err("blob cleanup failure");
assert!(
store
.list_files(
"session",
&[crate::backend::session_files::SessionFileOrigin::Upload]
)
.await
.expect("list uploads")
.is_empty()
);
}
#[tokio::test]
async fn accepts_250_mib_and_rejects_larger_files() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let limits = session_file_limits();
let pending = store
.begin_upload(
"session",
"large.bin".into(),
limits.max_file_bytes,
"application/octet-stream".into(),
)
.await
.expect("250 MiB upload");
drop(pending);
assert!(
store
.begin_upload(
"session",
"too-large.bin".into(),
limits.max_file_bytes + 1,
"application/octet-stream".into(),
)
.await
.is_err()
);
}
#[tokio::test]
async fn delete_session_rejects_an_active_upload() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let pending = store
.begin_upload("session", "pending.txt".into(), 1, "text/plain".into())
.await
.expect("begin upload");
assert!(store.delete_session("session").await.is_err());
drop(pending);
store
.delete_session("session")
.await
.expect("delete released session files");
}
#[tokio::test]
async fn attachment_cleanup_remains_safe_after_real_workspace_replacement_and_restart() {
let state = tempfile::tempdir().expect("state");
let root = tempfile::tempdir().expect("workspace root");
let workspace = root.path().join("workspace");
let replacement = root.path().join("replacement");
let session_id = "session";
let staged = workspace
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id));
let replacement_staged = replacement
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id));
std::fs::create_dir_all(&staged).expect("staged attachments");
std::fs::create_dir_all(&replacement_staged).expect("replacement attachments");
let pinned_workspace = Dir::open_ambient_dir(&workspace, ambient_authority()).expect("pin");
SessionFileStore::new(state.path(), None)
.register_attachment_workspace(session_id, &pinned_workspace, &workspace)
.await
.expect("register workspace");
std::fs::rename(&workspace, root.path().join("original")).expect("move original");
std::fs::rename(&replacement, &workspace).expect("install replacement");
SessionFileStore::new(state.path(), None)
.delete_session(session_id)
.await
.expect("delete session");
assert!(
workspace
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id))
.is_dir()
);
}
#[cfg(unix)]
#[tokio::test]
async fn attachment_cleanup_remains_safe_after_symlink_workspace_replacement_and_restart() {
use std::os::unix::fs::symlink;
let state = tempfile::tempdir().expect("state");
let root = tempfile::tempdir().expect("workspace root");
let workspace = root.path().join("workspace");
let outside = root.path().join("outside");
let session_id = "session";
let outside_staged = outside
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id));
std::fs::create_dir_all(
workspace
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id)),
)
.expect("staged attachments");
std::fs::create_dir_all(&outside_staged).expect("outside attachments");
let pinned_workspace = Dir::open_ambient_dir(&workspace, ambient_authority()).expect("pin");
SessionFileStore::new(state.path(), None)
.register_attachment_workspace(session_id, &pinned_workspace, &workspace)
.await
.expect("register workspace");
std::fs::rename(&workspace, root.path().join("original")).expect("move original");
symlink(&outside, &workspace).expect("install symlink");
SessionFileStore::new(state.path(), None)
.delete_session(session_id)
.await
.expect("delete session");
assert!(outside_staged.is_dir());
}
#[tokio::test]
async fn attachment_cleanup_removes_staged_files_for_the_registered_workspace() {
let state = tempfile::tempdir().expect("state");
let workspace_root = tempfile::tempdir().expect("workspace");
let workspace = workspace_root.path();
let session_id = "session";
let staged = workspace
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id));
std::fs::create_dir_all(&staged).expect("staged attachments");
let pinned_workspace = Dir::open_ambient_dir(workspace, ambient_authority()).expect("pin");
let store = SessionFileStore::new(state.path(), None);
store
.register_attachment_workspace(session_id, &pinned_workspace, workspace)
.await
.expect("register workspace");
store
.delete_session(session_id)
.await
.expect("delete session");
assert!(!staged.exists());
}
#[cfg(target_os = "macos")]
#[tokio::test]
async fn attachment_workspace_persists_the_volume_uuid_instead_of_the_mount_device() {
let state = tempfile::tempdir().expect("state");
let workspace = tempfile::tempdir().expect("workspace");
let directory =
Dir::open_ambient_dir(workspace.path(), ambient_authority()).expect("workspace");
let store = SessionFileStore::new(state.path(), None);
store
.register_attachment_workspace("session", &directory, workspace.path())
.await
.expect("register workspace");
let marker =
tokio::fs::read_to_string(store.session_dir("session").join(ATTACHMENT_WORKSPACE_FILE))
.await
.expect("workspace marker");
let marker: serde_json::Value = serde_json::from_str(&marker).expect("valid marker");
assert!(marker["identity"]["volume_uuid"].is_string());
assert!(marker["identity"].get("device").is_none());
}
#[tokio::test]
async fn empty_deletion_does_not_initialize_or_scan_file_storage() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let mut deletion = store
.prepare_delete_sessions(&[], SessionFileSelection::All)
.await
.expect("prepare");
deletion.stage().await.expect("stage");
deletion.delete().await.expect("delete");
assert!(!store.root.exists());
}
#[tokio::test]
async fn staged_deletion_releases_uploads_and_resumes_after_restart() {
let state = tempfile::tempdir().expect("state");
let workspace = tempfile::tempdir().expect("workspace");
let staged = workspace
.path()
.join(".mobius/attachments")
.join(session_storage_key("deleted"));
std::fs::create_dir_all(&staged).expect("staged attachment directory");
let directory =
Dir::open_ambient_dir(workspace.path(), ambient_authority()).expect("workspace");
let store = SessionFileStore::new(state.path(), None);
store
.register_attachment_workspace("deleted", &directory, workspace.path())
.await
.expect("register");
let mut deletion = store
.prepare_delete_sessions(&["deleted".into()], SessionFileSelection::All)
.await
.expect("prepare");
deletion.stage().await.expect("stage deletion");
assert!(
staged.is_dir(),
"staging does not touch the external workspace"
);
tokio::time::timeout(
std::time::Duration::from_secs(1),
store.publish_artifact(
"retained",
"result.txt".into(),
"text/plain".into(),
b"result",
),
)
.await
.expect("other uploads remain available")
.expect("publish");
drop(deletion);
let reopened = SessionFileStore::new(state.path(), None);
reopened
.cleanup_deleted_sessions()
.await
.expect("resume cleanup");
assert!(!staged.exists());
assert_eq!(
reopened
.list_files(
"retained",
&[crate::backend::session_files::SessionFileOrigin::Artifact]
)
.await
.expect("retained files")
.len(),
1
);
}
#[tokio::test]
async fn attachment_registration_handles_path_replacement_safely() {
let state = tempfile::tempdir().expect("state");
let root = tempfile::tempdir().expect("workspace root");
let workspace = root.path().join("workspace");
let replacement = root.path().join("replacement");
let session_id = "session";
let replacement_staged = replacement
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id));
std::fs::create_dir_all(&workspace).expect("workspace");
std::fs::create_dir_all(&replacement_staged).expect("replacement attachments");
let pinned_workspace = Dir::open_ambient_dir(&workspace, ambient_authority()).expect("pin");
std::fs::rename(&workspace, root.path().join("original")).expect("move original");
std::fs::rename(&replacement, &workspace).expect("install replacement");
let registration = SessionFileStore::new(state.path(), None)
.register_attachment_workspace(session_id, &pinned_workspace, &workspace)
.await;
#[cfg(target_os = "macos")]
{
let error = registration.expect_err("replaced workspace must be rejected");
assert!(error.to_string().contains("workspace changed"));
}
#[cfg(not(target_os = "macos"))]
{
registration.expect("register pinned workspace");
SessionFileStore::new(state.path(), None)
.delete_session(session_id)
.await
.expect("delete session");
}
assert!(
workspace
.join(".mobius")
.join("attachments")
.join(session_storage_key(session_id))
.is_dir()
);
}
#[tokio::test]
async fn display_names_never_select_internal_storage_paths() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
for name in [METADATA_FILE, ".SESSION-FILE.JSON"] {
let mut pending = store
.begin_upload("session", name.into(), 1, "application/octet-stream".into())
.await
.expect("begin");
pending.append(0, b"x").await.expect("append");
let file = pending.finish().await.expect("finish");
assert_eq!(file.name, name);
assert_eq!(
store
.read_chunk("session", &file.id, 0, MAX_READ_CHUNK_BYTES)
.await
.expect("read")
.data,
b"x"
);
}
}
#[tokio::test]
async fn artifacts_are_downloadable_but_excluded_from_upload_access() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let file = store
.publish_artifact(
"session",
"report.xlsx".into(),
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet".into(),
&[0, 255, 1],
)
.await
.expect("publish");
assert!(
store
.list_files(
"session",
&[crate::backend::session_files::SessionFileOrigin::Upload]
)
.await
.expect("uploads")
.is_empty()
);
assert_eq!(
store
.read_chunk("session", &file.id, 0, 16)
.await
.expect("chunk")
.data,
[0, 255, 1]
);
assert!(
store
.read_chunk("another-session", &file.id, 0, 16)
.await
.is_err()
);
assert!(store.verify_upload("session", &file).await.is_err());
let reopened = SessionFileStore::new(state.path(), None);
assert_eq!(
reopened
.list_files(
"session",
&[crate::backend::session_files::SessionFileOrigin::Artifact]
)
.await
.expect("reopened artifacts"),
[(SessionFileOrigin::Artifact, file)]
);
}
#[tokio::test]
async fn upload_rejects_traversal_names() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
assert!(
store
.begin_upload("session", "../secret".into(), 1, "text/plain".into())
.await
.is_err()
);
}
#[tokio::test]
async fn pending_uploads_reserve_session_quota_and_release_on_drop() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let limits = session_file_limits();
let mut pending = Vec::new();
for index in 0..(limits.max_session_bytes / limits.max_file_bytes) {
pending.push(
store
.begin_upload(
"session",
format!("{index}.bin"),
limits.max_file_bytes,
"application/octet-stream".into(),
)
.await
.expect("reserve upload"),
);
}
assert!(
store
.begin_upload(
"session",
"overflow.bin".into(),
1,
"application/octet-stream".into(),
)
.await
.is_err()
);
drop(pending.pop());
assert!(
store
.begin_upload(
"session",
"replacement.bin".into(),
limits.max_file_bytes,
"application/octet-stream".into(),
)
.await
.is_ok()
);
}
#[tokio::test]
async fn advertised_file_count_is_enforced() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let limits = session_file_limits();
let mut pending = Vec::new();
for index in 0..limits.max_session_files {
pending.push(
store
.begin_upload(
"session",
format!("{index}.bin"),
1,
"application/octet-stream".into(),
)
.await
.expect("reserve file slot"),
);
}
assert!(
store
.begin_upload(
"session",
"overflow.bin".into(),
1,
"application/octet-stream".into(),
)
.await
.is_err()
);
}
#[tokio::test]
async fn advertised_upload_chunk_size_is_enforced() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let limits = session_file_limits();
let accepted = vec![0; limits.max_upload_chunk_bytes];
let rejected = vec![0; limits.max_upload_chunk_bytes + 1];
let mut upload = store
.begin_upload(
"session",
"large.bin".into(),
u64::try_from(rejected.len()).expect("upload size"),
"application/octet-stream".into(),
)
.await
.expect("begin upload");
upload.append(0, &accepted).await.expect("maximum chunk");
let offset = u64::try_from(accepted.len()).expect("chunk offset");
assert!(upload.append(offset, &rejected).await.is_err());
}
#[tokio::test]
async fn first_store_access_removes_crash_leftovers() {
let state = tempfile::tempdir().expect("state");
let store = SessionFileStore::new(state.path(), None);
let session_id = Uuid::new_v4().to_string();
let session = store.session_dir(&session_id);
std::fs::create_dir_all(&session).expect("session directory");
let temporary = session.join(".tmp-upload");
std::fs::write(&temporary, b"partial").expect("temporary upload");
let staging = session.join(format!(".{}-partial", Uuid::new_v4()));
std::fs::create_dir(&staging).expect("staging directory");
std::fs::write(staging.join("payload"), b"partial").expect("staged file");
assert!(
store
.list_files(
&session_id,
&[crate::backend::session_files::SessionFileOrigin::Upload]
)
.await
.expect("list")
.is_empty()
);
assert!(!temporary.exists());
assert!(!staging.exists());
}
fn blob_entries(store: &SessionFileStore) -> usize {
std::fs::read_dir(store.blob_dir())
.expect("blob directory")
.count()
}
#[tokio::test]
async fn text_forks_need_no_store_and_media_forks_fail_closed() {
let text = serde_json::json!({"type":"message","role":"user","content":[{"type":"input_text","text":"hello"}]});
grant_context(None, "parent", "child", &[text])
.await
.expect("text fork");
for part in [
serde_json::json!({"type":"input_image"}),
serde_json::json!({"type":"file","file":{}}),
] {
let input = serde_json::json!({"type":"message","role":"user","content":[part]});
assert!(
grant_context(None, "parent", "child", &[input])
.await
.is_err()
);
}
}
#[tokio::test]
async fn usage_counts_observations_and_shared_references_without_exposing_hidden_files() {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), None);
let mut pending = store
.begin_upload("source", "input.txt".into(), 5, "text/plain".into())
.await
.unwrap();
pending.append(0, b"input").await.unwrap();
let upload = pending.finish().await.unwrap();
store
.publish_artifact(
"source",
"result.txt".into(),
"text/plain".into(),
b"result",
)
.await
.unwrap();
store
.grant_file("source", "observer", &upload)
.await
.unwrap();
assert_eq!(
store
.list_files("source", &[SessionFileOrigin::Upload])
.await
.unwrap(),
vec![(SessionFileOrigin::Upload, upload.clone())]
);
assert_eq!(
store
.list_files("observer", &[SessionFileOrigin::Observation])
.await
.unwrap(),
vec![(SessionFileOrigin::Observation, upload)]
);
}
#[tokio::test]
async fn gateway_quota_counts_cross_session_reservations_and_releases_after_purge() {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), Some(5));
let mut pending = store
.begin_upload("first", "input.txt".into(), 5, "text/plain".into())
.await
.unwrap();
assert!(matches!(
store
.begin_upload("second", "extra.txt".into(), 1, "text/plain".into())
.await,
Err(Error::StorageFull)
));
pending.append(0, b"input").await.unwrap();
pending.finish().await.unwrap();
assert!(matches!(
store
.publish_artifact("second", "extra.txt".into(), "text/plain".into(), b"x")
.await,
Err(Error::StorageFull)
));
store
.prepare_delete_sessions(
&["first".into()],
SessionFileSelection::Origins(vec![SessionFileOrigin::Upload]),
)
.await
.unwrap()
.delete()
.await
.unwrap();
store
.publish_artifact("second", "extra.txt".into(), "text/plain".into(), b"x")
.await
.unwrap();
}
#[tokio::test]
async fn origin_purge_preserves_uploads_and_shared_fork_blobs_and_removes_staged_copies() {
let state = tempfile::tempdir().unwrap();
let workspace = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), None);
let mut pending = store
.begin_upload("parent", "input.txt".into(), 5, "text/plain".into())
.await
.unwrap();
pending.append(0, b"input").await.unwrap();
let upload = pending.finish().await.unwrap();
store
.publish_artifact(
"parent",
"result.txt".into(),
"text/plain".into(),
b"result",
)
.await
.unwrap();
store.grant_file("parent", "fork", &upload).await.unwrap();
let observation = store
.publish_artifact(
"source",
"observed.txt".into(),
"text/plain".into(),
b"observed",
)
.await
.unwrap();
store
.grant_file("source", "parent", &observation)
.await
.unwrap();
store.delete_session("source").await.unwrap();
assert_eq!(
store
.list_files("parent", &[SessionFileOrigin::Observation])
.await
.unwrap()
.len(),
1
);
let pinned = Dir::open_ambient_dir(workspace.path(), ambient_authority()).unwrap();
store
.register_attachment_workspace("parent", &pinned, workspace.path())
.await
.unwrap();
let staged = workspace
.path()
.join(".mobius/attachments")
.join(session_storage_key("parent"))
.join(&upload.id);
std::fs::create_dir_all(&staged).unwrap();
std::fs::write(staged.join("input.txt"), b"input").unwrap();
store
.prepare_delete_sessions(
&["parent".into()],
SessionFileSelection::Origins(vec![
SessionFileOrigin::Artifact,
SessionFileOrigin::Observation,
]),
)
.await
.unwrap()
.delete()
.await
.unwrap();
assert_eq!(
store
.list_files("parent", &[SessionFileOrigin::Upload])
.await
.unwrap()
.len(),
1
);
assert!(
store
.list_files(
"parent",
&[SessionFileOrigin::Artifact, SessionFileOrigin::Observation]
)
.await
.unwrap()
.is_empty()
);
store
.prepare_delete_sessions(
&["parent".into()],
SessionFileSelection::Origins(vec![SessionFileOrigin::Upload]),
)
.await
.unwrap()
.delete()
.await
.unwrap();
assert!(!staged.exists());
assert_eq!(store.blob_bytes().await.unwrap(), 5);
assert_eq!(
store
.list_files("fork", &[SessionFileOrigin::Observation])
.await
.unwrap()
.len(),
1
);
store.delete_session("fork").await.unwrap();
assert_eq!(store.blob_bytes().await.unwrap(), 0);
}
#[tokio::test]
async fn stored_references_can_be_granted_and_published_at_capacity() {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), Some(5));
let mut upload = store
.begin_upload("source", "input.txt".into(), 5, "text/plain".into())
.await
.unwrap();
upload.append(0, b"input").await.unwrap();
let file = upload.finish().await.unwrap();
store.grant_file("source", "fork", &file).await.unwrap();
store
.grant_upload("source", "execution", &file)
.await
.unwrap();
let artifact = store.publish_reference("source", &file).await.unwrap();
store
.share_artifact("source", "target", &artifact)
.await
.unwrap();
assert_eq!(store.stored_bytes().await.unwrap(), 5);
assert_eq!(store.read_file("fork", &file).await.unwrap(), b"input");
assert!(matches!(
store
.begin_upload("new", "extra.txt".into(), 1, "text/plain".into())
.await,
Err(Error::StorageFull)
));
}
#[tokio::test]
async fn simultaneous_upload_admission_cannot_overbook_gateway_capacity() {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), Some(5));
let (first, second) = tokio::join!(
store.begin_upload("first", "input.txt".into(), 5, "text/plain".into()),
store.begin_upload("second", "input.txt".into(), 5, "text/plain".into()),
);
assert!(matches!(
(&first, &second),
(Ok(_), Err(Error::StorageFull)) | (Err(Error::StorageFull), Ok(_))
));
drop((first, second));
store
.begin_upload("retry", "input.txt".into(), 5, "text/plain".into())
.await
.unwrap();
}
#[tokio::test]
async fn deleting_one_completed_file_preserves_another_pending_upload() {
for by_origin in [false, true] {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), None);
let mut completed = store
.begin_upload("session", "first.txt".into(), 1, "text/plain".into())
.await
.unwrap();
completed.append(0, b"a").await.unwrap();
let first = completed.finish().await.unwrap();
let mut pending = store
.begin_upload("session", "second.txt".into(), 1, "text/plain".into())
.await
.unwrap();
pending.append(0, b"b").await.unwrap();
store
.prepare_delete_sessions(
&["session".into()],
if by_origin {
SessionFileSelection::Origins(vec![SessionFileOrigin::Upload])
} else {
SessionFileSelection::Ids(vec![first.id.clone()])
},
)
.await
.unwrap()
.delete()
.await
.unwrap();
let second = pending.finish().await.unwrap();
assert!(store.read_file("session", &first).await.is_err());
assert_eq!(store.read_file("session", &second).await.unwrap(), b"b");
}
}
#[tokio::test]
async fn empty_file_selections_are_rejected() {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), None);
for selection in [
SessionFileSelection::Ids(vec![]),
SessionFileSelection::Origins(vec![]),
] {
assert!(
store
.prepare_delete_sessions(&["session".into()], selection)
.await
.is_err()
);
}
}
#[tokio::test]
async fn selected_deletion_handles_damaged_payloads_and_missing_metadata() {
for by_origin in [false, true] {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), None);
let file = store
.publish_artifact(
"session",
"damaged.txt".into(),
"text/plain".into(),
b"payload",
)
.await
.unwrap();
let record = load_metadata(
&store
.session_dir("session")
.join(&file.id)
.join(METADATA_FILE),
)
.await
.unwrap();
if by_origin {
tokio::fs::remove_file(store.blob_path(&record.content_hash))
.await
.unwrap();
} else {
tokio::fs::write(
store
.session_dir("session")
.join(&file.id)
.join(METADATA_FILE),
b"invalid json",
)
.await
.unwrap();
}
let selection = if by_origin {
SessionFileSelection::Origins(vec![SessionFileOrigin::Artifact])
} else {
SessionFileSelection::Ids(vec![file.id])
};
let mut deletion = store
.prepare_delete_sessions(&["session".into()], selection)
.await
.unwrap();
deletion.delete().await.unwrap();
assert!(
store
.list_files("session", &[SessionFileOrigin::Artifact])
.await
.unwrap()
.is_empty()
);
let _commit = store.commits.lock().await;
tokio::fs::create_dir(store.blob_dir().join("invalid-blob"))
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(1), deletion.delete())
.await
.unwrap()
.unwrap();
}
}
#[tokio::test]
async fn generated_images_survive_selective_cleanup_until_the_chat_is_deleted() {
for legacy in [false, true] {
let state = tempfile::tempdir().unwrap();
let store = SessionFileStore::new(state.path(), None);
let mut encoded = std::io::Cursor::new(Vec::new());
image::DynamicImage::ImageRgba8(image::RgbaImage::new(2, 2))
.write_to(&mut encoded, image::ImageFormat::Png)
.unwrap();
let bytes = encoded.into_inner();
let name = if legacy {
"generated-image.png"
} else {
"custom-name.png"
};
let generated = store
.publish_image("session", name.into(), "image/png", bytes.clone())
.await
.unwrap();
if legacy {
let path = store
.session_dir("session")
.join(&generated.id)
.join(METADATA_FILE);
let mut metadata: serde_json::Value =
serde_json::from_slice(&tokio::fs::read(&path).await.unwrap()).unwrap();
metadata
.as_object_mut()
.unwrap()
.remove("protected_from_cleanup");
tokio::fs::write(path, serde_json::to_vec(&metadata).unwrap())
.await
.unwrap();
}
let ordinary = store
.publish_artifact("session", "ordinary.png".into(), "image/png".into(), &bytes)
.await
.unwrap();
let origins = [SessionFileOrigin::Artifact];
assert_eq!(
store.list_files("session", &origins).await.unwrap().len(),
2
);
assert_eq!(
store.read_image("session", &generated).await.unwrap(),
bytes
);
assert_eq!(
store.list_cleanup_files("session", &origins).await.unwrap(),
vec![(SessionFileOrigin::Artifact, ordinary.clone())]
);
let error = store
.prepare_delete_sessions(
&["session".into()],
SessionFileSelection::Ids(vec![ordinary.id.clone(), generated.id.clone()]),
)
.await
.unwrap()
.delete()
.await
.unwrap_err();
assert!(error.to_string().contains("generated images"));
assert_eq!(
store.list_files("session", &origins).await.unwrap().len(),
2
);
assert_eq!(store.read_image("session", &ordinary).await.unwrap(), bytes);
assert_eq!(
store.read_image("session", &generated).await.unwrap(),
bytes
);
store
.prepare_delete_sessions(
&["session".into()],
SessionFileSelection::Origins(origins.to_vec()),
)
.await
.unwrap()
.delete()
.await
.unwrap();
assert_eq!(
store.list_files("session", &origins).await.unwrap(),
vec![(SessionFileOrigin::Artifact, generated.clone())]
);
assert!(
store
.list_cleanup_files("session", &origins)
.await
.unwrap()
.is_empty()
);
assert_eq!(
store.read_image("session", &generated).await.unwrap(),
bytes
);
assert!(store.read_image("session", &ordinary).await.is_err());
store.delete_session("session").await.unwrap();
assert!(
store
.list_files("session", &origins)
.await
.unwrap()
.is_empty()
);
assert!(store.read_image("session", &generated).await.is_err());
assert_eq!(store.stored_bytes().await.unwrap(), 0);
}
}