def test_stress_remote_runtime_eval(n_eval_workers: int = 64):
    """Mimic evaluation setting to test remote runtime in a multi-processing setting."""

    def _initialize_runtime(
        runtime: Runtime,
    ):
        """Initialize the runtime for the agent.

        This function is called before the runtime is used to run the agent.
        """
        logger.info('-' * 30)
        logger.info('BEGIN Runtime Initialization Fn')
        logger.info('-' * 30)
        obs: CmdOutputObservation

        action = CmdRunAction(command="""export USER=$(whoami); echo USER=${USER} """)
        action.set_hard_timeout(600)
        logger.info(action, extra={'msg_type': 'ACTION'})
        obs = runtime.run_action(action)
        logger.info(obs, extra={'msg_type': 'OBSERVATION'})
        assert_and_raise(obs.exit_code == 0, f'Failed to export USER: {str(obs)}')

        action = CmdRunAction(command='mkdir -p /dummy_dir')
        action.set_hard_timeout(600)
        logger.info(action, extra={'msg_type': 'ACTION'})
        obs = runtime.run_action(action)
        logger.info(obs, extra={'msg_type': 'OBSERVATION'})
        assert_and_raise(
            obs.exit_code == 0,
            f'Failed to create /dummy_dir: {str(obs)}',
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            # Construct the full path for the desired file name within the temporary directory
            temp_file_path = os.path.join(temp_dir, 'dummy_file')
            # Write to the file with the desired name within the temporary directory
            with open(temp_file_path, 'w') as f:
                f.write('dummy content')

            # Copy the file to the desired location
            runtime.copy_to(temp_file_path, '/dummy_dir/')

        logger.info('-' * 30)
        logger.info('END Runtime Initialization Fn')
        logger.info('-' * 30)

    def _process_instance(
        instance: pd.Series,
        metadata: EvalMetadata,
        reset_logger: bool = True,
    ) -> EvalOutput:
        config = get_config()

        # Setup the logger properly, so you can run multi-processing to parallelize the evaluation
        if reset_logger:
            log_dir = os.path.join(metadata.eval_output_dir, 'infer_logs')
            reset_logger_for_multiprocessing(logger, instance.instance_id, log_dir)
        else:
            logger.info(f'Starting evaluation for instance {instance.instance_id}.')

        runtime = create_runtime(config, headless_mode=True)
        call_async_from_sync(runtime.connect)

        try:
            _initialize_runtime(runtime)

            instruction = 'dummy instruction'
            agent = Agent.get_cls(metadata.agent_class)(
                llm=LLM(config=metadata.llm_config),
                config=config.get_agent_config(metadata.agent_class),
            )

            def next_command(*args, **kwargs):
                return CmdRunAction(command='ls -lah')

            agent.step = MagicMock(side_effect=next_command)

            # Here's how you can run the agent (similar to the `main` function) and get the final task state
            state: State | None = asyncio.run(
                run_controller(
                    config=config,
                    initial_user_action=MessageAction(content=instruction),
                    runtime=runtime,
                    fake_user_response_fn=AGENT_CLS_TO_FAKE_USER_RESPONSE_FN[
                        metadata.agent_class
                    ],
                    agent=agent,
                )
            )

            # if fatal error, throw EvalError to trigger re-run
            if (
                state.last_error
                and 'fatal error during agent execution' in state.last_error
                and 'stuck in a loop' not in state.last_error
            ):
                raise EvalException('Fatal error detected: ' + state.last_error)

        finally:
            runtime.close()

        test_result = {}
        if state is None:
            raise ValueError('State should not be None.')
        histories = [event_to_dict(event) for event in state.history]
        metrics = state.metrics.get() if state.metrics else None

        # Save the output
        output = EvalOutput(
            instance_id=instance.instance_id,
            instruction=instruction,
            instance=instance.to_dict(),  # SWE Bench specific
            test_result=test_result,
            metadata=metadata,
            history=histories,
            metrics=metrics,
            error=state.last_error if state and state.last_error else None,
        )
        return output

    llm_config = LLMConfig()
    metadata = make_metadata(
        llm_config,
        'dummy_dataset_descrption',
        'CodeActAgent',
        max_iterations=10,
        eval_note='dummy_eval_note',
        eval_output_dir='./dummy_eval_output_dir',
        details={},
    )

    # generate 300 random dummy instances
    dummy_instance = pd.DataFrame(
        {
            'instance_id': [f'dummy_instance_{i}' for i in range(300)],
        }
    )

    output_file = os.path.join(metadata.eval_output_dir, 'output.jsonl')
    instances = prepare_dataset(
        dummy_instance, output_file, eval_n_limit=len(dummy_instance)
    )

    run_evaluation(instances, metadata, output_file, n_eval_workers, _process_instance)
Mimic evaluation setting to test remote runtime in a multi-processing setting.
tests/runtime/test_stress_remote_runtime.py