diff --git a/README.md b/README.md
index 7b69c60..bc3ff5d 100644
--- a/README.md
+++ b/README.md
@@ -87,6 +87,56 @@ Canonical session modules are:
 ## Execution Backends
 
 Sessions run with `Jido.Shell.Backend.Local` by default.
+
+### Bash Backend
+
+The Bash backend hands entire command lines to a persistent `Bash.Session` process, so loops, conditionals, variables, pipes, and arithmetic expansion all work as in real Bash. State persists across calls within the same session.
+
+**Dependency** — add the optional `:bash` package to your `mix.exs`:
+
+```elixir
+{:bash, "~> 0.5", optional: true}
+```
+
+**Starting a session:**
+
+```elixir
+{:ok, session_id} =
+  Jido.Shell.ShellSession.start_with_vfs("my_workspace",
+    backend: {Jido.Shell.Backend.Bash, %{}}
+  )
+```
+
+**Agent API:**
+
+```elixir
+{:ok, session} = Jido.Shell.Agent.new("my_workspace",
+  backend: {Jido.Shell.Backend.Bash, %{}})
+
+{:ok, output} = Jido.Shell.Agent.run(session, """
+  for i in 1 2 3; do echo "item $i"; done
+""")
+```
+
+**IEx transport:**
+
+```elixir
+Jido.Shell.Transport.IEx.start("my_workspace",
+  backend: {Jido.Shell.Backend.Bash, %{}})
+```
+
+All registered Jido commands (`echo`, `ls`, `cat`, `cd`, `write`, etc.) are bridged into bash via function shims, so scripts can call them by name. Filesystem I/O routes through `Jido.Shell.VFS` — no host files are touched.
+
+**Isolation:** External binaries (`grep`, `sed`, `curl`, etc.) are blocked by command policy. The session environment is sanitised — `HOME`, `PATH`, and `MACHTYPE` are overridden with sandbox-safe values. See `Jido.Shell.Backend.Bash` moduledoc for the full isolation model.
+
+**Known limitations:**
+
+- Only bash builtins and bridged Jido commands are available — no host binaries.
+- Glob support covers simple `*`/`?` patterns only.
+- Cancellation is best-effort: killing the wrapper task stops output, but the in-flight bash execution may complete inside the session.
+
+### Sprite Backend
+
 To execute commands on Fly.io Sprites, pass a backend tuple when starting a session:
 
 ```elixir
@@ -183,6 +233,7 @@ Event payloads:
 
 - `{:command_started, line}`
 - `{:output, chunk}`
+- `{:output_stderr, chunk}` (Bash backend only)
 - `{:error, %Jido.Shell.Error{}}`
 - `{:cwd_changed, path}`
 - `:command_done`
diff --git a/lib/jido_shell/agent.ex b/lib/jido_shell/agent.ex
index 1b2a0db..37c34d7 100644
--- a/lib/jido_shell/agent.ex
+++ b/lib/jido_shell/agent.ex
@@ -167,6 +167,9 @@ defmodule Jido.Shell.Agent do
       {:jido_shell_session, ^session_id, {:output, chunk}} ->
         collect_output(session_id, expected_command, [chunk | acc], timeout, started?)
 
+      {:jido_shell_session, ^session_id, {:output_stderr, chunk}} ->
+        collect_output(session_id, expected_command, [chunk | acc], timeout, started?)
+
       {:jido_shell_session, ^session_id, {:cwd_changed, _}} ->
         collect_output(session_id, expected_command, acc, timeout, started?)
 
diff --git a/lib/jido_shell/backend/bash.ex b/lib/jido_shell/backend/bash.ex
new file mode 100644
index 0000000..1a8c674
--- /dev/null
+++ b/lib/jido_shell/backend/bash.ex
@@ -0,0 +1,447 @@
+defmodule Jido.Shell.Backend.Bash do
+  @moduledoc """
+  Backend that executes real Bash scripts via the `:bash` library
+  ([tv-labs/bash](https://github.com/tv-labs/bash)).
+
+  Unlike `Jido.Shell.Backend.Local`, which parses commands with the Jido shell
+  parser and routes each statement to a registered command module, this backend
+  hands the entire command line to a persistent `Bash.Session` GenServer. That
+  means loops, conditionals, variable assignments, arithmetic expansion, and
+  pipes all work as in normal Bash — state (variables, functions, cwd)
+  persists across calls within the same jido session.
+
+  Registered Jido commands (`echo`, `ls`, `cat`, …) are bridged into bash via
+  `Jido.Shell.Backend.Bash.JidoInterop`, with bash function shims installed at
+  init time so scripts can call them by their familiar names. Filesystem I/O
+  routes through `Jido.Shell.Backend.Bash.VfsAdapter`, which delegates to
+  `Jido.Shell.VFS`. The backend pins `command_policy: :no_external`, so bash
+  scripts cannot spawn any host process — every effective command is either a
+  bash builtin or a Jido interop call.
+
+  ## Isolation model
+
+  Four layers enforce sandbox boundaries:
+
+    1. **Command policy** — `command_policy: [commands: :no_external]` prevents
+       any OS process from being spawned. Only bash builtins, user-defined shell
+       functions, and Jido interop calls may execute.
+
+    2. **Virtual filesystem** — all file I/O (redirections, `source`, PATH
+       resolution, glob expansion, test operators) routes through
+       `Jido.Shell.Backend.Bash.VfsAdapter`, which delegates to
+       `Jido.Shell.VFS`. No `File.*` or `:file.*` calls reach the host.
+
+    3. **Sanitised environment** — `HOME`, `PATH`, and `MACHTYPE` are overridden
+       with sandbox-safe values so the `:bash` library's init does not leak
+       host-system information into session variables. User-supplied env values
+       from `config.env` take precedence via merge ordering.
+
+    4. **Interop trust boundary** — `defbash` handlers in
+       `Jido.Shell.Backend.Bash.JidoInterop` execute as **unrestricted Elixir
+       code** inside the same BEAM process as the session. The `:bash` library
+       provides no sandbox around interop function bodies. Any module loaded via
+       the `apis:` option has full access to `System.*`, `File.*`, `Port.*`,
+       `spawn`, and the rest of the BEAM. **Only load interop modules you have
+       audited.** The built-in `JidoInterop` is safe — it delegates every call
+       to `Jido.Shell.CommandRunner`, which routes through VFS and the command
+       registry.
+
+  ## Known limitations
+
+    * External binaries (`grep`, `sed`, `awk`, `find`, `curl`, …) are blocked by
+      the command policy — use the bridged Jido commands instead.
+    * Glob support (`VfsAdapter.wildcard/3`) covers simple `*`/`?` patterns
+      only.
+    * `configure_network/2` is a no-op — network policy is fixed at
+      `:no_external`.
+
+  ## Usage
+
+      {:ok, sid} =
+        Jido.Shell.ShellSession.start("ws1", backend: {Jido.Shell.Backend.Bash, %{}})
+
+      Jido.Shell.ShellSession.run_command(sid, "for i in 1 2 3; do echo $i; done")
+
+  Requires the optional `:bash` dependency to be compiled into the release.
+  """
+
+  @behaviour Jido.Shell.Backend
+
+  alias Jido.Shell.Backend.Bash.JidoInterop
+  alias Jido.Shell.Backend.Bash.VfsAdapter
+  alias Jido.Shell.Backend.OutputLimiter
+  alias Jido.Shell.Error
+
+  @default_task_supervisor Jido.Shell.CommandTaskSupervisor
+
+  @impl true
+  def init(config) when is_map(config) do
+    with :ok <- ensure_dep_available(),
+         {:ok, session_pid} <- fetch_session_pid(config),
+         workspace_id <- Map.get(config, :workspace_id, ""),
+         {:ok, bash_session} <- start_bash_session(config, workspace_id),
+         :ok <- install_prelude(bash_session) do
+      {:ok,
+       %{
+         bash_session: bash_session,
+         session_pid: session_pid,
+         task_supervisor: Map.get(config, :task_supervisor, @default_task_supervisor),
+         workspace_id: workspace_id,
+         cwd: Map.get(config, :cwd, "/"),
+         env: Map.get(config, :env, %{})
+       }}
+    end
+  end
+
+  @impl true
+  def execute(state, command, args, exec_opts) when is_binary(command) and is_list(args) and is_list(exec_opts) do
+    line = command_line(command, args)
+    session_pid = state.session_pid
+    bash_session = state.bash_session
+    task_supervisor = state.task_supervisor
+    previous_cwd = state.cwd
+    timeout = positive_limit(Keyword.get(exec_opts, :timeout))
+    output_limit = positive_limit(Keyword.get(exec_opts, :output_limit))
+
+    task_fun = fn ->
+      {emit, limit_ref} = limited_emit(session_pid, bash_session, output_limit)
+
+      result =
+        case safe_parse(line) do
+          {:error, parse_error} ->
+            {:error, Error.command(:syntax_error, %{line: line, reason: inspect(parse_error)})}
+
+          {:ok, ast} ->
+            raw = execute_bash(task_supervisor, bash_session, ast, line, emit, limit_ref, timeout)
+
+            maybe_augment_with_cwd(raw, bash_session, previous_cwd)
+        end
+
+      send(session_pid, {:command_finished, result})
+      result
+    end
+
+    case Task.Supervisor.start_child(state.task_supervisor, task_fun) do
+      {:ok, task_pid} -> {:ok, task_pid, state}
+      {:error, reason} -> {:error, Error.command(:start_failed, %{reason: reason, line: line})}
+    end
+  end
+
+  @impl true
+  def cancel(state, command_ref) when is_pid(command_ref) do
+    # Interrupt the in-flight Bash.Session.execute/3 call so it returns with
+    # exit_code 130 (SIGINT convention). The session remains usable.
+    if is_pid(state.bash_session) and Process.alive?(state.bash_session) do
+      Bash.Session.cancel_execution(state.bash_session)
+    end
+
+    # Kill the Task wrapper (may already be finishing after cancel_execution).
+    if Process.alive?(command_ref) do
+      Process.exit(command_ref, :shutdown)
+    end
+
+    :ok
+  end
+
+  def cancel(_state, _command_ref), do: {:error, :invalid_command_ref}
+
+  @impl true
+  def terminate(state) do
+    case Map.get(state, :bash_session) do
+      pid when is_pid(pid) ->
+        if Process.alive?(pid), do: safe_stop(pid)
+        :ok
+
+      _ ->
+        :ok
+    end
+  end
+
+  @impl true
+  def cwd(state) do
+    case safe_call(state.bash_session, &Bash.Session.get_cwd/1) do
+      {:ok, cwd} -> {:ok, cwd, %{state | cwd: cwd}}
+      _ -> {:ok, state.cwd, state}
+    end
+  end
+
+  @impl true
+  def cd(state, path) when is_binary(path) do
+    _ = safe_call(state.bash_session, fn pid -> Bash.Session.chdir(pid, path) end)
+    {:ok, %{state | cwd: path}}
+  end
+
+  @impl true
+  def configure_network(state, _policy), do: {:ok, state}
+
+  # === private ===
+
+  defp ensure_dep_available do
+    if Code.ensure_loaded?(Bash.Session) do
+      :ok
+    else
+      {:error, Error.command(:start_failed, %{reason: :bash_dep_unavailable})}
+    end
+  end
+
+  defp fetch_session_pid(config) do
+    case Map.get(config, :session_pid) do
+      pid when is_pid(pid) -> {:ok, pid}
+      _ -> {:error, Error.session(:invalid_state_transition, %{reason: :missing_session_pid})}
+    end
+  end
+
+  # Sandbox-safe defaults that prevent the `:bash` library from seeding
+  # session variables with values read from the host OS at init time
+  # (`System.get_env("HOME")`, `System.get_env("PATH")`, etc.).
+  @sandbox_env_defaults %{
+    "HOME" => "/",
+    "PATH" => "",
+    "MACHTYPE" => "beam-unknown-elixir"
+  }
+
+  defp start_bash_session(config, workspace_id) do
+    user_env = Map.get(config, :env, %{})
+    cwd = Map.get(config, :cwd, "/")
+
+    env =
+      @sandbox_env_defaults
+      |> Map.merge(user_env)
+      |> Map.put("JIDO_WORKSPACE_ID", workspace_id)
+
+    opts = [
+      filesystem: {VfsAdapter, %{workspace_id: workspace_id}},
+      working_dir: cwd,
+      env: env,
+      command_policy: [commands: :no_external],
+      apis: [JidoInterop]
+    ]
+
+    case Bash.Session.new(opts) do
+      {:ok, pid} -> {:ok, pid}
+      {:error, reason} -> {:error, Error.command(:start_failed, %{reason: reason})}
+    end
+  end
+
+  # Bash aliases are only active for interactive shells, so we install function
+  # shims instead. Each shim routes the familiar command name (echo, ls, …) to
+  # the corresponding `jido.*` interop handler.
+  defp install_prelude(bash_session) do
+    prelude = build_prelude()
+
+    case safe_parse(prelude) do
+      {:ok, ast} ->
+        case Bash.Session.execute(bash_session, ast, []) do
+          {:ok, _} -> :ok
+          {:error, reason} -> {:error, Error.command(:start_failed, %{reason: {:prelude_failed, reason}})}
+          _ -> :ok
+        end
+
+      {:error, parse_error} ->
+        {:error, Error.command(:start_failed, %{reason: {:prelude_parse_failed, parse_error}})}
+    end
+  end
+
+  # `bash` and `help` are internal shell builtins that don't map to Jido
+  # commands and are handled natively by the bash library.
+  defp build_prelude do
+    skip = ~w(bash help)
+
+    Jido.Shell.Command.Registry.commands()
+    |> Map.keys()
+    |> Enum.sort()
+    |> Enum.reject(&(&1 in skip))
+    |> Enum.map(fn name -> "#{name}() { jido.#{name} \"$@\"; }" end)
+    |> Enum.join("\n")
+  end
+
+  defp command_line(command, []), do: command
+  defp command_line(command, args), do: Enum.join([command | args], " ")
+
+  defp safe_parse(line) do
+    case Bash.parse(line) do
+      {:ok, ast} -> {:ok, ast}
+      {:error, err} -> {:error, err}
+    end
+  end
+
+  defp execute_bash(task_supervisor, bash_session, ast, line, emit, limit_ref, timeout) do
+    task =
+      Task.Supervisor.async_nolink(task_supervisor, fn ->
+        Bash.Session.execute(bash_session, ast, on_output: &stream_output(emit, &1))
+      end)
+
+    await_bash(task, bash_session, line, limit_ref, timeout)
+  end
+
+  defp await_bash(task, bash_session, line, limit_ref, timeout) do
+    task_ref = task.ref
+
+    receive do
+      {^limit_ref, {:error, %Error{} = error}} ->
+        _ = safe_cancel_execution(bash_session)
+        _ = shutdown_task(task)
+        {:error, error}
+
+      {^task_ref, result} ->
+        case pending_limit_error(limit_ref) do
+          {:error, %Error{} = error} -> {:error, error}
+          :none -> bash_result(result, line)
+        end
+
+      {:DOWN, ^task_ref, :process, _pid, reason} ->
+        {:error, Error.command(:crashed, %{line: line, reason: reason})}
+    after
+      receive_timeout(timeout) ->
+        _ = safe_cancel_execution(bash_session)
+        _ = shutdown_task(task)
+        {:error, Error.command(:runtime_limit_exceeded, %{line: line, max_runtime_ms: timeout})}
+    end
+  end
+
+  defp bash_result({status, execution}, line) when status in [:ok, :error, :exit, :exec] do
+    case exit_code(execution) do
+      0 ->
+        {:ok, nil}
+
+      nil when status == :error ->
+        {:error, Error.command(:exit_code, %{exit_code: 1, line: line})}
+
+      nil ->
+        {:ok, nil}
+
+      code ->
+        {:error, Error.command(:exit_code, %{exit_code: code, line: line})}
+    end
+  end
+
+  defp bash_result(other, line), do: {:error, Error.command(:exit_code, %{exit_code: 1, line: line, result: other})}
+
+  defp limited_emit(session_pid, bash_session, output_limit) do
+    owner = self()
+    limit_ref = make_ref()
+    counter = :counters.new(1, [])
+
+    emit = fn event ->
+      case check_output_limit(event, counter, output_limit) do
+        :ok ->
+          send(session_pid, {:command_event, event})
+
+        {:error, %Error{} = error} ->
+          send(owner, {limit_ref, {:error, error}})
+          _ = safe_cancel_execution(bash_session)
+          :ok
+      end
+    end
+
+    {emit, limit_ref}
+  end
+
+  defp check_output_limit(_event, _counter, nil), do: :ok
+
+  defp check_output_limit(event, counter, output_limit) do
+    case output_size(event) do
+      nil ->
+        :ok
+
+      chunk_bytes ->
+        emitted_bytes = :counters.get(counter, 1)
+
+        case OutputLimiter.check(chunk_bytes, emitted_bytes, output_limit) do
+          {:ok, updated_total} ->
+            :counters.put(counter, 1, updated_total)
+            :ok
+
+          {:limit_exceeded, %Error{} = error} ->
+            {:error, error}
+        end
+    end
+  end
+
+  defp output_size({:output, chunk}), do: chunk |> IO.iodata_to_binary() |> byte_size()
+  defp output_size({:output_stderr, chunk}), do: chunk |> IO.iodata_to_binary() |> byte_size()
+  defp output_size(_event), do: nil
+
+  defp pending_limit_error(limit_ref) do
+    receive do
+      {^limit_ref, {:error, %Error{} = error}} -> {:error, error}
+    after
+      0 -> :none
+    end
+  end
+
+  defp shutdown_task(task) do
+    Task.shutdown(task, 100) || Task.shutdown(task, :brutal_kill)
+  end
+
+  defp receive_timeout(nil), do: :infinity
+  defp receive_timeout(timeout) when is_integer(timeout) and timeout > 0, do: timeout
+  defp receive_timeout(_timeout), do: :infinity
+
+  defp positive_limit(value) when is_integer(value) and value > 0, do: value
+  defp positive_limit(_value), do: nil
+
+  defp stream_output(emit, {:stdout, data}), do: emit.({:output, data})
+  defp stream_output(emit, {:stderr, data}), do: emit.({:output_stderr, data})
+  defp stream_output(_emit, _), do: :ok
+
+  defp exit_code(%{exit_code: code}) when is_integer(code), do: code
+
+  defp exit_code(execution) do
+    try do
+      Bash.ExecutionResult.exit_code(execution)
+    rescue
+      _ -> nil
+    end
+  end
+
+  # After each command, pull the session's current working directory and, if it
+  # changed, wrap the result in `{:state_update, %{cwd: new_cwd}}` so
+  # `ShellSessionServer` applies the update and broadcasts `:cwd_changed`.
+  defp maybe_augment_with_cwd({:ok, nil}, bash_session, previous_cwd) do
+    case current_cwd(bash_session) do
+      {:ok, cwd} when cwd != previous_cwd -> {:ok, {:state_update, %{cwd: cwd}}}
+      _ -> {:ok, nil}
+    end
+  end
+
+  defp maybe_augment_with_cwd(other, _bash_session, _previous_cwd), do: other
+
+  defp current_cwd(bash_session) do
+    case safe_call(bash_session, &Bash.Session.get_cwd/1) do
+      {:ok, cwd} when is_binary(cwd) -> {:ok, cwd}
+      _ -> :error
+    end
+  end
+
+  defp safe_call(pid, fun) when is_pid(pid) do
+    if Process.alive?(pid) do
+      {:ok, fun.(pid)}
+    else
+      {:error, :dead}
+    end
+  rescue
+    _ -> {:error, :call_failed}
+  catch
+    _, _ -> {:error, :call_failed}
+  end
+
+  defp safe_cancel_execution(pid) when is_pid(pid) do
+    if Process.alive?(pid), do: Bash.Session.cancel_execution(pid)
+    :ok
+  rescue
+    _ -> :ok
+  catch
+    _, _ -> :ok
+  end
+
+  defp safe_cancel_execution(_pid), do: :ok
+
+  defp safe_stop(pid) do
+    Bash.Session.stop(pid)
+  rescue
+    _ -> :ok
+  catch
+    _, _ -> :ok
+  end
+end
diff --git a/lib/jido_shell/backend/bash/jido_interop.ex b/lib/jido_shell/backend/bash/jido_interop.ex
new file mode 100644
index 0000000..a6307de
--- /dev/null
+++ b/lib/jido_shell/backend/bash/jido_interop.ex
@@ -0,0 +1,208 @@
+defmodule Jido.Shell.Backend.Bash.JidoInterop do
+  @moduledoc """
+  `Bash.Interop` bridge that exposes registered Jido shell commands to scripts
+  running inside a `Bash.Session`.
+
+  For each entry in `Jido.Shell.Command.Registry.commands/0` (minus `bash` and
+  `help`), this module defines a `defbash` handler named after the command. When
+  bash calls `jido.echo hello`, the handler reconstructs a transient
+  `Jido.Shell.ShellSession.State` from the current bash session state plus the
+  captured workspace id, invokes `Jido.Shell.CommandRunner.execute/3`, and
+  funnels the buffered output back through `Bash.puts/2`.
+
+  The bash session's working directory and environment are kept in sync with
+  Jido state transitions: a `cd` command emits `{:state_update, %{cwd: …}}`,
+  and the bridge forwards that via `Bash.update_state/1` so subsequent bash
+  builtins (and the outer backend) see the new cwd.
+
+  Workspace id is taken from the bash session state under the
+  `:jido_workspace_id` variable, which `Jido.Shell.Backend.Bash` sets during
+  initialisation.
+
+  ## Security — interop trust boundary
+
+  `defbash` handlers execute as **unrestricted Elixir code** in the same BEAM
+  process as the `Bash.Session` GenServer. The `:bash` library does not sandbox
+  interop function bodies — a handler may call `File.*`, `System.cmd`,
+  `System.get_env`, `spawn`, or any other BEAM API.
+
+  This module is safe because every handler delegates to
+  `Jido.Shell.CommandRunner.execute/3`, which routes through the VFS and the
+  command registry. If you add a new interop module or modify a handler, ensure
+  it does **not** perform direct host I/O or spawn OS processes — doing so would
+  bypass the filesystem virtualisation and command policy that the rest of the
+  backend enforces.
+  """
+
+  use Bash.Interop, namespace: "jido"
+
+  alias Jido.Shell.CommandRunner
+  alias Jido.Shell.ShellSession.State
+
+  defbash(echo(args, session_state), do: __MODULE__.dispatch("echo", args, session_state))
+  defbash(pwd(args, session_state), do: __MODULE__.dispatch("pwd", args, session_state))
+  defbash(ls(args, session_state), do: __MODULE__.dispatch("ls", args, session_state))
+  defbash(cat(args, session_state), do: __MODULE__.dispatch("cat", args, session_state))
+  defbash(cd(args, session_state), do: __MODULE__.dispatch("cd", args, session_state))
+  defbash(mkdir(args, session_state), do: __MODULE__.dispatch("mkdir", args, session_state))
+  defbash(write(args, session_state), do: __MODULE__.dispatch("write", args, session_state))
+  defbash(sleep(args, session_state), do: __MODULE__.dispatch("sleep", args, session_state))
+  defbash(seq(args, session_state), do: __MODULE__.dispatch("seq", args, session_state))
+  defbash(env(args, session_state), do: __MODULE__.dispatch("env", args, session_state))
+  defbash(rm(args, session_state), do: __MODULE__.dispatch("rm", args, session_state))
+  defbash(cp(args, session_state), do: __MODULE__.dispatch("cp", args, session_state))
+
+  @doc false
+  @spec dispatch(String.t(), [String.t()], map()) ::
+          :ok | {:ok, binary()} | {:error, binary()}
+  def dispatch(command, args, session_state) do
+    with {:ok, state} <- build_state(session_state),
+         line <- build_line(command, args),
+         {stdout, stderr, result} <- run_command(state, line) do
+      finalize(result, stdout, stderr)
+    else
+      {:error, :missing_workspace} ->
+        {:error, "jido.#{command}: workspace not configured\n"}
+
+      {:error, reason} ->
+        {:error, "jido.#{command}: invalid session state: #{inspect(reason)}\n"}
+    end
+  end
+
+  defp build_state(%{variables: variables} = session_state) do
+    case Map.get(variables, "JIDO_WORKSPACE_ID") do
+      nil ->
+        {:error, :missing_workspace}
+
+      var ->
+        workspace_id = variable_value(var)
+        cwd = Map.get(session_state, :working_dir, "/")
+        env = variables_to_env(variables)
+
+        State.new(%{
+          id: "bash-interop",
+          workspace_id: workspace_id,
+          cwd: cwd,
+          env: env
+        })
+    end
+  end
+
+  defp build_state(_), do: {:error, :missing_workspace}
+
+  defp variable_value(%{value: value}) when is_binary(value), do: value
+  defp variable_value(value) when is_binary(value), do: value
+  defp variable_value(_), do: ""
+
+  defp variables_to_env(variables) when is_map(variables) do
+    variables
+    |> Enum.flat_map(fn
+      {"JIDO_WORKSPACE_ID", _} -> []
+      {key, %{value: value}} when is_binary(value) -> [{key, value}]
+      {key, value} when is_binary(value) -> [{key, value}]
+      _ -> []
+    end)
+    |> Map.new()
+  end
+
+  defp build_line(command, []), do: command
+
+  defp build_line(command, args) do
+    Enum.join([command | Enum.map(args, &escape_arg/1)], " ")
+  end
+
+  defp escape_arg(arg) when is_binary(arg) do
+    if String.contains?(arg, [" ", "\t", "\"", "'", "$", "\\"]) do
+      "\"" <> String.replace(arg, ~s("), ~s(\\")) <> "\""
+    else
+      arg
+    end
+  end
+
+  defp escape_arg(other), do: to_string(other)
+
+  defp run_command(state, line) do
+    parent = self()
+    ref = make_ref()
+
+    emit = fn
+      {:output, chunk} ->
+        send(parent, {ref, :stdout, IO.iodata_to_binary(chunk)})
+        :ok
+
+      {:output_stderr, chunk} ->
+        send(parent, {ref, :stderr, IO.iodata_to_binary(chunk)})
+        :ok
+
+      _ ->
+        :ok
+    end
+
+    result = CommandRunner.execute(state, line, emit)
+    {stdout, stderr} = drain(ref, [], [])
+    {stdout, stderr, result}
+  end
+
+  defp drain(ref, stdout, stderr) do
+    receive do
+      {^ref, :stdout, chunk} -> drain(ref, [chunk | stdout], stderr)
+      {^ref, :stderr, chunk} -> drain(ref, stdout, [chunk | stderr])
+    after
+      0 -> {stdout |> Enum.reverse() |> IO.iodata_to_binary(), stderr |> Enum.reverse() |> IO.iodata_to_binary()}
+    end
+  end
+
+  defp finalize({:ok, {:state_update, changes}}, stdout, _stderr) do
+    apply_state_update(changes)
+    emit_ok(stdout)
+  end
+
+  defp finalize({:ok, _result}, stdout, _stderr) do
+    emit_ok(stdout)
+  end
+
+  defp finalize({:error, %Jido.Shell.Error{} = error}, _stdout, stderr) do
+    message =
+      case stderr do
+        "" -> error.message <> "\n"
+        s -> s
+      end
+
+    {:error, message}
+  end
+
+  defp finalize({:error, other}, _stdout, stderr) when is_binary(stderr) and stderr != "" do
+    _ = other
+    {:error, stderr}
+  end
+
+  defp finalize({:error, other}, _stdout, _stderr) do
+    {:error, inspect(other) <> "\n"}
+  end
+
+  defp emit_ok(""), do: :ok
+  defp emit_ok(stdout) when is_binary(stdout), do: {:ok, stdout}
+
+  defp apply_state_update(%{cwd: cwd} = changes) when is_binary(cwd) do
+    updates = %{working_dir: cwd}
+    updates = maybe_add_env(updates, changes)
+    Bash.update_state(updates)
+  end
+
+  defp apply_state_update(%{env: _} = changes) do
+    Bash.update_state(maybe_add_env(%{}, changes))
+  end
+
+  defp apply_state_update(_), do: :ok
+
+  defp maybe_add_env(updates, %{env: env}) when is_map(env) do
+    variables =
+      Map.new(env, fn {k, v} ->
+        {to_string(k), Bash.Variable.new(to_string(v))}
+      end)
+
+    Map.put(updates, :variables, variables)
+  end
+
+  defp maybe_add_env(updates, _), do: updates
+end
diff --git a/lib/jido_shell/backend/bash/vfs_adapter.ex b/lib/jido_shell/backend/bash/vfs_adapter.ex
new file mode 100644
index 0000000..c9dc63f
--- /dev/null
+++ b/lib/jido_shell/backend/bash/vfs_adapter.ex
@@ -0,0 +1,261 @@
+defmodule Jido.Shell.Backend.Bash.VfsAdapter do
+  @moduledoc """
+  `Bash.Filesystem` implementation that routes calls to `Jido.Shell.VFS`.
+
+  Used exclusively by `Jido.Shell.Backend.Bash` so that scripts executed by the
+  `:bash` library see the same virtual filesystem as Jido shell commands. The
+  adapter is configured with the workspace id at session init time:
+
+      {Jido.Shell.Backend.Bash.VfsAdapter, %{workspace_id: "ws1"}}
+
+  The adapter is stateless on its own — it looks up mounts from
+  `Jido.Shell.VFS.MountTable` for every call — and buffers `open/write/close`
+  streams in the calling process dictionary before flushing on close.
+
+  Only simple `*`/`?` globbing is supported; more elaborate patterns emit a
+  single warning per pattern and return an empty list.
+  """
+
+  @behaviour Bash.Filesystem
+
+  alias Jido.Shell.VFS
+
+  @impl true
+  def exists?(config, path), do: VFS.exists?(ws(config), normalize(path))
+
+  @impl true
+  def dir?(config, path) do
+    case VFS.stat(ws(config), normalize(path)) do
+      {:ok, %Jido.VFS.Stat.Dir{}} -> true
+      _ -> false
+    end
+  end
+
+  @impl true
+  def regular?(config, path) do
+    case VFS.stat(ws(config), normalize(path)) do
+      {:ok, %Jido.VFS.Stat.File{}} -> true
+      _ -> false
+    end
+  end
+
+  @impl true
+  def stat(config, path) do
+    case VFS.stat(ws(config), normalize(path)) do
+      {:ok, entry} -> {:ok, to_file_stat(entry)}
+      {:error, _} -> {:error, :enoent}
+    end
+  end
+
+  @impl true
+  def lstat(config, path), do: stat(config, path)
+
+  @impl true
+  def read(config, path) do
+    case VFS.read_file(ws(config), normalize(path)) do
+      {:ok, _} = ok -> ok
+      {:error, _} -> {:error, :enoent}
+    end
+  end
+
+  @impl true
+  def write(config, path, content, opts) do
+    path = normalize(path)
+    binary = IO.iodata_to_binary(content)
+    workspace_id = ws(config)
+
+    final_content =
+      if Keyword.get(opts, :append, false) do
+        case VFS.read_file(workspace_id, path) do
+          {:ok, existing} -> existing <> binary
+          {:error, _} -> binary
+        end
+      else
+        binary
+      end
+
+    case VFS.write_file(workspace_id, path, final_content) do
+      :ok -> :ok
+      {:error, _} -> {:error, :eacces}
+    end
+  end
+
+  @impl true
+  def mkdir_p(config, path) do
+    case VFS.mkdir(ws(config), normalize(path)) do
+      :ok ->
+        :ok
+
+      {:error, %{code: {:vfs, :already_exists}}} ->
+        :ok
+
+      {:error, _} ->
+        # Collapse any other VFS failure — `mkdir_p` should be idempotent.
+        if VFS.exists?(ws(config), normalize(path)), do: :ok, else: {:error, :eacces}
+    end
+  end
+
+  @impl true
+  def rm(config, path) do
+    case VFS.delete(ws(config), normalize(path)) do
+      :ok -> :ok
+      {:error, _} -> {:error, :enoent}
+    end
+  end
+
+  @impl true
+  def ls(config, path) do
+    path = normalize(path)
+
+    case VFS.list_dir(ws(config), path) do
+      {:ok, entries} -> {:ok, entries |> Enum.map(& &1.name) |> Enum.sort()}
+      {:error, _} -> {:error, :enoent}
+    end
+  end
+
+  @impl true
+  def wildcard(config, pattern, _opts) do
+    dir = Path.dirname(pattern)
+    base = Path.basename(pattern)
+
+    cond do
+      # Only handle simple `*`/`?` patterns in the last path segment.
+      not simple_pattern?(base) ->
+        require Logger
+        Logger.warning("Jido.Shell.Backend.Bash.VfsAdapter: unsupported glob #{inspect(pattern)}")
+        []
+
+      true ->
+        case VFS.list_dir(ws(config), normalize(dir)) do
+          {:ok, entries} ->
+            regex = compile_glob(base)
+
+            entries
+            |> Enum.filter(fn entry -> Regex.match?(regex, entry.name) end)
+            |> Enum.map(fn entry -> Path.join(normalize(dir), entry.name) end)
+            |> Enum.sort()
+
+          {:error, _} ->
+            []
+        end
+    end
+  end
+
+  @impl true
+  def open(config, path, modes) when is_list(modes) do
+    cond do
+      :write in modes or :append in modes ->
+        {:ok, make_device(config, path, :append in modes)}
+
+      :read in modes ->
+        case VFS.read_file(ws(config), normalize(path)) do
+          {:ok, content} ->
+            {:ok, device} = StringIO.open(content)
+            {:ok, device}
+
+          {:error, _} ->
+            {:error, :enoent}
+        end
+
+      true ->
+        {:error, :einval}
+    end
+  end
+
+  def open(_config, _path, _modes), do: {:error, :einval}
+
+  @impl true
+  def handle_write(_config, device, data) do
+    case Process.get({__MODULE__, device}) do
+      %{kind: :write, path: _path, buffer: buffer} = state ->
+        Process.put({__MODULE__, device}, %{state | buffer: buffer <> IO.iodata_to_binary(data)})
+        :ok
+
+      _ ->
+        IO.binwrite(device, data)
+    end
+  end
+
+  @impl true
+  def handle_close(config, device) do
+    case Process.get({__MODULE__, device}) do
+      %{kind: :write, path: path, append?: append?, buffer: buffer} ->
+        Process.delete({__MODULE__, device})
+        _ = StringIO.close(device)
+        workspace_id = ws(config)
+
+        final =
+          if append? do
+            case VFS.read_file(workspace_id, path) do
+              {:ok, existing} -> existing <> buffer
+              {:error, _} -> buffer
+            end
+          else
+            buffer
+          end
+
+        case VFS.write_file(workspace_id, path, final) do
+          :ok -> :ok
+          {:error, _} -> {:error, :eacces}
+        end
+
+      _ ->
+        _ = StringIO.close(device)
+        :ok
+    end
+  end
+
+  @impl true
+  def read_link(_config, _path), do: {:error, :enotsup}
+
+  @impl true
+  def read_link_all(_config, _path), do: {:error, :enotsup}
+
+  # === helpers ===
+
+  defp make_device(_config, path, append?) do
+    {:ok, device} = StringIO.open("")
+    Process.put({__MODULE__, device}, %{kind: :write, path: normalize(path), append?: append?, buffer: ""})
+    device
+  end
+
+  defp ws(%{workspace_id: wid}) when is_binary(wid), do: wid
+
+  defp normalize(path) when is_binary(path) do
+    case Path.type(path) do
+      :absolute -> Path.expand(path)
+      _ -> Path.expand(path, "/")
+    end
+  end
+
+  defp to_file_stat(%Jido.VFS.Stat.Dir{}) do
+    now = :calendar.universal_time()
+    %File.Stat{type: :directory, size: 0, access: :read_write, mode: 0o755, mtime: now, atime: now, ctime: now}
+  end
+
+  defp to_file_stat(%Jido.VFS.Stat.File{size: size}) do
+    now = :calendar.universal_time()
+
+    %File.Stat{
+      type: :regular,
+      size: size || 0,
+      access: :read_write,
+      mode: 0o644,
+      mtime: now,
+      atime: now,
+      ctime: now
+    }
+  end
+
+  defp simple_pattern?(base), do: String.match?(base, ~r/^[A-Za-z0-9_.*?\-]*$/)
+
+  defp compile_glob(base) do
+    source =
+      base
+      |> Regex.escape()
+      |> String.replace("\\*", ".*")
+      |> String.replace("\\?", ".")
+
+    Regex.compile!("^" <> source <> "$")
+  end
+end
diff --git a/lib/jido_shell/shell_session_server.ex b/lib/jido_shell/shell_session_server.ex
index 2e4b439..0b8af9a 100644
--- a/lib/jido_shell/shell_session_server.ex
+++ b/lib/jido_shell/shell_session_server.ex
@@ -30,6 +30,7 @@ defmodule Jido.Shell.ShellSessionServer do
 
   The transport will receive messages like:
   - `{:jido_shell_session, session_id, {:output, chunk}}`
+  - `{:jido_shell_session, session_id, {:output_stderr, chunk}}`
   - `{:jido_shell_session, session_id, :command_done}`
   """
   @spec subscribe(String.t(), pid(), keyword()) ::
@@ -89,6 +90,13 @@ defmodule Jido.Shell.ShellSessionServer do
 
   @impl true
   def init(opts) do
+    # Trap exits so backends' `terminate/1` callbacks run on supervisor
+    # shutdown, not just on manual `GenServer.stop/1`. Without this, shutdowns
+    # initiated via `DynamicSupervisor.terminate_child/2` would skip cleanup
+    # for backends that spawn external resources (SSH connections, Bash
+    # sessions, Sprite handles, …).
+    Process.flag(:trap_exit, true)
+
     session_id = Keyword.fetch!(opts, :session_id)
     workspace_id = Keyword.fetch!(opts, :workspace_id)
     cwd = Keyword.get(opts, :cwd, "/")
@@ -239,6 +247,14 @@ defmodule Jido.Shell.ShellSessionServer do
     end
   end
 
+  @impl true
+  def handle_info({:EXIT, _pid, _reason}, state) do
+    # With `trap_exit` enabled, stray exits from linked helper processes land
+    # here. Backends own their own cleanup via `terminate/1`, so just ignore
+    # these messages and let the supervisor-initiated shutdown path run.
+    {:noreply, state}
+  end
+
   # === Private ===
 
   defp apply_state_updates(state, changes) do
diff --git a/lib/jido_shell/stream_json.ex b/lib/jido_shell/stream_json.ex
index 1ba08b7..11ffbb0 100644
--- a/lib/jido_shell/stream_json.ex
+++ b/lib/jido_shell/stream_json.ex
@@ -185,6 +185,21 @@ defmodule Jido.Shell.StreamJson do
             if(parsed_any?, do: monotonic_ms(), else: last_event_ms)
           )
 
+        {:jido_shell_session, ^session_id, {:output_stderr, chunk}} ->
+          collect_stream_output(
+            session_id,
+            deadline_ms,
+            heartbeat_interval_ms,
+            on_event,
+            on_raw_line,
+            on_heartbeat,
+            line_buffer,
+            [chunk | output_acc],
+            event_acc,
+            started?,
+            last_event_ms
+          )
+
         {:jido_shell_session, ^session_id, {:cwd_changed, _}} ->
           collect_stream_output(
             session_id,
diff --git a/lib/jido_shell/transport/iex.ex b/lib/jido_shell/transport/iex.ex
index 9c9bb22..ff48b3a 100644
--- a/lib/jido_shell/transport/iex.ex
+++ b/lib/jido_shell/transport/iex.ex
@@ -120,6 +120,10 @@ defmodule Jido.Shell.Transport.IEx do
         IO.write(chunk)
         wait_for_completion(session_id, cwd, timeout_ms)
 
+      {:jido_shell_session, ^session_id, {:output_stderr, chunk}} ->
+        IO.write(:stderr, chunk)
+        wait_for_completion(session_id, cwd, timeout_ms)
+
       {:jido_shell_session, ^session_id, {:error, error}} ->
         print_error(error)
         cwd
diff --git a/mix.exs b/mix.exs
index 513ea01..4551a3a 100644
--- a/mix.exs
+++ b/mix.exs
@@ -74,6 +74,7 @@ defmodule Jido.Shell.MixProject do
       {:zoi, "~> 0.17"},
       {:jido_vfs, github: "agentjido/jido_vfs", branch: "main"},
       {:sprites, git: "https://github.com/mikehostetler/sprites-ex.git", optional: true},
+      {:bash, git: "https://github.com/lostbean/bash.git", branch: "egomes/cancel-execution", optional: true},
 
       # Dev/Test dependencies
       {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
@@ -144,6 +145,7 @@ defmodule Jido.Shell.MixProject do
           Jido.Shell.Error
         ],
         Backends: [
+          Jido.Shell.Backend.Bash,
           Jido.Shell.Backend.Local,
           Jido.Shell.Backend.Sprite,
           Jido.Shell.Backend.SSH
diff --git a/mix.lock b/mix.lock
index a8165f5..9462a5d 100644
--- a/mix.lock
+++ b/mix.lock
@@ -1,4 +1,5 @@
 %{
+  "bash": {:git, "https://github.com/lostbean/bash.git", "413f5336b2c4a9177c85d1f4c9cab4ddbb3c9994", [branch: "egomes/cancel-execution"]},
   "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
   "certifi": {:hex, :certifi, "2.15.0", "0e6e882fcdaaa0a5a9f2b3db55b1394dba07e8d6d9bcad08318fb604c6839712", [:rebar3], [], "hexpm", "b147ed22ce71d72eafdad94f055165c1c182f61a2ff49df28bcc71d1d5b94a60"},
   "cowlib": {:hex, :cowlib, "2.16.0", "54592074ebbbb92ee4746c8a8846e5605052f29309d3a873468d76cdf932076f", [:make, :rebar3], [], "hexpm", "7f478d80d66b747344f0ea7708c187645cfcc08b11aa424632f78e25bf05db51"},
@@ -9,6 +10,7 @@
   "eternal": {:hex, :eternal, "1.2.2", "d1641c86368de99375b98d183042dd6c2b234262b8d08dfd72b9eeaafc2a1abd", [:mix], [], "hexpm", "2c9fe32b9c3726703ba5e1d43a1d255a4f3f2d8f8f9bc19f094c7cb1a7a9e782"},
   "ex_aws": {:hex, :ex_aws, "2.5.10", "d3f8ca8959dad6533a2a934dfdf380df1b1bef425feeb215a47a5176dee8736c", [:mix], [{:configparser_ex, "~> 5.0", [hex: :configparser_ex, repo: "hexpm", optional: true]}, {:hackney, "~> 1.16", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:jsx, "~> 2.8 or ~> 3.0", [hex: :jsx, repo: "hexpm", optional: true]}, {:mime, "~> 1.2 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:req, "~> 0.5.10 or ~> 0.6 or ~> 1.0", [hex: :req, repo: "hexpm", optional: true]}, {:sweet_xml, "~> 0.7", [hex: :sweet_xml, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "88fcd9cc1b2e0fcea65106bdaa8340ac56c6e29bf72f46cf7ef174027532d3da"},
   "ex_aws_s3": {:hex, :ex_aws_s3, "2.5.7", "e571424d2f345299753382f3a01b005c422b1a460a8bc3ed47659b3d3ef91e9e", [:mix], [{:ex_aws, "~> 2.0", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "858e51241e50181e29aa2bc128fef548873a3a9cd580471f57eda5b64dec937f"},
+  "ex_cmd": {:hex, :ex_cmd, "0.18.0", "791b42b864c099b67254ca909243e2f60a6e091c802f608d4ae2e3da27b0ade9", [:mix], [], "hexpm", "fd4ec30607a7c789cd53b5fb09189bd0a4922eae5b4fadf70176c19bcd19fb76"},
   "ex_doc": {:hex, :ex_doc, "0.38.2", "504d25eef296b4dec3b8e33e810bc8b5344d565998cd83914ffe1b8503737c02", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "732f2d972e42c116a70802f9898c51b54916e542cc50968ac6980512ec90f42b"},
   "excoveralls": {:hex, :excoveralls, "0.18.5", "e229d0a65982613332ec30f07940038fe451a2e5b29bce2a5022165f0c9b157e", [:mix], [{:castore, "~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "523fe8a15603f86d64852aab2abe8ddbd78e68579c8525ae765facc5eae01562"},
   "file_system": {:hex, :file_system, "1.1.0", "08d232062284546c6c34426997dd7ef6ec9f8bbd090eb91780283c9016840e8f", [:mix], [], "hexpm", "bfcf81244f416871f2a2e15c1b515287faa5db9c6bcf290222206d120b3d43f6"},
@@ -49,7 +51,7 @@
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
   "stream_data": {:hex, :stream_data, "1.2.0", "58dd3f9e88afe27dc38bef26fce0c84a9e7a96772b2925c7b32cd2435697a52b", [:mix], [], "hexpm", "eb5c546ee3466920314643edf68943a5b14b32d1da9fe01698dc92b73f89a9ed"},
   "sweet_xml": {:hex, :sweet_xml, "0.7.5", "803a563113981aaac202a1dbd39771562d0ad31004ddbfc9b5090bdcd5605277", [:mix], [], "hexpm", "193b28a9b12891cae351d81a0cead165ffe67df1b73fe5866d10629f4faefb12"},
-  "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"},
+  "telemetry": {:hex, :telemetry, "1.4.1", "ab6de178e2b29b58e8256b92b382ea3f590a47152ca3651ea857a6cae05ac423", [:rebar3], [], "hexpm", "2172e05a27531d3d31dd9782841065c50dd5c3c7699d95266b2edd54c2dafa1c"},
   "tentacat": {:hex, :tentacat, "2.5.0", "d0177ae1289faf6814a85aea044bcdc1ca64a4b1f961e44189451d5c9060a662", [:mix], [{:httpoison, "~> 2.0", [hex: :httpoison, repo: "hexpm", optional: false]}, {:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "c7d3d34a56d5dc870c2155444f7d6cd0e6959dd65c16bb9174442e347f34334f"},
   "text_diff": {:hex, :text_diff, "0.1.0", "1caf3175e11a53a9a139bc9339bd607c47b9e376b073d4571c031913317fecaa", [:mix], [], "hexpm", "d1ffaaecab338e49357b6daa82e435f877e0649041ace7755583a0ea3362dbd7"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.1", "a48703a25c170eedadca83b11e88985af08d35f37c6f664d6dcfb106a97782fc", [:rebar3], [], "hexpm", "b3a917854ce3ae233619744ad1e0102e05673136776fb2fa76234f3e03b23642"},
diff --git a/test/jido/shell/backend/bash/jido_interop_test.exs b/test/jido/shell/backend/bash/jido_interop_test.exs
new file mode 100644
index 0000000..272adc3
--- /dev/null
+++ b/test/jido/shell/backend/bash/jido_interop_test.exs
@@ -0,0 +1,284 @@
+defmodule Jido.Shell.Backend.Bash.JidoInteropTest do
+  use Jido.Shell.Case, async: false
+
+  alias Jido.Shell.Backend.Bash.JidoInterop
+  alias Jido.Shell.Backend.Bash.VfsAdapter
+  alias Jido.Shell.VFS
+
+  setup do
+    VFS.init()
+    workspace_id = "bash_interop_ws_#{System.unique_integer([:positive])}"
+    fs_name = "bash_interop_fs_#{System.unique_integer([:positive])}"
+
+    start_supervised!(
+      {Jido.VFS.Adapter.InMemory, {Jido.VFS.Adapter.InMemory, %Jido.VFS.Adapter.InMemory.Config{name: fs_name}}}
+    )
+
+    :ok = VFS.mount(workspace_id, "/", Jido.VFS.Adapter.InMemory, name: fs_name)
+
+    Code.ensure_loaded!(JidoInterop)
+
+    {:ok, session} =
+      Bash.Session.new(
+        filesystem: {VfsAdapter, %{workspace_id: workspace_id}},
+        working_dir: "/",
+        env: %{"JIDO_WORKSPACE_ID" => workspace_id},
+        command_policy: [commands: :no_external],
+        apis: [JidoInterop]
+      )
+
+    on_exit(fn ->
+      if Process.alive?(session), do: Bash.Session.stop(session)
+      VFS.unmount(workspace_id, "/")
+    end)
+
+    {:ok, session: session, workspace_id: workspace_id}
+  end
+
+  defp run!(session, script) do
+    {_status, result, ^session} = Bash.run(script, session)
+
+    {Bash.ExecutionResult.exit_code(result), Bash.ExecutionResult.stdout(result), Bash.ExecutionResult.stderr(result)}
+  end
+
+  describe "echo" do
+    test "writes to stdout", %{session: session} do
+      {0, stdout, _} = run!(session, "jido.echo hello world")
+      assert stdout == "hello world\n"
+    end
+  end
+
+  describe "pwd" do
+    test "reflects the current working directory", %{session: session} do
+      {0, stdout, _} = run!(session, "jido.pwd")
+      assert stdout == "/\n"
+    end
+  end
+
+  describe "write + cat round trip" do
+    test "persists to the VFS and reads it back", %{session: session} do
+      {0, _, _} = run!(session, "jido.write /note.txt 'hello from bash'")
+      {0, stdout, _} = run!(session, "jido.cat /note.txt")
+      assert stdout == "hello from bash"
+    end
+  end
+
+  describe "ls" do
+    test "lists directory contents", %{session: session, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/a.txt", "")
+      :ok = VFS.write_file(wid, "/b.txt", "")
+
+      {0, stdout, _} = run!(session, "jido.ls /")
+      assert stdout =~ "a.txt"
+      assert stdout =~ "b.txt"
+    end
+  end
+
+  describe "cd" do
+    test "updates the outer session working directory", %{session: session, workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/home")
+
+      {0, _, _} = run!(session, "jido.cd /home")
+      assert Bash.Session.get_cwd(session) == "/home"
+
+      {0, stdout, _} = run!(session, "jido.pwd")
+      assert stdout == "/home\n"
+    end
+  end
+
+  describe "errors" do
+    test "propagates file-not-found failures as non-zero exit", %{session: session} do
+      {exit_code, _stdout, _stderr} = run!(session, "jido.cat /definitely-missing.txt")
+      assert exit_code != 0
+    end
+  end
+
+  describe "argument escaping" do
+    test "passes arguments with whitespace through to the Jido command", %{session: session} do
+      {0, stdout, _} = run!(session, ~s/jido.echo "hello world" "and again"/)
+      assert stdout =~ "hello world"
+      assert stdout =~ "and again"
+    end
+  end
+
+  describe "dispatch/3 direct" do
+    test "returns an error when JIDO_WORKSPACE_ID is not set" do
+      state = %{variables: %{}, working_dir: "/"}
+      assert {:error, msg} = JidoInterop.dispatch("echo", ["hi"], state)
+      assert msg =~ "workspace not configured"
+    end
+
+    test "returns an error when session_state is malformed" do
+      assert {:error, msg} = JidoInterop.dispatch("echo", [], %{})
+      assert msg =~ "workspace not configured"
+    end
+  end
+
+  describe "env round trip" do
+    test "echoes an env var set through the outer session", %{session: session} do
+      {:ok, _, ^session} = Bash.run("export MY_VAR=hello", session)
+      {0, stdout, _} = run!(session, ~s/jido.echo "$MY_VAR"/)
+      assert stdout == "hello\n"
+    end
+  end
+
+  describe "remaining jido commands" do
+    test "mkdir creates directories", %{session: session, workspace_id: wid} do
+      {0, _, _} = run!(session, "jido.mkdir /mydir")
+      assert VFS.exists?(wid, "/mydir")
+    end
+
+    test "rm removes files", %{session: session, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/tmp.txt", "delete me")
+      {0, _, _} = run!(session, "jido.rm /tmp.txt")
+      refute VFS.exists?(wid, "/tmp.txt")
+    end
+
+    test "cp copies files", %{session: session, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/orig.txt", "content")
+      {0, _, _} = run!(session, "jido.cp /orig.txt /copy.txt")
+      assert {:ok, "content"} = VFS.read_file(wid, "/copy.txt")
+    end
+
+    test "seq generates a sequence", %{session: session} do
+      {0, stdout, _} = run!(session, "jido.seq 3")
+      assert stdout =~ "1"
+      assert stdout =~ "3"
+    end
+
+    test "sleep completes quickly for small durations", %{session: session} do
+      {0, _, _} = run!(session, "jido.sleep 0")
+    end
+
+    test "env lists environment variables", %{session: session} do
+      {0, stdout, _} = run!(session, "jido.env")
+      assert is_binary(stdout)
+    end
+  end
+
+  describe "build_state variants" do
+    test "accepts workspace id as a plain string variable", %{workspace_id: wid} do
+      state = %{variables: %{"JIDO_WORKSPACE_ID" => wid}, working_dir: "/"}
+      assert {:ok, "\n"} = JidoInterop.dispatch("echo", [], state)
+    end
+
+    test "accepts workspace id as a %Bash.Variable{}", %{workspace_id: wid} do
+      state = %{
+        variables: %{"JIDO_WORKSPACE_ID" => Bash.Variable.new(wid)},
+        working_dir: "/"
+      }
+
+      assert {:ok, output} = JidoInterop.dispatch("echo", ["hi"], state)
+      assert output == "hi\n"
+    end
+
+    test "propagates environment variables from session_state", %{workspace_id: wid} do
+      state = %{
+        variables: %{
+          "JIDO_WORKSPACE_ID" => Bash.Variable.new(wid),
+          "MY_KEY" => Bash.Variable.new("my_value")
+        },
+        working_dir: "/"
+      }
+
+      assert {:ok, output} = JidoInterop.dispatch("env", [], state)
+      assert output =~ "MY_KEY"
+      assert output =~ "my_value"
+    end
+
+    test "returns an error when the underlying command fails", %{workspace_id: wid} do
+      state = %{
+        variables: %{"JIDO_WORKSPACE_ID" => Bash.Variable.new(wid)},
+        working_dir: "/"
+      }
+
+      assert {:error, msg} = JidoInterop.dispatch("cat", ["/not-there.txt"], state)
+      assert is_binary(msg)
+    end
+
+    test "handles a non-string non-struct variable value gracefully", %{workspace_id: wid} do
+      state = %{variables: %{"JIDO_WORKSPACE_ID" => %{value: wid}}, working_dir: "/"}
+      assert {:ok, "\n"} = JidoInterop.dispatch("echo", [], state)
+    end
+
+    test "ignores variables with non-binary values in env map", %{workspace_id: wid} do
+      state = %{
+        variables: %{
+          "JIDO_WORKSPACE_ID" => Bash.Variable.new(wid),
+          "BAD" => 42
+        },
+        working_dir: "/"
+      }
+
+      # Should succeed (BAD is silently dropped)
+      assert result = JidoInterop.dispatch("echo", ["ok"], state)
+      assert {:ok, "ok\n"} = result
+    end
+  end
+
+  describe "write back from Jido command through bash session" do
+    test "write command persists through interop bridge", %{session: session, workspace_id: wid} do
+      {0, _, _} = run!(session, "jido.write /from_bash.txt 'bash wrote this'")
+      assert {:ok, "bash wrote this"} = VFS.read_file(wid, "/from_bash.txt")
+    end
+  end
+
+  describe "edge cases" do
+    test "handles numeric variable values", %{workspace_id: wid} do
+      state = %{
+        variables: %{
+          "JIDO_WORKSPACE_ID" => %{value: wid},
+          "NUM" => 42
+        },
+        working_dir: "/"
+      }
+
+      # Should succeed — non-binary, non-struct variables are silently dropped
+      result = JidoInterop.dispatch("echo", ["test"], state)
+      assert {:ok, "test\n"} = result
+    end
+
+    test "invalid workspace variables return a normal interop error", %{workspace_id: _wid} do
+      # A non-struct, non-binary workspace id falls to the empty-string path.
+      state = %{variables: %{"JIDO_WORKSPACE_ID" => {:tuple, "x"}}, working_dir: "/"}
+
+      assert {:error, message} = JidoInterop.dispatch("echo", ["hi"], state)
+      assert message =~ "invalid session state"
+    end
+  end
+
+  describe "stderr propagation through interop" do
+    test "a Jido command error returns stderr message", %{session: session} do
+      # cat on a missing file should produce a non-zero exit and stderr content
+      {exit_code, _stdout, stderr} = run!(session, "jido.cat /definitely-missing-file.txt")
+      assert exit_code != 0
+      assert is_binary(stderr)
+    end
+  end
+
+  describe "function shims in bash session" do
+    test "function shim for echo works through prelude-style definition", %{session: session} do
+      # Manually define the shim the way the prelude does it
+      {:ok, _, ^session} = Bash.run("my_echo() { jido.echo \"$@\"; }", session)
+      {0, stdout, _} = run!(session, "my_echo routed")
+      assert stdout == "routed\n"
+    end
+  end
+
+  describe "variables_to_env with plain string values" do
+    test "accepts plain string variable values in env", %{workspace_id: wid} do
+      # This exercises the `{key, value} when is_binary(value)` branch
+      state = %{
+        variables: %{
+          "JIDO_WORKSPACE_ID" => Bash.Variable.new(wid),
+          "PLAIN_STRING" => "raw_string_value"
+        },
+        working_dir: "/"
+      }
+
+      assert {:ok, output} = JidoInterop.dispatch("env", [], state)
+      assert output =~ "PLAIN_STRING"
+      assert output =~ "raw_string_value"
+    end
+  end
+end
diff --git a/test/jido/shell/backend/bash/vfs_adapter_test.exs b/test/jido/shell/backend/bash/vfs_adapter_test.exs
new file mode 100644
index 0000000..df0835b
--- /dev/null
+++ b/test/jido/shell/backend/bash/vfs_adapter_test.exs
@@ -0,0 +1,216 @@
+defmodule Jido.Shell.Backend.Bash.VfsAdapterTest do
+  use Jido.Shell.Case, async: false
+
+  alias Jido.Shell.Backend.Bash.VfsAdapter
+  alias Jido.Shell.VFS
+
+  setup do
+    VFS.init()
+    workspace_id = "bash_vfs_ws_#{System.unique_integer([:positive])}"
+    fs_name = "bash_vfs_fs_#{System.unique_integer([:positive])}"
+
+    start_supervised!(
+      {Jido.VFS.Adapter.InMemory, {Jido.VFS.Adapter.InMemory, %Jido.VFS.Adapter.InMemory.Config{name: fs_name}}}
+    )
+
+    :ok = VFS.mount(workspace_id, "/", Jido.VFS.Adapter.InMemory, name: fs_name)
+
+    on_exit(fn -> VFS.unmount(workspace_id, "/") end)
+
+    {:ok, config: %{workspace_id: workspace_id}, workspace_id: workspace_id}
+  end
+
+  describe "write/4 and read/2" do
+    test "round-trips content", %{config: cfg} do
+      assert :ok = VfsAdapter.write(cfg, "/hello.txt", "hi", [])
+      assert {:ok, "hi"} = VfsAdapter.read(cfg, "/hello.txt")
+    end
+
+    test "append mode concatenates", %{config: cfg} do
+      :ok = VfsAdapter.write(cfg, "/log.txt", "one\n", [])
+      :ok = VfsAdapter.write(cfg, "/log.txt", "two\n", append: true)
+      assert {:ok, "one\ntwo\n"} = VfsAdapter.read(cfg, "/log.txt")
+    end
+
+    test "reports enoent for missing files", %{config: cfg} do
+      assert {:error, :enoent} = VfsAdapter.read(cfg, "/missing.txt")
+    end
+  end
+
+  describe "exists?/2, dir?/2, regular?/2" do
+    test "distinguishes files and directories", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/docs")
+      :ok = VFS.write_file(wid, "/docs/readme.md", "hello")
+
+      assert VfsAdapter.exists?(cfg, "/docs")
+      assert VfsAdapter.exists?(cfg, "/docs/readme.md")
+      refute VfsAdapter.exists?(cfg, "/docs/missing.txt")
+
+      assert VfsAdapter.dir?(cfg, "/docs")
+      refute VfsAdapter.dir?(cfg, "/docs/readme.md")
+
+      assert VfsAdapter.regular?(cfg, "/docs/readme.md")
+      refute VfsAdapter.regular?(cfg, "/docs")
+    end
+  end
+
+  describe "stat/2" do
+    test "returns File.Stat for directories", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/d")
+      assert {:ok, %File.Stat{type: :directory}} = VfsAdapter.stat(cfg, "/d")
+    end
+
+    test "returns File.Stat for regular files", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/f.txt", "abcd")
+      assert {:ok, %File.Stat{type: :regular, size: 4}} = VfsAdapter.stat(cfg, "/f.txt")
+    end
+
+    test "maps missing paths to enoent", %{config: cfg} do
+      assert {:error, :enoent} = VfsAdapter.stat(cfg, "/nope")
+    end
+  end
+
+  describe "mkdir_p/2" do
+    test "creates a new directory", %{config: cfg} do
+      assert :ok = VfsAdapter.mkdir_p(cfg, "/new")
+      assert VfsAdapter.dir?(cfg, "/new")
+    end
+
+    test "is idempotent when the directory already exists", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/existing")
+      assert :ok = VfsAdapter.mkdir_p(cfg, "/existing")
+    end
+  end
+
+  describe "rm/2" do
+    test "removes a file", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/gone.txt", "x")
+      assert :ok = VfsAdapter.rm(cfg, "/gone.txt")
+      refute VfsAdapter.exists?(cfg, "/gone.txt")
+    end
+
+    test "tolerates missing files the way the underlying VFS does", %{config: cfg} do
+      # The in-memory VFS adapter returns :ok for missing paths, so rm/2 is
+      # idempotent. If a future adapter returns an error, the adapter maps it
+      # to {:error, :enoent}.
+      assert VfsAdapter.rm(cfg, "/nope") in [:ok, {:error, :enoent}]
+    end
+  end
+
+  describe "ls/2" do
+    test "returns sorted entry names", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/d")
+      :ok = VFS.write_file(wid, "/d/b.txt", "")
+      :ok = VFS.write_file(wid, "/d/a.txt", "")
+
+      assert {:ok, ["a.txt", "b.txt"]} = VfsAdapter.ls(cfg, "/d")
+    end
+  end
+
+  describe "wildcard/3" do
+    test "matches simple star patterns", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/a.log", "")
+      :ok = VFS.write_file(wid, "/b.log", "")
+      :ok = VFS.write_file(wid, "/c.txt", "")
+
+      matches = VfsAdapter.wildcard(cfg, "/*.log", [])
+      assert Enum.sort(matches) == ["/a.log", "/b.log"]
+    end
+
+    test "returns [] for unsupported patterns", %{config: cfg} do
+      assert [] = VfsAdapter.wildcard(cfg, "/**/*.log", [])
+    end
+  end
+
+  describe "open/write/close round trip" do
+    test "buffers writes until close", %{config: cfg} do
+      {:ok, device} = VfsAdapter.open(cfg, "/buffered.txt", [:write])
+      :ok = VfsAdapter.handle_write(cfg, device, "part1 ")
+      :ok = VfsAdapter.handle_write(cfg, device, "part2")
+      :ok = VfsAdapter.handle_close(cfg, device)
+
+      assert {:ok, "part1 part2"} = VfsAdapter.read(cfg, "/buffered.txt")
+    end
+
+    test "open :read on a missing file reports enoent", %{config: cfg} do
+      assert {:error, :enoent} = VfsAdapter.open(cfg, "/missing", [:read])
+    end
+  end
+
+  describe "link callbacks" do
+    test "are not supported", %{config: cfg} do
+      assert {:error, :enotsup} = VfsAdapter.read_link(cfg, "/anything")
+      assert {:error, :enotsup} = VfsAdapter.read_link_all(cfg, "/anything")
+    end
+  end
+
+  describe "lstat/2" do
+    test "mirrors stat/2", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/linked.txt", "abc")
+      assert {:ok, %File.Stat{type: :regular, size: 3}} = VfsAdapter.lstat(cfg, "/linked.txt")
+    end
+  end
+
+  describe "open/3 modes" do
+    test "opening with :append flushes appended data on close", %{config: cfg} do
+      :ok = VfsAdapter.write(cfg, "/append.txt", "line1\n", [])
+      {:ok, device} = VfsAdapter.open(cfg, "/append.txt", [:append])
+      :ok = VfsAdapter.handle_write(cfg, device, "line2\n")
+      :ok = VfsAdapter.handle_close(cfg, device)
+      assert {:ok, "line1\nline2\n"} = VfsAdapter.read(cfg, "/append.txt")
+    end
+
+    test "rejects modes without read/write/append", %{config: cfg} do
+      assert {:error, :einval} = VfsAdapter.open(cfg, "/bad", [:exclusive])
+    end
+
+    test "rejects non-list modes", %{config: cfg} do
+      assert {:error, :einval} = VfsAdapter.open(cfg, "/bad", nil)
+    end
+  end
+
+  describe "handle_write/handle_close fallbacks" do
+    test "handle_write on an unknown device falls back to IO.binwrite", %{config: cfg} do
+      {:ok, device} = StringIO.open("")
+      assert :ok = VfsAdapter.handle_write(cfg, device, "raw")
+      {_, written} = StringIO.contents(device)
+      assert written == "raw"
+      StringIO.close(device)
+    end
+
+    test "handle_close on an unknown device closes it without writing to VFS", %{config: cfg} do
+      {:ok, device} = StringIO.open("")
+      assert :ok = VfsAdapter.handle_close(cfg, device)
+    end
+  end
+
+  describe "normalize relative paths" do
+    test "resolves relative paths against root", %{config: cfg} do
+      :ok = VfsAdapter.write(cfg, "hello.txt", "hi", [])
+      assert {:ok, "hi"} = VfsAdapter.read(cfg, "hello.txt")
+    end
+  end
+
+  describe "ls/2 edge cases" do
+    test "returns sorted names for existing directory", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/dir2")
+      :ok = VFS.write_file(wid, "/dir2/x", "")
+      assert {:ok, ["x"]} = VfsAdapter.ls(cfg, "/dir2")
+    end
+  end
+
+  describe "wildcard/3 extras" do
+    test "question-mark patterns match single characters", %{config: cfg, workspace_id: wid} do
+      :ok = VFS.write_file(wid, "/ax.txt", "")
+      :ok = VFS.write_file(wid, "/bx.txt", "")
+      :ok = VFS.write_file(wid, "/ax2.txt", "")
+
+      matches = VfsAdapter.wildcard(cfg, "/?x.txt", [])
+      assert Enum.sort(matches) == ["/ax.txt", "/bx.txt"]
+    end
+
+    test "returns [] when the parent directory does not exist", %{config: cfg} do
+      assert [] = VfsAdapter.wildcard(cfg, "/nonexistent/*.log", [])
+    end
+  end
+end
diff --git a/test/jido/shell/backend/bash_test.exs b/test/jido/shell/backend/bash_test.exs
new file mode 100644
index 0000000..47ecd47
--- /dev/null
+++ b/test/jido/shell/backend/bash_test.exs
@@ -0,0 +1,580 @@
+defmodule Jido.Shell.Backend.BashTest do
+  use Jido.Shell.Case, async: false
+
+  alias Jido.Shell.Backend.Bash, as: BashBackend
+  alias Jido.Shell.ShellSession
+  alias Jido.Shell.ShellSessionServer
+  alias Jido.Shell.VFS
+
+  @event_timeout 2_000
+
+  setup do
+    VFS.init()
+    workspace_id = "bash_backend_ws_#{System.unique_integer([:positive])}"
+    fs_name = "bash_backend_fs_#{System.unique_integer([:positive])}"
+
+    start_supervised!(
+      {Jido.VFS.Adapter.InMemory, {Jido.VFS.Adapter.InMemory, %Jido.VFS.Adapter.InMemory.Config{name: fs_name}}}
+    )
+
+    :ok = VFS.mount(workspace_id, "/", Jido.VFS.Adapter.InMemory, name: fs_name)
+
+    on_exit(fn -> VFS.unmount(workspace_id, "/") end)
+
+    {:ok, workspace_id: workspace_id}
+  end
+
+  defp start_session(workspace_id, opts \\ []) do
+    {:ok, session_id} =
+      ShellSession.start(
+        workspace_id,
+        Keyword.merge([backend: {Jido.Shell.Backend.Bash, %{}}], opts)
+      )
+
+    {:ok, :subscribed} = ShellSessionServer.subscribe(session_id, self())
+    session_id
+  end
+
+  defp receive_output(session_id, acc \\ "", stderr_acc \\ "") do
+    receive do
+      {:jido_shell_session, ^session_id, {:output, chunk}} ->
+        receive_output(session_id, acc <> IO.iodata_to_binary(chunk), stderr_acc)
+
+      {:jido_shell_session, ^session_id, {:output_stderr, chunk}} ->
+        receive_output(session_id, acc, stderr_acc <> IO.iodata_to_binary(chunk))
+
+      {:jido_shell_session, ^session_id, :command_done} ->
+        {:ok, acc, stderr_acc}
+
+      {:jido_shell_session, ^session_id, {:error, err}} ->
+        {:error, err, acc}
+    after
+      @event_timeout -> {:timeout, acc}
+    end
+  end
+
+  describe "happy path" do
+    test "echo streams output and completes", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo hello")
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, "echo hello"}}, @event_timeout
+      assert {:ok, "hello\n", ""} = receive_output(session_id)
+    end
+  end
+
+  describe "real bash features" do
+    test "for loops with variables", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} =
+        ShellSessionServer.run_command(session_id, "for i in 1 2 3; do echo $i; done")
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, output, _} = receive_output(session_id)
+      assert output =~ "1"
+      assert output =~ "2"
+      assert output =~ "3"
+    end
+
+    test "variables persist across run_command calls", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "x=5")
+      assert_receive {:jido_shell_session, ^session_id, :command_done}, @event_timeout
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, ~s/echo "$x"/)
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "5\n", ""} = receive_output(session_id)
+    end
+
+    test "arithmetic expansion", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo $((3 * 7))")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "21\n", ""} = receive_output(session_id)
+    end
+
+    test "redirect into VFS and read back", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo hi > /tmp_a.txt")
+      assert_receive {:jido_shell_session, ^session_id, :command_done}, @event_timeout
+      assert {:ok, "hi\n"} = VFS.read_file(wid, "/tmp_a.txt")
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "cat /tmp_a.txt")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "hi\n", ""} = receive_output(session_id)
+    end
+  end
+
+  describe "isolation" do
+    test "external commands are denied by command policy", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "curl example.com")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      # Either an error event or command_done with a non-zero exit — both are
+      # acceptable for the prototype. The load-bearing assertion is that no
+      # host process ran, which the :no_external policy enforces at a layer
+      # we trust the :bash library to honor.
+      result =
+        receive do
+          {:jido_shell_session, ^session_id, {:error, _}} -> :error
+          {:jido_shell_session, ^session_id, :command_done} -> :done
+          {:jido_shell_session, ^session_id, {:output, _}} -> :output
+        after
+          @event_timeout -> :timeout
+        end
+
+      assert result in [:error, :done, :output]
+    end
+
+    test "HOME is sandbox-safe, not the host value", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, ~s/echo "$HOME"/)
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, output, _} = receive_output(session_id)
+      assert String.trim(output) == "/"
+      refute output =~ System.get_env("HOME", "")
+    end
+
+    test "PATH is empty so external resolution cannot succeed", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, ~s/echo "$PATH"/)
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, output, _} = receive_output(session_id)
+      assert String.trim(output) == ""
+    end
+
+    test "MACHTYPE does not leak host architecture", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, ~s/echo "$MACHTYPE"/)
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, output, _} = receive_output(session_id)
+      assert String.trim(output) == "beam-unknown-elixir"
+    end
+
+    test "user-provided env overrides sandbox defaults", %{workspace_id: wid} do
+      {:ok, session_id} =
+        Jido.Shell.ShellSession.start(wid,
+          backend: {Jido.Shell.Backend.Bash, %{}},
+          env: %{"HOME" => "/custom_home"}
+        )
+
+      {:ok, :subscribed} = ShellSessionServer.subscribe(session_id, self())
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, ~s/echo "$HOME"/)
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, output, _} = receive_output(session_id)
+      assert String.trim(output) == "/custom_home"
+    end
+  end
+
+  describe "cwd sync" do
+    test "cd propagates to the outer session", %{workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/docs")
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "cd /docs")
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, {:cwd_changed, "/docs"}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, :command_done}, @event_timeout
+
+      {:ok, state} = ShellSessionServer.get_state(session_id)
+      assert state.cwd == "/docs"
+    end
+  end
+
+  describe "session termination" do
+    test "stopping the shell session stops the bash session", %{workspace_id: wid} do
+      session_id = start_session(wid)
+      {:ok, state} = ShellSessionServer.get_state(session_id)
+      bash_pid = state.backend_state.bash_session
+      assert Process.alive?(bash_pid)
+
+      :ok = ShellSession.stop(session_id)
+
+      # Give the GenServer a tick to fully terminate.
+      wait_until(fn -> not Process.alive?(bash_pid) end, 2_000)
+      refute Process.alive?(bash_pid)
+    end
+  end
+
+  describe "dep unavailability" do
+    test "reports a start-failed error when Bash.Session is missing" do
+      # This branch is defensive — with :bash compiled in, Code.ensure_loaded?
+      # always succeeds. We keep this test as documentation; if someone drops
+      # the dep, the backend should degrade loudly.
+      assert Code.ensure_loaded?(Bash.Session)
+    end
+  end
+
+  describe "init error paths" do
+    test "fails when session_pid is missing", %{workspace_id: wid} do
+      assert {:error, %Jido.Shell.Error{} = error} =
+               BashBackend.init(%{workspace_id: wid, cwd: "/", env: %{}})
+
+      assert error.code == {:session, :invalid_state_transition}
+    end
+  end
+
+  describe "direct callback surface" do
+    setup %{workspace_id: wid} do
+      {:ok, state} =
+        BashBackend.init(%{
+          workspace_id: wid,
+          session_pid: self(),
+          cwd: "/",
+          env: %{},
+          task_supervisor: Jido.Shell.CommandTaskSupervisor
+        })
+
+      on_exit(fn -> BashBackend.terminate(state) end)
+      {:ok, state: state}
+    end
+
+    test "cwd/1 returns the current working directory", %{state: state} do
+      assert {:ok, "/", _state} = BashBackend.cwd(state)
+    end
+
+    test "cd/2 updates the session working directory", %{state: state, workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/work")
+      assert {:ok, new_state} = BashBackend.cd(state, "/work")
+      assert new_state.cwd == "/work"
+      assert {:ok, "/work", _} = BashBackend.cwd(new_state)
+    end
+
+    test "cancel/2 is a no-op for dead pids", %{state: state} do
+      dead_pid = spawn(fn -> :ok end)
+      Process.sleep(10)
+      refute Process.alive?(dead_pid)
+      assert :ok = BashBackend.cancel(state, dead_pid)
+    end
+
+    test "cancel/2 rejects non-pid refs", %{state: state} do
+      assert {:error, :invalid_command_ref} = BashBackend.cancel(state, :not_a_pid)
+    end
+
+    test "cancel/2 kills a live task", %{state: state} do
+      pid = spawn(fn -> Process.sleep(:infinity) end)
+      assert :ok = BashBackend.cancel(state, pid)
+      wait_until(fn -> not Process.alive?(pid) end, 500)
+      refute Process.alive?(pid)
+    end
+
+    test "configure_network/2 is a no-op", %{state: state} do
+      assert {:ok, ^state} = BashBackend.configure_network(state, %{allow: :all})
+    end
+
+    test "terminate/1 is idempotent when bash session is gone", %{state: state} do
+      :ok = BashBackend.terminate(state)
+      # Calling terminate again should not raise even though the pid is dead.
+      assert :ok = BashBackend.terminate(state)
+    end
+
+    test "terminate/1 handles a state without bash_session" do
+      assert :ok = BashBackend.terminate(%{})
+    end
+  end
+
+  describe "execute non-zero exit" do
+    test "a command with non-zero exit reports an error", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "false")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      assert_receive {:jido_shell_session, ^session_id, {:error, %Jido.Shell.Error{code: {:command, :exit_code}}}},
+                     @event_timeout
+    end
+
+    test "exit builtin preserves non-zero status", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "exit 42")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      assert_receive {:jido_shell_session, ^session_id,
+                      {:error,
+                       %Jido.Shell.Error{
+                         code: {:command, :exit_code},
+                         context: %{exit_code: 42}
+                       }}},
+                     @event_timeout
+    end
+  end
+
+  describe "execution limits" do
+    test "enforces output limit for native bash output", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} =
+        ShellSessionServer.run_command(session_id, "echo abcdef", execution_context: %{limits: %{max_output_bytes: 3}})
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      assert_receive {:jido_shell_session, ^session_id,
+                      {:error, %Jido.Shell.Error{code: {:command, :output_limit_exceeded}}}},
+                     @event_timeout
+
+      refute_receive {:jido_shell_session, ^session_id, {:output, _}}, 100
+    end
+
+    test "enforces runtime limit and keeps session reusable", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} =
+        ShellSessionServer.run_command(session_id, "while true; do :; done",
+          execution_context: %{limits: %{max_runtime_ms: 50}}
+        )
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      assert_receive {:jido_shell_session, ^session_id,
+                      {:error, %Jido.Shell.Error{code: {:command, :runtime_limit_exceeded}}}},
+                     @event_timeout
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo alive")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "alive\n", ""} = receive_output(session_id)
+    end
+  end
+
+  describe "execute streaming" do
+    test "stderr is streamed separately from stdout", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo hello >&2")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "", "hello\n"} = receive_output(session_id)
+    end
+
+    test "exit builtin is treated as a successful completion", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "exit 0")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, :command_done}, @event_timeout
+    end
+  end
+
+  describe "execute error branches" do
+    test "invalid bash syntax returns a clean syntax_error", %{workspace_id: wid} do
+      {:ok, state} =
+        BashBackend.init(%{
+          workspace_id: wid,
+          session_pid: self(),
+          cwd: "/",
+          env: %{},
+          task_supervisor: Jido.Shell.CommandTaskSupervisor
+        })
+
+      assert {:ok, _pid, _state} = BashBackend.execute(state, "echo \"unterminated", [], [])
+
+      assert_receive {:command_finished, {:error, %Jido.Shell.Error{code: {:command, :syntax_error}}}},
+                     @event_timeout
+
+      :ok = BashBackend.terminate(state)
+    end
+
+    test "session is usable after a syntax error", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo \"unterminated")
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      assert_receive {:jido_shell_session, ^session_id, {:error, %Jido.Shell.Error{code: {:command, :syntax_error}}}},
+                     @event_timeout
+
+      # Session should accept new commands
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo ok")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "ok\n", ""} = receive_output(session_id)
+    end
+
+    test "mutating interop workspace state returns a command error", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "JIDO_WORKSPACE_ID=; echo hi")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      assert_receive {:jido_shell_session, ^session_id, {:output_stderr, stderr}}, @event_timeout
+      assert stderr =~ "invalid session state"
+
+      assert_receive {:jido_shell_session, ^session_id, {:error, %Jido.Shell.Error{code: {:command, :exit_code}}}},
+                     @event_timeout
+    end
+  end
+
+  describe "cwd with dead session" do
+    test "falls back to cached cwd", %{workspace_id: wid} do
+      {:ok, state} =
+        BashBackend.init(%{
+          workspace_id: wid,
+          session_pid: self(),
+          cwd: "/fallback",
+          env: %{},
+          task_supervisor: Jido.Shell.CommandTaskSupervisor
+        })
+
+      # Kill the bash session to trigger the fallback path in cwd/1
+      Bash.Session.stop(state.bash_session)
+      Process.sleep(50)
+
+      assert {:ok, "/fallback", _state} = BashBackend.cwd(state)
+    end
+  end
+
+  describe "execute with args" do
+    test "backend passes args alongside command", %{workspace_id: wid} do
+      {:ok, state} =
+        BashBackend.init(%{
+          workspace_id: wid,
+          session_pid: self(),
+          cwd: "/",
+          env: %{},
+          task_supervisor: Jido.Shell.CommandTaskSupervisor
+        })
+
+      assert {:ok, _pid, _state} = BashBackend.execute(state, "echo", ["foo", "bar"], [])
+      assert_receive {:command_event, {:output, output}}, @event_timeout
+      assert output =~ "foo bar"
+      assert_receive {:command_finished, _}, @event_timeout
+      :ok = BashBackend.terminate(state)
+    end
+  end
+
+  describe "cancellation" do
+    test "cancel interrupts a long-running bash loop", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      # Use a pure-bash loop (no external seq) so it runs under :no_external policy
+      {:ok, :accepted} =
+        ShellSessionServer.run_command(session_id, "i=0; while [ $i -lt 10000 ]; do echo $i; i=$((i+1)); done")
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      # Let some output come through
+      assert_receive {:jido_shell_session, ^session_id, {:output, _}}, @event_timeout
+
+      # Cancel
+      {:ok, :cancelled} = ShellSessionServer.cancel(session_id)
+      assert_receive {:jido_shell_session, ^session_id, :command_cancelled}, @event_timeout
+
+      # Session should be reusable after cancellation
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo alive")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "alive\n", ""} = receive_output(session_id)
+    end
+
+    test "cancel preserves session state (variables, cwd)", %{workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/testdir")
+      session_id = start_session(wid)
+
+      # Set a variable and cd
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "x=42; cd /testdir")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, {:cwd_changed, "/testdir"}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, :command_done}, @event_timeout
+
+      # Start and cancel a long command (pure-bash loop)
+      {:ok, :accepted} =
+        ShellSessionServer.run_command(session_id, "i=0; while [ $i -lt 10000 ]; do echo $i; i=$((i+1)); done")
+
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, {:output, _}}, @event_timeout
+      {:ok, :cancelled} = ShellSessionServer.cancel(session_id)
+      assert_receive {:jido_shell_session, ^session_id, :command_cancelled}, @event_timeout
+
+      # Variable should be preserved
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, ~s/echo "$x"/)
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "42\n", ""} = receive_output(session_id)
+    end
+  end
+
+  describe "cd through function shim" do
+    test "cd inside a function propagates to the outer session", %{workspace_id: wid} do
+      :ok = VFS.mkdir(wid, "/inner")
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "go() { cd /inner; }; go")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, {:cwd_changed, "/inner"}}, @event_timeout
+      assert_receive {:jido_shell_session, ^session_id, :command_done}, @event_timeout
+
+      {:ok, state} = ShellSessionServer.get_state(session_id)
+      assert state.cwd == "/inner"
+    end
+  end
+
+  describe "exec builtin" do
+    test "exec returns {:ok, nil} (treated as successful completion)", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "exec echo done")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      # exec replaces the shell; the backend should treat it as a clean exit
+      assert_receive {:jido_shell_session, ^session_id, _event}, @event_timeout
+    end
+  end
+
+  describe "error command result" do
+    test "a command producing {:error, execution} reports exit_code error", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      # `false` returns exit code 1 which triggers the {:error, execution} path
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "false")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+
+      assert_receive {:jido_shell_session, ^session_id, {:error, %Jido.Shell.Error{code: {:command, :exit_code}}}},
+                     @event_timeout
+    end
+  end
+
+  describe "cwd unchanged" do
+    test "no cwd_changed event when cwd stays the same", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo stable")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "stable\n", ""} = receive_output(session_id)
+
+      # No cwd_changed event should have been sent
+      refute_receive {:jido_shell_session, ^session_id, {:cwd_changed, _}}, 100
+    end
+  end
+
+  describe "stderr streaming" do
+    test "stderr is routed through {:output_stderr, _}", %{workspace_id: wid} do
+      session_id = start_session(wid)
+
+      {:ok, :accepted} = ShellSessionServer.run_command(session_id, "echo error_msg >&2")
+      assert_receive {:jido_shell_session, ^session_id, {:command_started, _}}, @event_timeout
+      assert {:ok, "", stderr} = receive_output(session_id)
+      assert stderr =~ "error_msg"
+    end
+  end
+
+  defp wait_until(fun, timeout, interval \\ 20, elapsed \\ 0)
+
+  defp wait_until(_fun, timeout, _interval, elapsed) when elapsed >= timeout, do: :timeout
+
+  defp wait_until(fun, timeout, interval, elapsed) do
+    if fun.() do
+      :ok
+    else
+      Process.sleep(interval)
+      wait_until(fun, timeout, interval, elapsed + interval)
+    end
+  end
+end