"""SubprocessAgentExecutor — asyncio subprocess implementation of AgentExecutor."""
from __future__ import annotations
import asyncio
import contextlib
import os
import time
from pathlib import Path
from subprocess import PIPE as _PIPE
from subprocess import STDOUT as _STDOUT
from typing import TYPE_CHECKING
from ralph.agents.executor import ExecutorError, WorkerResult
from ralph.display.activity_router import ActivityRouter, detect_provider_from_command
from ralph.display.line_sanitizer import sanitize_display_line
from ralph.display.raw_overflow import DEFAULT_MAX_OVERFLOW_FILE_BYTES, RawOverflowLog
from ralph.mcp.protocol.env import AGENT_LABEL_SCOPE_ENV
from ralph.mcp.server._activity_sink import (
reset_subagent_sink,
set_subagent_sink,
)
from ralph.pipeline.worker_state import WorkerStatus
from ralph.process.manager import ProcessManager, SpawnOptions, get_process_manager
from ralph.process.manager._process_status import _TERMINAL_STATUSES
if TYPE_CHECKING:
from collections.abc import Callable, Mapping, Sequence
from contextvars import Token
from ralph.display.activity_model import ActivityProvider
from ralph.interrupt.asyncio_bridge import SignalBridge
from ralph.pipeline.work_units import WorkUnit
from ralph.process.manager._managed_async_process import ManagedAsyncProcess
def agent_process_label(unit_id: str, env: dict[str, str] | None = None) -> str:
"""Return the full process label for the root subprocess of a work unit."""
scope = None if env is None else env.get(str(AGENT_LABEL_SCOPE_ENV))
if scope:
return f"agent:{scope}:{unit_id}:root"
return f"agent:{unit_id}:root"
def agent_process_label_prefix(unit_id: str, env: dict[str, str] | None = None) -> str:
"""Return the label prefix shared by all child processes of a work unit."""
scope = None if env is None else env.get(str(AGENT_LABEL_SCOPE_ENV))
if scope:
return f"agent:{scope}:{unit_id}:"
return f"agent:{unit_id}:"
[docs]
class SubprocessAgentExecutor:
"""AgentExecutor that spawns a subprocess in its own process group.
Uses ProcessManager.spawn_async with start_new_session=True so the child
gets its own process group, enabling escalating tree-kill on cancellation.
Success or failure is determined by the coordinator from empirical evidence
(artifact submission, git changes) — never from this executor's exit code.
"""
def __init__(
self,
command: Sequence[str],
*,
signal_bridge: SignalBridge | None = None,
cwd: Path | None = None,
extra_env: Mapping[str, str] | None = None,
activity_router: ActivityRouter | None = None,
raw_overflow_root: Path | None = None,
subagent_sink: Callable[[str], None] | None = None,
_pm: ProcessManager | None = None,
) -> None:
self._command = tuple(command)
self._signal_bridge = signal_bridge
self._cwd = cwd
self._extra_env = extra_env
self.activity_router = activity_router
self._raw_overflow_root = raw_overflow_root
# Optional watchdog-style subagent activity sink. The
# ``ActivityRouter.push_raw_line`` path calls
# ``invoke_subagent_sink(summary)`` from the
# ``_activity_sink`` contextvar, so the executor registers
# the supplied sink into that contextvar at run() time and
# resets it on exit. This way the production
# ``SubprocessAgentExecutor -> ActivityRouter`` path keeps
# the watchdog's ``record_subagent_work`` channel fresh
# without relying on tests to install the sink manually.
self._subagent_sink = subagent_sink
self._subagent_sink_token: Token[Callable[[str], None] | None] | None = None
self._raw_logs: dict[str, RawOverflowLog] = {} # bounded-accumulator-ok: drained
self._pm = _pm
def _get_raw_log(self, unit_id: str) -> RawOverflowLog:
if unit_id not in self._raw_logs:
root = self._raw_overflow_root
if root is None:
root = self._cwd if self._cwd is not None else Path.cwd()
self._raw_logs[unit_id] = RawOverflowLog(
root, unit_id, max_bytes=DEFAULT_MAX_OVERFLOW_FILE_BYTES
)
return self._raw_logs[unit_id]
[docs]
def drop_unit(self, unit_id: str) -> None:
"""Release per-unit state so long parallel sessions don't accumulate state across waves.
Removes the unit's raw overflow log entry from ``self._raw_logs``
so the memory the log holds (up to ``DEFAULT_MAX_OVERFLOW_FILE_BYTES``
per unit) is released when the unit is no longer needed. Calls
``close()`` on the log first so any buffered tail bytes reach
disk deterministically. Safe to call for a unit that was never
added; it just no-ops.
"""
raw_log = self._raw_logs.pop(unit_id, None)
if raw_log is not None:
raw_log.close()
async def run(
self,
unit: WorkUnit,
*,
on_output: Callable[[str], None],
on_status: Callable[[WorkerStatus], None],
) -> WorkerResult:
on_status(WorkerStatus.RUNNING)
start_time = time.monotonic()
last_line: str = ""
activity_provider: ActivityProvider = detect_provider_from_command(list(self._command))
# Register the watchdog-style subagent sink so the
# ``ActivityRouter.push_raw_line -> invoke_subagent_sink``
# path actually reaches a sink. The token is captured so the
# finally block can reset the contextvar even when the
# executor is cancelled mid-drain.
if self._subagent_sink is not None:
self._subagent_sink_token = set_subagent_sink(self._subagent_sink)
env = {**os.environ, **self._extra_env} if self._extra_env else None
pm = self._pm if self._pm is not None else get_process_manager()
handle: ManagedAsyncProcess | None = None
try:
handle = await pm.spawn_async(
self._command,
SpawnOptions(
cwd=str(self._cwd) if self._cwd is not None else None,
env=env,
stdout=_PIPE,
stderr=_STDOUT,
start_new_session=True,
label=agent_process_label(unit.unit_id, env),
),
)
except OSError as exc:
on_status(WorkerStatus.FAILED)
raise ExecutorError(f"Failed to start subprocess: {exc}") from exc
async def drain_output() -> None:
nonlocal last_line
assert handle is not None
assert handle.stdout is not None
async for raw_line in handle.stdout:
line = sanitize_display_line(raw_line.rstrip(b"\n"))
if self.activity_router is not None:
raw_log = self._get_raw_log(unit.unit_id)
raw_log.append(line)
raw_ref = raw_log.relative_reference(
self._raw_overflow_root or self._cwd or Path.cwd()
)
for parsed_line in line.splitlines():
stripped_line = parsed_line.strip()
if not stripped_line:
continue
self.activity_router.push_raw_line(
unit.unit_id,
stripped_line,
provider=activity_provider,
raw_reference=raw_ref,
)
else:
on_output(line)
last_line = line
try:
try:
assert handle is not None
# The handle.wait() inside the gather is bounded by the
# activity-aware idle watchdog teardown (see
# audit_activity_aware_watchdog.py — teardown_subtree is
# enforced on every fire path) and the surrounding finally
# block always terminates a non-terminal handle, so a
# healthy-but-slow agent is not killed by a hard ceiling.
# The drain_output() coroutine exits on stdout EOF.
await asyncio.gather(
drain_output(),
handle.wait(), # mcp-timeout-ok: idle-wd-bounded
)
except asyncio.CancelledError:
assert handle is not None
await handle.terminate(grace_period_s=0)
raise
finally:
if self._subagent_sink_token is not None:
with contextlib.suppress(Exception):
reset_subagent_sink(self._subagent_sink_token)
self._subagent_sink_token = None
if handle is not None and handle.record.status not in _TERMINAL_STATUSES:
with contextlib.suppress(Exception):
await handle.terminate(grace_period_s=0)
with contextlib.suppress(Exception):
await asyncio.wait_for(handle.wait(), timeout=0.5) # mcp-timeout-ok: wf-bounded
duration_ms = int((time.monotonic() - start_time) * 1000)
exit_code = handle.returncode if handle.returncode is not None else 0
return WorkerResult(
unit_id=unit.unit_id,
exit_code=exit_code,
final_message=last_line,
duration_ms=duration_ms,
)
__all__ = ["SubprocessAgentExecutor"]