Upload folder using huggingface_hub

server/gradio_ui.py

@@ -14,10 +14,70 @@ from typing import Any, Optional
 import gradio as gr
 
 
+TOOL_SPECS: dict[str, dict[str, Any]] = {
+    "bash": {
+        "icon": "🖥️",
+        "description": "Execute a bash command inside the E2B sandbox.",
+        "signature": "bash(command: str, timeout: float = 30)",
+    },
+    "read": {
+        "icon": "📖",
+        "description": "Read a file from the sandbox. Optional `offset` and `limit` slice the lines.",
+        "signature": "read(file_path: str, offset?: int, limit?: int)",
+    },
+    "write": {
+        "icon": "✏️",
+        "description": "Create or overwrite a file with `content`.",
+        "signature": "write(file_path: str, content: str)",
+    },
+    "edit": {
+        "icon": "🔧",
+        "description": "Find-and-replace `old_string` with `new_string` in a file. Use `replace_all` for every occurrence.",
+        "signature": "edit(file_path, old_string, new_string, replace_all=False)",
+    },
+    "multi_edit": {
+        "icon": "🔁",
+        "description": "Apply a list of {old_string, new_string, replace_all?} edits sequentially in one file.",
+        "signature": "multi_edit(file_path, edits: list[dict])",
+    },
+    "glob": {
+        "icon": "🔍",
+        "description": "Glob for files matching a shell-style pattern, optionally rooted at `path`.",
+        "signature": "glob(pattern: str, path?: str)",
+    },
+    "grep": {
+        "icon": "🧵",
+        "description": "Regex search across files. `include` is an optional file glob filter.",
+        "signature": "grep(pattern: str, path?: str, include?: str)",
+    },
+    "ls": {
+        "icon": "📂",
+        "description": "List a directory. `ignore` is a comma-separated list of patterns to skip.",
+        "signature": "ls(path: str = '.', ignore?: list[str])",
+    },
+    "todo_write": {
+        "icon": "🗒️",
+        "description": "Replace the agent's todo list. Each item: {id, content, status, priority}. Only one `in_progress` at a time.",
+        "signature": "todo_write(todos: list[dict])",
+    },
+    "submit_solution": {
+        "icon": "🚀",
+        "description": "Run the configured verify commands and finalise the episode with a reward.",
+        "signature": "submit_solution()",
+    },
+}
+
+TOOL_CHOICES = [(f"{spec['icon']}  {name}", name) for name, spec in TOOL_SPECS.items()]
+
+
 def _lines(value: str) -> list[str]:
     return [line.strip() for line in value.splitlines() if line.strip()]
 
 
+def _csv_list(value: str) -> list[str]:
+    return [item.strip() for item in value.split(",") if item.strip()]
+
+
 def _extract_tool_text(result: dict[str, Any]) -> str:
     obs = result.get("observation", {})
     if not isinstance(obs, dict):
@@ -33,6 +93,46 @@ def _extract_tool_text(result: dict[str, Any]) -> str:
     return str(tool_result.get("data", ""))
 
 
