feat(sandbox): stateful executor toolset + research_demo data_analyst

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",
     ),
 ]

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

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"))

src/agentic_cli/tools/factories.py

@@ -365,24 +365,27 @@ 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.
 
         Args:
             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,

src/agentic_cli/tools/sandbox/__init__.py

@@ -18,41 +18,47 @@
     # 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.
 
     Args:
         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,

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.
 

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)

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,13 +113,24 @@ 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
         valid, error = kernel_exec.validate_code(code)
         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)