HostExecutionPolicy Class — langchain Architecture

Architecture documentation for the HostExecutionPolicy class in _execution.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  489b9db3_8221_6964_1dd6_b6ee6fd2e645["HostExecutionPolicy"]
  54947cfa_44ad_df6e_ebe6_8908dfdafc92["BaseExecutionPolicy"]
  489b9db3_8221_6964_1dd6_b6ee6fd2e645 -->|extends| 54947cfa_44ad_df6e_ebe6_8908dfdafc92
  a7063f14_dbac_d9f7_9cc6_a20aa09f5461["_execution.py"]
  489b9db3_8221_6964_1dd6_b6ee6fd2e645 -->|defined in| a7063f14_dbac_d9f7_9cc6_a20aa09f5461
  67f4d214_e3bb_d1b8_e7f8_0f36f4a3e6aa["__post_init__()"]
  489b9db3_8221_6964_1dd6_b6ee6fd2e645 -->|method| 67f4d214_e3bb_d1b8_e7f8_0f36f4a3e6aa
  ea98a366_f701_dc2d_901d_9954c6c58e48["spawn()"]
  489b9db3_8221_6964_1dd6_b6ee6fd2e645 -->|method| ea98a366_f701_dc2d_901d_9954c6c58e48
  4369ac03_9b6d_38a2_6328_517f16dfa130["_create_preexec_fn()"]
  489b9db3_8221_6964_1dd6_b6ee6fd2e645 -->|method| 4369ac03_9b6d_38a2_6328_517f16dfa130
  6592225b_0ebb_d535_96fb_1f481b669d2c["_apply_post_spawn_limits()"]
  489b9db3_8221_6964_1dd6_b6ee6fd2e645 -->|method| 6592225b_0ebb_d535_96fb_1f481b669d2c
  bad031b8_5bd0_bccc_ddb5_454177e775ee["_can_use_prlimit()"]
  489b9db3_8221_6964_1dd6_b6ee6fd2e645 -->|method| bad031b8_5bd0_bccc_ddb5_454177e775ee

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/_execution.py lines 92–187

class HostExecutionPolicy(BaseExecutionPolicy):
    """Run the shell directly on the host process.

    This policy is best suited for trusted or single-tenant environments (CI jobs,
    developer workstations, pre-sandboxed containers) where the agent must access the
    host filesystem and tooling without additional isolation. Enforces optional CPU and
    memory limits to prevent runaway commands but offers **no** filesystem or network
    sandboxing; commands can modify anything the process user can reach.

    On Linux platforms resource limits are applied with `resource.prlimit` after the
    shell starts. On macOS, where `prlimit` is unavailable, limits are set in a
    `preexec_fn` before `exec`. In both cases the shell runs in its own process group
    so timeouts can terminate the full subtree.
    """

    cpu_time_seconds: int | None = None
    memory_bytes: int | None = None
    create_process_group: bool = True

    _limits_requested: bool = field(init=False, repr=False, default=False)

    def __post_init__(self) -> None:
        super().__post_init__()
        if self.cpu_time_seconds is not None and self.cpu_time_seconds <= 0:
            msg = "cpu_time_seconds must be positive if provided."
            raise ValueError(msg)
        if self.memory_bytes is not None and self.memory_bytes <= 0:
            msg = "memory_bytes must be positive if provided."
            raise ValueError(msg)
        self._limits_requested = any(
            value is not None for value in (self.cpu_time_seconds, self.memory_bytes)
        )
        if self._limits_requested and not _HAS_RESOURCE:
            msg = (
                "HostExecutionPolicy cpu/memory limits require the Python 'resource' module. "
                "Either remove the limits or run on a POSIX platform."
            )
            raise RuntimeError(msg)

    def spawn(
        self,
        *,
        workspace: Path,
        env: Mapping[str, str],
        command: Sequence[str],
    ) -> subprocess.Popen[str]:
        process = _launch_subprocess(
            list(command),
            env=env,
            cwd=workspace,
            preexec_fn=self._create_preexec_fn(),
            start_new_session=self.create_process_group,
        )
        self._apply_post_spawn_limits(process)
        return process

    def _create_preexec_fn(self) -> typing.Callable[[], None] | None:
        if not self._limits_requested or self._can_use_prlimit():
            return None

        def _configure() -> None:  # pragma: no cover - depends on OS
            if self.cpu_time_seconds is not None:
                limit = (self.cpu_time_seconds, self.cpu_time_seconds)
                resource.setrlimit(resource.RLIMIT_CPU, limit)
            if self.memory_bytes is not None:
                limit = (self.memory_bytes, self.memory_bytes)
                if hasattr(resource, "RLIMIT_AS"):
                    resource.setrlimit(resource.RLIMIT_AS, limit)
                elif hasattr(resource, "RLIMIT_DATA"):
                    resource.setrlimit(resource.RLIMIT_DATA, limit)

        return _configure

    def _apply_post_spawn_limits(self, process: subprocess.Popen[str]) -> None:
        if not self._limits_requested or not self._can_use_prlimit():
            return
        if not _HAS_RESOURCE:  # pragma: no cover - defensive
            return
        pid = process.pid
        try:
            prlimit = typing.cast("typing.Any", resource).prlimit