Function run_single

examples/run_examples.py:666–768

(example: ExampleScript)

            write_main_log_line(main_log, line)

    def run_single(example: ExampleScript) -> ExampleResult:
        relpath = example.relpath
        log_filename = f"{relpath.replace('/', '__')}.log"
        log_path = logs_dir / log_filename
        ensure_dirs(log_path, is_file=True)

        env = os.environ.copy()
        env["PATH"] = command_path
        env["PYTHONPATH"] = build_python_path(env.get("PYTHONPATH"))
        env["EXAMPLES_ARTIFACTS_DIR"] = str(artifact_dir_for_example(relpath, artifacts_dir))
        if auto_mode:
            env["EXAMPLES_INTERACTIVE_MODE"] = "auto"
            env["APPLY_PATCH_AUTO_APPROVE"] = "1"
            env.setdefault("SHELL_AUTO_APPROVE", "1")
            env.setdefault("AUTO_APPROVE_MCP", "1")

        force_prompt_stream = (not auto_mode) and ("interactive" in example.tags)
        buffer_output_local = buffer_output and not force_prompt_stream
        buffer_lines: list[str] = []
        redis_server: TemporaryRedisServer | None = None
        service_messages: list[str] = []

        with log_path.open("w", encoding="utf-8") as per_log:
            redis_server, service_messages = prepare_redis_for_example(relpath, env)
            for message in service_messages:
                per_log.write(f"[runner] {message}\n")
                if not buffer_output_local:
                    with output_lock:
                        sys.stdout.write(f"[{relpath}] [runner] {message}\n")

            try:
                proc = subprocess.Popen(
                    example.command,
                    cwd=ROOT_DIR,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    text=True,
                    env=env,
                )
                assert proc.stdout is not None

                if force_prompt_stream:
                    at_line_start = True
                    while True:
                        char = proc.stdout.read(1)
                        if char == "":
                            break
                        per_log.write(char)
                        with output_lock:
                            if at_line_start:
                                sys.stdout.write(f"[{relpath}] ")
                            sys.stdout.write(char)
                            sys.stdout.flush()
                        at_line_start = char == "\n"
                else:
                    for line in proc.stdout:
                        per_log.write(line)
                        if buffer_output_local: