lywsvip/ironclaw
1
0
Fork 0
mirror of https://github.com/nearai/ironclaw.git synced 2026-09-03 08:06:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(processes): enforce tenant isolation via ScopedFilesystem in FilesystemProcessStore

Browse Source 
Mirrors the engine migration in commit ac8e677f9 and the parallel
secrets/authorization/outbound refactors: `FilesystemProcessStore` and
`FilesystemProcessResultStore` previously held `Arc<F: RootFilesystem>`
and manually encoded tenant/user identity in every path via
`resource_owner_root` (formatted
`/engine/tenants/<tenant>/users/<user>/agents/.../projects/...`). Any
composition layer that forgot to wrap the backend in a tenant scope
would leak across tenants with the type system saying nothing — the
HIGH-severity class of finding tracked in PR #3679 for `ironclaw_engine`.

Migrated the process lifecycle / result stores so tenant isolation is
structural rather than something this crate has to re-derive from
`ResourceScope.tenant_id` / `user_id`:

- `FilesystemProcessStore::new` and `from_arc`, plus
  `FilesystemProcessResultStore::new` and `from_arc`, now take
  `Arc<ScopedFilesystem<F>>`. The internal `FilesystemHandle`
  borrow/Arc enum is gone — both surfaces share the same shape.
- Path helpers (`process_record_path`, `process_records_root`,
  `process_result_path`, `process_output_path`) return `ScopedPath`
  instead of `VirtualPath`. The on-disk layout under the `/processes`
  mount alias is:

      /processes[/agents/<agent>][/projects/<project>][/missions/<mission>][/threads/<thread>]/records/<process_id>.json
      /processes[/agents/<agent>][/...]/results/<process_id>.json
      /processes[/agents/<agent>][/...]/outputs/<process_id>/output.json

  The leading `/engine/tenants/<tenant>/users/<user>` prefix is gone;
  the caller's `MountView` resolves the `/processes` alias to the
  tenant/user-scoped target at every op. Sub-scope axes
  (agent/project/mission/thread) stay in the alias-relative path
  because they are within-tenant scoping, not covered by the
  per-tenant MountAlias.
- `put_with_byte_fallback` now takes `&ScopedFilesystem<F>` /
  `&ScopedPath`. The fallback semantics are unchanged: record-shaped
  entries and non-`Any` CAS are stripped/downgraded if the backend
  reports `Unsupported`, so byte-only mounts (LocalFilesystem) keep
  working through the single-instance `transition_lock`.
- `records_for_scope_via_list` reconstructs each child `ScopedPath`
  from the `list_dir`-returned `VirtualPath` leaf so the per-op ACL
  applies to the follow-up `get` (mirrors the engine store's
  `join_scoped` shape).
- `FilesystemProcessResultStore` resolves the `ScopedPath` output blob
  to a `VirtualPath` via `ScopedFilesystem::mounts().resolve()` before
  recording it on the `ProcessResultRecord`, so the on-wire
  `output_ref` shape stays a tenant-scoped `VirtualPath` and the
  existing forged-ref rejection in `output()` still works (now
  comparing against the resolved expected path, not a hand-formatted
  string).
- `ProcessServices::filesystem` constructor takes
  `Arc<ScopedFilesystem<F>>` instead of `Arc<F>`. Composition wiring
  in `ironclaw_reborn_composition` is tracked separately under the
  scoped-filesystem epic and is not touched here.

Tests:

- `tests/process_store_contract.rs` and
  `tests/process_services_contract.rs` now construct a
  `ScopedFilesystem` over `LocalFilesystem` / `InMemoryBackend` with a
  `MountPermissions::read_write_list_delete()` grant on the
  `/processes` alias pointing at a tenant1/user1 target. The existing
  cross-scope assertions (`filesystem_process_result_store_persists_under_resource_scope`,
  `filesystem_process_result_store_rejects_unexpected_output_refs`,
  etc.) keep working because the post-read `same_scope_owner` check
  still filters out forged records whose in-body scope differs from
  the request scope.
- New regression test
  `filesystem_process_store_isolates_two_tenants_with_same_user_project_ids`
  wires two `FilesystemProcessStore`s over one `InMemoryBackend` with
  different `MountView` targets but identical `user_id` / `project_id`
  request scopes. Writing on tenant A must not be visible from tenant
  B — fails closed if the ScopedFilesystem wrapping ever regresses to
  raw `Arc<F: RootFilesystem>`.
- Reworked
  `filesystem_process_store_records_for_scope_uses_index_on_record_backend`
  to drive within-tenant project discrimination (the new
  scope axis that lives in the path), since cross-tenant
  discrimination via the path is now a separate-stores test rather
  than a single-store test.

`cargo clippy -p ironclaw_processes --all-features --tests -- -D warnings`
and `cargo test -p ironclaw_processes --all-features` both clean
(65 integration tests + 1 unit test).
This commit is contained in:
ilblackdragon@gmail.com
2026-05-16 17:11:25 -07:00
parent 4ae56769b0
commit 81664dd293
4 changed files with 448 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

369
crates/ironclaw_processes/src/filesystem_store.rs
View File