+def _extract_tool_error(result: dict[str, Any]) -> bool:
+    obs = result.get("observation", {})
+    if not isinstance(obs, dict):
+        return False
+    tool_result = obs.get("result", {})
+    if not isinstance(tool_result, dict):
+        return False
+    return bool(tool_result.get("is_error", False))
+
+
+def _format_status(state: dict[str, Any]) -> str:
+    if not state:
+        return "**No active session.** Configure setup/verify and click *Reset sandbox*."
+    sandbox_id = state.get("sandbox_id") or "—"
+    step_count = state.get("step_count", 0)
+    submitted = state.get("submitted", False)
+    last_reward = state.get("last_reward")
+    reward_str = "—" if last_reward is None else f"{last_reward}"
+    submitted_str = "✅ submitted" if submitted else "⏳ in progress"
+    return (
+        f"**Session active** · sandbox `{sandbox_id}` · "
+        f"steps `{step_count}` · {submitted_str} · reward `{reward_str}`"
+    )
+
+
+def _format_history(state: dict[str, Any]) -> list[list[str]]:
+    history = state.get("tool_history", []) or []
+    rows: list[list[str]] = []
+    for entry in history[-25:][::-1]:
+        if not isinstance(entry, dict):
+            continue
+        tool = str(entry.get("tool", ""))
+        ok = "✅" if entry.get("ok") else "❌"
+        output = str(entry.get("output") or entry.get("error") or "")
+        if len(output) > 140:
+            output = output[:137] + "…"
+        rows.append([ok, tool, output])
+    return rows
+
+
 def coding_tools_ui_builder(
     web_manager,
     action_fields: list[dict[str, Any]],
@@ -47,171 +147,376 @@ def coding_tools_ui_builder(
         except RuntimeError:
             return {}
 
-    with gr.Blocks(title=f"{title} - Coding Tools") as demo:
-        gr.Markdown(f"# {title}")
-        gr.Markdown("E2B-backed coding tools environment with SETA-style tool surface.")
+    with gr.Blocks(
+        title=f"{title} - Coding Tools",
+        theme=gr.themes.Soft(primary_hue="blue", neutral_hue="slate"),
+    ) as demo:
+        gr.Markdown(f"# 🧰 {title}")
+        gr.Markdown(
+            "E2B-backed coding environment exposing a SETA-style tool surface. "
+            "Pick a tool, fill the inputs, and inspect the output, state, and call history."
+        )
+
+        status_md = gr.Markdown(_format_status({}))
 
         with gr.Row():
-            with gr.Column(scale=1):
-                gr.Markdown("### Session")
-                setup_input = gr.Textbox(
-                    label="Setup commands",
-                    value="mkdir -p /home/user/work",
-                    lines=4,
+            # ───────── Left: Session controls ─────────
+            with gr.Column(scale=1, min_width=320):
+                with gr.Accordion("Session controls", open=True):
+                    setup_input = gr.Textbox(
+                        label="Setup commands (one per line)",
+                        value="mkdir -p /home/user/work",
+                        lines=3,
+                    )
+                    verify_input = gr.Textbox(
+                        label="Verify commands (one per line)",
+                        value="pytest -q",
+                        lines=2,
+                    )
+                    with gr.Row():
+                        reset_btn = gr.Button("🔄 Reset sandbox", variant="primary")
+                        close_btn = gr.Button("🛑 Close", variant="secondary")
+
+            # ───────── Middle: Tool selection + dynamic inputs ─────────
+            with gr.Column(scale=2, min_width=420):
+                gr.Markdown("### Tool")
+                tool_dropdown = gr.Dropdown(
+                    choices=TOOL_CHOICES,
+                    value="bash",
+                    label="Select a tool",
+                    interactive=True,
                 )
-                verify_input = gr.Textbox(
-                    label="Verify commands",
-                    value="pytest -q",
-                    lines=3,
+                tool_help = gr.Markdown(
+                    f"**{TOOL_SPECS['bash']['signature']}**  \n{TOOL_SPECS['bash']['description']}"
                 )
-                reset_btn = gr.Button("Reset sandbox", variant="primary")
-                close_btn = gr.Button("Stop / Close session", variant="secondary")
-
-                gr.Markdown("### Bash")
-                bash_input = gr.Textbox(label="Command", value="pwd && ls -la", lines=3)
-                timeout_input = gr.Number(label="Timeout (seconds)", value=30)
-                bash_btn = gr.Button("Run bash", variant="primary")
-
-                gr.Markdown("### Submit")
-                submit_btn = gr.Button("Submit solution", variant="primary")
-
-            with gr.Column(scale=1):
-                gr.Markdown("### File tools")
-                file_path = gr.Textbox(label="File path", value="/home/user/work/main.py")
-                read_btn = gr.Button("Read")
-                write_content = gr.Textbox(label="Write content", lines=5)
-                write_btn = gr.Button("Write")
-                old_string = gr.Textbox(label="Old string", lines=2)
-                new_string = gr.Textbox(label="New string", lines=2)
-                replace_all = gr.Checkbox(label="Replace all", value=False)
-                edit_btn = gr.Button("Edit")
-
-                gr.Markdown("### Search tools")
-                glob_pattern = gr.Textbox(label="Glob pattern", value="**/*.py")
-                glob_path = gr.Textbox(label="Glob path", value="/home/user/work")
-                glob_btn = gr.Button("Glob")
-                grep_pattern = gr.Textbox(label="Grep pattern", value="TODO")
-                grep_path = gr.Textbox(label="Grep path", value="/home/user/work")
-                grep_include = gr.Textbox(label="Grep include (optional)")
-                grep_btn = gr.Button("Grep")
-
-            with gr.Column(scale=1):
-                gr.Markdown("### Directory and Todo")
-                ls_path = gr.Textbox(label="LS path", value="/home/user/work")
-                ls_ignore = gr.Textbox(label="LS ignore (comma-separated)")
-                ls_btn = gr.Button("LS")
-                todos_json = gr.Code(
-                    label="Todos JSON",
-                    language="json",
-                    value='[{"id":"1","content":"Inspect files","status":"in_progress","priority":"high"}]',
-                    lines=8,
+
+                # One Group per tool. Visibility toggled by tool_dropdown.
+                with gr.Group(visible=True) as g_bash:
+                    bash_command = gr.Textbox(
+                        label="command", value="pwd && ls -la", lines=3
+                    )
+                    bash_timeout = gr.Number(label="timeout (seconds)", value=30)
+
+                with gr.Group(visible=False) as g_read:
+                    read_path = gr.Textbox(
+                        label="file_path", value="/home/user/work/main.py"
+                    )
+                    with gr.Row():
+                        read_offset = gr.Number(label="offset (optional)", value=None)
+                        read_limit = gr.Number(label="limit (optional)", value=None)
+
+                with gr.Group(visible=False) as g_write:
+                    write_path = gr.Textbox(
+                        label="file_path", value="/home/user/work/main.py"
+                    )
+                    write_content = gr.Textbox(label="content", lines=8)
+
+                with gr.Group(visible=False) as g_edit:
+                    edit_path = gr.Textbox(
+                        label="file_path", value="/home/user/work/main.py"
+                    )
+                    edit_old = gr.Textbox(label="old_string", lines=3)
+                    edit_new = gr.Textbox(label="new_string", lines=3)
+                    edit_replace_all = gr.Checkbox(label="replace_all", value=False)
+
+                with gr.Group(visible=False) as g_multi_edit:
+                    multi_edit_path = gr.Textbox(
+                        label="file_path", value="/home/user/work/main.py"
+                    )
+                    multi_edit_json = gr.Code(
+                        label="edits (JSON array)",
+                        language="json",
+                        value=(
+                            '[\n'
+                            '  {"old_string": "TODO", "new_string": "DONE", "replace_all": false}\n'
+                            ']'
+                        ),
+                        lines=8,
+                    )
+
+                with gr.Group(visible=False) as g_glob:
+                    glob_pattern = gr.Textbox(label="pattern", value="**/*.py")
+                    glob_path = gr.Textbox(
+                        label="path (optional)", value="/home/user/work"
+                    )
+
+                with gr.Group(visible=False) as g_grep:
+                    grep_pattern = gr.Textbox(label="pattern", value="TODO")
+                    grep_path = gr.Textbox(
+                        label="path (optional)", value="/home/user/work"
+                    )
+                    grep_include = gr.Textbox(
+                        label="include (optional file glob)", value=""
+                    )
+
+                with gr.Group(visible=False) as g_ls:
+                    ls_path = gr.Textbox(label="path", value="/home/user/work")
+                    ls_ignore = gr.Textbox(
+                        label="ignore (comma-separated globs)", value=""
+                    )
+
+                with gr.Group(visible=False) as g_todo:
+                    todo_json = gr.Code(
+                        label="todos (JSON array)",
+                        language="json",
+                        value=(
+                            '[\n'
+                            '  {"id":"1","content":"Inspect files",'
+                            '"status":"in_progress","priority":"high"}\n'
+                            ']'
+                        ),
+                        lines=8,
+                    )
+
+                with gr.Group(visible=False) as g_submit:
+                    gr.Markdown(
+                        "_Submit runs the configured verify commands and ends the "
+                        "episode with a reward._"
+                    )
+
+                run_btn = gr.Button("▶️  Run tool", variant="primary", size="lg")
+
+            # ───────── Right: Output + State ─────────
+            with gr.Column(scale=2, min_width=420):
+                gr.Markdown("### Output")
+                output_status = gr.Markdown("_Awaiting first call._")
+                output_view = gr.Code(
+                    label="Tool output (text)",
+                    language=None,
+                    value="",
+                    lines=14,
                 )
-                todo_btn = gr.Button("Update todos")
-                state_view = gr.JSON(label="Session state", value={})
+                with gr.Accordion("Raw step response", open=False):
+                    raw_response = gr.Code(
+                        label="raw JSON",
+                        language="json",
+                        value="",
+                        lines=14,
+                    )
+
+                gr.Markdown("### State")
+                with gr.Tabs():
+                    with gr.Tab("Summary"):
+                        state_summary = gr.Markdown(_format_status({}))
+                        history_table = gr.Dataframe(
+                            headers=["✓", "tool", "output"],
+                            datatype=["str", "str", "str"],
+                            row_count=(0, "dynamic"),
+                            wrap=True,
+                            label="Tool history (latest 25)",
+                        )
+                    with gr.Tab("JSON"):
+                        state_json = gr.Code(
+                            label="full session state",
+                            language="json",
+                            value="{}",
+                            lines=18,
+                        )
+
+        # ───────── Tool dropdown → input form visibility ─────────
+        TOOL_GROUPS = {
+            "bash": g_bash,
+            "read": g_read,
+            "write": g_write,
+            "edit": g_edit,
+            "multi_edit": g_multi_edit,
+            "glob": g_glob,
+            "grep": g_grep,
+            "ls": g_ls,
+            "todo_write": g_todo,
+            "submit_solution": g_submit,
+        }
+        group_components = list(TOOL_GROUPS.values())
 
-        output = gr.Textbox(label="Tool output", lines=12, interactive=False)
-        status = gr.Textbox(label="Status", value="Ready.", interactive=False)
+        def on_tool_change(tool: str):
+            spec = TOOL_SPECS.get(tool, {})
+            help_md = (
+                f"**{spec.get('signature', tool)}**  \n{spec.get('description', '')}"
+            )
+            updates = [gr.update(visible=(name == tool)) for name in TOOL_GROUPS]
+            return [help_md, *updates]
+
+        tool_dropdown.change(
+            on_tool_change, inputs=[tool_dropdown], outputs=[tool_help, *group_components]
+        )
+
+        # ───────── Result rendering helper ─────────
+        def render_result(tool: str, raw: dict[str, Any]) -> tuple[str, str, str, str, str, list[list[str]]]:
+            text = _extract_tool_text(raw)
+            is_error = _extract_tool_error(raw) or text.startswith("ERROR:") or text.startswith("Error:")
+            badge = "❌ error" if is_error else "✅ ok"
+            status_line = f"**{tool}** — {badge}"
+            state = state_payload()
+            return (
+                status_line,                       # output_status
+                text,                              # output_view
+                json.dumps(raw, indent=2),         # raw_response
+                _format_status(state),             # state_summary (top + summary panel — same content)
+                json.dumps(state, indent=2, default=str),  # state_json
+                _format_history(state),            # history_table
+            )
 
+        # ───────── Session handlers ─────────
         async def on_reset(setup_text: str, verify_text: str):
-            result = await web_manager.reset_environment(
+            raw = await web_manager.reset_environment(
                 {"setup": _lines(setup_text), "verify": _lines(verify_text)}
             )
             state = state_payload()
-            return state, json.dumps(result, indent=2), "Sandbox reset."
+            obs = raw.get("observation", {}) if isinstance(raw, dict) else {}
+            meta = obs.get("metadata", {}) if isinstance(obs, dict) else {}
+            err = meta.get("error") if isinstance(meta, dict) else None
+            sandbox_id = state.get("sandbox_id") or meta.get("sandbox_id") or "—"
+            if err:
+                status_line = f"**reset** — ❌ {err}"
+                text = err
+            else:
+                status_line = f"**reset** — ✅ sandbox `{sandbox_id}` ready"
+                text = f"Sandbox ready: {sandbox_id}"
+            return (
+                _format_status(state),
+                status_line,
+                text,
+                json.dumps(raw, indent=2),
+                _format_status(state),
+                json.dumps(state, indent=2, default=str),
+                _format_history(state),
+            )
 
         async def on_close():
             await web_manager._run_sync_in_thread_pool(web_manager.env.close)
-            return {}, "Session closed.", "Session closed."
-
-        async def on_bash(command: str, timeout: float):
-            result = await web_manager.step_environment(
-                {
-                    "tool_name": "bash",
-                    "arguments": {"command": command, "timeout": timeout},
-                }
-            )
-            return state_payload(), _extract_tool_text(result), "bash executed."
-
-        async def on_read(path: str):
-            result = await web_manager.step_environment(
-                {"tool_name": "read", "arguments": {"file_path": path}}
+            return (
+                _format_status({}),
+                "**close** — 🛑 session closed",
+                "Session closed.",
+                "{}",
+                _format_status({}),
+                "{}",
+                [],
             )
-            return state_payload(), _extract_tool_text(result), "read executed."
-
-        async def on_write(path: str, content: str):
-            result = await web_manager.step_environment(
-                {
-                    "tool_name": "write",
-                    "arguments": {"file_path": path, "content": content},
-                }
-            )
-            return state_payload(), _extract_tool_text(result), "write executed."
-
-        async def on_edit(path: str, old: str, new: str, all_matches: bool):
-            result = await web_manager.step_environment(
-                {
-                    "tool_name": "edit",
-                    "arguments": {
-                        "file_path": path,
-                        "old_string": old,
-                        "new_string": new,
-                        "replace_all": all_matches,
-                    },
-                }
-            )
-            return state_payload(), _extract_tool_text(result), "edit executed."
 
-        async def on_glob(pattern: str, path: str):
-            result = await web_manager.step_environment(
-                {"tool_name": "glob", "arguments": {"pattern": pattern, "path": path}}
-            )
-            return state_payload(), _extract_tool_text(result), "glob executed."
-
-        async def on_grep(pattern: str, path: str, include: str):
-            args: dict[str, Any] = {"pattern": pattern, "path": path}
-            if include.strip():
-                args["include"] = include
-            result = await web_manager.step_environment(
-                {"tool_name": "grep", "arguments": args}
-            )
-            return state_payload(), _extract_tool_text(result), "grep executed."
+        # ───────── Universal run handler ─────────
+        async def on_run(
+            tool: str,
+            # bash
+            bash_command: str, bash_timeout: float,
+            # read
+            read_path: str, read_offset: float | None, read_limit: float | None,
+            # write
+            write_path: str, write_content: str,
+            # edit
+            edit_path: str, edit_old: str, edit_new: str, edit_replace_all: bool,
+            # multi_edit
+            multi_edit_path: str, multi_edit_json: str,
+            # glob
+            glob_pattern: str, glob_path: str,
+            # grep
+            grep_pattern: str, grep_path: str, grep_include: str,
+            # ls
+            ls_path: str, ls_ignore: str,
+            # todo_write
+            todo_json: str,
+        ):
+            try:
+                if tool == "bash":
+                    args = {
+                        "command": bash_command,
+                        "timeout": float(bash_timeout) if bash_timeout else 30,
+                    }
+                elif tool == "read":
+                    args = {"file_path": read_path}
+                    if read_offset not in (None, "", 0):
+                        args["offset"] = int(read_offset)
+                    if read_limit not in (None, ""):
+                        args["limit"] = int(read_limit)
+                elif tool == "write":
+                    args = {"file_path": write_path, "content": write_content}
+                elif tool == "edit":
+                    args = {
+                        "file_path": edit_path,
+                        "old_string": edit_old,
+                        "new_string": edit_new,
+                        "replace_all": bool(edit_replace_all),
+                    }
+                elif tool == "multi_edit":
+                    args = {
+                        "file_path": multi_edit_path,
+                        "edits": json.loads(multi_edit_json),
+                    }
+                elif tool == "glob":
+                    args = {"pattern": glob_pattern}
+                    if glob_path.strip():
+                        args["path"] = glob_path
+                elif tool == "grep":
+                    args = {"pattern": grep_pattern}
+                    if grep_path.strip():
+                        args["path"] = grep_path
+                    if grep_include.strip():
+                        args["include"] = grep_include
+                elif tool == "ls":
+                    args = {"path": ls_path or "."}
+                    ignore = _csv_list(ls_ignore)
+                    if ignore:
+                        args["ignore"] = ignore
+                elif tool == "todo_write":
+                    args = {"todos": json.loads(todo_json)}
+                elif tool == "submit_solution":
+                    args = {}
+                else:
+                    return (
+                        _format_status(state_payload()),
+                        f"**{tool}** — ❌ unknown tool",
+                        f"unknown tool: {tool}",
+                        "{}",
+                        _format_status(state_payload()),
+                        "{}",
+                        [],
+                    )
+            except (json.JSONDecodeError, ValueError) as exc:
+                state = state_payload()
+                return (
+                    _format_status(state),
+                    f"**{tool}** — ❌ bad input: {exc}",
+                    f"Input parse error: {exc}",
+                    "{}",
+                    _format_status(state),
+                    json.dumps(state, indent=2, default=str),
+                    _format_history(state),
+                )
 
-        async def on_ls(path: str, ignore_csv: str):
-            ignore = [item.strip() for item in ignore_csv.split(",") if item.strip()]
-            result = await web_manager.step_environment(
-                {"tool_name": "ls", "arguments": {"path": path, "ignore": ignore or None}}
+            raw = await web_manager.step_environment(
+                {"tool_name": tool, "arguments": args}
             )
-            return state_payload(), _extract_tool_text(result), "ls executed."
+            state = state_payload()
+            rendered = render_result(tool, raw)
+            return (_format_status(state), *rendered)
 
-        async def on_todo(todo_raw: str):
-            todos = json.loads(todo_raw)
-            result = await web_manager.step_environment(
-                {"tool_name": "todo_write", "arguments": {"todos": todos}}
-            )
-            return state_payload(), _extract_tool_text(result), "todo_write executed."
+        # ───────── Wire up events ─────────
+        all_inputs = [
+            tool_dropdown,
+            bash_command, bash_timeout,
+            read_path, read_offset, read_limit,
+            write_path, write_content,
+            edit_path, edit_old, edit_new, edit_replace_all,
+            multi_edit_path, multi_edit_json,
+            glob_pattern, glob_path,
+            grep_pattern, grep_path, grep_include,
+            ls_path, ls_ignore,
+            todo_json,
+        ]
+        all_outputs = [
+            status_md,
+            output_status,
+            output_view,
+            raw_response,
+            state_summary,
+            state_json,
+            history_table,
+        ]
 
-        async def on_submit():
-            result = await web_manager.step_environment(
-                {"tool_name": "submit_solution", "arguments": {}}
-            )
-            return state_payload(), _extract_tool_text(result), "submit_solution executed."
-
-        reset_btn.click(on_reset, [setup_input, verify_input], [state_view, output, status])
-        close_btn.click(on_close, outputs=[state_view, output, status])
-        bash_btn.click(on_bash, [bash_input, timeout_input], [state_view, output, status])
-        read_btn.click(on_read, [file_path], [state_view, output, status])
-        write_btn.click(on_write, [file_path, write_content], [state_view, output, status])
-        edit_btn.click(
-            on_edit,
-            [file_path, old_string, new_string, replace_all],
-            [state_view, output, status],
-        )
-        glob_btn.click(on_glob, [glob_pattern, glob_path], [state_view, output, status])
-        grep_btn.click(
-            on_grep, [grep_pattern, grep_path, grep_include], [state_view, output, status]
+        run_btn.click(on_run, inputs=all_inputs, outputs=all_outputs)
+        reset_btn.click(
+            on_reset, inputs=[setup_input, verify_input], outputs=all_outputs
         )
-        ls_btn.click(on_ls, [ls_path, ls_ignore], [state_view, output, status])
-        todo_btn.click(on_todo, [todos_json], [state_view, output, status])
-        submit_btn.click(on_submit, outputs=[state_view, output, status])
+        close_btn.click(on_close, outputs=all_outputs)
 
     return demo