DaytonaCodeExecutor

def execute_code_blocks(self, code_blocks: list[CodeBlock]) -> DaytonaCodeResult:
    """Execute code blocks sequentially inside the Daytona sandbox.

    Python blocks are executed via process.code_run(). All other languages
    are written to a temp file and run with the appropriate runtime binary
    via process.exec().

    Execution stops on the first failure and returns that block's result.
    On full success, outputs from all blocks are joined with newlines.

    Args:
        code_blocks: The code blocks to execute.

    Returns:
        DaytonaCodeResult with exit_code, combined output, and sandbox_id.
    """
    if not code_blocks:
        return DaytonaCodeResult(exit_code=0, output="", sandbox_id=self._sandbox.id)

    outputs: list[str] = []

    for code_block in code_blocks:
        lang = self._normalize_language(code_block.language)

        if lang not in self.SUPPORTED_LANGUAGES:
            return DaytonaCodeResult(
                exit_code=1,
                output=(
                    f"Unsupported language: {code_block.language!r}. "
                    f"Supported languages: {', '.join(self.SUPPORTED_LANGUAGES)}"
                ),
                sandbox_id=self._sandbox.id,
            )

        logger.info("Executing %s code block in sandbox %s", lang, self._sandbox.id)
        try:
            if lang == "python":
                response = self._sandbox.process.code_run(code_block.code, timeout=self._timeout)
            else:
                ext = self._LANG_EXT[lang]
                script_path = f"/tmp/ag2_{uuid.uuid4().hex}.{ext}"

                self._sandbox.fs.upload_file(code_block.code.encode("utf-8"), script_path)

                cmd = f"{self._LANG_CMD[lang]} {script_path}"
                response = self._sandbox.process.exec(cmd, timeout=self._timeout)

                # Best-effort cleanup — don't let a stale file mask the real result.
                try:
                    self._sandbox.fs.delete_file(script_path)
                except DaytonaNotFoundError:
                    pass  # File already gone — expected
                except Exception as e:
                    logger.debug("Failed to delete temp script %s: %s", script_path, e)

        except DaytonaTimeoutError as e:
            logger.warning("Execution timed out for %s code block in sandbox %s: %s", lang, self._sandbox.id, e)
            return DaytonaCodeResult(
                exit_code=1,
                output=f"Execution timed out: {e}",
                sandbox_id=self._sandbox.id,
            )
        except DaytonaRateLimitError as e:
            logger.warning("Rate limit hit executing %s code block in sandbox %s: %s", lang, self._sandbox.id, e)
            return DaytonaCodeResult(
                exit_code=1,
                output=f"Daytona rate limit exceeded: {e}",
                sandbox_id=self._sandbox.id,
            )
        except DaytonaError as e:
            logger.warning("Daytona error executing %s code block in sandbox %s: %s", lang, self._sandbox.id, e)
            return DaytonaCodeResult(
                exit_code=1,
                output=f"Daytona error: {e}",
                sandbox_id=self._sandbox.id,
            )
        except Exception as e:
            logger.warning("Unexpected error executing %s code block in sandbox %s: %s", lang, self._sandbox.id, e)
            return DaytonaCodeResult(
                exit_code=1,
                output=f"Error executing code block: {e}",
                sandbox_id=self._sandbox.id,
            )
        if response.exit_code != 0:
            return DaytonaCodeResult(
                exit_code=response.exit_code,
                output=response.result,
                sandbox_id=self._sandbox.id,
            )

        outputs.append(response.result)

    return DaytonaCodeResult(
        exit_code=0,
        output="\n".join(outputs),
        sandbox_id=self._sandbox.id,
    )

def restart(self) -> None:
    """Reset the executor by deleting the current sandbox and creating a fresh one.

    Called by agent.reset() when starting a new conversation. Clears all
    accumulated state — created files, installed packages, process state.
    """
    logger.info("Restarting Daytona executor — deleting sandbox %s", self._sandbox.id if self._sandbox else "None")
    try:
        self._sandbox.delete()
    except DaytonaNotFoundError:
        pass  # Already deleted — expected
    except Exception as e:
        logger.debug("Suppressed exception during sandbox deletion on restart: %s", e)
    self._sandbox = self._create_sandbox()

def delete(self) -> None:
    """Delete the sandbox and release all associated resources.

    Safe to call multiple times. Called automatically on process exit via atexit.
    Unregisters the atexit handler to prevent double-execution and handler accumulation.
    """
    atexit.unregister(self.delete)
    if self._sandbox is not None:
        logger.info("Deleting Daytona sandbox %s", self._sandbox.id)
        try:
            self._sandbox.delete()
        except DaytonaNotFoundError:
            pass  # Already deleted — expected
        except Exception as e:
            logger.debug("Suppressed exception during sandbox deletion: %s", e)
        self._sandbox = None  # type: ignore[assignment]