diff --git a/examples/research_demo/agents.py b/examples/research_demo/agents.py
index 629bc1b..360c665 100644
--- a/examples/research_demo/agents.py
+++ b/examples/research_demo/agents.py
@@ -1,9 +1,10 @@
 """Agent configuration for the Research Demo application.
 
-Multi-agent architecture:
+Multi-agent architecture with three agents:
 - research_coordinator: Root agent that owns workflow state (planning, tasks, HITL)
-  and delegates academic paper research to the arXiv specialist.
+  and delegates to the arXiv specialist and data analyst.
 - arxiv_specialist: Leaf agent focused on arXiv paper search, analysis, and ingestion.
+- data_analyst: Leaf agent that runs multi-step data analysis in a stateful executor.
 
 Uses framework-provided tools exclusively — no app-specific tools needed.
 """
@@ -26,6 +27,7 @@
     grep,
     glob,
 )
+from agentic_cli.tools.sandbox import sandbox_execute
 
 
 # ---------------------------------------------------------------------------
@@ -102,6 +104,27 @@
 """
 
 
+# ---------------------------------------------------------------------------
+# Data Analyst (leaf agent)
+# ---------------------------------------------------------------------------
+
+DATA_ANALYST_PROMPT = """You are a data-analysis specialist. You run multi-step Python analysis in a stateful, isolated executor (variables and DataFrames persist across calls).
+
+## Data
+- Pre-mounted sample datasets are read-only under `data/samples/` (e.g. `data/samples/benchmarks.csv`). Discover them with `os.listdir('data/samples')`.
+- Files handed to you by the coordinator arrive via the tool's `inputs` argument and appear at `inputs/<filename>`. Load them by that relative path — never by a host path.
+
+## Working style
+1. Explore first: `df = pd.read_csv('data/samples/benchmarks.csv'); print(df.info()); print(df.describe())`.
+2. Transform/aggregate step by step — the session remembers your DataFrames between calls.
+3. Plot with matplotlib (figures are captured automatically).
+4. Write FINAL deliverables (cleaned tables, key figures) to `outputs/` — those persist and are shared with other agents. Keep scratch in the working directory.
+5. Save a short narrative findings report with `write_file`.
+
+Report what you found with concrete numbers, and name the files you wrote to `outputs/`.
+"""
+
+
 # ---------------------------------------------------------------------------
 # Research Coordinator (root agent)
 # ---------------------------------------------------------------------------
@@ -154,6 +177,7 @@
 6. **IMMEDIATELY show the plan** to the user in your response.
 7. **WAIT for user confirmation** before executing tasks.
 8. For arXiv paper research, **delegate to arxiv_specialist** (it has KB writer access and writes concept pages when 3+ related papers accumulate).
+- For multi-step data analysis (datasets, DataFrames, plots), delegate to **data_analyst**. Use `execute_python` only for quick one-off calculations.
 9. Execute ONE task at a time, updating the plan after each.
 10. Use `web_fetch` to extract information from specific URLs found during research.
 11. Use `execute_python` for quick calculations and data validation.
@@ -216,6 +240,14 @@
         ],
         description="arXiv paper research specialist: search, analyze, save, and catalog academic papers",
     ),
+    # Leaf agent: data analyst (must be listed before coordinator)
+    AgentConfig(
+        name="data_analyst",
+        prompt=DATA_ANALYST_PROMPT,
+        include_state_tools=False,
+        tools=[sandbox_execute, read_file, write_file, ask_clarification],
+        description="Stateful data-analysis specialist: loads datasets and runs multi-step pandas/plotting analysis in an isolated executor.",
+    ),
     # Root agent: research coordinator (owns workflow state, delegates arXiv work)
     AgentConfig(
         name="research_coordinator",
@@ -241,7 +273,7 @@
             grep,
             diff_compare,
         ],
-        sub_agents=["arxiv_specialist"],
+        sub_agents=["arxiv_specialist", "data_analyst"],
         description="Research coordinator with memory, planning, task management, knowledge base, and HITL capabilities",
     ),
 ]
diff --git a/examples/research_demo/data/benchmarks.csv b/examples/research_demo/data/benchmarks.csv
new file mode 100644
index 0000000..516b77b
--- /dev/null
+++ b/examples/research_demo/data/benchmarks.csv
@@ -0,0 +1,6 @@
+model,dataset,accuracy,params_millions,year
+resnet50,imagenet,76.1,25.6,2015
+vit_b16,imagenet,77.9,86.6,2020
+convnext_t,imagenet,82.1,28.6,2022
+efficientnet_b0,imagenet,77.1,5.3,2019
+swin_t,imagenet,81.3,28.3,2021
diff --git a/examples/research_demo/settings.py b/examples/research_demo/settings.py
index cf86af4..12b922c 100644
--- a/examples/research_demo/settings.py
+++ b/examples/research_demo/settings.py
@@ -42,3 +42,8 @@ def model_post_init(self, __context):
         """
         if "verbose_thinking" not in self.model_fields_set:
             object.__setattr__(self, "verbose_thinking", False)
+        if "sandbox_data_mounts" not in self.model_fields_set:
+            data_dir = Path(__file__).parent / "data"
+            object.__setattr__(self, "sandbox_data_mounts", [f"{data_dir}:samples"])
+        if "sandbox_outputs_dir" not in self.model_fields_set:
+            object.__setattr__(self, "sandbox_outputs_dir", str(Path(self.workspace_dir) / "artifacts"))
diff --git a/src/agentic_cli/tools/factories.py b/src/agentic_cli/tools/factories.py
index 712700a..3b8ff3f 100644
--- a/src/agentic_cli/tools/factories.py
+++ b/src/agentic_cli/tools/factories.py
@@ -365,6 +365,7 @@ def sandbox_execute(
         code: str,
         session_id: str = "default",
         timeout_seconds: int = 120,
+        inputs: list[str] | None = None,
     ) -> dict[str, Any]:
         """Execute Python code in a stateful sandbox.
 
@@ -372,17 +373,19 @@ def sandbox_execute(
             code: Python code to execute.
             session_id: Session identifier for state persistence.
             timeout_seconds: Maximum execution time in seconds.
+            inputs: Optional list of host file paths to stage into
+                inputs/<basename> inside the session before execution.
 
         Returns:
             Dictionary with execution results.
         """
         # Opt-in gate — the workflow binds THIS tool (base_manager), so the gate
         # must live here, not only on the module-level tool. Without it the
-        # sandbox_execute_enabled switch is inert in the real path.
+        # stateful_executor_backend setting is inert in the real path.
         from agentic_cli.config import get_settings
         from agentic_cli.tools.sandbox.manager import sandbox_disabled_reason
 
-        if not getattr(get_settings(), "sandbox_execute_enabled", False):
+        if getattr(get_settings(), "stateful_executor_backend", "none") == "none":
             return {"success": False, "error": sandbox_disabled_reason(get_settings())}
 
         # Namespace the default session to the active conversation so distinct
