How GenericAgent's code_run Tool Safely Executes Python with Timeout and Streaming Output

GenericAgent's code_run tool isolates untrusted Python code in a subprocess, streams output in real-time via daemon threads, and enforces hard timeouts through process termination to prevent blocking the agent's main loop.

The code_run tool is a core utility in the lsdefine/GenericAgent repository that enables AI agents to execute arbitrary Python or shell commands safely. Implemented in ga.py, this tool creates an isolated execution environment that prevents user code from polluting the agent's runtime while providing live feedback and automatic resource cleanup.

Subprocess Isolation and Temporary File Management

The foundation of GenericAgent's safe execution model relies on process isolation rather than in-process evaluation.

In ga.py lines 11-80, the code_run function creates a temporary file with the .ai.py extension using tempfile.NamedTemporaryFile. The supplied code is written to this file, ensuring it runs in a fresh interpreter process rather than the agent's own Python runtime. This prevents import side-effects, global state pollution, and memory leaks from affecting the agent's stability.

The subprocess command is constructed as:

cmd = [sys.executable, "-X", "utf8", "-u", tmp_path]

The -u flag forces unbuffered binary stdout and stderr, while -X utf8 ensures consistent encoding behavior. On Windows systems (os.name == 'nt'), a STARTUPINFO object with SW_HIDE is attached to prevent console windows from appearing during execution.

Real-Time Streaming Output Architecture

Unlike standard subprocess calls that wait for completion, code_run implements incremental output streaming to mimic an interactive console experience.

Daemon Thread Line Reader

A daemon thread runs the internal stream_reader function, which continuously calls proc.stdout.readline(). Each line is decoded from UTF-8 (with GBK fallback for compatibility), appended to full_stdout, and printed immediately. This architecture ensures that long-running scripts provide feedback without waiting for the process to terminate, as implemented in the main execution loop of ga.py.

Encoding and Buffer Handling

The subprocess is initialized with bufsize=0 (unbuffered) and stderr=subprocess.STDOUT to merge error streams into the output pipe. This guarantees that both standard output and errors appear in the correct chronological order in the streamed response.

Safety Mechanisms and Timeout Enforcement

Safety is enforced through multiple layers of protection designed to handle runaway code and external interruptions.

Hard Timeout Implementation

The main execution loop monitors time.time() - start_t against the specified timeout parameter. If the threshold is exceeded, process.kill() is invoked immediately, and [Timeout Error] 超时强制终止 is appended to the output log. This prevents infinite loops from blocking the agent indefinitely.

External Stop Signals

Beyond timeouts, code_run accepts a stop_signal list reference. When any truthy value is appended to this list from an external source, the main loop terminates the process and returns [Stopped] in the output. This enables cooperative cancellation between the agent's UI and the execution engine.

Header Injection for Nested Safety

If assets/code_run_header.py exists, its contents are prepended to the temporary script. This header monkey-patches subprocess.run to enforce text mode and adds a custom exception hook, ensuring that even if the user code spawns additional subprocesses, they inherit safe handling characteristics.

Resource Cleanup

The temporary script file is removed via os.remove(tmp_path) in a finally block, guaranteeing no artifacts remain on the host system regardless of execution success or failure.

Practical Usage Examples

Basic Python Execution with Streaming

from ga import code_run

script = """
import time
for i in range(3):
    print(f"Step {i}")
    time.sleep(0.5)
"""

# Execute with 10-second timeout

generator = code_run(script, code_type="python", timeout=10)
for update in generator:
    print(update)  # Live streaming output

Handling Timeout Errors

# Infinite loop that will be terminated

infinite_script = "while True: pass"

result = list(code_run(infinite_script, timeout=2))[-1]
print(result["status"])  # 'error'

print(result["stdout"])  # Contains '[Timeout Error] 超时强制终止'

Tool Integration via do_code_run

The higher-level wrapper do_code_run (lines 96-118 in ga.py) handles JSON tool calls from the LLM:

# Inside GenericAgentHandler

def tool_call_example(self):
    # Simulates receiving a tool call from the LLM

    result = self.do_code_run(
        type="python",
        code="print('Hello from GenericAgent')",
        timeout=5
    )
    # Returns structured output with exit code and status

Summary

• Process Isolation: GenericAgent's code_run executes code in separate subprocesses via temporary files, preventing state pollution in the main agent runtime.
• Live Streaming: Daemon threads read stdout line-by-line, providing real-time output without waiting for process completion.
• Timeout Protection: Hard timeouts are enforced through process.kill() when execution exceeds the specified limit, with optional external stop signals for manual interruption.
• Safe Cleanup: Temporary files are removed in finally blocks, and optional header injection ensures nested subprocess calls inherit safe execution parameters.
• Source Locations: Core logic resides in ga.py lines 11-80 (code_run) and lines 96-118 (do_code_run), with safety headers defined in assets/code_run_header.py.

Frequently Asked Questions

How does GenericAgent prevent infinite loops from blocking the agent?

The code_run function monitors execution time in a main loop and calls process.kill() when time.time() - start_t exceeds the specified timeout parameter. This hard termination occurs in the main thread while a daemon thread handles output streaming, ensuring the agent remains responsive even if user code enters an infinite loop.

What encoding does GenericAgent use for subprocess output?

GenericAgent forces UTF-8 encoding via the -X utf8 interpreter flag and attempts UTF-8 decoding on each output line, falling back to GBK if necessary. This dual-encoding approach handles both standard English output and Chinese character sets commonly encountered in multilingual environments.

Can GenericAgent execute languages other than Python?

Yes, the code_run tool supports multiple code types. While Python uses sys.executable with specific flags, the implementation also builds command lists for bash and powershell by checking the code_type parameter, allowing the agent to execute shell scripts with the same timeout and streaming protections.

Why does GenericAgent use temporary files instead of exec() or eval()?

Executing code via exec() or eval() runs within the agent's own Python process, risking global namespace pollution, memory leaks, and security vulnerabilities from __import__ or os.system calls. By writing code to temporary .ai.py files and launching fresh interpreter processes via subprocess.Popen, GenericAgent achieves true isolation where crashes or resource exhaustion in user code cannot affect the agent's stability.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}