diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index f48ea1b..e5e6407 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -10,7 +10,7 @@ permissions:
   contents: read
 
 jobs:
-  test:
+  unit-tests:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -20,5 +20,42 @@ jobs:
           pixi-version: latest
           environments: dev
 
-      - name: Run tests
+      - name: Run unit tests
         run: pixi run -e dev test
+
+  integration-tests:
+    runs-on: ubuntu-latest
+    services:
+      minio:
+        image: bitnamilegacy/minio:latest@sha256:b3d51900e846b92f7503ca6be07d2e8c56ebb6a13a60bc71b8777c716c074bcf
+        ports:
+          - 9000:9000
+        env:
+          MINIO_ROOT_USER: minioadmin
+          MINIO_ROOT_PASSWORD: minioadmin
+          MINIO_DEFAULT_BUCKETS: josh-test-bucket:public
+          MINIO_SCHEME: http
+        options: >-
+          --health-cmd "curl -f http://localhost:9000/minio/health/ready || curl -f http://localhost:9000/minio/health/live"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - uses: prefix-dev/setup-pixi@a0af7a228712d6121d37aba47adf55c1332c9c2e # v0.9.4
+        with:
+          pixi-version: latest
+          environments: dev
+
+      - name: Verify MinIO is ready
+        run: curl -f http://localhost:9000/minio/health/ready
+
+      - name: Download Josh JAR
+        run: pixi run get-jars
+
+      - name: Run integration tests
+        run: pixi run -e dev test-integration
diff --git a/BATCH_INTEGRATION.md b/BATCH_INTEGRATION.md
new file mode 100644
index 0000000..b38fdb0
--- /dev/null
+++ b/BATCH_INTEGRATION.md
@@ -0,0 +1,344 @@
+# Plan: Batch Remote Execution for joshpy
+
+Tracking issue: [joshpy#31](https://github.com/SchmidtDSE/joshpy/issues/31)
+Companion Java plan: [josh#374](https://github.com/SchmidtDSE/josh/issues/374)
+Dependency: [josh#406](https://github.com/SchmidtDSE/josh/issues/406) — `pollBatch` CLI command for async job status polling
+
+## Context
+
+joshsim (Java) has added `batchRemote` — a parallel execution path using MinIO staging and target profiles instead of HTTP streaming. PRs 1-7 are merged on the Java side ([josh#374](https://github.com/SchmidtDSE/josh/issues/374)). joshpy needs to wrap these new capabilities and provide efficient Python-level orchestration for parameter sweeps.
+
+**Immediate motivation:** A production run has 5 of 6 replicate CSVs sitting in MinIO (the 6th OOM'd). The run is registered in the local RunRegistry with a label. We need a way to recover those results NOW — look up the run by label, discover the `minio://` export paths, read the CSVs directly into DuckDB via S3, and load them into the registry. This drives the PR ordering: result ingestion first, then the rest of the batch infrastructure.
+
+**Access model (Model A):** MinIO/S3 CSVs are the source of truth. The local `.duckdb` is a materialized cache that any machine can rebuild by re-ingesting from S3. DuckDB reads CSVs directly from S3 via `httpfs` — no download, no local disk needed for the CSV data. This supports future access patterns: browser WASM reading S3, serverless aggregators attaching `.duckdb`, multi-machine access.
+
+**State ownership:** josh is stateless/ephemeral — it dispatches jobs and can check their status, but holds no long-running state. joshpy owns all state via RunRegistry (what was run, parameters, label, job ID). When joshpy dispatches a `--no-wait` batch job, it stores the `batch_job_id` in `job_runs.metadata`. To poll, joshpy calls josh's `pollBatch` CLI command ([josh#406](https://github.com/SchmidtDSE/josh/issues/406)) which knows HOW to check status for each target type (MinIO status file, K8s Job API, etc.). joshpy doesn't know or care about the polling mechanism internals — it just gets back "running" / "complete" / "error".
+
+**Key design decisions:**
+- `batchRemote` has no `--data` flags — files stage to/from MinIO. First positional arg can be a `.josh` file OR a directory. The caller stages data; the worker pulls via `stageFromMinio`.
+- Auto-pull results from MinIO after jobs complete, with opt-out for fire-and-forget. Plus a generic "ingest CSVs after the fact" code path that works for both batch remote AND local OOM recovery (DRY).
+- Target config system is SHARED between josh and joshpy — joshpy reads AND creates `~/.josh/targets/<name>.json`.
+- MinIO cred resolution hierarchy (mirrors joshsim's `HierarchyConfig`): CLI flags > profile JSON > env vars (`MINIO_ENDPOINT`, `MINIO_ACCESS_KEY`, `MINIO_SECRET_KEY`, `MINIO_BUCKET`). Secrets don't need to live in profile JSON.
+- K8s targets have a separate `pod_minio_endpoint` — the in-cluster MinIO endpoint pods use, which may differ from the outer `minio_endpoint` used for host-side staging.
+- For sweeps: stage shared data (.josh, .jshd) to MinIO ONCE, then per-job stage only the unique .jshc config. joshpy orchestrates staging directly (not via `batchRemote`).
+- Dev JARs are outdated — implement against spec, test when updated.
+
+---
+
+## PR Plan
+
+```
+PR1 (S3-native ingest) → PR2 (target profiles) → PR3 (CLI wrappers) → PR4 (sweep integration) → PR5 (shared staging optimization) → PR6 (polish)
+```
+
+### Regression gates (every PR)
+- `pixi run pytest` passes
+- Existing `runRemote` path completely untouched
+
+---
+
+### PR 1: Result Recovery — S3-native `ingest_results()`
+
+**Solves the immediate need.** Enables recovering results from MinIO into the registry by label. DuckDB reads CSVs directly from S3 via httpfs — no download, no local disk needed for the CSV data. Also provides `download=True` fallback via `stageFromMinio` for users who want local copies.
+
+#### New utility: `configure_s3()` in `joshpy/registry.py` (or `joshpy/s3.py`)
+
+Reusable DuckDB S3/MinIO connection setup — the foundation for all future S3 access (serverless aggregators, WASM, multi-machine):
+
+```python
+def configure_s3(conn, endpoint: str, access_key: str, secret_key: str, url_style: str = "path") -> None:
+    """Configure DuckDB connection for S3/MinIO access via httpfs."""
+    conn.execute("INSTALL httpfs; LOAD httpfs;")
+    conn.execute(f"""
+        CREATE OR REPLACE SECRET (
+            TYPE s3,
+            KEY_ID '{access_key}',
+            SECRET '{secret_key}',
+            ENDPOINT '{endpoint}',
+            URL_STYLE '{url_style}',
+            USE_SSL true
+        )
+    """)
+```
+
+S3 credentials resolve via hierarchy: explicit args > env vars (`MINIO_ENDPOINT`, `MINIO_ACCESS_KEY`, `MINIO_SECRET_KEY`). The function takes explicit args; the caller (`ingest_results`) handles the env var fallback.
+
+#### Modify `CellDataLoader.load_csv()` in `joshpy/cell_data.py`
+
+Accept `str` (S3 URL) in addition to `Path`:
+```python
+def load_csv(self, csv_path: Path | str, run_id: str, run_hash: str, ...) -> int:
+    if isinstance(csv_path, str) and csv_path.startswith("s3://"):
+        csv_path_str = csv_path  # S3 URL — pass directly to read_csv_auto
+    else:
+        csv_path_str = str(Path(csv_path).resolve())  # local path (existing behavior)
+    # ... rest unchanged — read_csv_auto handles both
+```
+
+The existing `read_csv_auto()` call works with S3 URLs natively once httpfs is loaded.
+
+#### New function: `ingest_results()` in `joshpy/sweep.py`
+
+The core recovery function. Works by label (not by `JobSet`):
+
+```python
+def ingest_results(
+    cli: JoshCLI,
+    registry: RunRegistry,
+    label_or_hash: str,
+    *,
+    export_type: str = "patch",
+    download: bool = False,           # if True, download via stageFromMinio instead of S3 read
+    output_dir: Path | None = None,   # download destination (only used when download=True)
+    minio_bucket: str | None = None,  # override bucket (else from ExportFileInfo.host)
+    quiet: bool = False,
+) -> int:
+```
+
+**Flow:**
+1. `registry._resolve_label_or_hash(label_or_hash)` -> `run_hash`
+2. `registry.get_config_by_hash(run_hash)` -> `ConfigInfo` (josh_path, josh_content, parameters, label)
+3. `registry.get_session(config.session_id)` -> `SessionInfo` (simulation, total_replicates)
+4. Get josh source on disk: if `config.josh_path` exists use it; otherwise write `config.josh_content` to a temp file
+5. `cli.inspect_exports(script, simulation)` -> `ExportPaths`
+6. Get `ExportFileInfo` for `export_type` -> check `info.protocol`
+7. If `protocol == "minio"` and NOT `download`:
+   - Configure S3 on registry connection: `configure_s3(registry.conn, endpoint, access_key, secret_key)`
+   - Translate `minio://bucket/path` to `s3://bucket/path` for DuckDB
+8. If `protocol == "minio"` and `download`:
+   - Call `cli.stage_from_minio(...)` to download locally (fallback path)
+9. `registry._resolve_run_id_for_hash(run_hash)` -> `run_id` for the latest execution
+10. For each replicate 0..`total_replicates-1`:
+    - Build template vars: `{simulation, replicate, **config.parameters, label: config.label}`
+    - Resolve path template -> concrete path
+    - If minio (no download): remap to `s3://bucket/resolved_path`
+    - If minio (download): remap to `output_dir / filename`
+    - Load via `CellDataLoader.load_csv(csv_path_or_url, run_id, run_hash)`
+    - If file/object doesn't exist -> skip gracefully (the OOM'd replicate), print which one
+11. Return total rows loaded
+
+#### Also in this PR: `StageFromMinioConfig` + `stage_from_minio()` for `download=True` path
+
+```python
+@dataclass(frozen=True)
+class StageFromMinioConfig:
+    output_dir: Path
+    prefix: str
+    minio_endpoint: str | None = None
+    minio_access_key: str | None = None
+    minio_secret_key: str | None = None
+    minio_bucket: str | None = None
+```
+
+Plus `JoshCLI.stage_from_minio()` method wrapping `stageFromMinio --output-dir=... --prefix=... [--minio-* options]`.
+
+#### New method: `SweepManager.ingest()` in `joshpy/sweep.py`
+
+```python
+def ingest(self, export_type="patch", download=False, output_dir=None, quiet=False) -> int:
+    label = getattr(self, '_label', None) or self.job_set.jobs[0].run_hash
+    return ingest_results(self.cli, self.registry, label, export_type=export_type,
+                          download=download, output_dir=output_dir, quiet=quiet)
+```
+
+#### Exports: `joshpy/__init__.py`
+- Add `StageFromMinioConfig`, `configure_s3` to CLI exports
+- Add `ingest_results` to sweep exports
+
+#### Tests
+- `tests/test_cli.py`: `StageFromMinioConfig` defaults, `stage_from_minio()` arg building
+- `tests/test_sweep.py`: `ingest_results` with mocked registry + mocked DuckDB (S3 URL construction, download fallback, missing replicate skip, josh_content temp file fallback)
+
+#### User-facing example (pixi task in josh-models)
+
+```toml
+recover = { cmd = "python scripts/recover.py", env = { JOSH_LABEL = "{{ LABEL }}" }, args = [{ arg = "LABEL" }], description = "Recover results from MinIO: pixi run recover <LABEL>." }
+```
+
+```python
+# scripts/recover.py
+import os
+from dotenv import load_dotenv
+load_dotenv()
+
+from joshpy.cli import JoshCLI
+from joshpy.jar import JarMode
+from joshpy.registry import RunRegistry
+from joshpy.sweep import ingest_results
+
+registry = RunRegistry(os.environ["JOSH_REGISTRY"])
+cli = JoshCLI(josh_jar=JarMode[os.environ.get("JOSH_JAR_MODE", "DEV")])
+
+rows = ingest_results(cli, registry, os.environ["JOSH_LABEL"])
+print(f"Done: {rows} rows loaded for '{os.environ['JOSH_LABEL']}'")
+registry.close()
+```
+
+**Polling:** PR 1 does NOT poll — `ingest_results()` assumes the job is already done (called after blocking `batchRemote`, or manually by user for recovery). It reads whatever CSVs exist in S3 and skips missing ones. Async polling comes in PR 4/5 via josh's `pollBatch` CLI ([josh#406](https://github.com/SchmidtDSE/josh/issues/406)).
+
+**`batch_job_id` in registry:** When batch remote jobs are dispatched, the job ID is stored in `job_runs.metadata` as `{"batch_job_id": "...", "target": "..."}`. This field is optional — absent for local runs and blocking batch runs where the ID isn't needed. `ingest_results()` does not require it; it works by label/run_hash alone.
+
+**Risk: LOW — additive. Existing load_csv() local path behavior unchanged. httpfs is opt-in (only configured when minio:// detected).**
+
+---
+
+### PR 2: Target Profile System
+
+New file `joshpy/targets.py`. joshpy reads AND writes `~/.josh/targets/<name>.json` — shared config between josh and joshpy.
+
+**Dataclasses:** `TargetProfile`, `HttpTargetConfig`, `KubernetesTargetConfig` (mirrors joshsim JSON structure).
+
+**JSON serialization:** Python snake_case <-> JSON camelCase where joshsim expects it (`api_key` -> `apiKey`, `timeout_seconds` -> `timeoutSeconds`).
+
+**Functions:**
+- `load_target(name)` / `save_target(name, profile)` — read/write `~/.josh/targets/<name>.json`
+- `list_targets()` / `delete_target(name)` — manage profiles
+- `resolve_minio_creds(target=None)` — hierarchy: profile JSON -> env vars
+
+**K8s note:** `pod_minio_endpoint` in `KubernetesTargetConfig` — the in-cluster MinIO endpoint pods use, distinct from outer `minio_endpoint`.
+
+**Tests:** `tests/test_targets.py` — round-trip, hierarchy, validation, auto-create dirs.
+
+**Risk: LOW — all new files, no modifications to existing code.**
+
+---
+
+### PR 3: Remaining CLI Wrappers — `batch_remote()`, `stage_to_minio()`
+
+`stage_from_minio()` already shipped in PR 1. This adds the remaining two.
+
+**`BatchRemoteConfig`:**
+```python
+@dataclass(frozen=True)
+class BatchRemoteConfig:
+    script_or_dir: Path    # .josh file or directory
+    simulation: str
+    target: str            # required — profile name
+    replicates: int = 1
+    no_wait: bool = False
+    poll_interval: int | None = None
+    timeout: int | None = None
+```
+
+**`StageToMinioConfig`:** `input_dir`, `prefix`, optional `minio_*` creds.
+
+**Methods:** `JoshCLI.batch_remote()`, `JoshCLI.stage_to_minio()`.
+
+**Tests:** `tests/test_cli.py` — mock subprocess, verify arg building.
+
+**Risk: LOW — additive, follows existing `run_remote()` pattern exactly.**
+
+---
+
+### PR 4: Sweep Integration — `run_sweep()` + `SweepManager` + adaptive
+
+Wires batch remote into the sweep loop. Two modes:
+
+**Blocking mode (default, `batch_no_wait=False`):** Each job calls `batchRemote` without `--no-wait`. The subprocess blocks until josh finishes polling internally. joshpy gets exit code, records in registry, then calls `ingest_results()` to read CSVs from S3. Sequential but simple — same pattern as existing `run_remote()`.
+
+**Async mode (`batch_no_wait=True`):** Each job calls `batchRemote --no-wait`, gets back a job ID. joshpy stores `batch_job_id` in `job_runs.metadata`, then dispatches the next job. After all jobs are dispatched, joshpy polls via `cli.poll_batch(job_id, target)` ([josh#406](https://github.com/SchmidtDSE/josh/issues/406)) until all complete. Then ingests results. This is the path to parallel runs on big-memory machines.
+
+**New CLI wrapper (depends on [josh#406](https://github.com/SchmidtDSE/josh/issues/406)):**
+```python
+@dataclass(frozen=True)
+class PollBatchConfig:
+    job_id: str
+    target: str
+
+def poll_batch(self, config: PollBatchConfig, timeout: float | None = None) -> CLIResult:
+    # calls: java -jar joshsim.jar pollBatch <jobId> --target=<name>
+    # exit code: 0=complete, 1=error, 2=running
+```
+
+**New parameters on `run_sweep()`:** `batch_remote`, `target`, `poll_interval`, `batch_timeout`, `batch_no_wait`, `auto_pull`.
+
+**New functions in `joshpy/jobs.py`:**
+- `assemble_batch_workdir(job, workdir)` — creates per-job dir with symlinked shared files + written .jshc
+- `to_batch_remote_config(job, target, workdir)` — converts `ExpandedJob` to `BatchRemoteConfig`
+
+**Validation:** `batch_remote` and `remote` mutually exclusive; `target` required when `batch_remote=True`.
+
+**Extends:** `SweepManager.run()`, `run_adaptive_sweep()` with same parameters.
+
+**Tests:** `tests/test_jobs.py`, `tests/test_sweep.py`, `tests/test_strategies.py`.
+
+**Risk: LOW — mostly new code. Small modifications to existing function signatures (additive parameters). Async mode depends on josh#406.**
+
+---
+
+### PR 5: Shared Staging Optimization for Sweeps
+
+Stage shared data (.josh, .jshd) to MinIO ONCE, per-job stage only the unique .jshc config. Avoids re-uploading GBs of .jshd per job.
+
+New file `joshpy/batch_orchestrator.py` with `BatchOrchestrator`:
+- `stage_shared(jobs)` — stage shared files once
+- `dispatch_job(job, shared_prefix)` — stage per-job config + dispatch via HTTP POST to `/runBatch`
+- `poll(job_id)` / `pull_results(job_id, output_dir)`
+
+**Dispatch approach:** HTTP POST to `/runBatch` directly from Python (~20 lines with `urllib.request`). Self-contained, no joshsim changes needed, `/runBatch` endpoint already exists.
+
+**Risk: MEDIUM — introduces direct HTTP dispatch from joshpy. Well-isolated in new file.**
+
+---
+
+### PR 6: Polish — Builder, Docs, Bottle Metadata
+
+- `SweepManagerBuilder.with_batch_remote(target, ...)` convenience method
+- MinIO metadata in bottle manifest
+- Update `llms-full.txt` with all new APIs
+
+**Risk: LOW**
+
+---
+
+## Files Modified (all PRs)
+
+| File | PRs | Changes |
+|------|-----|---------|
+| `joshpy/cli.py` | 1, 3, 4 | `StageFromMinioConfig` + `stage_from_minio()` (PR 1); `BatchRemoteConfig` + `batch_remote()`, `StageToMinioConfig` + `stage_to_minio()` (PR 3); `PollBatchConfig` + `poll_batch()` (PR 4) |
+| `joshpy/cell_data.py` | 1 | `load_csv()` accepts `str` (S3 URL) in addition to `Path` |
+| `joshpy/registry.py` | 1 | `configure_s3()` utility for DuckDB httpfs + S3 credential setup |
+| `joshpy/sweep.py` | 1, 4, 6 | `ingest_results()` + `SweepManager.ingest()` (PR 1); extend `.run()` (PR 4); builder (PR 6) |
+| **NEW** `joshpy/targets.py` | 2 | Target profile system (read/write/list/creds hierarchy) |
+| `joshpy/jobs.py` | 4 | `assemble_batch_workdir`, `to_batch_remote_config`, extend `run_sweep()` |
+| `joshpy/strategies.py` | 4 | Extend `run_adaptive_sweep()` |
+| **NEW** `joshpy/batch_orchestrator.py` | 5 | Shared staging orchestration |
+| `joshpy/bottle.py` | 6 | MinIO metadata in manifest |
+| `joshpy/__init__.py` | 1-3 | Export new symbols |
+| `tests/test_cli.py` | 1, 3 | `StageFromMinio` tests (PR 1); remaining CLI tests (PR 3) |
+| `tests/test_sweep.py` | 1, 4 | `ingest_results` tests (PR 1); SweepManager batch_remote tests (PR 4) |
+| **NEW** `tests/test_targets.py` | 2 | Target profile tests |
+| `tests/test_jobs.py` | 4 | Workdir, converter, sweep tests |
+| `tests/test_strategies.py` | 4 | Adaptive batch remote tests |
+
+---
+
+## Verification
+
+PR 1 end-to-end (immediate need):
+```bash
+# In josh-models repo:
+pixi run recover my-label
+# -> Looks up "my-label" in registry
+# -> Discovers minio:// export paths via inspect-exports
+# -> Configures DuckDB httpfs with S3 creds from env vars
+# -> Reads CSVs directly from S3 into DuckDB (no download)
+# -> Loads into registry, skipping missing replicate from OOM
+# -> Prints: "Done: 1234567 rows loaded for 'my-label'"
+```
+
+Full integration (when dev JARs update):
+```python
+# Sweep with batch remote
+manager = SweepManager.from_config(config, registry="exp.duckdb")
+results = manager.run(batch_remote=True, target="my-server")
+manager.load_results()
+
+# Fire-and-forget -> recover later
+results = manager.run(batch_remote=True, target="my-server", batch_no_wait=True)
+# ... later ...
+manager.ingest()
+
+# Or download locally
+manager.ingest(download=True, output_dir=Path("./local_results"))
+```
diff --git a/joshpy/__init__.py b/joshpy/__init__.py
index 50270ae..ccff0d0 100644
--- a/joshpy/__init__.py
+++ b/joshpy/__init__.py
@@ -37,6 +37,7 @@
     InspectExportsConfig,
     ExportFileInfo,
     ExportPaths,
+    StageFromMinioConfig,
 )
 
 # JFR diagnostics (always available, no external deps)
@@ -119,6 +120,7 @@
         RunInfo,
         SessionSummary,
         DataSummary,
+        configure_s3,
     )
     from joshpy.cell_data import (
         CellDataLoader,
@@ -136,6 +138,7 @@
         SweepManagerBuilder,
         recover_sweep_results,
         load_job_results,
+        ingest_results,
         LoadConfig,
         ResultLoadError,
     )
@@ -182,6 +185,7 @@
     "InspectExportsConfig",
     "ExportFileInfo",
     "ExportPaths",
+    "StageFromMinioConfig",
     # JFR diagnostics
     "ResourceProfile",
     "CpuProfile",
@@ -243,6 +247,7 @@
     "RunInfo",
     "SessionSummary",
     "DataSummary",
+    "configure_s3",
     "CellDataLoader",
     "DiagnosticQueries",
     "SimulationDiagnostics",
@@ -252,6 +257,7 @@
     "SweepManagerBuilder",
     "recover_sweep_results",
     "load_job_results",
+    "ingest_results",
     "LoadConfig",
     "ResultLoadError",
     "HAS_SWEEP",
diff --git a/joshpy/cell_data.py b/joshpy/cell_data.py
index 7b70b59..ee02ba8 100644
--- a/joshpy/cell_data.py
+++ b/joshpy/cell_data.py
@@ -148,7 +148,7 @@ def __init__(self, registry: Any):
 
     def load_csv(
         self,
-        csv_path: Path,
+        csv_path: "Path | str",
         run_id: str,
         run_hash: str,
         entity_type: str = "patch",
@@ -166,10 +166,12 @@ def load_csv(
         quoted identifiers (e.g., 'avg.height' stays as "avg.height"), requiring
         double quotes when referenced with direct calls to DuckDB.
 
-        Uses DuckDB's native CSV reader for optimal performance.
+        Uses DuckDB's native CSV reader for optimal performance.  Accepts both
+        local ``Path`` objects and ``s3://`` URL strings (requires httpfs to be
+        loaded on the connection -- see ``configure_s3()``).
 
         Args:
-            csv_path: Path to the CSV file.
+            csv_path: Path to the CSV file, or an ``s3://`` URL string.
             run_id: The run ID this data belongs to.
             run_hash: Run hash for this run.
             entity_type: Type of entity being exported (default: "patch").
@@ -178,14 +180,18 @@ def load_csv(
             Number of rows loaded.
 
         Raises:
-            FileNotFoundError: If csv_path doesn't exist.
+            FileNotFoundError: If csv_path is a local path that doesn't exist.
             ValueError: If CSV is missing required columns or type mismatch.
         """
-        if not csv_path.exists():
-            raise FileNotFoundError(f"CSV not found: {csv_path}")
+        if isinstance(csv_path, str) and csv_path.startswith("s3://"):
+            csv_path_str = csv_path
+        else:
+            csv_path = Path(csv_path)
+            if not csv_path.exists():
+                raise FileNotFoundError(f"CSV not found: {csv_path}")
+            csv_path_str = str(csv_path.resolve())
 
         conn = self.registry.conn
-        csv_path_str = str(csv_path.resolve())
 
         # Read CSV header to identify columns using DuckDB
         header_result = conn.execute(f"SELECT * FROM read_csv_auto('{csv_path_str}') LIMIT 0")
diff --git a/joshpy/cli.py b/joshpy/cli.py
index 2debb12..f0fbc8d 100644
--- a/joshpy/cli.py
+++ b/joshpy/cli.py
@@ -398,6 +398,32 @@ class InspectJshdConfig:
     y: int
 
 
+@dataclass(frozen=True)
+class StageFromMinioConfig:
+    """Arguments for 'java -jar joshsim.jar stageFromMinio' command.
+
+    Downloads all objects under a MinIO prefix to a local directory.
+    MinIO credentials are optional -- joshsim falls back to environment
+    variables (MINIO_ENDPOINT, MINIO_ACCESS_KEY, MINIO_SECRET_KEY,
+    MINIO_BUCKET) via its HierarchyConfig.
+
+    Attributes:
+        output_dir: Local directory to download files into.
+        prefix: MinIO object prefix to download from.
+        minio_endpoint: MinIO endpoint URL (optional).
+        minio_access_key: MinIO access key (optional).
+        minio_secret_key: MinIO secret key (optional).
+        minio_bucket: MinIO bucket name (optional).
+    """
+
+    output_dir: Path
+    prefix: str
+    minio_endpoint: str | None = None
+    minio_access_key: str | None = None
+    minio_secret_key: str | None = None
+    minio_bucket: str | None = None
+
+
 @dataclass(frozen=True)
 class InspectExportsConfig:
     """Arguments for 'java -jar joshsim.jar inspect-exports' command.
@@ -624,6 +650,37 @@ def _execute(
                 command=cmd,
             )
 
+    def stage_from_minio(
+        self,
+        config: StageFromMinioConfig,
+        timeout: float | None = None,
+    ) -> CLIResult:
+        """Download files from MinIO to a local directory.
+
+        Args:
+            config: Stage-from-MinIO configuration.
+            timeout: Timeout in seconds.
+
+        Returns:
+            CLIResult with execution details.
+        """
+        args = [
+            "stageFromMinio",
+            "--output-dir", str(config.output_dir.resolve()),
+            "--prefix", config.prefix,
+        ]
+
+        if config.minio_endpoint:
+            args.extend(["--minio-endpoint", config.minio_endpoint])
+        if config.minio_access_key:
+            args.extend(["--minio-access-key", config.minio_access_key])
+        if config.minio_secret_key:
+            args.extend(["--minio-secret-key", config.minio_secret_key])
+        if config.minio_bucket:
+            args.extend(["--minio-bucket", config.minio_bucket])
+
+        return self._execute(args, timeout=timeout)
+
     def _execute_streaming(
         self,
         cmd: list[str],
diff --git a/joshpy/registry.py b/joshpy/registry.py
index cda9f87..cc50594 100644
--- a/joshpy/registry.py
+++ b/joshpy/registry.py
@@ -72,6 +72,48 @@ def _check_duckdb() -> None:
         )
 
 
+def configure_s3(
+    conn: Any,
+    endpoint: str,
+    access_key: str,
+    secret_key: str,
+    url_style: str = "path",
+    use_ssl: bool = True,
+) -> None:
+    """Configure a DuckDB connection for S3/MinIO access via httpfs.
+
+    Installs and loads the httpfs extension, then creates an S3 secret
+    so ``read_csv_auto('s3://bucket/key.csv')`` works transparently.
+
+    Credential resolution is the caller's responsibility -- this function
+    takes explicit values.  ``ingest_results()`` resolves credentials from
+    environment variables (``MINIO_ENDPOINT``, ``MINIO_ACCESS_KEY``,
+    ``MINIO_SECRET_KEY``) before calling here.
+
+    Args:
+        conn: DuckDB connection object.
+        endpoint: S3-compatible endpoint (e.g. ``"storage.googleapis.com"``).
+        access_key: Access key / key ID.
+        secret_key: Secret key.
+        url_style: ``"path"`` (default, MinIO) or ``"vhost"`` (AWS).
+        use_ssl: Use HTTPS (default True).
+    """
+    conn.execute("INSTALL httpfs; LOAD httpfs;")
+    conn.execute(
+        """
+        CREATE OR REPLACE SECRET (
+            TYPE s3,
+            KEY_ID ?,
+            SECRET ?,
+            ENDPOINT ?,
+            URL_STYLE ?,
+            USE_SSL ?
+        )
+        """,
+        [access_key, secret_key, endpoint, url_style, use_ssl],
+    )
+
+
 def _get_git_hash() -> str | None:
     """Get current git HEAD hash, or None if not in a git repo."""
     try:
diff --git a/joshpy/sweep.py b/joshpy/sweep.py
index 6d7eba7..3f9c992 100644
--- a/joshpy/sweep.py
+++ b/joshpy/sweep.py
@@ -37,6 +37,8 @@
 
 from __future__ import annotations
 
+import os
+import tempfile
 import time
 from collections.abc import Callable
 from dataclasses import dataclass, field
@@ -46,7 +48,7 @@
 import pandas as pd
 
 from joshpy.cell_data import CellDataLoader, DiagnosticQueries
-from joshpy.cli import ExportPaths, InspectExportsConfig, JoshCLI
+from joshpy.cli import ExportPaths, InspectExportsConfig, JoshCLI, StageFromMinioConfig
 from joshpy.jobs import (
     ExpandedJob,
     JobConfig,
@@ -55,7 +57,7 @@
     SweepResult,
     run_sweep,
 )
-from joshpy.registry import RunRegistry
+from joshpy.registry import RunRegistry, configure_s3
 
 
 @dataclass
@@ -421,6 +423,316 @@ def _get_export_path(export_paths: ExportPaths, export_type: str) -> str | None:
         raise ValueError(f"Unknown export_type: {export_type}. Use 'patch', 'meta', or 'entity'.")
 
 
+@dataclass
+class _IngestMetadata:
+    """Resolved metadata for an ingest operation."""
+
+    run_hash: str
+    config: Any
+    simulation: str
+    total_replicates: int
+    label: str | None
+
+
+def _resolve_ingest_metadata(
+    registry: RunRegistry,
+    label_or_hash: str,
+    *,
+    quiet: bool = False,
+) -> _IngestMetadata:
+    """Resolve a label or hash to the run metadata needed for ingestion."""
+    run_hash = registry._resolve_label_or_hash(label_or_hash)
+    config = registry.get_config_by_hash(run_hash)
+    if config is None:
+        raise KeyError(f"No config found for run hash: {run_hash}")
+
+    session = registry.get_session(config.session_id)
+    if session is None:
+        raise KeyError(f"No session found: {config.session_id}")
+
+    simulation = session.simulation
+
+    # Determine replicate count: session metadata > job_runs count > fallback to 1
+    total_replicates = session.total_replicates
+    if not total_replicates:
+        job_config = session.job_config
+        if job_config is not None:
+            total_replicates = getattr(job_config, "replicates", None)
+    if not total_replicates:
+        runs = registry.get_runs_for_hash(run_hash)
+        total_replicates = len(runs) if runs else 1
+
+    if not quiet:
+        label_str = f" ({config.label})" if config.label else ""
+        print(f"Ingesting results for {run_hash}{label_str}")
+        print(f"  Simulation: {simulation}, Replicates: {total_replicates}")
+
+    return _IngestMetadata(
+        run_hash=run_hash,
+        config=config,
+        simulation=simulation,
+        total_replicates=total_replicates,
+        label=config.label,
+    )
+
+
+def _get_josh_source(config: Any, run_hash: str) -> tuple[Path, str | None]:
+    """Get josh source file on disk, creating a temp file if needed.
+
+    Returns:
+        ``(josh_path, temp_file_path_or_None)``.  Caller must clean up
+        the temp file when non-None.
+    """
+    if config.josh_path and Path(config.josh_path).exists():
+        return Path(config.josh_path), None
+
+    if config.josh_content:
+        fd, temp_path = tempfile.mkstemp(suffix=".josh")
+        os.close(fd)
+        Path(temp_path).write_text(config.josh_content)
+        return Path(temp_path), temp_path
+
+    raise RuntimeError(
+        f"Cannot inspect exports: no josh source available for {run_hash}. "
+        "Neither josh_path exists on disk nor josh_content stored in registry."
+    )
+
+
+def _configure_minio_access(
+    cli: JoshCLI,
+    registry: RunRegistry,
+    export_info: Any,
+    path_template: str,
+    *,
+    download: bool,
+    output_dir: Path | None,
+    minio_bucket: str | None,
+    quiet: bool,
+) -> tuple[str, Path | None]:
+    """Configure S3 direct read or download from MinIO.
+
+    Returns:
+        ``(bucket_name, download_dir_or_None)``.
+    """
+    bucket = minio_bucket or export_info.host
+
+    if not download:
+        endpoint = os.environ.get("MINIO_ENDPOINT", "")
+        access_key = os.environ.get("MINIO_ACCESS_KEY", "")
+        secret_key = os.environ.get("MINIO_SECRET_KEY", "")
+
+        if not endpoint or not access_key or not secret_key:
+            raise RuntimeError(
+                "MINIO_ENDPOINT, MINIO_ACCESS_KEY, and MINIO_SECRET_KEY "
+                "environment variables are required for S3 reads."
+            )
+
+        configure_s3(registry.conn, endpoint, access_key, secret_key)
+
+        if not quiet:
+            print(f"  Reading directly from S3 (bucket: {bucket})")
+
+        return bucket, None
+
+    # download=True: stage files locally via stageFromMinio
+    prefix = str(Path(path_template).parent).lstrip("/")
+    if prefix and not prefix.endswith("/"):
+        prefix += "/"
+
+    dl_dir = Path(output_dir) if output_dir else Path(tempfile.mkdtemp(prefix="joshpy-ingest-"))
+    dl_dir.mkdir(parents=True, exist_ok=True)
+
+    if not quiet:
+        print(f"  Downloading from minio://{bucket}/{prefix} to {dl_dir}")
+
+    stage_result = cli.stage_from_minio(
+        StageFromMinioConfig(
+            output_dir=dl_dir,
+            prefix=prefix,
+            minio_bucket=bucket,
+        )
+    )
+    if not stage_result.success:
+        raise RuntimeError(
+            f"stageFromMinio failed (exit {stage_result.exit_code}): "
+            f"{stage_result.stderr}"
+        )
+
+    return bucket, dl_dir
+
+
+def _load_ingest_replicates(
+    registry: RunRegistry,
+    export_paths: ExportPaths,
+    path_template: str,
+    *,
+    is_minio: bool,
+    download: bool,
+    bucket: str,
+    dl_dir: Path | None,
+    meta: _IngestMetadata,
+    run_id: str,
+    export_type: str,
+    quiet: bool,
+) -> int:
+    """Load CSVs for each replicate into the registry."""
+    loader = CellDataLoader(registry)
+    total_rows = 0
+    loaded = 0
+    skipped = 0
+
+    template_vars_base: dict[str, Any] = {"simulation": meta.simulation}
+    if meta.config.parameters:
+        template_vars_base.update(meta.config.parameters)
+    if meta.label:
+        template_vars_base["label"] = meta.label
+
+    for rep in range(meta.total_replicates):
+        template_vars = {**template_vars_base, "replicate": rep}
+
+        try:
+            resolved = export_paths.resolve_path(path_template, **template_vars)
+        except KeyError:
+            if not quiet:
+                print(f"  Replicate {rep}: template variable missing, skipping")
+            skipped += 1
+            continue
+
+        # Determine the actual path/URL to load
+        if is_minio and not download:
+            resolved_path = resolved.as_posix().lstrip("/")
+            csv_target: Path | str = f"s3://{bucket}/{resolved_path}"
+        elif is_minio and download:
+            csv_target = dl_dir / resolved.name
+        else:
+            csv_target = resolved
+
+        # Try loading -- skip gracefully if missing
+        try:
+            rows = loader.load_csv(
+                csv_path=csv_target,
+                run_id=run_id,
+                run_hash=meta.run_hash,
+                entity_type=export_type,
+            )
+            total_rows += rows
+            loaded += 1
+            if not quiet:
+                print(f"  Replicate {rep}: {rows:,} rows loaded")
+        except FileNotFoundError:
+            skipped += 1
+            if not quiet:
+                print(f"  Replicate {rep}: not found, skipping")
+        except Exception as e:
+            skipped += 1
+            if not quiet:
+                err_str = str(e)
+                # DuckDB raises IOException for missing S3 objects
+                if "HTTP 404" in err_str or "NoSuchKey" in err_str:
+                    print(f"  Replicate {rep}: not found in S3, skipping")
+                else:
+                    print(f"  Replicate {rep}: error loading: {e}")
+
+    if not quiet:
+        print(f"\nDone: {total_rows:,} rows loaded ({loaded} replicates, {skipped} skipped)")
+
+    return total_rows
+
+
+def ingest_results(
+    cli: JoshCLI,
+    registry: RunRegistry,
+    label_or_hash: str,
+    *,
+    export_type: str = "patch",
+    download: bool = False,
+    output_dir: Path | None = None,
+    minio_bucket: str | None = None,
+    quiet: bool = False,
+) -> int:
+    """Recover and ingest results into the registry by label or run hash.
+
+    Looks up the run in the registry, discovers export paths via
+    ``inspect_exports``, and loads CSVs into the ``cell_data`` table.
+
+    For ``minio://`` export paths the default behaviour reads CSVs directly
+    from S3 into DuckDB via ``httpfs`` (no local download).  Set
+    ``download=True`` to download via ``stageFromMinio`` first.
+
+    Missing CSVs (e.g. from an OOM'd replicate) are skipped gracefully.
+
+    Args:
+        cli: JoshCLI instance.
+        registry: RunRegistry where results will be loaded.
+        label_or_hash: Human-readable label or 12-char run hash.
+        export_type: Type of export to load (``"patch"``, ``"meta"``, ``"entity"``).
+        download: If True, download CSVs locally via ``stageFromMinio``
+            instead of reading directly from S3.
+        output_dir: Local directory for downloads (temp dir if None).
+            Only used when ``download=True``.
+        minio_bucket: Override the MinIO bucket (default: parsed from
+            the ``minio://`` export path).
+        quiet: Suppress progress output.
+
+    Returns:
+        Total number of rows loaded.
+
+    Raises:
+        KeyError: If label/hash not found in registry.
+        RuntimeError: If no export path configured for *export_type*, or
+            if ``inspect_exports`` fails.
+
+    Examples:
+        >>> # Recover results for a labeled run (reads from S3)
+        >>> rows = ingest_results(cli, registry, "my-label")
+
+        >>> # Download locally first, then load
+        >>> rows = ingest_results(cli, registry, "my-label", download=True)
+    """
+    meta = _resolve_ingest_metadata(registry, label_or_hash, quiet=quiet)
+    josh_path, temp_josh = _get_josh_source(meta.config, meta.run_hash)
+
+    try:
+        export_paths = cli.inspect_exports(
+            InspectExportsConfig(script=josh_path, simulation=meta.simulation)
+        )
+
+        export_info = export_paths.export_files.get(export_type)
+        if export_info is None:
+            raise RuntimeError(
+                f"No {export_type} export configured in {josh_path}. "
+                f"Check that exportFiles.{export_type} is set in your simulation."
+            )
+
+        path_template = export_info.path
+        is_minio = export_info.protocol == "minio"
+
+        if not quiet:
+            proto = f"minio://{export_info.host}" if is_minio else "local"
+            print(f"  Export path: {proto}{path_template}")
+
+        bucket: str = ""
+        dl_dir: Path | None = None
+        if is_minio:
+            bucket, dl_dir = _configure_minio_access(
+                cli, registry, export_info, path_template,
+                download=download, output_dir=output_dir,
+                minio_bucket=minio_bucket, quiet=quiet,
+            )
+
+        run_id = registry._resolve_run_id_for_hash(meta.run_hash)
+
+        return _load_ingest_replicates(
+            registry, export_paths, path_template,
+            is_minio=is_minio, download=download, bucket=bucket,
+            dl_dir=dl_dir, meta=meta, run_id=run_id,
+            export_type=export_type, quiet=quiet,
+        )
+    finally:
+        if temp_josh:
+            Path(temp_josh).unlink(missing_ok=True)
+
+
 @dataclass
 class SweepManager:
     """Convenience orchestrator for parameter sweeps.
@@ -698,6 +1010,49 @@ def load_results(
             quiet=quiet,
         )
 
+    def ingest(
+        self,
+        *,
+        export_type: str = "patch",
+        download: bool = False,
+        output_dir: Path | None = None,
+        minio_bucket: str | None = None,
+        quiet: bool = False,
+    ) -> int:
+        """Recover and ingest results from MinIO (or local) by label.
+
+        Uses ``ingest_results()`` to look up the run by label, discover
+        export paths, and load CSVs into the registry.  Unlike
+        ``load_results()`` this does not require a prior ``run()`` call --
+        it works from the registry alone.
+
+        Args:
+            export_type: Type of export to load ("patch", "meta", "entity").
+            download: If True, download CSVs locally instead of S3 direct read.
+            output_dir: Download destination (only used with download=True).
+            minio_bucket: Override MinIO bucket name.
+            quiet: Suppress progress output.
+
+        Returns:
+            Total number of rows loaded.
+
+        Examples:
+            >>> manager.ingest()  # reads directly from S3
+            >>> manager.ingest(download=True, output_dir=Path("./local"))
+        """
+        label = self._label if hasattr(self, "_label") and self._label else None
+        identifier = label or self.job_set.jobs[0].run_hash
+        return ingest_results(
+            cli=self.cli,
+            registry=self.registry,
+            label_or_hash=identifier,
+            export_type=export_type,
+            download=download,
+            output_dir=output_dir,
+            minio_bucket=minio_bucket,
+            quiet=quiet,
+        )
+
     def query(
         self,
         variable: str,
diff --git a/pixi.toml b/pixi.toml
index 3c29df6..db36599 100644
--- a/pixi.toml
+++ b/pixi.toml
@@ -49,7 +49,8 @@ dev = { features = ["dev"], solve-group = "default" }
 [tasks]
 install = "pip install -e '.[full]' --quiet"
 install-dev = "pip install -e '.[dev]' --quiet"
-test = { cmd = "pytest tests/ -v", depends-on = ["install-dev"] }
+test = { cmd = "pytest tests/ -v -m 'not integration'", depends-on = ["install-dev"] }
+test-integration = { cmd = "pytest tests/ -v -m integration", depends-on = ["install-dev"] }
 lint = "ruff check joshpy/"
 typecheck = "mypy joshpy/"
 format = "ruff format joshpy/"
diff --git a/pyproject.toml b/pyproject.toml
index a06f90d..6bce7d1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -92,6 +92,9 @@ testpaths = ["tests"]
 python_files = ["test_*.py"]
 python_functions = ["test_*"]
 addopts = "-v --tb=short"
+markers = [
+    "integration: marks tests requiring external services (MinIO)",
+]
 
 [tool.mypy]
 python_version = "3.10"
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..3daf8d7
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,186 @@
+"""Shared fixtures and configuration for joshpy tests."""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Pytest marker registration
+# ---------------------------------------------------------------------------
+
+
+def pytest_configure(config):
+    config.addinivalue_line(
+        "markers",
+        "integration: marks tests requiring external services (MinIO)",
+    )
+
+
+# ---------------------------------------------------------------------------
+# MinIO integration test constants (bitnami test image defaults)
+# ---------------------------------------------------------------------------
+
+MINIO_ENDPOINT = "localhost:9000"
+MINIO_ACCESS_KEY = "minioadmin"
+MINIO_SECRET_KEY = "minioadmin"
+TEST_BUCKET = "josh-test-bucket"
+
+
+# ---------------------------------------------------------------------------
+# Session-scoped guards — skip the entire suite when infra is missing
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def minio_available():
+    """Skip if the MinIO test container is not reachable."""
+    import requests
+
+    try:
+        resp = requests.get(
+            f"http://{MINIO_ENDPOINT}/minio/health/ready", timeout=3
+        )
+        if resp.status_code != 200:
+            pytest.skip(f"MinIO not ready (HTTP {resp.status_code})")
+    except requests.ConnectionError:
+        pytest.skip("MinIO not available at localhost:9000")
+
+
+@pytest.fixture(scope="session")
+def jar_available():
+    """Skip if the Josh JAR has not been downloaded."""
+    from joshpy.jar import JarManager, JarMode
+
+    manager = JarManager()
+    try:
+        manager.get_jar(JarMode.DEV, auto_download=False)
+    except FileNotFoundError:
+        pytest.skip(
+            "Josh JAR not found — run `pixi run get-jars` first"
+        )
+
+
+# ---------------------------------------------------------------------------
+# Bucket name
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def test_bucket():
+    return TEST_BUCKET
+
+
+# ---------------------------------------------------------------------------
+# DuckDB connection with S3 configured for the test MinIO
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def minio_conn(minio_available):
+    """Fresh DuckDB connection with httpfs configured for test MinIO."""
+    import duckdb
+    from joshpy.registry import configure_s3
+
+    conn = duckdb.connect(":memory:")
+    configure_s3(
+        conn,
+        endpoint=MINIO_ENDPOINT,
+        access_key=MINIO_ACCESS_KEY,
+        secret_key=MINIO_SECRET_KEY,
+        use_ssl=False,
+    )
+    yield conn
+    conn.close()
+
+
+# ---------------------------------------------------------------------------
+# RunRegistry with S3 pre-configured
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def minio_registry(minio_available):
+    """In-memory RunRegistry whose DuckDB connection can read S3."""
+    from joshpy.registry import RunRegistry, configure_s3
+
+    registry = RunRegistry(":memory:")
+    configure_s3(
+        registry.conn,
+        endpoint=MINIO_ENDPOINT,
+        access_key=MINIO_ACCESS_KEY,
+        secret_key=MINIO_SECRET_KEY,
+        use_ssl=False,
+    )
+    yield registry
+    registry.close()
+
+
+# ---------------------------------------------------------------------------
+# CSV seeding helper — writes to MinIO via DuckDB COPY
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def seed_csv(minio_conn, test_bucket):
+    """Return a callable that writes CSV content to MinIO.
+
+    Usage::
+
+        url = seed_csv("level1/test.csv", "step,replicate,val\\n0,0,1.0\\n")
+    """
+    cleanup: list[str] = []
+
+    def _seed(key: str, csv_content: str) -> str:
+        s3_url = f"s3://{test_bucket}/{key}"
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".csv", delete=False
+        ) as f:
+            f.write(csv_content)
+            local_path = f.name
+        try:
+            minio_conn.execute(
+                f"COPY (SELECT * FROM read_csv_auto('{local_path}')) "
+                f"TO '{s3_url}' (FORMAT CSV, HEADER)"
+            )
+        finally:
+            Path(local_path).unlink(missing_ok=True)
+        cleanup.append(s3_url)
+        return s3_url
+
+    yield _seed
+
+
+# ---------------------------------------------------------------------------
+# Real JoshCLI (session-scoped — JAR doesn't change)
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def josh_cli(jar_available):
+    """JoshCLI backed by the real downloaded JAR."""
+    from joshpy.cli import JoshCLI
+    from joshpy.jar import JarMode
+
+    return JoshCLI(josh_jar=JarMode.DEV)
+
+
+# ---------------------------------------------------------------------------
+# Monkeypatch for configure_s3 → use_ssl=False
+# (needed by ingest_results which calls configure_s3 without use_ssl kwarg)
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def patch_s3_no_ssl(monkeypatch):
+    """Patch configure_s3 in the sweep module so it uses use_ssl=False."""
+    from joshpy.registry import configure_s3 as real_configure_s3
+
+    def _no_ssl(conn, endpoint, access_key, secret_key, **kwargs):
+        real_configure_s3(
+            conn, endpoint, access_key, secret_key, use_ssl=False
+        )
+
+    monkeypatch.setattr("joshpy.sweep.configure_s3", _no_ssl)
diff --git a/tests/fixtures/minio_export.josh b/tests/fixtures/minio_export.josh
new file mode 100644
index 0000000..6ae6c27
--- /dev/null
+++ b/tests/fixtures/minio_export.josh
@@ -0,0 +1,42 @@
+# Minimal simulation for MinIO integration tests.
+# Writes CSV results to minio://josh-test-bucket/results/output_{replicate}.csv
+# Tiny grid, 5 timesteps — completes in seconds.
+
+start simulation Main
+
+  grid.size = 1000 m
+  grid.low = 33.7 degrees latitude, -115.4 degrees longitude
+  grid.high = 34.0 degrees latitude, -116.4 degrees longitude
+  grid.patch = "Default"
+
+  steps.low = 0 count
+  steps.high = 5 count
+
+  exportFiles.patch = "minio://josh-test-bucket/results/output_{replicate}.csv"
+
+end simulation
+
+start patch Default
+
+  ForeverTree.init = create 5 count of ForeverTree
+
+  export.treeCount.step = count(ForeverTree)
+  export.averageHeight.step = mean(ForeverTree.height)
+
+end patch
+
+start organism ForeverTree
+
+  age.init = 0 year
+  age.step = prior.age + 1 year
+
+  height.init = 0 meters
+  height.step = prior.height + sample uniform from 0 meters to 1 meters
+
+end organism
+
+start unit year
+
+  alias years
+
+end unit
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 7fc84a7..4d9f3cb 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -446,9 +446,9 @@ def test_run_with_data_files(self, mock_run):
 
         cmd = mock_run.call_args[0][0]
         self.assertIn("--data", cmd)
-        # Find the data value
+        # Find the data value — name gets extension appended when missing
         data_idx = cmd.index("--data")
-        self.assertIn("editor=", cmd[data_idx + 1])
+        self.assertIn("editor.jshc=", cmd[data_idx + 1])
 
     @patch("subprocess.run")
     def test_run_with_custom_tags(self, mock_run):
@@ -1446,5 +1446,99 @@ def test_stream_output_run_remote(self, mock_popen):
         self.assertIn("remote step", result.stdout)
 
 
+class TestStageFromMinioConfig(unittest.TestCase):
+    """Tests for StageFromMinioConfig."""
+
+    def test_defaults(self):
+        from joshpy.cli import StageFromMinioConfig
+
+        config = StageFromMinioConfig(
+            output_dir=Path("/tmp/out"),
+            prefix="batch-jobs/abc/inputs/",
+        )
+        self.assertEqual(config.output_dir, Path("/tmp/out"))
+        self.assertEqual(config.prefix, "batch-jobs/abc/inputs/")
+        self.assertIsNone(config.minio_endpoint)
+        self.assertIsNone(config.minio_access_key)
+        self.assertIsNone(config.minio_secret_key)
+        self.assertIsNone(config.minio_bucket)
+
+    def test_frozen(self):
+        from joshpy.cli import StageFromMinioConfig
+
+        config = StageFromMinioConfig(output_dir=Path("/tmp"), prefix="p/")
+        with self.assertRaises(AttributeError):
+            config.prefix = "other/"
+
+
+class TestStageFromMinio(unittest.TestCase):
+    """Tests for JoshCLI.stage_from_minio()."""
+
+    JAR_MODE = JarMode.LOCAL
+
+    @patch("joshpy.jar.JarManager.get_jar", return_value=Path("/fake/joshsim-fat.jar"))
+    @patch("subprocess.run")
+    def test_basic_args(self, mock_run, _mock_jar):
+        """stage_from_minio() should build correct CLI args."""
+        from joshpy.cli import StageFromMinioConfig
+
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+
+        cli = JoshCLI(josh_jar=self.JAR_MODE)
+        config = StageFromMinioConfig(
+            output_dir=Path("/tmp/out"),
+            prefix="batch-jobs/abc/inputs/",
+        )
+        cli.stage_from_minio(config)
+
+        cmd = mock_run.call_args[0][0]
+        self.assertIn("stageFromMinio", cmd)
+        self.assertIn("--output-dir", cmd)
+        self.assertIn("--prefix", cmd)
+        prefix_idx = cmd.index("--prefix")
+        self.assertEqual(cmd[prefix_idx + 1], "batch-jobs/abc/inputs/")
+
+    @patch("joshpy.jar.JarManager.get_jar", return_value=Path("/fake/joshsim-fat.jar"))
+    @patch("subprocess.run")
+    def test_minio_flags_only_when_set(self, mock_run, _mock_jar):
+        """Only non-None minio flags should be passed."""
+        from joshpy.cli import StageFromMinioConfig
+
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+
+        cli = JoshCLI(josh_jar=self.JAR_MODE)
+
+        # With no minio flags
+        config_no_minio = StageFromMinioConfig(
+            output_dir=Path("/tmp/out"), prefix="p/"
+        )
+        cli.stage_from_minio(config_no_minio)
+        cmd = mock_run.call_args[0][0]
+        self.assertNotIn("--minio-endpoint", cmd)
+        self.assertNotIn("--minio-access-key", cmd)
+        self.assertNotIn("--minio-secret-key", cmd)
+        self.assertNotIn("--minio-bucket", cmd)
+
+        # With all minio flags
+        config_with_minio = StageFromMinioConfig(
+            output_dir=Path("/tmp/out"),
+            prefix="p/",
+            minio_endpoint="https://storage.example.com",
+            minio_access_key="AKID",
+            minio_secret_key="SECRET",
+            minio_bucket="my-bucket",
+        )
+        cli.stage_from_minio(config_with_minio)
+        cmd = mock_run.call_args[0][0]
+        self.assertIn("--minio-endpoint", cmd)
+        self.assertIn("--minio-access-key", cmd)
+        self.assertIn("--minio-secret-key", cmd)
+        self.assertIn("--minio-bucket", cmd)
+        ep_idx = cmd.index("--minio-endpoint")
+        self.assertEqual(cmd[ep_idx + 1], "https://storage.example.com")
+        bucket_idx = cmd.index("--minio-bucket")
+        self.assertEqual(cmd[bucket_idx + 1], "my-bucket")
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_diff.py b/tests/test_diff.py
index 967dfa3..8a2cd82 100644
--- a/tests/test_diff.py
+++ b/tests/test_diff.py
@@ -319,7 +319,8 @@ def test_main_view(self):
             file_registry.label_run("h1", "run_a")
             file_registry.close()
 
-            with patch("sys.argv", ["prog", str(db_path), "--view", "run_a"]):
+            with patch("sys.argv", ["prog", str(db_path), "--view", "run_a"]), \
+                 patch("joshpy.inspect._core._launch_ide"):
                 result = main()
 
             self.assertEqual(result, 0)
diff --git a/tests/test_minio_integration.py b/tests/test_minio_integration.py
new file mode 100644
index 0000000..6611337
--- /dev/null
+++ b/tests/test_minio_integration.py
@@ -0,0 +1,691 @@
+"""MinIO integration tests for joshpy.
+
+Escalating levels of integration testing against a real MinIO service:
+
+- Level 1: DuckDB writes CSV to MinIO and reads it back
+- Level 2: Josh JAR runs a simulation that exports to MinIO, Python reads it
+- Level 3: CellDataLoader.load_csv() ingests JAR output from S3 into registry
+- Level 4: End-to-end ingest_results() from MinIO by label
+- Level 5: Partial/interrupted sweep recovery from MinIO
+- Edge cases: bad creds, missing bucket, namespace isolation
+
+Requires:
+    - MinIO running at localhost:9000 (bitnamilegacy/minio with josh-test-bucket:public)
+    - Josh JAR downloaded (pixi run get-jars)
+
+Run with: pixi run -e dev test-integration
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from tests.conftest import (
+    MINIO_ACCESS_KEY,
+    MINIO_ENDPOINT,
+    MINIO_SECRET_KEY,
+    TEST_BUCKET,
+)
+
+# All tests in this file require MinIO
+pytestmark = pytest.mark.integration
+
+
+# ---------------------------------------------------------------------------
+# Test CSV data
+# ---------------------------------------------------------------------------
+
+SIMPLE_CSV = "step,replicate,position.x,position.y,treeCount,averageHeight\n"
+
+
+def _make_csv(replicate: int = 0, steps: int = 5, n_patches: int = 1) -> str:
+    """Generate a CSV matching Josh export format."""
+    lines = [SIMPLE_CSV.rstrip("\n")]
+    for step in range(steps):
+        for _ in range(n_patches):
+            lines.append(
+                f"{step},{replicate},0.0,0.0,{10 + step},{5.0 + step * 0.5}"
+            )
+    return "\n".join(lines) + "\n"
+
+
+# ===================================================================
+# Level 1: DuckDB httpfs writes to and reads from MinIO
+# ===================================================================
+
+
+class TestMinioWrite:
+    """Level 1: Prove DuckDB httpfs can write CSV to MinIO."""
+
+    def test_duckdb_copy_csv_to_s3(self, minio_conn, test_bucket):
+        """COPY ... TO 's3://...' should succeed without error."""
+        key = f"test-level1/{uuid.uuid4().hex[:8]}/write.csv"
+        s3_url = f"s3://{test_bucket}/{key}"
+
+        minio_conn.execute(
+            f"COPY (SELECT 1 as step, 0 as replicate, 42.0 as val) "
+            f"TO '{s3_url}' (FORMAT CSV, HEADER)"
+        )
+
+        # Verify by reading back
+        result = minio_conn.execute(
+            f"SELECT * FROM read_csv_auto('{s3_url}')"
+        ).fetchall()
+        assert len(result) == 1
+        assert result[0] == (1, 0, 42.0)
+
+    def test_write_then_read_roundtrip(self, seed_csv):
+        """seed_csv fixture writes CSV, read it back via DuckDB."""
+        csv_data = "a,b,c\n1,hello,3.14\n2,world,2.72\n"
+        key = f"test-level1/{uuid.uuid4().hex[:8]}/roundtrip.csv"
+        s3_url = seed_csv(key, csv_data)
+
+        import duckdb
+        from joshpy.registry import configure_s3
+
+        conn = duckdb.connect(":memory:")
+        configure_s3(
+            conn,
+            endpoint=MINIO_ENDPOINT,
+            access_key=MINIO_ACCESS_KEY,
+            secret_key=MINIO_SECRET_KEY,
+            use_ssl=False,
+        )
+        rows = conn.execute(
+            f"SELECT * FROM read_csv_auto('{s3_url}')"
+        ).fetchall()
+        conn.close()
+
+        assert len(rows) == 2
+        assert rows[0][1] == "hello"
+        assert rows[1][1] == "world"
+
+
+# ===================================================================
+# Level 2: Josh JAR writes to MinIO, Python reads
+# ===================================================================
+
+
+class TestMinioJarWrite:
+    """Level 2: Run a real simulation that exports to MinIO, verify from Python."""
+
+    SCRIPT = Path(__file__).parent / "fixtures" / "minio_export.josh"
+
+    @pytest.fixture(autouse=True, scope="class")
+    def _run_simulation(self, request, josh_cli, minio_available, jar_available):
+        """Run the test simulation once for the whole class."""
+        env_backup = {}
+        for k, v in {
+            "MINIO_ENDPOINT": f"http://{MINIO_ENDPOINT}",
+            "MINIO_ACCESS_KEY": MINIO_ACCESS_KEY,
+            "MINIO_SECRET_KEY": MINIO_SECRET_KEY,
+        }.items():
+            env_backup[k] = os.environ.get(k)
+            os.environ[k] = v
+
+        from joshpy.cli import RunConfig
+
+        result = josh_cli.run(
+            RunConfig(
+                script=self.SCRIPT,
+                simulation="Main",
+                replicates=2,
+                seed=42,
+            )
+        )
+
+        # Store result on the class for tests to inspect
+        request.cls.jar_result = result
+
+        yield
+
+        # Restore env
+        for k, orig in env_backup.items():
+            if orig is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = orig
+
+    def test_jar_run_succeeds(self):
+        """The Josh JAR should complete the simulation without error."""
+        assert self.jar_result.success, (
+            f"JAR failed (exit {self.jar_result.exit_code}): "
+            f"{self.jar_result.stderr}"
+        )
+
+    def test_jar_inspect_exports_minio(self, josh_cli):
+        """inspect_exports should parse the minio:// export path."""
+        from joshpy.cli import InspectExportsConfig
+
+        exports = josh_cli.inspect_exports(
+            InspectExportsConfig(script=self.SCRIPT, simulation="Main")
+        )
+        patch_info = exports.export_files["patch"]
+        assert patch_info is not None
+        assert patch_info.protocol == "minio"
+        assert patch_info.host == TEST_BUCKET
+        assert "{replicate}" in patch_info.path
+
+    def test_jar_output_readable_from_s3(self, minio_conn):
+        """CSV written by the JAR should be readable via DuckDB S3."""
+        s3_url = f"s3://{TEST_BUCKET}/results/output_0.csv"
+        rows = minio_conn.execute(
+            f"SELECT * FROM read_csv_auto('{s3_url}')"
+        ).fetchall()
+
+        assert len(rows) > 0
+
+        # Check expected columns exist
+        cols = [
+            desc[0]
+            for desc in minio_conn.execute(
+                f"SELECT * FROM read_csv_auto('{s3_url}') LIMIT 0"
+            ).description
+        ]
+        assert "step" in cols
+        assert "replicate" in cols
+        assert "treeCount" in cols
+        assert "averageHeight" in cols
+
+
+# ===================================================================
+# Level 3: CellDataLoader loads JAR output from S3
+# ===================================================================
+
+
+class TestMinioCellDataLoader:
+    """Level 3: CellDataLoader.load_csv with s3:// URL."""
+
+    def _setup_registry_for_load(self, registry):
+        """Register a minimal run so load_csv has a valid run_id."""
+        from joshpy.jobs import JobConfig
+
+        config = JobConfig(
+            source_path=Path("/tmp/sim.josh"),
+            simulation="Main",
+            replicates=1,
+        )
+        session_id = registry.create_session(
+            config=config, experiment_name="test"
+        )
+        registry.register_run(
+            session_id=session_id,
+            run_hash="load_test_hash",
+            josh_path="/tmp/sim.josh",
+            config_content="test",
+            file_mappings=None,
+            parameters={},
+        )
+        run_id = registry.start_run("load_test_hash", session_id=session_id)
+        registry.complete_run(run_id, exit_code=0)
+        return run_id
+
+    def test_load_csv_from_s3_url(self, minio_registry, seed_csv, test_bucket):
+        """load_csv with an s3:// URL should insert rows into cell_data."""
+        from joshpy.cell_data import CellDataLoader
+
+        run_id = self._setup_registry_for_load(minio_registry)
+        csv_data = _make_csv(replicate=0, steps=3)
+        key = f"test-level3/{uuid.uuid4().hex[:8]}/export.csv"
+        s3_url = seed_csv(key, csv_data)
+
+        loader = CellDataLoader(minio_registry)
+        rows = loader.load_csv(
+            csv_path=s3_url,
+            run_id=run_id,
+            run_hash="load_test_hash",
+        )
+
+        assert rows == 3
+
+        # Verify data in registry
+        result = minio_registry.conn.execute(
+            "SELECT step, replicate, \"treeCount\", \"averageHeight\" "
+            "FROM cell_data ORDER BY step"
+        ).fetchall()
+        assert len(result) == 3
+        assert result[0][0] == 0  # step
+        assert result[0][1] == 0  # replicate
+        assert result[0][2] == 10  # treeCount at step 0
+
+    def test_load_csv_creates_variable_columns(
+        self, minio_registry, seed_csv, test_bucket
+    ):
+        """Variable columns from the S3 CSV should be auto-created."""
+        from joshpy.cell_data import CellDataLoader
+
+        run_id = self._setup_registry_for_load(minio_registry)
+        csv_data = _make_csv(replicate=0, steps=2)
+        key = f"test-level3/{uuid.uuid4().hex[:8]}/vars.csv"
+        s3_url = seed_csv(key, csv_data)
+
+        CellDataLoader(minio_registry).load_csv(
+            csv_path=s3_url, run_id=run_id, run_hash="load_test_hash"
+        )
+
+        var_cols = minio_registry.list_variable_columns()
+        assert "treeCount" in var_cols
+        assert "averageHeight" in var_cols
+
+    def test_load_csv_s3_nonexistent_key(self, minio_registry):
+        """Missing S3 object should raise a recognizable error."""
+        from joshpy.cell_data import CellDataLoader
+
+        run_id = self._setup_registry_for_load(minio_registry)
+        loader = CellDataLoader(minio_registry)
+
+        with pytest.raises(Exception, match="HTTP|404|NoSuchKey|IOException"):
+            loader.load_csv(
+                csv_path=f"s3://{TEST_BUCKET}/nonexistent/{uuid.uuid4()}.csv",
+                run_id=run_id,
+                run_hash="load_test_hash",
+            )
+
+    def test_load_csv_s3_missing_required_columns(
+        self, minio_registry, seed_csv
+    ):
+        """CSV without step/replicate should raise ValueError even from S3."""
+        from joshpy.cell_data import CellDataLoader
+
+        run_id = self._setup_registry_for_load(minio_registry)
+        bad_csv = "a,b,c\n1,2,3\n"
+        key = f"test-level3/{uuid.uuid4().hex[:8]}/bad.csv"
+        s3_url = seed_csv(key, bad_csv)
+
+        loader = CellDataLoader(minio_registry)
+        with pytest.raises(ValueError, match="step.*replicate"):
+            loader.load_csv(
+                csv_path=s3_url,
+                run_id=run_id,
+                run_hash="load_test_hash",
+            )
+
+
+# ===================================================================
+# Level 4: End-to-end ingest_results() from MinIO
+# ===================================================================
+
+
+def _make_ingest_registry(minio_registry, josh_content, replicates=2):
+    """Set up registry metadata for ingest_results() tests.
+
+    Creates session, registers run with josh_content, labels it,
+    and creates completed job_runs.  Returns (run_hash, run_id).
+    """
+    from joshpy.jobs import JobConfig
+
+    run_hash = f"ingest_{uuid.uuid4().hex[:8]}"
+
+    config = JobConfig(
+        source_path=Path("/tmp/sim.josh"),
+        simulation="Main",
+        replicates=replicates,
+    )
+    session_id = minio_registry.create_session(
+        config=config, experiment_name="ingest-test"
+    )
+    minio_registry.register_run(
+        session_id=session_id,
+        run_hash=run_hash,
+        josh_path="/tmp/sim.josh",
+        config_content="test",
+        file_mappings=None,
+        parameters={},
+        josh_content=josh_content,
+    )
+    minio_registry.label_run(run_hash, f"label-{run_hash}")
+
+    run_id = None
+    for _ in range(replicates):
+        run_id = minio_registry.start_run(run_hash, session_id=session_id)
+        minio_registry.complete_run(run_id, exit_code=0)
+
+    return run_hash, run_id
+
+
+class TestMinioIngestResults:
+    """Level 4: Full ingest_results() reading real CSVs from MinIO."""
+
+    JOSH_CONTENT = (Path(__file__).parent / "fixtures" / "minio_export.josh").read_text()
+
+    def test_ingest_all_replicates(
+        self,
+        minio_registry,
+        seed_csv,
+        test_bucket,
+        patch_s3_no_ssl,
+        monkeypatch,
+    ):
+        """ingest_results() should load all replicates from S3."""
+        from joshpy.cli import ExportFileInfo, ExportPaths
+        from joshpy.sweep import ingest_results
+
+        run_hash, _ = _make_ingest_registry(
+            minio_registry, self.JOSH_CONTENT, replicates=3
+        )
+        label = f"label-{run_hash}"
+
+        # Seed 3 replicate CSVs
+        prefix = f"test-level4/{run_hash}"
+        for rep in range(3):
+            csv_data = _make_csv(replicate=rep, steps=4)
+            seed_csv(f"{prefix}/output_{rep}.csv", csv_data)
+
+        # Mock CLI — only inspect_exports needs the JAR
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw=f"minio://{test_bucket}/{prefix}/output_{{replicate}}.csv",
+                    protocol="minio",
+                    host=test_bucket,
+                    path=f"/{prefix}/output_{{replicate}}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={
+                "organism": None,
+                "patch": None,
+                "agent": None,
+                "disturbance": None,
+            },
+        )
+
+        monkeypatch.setenv("MINIO_ENDPOINT", MINIO_ENDPOINT)
+        monkeypatch.setenv("MINIO_ACCESS_KEY", MINIO_ACCESS_KEY)
+        monkeypatch.setenv("MINIO_SECRET_KEY", MINIO_SECRET_KEY)
+
+        rows = ingest_results(mock_cli, minio_registry, label, quiet=True)
+
+        # 3 replicates x 4 steps x 1 patch = 12 rows
+        assert rows == 12
+
+        # Verify data is queryable
+        result = minio_registry.conn.execute(
+            "SELECT DISTINCT replicate FROM cell_data ORDER BY replicate"
+        ).fetchall()
+        assert [r[0] for r in result] == [0, 1, 2]
+
+    def test_ingest_results_queryable(
+        self,
+        minio_registry,
+        seed_csv,
+        test_bucket,
+        patch_s3_no_ssl,
+        monkeypatch,
+    ):
+        """After ingest, cell_data should be queryable with aggregates."""
+        from joshpy.cli import ExportFileInfo, ExportPaths
+        from joshpy.sweep import ingest_results
+
+        run_hash, _ = _make_ingest_registry(
+            minio_registry, self.JOSH_CONTENT, replicates=2
+        )
+        label = f"label-{run_hash}"
+
+        prefix = f"test-level4-query/{run_hash}"
+        for rep in range(2):
+            seed_csv(f"{prefix}/output_{rep}.csv", _make_csv(replicate=rep, steps=5))
+
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw=f"minio://{test_bucket}/{prefix}/output_{{replicate}}.csv",
+                    protocol="minio",
+                    host=test_bucket,
+                    path=f"/{prefix}/output_{{replicate}}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={
+                "organism": None,
+                "patch": None,
+                "agent": None,
+                "disturbance": None,
+            },
+        )
+
+        monkeypatch.setenv("MINIO_ENDPOINT", MINIO_ENDPOINT)
+        monkeypatch.setenv("MINIO_ACCESS_KEY", MINIO_ACCESS_KEY)
+        monkeypatch.setenv("MINIO_SECRET_KEY", MINIO_SECRET_KEY)
+
+        ingest_results(mock_cli, minio_registry, label, quiet=True)
+
+        # Aggregate query
+        avg = minio_registry.conn.execute(
+            'SELECT AVG("treeCount") FROM cell_data WHERE run_hash = ?',
+            [run_hash],
+        ).fetchone()[0]
+        assert avg is not None
+        assert avg > 0
+
+
+# ===================================================================
+# Level 5: Partial / interrupted sweep recovery
+# ===================================================================
+
+
+class TestMinioPartialRecovery:
+    """Level 5: Graceful recovery when some replicates are missing."""
+
+    JOSH_CONTENT = (Path(__file__).parent / "fixtures" / "minio_export.josh").read_text()
+
+    def _run_ingest(
+        self,
+        minio_registry,
+        seed_csv,
+        test_bucket,
+        monkeypatch,
+        *,
+        replicates_registered: int,
+        replicates_seeded: list[int],
+        steps: int = 3,
+    ) -> tuple[int, str]:
+        """Helper: set up registry, seed some replicates, call ingest_results."""
+        from joshpy.cli import ExportFileInfo, ExportPaths
+        from joshpy.sweep import ingest_results
+
+        run_hash, _ = _make_ingest_registry(
+            minio_registry, self.JOSH_CONTENT, replicates=replicates_registered
+        )
+        label = f"label-{run_hash}"
+
+        prefix = f"test-level5/{run_hash}"
+        for rep in replicates_seeded:
+            seed_csv(
+                f"{prefix}/output_{rep}.csv",
+                _make_csv(replicate=rep, steps=steps),
+            )
+
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw=f"minio://{test_bucket}/{prefix}/output_{{replicate}}.csv",
+                    protocol="minio",
+                    host=test_bucket,
+                    path=f"/{prefix}/output_{{replicate}}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={
+                "organism": None,
+                "patch": None,
+                "agent": None,
+                "disturbance": None,
+            },
+        )
+
+        monkeypatch.setenv("MINIO_ENDPOINT", MINIO_ENDPOINT)
+        monkeypatch.setenv("MINIO_ACCESS_KEY", MINIO_ACCESS_KEY)
+        monkeypatch.setenv("MINIO_SECRET_KEY", MINIO_SECRET_KEY)
+
+        rows = ingest_results(mock_cli, minio_registry, label, quiet=True)
+        return rows, run_hash
+
+    def test_partial_replicates_graceful(
+        self, minio_registry, seed_csv, test_bucket, patch_s3_no_ssl, monkeypatch
+    ):
+        """Only 2 of 3 replicates exist — should load 2, skip 1, no error."""
+        rows, run_hash = self._run_ingest(
+            minio_registry,
+            seed_csv,
+            test_bucket,
+            monkeypatch,
+            replicates_registered=3,
+            replicates_seeded=[0, 2],  # replicate 1 missing
+            steps=4,
+        )
+
+        # 2 replicates x 4 steps = 8 rows
+        assert rows == 8
+
+        # Verify only replicates 0 and 2 present
+        reps = minio_registry.conn.execute(
+            "SELECT DISTINCT replicate FROM cell_data "
+            "WHERE run_hash = ? ORDER BY replicate",
+            [run_hash],
+        ).fetchall()
+        assert [r[0] for r in reps] == [0, 2]
+
+    def test_zero_replicates_available(
+        self, minio_registry, seed_csv, test_bucket, patch_s3_no_ssl, monkeypatch
+    ):
+        """No CSVs in MinIO — should return 0 rows, no exception."""
+        rows, _ = self._run_ingest(
+            minio_registry,
+            seed_csv,
+            test_bucket,
+            monkeypatch,
+            replicates_registered=3,
+            replicates_seeded=[],  # nothing written
+        )
+        assert rows == 0
+
+    def test_single_replicate_of_many(
+        self, minio_registry, seed_csv, test_bucket, patch_s3_no_ssl, monkeypatch
+    ):
+        """1 of 10 replicates available — should load only that one."""
+        rows, run_hash = self._run_ingest(
+            minio_registry,
+            seed_csv,
+            test_bucket,
+            monkeypatch,
+            replicates_registered=10,
+            replicates_seeded=[7],
+            steps=3,
+        )
+        assert rows == 3
+
+        reps = minio_registry.conn.execute(
+            "SELECT DISTINCT replicate FROM cell_data WHERE run_hash = ?",
+            [run_hash],
+        ).fetchall()
+        assert [r[0] for r in reps] == [7]
+
+
+# ===================================================================
+# Edge cases
+# ===================================================================
+
+
+class TestMinioEdgeCases:
+    """Edge cases: bad credentials, missing bucket, namespace isolation."""
+
+    def test_bad_credentials_clear_error(self, minio_available, test_bucket):
+        """Wrong credentials should produce an actionable error."""
+        import duckdb
+        from joshpy.registry import configure_s3
+
+        conn = duckdb.connect(":memory:")
+        configure_s3(
+            conn,
+            endpoint=MINIO_ENDPOINT,
+            access_key="WRONG_KEY",
+            secret_key="WRONG_SECRET",
+            use_ssl=False,
+        )
+
+        with pytest.raises(Exception, match="403|AccessDenied|Forbidden|signature"):
+            conn.execute(
+                f"SELECT * FROM read_csv_auto('s3://{test_bucket}/results/output_0.csv')"
+            ).fetchall()
+
+        conn.close()
+
+    def test_nonexistent_bucket_clear_error(self, minio_conn):
+        """Reading from a missing bucket should raise a clear error."""
+        with pytest.raises(Exception, match="404|NoSuchBucket|NoSuchKey|not found"):
+            minio_conn.execute(
+                "SELECT * FROM read_csv_auto("
+                "'s3://this-bucket-does-not-exist/file.csv')"
+            ).fetchall()
+
+    def test_namespace_isolation(
+        self, minio_registry, seed_csv, test_bucket
+    ):
+        """Two run_hashes should not leak data into each other."""
+        from joshpy.cell_data import CellDataLoader
+        from joshpy.jobs import JobConfig
+
+        config = JobConfig(
+            source_path=Path("/tmp/sim.josh"),
+            simulation="Main",
+            replicates=1,
+        )
+        session_id = minio_registry.create_session(
+            config=config, experiment_name="isolation-test"
+        )
+
+        # Register two runs
+        for rh in ("hash_AAA", "hash_BBB"):
+            minio_registry.register_run(
+                session_id=session_id,
+                run_hash=rh,
+                josh_path="/tmp/sim.josh",
+                config_content="test",
+                file_mappings=None,
+                parameters={},
+            )
+
+        run_id_a = minio_registry.start_run("hash_AAA", session_id=session_id)
+        minio_registry.complete_run(run_id_a, exit_code=0)
+        run_id_b = minio_registry.start_run("hash_BBB", session_id=session_id)
+        minio_registry.complete_run(run_id_b, exit_code=0)
+
+        # Seed different CSVs
+        prefix = f"test-isolation/{uuid.uuid4().hex[:8]}"
+        csv_a = "step,replicate,position.x,position.y,val\n0,0,0.0,0.0,111\n"
+        csv_b = "step,replicate,position.x,position.y,val\n0,0,0.0,0.0,999\n"
+        url_a = seed_csv(f"{prefix}/a.csv", csv_a)
+        url_b = seed_csv(f"{prefix}/b.csv", csv_b)
+
+        loader = CellDataLoader(minio_registry)
+        loader.load_csv(csv_path=url_a, run_id=run_id_a, run_hash="hash_AAA")
+        loader.load_csv(csv_path=url_b, run_id=run_id_b, run_hash="hash_BBB")
+
+        # Query by hash — should be isolated
+        val_a = minio_registry.conn.execute(
+            'SELECT val FROM cell_data WHERE run_hash = ?', ["hash_AAA"]
+        ).fetchone()[0]
+        val_b = minio_registry.conn.execute(
+            'SELECT val FROM cell_data WHERE run_hash = ?', ["hash_BBB"]
+        ).fetchone()[0]
+
+        assert val_a == 111
+        assert val_b == 999
diff --git a/tests/test_sweep.py b/tests/test_sweep.py
index 092839c..5c2df1a 100644
--- a/tests/test_sweep.py
+++ b/tests/test_sweep.py
@@ -1008,5 +1008,273 @@ def test_with_label_on_collision_timestamp(self):
         registry.close()
 
 
+class TestIngestResults(unittest.TestCase):
+    """Tests for ingest_results()."""
+
+    def _make_registry_with_run(self, replicates=3):
+        """Create an in-memory registry with a labeled run for testing."""
+        registry = RunRegistry(":memory:")
+        config = JobConfig(
+            source_path=Path("/tmp/sim.josh"),
+            simulation="Main",
+            replicates=replicates,
+        )
+        session_id = registry.create_session(
+            config=config,
+            experiment_name="test",
+        )
+        # Register a config
+        registry.register_run(
+            session_id=session_id,
+            run_hash="abc123def456",
+            josh_path="/tmp/sim.josh",
+            config_content="config_here",
+            file_mappings=None,
+            parameters={"maxGrowth": 50},
+            josh_content="simulation Main { }",
+        )
+        registry.label_run("abc123def456", "test-label")
+
+        # Start runs so _resolve_run_id_for_hash works and replicate count is right
+        run_id = None
+        for _ in range(replicates):
+            run_id = registry.start_run("abc123def456", session_id=session_id)
+            registry.complete_run(run_id, exit_code=0)
+
+        return registry, session_id, run_id
+
+    @patch("joshpy.sweep.CellDataLoader")
+    def test_local_file_protocol(self, mock_loader_cls):
+        """ingest_results with file:// protocol loads local CSVs."""
+        from joshpy.sweep import ingest_results
+        from joshpy.cli import ExportFileInfo, ExportPaths
+
+        registry, _, run_id = self._make_registry_with_run()
+
+        mock_loader = MagicMock()
+        mock_loader.load_csv.return_value = 100
+        mock_loader_cls.return_value = mock_loader
+
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw="file:///tmp/output_{replicate}.csv",
+                    protocol="file",
+                    host="",
+                    path="/tmp/output_{replicate}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={"organism": None, "patch": None, "agent": None, "disturbance": None},
+        )
+
+        # Create fake CSV files
+        import tempfile, os
+        with tempfile.TemporaryDirectory() as tmpdir:
+            for rep in range(3):
+                csv_path = Path(f"/tmp/output_{rep}.csv")
+                csv_path.write_text("step,replicate,val\n0,0,1.0\n")
+
+            try:
+                rows = ingest_results(mock_cli, registry, "test-label", quiet=True)
+                # Should have called load_csv 3 times
+                self.assertEqual(mock_loader.load_csv.call_count, 3)
+            finally:
+                for rep in range(3):
+                    Path(f"/tmp/output_{rep}.csv").unlink(missing_ok=True)
+
+        registry.close()
+
+    @patch("joshpy.sweep.CellDataLoader")
+    def test_missing_replicate_skipped(self, mock_loader_cls):
+        """Missing CSVs should be skipped gracefully."""
+        from joshpy.sweep import ingest_results
+        from joshpy.cli import ExportFileInfo, ExportPaths
+
+        registry, _, _ = self._make_registry_with_run()
+
+        mock_loader = MagicMock()
+        mock_loader.load_csv.side_effect = FileNotFoundError("not found")
+        mock_loader_cls.return_value = mock_loader
+
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw="file:///tmp/missing_{replicate}.csv",
+                    protocol="file",
+                    host="",
+                    path="/tmp/missing_{replicate}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={"organism": None, "patch": None, "agent": None, "disturbance": None},
+        )
+
+        rows = ingest_results(mock_cli, registry, "test-label", quiet=True)
+        self.assertEqual(rows, 0)
+        registry.close()
+
+    def test_unknown_label_raises(self):
+        """ingest_results should raise KeyError for unknown label."""
+        from joshpy.sweep import ingest_results
+
+        registry = RunRegistry(":memory:")
+        mock_cli = MagicMock()
+
+        with self.assertRaises(KeyError):
+            ingest_results(mock_cli, registry, "nonexistent-label")
+        registry.close()
+
+    @patch("joshpy.sweep.CellDataLoader")
+    def test_minio_protocol_configures_s3(self, mock_loader_cls):
+        """minio:// protocol should call configure_s3 and build s3:// URLs."""
+        from joshpy.sweep import ingest_results
+        from joshpy.cli import ExportFileInfo, ExportPaths
+
+        registry, _, _ = self._make_registry_with_run()
+
+        mock_loader = MagicMock()
+        mock_loader.load_csv.return_value = 50
+        mock_loader_cls.return_value = mock_loader
+
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw="minio://my-bucket/results/output_{replicate}.csv",
+                    protocol="minio",
+                    host="my-bucket",
+                    path="/results/output_{replicate}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={"organism": None, "patch": None, "agent": None, "disturbance": None},
+        )
+
+        env = {
+            "MINIO_ENDPOINT": "storage.example.com",
+            "MINIO_ACCESS_KEY": "AKID",
+            "MINIO_SECRET_KEY": "SECRET",
+        }
+        with patch("joshpy.sweep.configure_s3") as mock_configure, \
+             patch.dict("os.environ", env):
+            rows = ingest_results(mock_cli, registry, "test-label", quiet=True)
+
+            # Should have configured S3
+            mock_configure.assert_called_once()
+            call_args = mock_configure.call_args
+            self.assertEqual(call_args[0][1], "storage.example.com")
+
+            # load_csv should have been called with s3:// URLs
+            for call in mock_loader.load_csv.call_args_list:
+                csv_arg = call[1].get("csv_path") or call[0][0]
+                self.assertTrue(str(csv_arg).startswith("s3://my-bucket/"))
+
+        registry.close()
+
+    @patch("joshpy.sweep.CellDataLoader")
+    def test_minio_missing_creds_raises(self, mock_loader_cls):
+        """minio:// without env vars should raise RuntimeError."""
+        from joshpy.sweep import ingest_results
+        from joshpy.cli import ExportFileInfo, ExportPaths
+
+        registry, _, _ = self._make_registry_with_run()
+
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw="minio://bucket/out_{replicate}.csv",
+                    protocol="minio",
+                    host="bucket",
+                    path="/out_{replicate}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={"organism": None, "patch": None, "agent": None, "disturbance": None},
+        )
+
+        # Clear any minio env vars
+        clean_env = {k: v for k, v in __import__("os").environ.items()
+                     if not k.startswith("MINIO_")}
+        with patch.dict("os.environ", clean_env, clear=True):
+            with self.assertRaises(RuntimeError):
+                ingest_results(mock_cli, registry, "test-label", quiet=True)
+
+        registry.close()
+
+    @patch("joshpy.sweep.CellDataLoader")
+    def test_josh_content_fallback(self, mock_loader_cls):
+        """Should use josh_content from registry when josh_path doesn't exist."""
+        from joshpy.sweep import ingest_results
+        from joshpy.cli import ExportFileInfo, ExportPaths
+
+        registry, _, _ = self._make_registry_with_run()
+
+        mock_loader = MagicMock()
+        mock_loader.load_csv.return_value = 10
+        mock_loader_cls.return_value = mock_loader
+
+        mock_cli = MagicMock()
+        mock_cli.inspect_exports.return_value = ExportPaths(
+            simulation="Main",
+            export_files={
+                "patch": ExportFileInfo(
+                    raw="file:///tmp/out_{replicate}.csv",
+                    protocol="file",
+                    host="",
+                    path="/tmp/out_{replicate}.csv",
+                    file_type="csv",
+                ),
+                "meta": None,
+                "entity": None,
+            },
+            debug_files={"organism": None, "patch": None, "agent": None, "disturbance": None},
+        )
+
+        # josh_path is /tmp/sim.josh which doesn't exist — should fall back to josh_content
+        rows = ingest_results(mock_cli, registry, "test-label", quiet=True)
+
+        # inspect_exports should have been called with a temp file (not /tmp/sim.josh)
+        call_config = mock_cli.inspect_exports.call_args[0][0]
+        self.assertNotEqual(str(call_config.script), "/tmp/sim.josh")
+        # Temp file has .josh suffix
+        self.assertTrue(str(call_config.script).endswith(".josh"))
+
+        registry.close()
+
+
+class TestConfigureS3(unittest.TestCase):
+    """Tests for configure_s3()."""
+
+    def test_executes_install_and_create_secret(self):
+        """configure_s3 should call INSTALL httpfs and CREATE SECRET."""
+        from joshpy.registry import configure_s3
+
+        mock_conn = MagicMock()
+        configure_s3(mock_conn, "storage.example.com", "AKID", "SECRET")
+
+        # Should have called execute twice: INSTALL + CREATE SECRET
+        self.assertEqual(mock_conn.execute.call_count, 2)
+        first_call = mock_conn.execute.call_args_list[0]
+        self.assertIn("INSTALL httpfs", first_call[0][0])
+        second_call = mock_conn.execute.call_args_list[1]
+        self.assertIn("CREATE OR REPLACE SECRET", second_call[0][0])
+
+
 if __name__ == "__main__":
     unittest.main()