@@ -398,6 +401,7 @@ def sandbox_execute(
             code=code,
             session_id=sid,
             timeout_seconds=timeout_seconds,
+            inputs=inputs,
         )
         return {
             "success": result.success,
diff --git a/src/agentic_cli/tools/sandbox/__init__.py b/src/agentic_cli/tools/sandbox/__init__.py
index 852d313..1509e65 100644
--- a/src/agentic_cli/tools/sandbox/__init__.py
+++ b/src/agentic_cli/tools/sandbox/__init__.py
@@ -18,22 +18,26 @@
     # unsandboxed and stateful, so an "Allow always" for the stateless scratchpad
     # must NOT silently authorize it. A deliberate ``python.*`` grant still covers
     # both.
-    capabilities=[Capability("python.exec.stateful")],
+    capabilities=[Capability("python.exec.stateful"), Capability("filesystem.read", target_arg="inputs")],
     description=(
         "Execute Python code in a stateful session. "
         "State (variables, imports) persists across calls within the same session. "
-        "Isolation depends on sandbox_backend: 'jupyter_docker' runs in a "
+        "Isolation depends on stateful_executor_backend: 'docker' runs in a "
         "network-isolated container (no network egress; memory/CPU/PID-capped, "
-        "though disk is not); "
-        "'jupyter_local' runs with host privileges and shared filesystem. "
+        "though disk is not); 'local' runs with host privileges and shared filesystem. "
         "Disabled unless explicitly enabled. Use for data analysis, prototyping, "
-        "and producing work output. Use execute_python for quick stateless calculations."
+        "and producing work output. Use execute_python for quick stateless calculations. "
+        "Each `inputs` file is copied to `inputs/<filename>` inside the session before "
+        "the code runs; load it by that relative path (e.g. open('inputs/data.csv')). "
+        "Write scratch/intermediate files to the working directory; write FINAL deliverables "
+        "(figures, tables) to `outputs/` — those persist and are shared with other agents."
     ),
 )
 def sandbox_execute(
     code: str,
     session_id: str = "default",
     timeout_seconds: int = 120,
+    inputs: list[str] | None = None,
 ) -> dict[str, Any]:
     """Execute Python code in a stateful sandbox.
 
@@ -41,18 +45,20 @@ def sandbox_execute(
         code: Python code to execute.
         session_id: Session identifier for state persistence (default: "default").
         timeout_seconds: Maximum execution time in seconds.
+        inputs: Optional list of host file paths to stage into
+            inputs/<basename> inside the session before execution.
 
     Returns:
         Dictionary with execution results.
     """
-    # Opt-in (see sandbox_execute_enabled). Gate before touching the service so a
-    # disabled deployment fails fast. NOTE: the workflow binds the factory tool
-    # (tools/factories.py), which gates identically — this gate covers the
-    # module-level tool used outside the workflow.
+    # Opt-in gate. Gate before touching the service so a disabled deployment
+    # fails fast. NOTE: the workflow binds the factory tool (tools/factories.py),
+    # which gates identically — this gate covers the module-level tool used
+    # outside the workflow.
     from agentic_cli.tools.sandbox.manager import sandbox_disabled_reason
 
     settings = get_settings()
-    if not getattr(settings, "sandbox_execute_enabled", False):
+    if getattr(settings, "stateful_executor_backend", "none") == "none":
         return {"success": False, "error": sandbox_disabled_reason(settings)}
 
     manager = require_service(SANDBOX_MANAGER)
