Shell Abstract Base Class

Asynchronous-ready command dispatcher  •  Virtual state management  •  Extensible shell emulation

📘 Overview

Shell is an abstract foundation for building command-line interpreters or remote execution engines. It separates state-changing virtual commands (cd, export) from external process execution. The class provides built-in context (SessionContext), observer hooks, timeout control, and a clean dispatcher pattern.

🔍 Key Design: Commands like cd and export never touch the real OS — they update the internal SessionContext (cwd, environment). All other commands are delegated to subprocesses using platform‑specific formatting via the abstract _format_command method.

🏛️ Class Architecture

ComponentDescription
context: SessionContextHolds working directory, environment variables, encoding, and runtime state.
observer: ShellObserverCallback handler for session events, command lifecycle, errors, and context changes.
default_timeout: floatDefault timeout (seconds) for external commands.
_virtual_builtins: Dict[str, Callable]Registry mapping built-in names to internal handlers (cd, export).

⚙️ Initializer

__init__(context: Optional[SessionContext] = None, observer: Optional[ShellObserver] = None, default_timeout: float = 30.0)

Creates a new shell instance with custom context and observer. If no context is provided, a fresh SessionContext is created. The observer defaults to a no‑op ShellObserver. The default_timeout defines the maximum runtime for any external subprocess.

✨ What's initialized internally

• self.context – holds environment and current working directory.
• self.observer – tracks events: session start/end, context changes, command results.
• self._virtual_builtins – maps "cd" → _handle_cd and "export" → _handle_export.

🔁 Core Public API

run(command, timeout=None) → CommandResult

Main dispatcher. Checks if command.executable matches a virtual built‑in. If yes, invokes the internal handler; otherwise, delegates to external execution (_run_external). Returns a CommandResult containing stdout, stderr, return code, and execution time.

__enter__() / __exit__(...)

Context manager support. On enter, calls observer.on_session_start(class_name). On exit, calls observer.on_session_end(...) with any exception details. Ideal for scoped shell sessions.

_format_command(executable, args) → list[str] abstract

Each child shell must implement this method to transform the command into the target shell syntax (e.g., Bash: ["-c", "executable args"], PowerShell: ["-Command", ...]). Used during external execution to build the final argument vector.

📁 Virtual Built‑in Commands (State Management)

Intercepted commands that mutate SessionContext without spawning OS processes.

_handle_cd(command) → CommandResult

Changes the current working directory inside the context. Resolves relative paths against current cwd, expands ~ to home directory, and validates that the target exists and is a directory. On success, updates self.context (using replace() for immutability) and notifies observer with on_context_change("cwd", new_path). Returns success message or error.

_handle_export(command) → CommandResult

Updates environment variables. If no arguments are given, returns a string representation of the current environment. Otherwise, parses key=value pairs via _parse_env_vars and merges them into a new environment dict. For each updated variable, triggers observer.on_context_change(f"env.{key}", value). Returns a count of updated variables.

🖥️ External Execution Logic

_run_external(command, timeout) → CommandResult

Handles any non‑virtual command by executing a real OS subprocess.

Step by step:
1️⃣ _validate_executable – resolves full path using shutil.which (raises CommandNotFoundError if missing).
2️⃣ Builds final arguments using ArgumentBuilder + _format_command to respect shell‑specific syntax.
3️⃣ Notifies observer via on_command_start before execution.
4️⃣ Runs subprocess.run with current context’s cwd, env, encoding, and timeout.
5️⃣ Captures stdout/stderr, return code, execution time, and returns a CommandResult.
6️⃣ Catches CommandNotFoundError (return code 127) and TimeoutExpired errors gracefully.

🛠️ Helper Methods & Utilities

MethodDescription
_validate_executable(executable)Uses shutil.which to locate absolute path; raises CommandNotFoundError if not found.
_resolve_path(args)Converts command arguments into a Path object. Defaults to home directory. Expands user (~) and resolves relative paths against current cwd.
_parse_env_vars(args)Parses a list of strings like ["KEY1=value", "KEY2=value2"] into a dictionary. Splits on the first '=' only.
_notify_error(message, error, return_code)Triggers observer.on_error and returns a CommandResult with error message and given return code.
⚡ Performance & safety: Context updates use replace() (immutable approach). Virtual commands never invoke the OS, making state changes extremely fast and side‑effect free.

📦 Supporting Classes (inferred)

While not fully implemented in this snippet, the following types are essential for the Shell abstraction:

🧪 Example: Implementing a Concrete Shell (Bash)

class BashShell(Shell):
    def _format_command(self, executable: str, args: list[str]) -> list[str]:
        # For bash, we forward the full command as a single string via -c
        full_cmd = f"{executable} {' '.join(args)}" if args else executable
        return ["-c", full_cmd]

# Usage
with BashShell(default_timeout=10.0) as shell:
    result = shell.run(Command("ls", ["-la"]))
    print(result.standard_output)
    shell.run(Command("cd", ["/tmp"]))   # virtual cd
    shell.run(Command("export", ["EDITOR=nano"]))

The observer can be used to log every command or track real-time context changes.

📐 Design & Extension Points