@@ -1,12 +1,24 @@
//! Filesystem-backed process and process-result stores.
//!
//! Records are stored as JSON under the exact resource-owner path
//! `tenants/<tenant>/users/<user>[/agents/<agent>][/projects/<project>][/missions/<mission>][/threads/<thread>]/`,
//! split into:
//! Records live under the `/processes` mount alias on a
//! [`ScopedFilesystem`](ironclaw_filesystem::ScopedFilesystem). The paths in
//! this module are alias-relative [`ScopedPath`] strings — at every op the
//! [`ScopedFilesystem`] resolves the alias against its caller-supplied
//! [`MountView`](ironclaw_host_api::MountView) and enforces per-grant ACL
//! before backend dispatch. The composition layer wires the alias to a
//! tenant/user-scoped [`VirtualPath`](ironclaw_host_api::VirtualPath), so
//! tenant isolation is structural rather than something this crate must
//! re-derive from `ResourceScope.tenant_id`/`user_id`.
//!
//! - `processes/<process_id>.json` — lifecycle records ([`FilesystemProcessStore`])
//! - `process-results/<process_id>.json` — terminal result metadata
//! - `process-outputs/<process_id>/output.json` — large/sensitive output bodies
//! Within the alias, sub-scope (`agent_id`, `project_id`, `mission_id`,
//! `thread_id`) is still encoded in the path so a single tenant/user can
//! own multiple agent/project/mission/thread cells:
//!
//! ```text
//! /processes[/agents/<agent>][/projects/<project>][/missions/<mission>][/threads/<thread>]/records/<process_id>.json
//! /processes[/agents/<agent>][/projects/<project>][/missions/<mission>][/threads/<thread>]/results/<process_id>.json
//! /processes[/agents/<agent>][/projects/<project>][/missions/<mission>][/threads/<thread>]/outputs/<process_id>/output.json
//! ```
//!
//! All path/serde helpers are private to this module since they are tied to
//! the on-disk layout above.
@@ -17,9 +29,9 @@ use async_trait::async_trait;
use ironclaw_events::sanitize_error_kind;
use ironclaw_filesystem::{
CasExpectation, ContentType, Entry, FilesystemError, Filter, IndexKey, IndexKind, IndexName,
IndexSpec, IndexValue, Page, RootFilesystem,
IndexSpec, IndexValue, Page, RootFilesystem, ScopedFilesystem,
};
use ironclaw_host_api::{ProcessId, ResourceScope, VirtualPath};
use ironclaw_host_api::{ProcessId, ResourceScope, ScopedPath, VirtualPath};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::Mutex as AsyncMutex;
@@ -29,35 +41,25 @@ use crate::types::{
ProcessStatus, ProcessStore, ensure_status_transition, invalid_path, same_scope_owner,
};
pub(crate) enum FilesystemHandle<'a, F>
/// Filesystem-backed [`ProcessStore`].
///
/// Construct with a [`ScopedFilesystem`] over any [`RootFilesystem`] —
/// typically a composite root in production or the in-memory backend in
/// tests. The [`ScopedFilesystem`] enforces the caller's
/// [`MountView`](ironclaw_host_api::MountView) per-operation ACL and
/// resolves the `/processes` alias to a tenant-scoped
/// [`VirtualPath`](ironclaw_host_api::VirtualPath) before any backend
/// dispatch — so tenant isolation is structural, not a convention this
/// crate has to remember.
pub struct FilesystemProcessStore<F>
where
F: RootFilesystem,
{
Borrowed(&'a F),
Shared(Arc<F>),
}
impl<F> FilesystemHandle<'_, F>
where
F: RootFilesystem,
{
fn as_ref(&self) -> &F {
match self {
Self::Borrowed(filesystem) => filesystem,
Self::Shared(filesystem) => filesystem.as_ref(),
}
}
}
pub struct FilesystemProcessStore<'a, F>
where
F: RootFilesystem,
{
filesystem: FilesystemHandle<'a, F>,
filesystem: Arc<ScopedFilesystem<F>>,
transition_lock: AsyncMutex<()>,
}
impl<'a, F> FilesystemProcessStore<'a, F>
impl<F> FilesystemProcessStore<F>
where
F: RootFilesystem,
{
@@ -69,23 +71,22 @@ where
/// concurrently against the same on-disk root is unsupported and will
/// race on the JSON record files. Construct the store once and share via
/// `Arc` (see [`from_arc`](Self::from_arc)).
pub fn new(filesystem: &'a F) -> Self {
pub fn new(filesystem: Arc<ScopedFilesystem<F>>) -> Self {
Self {
filesystem: FilesystemHandle::Borrowed(filesystem),
filesystem,
transition_lock: AsyncMutex::new(()),
}
}
/// Construct an owned (`'static`) variant from a shared filesystem handle.
/// Convenience constructor mirroring [`new`](Self::new) — kept so call
/// sites that previously held an [`Arc<ScopedFilesystem<F>>`] separately
/// from a borrow can continue to use the same shape.
///
/// The same single-instance invariant from [`new`](Self::new) applies:
/// share the resulting store via `Arc` rather than constructing multiple
/// instances pointed at the same root.
pub fn from_arc(filesystem: Arc<F>) -> FilesystemProcessStore<'static, F> {
FilesystemProcessStore {
filesystem: FilesystemHandle::Shared(filesystem),
transition_lock: AsyncMutex::new(()),
}
pub fn from_arc(filesystem: Arc<ScopedFilesystem<F>>) -> Self {
Self::new(filesystem)
}
async fn write_record(&self, record: &ProcessRecord) -> Result<(), ProcessError> {
@@ -93,46 +94,46 @@ where
let body = serialize_pretty(record)?;
self.ensure_indexes(&record.scope).await?;
let entry = process_record_entry(body, record);
put_with_byte_fallback(self.filesystem.as_ref(), &path, entry, CasExpectation::Any).await?;
put_with_byte_fallback(&self.filesystem, &path, entry, CasExpectation::Any).await?;
Ok(())
}
/// Declare the indexed-projection fields on the per-owner `processes/`
/// Declare the indexed-projection fields on the per-owner `records/`
/// prefix so `records_for_scope` can use a native `query` filter.
/// Tolerates `Unsupported` for byte-only backends (e.g. LocalFilesystem)
/// so the existing list+get fallback path is still reachable.
async fn ensure_indexes(&self, scope: &ResourceScope) -> Result<(), ProcessError> {
let prefix = process_records_root(scope)?;
ensure_exact_index(
self.filesystem.as_ref(),
&self.filesystem,
&prefix,
index_name("processes_by_tenant"),
index_key_tenant_id(),
)
.await?;
ensure_exact_index(
self.filesystem.as_ref(),
&self.filesystem,
&prefix,
index_name("processes_by_user"),
index_key_user_id(),
)
.await?;
ensure_exact_index(
self.filesystem.as_ref(),
&self.filesystem,
&prefix,
index_name("processes_by_status"),
index_key_status(),
)
.await?;
ensure_exact_index(
self.filesystem.as_ref(),
&self.filesystem,
&prefix,
index_name("processes_by_extension"),
index_key_extension_id(),
)
.await?;
ensure_exact_index(
self.filesystem.as_ref(),
&self.filesystem,
&prefix,
index_name("processes_by_parent"),
index_key_parent_process_id(),
@@ -152,7 +153,7 @@ where
///
/// Backends without versioning (LocalFilesystem) return version `0`
/// for every read and reject `CasExpectation::Version` with
/// `Unsupported`; for those, `put_with_byte_fallback_versioned` falls
/// `Unsupported`; for those, [`put_with_byte_fallback`] falls
/// through to `CasExpectation::Any` so the existing single-instance
/// guarantee from `transition_lock` carries the safety invariant.
async fn update_status(
@@ -165,7 +166,7 @@ where
let _guard = self.transition_lock.lock().await;
for _ in 0..MAX_CAS_RETRIES {
let path = process_record_path(scope, process_id)?;
let Some(versioned) = self.filesystem.as_ref().get(&path).await? else {
let Some(versioned) = self.filesystem.get(&path).await? else {
return Err(ProcessError::UnknownProcess { process_id });
};
let mut record = deserialize::<ProcessRecord>(&versioned.entry.body)?;
@@ -180,7 +181,7 @@ where
let body = serialize_pretty(&record)?;
let entry = process_record_entry(body, &record);
match put_with_byte_fallback(
self.filesystem.as_ref(),
&self.filesystem,
&path,
entry,
CasExpectation::Version(versioned.version),
@@ -206,9 +207,9 @@ where
const MAX_CAS_RETRIES: usize = 5;
#[async_trait]
impl<F> ProcessStore for FilesystemProcessStore<'_, F>
impl<F> ProcessStore for FilesystemProcessStore<F>
where
F: RootFilesystem,
F: RootFilesystem + 'static,
{
async fn start(&self, start: ProcessStart) -> Result<ProcessRecord, ProcessError> {
let _guard = self.transition_lock.lock().await;
@@ -218,7 +219,7 @@ where
// transition_lock per the single-instance invariant in this struct's
// docstring. A future migration can switch to `CasExpectation::Absent`
// once every backend in production exposes native put.
if self.filesystem.as_ref().get(&path).await?.is_some() {
if self.filesystem.get(&path).await?.is_some() {
return Err(ProcessError::ProcessAlreadyExists {
process_id: start.process_id,
});
@@ -281,7 +282,7 @@ where
process_id: ProcessId,
) -> Result<Option<ProcessRecord>, ProcessError> {
let path = process_record_path(scope, process_id)?;
let Some(versioned) = self.filesystem.as_ref().get(&path).await? else {
let Some(versioned) = self.filesystem.get(&path).await? else {
return Ok(None);
};
let record = deserialize::<ProcessRecord>(&versioned.entry.body)?;
@@ -299,12 +300,17 @@ where
) -> Result<Vec<ProcessRecord>, ProcessError> {
let root = process_records_root(scope)?;
// Try the indexed query path first. The `tenant_id` + `user_id`
// pair narrows backend-side to the same owner the path encodes,
// and the post-query `same_scope_owner` check guards the
// remaining sub-scope (agent/project/mission/thread) axes that
// are not in the index spec yet. Backends without index support
// (LocalFilesystem) return `Unsupported` and we fall back to the
// legacy list+get scan so behaviour is identical.
// pair is still projected onto each record so a backend serving
// a shared root (e.g. tests reusing one InMemoryBackend across
// mount views) can distinguish records. With the ScopedFilesystem
// refactor the path itself already encodes tenant/user via the
// MountView, but the indexed projection stays as belt-and-braces
// — backends that share storage across MountViews must still
// produce the right rows. The post-query `same_scope_owner`
// check guards the remaining sub-scope (agent/project/mission/
// thread) axes that are not in the index spec yet. Backends
// without index support (LocalFilesystem) return `Unsupported`
// and we fall back to the legacy list+get scan.
self.ensure_indexes(scope).await?;
let filter = Filter::And(vec![
Filter::Eq {
@@ -316,7 +322,7 @@ where
value: IndexValue::Text(scope.user_id.as_str().to_string()),
},
]);
match query_all_records(self.filesystem.as_ref(), &root, &filter).await {
match query_all_records(&self.filesystem, &root, &filter).await {
Ok(records) => {
let mut filtered = records
.into_iter()
@@ -333,7 +339,7 @@ where
}
}
impl<F> FilesystemProcessStore<'_, F>
impl<F> FilesystemProcessStore<F>
where
F: RootFilesystem,
{
@@ -344,32 +350,39 @@ where
async fn records_for_scope_via_list(
&self,
scope: &ResourceScope,
root: &VirtualPath,
root: &ScopedPath,
) -> Result<Vec<ProcessRecord>, ProcessError> {
let entries = match self.filesystem.as_ref().list_dir(root).await {
let entries = match self.filesystem.list_dir(root).await {
Ok(entries) => entries,
Err(error) if is_not_found(&error) => return Ok(Vec::new()),
Err(error) => return Err(error.into()),
};
let mut records = Vec::new();
for entry in entries {
if entry.name.ends_with(".json") {
// Reviewer (PR #3666) flagged: a `get` returning `None` after
// `list_dir` enumerated the path indicates a race or backend
// inconsistency. Returning a partial process list silently
// hides this; surface it as `NotFound` so callers see the
// same failure shape they got with the legacy `read_file`
// path.
let Some(versioned) = self.filesystem.as_ref().get(&entry.path).await? else {
return Err(ProcessError::Filesystem(format!(
"process record listed but missing at {}",
entry.path
)));
};
let record = deserialize::<ProcessRecord>(&versioned.entry.body)?;
if same_scope_owner(&record.scope, scope) {
records.push(record);
}
if !entry.name.ends_with(".json") {
continue;
}
// `list_dir` returns `VirtualPath`s because resolution has
// already happened. We reconstruct the child as a
// [`ScopedPath`] under the same alias-relative prefix so the
// per-op ACL still runs on the follow-up `get` (mirrors the
// engine store's `list_subdir_names` shape).
let scoped_child = join_scoped(root, &entry.name)?;
// Reviewer (PR #3666) flagged: a `get` returning `None` after
// `list_dir` enumerated the path indicates a race or backend
// inconsistency. Returning a partial process list silently
// hides this; surface it as a filesystem error so callers see
// the same failure shape they got with the legacy `read_file`
// path.
let Some(versioned) = self.filesystem.get(&scoped_child).await? else {
return Err(ProcessError::Filesystem(format!(
"process record listed but missing at {}",
scoped_child.as_str()
)));
};
let record = deserialize::<ProcessRecord>(&versioned.entry.body)?;
if same_scope_owner(&record.scope, scope) {
records.push(record);
}
}
records.sort_by_key(|record| record.process_id.as_uuid());
@@ -377,27 +390,25 @@ where
}
}
pub struct FilesystemProcessResultStore<'a, F>
pub struct FilesystemProcessResultStore<F>
where
F: RootFilesystem,
{
filesystem: FilesystemHandle<'a, F>,
filesystem: Arc<ScopedFilesystem<F>>,
}
impl<'a, F> FilesystemProcessResultStore<'a, F>
impl<F> FilesystemProcessResultStore<F>
where
F: RootFilesystem,
{
pub fn new(filesystem: &'a F) -> Self {
Self {
filesystem: FilesystemHandle::Borrowed(filesystem),
}
pub fn new(filesystem: Arc<ScopedFilesystem<F>>) -> Self {
Self { filesystem }
}
pub fn from_arc(filesystem: Arc<F>) -> FilesystemProcessResultStore<'static, F> {
FilesystemProcessResultStore {
filesystem: FilesystemHandle::Shared(filesystem),
}
/// Convenience constructor mirroring [`new`](Self::new); preserved so
/// existing call sites (composition factories, tests) keep their shape.
pub fn from_arc(filesystem: Arc<ScopedFilesystem<F>>) -> Self {
Self::new(filesystem)
}
async fn write_result(&self, record: &ProcessResultRecord) -> Result<(), ProcessError> {
@@ -405,7 +416,6 @@ where
let body = serialize_pretty(record)?;
let entry = Entry::bytes(body).with_content_type(ContentType::json());
self.filesystem
.as_ref()
.put(&path, entry, CasExpectation::Any)
.await?;
Ok(())
@@ -424,10 +434,19 @@ where
// backwards-compatible with any caller that uses `read_file`.
let entry = Entry::bytes(body).with_content_type(ContentType::json());
self.filesystem
.as_ref()
.put(&path, entry, CasExpectation::Any)
.await?;
Ok(path)
// The on-disk `output_ref` recorded in the result record must be a
// [`VirtualPath`] (it is part of the wire surface of
// [`ProcessResultRecord`]) — resolve the alias through the
// caller's MountView so the value persisted matches the
// resolution any future reader would see for the same scope.
let virtual_path = self
.filesystem
.mounts()
.resolve(&path)
.map_err(invalid_path)?;
Ok(virtual_path)
}
async fn store_result(
@@ -453,19 +472,19 @@ where
}
#[async_trait]
impl<F> ProcessResultStore for FilesystemProcessResultStore<'_, F>
impl<F> ProcessResultStore for FilesystemProcessResultStore<F>
where
F: RootFilesystem,
F: RootFilesystem + 'static,
{
/// Persist a successful terminal record and its output blob.
///
/// Writes happen in two steps (`write_output` then `write_result`); if
/// the second write fails, the output blob at
/// `process-outputs/<process_id>/output.json` is left on disk as an
/// `outputs/<process_id>/output.json` is left on disk as an
/// orphan. Cleanup of orphaned output blobs is the caller's responsibility
/// (typically swept during operator-initiated reconciliation rather than
/// inline, since orphans are observable via missing
/// `process-results/<process_id>.json`).
/// `results/<process_id>.json`).
async fn complete(
&self,
scope: &ResourceScope,
@@ -516,7 +535,7 @@ where
process_id: ProcessId,
) -> Result<Option<ProcessResultRecord>, ProcessError> {
let path = process_result_path(scope, process_id)?;
let Some(versioned) = self.filesystem.as_ref().get(&path).await? else {
let Some(versioned) = self.filesystem.get(&path).await? else {
return Ok(None);
};
let record = deserialize::<ProcessResultRecord>(&versioned.entry.body)?;
@@ -542,95 +561,121 @@ where
let Some(output_ref) = record.output_ref else {
return Ok(None);
};
let expected_output_ref = process_output_path(scope, process_id)?;
if output_ref != expected_output_ref {
// The stored `output_ref` is a tenant-scoped [`VirtualPath`]; we
// compare it against the resolved view of the path the current
// scope would produce, so a forged record whose `output_ref`
// points at a sibling tenant/scope's blob is rejected before any
// read. After the match passes, we read the blob through the
// scoped path (going through the per-op ACL) rather than the raw
// `VirtualPath` so backends with stricter scopes still apply
// their checks.
let expected_scoped = process_output_path(scope, process_id)?;
let expected_virtual = self
.filesystem
.mounts()
.resolve(&expected_scoped)
.map_err(invalid_path)?;
if output_ref != expected_virtual {
return Err(invalid_stored_record(format!(
"process result output ref {} does not match expected {}",
output_ref.as_str(),
expected_output_ref.as_str()
expected_virtual.as_str()
)));
}
let Some(versioned) = self.filesystem.as_ref().get(&output_ref).await? else {
let Some(versioned) = self.filesystem.get(&expected_scoped).await? else {
return Ok(None);
};
deserialize::<Value>(&versioned.entry.body).map(Some)
}
}
// ── Paths ──────────────────────────────────────────────────────
//
// Every path returned here is alias-relative to the `/processes` mount
// alias on the caller's [`ScopedFilesystem`]. The leading tenant/user
// segment that the legacy implementation hand-formatted into the path
// is gone: the MountView's `/processes -> /tenants/<tenant>/users/<user>/processes`
// grant supplies it at every op. Sub-scope axes (agent/project/mission/
// thread) remain in the alias-relative path because they are *within*-
// tenant scoping and are not covered by the per-tenant MountAlias.
const PROCESSES_PREFIX: &str = "/processes";
fn process_record_path(
scope: &ResourceScope,
process_id: ProcessId,
) -> Result<VirtualPath, ProcessError> {
VirtualPath::new(format!(
) -> Result<ScopedPath, ProcessError> {
scoped_path(&format!(
"{}/{process_id}.json",
process_records_root(scope)?.as_str()
process_records_root_string(scope)
))
.map_err(invalid_path)
}
fn process_records_root(scope: &ResourceScope) -> Result<VirtualPath, ProcessError> {
VirtualPath::new(format!("{}/processes", resource_owner_root(scope))).map_err(invalid_path)
fn process_records_root(scope: &ResourceScope) -> Result<ScopedPath, ProcessError> {
scoped_path(&process_records_root_string(scope))
}
fn process_records_root_string(scope: &ResourceScope) -> String {
format!("{}/records", scope_owner_root_string(scope))
}
fn process_result_path(
scope: &ResourceScope,
process_id: ProcessId,
) -> Result<VirtualPath, ProcessError> {
VirtualPath::new(format!(
"{}/{process_id}.json",
process_results_root(scope)?.as_str()
) -> Result<ScopedPath, ProcessError> {
scoped_path(&format!(
"{}/results/{process_id}.json",
scope_owner_root_string(scope)
))
.map_err(invalid_path)
}
fn process_results_root(scope: &ResourceScope) -> Result<VirtualPath, ProcessError> {
VirtualPath::new(format!("{}/process-results", resource_owner_root(scope)))
.map_err(invalid_path)
}
fn process_output_path(
scope: &ResourceScope,
process_id: ProcessId,
) -> Result<VirtualPath, ProcessError> {
VirtualPath::new(format!(
"{}/output.json",
process_outputs_root(scope, process_id)?.as_str()
) -> Result<ScopedPath, ProcessError> {
scoped_path(&format!(
"{}/outputs/{process_id}/output.json",
scope_owner_root_string(scope)
))
.map_err(invalid_path)
}
fn process_outputs_root(
scope: &ResourceScope,
process_id: ProcessId,
) -> Result<VirtualPath, ProcessError> {
VirtualPath::new(format!(
"{}/process-outputs/{process_id}",
resource_owner_root(scope)
))
.map_err(invalid_path)
}
fn resource_owner_root(scope: &ResourceScope) -> String {
let mut base = format!(
"/engine/tenants/{}/users/{}",
scope.tenant_id.as_str(),
scope.user_id.as_str()
);
/// Build the alias-relative prefix for a given sub-scope under
/// `/processes`. The tenant/user prefix is supplied by the caller's
/// MountView at op time and intentionally absent here.
fn scope_owner_root_string(scope: &ResourceScope) -> String {
let mut base = String::from(PROCESSES_PREFIX);
if let Some(agent_id) = &scope.agent_id {
base = format!("{base}/agents/{}", agent_id.as_str());
base.push_str("/agents/");
base.push_str(agent_id.as_str());
}
if let Some(project_id) = &scope.project_id {
base = format!("{base}/projects/{}", project_id.as_str());
base.push_str("/projects/");
base.push_str(project_id.as_str());
}
if let Some(mission_id) = &scope.mission_id {
base = format!("{base}/missions/{}", mission_id.as_str());
base.push_str("/missions/");
base.push_str(mission_id.as_str());
}
if let Some(thread_id) = &scope.thread_id {
base = format!("{base}/threads/{}", thread_id.as_str());
base.push_str("/threads/");
base.push_str(thread_id.as_str());
}
base
}
fn scoped_path(raw: &str) -> Result<ScopedPath, ProcessError> {
ScopedPath::new(raw).map_err(invalid_path)
}
/// Join a leaf segment onto a [`ScopedPath`] prefix. Used when
/// reconstructing a child path after `list_dir` (which returns
/// [`VirtualPath`]s) so the per-op ACL check still runs on the follow-up
/// `get` — mirrors the engine store's `join_scoped` helper.
fn join_scoped(prefix: &ScopedPath, leaf: &str) -> Result<ScopedPath, ProcessError> {
let joined = format!("{}/{}", prefix.as_str().trim_end_matches('/'), leaf);
ScopedPath::new(joined).map_err(invalid_path)
}
fn ensure_process_record_matches(
record: &ProcessRecord,
process_id: ProcessId,
@@ -734,19 +779,21 @@ fn process_status_label(status: ProcessStatus) -> &'static str {
/// `put` with a fallback to an opaque (byte-only) entry on `Unsupported`.
///
/// Backends that don't yet implement records (LocalFilesystem with no
/// sidecar metadata) reject `kind = Some(_)` or any non-empty
/// `indexed` projection. We try the indexed write first so SQL and
/// in-memory backends get the projection, then retry with the same body
/// stripped of metadata so the legacy byte-only path keeps working
/// during the consumer migration.
/// sidecar metadata) reject `kind = Some(_)` or any non-`Any` CAS
/// expectation. We try the indexed write first so SQL and in-memory
/// backends get the projection, then retry with the same body stripped
/// of metadata and the CAS downgraded to `Any` so the legacy byte-only
/// path keeps working during the consumer migration. The single-instance
/// `transition_lock` on the caller carries the ordering safety
/// invariant that CAS would otherwise provide.
async fn put_with_byte_fallback<F>(
filesystem: &F,
path: &VirtualPath,
filesystem: &ScopedFilesystem<F>,
path: &ScopedPath,
entry: Entry,
cas: CasExpectation,
) -> Result<(), FilesystemError>
where
F: RootFilesystem + ?Sized,
F: RootFilesystem,
{
match filesystem.put(path, entry.clone(), cas).await {
Ok(_) => Ok(()),
@@ -754,9 +801,7 @@ where
// Byte-only backends (LocalFilesystem) reject BOTH record-shaped
// entries AND non-`Any` CAS in a single `Unsupported` response.
// Strip the record metadata and downgrade the CAS expectation to
// `Any` so the legacy byte-only path stays writeable. The
// single-instance `transition_lock` on the caller carries the
// ordering safety invariant that CAS would otherwise provide.
// `Any` so the legacy byte-only path stays writeable.
let opaque = Entry::bytes(entry.body).with_content_type(entry.content_type);
filesystem
.put(path, opaque, CasExpectation::Any)
@@ -772,13 +817,13 @@ where
/// `ensure_exact_index` shape so backends without index support degrade
/// to the list+get fallback path instead of failing closed.
async fn ensure_exact_index<F>(
filesystem: &F,
prefix: &VirtualPath,
filesystem: &ScopedFilesystem<F>,
prefix: &ScopedPath,
name: IndexName,
key: IndexKey,
) -> Result<(), ProcessError>
where
F: RootFilesystem + ?Sized,
F: RootFilesystem,
{
let spec = IndexSpec::new(name, vec![key], IndexKind::Exact);
match filesystem.ensure_index(prefix, &spec).await {
@@ -791,12 +836,12 @@ where
/// Drain a paginated `query` against `prefix` with `filter`, materializing
/// every matched [`ProcessRecord`].
async fn query_all_records<F>(
filesystem: &F,
prefix: &VirtualPath,
filesystem: &ScopedFilesystem<F>,
prefix: &ScopedPath,
filter: &Filter,
) -> Result<Vec<ProcessRecord>, FilesystemError>
where
F: RootFilesystem + ?Sized,
F: RootFilesystem,
{
let mut out = Vec::new();
let mut offset: u64 = 0;

7
crates/ironclaw_processes/src/services.rs
View File

@@ -17,7 +17,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use futures::FutureExt;
use ironclaw_events::sanitize_error_kind;
use ironclaw_filesystem::RootFilesystem;
use ironclaw_filesystem::{RootFilesystem, ScopedFilesystem};
use ironclaw_host_api::{ProcessId, ResourceReservation, ResourceScope};
use crate::cancellation::ProcessCancellationRegistry;
@@ -147,12 +147,11 @@ impl ProcessServices<InMemoryProcessStore, InMemoryProcessResultStore> {
}
}
impl<F>
ProcessServices<FilesystemProcessStore<'static, F>, FilesystemProcessResultStore<'static, F>>
impl<F> ProcessServices<FilesystemProcessStore<F>, FilesystemProcessResultStore<F>>
where
F: RootFilesystem + 'static,
{
pub fn filesystem(filesystem: Arc<F>) -> Self {
pub fn filesystem(filesystem: Arc<ScopedFilesystem<F>>) -> Self {
Self::new(
Arc::new(FilesystemProcessStore::from_arc(Arc::clone(&filesystem))),
Arc::new(FilesystemProcessResultStore::from_arc(filesystem)),

24
crates/ironclaw_processes/tests/process_services_contract.rs
View File

@@ -7,7 +7,7 @@ use std::{
};
use async_trait::async_trait;
use ironclaw_filesystem::LocalFilesystem;
use ironclaw_filesystem::{LocalFilesystem, RootFilesystem, ScopedFilesystem};
use ironclaw_host_api::*;
use ironclaw_processes::*;
use serde_json::json;
@@ -80,7 +80,7 @@ async fn process_services_share_cancellation_registry_between_host_and_manager()
#[tokio::test]
async fn filesystem_process_services_store_output_refs() {
let services = ProcessServices::filesystem(Arc::new(engine_filesystem()));
let services = ProcessServices::filesystem(engine_filesystem());
let manager = services.background_manager(Arc::new(SuccessExecutor));
let invocation_id = InvocationId::new();
let process_id = ProcessId::new();
@@ -251,7 +251,7 @@ fn process_start(
}
}
fn engine_filesystem() -> LocalFilesystem {
fn engine_filesystem() -> Arc<ScopedFilesystem<LocalFilesystem>> {
let storage = tempfile::tempdir().unwrap().keep();
let mut fs = LocalFilesystem::new();
fs.mount_local(
@@ -259,7 +259,23 @@ fn engine_filesystem() -> LocalFilesystem {
HostPath::from_path_buf(storage),
)
.unwrap();
fs
scoped_processes_filesystem(
Arc::new(fs),
"/engine/tenants/tenant1/users/user1/processes",
)
}
fn scoped_processes_filesystem<F>(backend: Arc<F>, target_root: &str) -> Arc<ScopedFilesystem<F>>
where
F: RootFilesystem,
{
let mounts = MountView::new(vec![MountGrant::new(
MountAlias::new("/processes").expect("alias"),
VirtualPath::new(target_root).expect("target"),
MountPermissions::read_write_list_delete(),
)])
.expect("mount view");
Arc::new(ScopedFilesystem::new(backend, mounts))
}
fn sample_scope(invocation_id: InvocationId, tenant: &str, user: &str) -> ResourceScope {

295
crates/ironclaw_processes/tests/process_store_contract.rs
View File

@@ -11,7 +11,7 @@ use async_trait::async_trait;
use ironclaw_events::{InMemoryEventSink, RuntimeEventKind};
use ironclaw_filesystem::{
DirEntry, FileStat, FilesystemError, FilesystemOperation, InMemoryBackend, LocalFilesystem,
RootFilesystem,
RootFilesystem, ScopedFilesystem,
};
use ironclaw_host_api::*;
use ironclaw_processes::*;
@@ -493,7 +493,7 @@ async fn process_host_kill_does_not_cancel_other_tenant_process() {
#[tokio::test]
async fn background_process_manager_can_use_owned_filesystem_store() {
let filesystem = Arc::new(engine_filesystem());
let filesystem = engine_filesystem();
let store = Arc::new(FilesystemProcessStore::from_arc(filesystem));
let executor = Arc::new(CountingExecutor::success());
let manager = BackgroundProcessManager::new(store.clone(), executor);
@@ -512,7 +512,7 @@ async fn background_process_manager_can_use_owned_filesystem_store() {
#[tokio::test]
async fn filesystem_process_store_rejects_terminal_status_overwrite() {
let fs = engine_filesystem();
let store = FilesystemProcessStore::new(&fs);
let store = FilesystemProcessStore::new(Arc::clone(&fs));
let invocation_id = InvocationId::new();
let process_id = ProcessId::new();
let scope = sample_scope(invocation_id, "tenant1", "user1");
@@ -1318,7 +1318,7 @@ async fn process_result_lookup_is_resource_scope_scoped() {
#[tokio::test]
async fn filesystem_process_result_store_persists_under_resource_scope() {
let fs = engine_filesystem();
let store = FilesystemProcessResultStore::new(&fs);
let store = FilesystemProcessResultStore::new(Arc::clone(&fs));
let invocation_id = InvocationId::new();
let process_id = ProcessId::new();
let scope = sample_scope(invocation_id, "tenant1", "user1");
@@ -1330,7 +1330,7 @@ async fn filesystem_process_result_store_persists_under_resource_scope() {
.await
.unwrap();
let reloaded = FilesystemProcessResultStore::new(&fs)
let reloaded = FilesystemProcessResultStore::new(Arc::clone(&fs))
.get(&scope, process_id)
.await
.unwrap()
@@ -1339,45 +1339,38 @@ async fn filesystem_process_result_store_persists_under_resource_scope() {
assert_eq!(reloaded.output, None);
assert_eq!(
reloaded.output_ref,
Some(
VirtualPath::new(format!(
"{}/process-outputs/{}/output.json",
stored_process_owner_root(&scope),
process_id
))
.unwrap()
)
Some(stored_process_output_path(&scope, process_id)),
);
assert_eq!(
FilesystemProcessResultStore::new(&fs)
FilesystemProcessResultStore::new(Arc::clone(&fs))
.output(&scope, process_id)
.await
.unwrap(),
Some(serde_json::json!({"ok": true}))
);
assert!(
FilesystemProcessResultStore::new(&fs)
FilesystemProcessResultStore::new(Arc::clone(&fs))
.get(&other_scope, process_id)
.await
.unwrap()
.is_none()
);
assert!(
FilesystemProcessResultStore::new(&fs)
FilesystemProcessResultStore::new(Arc::clone(&fs))
.output(&other_scope, process_id)
.await
.unwrap()
.is_none()
);
assert!(
FilesystemProcessResultStore::new(&fs)
FilesystemProcessResultStore::new(Arc::clone(&fs))
.get(&other_project, process_id)
.await
.unwrap()
.is_none()
);
assert!(
FilesystemProcessResultStore::new(&fs)
FilesystemProcessResultStore::new(Arc::clone(&fs))
.output(&other_project, process_id)
.await
.unwrap()
@@ -1387,7 +1380,7 @@ async fn filesystem_process_result_store_persists_under_resource_scope() {
#[tokio::test]
async fn background_process_manager_stores_filesystem_output_ref() {
let fs = Arc::new(engine_filesystem());
let fs = engine_filesystem();
let store = Arc::new(InMemoryProcessStore::new());
let result_store = Arc::new(FilesystemProcessResultStore::from_arc(fs));
let manager =
@@ -1417,8 +1410,11 @@ async fn background_process_manager_stores_filesystem_output_ref() {
#[tokio::test]
async fn filesystem_process_store_propagates_backend_errors_that_mention_not_found() {
let fs = BackendErrorFilesystem;
let store = FilesystemProcessStore::new(&fs);
let fs = scoped_processes_filesystem(
Arc::new(BackendErrorFilesystem),
&default_mount_target_string(),
);
let store = FilesystemProcessStore::new(fs);
let invocation_id = InvocationId::new();
let process_id = ProcessId::new();
let scope = sample_scope(invocation_id, "tenant1", "user1");
@@ -1434,7 +1430,7 @@ async fn filesystem_process_store_propagates_backend_errors_that_mention_not_fou
#[tokio::test]
async fn filesystem_process_store_rejects_record_id_mismatches() {
let fs = engine_filesystem();
let store = FilesystemProcessStore::new(&fs);
let store = FilesystemProcessStore::new(Arc::clone(&fs));
let invocation_id = InvocationId::new();
let requested_process_id = ProcessId::new();
let stored_process_id = ProcessId::new();
@@ -1442,9 +1438,13 @@ async fn filesystem_process_store_rejects_record_id_mismatches() {
let mut forged = process_record(stored_process_id, invocation_id, scope.clone());
forged.status = ProcessStatus::Completed;
fs.write_file(
&stored_process_record_path(&scope, requested_process_id),
&serde_json::to_vec_pretty(&forged).unwrap(),
// Inject a forged record at the alias-relative ScopedPath for
// `requested_process_id`. Going through the scoped filesystem (vs.
// a raw `write_file`) keeps the test honest about the actual on-disk
// surface the store will read from.
fs.write_bytes(
&scoped_record_path(&scope, requested_process_id),
serde_json::to_vec_pretty(&forged).unwrap(),
)
.await
.unwrap();
@@ -1457,7 +1457,7 @@ async fn filesystem_process_store_rejects_record_id_mismatches() {
#[tokio::test]
async fn filesystem_process_result_store_rejects_unexpected_output_refs() {
let fs = engine_filesystem();
let store = FilesystemProcessResultStore::new(&fs);
let store = FilesystemProcessResultStore::new(Arc::clone(&fs));
let owner_invocation_id = InvocationId::new();
let owner_process_id = ProcessId::new();
let owner_scope = sample_scope(owner_invocation_id, "tenant1", "user1");
@@ -1473,6 +1473,9 @@ async fn filesystem_process_result_store_rejects_unexpected_output_refs() {
)
.await
.unwrap();
// Forged record whose `output_ref` points at a *different* on-disk
// location than the owner's expected output path. The store must
// reject the read instead of dereferencing the forged ref.
let forged = ProcessResultRecord {
process_id: owner_process_id,
scope: owner_scope.clone(),
@@ -1481,9 +1484,9 @@ async fn filesystem_process_result_store_rejects_unexpected_output_refs() {
output_ref: Some(stored_process_output_path(&other_scope, other_process_id)),
error_kind: None,
};
fs.write_file(
&stored_process_result_path(&owner_scope, owner_process_id),
&serde_json::to_vec_pretty(&forged).unwrap(),
fs.write_bytes(
&scoped_result_path(&owner_scope, owner_process_id),
serde_json::to_vec_pretty(&forged).unwrap(),
)
.await
.unwrap();
@@ -1499,7 +1502,7 @@ async fn filesystem_process_result_store_rejects_unexpected_output_refs() {
#[tokio::test]
async fn filesystem_process_store_persists_under_resource_scope_engine_processes() {
let fs = engine_filesystem();
let store = FilesystemProcessStore::new(&fs);
let store = FilesystemProcessStore::new(Arc::clone(&fs));
let invocation_id = InvocationId::new();
let process_id = ProcessId::new();
let scope = sample_scope(invocation_id, "tenant1", "user1");
@@ -1510,14 +1513,14 @@ async fn filesystem_process_store_persists_under_resource_scope_engine_processes
.unwrap();
store.complete(&scope, process_id).await.unwrap();
let reloaded = FilesystemProcessStore::new(&fs)
let reloaded = FilesystemProcessStore::new(Arc::clone(&fs))
.get(&scope, process_id)
.await
.unwrap()
.unwrap();
assert_eq!(reloaded.status, ProcessStatus::Completed);
assert_eq!(
FilesystemProcessStore::new(&fs)
FilesystemProcessStore::new(Arc::clone(&fs))
.records_for_scope(&scope)
.await
.unwrap()
@@ -1530,21 +1533,23 @@ async fn filesystem_process_store_persists_under_resource_scope_engine_processes
async fn filesystem_process_store_records_for_scope_uses_index_on_record_backend() {
// Drive `records_for_scope` against the in-memory backend (which
// supports `query` over indexed projections) so the indexed path
// exercised by SQL backends is covered. Asserts that records from
// a different tenant or user under the same `processes/` root are
// not returned, and that the result matches the legacy list+get
// fallback path used on byte-only backends.
// exercised by SQL backends is covered. With the ScopedFilesystem
// refactor, tenant/user isolation lives in the MountView (not the
// path), so this test focuses on the *within-tenant* sub-scope
// discrimination that does still live in the path: separate
// `project_id` cells must not bleed into each other through the
// indexed query path, even though they share one `/processes` mount.
let backend = Arc::new(InMemoryBackend::new());
let store = FilesystemProcessStore::from_arc(Arc::clone(&backend));
let fs = scoped_processes_filesystem(Arc::clone(&backend), &default_mount_target_string());
let store = FilesystemProcessStore::from_arc(fs);
let invocation_id = InvocationId::new();
let scope = sample_scope(invocation_id, "tenant1", "user1");
let other_user_scope = sample_scope(invocation_id, "tenant1", "user2");
let other_tenant_scope = sample_scope(invocation_id, "tenant2", "user1");
let other_project_scope =
sample_scope_with_project(invocation_id, "tenant1", "user1", "project2");
let mine_a = ProcessId::new();
let mine_b = ProcessId::new();
let other_user = ProcessId::new();
let other_tenant = ProcessId::new();
let other_project = ProcessId::new();
store
.start(process_start(mine_a, invocation_id, scope.clone()))
.await
@@ -1555,17 +1560,9 @@ async fn filesystem_process_store_records_for_scope_uses_index_on_record_backend
.unwrap();
store
.start(process_start(
other_user,
other_project,
invocation_id,
other_user_scope.clone(),
))
.await
.unwrap();
store
.start(process_start(
other_tenant,
invocation_id,
other_tenant_scope.clone(),
other_project_scope.clone(),
))
.await
.unwrap();
@@ -1577,13 +1574,79 @@ async fn filesystem_process_store_records_for_scope_uses_index_on_record_backend
expected.sort_by_key(|id| id.as_uuid());
assert_eq!(got, expected);
let theirs = store.records_for_scope(&other_user_scope).await.unwrap();
let theirs = store.records_for_scope(&other_project_scope).await.unwrap();
assert_eq!(theirs.len(), 1);
assert_eq!(theirs[0].process_id, other_user);
assert_eq!(theirs[0].process_id, other_project);
}
let cross_tenant = store.records_for_scope(&other_tenant_scope).await.unwrap();
assert_eq!(cross_tenant.len(), 1);
assert_eq!(cross_tenant[0].process_id, other_tenant);
/// Regression test for the tenant-isolation invariant: two
/// `FilesystemProcessStore`s sharing one backend but constructed with
/// different `MountView`s (i.e. different tenant/user mount targets)
/// must not see each other's records, even though their request scopes
/// share `user_id` / `project_id` / `agent_id` and the alias-relative
/// path is identical.
///
/// Before the migration to `Arc<ScopedFilesystem<F>>`, the store
/// hand-formatted `tenant_id`/`user_id` into the path string — so any
/// composition layer that forgot to do that (or did it differently in
/// one place) would silently share storage across tenants. With the
/// ScopedFilesystem refactor, the MountView resolves the leading
/// segment, and the type system makes it impossible for the store to
/// reach across mounts.
#[tokio::test]
async fn filesystem_process_store_isolates_two_tenants_with_same_user_project_ids() {
let backend = Arc::new(InMemoryBackend::new());
let store_a = FilesystemProcessStore::from_arc(scoped_processes_filesystem(
Arc::clone(&backend),
"/engine/tenants/a/users/alice/processes",
));
let store_b = FilesystemProcessStore::from_arc(scoped_processes_filesystem(
Arc::clone(&backend),
"/engine/tenants/b/users/alice/processes",
));
// Identical scope across both stores — the only thing separating them
// is the mount-time tenant prefix on each store's MountView.
let invocation_id = InvocationId::new();
let process_id = ProcessId::new();
let scope = sample_scope(invocation_id, "tenant-a", "alice");
store_a
.start(process_start(process_id, invocation_id, scope.clone()))
.await
.unwrap();
// Tenant A sees its own record.
assert!(
store_a
.get(&scope, process_id)
.await
.expect("store_a get succeeds")
.is_some(),
"tenant A must see the record it just wrote",
);
// Tenant B does NOT see tenant A's record, despite the identical
// request scope and process id.
assert!(
store_b
.get(&scope, process_id)
.await
.expect("store_b get succeeds")
.is_none(),
"tenant B must NOT see tenant A's record (cross-tenant leak)",
);
// Tenant B's records_for_scope must be empty under the shared
// request scope.
let b_records = store_b
.records_for_scope(&scope)
.await
.expect("store_b records_for_scope succeeds");
assert!(
b_records.is_empty(),
"tenant B records_for_scope must be empty under shared scope; got {} records",
b_records.len(),
);
}
enum UnownedTransition {
@@ -2359,36 +2422,44 @@ fn process_estimate() -> ResourceEstimate {
}
}
fn stored_process_record_path(scope: &ResourceScope, process_id: ProcessId) -> VirtualPath {
VirtualPath::new(format!(
"{}/processes/{process_id}.json",
stored_process_owner_root(scope)
))
.unwrap()
}
// ── Test path layout ───────────────────────────────────────────
//
// After the FilesystemProcessStore refactor onto `ScopedFilesystem`, the
// on-disk path layout is alias-relative: `/processes/...` is the alias
// and the caller's `MountView` resolves the leading segment to a
// tenant/user-scoped target. The test fixtures below construct a
// canonical MountView pointing the `/processes` alias at
// `/engine/tenants/<tenant>/users/<user>/processes` so existing tests
// that drive the store across multiple scope objects (different
// `tenant_id`/`user_id`) still exercise the post-read
// `same_scope_owner` check — even though the on-disk record lives at
// the *mount's* tenant/user, not the request scope's.
fn stored_process_result_path(scope: &ResourceScope, process_id: ProcessId) -> VirtualPath {
VirtualPath::new(format!(
"{}/process-results/{process_id}.json",
stored_process_owner_root(scope)
))
.unwrap()
/// Canonical `/processes` mount target for the default test scope
/// (`tenant1` / `user1`). Tests that drive cross-tenant filtering via
/// the in-record `scope` field rely on this default — see
/// [`engine_filesystem`].
const DEFAULT_TEST_MOUNT_TENANT: &str = "tenant1";
const DEFAULT_TEST_MOUNT_USER: &str = "user1";
fn default_mount_target_string() -> String {
format!("/engine/tenants/{DEFAULT_TEST_MOUNT_TENANT}/users/{DEFAULT_TEST_MOUNT_USER}/processes")
}
fn stored_process_output_path(scope: &ResourceScope, process_id: ProcessId) -> VirtualPath {
VirtualPath::new(format!(
"{}/process-outputs/{process_id}/output.json",
"{}/outputs/{process_id}/output.json",
stored_process_owner_root(scope)
))
.unwrap()
}
/// Resolved on-disk root for the default test mount. Tenant/user come
/// from the *mount* (always `tenant1`/`user1` in this fixture); the
/// scope's sub-axes (agent/project/mission/thread) come from the
/// request scope.
fn stored_process_owner_root(scope: &ResourceScope) -> String {
let mut base = format!(
"/engine/tenants/{}/users/{}",
scope.tenant_id.as_str(),
scope.user_id.as_str()
);
let mut base = default_mount_target_string();
if let Some(agent_id) = &scope.agent_id {
base = format!("{base}/agents/{}", agent_id.as_str());
}
@@ -2404,7 +2475,12 @@ fn stored_process_owner_root(scope: &ResourceScope) -> String {
base
}
fn engine_filesystem() -> LocalFilesystem {
/// Build a `Arc<ScopedFilesystem<LocalFilesystem>>` over a fresh tempdir
/// mounted at `/engine`, with the `/processes` alias resolving to the
/// default tenant1/user1 target. Tests that need a different mount
/// target (e.g. cross-tenant isolation tests) construct a
/// `ScopedFilesystem` directly with their own `MountView`.
fn engine_filesystem() -> Arc<ScopedFilesystem<LocalFilesystem>> {
let storage = tempfile::tempdir().unwrap().keep();
let mut fs = LocalFilesystem::new();
fs.mount_local(
@@ -2412,7 +2488,72 @@ fn engine_filesystem() -> LocalFilesystem {
HostPath::from_path_buf(storage),
)
.unwrap();
fs
let backend = Arc::new(fs);
scoped_processes_filesystem(backend, &default_mount_target_string())
}
/// Wrap a raw `RootFilesystem` backend in a `ScopedFilesystem` granting
/// full read/write/list/delete on the `/processes` alias mapped to
/// `target_root`. Used both by the default fixture above and by the
/// cross-tenant isolation regression tests below that need to wire two
/// different mount targets over one shared backend.
fn scoped_processes_filesystem<F>(backend: Arc<F>, target_root: &str) -> Arc<ScopedFilesystem<F>>
where
F: RootFilesystem,
{
let mounts = MountView::new(vec![MountGrant::new(
MountAlias::new("/processes").expect("alias"),
VirtualPath::new(target_root).expect("target"),
MountPermissions::read_write_list_delete(),
)])
.expect("mount view");
Arc::new(ScopedFilesystem::new(backend, mounts))
}
/// Alias-relative [`ScopedPath`] for a lifecycle record. Used by tests
/// that inject a forged record body via the scoped filesystem so the
/// production code path is still exercised on the read side.
fn scoped_record_path(scope: &ResourceScope, process_id: ProcessId) -> ScopedPath {
ScopedPath::new(format!(
"{}/records/{process_id}.json",
alias_relative_owner_root(scope)
))
.expect("scoped record path")
}
/// Alias-relative [`ScopedPath`] for a result record (sibling helper of
/// [`scoped_record_path`]).
fn scoped_result_path(scope: &ResourceScope, process_id: ProcessId) -> ScopedPath {
ScopedPath::new(format!(
"{}/results/{process_id}.json",
alias_relative_owner_root(scope)
))
.expect("scoped result path")
}
/// Build the alias-relative `/processes/...` owner prefix for a request
/// scope. Mirrors the production `scope_owner_root_string` in
/// `filesystem_store.rs` but lives in test code so a drift between
/// production and fixture path layouts shows up as a test failure.
fn alias_relative_owner_root(scope: &ResourceScope) -> String {
let mut base = String::from("/processes");
if let Some(agent_id) = &scope.agent_id {
base.push_str("/agents/");
base.push_str(agent_id.as_str());
}
if let Some(project_id) = &scope.project_id {
base.push_str("/projects/");
base.push_str(project_id.as_str());
}
if let Some(mission_id) = &scope.mission_id {
base.push_str("/missions/");
base.push_str(mission_id.as_str());
}
if let Some(thread_id) = &scope.thread_id {
base.push_str("/threads/");
base.push_str(thread_id.as_str());
}
base
}
fn sample_scope_with_agent(