@@ -62,6 +68,7 @@ def sandbox_execute(
         code=code,
         session_id=session_id,
         timeout_seconds=timeout_seconds,
+        inputs=inputs,
     )
     return {
         "success": result.success,
diff --git a/src/agentic_cli/tools/sandbox/backends/base.py b/src/agentic_cli/tools/sandbox/backends/base.py
index 8980516..6056c06 100644
--- a/src/agentic_cli/tools/sandbox/backends/base.py
+++ b/src/agentic_cli/tools/sandbox/backends/base.py
@@ -18,6 +18,7 @@ def execute(
         session_id: str,
         timeout_seconds: int = 120,
         working_dir: Path | None = None,
+        inputs: list[str] | None = None,
     ) -> ExecutionResult:
         """Execute code in the given session.
 
diff --git a/src/agentic_cli/tools/sandbox/backends/jupyter_docker.py b/src/agentic_cli/tools/sandbox/backends/jupyter_docker.py
index 4cbdd77..b2c12cb 100644
--- a/src/agentic_cli/tools/sandbox/backends/jupyter_docker.py
+++ b/src/agentic_cli/tools/sandbox/backends/jupyter_docker.py
@@ -5,6 +5,7 @@
 import json
 import os
 import queue
+import shutil
 import threading
 import time
 from pathlib import Path
@@ -212,6 +213,32 @@ def _ensure_runtime(self):
             self._runtime = DockerContainerRuntime(avail.runtime or "docker")
         return self._runtime
 
+    def _outputs_dir(self) -> Path:
+        configured = getattr(self._settings, "sandbox_outputs_dir", "") or ""
+        base = Path(configured) if configured else Path(self._settings.workspace_dir) / "artifacts"
+        base.mkdir(parents=True, exist_ok=True)
+        try:
+            os.chmod(base, 0o777)  # container runs as host uid; keep writable across sessions
+        except OSError:
+            pass
+        return base
+
+    def _parse_data_mounts(self) -> list[tuple[str, str]]:
+        """Parse sandbox_data_mounts into (host_path, sanitized_name) pairs.
+
+        Used by both _build_spec (to build Mount objects) and _start_session
+        (to pre-create host-side mount-point dirs before runtime.start).
+        """
+        result = []
+        for entry in self._settings.sandbox_data_mounts:
+            host, _, name = entry.partition(":")
+            # Sanitize the mount name so a hostile '..'/absolute value can't
+            # remap the mount outside /workspace/data/ (sanitize_filename maps
+            # '/' and '.' to '_').
+            name = sanitize_filename(name or Path(host).name) or "mount"
+            result.append((host, name))
+        return result
+
     def _build_spec(self, session_id: str, working_dir) -> ContainerSpec:
         s = self._settings
         here = Path(__file__).parent
@@ -223,12 +250,7 @@ def _build_spec(self, session_id: str, working_dir) -> ContainerSpec:
             Mount(str(here / "driver.py"), f"{_DRIVER_DIR}/driver.py", read_only=True),
             Mount(str(here / "kernel_exec.py"), f"{_DRIVER_DIR}/kernel_exec.py", read_only=True),
         ]
-        for entry in s.sandbox_data_mounts:
-            host, _, name = entry.partition(":")
-            # Sanitize the mount name so a hostile '..'/absolute value can't
-            # remap the mount outside /workspace/data/ (sanitize_filename maps
-            # '/' and '.' to '_').
-            name = sanitize_filename(name or Path(host).name) or "mount"
+        for host, name in self._parse_data_mounts():
             mounts.append(Mount(host, f"/workspace/data/{name}", read_only=True))
         env = {
             "HOME": "/tmp", "MPLCONFIGDIR": "/tmp", "IPYTHONDIR": "/tmp",
@@ -262,6 +284,23 @@ def _start_session(self, session_id: str, working_dir) -> ContainerSession:
         # permissions. No chmod needed.
         runtime = self._ensure_runtime()
         spec = self._build_spec(session_id, working_dir)
+        if working_dir is not None:
+            wd = Path(working_dir)
+            # Pre-create the /workspace/outputs mount point as the host user.
+            # Docker would otherwise create this nested bind-mount target as root,
+            # which pollutes the session dir and breaks host-side cleanup.
+            (wd / "outputs").mkdir(parents=True, exist_ok=True)
+            # Pre-create data-mount host-side mount points before runtime.start().
+            # Without this, Docker creates them as root inside the session dir,
+            # which breaks host-side cleanup and causes ACL issues on macOS.
+            for host, name in self._parse_data_mounts():
+                mount_point = wd / "data" / name
+                if Path(host).is_dir():
+                    mount_point.mkdir(parents=True, exist_ok=True)
+                else:
+                    mount_point.parent.mkdir(parents=True, exist_ok=True)
+                    if not mount_point.exists():
+                        mount_point.touch()
         handle = runtime.start(spec)
         name = spec.name
         session = ContainerSession(
@@ -281,7 +320,7 @@ def _start_session(self, session_id: str, working_dir) -> ContainerSession:
         self._sessions[session_id] = session
         return session
 
-    def execute(self, code, session_id, timeout_seconds=120, working_dir=None) -> ExecutionResult:
+    def execute(self, code, session_id, timeout_seconds=120, working_dir=None, inputs=None) -> ExecutionResult:
         avail = self._detect()
         if not avail.available:
             return ExecutionResult(
@@ -292,13 +331,32 @@ def execute(self, code, session_id, timeout_seconds=120, working_dir=None) -> Ex
         ok, msg = kernel_exec.validate_code(code)
         if not ok:
             return ExecutionResult(success=False, error=msg)
+        if inputs:
+            from agentic_cli.tools.sandbox.manager import stage_inputs
+            try:
+                stage_inputs(working_dir, inputs)
+            except ValueError as exc:
+                return ExecutionResult(success=False, error=f"input staging failed: {exc}")
         session = self._sessions.get(session_id)
         if session is None or session.status == "dead":
             try:
                 session = self._start_session(session_id, working_dir)
             except Exception as exc:
                 return ExecutionResult(success=False, error=f"Failed to start sandbox: {exc}")
-        return session.execute(code, timeout_seconds)
+        result = session.execute(code, timeout_seconds)
+        if result.success and working_dir is not None:
+            session_outs = Path(working_dir) / "outputs"
+            if session_outs.is_dir():
+                shared = self._outputs_dir()
+                extra: list[str] = []
+                for src in sorted(session_outs.iterdir()):
+                    if src.is_file():
+                        dst = shared / src.name
+                        shutil.copy2(src, dst)
+                        extra.append(str(dst))
+                if extra:
+                    result.artifacts = list(result.artifacts) + extra
+        return result
 
     def reset_session(self, session_id: str) -> None:
         session = self._sessions.pop(session_id, None)
diff --git a/src/agentic_cli/tools/sandbox/backends/jupyter_local.py b/src/agentic_cli/tools/sandbox/backends/jupyter_local.py
index 2804397..c834dfd 100644
--- a/src/agentic_cli/tools/sandbox/backends/jupyter_local.py
+++ b/src/agentic_cli/tools/sandbox/backends/jupyter_local.py
@@ -82,10 +82,10 @@ def _get_or_create_session(
             return self._sessions[session_id]
 
         km = KernelManager()
+        start_kwargs: dict = {}
         if working_dir:
-            km.cwd = str(working_dir)
-
-        km.start_kernel()
+            start_kwargs["cwd"] = str(working_dir)
+        km.start_kernel(**start_kwargs)
         kc = km.blocking_client()
         kc.start_channels()
         kc.wait_for_ready(timeout=30)
@@ -113,6 +113,7 @@ def execute(
         session_id: str,
         timeout_seconds: int = 120,
         working_dir: Path | None = None,
+        inputs: list[str] | None = None,
     ) -> ExecutionResult:
         """Execute code in a Jupyter kernel session."""
         # Pre-scan for blocked patterns
@@ -120,6 +121,16 @@ def execute(
         if not valid:
             return ExecutionResult(success=False, error=error)
 
+        if working_dir is not None:
+            (Path(working_dir) / "outputs").mkdir(parents=True, exist_ok=True)
+
+        if inputs:
+            from agentic_cli.tools.sandbox.manager import stage_inputs
+            try:
+                stage_inputs(working_dir, inputs)
+            except ValueError as exc:
+                return ExecutionResult(success=False, error=f"input staging failed: {exc}")
+
         _, kc = self._get_or_create_session(session_id, working_dir)
         msg_id = kc.execute(code)
         data = kernel_exec.collect_execution(kc, msg_id, timeout_seconds, working_dir)
diff --git a/src/agentic_cli/tools/sandbox/manager.py b/src/agentic_cli/tools/sandbox/manager.py
index e485fb5..f5f6f4d 100644
--- a/src/agentic_cli/tools/sandbox/manager.py
+++ b/src/agentic_cli/tools/sandbox/manager.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import atexit
+import shutil
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, TYPE_CHECKING
@@ -22,19 +23,41 @@
 logger = Loggers.tools()
 
 
+def stage_inputs(session_dir: Path, inputs: list[str]) -> None:
+    """Copy each host path into session_dir/inputs/<basename> (the in-sandbox
+    'inputs/<name>' contract). Raises ValueError on a missing file or a
+    basename collision (v1 does not support renaming)."""
+    if session_dir is None:
+        raise ValueError("session_dir is required for staging inputs")
+    if not inputs:
+        return
+    inputs_dir = Path(session_dir) / "inputs"
+    inputs_dir.mkdir(parents=True, exist_ok=True)
+    seen: set[str] = set()
+    for src in inputs:
+        p = Path(src).expanduser()
+        if not p.is_file():
+            raise ValueError(f"input file not found: {src}")
+        name = p.name
+        if name in seen:
+            raise ValueError(f"duplicate input basename {name!r}; rename one of the source files")
+        seen.add(name)
+        shutil.copy2(p, inputs_dir / name)
+
+
 def sandbox_disabled_reason(settings) -> str:
     """Backend-aware reason string for a disabled sandbox_execute. Shared by
     every entry point (module tool + factory tool) so the opt-in gate is
     applied consistently and worded accurately for the selected backend."""
-    backend = getattr(settings, "sandbox_backend", "jupyter_local")
-    if backend == "jupyter_docker":
-        detail = ("The 'jupyter_docker' backend runs it in a network-isolated, "
-                  "memory/CPU/PID-capped container.")
+    backend = getattr(settings, "stateful_executor_backend", "none")
+    if backend == "docker":
+        detail = "The 'docker' backend runs it in a network-isolated, memory/CPU/PID-capped container."
+    elif backend == "local":
+        detail = "The 'local' backend runs Python with host privileges and no OS sandbox."
     else:
-        detail = (f"The '{backend}' backend runs Python with host privileges and "
-                  "no OS sandbox (use 'jupyter_docker' for isolation).")
-    return (f"sandbox_execute is not enabled. {detail} "
-            "Enable sandbox_execute_enabled in settings to use it.")
+        detail = "No stateful executor is configured."
+    return (f"sandbox_execute is not available. {detail} "
+            "Set stateful_executor_backend to 'docker' (isolated) or 'local' (unsandboxed) to use it.")
 
 
 @dataclass
@@ -52,7 +75,7 @@ class SandboxManager:
     Args:
         settings: Application settings instance.
         backend: Optional backend for test injection. If None, created
-            lazily from settings.sandbox_backend.
+            lazily from settings.stateful_executor_backend.
     """
 
     def __init__(
@@ -70,18 +93,18 @@ def __init__(
     def _ensure_backend(self) -> "SandboxBackend":
         """Lazily create the backend if not injected."""
         if self._backend is None:
-            self._backend = self._create_backend(self._settings.sandbox_backend)
+            self._backend = self._create_backend(self._settings.stateful_executor_backend)
         return self._backend
 
     def _create_backend(self, backend_name: str) -> "SandboxBackend":
         """Create a backend by name."""
-        if backend_name == "jupyter_local":
+        if backend_name == "local":
             from agentic_cli.tools.sandbox.backends.jupyter_local import JupyterLocalBackend
             return JupyterLocalBackend()
-        if backend_name == "jupyter_docker":
+        if backend_name == "docker":
             from agentic_cli.tools.sandbox.backends.jupyter_docker import JupyterDockerBackend
             return JupyterDockerBackend(self._settings)
-        raise ValueError(f"Unknown sandbox backend: {backend_name!r}")
+        raise ValueError(f"Unknown stateful_executor_backend: {backend_name!r}")
 
     def _get_session_dir(self, session_id: str) -> Path:
         """Get or create the working directory for a session."""
@@ -96,6 +119,7 @@ def execute(
         code: str,
         session_id: str = "default",
         timeout_seconds: int | None = None,
+        inputs: list[str] | None = None,
     ) -> ExecutionResult:
         """Execute code in a sandbox session.
 
@@ -103,6 +127,8 @@ def execute(
             code: Python code to execute.
             session_id: Session identifier (default: "default").
             timeout_seconds: Execution timeout (uses settings default if None).
+            inputs: Optional list of host file paths to stage into
+                inputs/<basename> inside the session before execution.
 
         Returns:
             ExecutionResult with output and metadata.
@@ -139,6 +165,7 @@ def execute(
             session_id=session_id,
             timeout_seconds=timeout_seconds,
             working_dir=session.working_dir,
+            inputs=inputs,
         )
 
         # A failed start (docker down, startup error) must not leave phantom
diff --git a/src/agentic_cli/workflow/permissions/engine.py b/src/agentic_cli/workflow/permissions/engine.py
index 965efa9..a1d3b8b 100644
--- a/src/agentic_cli/workflow/permissions/engine.py
+++ b/src/agentic_cli/workflow/permissions/engine.py
@@ -171,9 +171,14 @@ def _resolve(
     ) -> list[ResolvedCapability]:
         resolved: list[ResolvedCapability] = []
         for cap in capabilities:
-            raw = "*" if cap.target_arg is None else str(args.get(cap.target_arg, ""))
-            target = "*" if cap.target_arg is None else get_matcher(cap.name).canonicalize(raw, self._ctx)
-            resolved.append(ResolvedCapability(cap.name, target))
+            if cap.target_arg is None:
+                resolved.append(ResolvedCapability(cap.name, "*"))
+                continue
+            value = args.get(cap.target_arg, "")
+            matcher = get_matcher(cap.name)
+            items = value if isinstance(value, (list, tuple)) else [value]
+            for item in items:
+                resolved.append(ResolvedCapability(cap.name, matcher.canonicalize(str(item), self._ctx)))
         return resolved
 
     def _evaluate(
diff --git a/src/agentic_cli/workflow/settings.py b/src/agentic_cli/workflow/settings.py
index 0c72af9..a797268 100644
--- a/src/agentic_cli/workflow/settings.py
+++ b/src/agentic_cli/workflow/settings.py
@@ -252,23 +252,17 @@ class WorkflowSettingsMixin:
     )
 
     # Sandbox executor (stateful Jupyter-backed execution)
-    sandbox_execute_enabled: bool = Field(
-        default=False,
-        title="Sandbox Execute Enabled",
+    stateful_executor_backend: Literal["none", "local", "docker"] = Field(
+        default="none",
+        title="Stateful Executor Backend",
         description=(
-            "Enable the stateful sandbox_execute tool. The default 'jupyter_local' "
-            "backend runs Python with host privileges (NOT OS-sandboxed); the "
-            "'jupyter_docker' backend runs in a network-isolated container. Pick "
-            "the backend via sandbox_backend accordingly."
+            "Backend for the stateful sandbox_execute tool. 'none' disables it; "
+            "'docker' runs in a network-isolated container (recommended); 'local' "
+            "runs a Jupyter kernel with host privileges (NOT OS-sandboxed). Future: "
+            "'modal', 'runpod'."
         ),
         json_schema_extra={"ui_order": 121},
     )
-    sandbox_backend: str = Field(
-        default="jupyter_local",
-        title="Sandbox Backend",
-        description="Backend for stateful sandbox execution",
-        json_schema_extra={"ui_order": 122},
-    )
     sandbox_timeout: int = Field(
         default=120,
         title="Sandbox Timeout",
@@ -335,6 +329,12 @@ class WorkflowSettingsMixin:
         description="Seconds to wait for container start + image pull + kernel readiness (docker backend).",
         json_schema_extra={"ui_order": 133},
     )
+    sandbox_outputs_dir: str = Field(
+        default="",
+        title="Sandbox Outputs Dir",
+        description="Shared host dir mounted at /workspace/outputs for FINAL deliverables (default: <workspace_dir>/artifacts).",
+        json_schema_extra={"ui_order": 134},
+    )
 
     @field_validator("sandbox_network")
     @classmethod
diff --git a/tests/examples/__init__.py b/tests/examples/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/examples/test_research_demo_agents.py b/tests/examples/test_research_demo_agents.py
new file mode 100644
index 0000000..534f4e4
--- /dev/null
+++ b/tests/examples/test_research_demo_agents.py
@@ -0,0 +1,16 @@
+"""Config-level tests for the research_demo example agents.
+
+Verifies that AGENT_CONFIGS is wired correctly without making any LLM calls.
+"""
+
+
+def test_data_analyst_wired():
+    from research_demo.agents import AGENT_CONFIGS
+
+    names = {a.name for a in AGENT_CONFIGS}
+    assert "data_analyst" in names
+    analyst = next(a for a in AGENT_CONFIGS if a.name == "data_analyst")
+    tool_names = {t.__name__ for t in analyst.tools}
+    assert "sandbox_execute" in tool_names
+    coord = next(a for a in AGENT_CONFIGS if a.name == "research_coordinator")
+    assert "data_analyst" in coord.sub_agents
diff --git a/tests/permissions/test_engine.py b/tests/permissions/test_engine.py
index 1d5b102..6b1f030 100644
--- a/tests/permissions/test_engine.py
+++ b/tests/permissions/test_engine.py
@@ -282,6 +282,41 @@ async def fake_input(request):
         assert ask_peak == 1  # never two asks in flight simultaneously
 
 
+class TestResolveListTargetArg:
+    def test_list_target_arg_resolves_per_item(self, ctx):
+        from agentic_cli.workflow.permissions.capabilities import ResolvedCapability
+        eng = PermissionEngine(settings=_stub_settings(), workflow=_stub_workflow(), ctx=ctx)
+        resolved = eng._resolve(
+            [Capability("filesystem.read", target_arg="inputs")],
+            {"inputs": ["/data/a.csv", "/data/b.csv"]},
+        )
+        assert all(isinstance(rc, ResolvedCapability) for rc in resolved)
+        targets = sorted(rc.target for rc in resolved)
+        assert len(resolved) == 2
+        assert any(t.endswith("a.csv") for t in targets)
+        assert any(t.endswith("b.csv") for t in targets)
+
+    def test_scalar_target_arg_still_yields_single_resolved(self, ctx):
+        """No regression: a scalar value must still produce exactly one ResolvedCapability."""
+        eng = PermissionEngine(settings=_stub_settings(), workflow=_stub_workflow(), ctx=ctx)
+        resolved = eng._resolve(
+            [Capability("filesystem.read", target_arg="path")],
+            {"path": "/data/x.csv"},
+        )
+        assert len(resolved) == 1
+        assert resolved[0].target.endswith("x.csv")
+
+    def test_none_target_arg_still_yields_star(self, ctx):
+        """No regression: target_arg=None must still produce a single ResolvedCapability with target='*'."""
+        eng = PermissionEngine(settings=_stub_settings(), workflow=_stub_workflow(), ctx=ctx)
+        resolved = eng._resolve(
+            [Capability("python.exec.stateful")],
+            {"code": "x = 1"},
+        )
+        assert len(resolved) == 1
+        assert resolved[0].target == "*"
+
+
 class TestTargetlessAllowAlwaysRegression:
     """Regression: after 'Allow always' on a targetless capability (target_arg=None),
     subsequent calls must not re-prompt.
diff --git a/tests/test_tools.py b/tests/test_tools.py
index 8654b89..0bfa2bb 100644
--- a/tests/test_tools.py
+++ b/tests/test_tools.py
@@ -1482,7 +1482,7 @@ def test_dangerous_tool_executes_directly(self):
         from tests.conftest import MockContext
         from tests.tools.test_sandbox import MockSandboxBackend
 
-        with MockContext(sandbox_execute_enabled=True) as ctx:
+        with MockContext(stateful_executor_backend="local") as ctx:
             backend = MockSandboxBackend(
                 ExecutionResult(success=True, stdout="ok\n", result="1")
             )
diff --git a/tests/tools/test_sandbox.py b/tests/tools/test_sandbox.py
index b72ba64..132778a 100644
--- a/tests/tools/test_sandbox.py
+++ b/tests/tools/test_sandbox.py
@@ -35,13 +35,14 @@ def __init__(self, result: ExecutionResult | None = None) -> None:
         self.execute_calls: list[dict] = []
         self.reset_calls: list[str] = []
 
-    def execute(self, code, session_id, timeout_seconds=120, working_dir=None):
+    def execute(self, code, session_id, timeout_seconds=120, working_dir=None, inputs=None):
         self._sessions.add(session_id)
         self.execute_calls.append({
             "code": code,
             "session_id": session_id,
             "timeout_seconds": timeout_seconds,
             "working_dir": working_dir,
+            "inputs": inputs,
         })
         return self._result
 
@@ -166,7 +167,7 @@ def test_failed_start_does_not_consume_session_slots(self, tmp_path):
         session metadata that fills sandbox_max_sessions with no real session."""
         class _FailingBackend(SandboxBackend):
             backend_name = "failing"
-            def execute(self, code, session_id, timeout_seconds=120, working_dir=None):
+            def execute(self, code, session_id, timeout_seconds=120, working_dir=None, inputs=None):
                 return ExecutionResult(success=False, error="Docker sandbox backend unavailable")
             def reset_session(self, session_id): pass
             def cleanup(self): pass
@@ -262,7 +263,7 @@ def test_working_dir_created(self, tmp_path):
 
 class TestSandboxTools:
     def test_sandbox_execute_success(self, tmp_path):
-        with MockContext(sandbox_execute_enabled=True) as ctx:
+        with MockContext(stateful_executor_backend="local") as ctx:
             from agentic_cli.workflow.service_registry import set_service_registry
 
             backend = MockSandboxBackend(
@@ -283,7 +284,7 @@ def test_sandbox_execute_success(self, tmp_path):
                 mgr.cleanup()
 
     def test_sandbox_execute_no_manager(self, tmp_path):
-        with MockContext(sandbox_execute_enabled=True):
+        with MockContext(stateful_executor_backend="local"):
             from agentic_cli.workflow.service_registry import set_service_registry
 
             token = set_service_registry({})
@@ -301,23 +302,23 @@ def test_sandbox_execute_disabled_by_default(self, tmp_path):
             from agentic_cli.tools.sandbox import sandbox_execute
             result = sandbox_execute("print('hi')")
             assert result["success"] is False
-            assert "enabled" in result["error"].lower()
+            assert "stateful" in result["error"].lower() or "backend" in result["error"].lower()
 
     def test_factory_tool_respects_enabled_flag(self, tmp_path):
         """CRITICAL regression: the workflow uses the factory-bound tool
         (base_manager wires make_sandbox_tool), which must honor the
-        sandbox_execute_enabled opt-in — not just the module-level tool."""
+        stateful_executor_backend opt-in — not just the module-level tool."""
         from agentic_cli.tools.factories import make_sandbox_tool
 
-        with MockContext(sandbox_execute_enabled=False) as ctx:
+        with MockContext(stateful_executor_backend="none") as ctx:
             mgr = SandboxManager(ctx.settings, backend=MockSandboxBackend())
             tool = make_sandbox_tool(mgr)
             r = tool(code="x = 1")
             assert r["success"] is False, "factory tool executed despite disabled flag"
-            assert "enabled" in r["error"].lower()
+            assert "stateful" in r["error"].lower() or "backend" in r["error"].lower()
             mgr.cleanup()
 
-        with MockContext(sandbox_execute_enabled=True) as ctx:
+        with MockContext(stateful_executor_backend="local") as ctx:
             mgr = SandboxManager(ctx.settings, backend=MockSandboxBackend())
             tool = make_sandbox_tool(mgr)
             r = tool(code="x = 1")
@@ -333,7 +334,7 @@ def test_factory_tool_namespaces_default_session_to_conversation(self, tmp_path)
         class _WF:
             active_session_id = "conv-abc"
 
-        with MockContext(sandbox_execute_enabled=True) as ctx:
+        with MockContext(stateful_executor_backend="local") as ctx:
             backend = MockSandboxBackend()
             mgr = SandboxManager(ctx.settings, backend=backend)
             tool = make_sandbox_tool(mgr, _WF())
@@ -347,17 +348,11 @@ class _WF:
             mgr.cleanup()
 
     def test_disabled_message_is_backend_aware(self, tmp_path):
-        """The disabled-tool message must reflect the selected backend: the local
-        backend is host-privileged, but the docker backend is container-isolated
-        — claiming 'host privileges' there would be a false, misleading warning."""
+        """The disabled-tool message must reflect the selected backend."""
         from agentic_cli.tools.sandbox import sandbox_execute
-        with MockContext(sandbox_backend="jupyter_local"):
+        with MockContext(stateful_executor_backend="none"):
             err = sandbox_execute("print('hi')")["error"].lower()
-            assert "host" in err  # honest for the unsandboxed local backend
-        with MockContext(sandbox_backend="jupyter_docker"):
-            err = sandbox_execute("print('hi')")["error"].lower()
-            assert ("container" in err or "isolat" in err)
-            assert "host privilege" not in err  # docker backend is NOT host-privileged
+            assert "stateful_executor_backend" in err
 
     def test_description_makes_no_false_network_claim(self):
         """The tool does NOT block network — the description must not claim it
@@ -379,15 +374,23 @@ def test_capability_distinct_from_execute_python(self):
         from agentic_cli.workflow.permissions.matchers import _cap_matches
 
         reg = get_registry()
-        sandbox_caps = [c.name for c in reg.get("sandbox_execute").capabilities]
+        sandbox_cap_names = [c.name for c in reg.get("sandbox_execute").capabilities]
         exec_caps = [c.name for c in reg.get("execute_python").capabilities]
 
         assert exec_caps == ["python.exec"]
-        assert sandbox_caps == ["python.exec.stateful"]
+        assert "python.exec.stateful" in sandbox_cap_names
         # An execute_python grant (rule 'python.exec') must not cover it.
-        assert _cap_matches("python.exec", sandbox_caps[0]) is False
+        assert _cap_matches("python.exec", "python.exec.stateful") is False
         # A deliberate broad 'python.*' grant still covers both.
-        assert _cap_matches("python.*", sandbox_caps[0]) is True
+        assert _cap_matches("python.*", "python.exec.stateful") is True
+
+    def test_inputs_declares_filesystem_read(self):
+        """sandbox_execute must declare filesystem.read for its inputs arg,
+        so each staged file path is permission-checked identically to read_file."""
+        from agentic_cli.tools.registry import get_registry
+        caps = {(c.name, c.target_arg) for c in get_registry().get("sandbox_execute").capabilities}
+        assert ("python.exec.stateful", None) in caps
+        assert ("filesystem.read", "inputs") in caps
 
 
 
@@ -398,9 +401,9 @@ def test_capability_distinct_from_execute_python(self):
 class TestBackendSelection:
     def test_create_jupyter_docker_backend(self):
         from agentic_cli.tools.sandbox.backends.jupyter_docker import JupyterDockerBackend
-        with MockContext(sandbox_backend="jupyter_docker") as ctx:
+        with MockContext(stateful_executor_backend="docker") as ctx:
             mgr = SandboxManager(ctx.settings)
-            backend = mgr._create_backend("jupyter_docker")
+            backend = mgr._create_backend("docker")
             assert isinstance(backend, JupyterDockerBackend)
 
     def test_unknown_backend_raises(self):
@@ -708,6 +711,34 @@ def test_safe_math_still_works(self, tmp_path):
             backend.cleanup()
 
 
+class TestStageInputs:
+    def test_copies_to_inputs_subdir(self, tmp_path):
+        from agentic_cli.tools.sandbox.manager import stage_inputs
+        src = tmp_path / "sales.csv"; src.write_text("a,b\n1,2\n")
+        sess = tmp_path / "sess"; sess.mkdir()
+        stage_inputs(sess, [str(src)])
+        assert (sess / "inputs" / "sales.csv").read_text() == "a,b\n1,2\n"
+
+    def test_missing_file_raises(self, tmp_path):
+        from agentic_cli.tools.sandbox.manager import stage_inputs
+        sess = tmp_path / "sess"; sess.mkdir()
+        with pytest.raises(ValueError):
+            stage_inputs(sess, [str(tmp_path / "nope.csv")])
+
+    def test_none_session_dir_raises_value_error(self, tmp_path):
+        from agentic_cli.tools.sandbox.manager import stage_inputs
+        with pytest.raises(ValueError, match="session_dir is required"):
+            stage_inputs(None, ["/x"])
+
+    def test_basename_collision_raises(self, tmp_path):
+        from agentic_cli.tools.sandbox.manager import stage_inputs
+        (tmp_path / "a").mkdir(); (tmp_path / "b").mkdir()
+        (tmp_path / "a" / "x.csv").write_text("1"); (tmp_path / "b" / "x.csv").write_text("2")
+        sess = tmp_path / "sess"; sess.mkdir()
+        with pytest.raises(ValueError):
+            stage_inputs(sess, [str(tmp_path / "a" / "x.csv"), str(tmp_path / "b" / "x.csv")])
+
+
 class TestJupyterLocalBackend:
     """Integration tests for JupyterLocalBackend."""
 
@@ -781,3 +812,19 @@ def test_has_session(self, tmp_path):
             assert backend.has_session("test") is False
         finally:
             backend.cleanup()
+
+    def test_outputs_dir_pre_created(self, tmp_path):
+        """outputs/ must be pre-created so open('outputs/x','w') works."""
+        from agentic_cli.tools.sandbox.backends.jupyter_local import JupyterLocalBackend
+
+        backend = JupyterLocalBackend()
+        try:
+            result = backend.execute(
+                "open('outputs/t.txt','w').write('hi'); print('ok')",
+                session_id="test",
+                working_dir=tmp_path,
+            )
+            assert result.success is True
+            assert (tmp_path / "outputs" / "t.txt").exists()
+        finally:
+            backend.cleanup()
diff --git a/tests/tools/test_sandbox_docker_backend.py b/tests/tools/test_sandbox_docker_backend.py
index 061710d..0123d36 100644
--- a/tests/tools/test_sandbox_docker_backend.py
+++ b/tests/tools/test_sandbox_docker_backend.py
@@ -36,7 +36,7 @@ def remove(self, name):
 
 
 def _backend(available=True):
-    ctx = MockContext(sandbox_backend="jupyter_docker").__enter__()
+    ctx = MockContext(stateful_executor_backend="docker").__enter__()
     rt = FakeRuntime()
     detect_fn = lambda: DockerAvailability(available, "docker" if available else "", "test")
     backend = JupyterDockerBackend(ctx.settings, runtime=rt, detect_fn=detect_fn)
@@ -176,7 +176,7 @@ def test_container_runs_as_host_uid_and_dir_not_world_writable(tmp_path):
 
 
 def test_explicit_container_user_overrides_host_uid(tmp_path):
-    ctx = MockContext(sandbox_backend="jupyter_docker",
+    ctx = MockContext(stateful_executor_backend="docker",
                       sandbox_container_user="1234:5678").__enter__()
     rt = FakeRuntime()
     backend = JupyterDockerBackend(
@@ -191,11 +191,105 @@ def test_explicit_container_user_overrides_host_uid(tmp_path):
         ctx.__exit__(None, None, None)
 
 
+def test_execute_stages_inputs_into_session_inputs_dir(tmp_path):
+    backend, rt, ctx = _backend()
+    try:
+        _feed_result(rt)
+        src = tmp_path / "sales.csv"; src.write_text("x\n1\n")
+        wd = tmp_path / "sess"; wd.mkdir()
+        backend.execute("print(1)", "s1", timeout_seconds=5, working_dir=wd, inputs=[str(src)])
+        assert (wd / "inputs" / "sales.csv").read_text() == "x\n1\n"
+    finally:
+        ctx.__exit__(None, None, None)
+
+
+def test_build_spec_has_no_nested_outputs_mount(tmp_path):
+    """outputs/ must NOT be a separate bind mount nested inside /workspace.
+    That pattern causes macOS Docker Desktop ACL/xattr issues that make the
+    session dir un-removable at teardown."""
+    backend, rt, ctx = _backend()
+    try:
+        _feed_result(rt)
+        backend.execute("print(1)", "s1", timeout_seconds=5, working_dir=tmp_path)
+        spec = rt.started[0]
+        assert not any(m.container == "/workspace/outputs" for m in spec.mounts), (
+            "outputs/ must not be a nested bind mount inside /workspace"
+        )
+    finally:
+        ctx.__exit__(None, None, None)
+
+
+def test_outputs_mountpoint_pre_created_as_host_user(tmp_path):
+    """Docker must not create /workspace/outputs as root.
+    The backend must pre-create <working_dir>/outputs before runtime.start() so
+    the mount-point directory is owned by the host user (not root), which allows
+    pytest teardown to remove it and keeps the session dir clean."""
+    backend, rt, ctx = _backend()
+    try:
+        _feed_result(rt)
+        wd = tmp_path / "sess"
+        wd.mkdir()
+        backend.execute("x = 1", "s1", timeout_seconds=5, working_dir=wd)
+        assert (wd / "outputs").exists(), "outputs/ mount-point must exist after execute"
+        assert (wd / "outputs").is_dir(), "outputs/ must be a directory, not a file"
+    finally:
+        ctx.__exit__(None, None, None)
+
+
+def test_execute_copies_outputs_to_shared_dir(tmp_path):
+    """Files written to <working_dir>/outputs/ by the kernel must be copied to
+    the shared outputs dir after a successful execute, and the shared path (not
+    the session path) must appear in ExecutionResult.artifacts."""
+    backend, rt, ctx = _backend()
+    try:
+        _feed_result(rt)
+        wd = tmp_path / "sess"
+        wd.mkdir()
+        # Simulate what the kernel would write inside the container
+        (wd / "outputs").mkdir()
+        (wd / "outputs" / "result.csv").write_text("a,b\n1,2\n")
+
+        result = backend.execute("print(1)", "s1", timeout_seconds=5, working_dir=wd)
+
+        shared_dir = backend._outputs_dir()
+        shared_file = shared_dir / "result.csv"
+        assert shared_file.exists(), "file must have been copied to shared outputs dir"
+        assert shared_file.read_text() == "a,b\n1,2\n"
+        assert str(shared_file) in result.artifacts, (
+            "shared path must appear in ExecutionResult.artifacts"
+        )
+    finally:
+        ctx.__exit__(None, None, None)
+
+
+def test_data_mount_points_pre_created_host_owned(tmp_path):
+    """Data-mount target dirs under working_dir/data/ must be pre-created by the
+    host before runtime.start() so Docker does not create them as root."""
+    some_dir = tmp_path / "mydata"
+    some_dir.mkdir()
+    ctx = MockContext(stateful_executor_backend="docker",
+                      sandbox_data_mounts=[f"{some_dir}:samples"]).__enter__()
+    rt = FakeRuntime()
+    backend = JupyterDockerBackend(
+        ctx.settings, runtime=rt,
+        detect_fn=lambda: DockerAvailability(True, "docker", "test"),
+    )
+    try:
+        _feed_result(rt)
+        wd = tmp_path / "sess"
+        wd.mkdir()
+        backend.execute("x = 1", "s1", timeout_seconds=5, working_dir=wd)
+        assert (wd / "data" / "samples").exists(), "data/samples mount point must be pre-created"
+        assert (wd / "data" / "samples").is_dir(), "data/samples must be a directory"
+    finally:
+        ctx.__exit__(None, None, None)
+
+
 def test_data_mount_name_cannot_escape_workspace(tmp_path):
     """A hostile data-mount name (traversal) must not remap the mount point
     outside /workspace/data/ inside the container."""
     import posixpath
-    ctx = MockContext(sandbox_backend="jupyter_docker",
+    ctx = MockContext(stateful_executor_backend="docker",
                       sandbox_data_mounts=[f"{tmp_path}:../../etc"]).__enter__()
     rt = FakeRuntime()
     backend = JupyterDockerBackend(
diff --git a/tests/tools/test_sandbox_docker_integration.py b/tests/tools/test_sandbox_docker_integration.py
index 097a0e6..89110d8 100644
--- a/tests/tools/test_sandbox_docker_integration.py
+++ b/tests/tools/test_sandbox_docker_integration.py
@@ -63,7 +63,7 @@ def remove(self, name: str) -> None:
 
 @pytest.fixture
 def backend(tmp_path):
-    with MockContext(sandbox_backend="jupyter_docker", sandbox_start_timeout=60) as ctx:
+    with MockContext(stateful_executor_backend="docker", sandbox_start_timeout=60) as ctx:
         b = JupyterDockerBackend(
             ctx.settings,
             runtime=LocalDriverRuntime(),
@@ -115,6 +115,16 @@ def test_user_code_cannot_forge_protocol_via_fd1(backend, tmp_path):
     backend.cleanup()
 
 
+def test_inputs_are_loadable_by_relative_path(backend, tmp_path):
+    """A staged input is available at inputs/<name> and loadable relatively."""
+    src = tmp_path.parent / "iris_like.csv"; src.write_text("a,b\n1,2\n3,4\n")
+    r = backend.execute("print(open('inputs/iris_like.csv').read().strip())",
+                        "s1", timeout_seconds=30, working_dir=tmp_path, inputs=[str(src)])
+    assert r.success is True, r.error
+    assert "1,2" in r.stdout
+    backend.cleanup()
+
+
 def test_interrupt_preserves_session_state(backend, tmp_path):
     """A runaway cell is aborted by the host's cooperative interrupt, but the
     session (kernel + prior state) survives and the next request runs cleanly.
diff --git a/tests/tools/test_sandbox_docker_live.py b/tests/tools/test_sandbox_docker_live.py
index fad9ed9..7d6e70f 100644
--- a/tests/tools/test_sandbox_docker_live.py
+++ b/tests/tools/test_sandbox_docker_live.py
@@ -17,6 +17,7 @@
 
 from agentic_cli.tools.sandbox.backends.detect import detect_docker, docker_available
 from agentic_cli.tools.sandbox.backends.jupyter_docker import JupyterDockerBackend
+from agentic_cli.tools.sandbox.manager import SandboxManager
 from tests.conftest import MockContext
 
 # Every test in this module is a docker test (selectable via -m docker).
@@ -37,7 +38,7 @@ def test_docker_runtime_present_when_required():
 
 @pytest.fixture
 def backend(tmp_path):
-    with MockContext(sandbox_backend="jupyter_docker") as ctx:
+    with MockContext(stateful_executor_backend="docker") as ctx:
         b = JupyterDockerBackend(ctx.settings)
         try:
             yield b
@@ -178,6 +179,33 @@ def test_sessions_are_isolated(backend, tmp_path):
     assert "NO_FILE" in rb.stdout
 
 
+# --------------------------------------------------------------------------
+# Stateful analysis: multi-call persistence
+# --------------------------------------------------------------------------
+
+@_requires_docker
+def test_analyst_flow_stage_analyze_output(tmp_path):
+    """Stage an input, run multi-step stateful analysis, write a figure to
+    outputs/, and confirm it appears on the shared host dir."""
+    from agentic_cli.tools.sandbox.manager import SandboxManager
+    outdir = tmp_path / "artifacts"
+    src = tmp_path / "rows.csv"; src.write_text("g,v\na,1\na,3\nb,10\n")
+    with MockContext(stateful_executor_backend="docker",
+                     sandbox_outputs_dir=str(outdir)) as ctx:
+        mgr = SandboxManager(ctx.settings)
+        try:
+            r1 = mgr.execute("import pandas as pd\ndf = pd.read_csv('inputs/rows.csv')\nprint(df.groupby('g').v.mean().to_dict())",
+                             session_id="an", inputs=[str(src)])
+            assert r1.success is True, r1.error
+            assert "'a': 2.0" in r1.stdout or '"a": 2.0' in r1.stdout
+            r2 = mgr.execute("df.groupby('g').v.mean().to_csv('outputs/means.csv'); print('wrote')",
+                             session_id="an")  # same session: df persists
+            assert r2.success is True, r2.error
+            assert (outdir / "means.csv").exists()
+        finally:
+            mgr.cleanup()
+
+
 # --------------------------------------------------------------------------
 # Cooperative interrupt: a runaway cell is aborted but the session survives
 # --------------------------------------------------------------------------
@@ -207,7 +235,7 @@ def test_interrupt_preserves_session_state(backend, tmp_path):
 def test_memory_cap_oom_kills(tmp_path):
     """A single allocation far past --memory (swap disabled) is OOM-killed;
     the backend surfaces failure rather than a clean success."""
-    with MockContext(sandbox_backend="jupyter_docker", sandbox_memory_mb=256) as ctx:
+    with MockContext(stateful_executor_backend="docker", sandbox_memory_mb=256) as ctx:
         b = JupyterDockerBackend(ctx.settings)
         try:
             r = b.execute("x = bytearray(1024 * 1024 * 1024)  # 1 GiB vs 256 MiB cap",
@@ -220,7 +248,7 @@ def test_memory_cap_oom_kills(tmp_path):
 @_requires_docker
 def test_pids_limit_caps_thread_bomb(tmp_path):
     """--pids-limit bounds the number of tasks; a thread bomb hits it."""
-    with MockContext(sandbox_backend="jupyter_docker", sandbox_pids_limit=128) as ctx:
+    with MockContext(stateful_executor_backend="docker", sandbox_pids_limit=128) as ctx:
         b = JupyterDockerBackend(ctx.settings)
         try:
             code = ("import threading, time\n"
@@ -242,10 +270,25 @@ def test_pids_limit_caps_thread_bomb(tmp_path):
 # Lifecycle: no leaked containers after cleanup
 # --------------------------------------------------------------------------
 
+@_requires_docker
+def test_outputs_dir_persists_to_shared_host_dir(tmp_path):
+    outdir = tmp_path / "shared_out"
+    with MockContext(stateful_executor_backend="docker", sandbox_outputs_dir=str(outdir)) as ctx:
+        b = JupyterDockerBackend(ctx.settings)
+        try:
+            r = b.execute("open('outputs/final.txt','w').write('done'); print('ok')",
+                          "out", timeout_seconds=60, working_dir=tmp_path)
+            assert r.success is True, r.error
+            assert (outdir / "final.txt").read_text() == "done"
+            assert any(a.endswith("final.txt") for a in r.artifacts)
+        finally:
+            b.cleanup()
+
+
 @_requires_docker
 def test_no_orphaned_containers_after_cleanup(tmp_path):
     runtime = detect_docker().runtime or "docker"
-    with MockContext(sandbox_backend="jupyter_docker") as ctx:
+    with MockContext(stateful_executor_backend="docker") as ctx:
         b = JupyterDockerBackend(ctx.settings)
         b.execute("x = 1", "orphan1", timeout_seconds=60, working_dir=tmp_path)
         b.execute("y = 2", "orphan2", timeout_seconds=60, working_dir=tmp_path)
diff --git a/tests/tools/test_sandbox_settings.py b/tests/tools/test_sandbox_settings.py
index 64729ca..c055e34 100644
--- a/tests/tools/test_sandbox_settings.py
+++ b/tests/tools/test_sandbox_settings.py
@@ -16,9 +16,17 @@ def test_docker_sandbox_defaults():
     assert s.sandbox_container_user == ""
     assert s.sandbox_data_mounts == []
     assert s.sandbox_start_timeout == 180
-    # unchanged safety defaults
-    assert s.sandbox_backend == "jupyter_local"
-    assert s.sandbox_execute_enabled is False
+    # unified backend config (replaces the old two-field enable+backend pattern)
+    assert s.stateful_executor_backend == "none"
+
+
+def test_stateful_executor_backend_default_and_values():
+    from pydantic import ValidationError
+    assert BaseSettings().stateful_executor_backend == "none"
+    assert BaseSettings(stateful_executor_backend="docker").stateful_executor_backend == "docker"
+    assert BaseSettings(stateful_executor_backend="local").stateful_executor_backend == "local"
+    with pytest.raises(ValidationError):
+        BaseSettings(stateful_executor_backend="bogus")
 
 
 def test_sandbox_network_must_be_none():