Source code for ralph._session_runtime_runtime

"""Runtime context for Ralph-managed standalone agent sessions."""

from __future__ import annotations

from dataclasses import replace
from typing import TYPE_CHECKING, cast
from uuid import uuid4

from loguru import logger

from ralph.agents.invoke import (
    InvokeOptions,
    InvokeRuntimeOptions,
    build_invoke_options_from_config,
)
from ralph.agents.invoke._direct_mcp_recovery import (
    default_direct_mcp_retry_limit,
    iter_with_direct_mcp_recovery,
    summarize_retry_failure_evidence,
)
from ralph.config.enums import AgentTransport
from ralph.mcp.protocol.env import AGENT_LABEL_SCOPE_ENV, MCP_ENDPOINT_ENV, MCP_RUN_ID_ENV
from ralph.mcp.protocol.session import AgentSession
from ralph.mcp.server.lifecycle import McpServerExtras, SessionBridgeLike
from ralph.mcp.session_plan import SessionMcpPlan, SessionModelOpts

from ._session_runtime_deps import ManagedAgentSessionDeps

if TYPE_CHECKING:
    from collections.abc import Callable, Iterable
    from pathlib import Path

    from ralph.agents.idle_watchdog import WaitingStatusListener
    from ralph.config.models import AgentConfig, UnifiedConfig
    from ralph.phases.required_artifacts import RequiredArtifact
    from ralph.policy.models import AgentsPolicy

    from ._session_runtime_request import ManagedAgentSessionRequest


[docs] class ManagedAgentSessionRuntime: """Host-owned context for running prompt-like mini workflows through Ralph. A :class:`ManagedAgentSessionRuntime` is the reusable runtime seam Ralph exposes for tools that need to drive a single, isolated agent session without entering the full policy-driven pipeline. Callers own the higher-level host loop (e.g. plan/verify helpers, ad-hoc prompt runners, or external tooling); the runtime owns the MCP bridge, the per-session environment, agent invocation wiring, retry handling, and optional system-prompt materialization. Instances are constructed via :meth:`open` (a classmethod) so the bridge and session id lifecycle are managed in one place. The runtime is a context manager: ``with runtime: ...`` shuts down the bridge on exit (also available directly via :meth:`close`). Attributes (set in :meth:`open`, treated as read-only afterwards): config: The fully-merged :class:`UnifiedConfig` driving the general settings of every :meth:`invoke_prompt_file` turn. workspace_root: The repository-relative workspace the agent subprocess treats as its working directory. agent_config: The selected :class:`AgentConfig` identifying the agent CLI to invoke. request: The :class:`ManagedAgentSessionRequest` that named the session; retained verbatim for diagnostics and checkpoint rehydration. bridge: The :class:`SessionBridgeLike` started by :meth:`open`. Owns the MCP server endpoint the agent reaches as ``MCP_ENDPOINT``. agent_session: The :class:`AgentSession` carrying the unique session id, run id, declared capabilities, and model identity. system_prompt_file: Resolved path to a system-prompt file when the request named one; ``None`` when the agent runs without an explicit system prompt. deps: The :class:`ManagedAgentSessionDeps` dependency bundle in use (production defaults, or the test stub passed to :meth:`open`). Invariants: - The class is constructed only via :meth:`open`; the regular ``__init__`` is reserved for the runtime's internal use to keep construction injectable. - The MCP bridge owned by the runtime is alive between construction and :meth:`close`; callers must invoke :meth:`close` (or use the context-manager protocol) regardless of how their host loop exits. """ def __init__( self, *, config: UnifiedConfig, workspace_root: Path, agent_config: AgentConfig, request: ManagedAgentSessionRequest, bridge: SessionBridgeLike, agent_session: AgentSession, system_prompt_file: str | None, deps: ManagedAgentSessionDeps, ) -> None: self._config = config self._workspace_root = workspace_root self._agent_config = agent_config self._request = request self._bridge = bridge self._agent_session = agent_session self._system_prompt_file = system_prompt_file self._deps = deps
[docs] @classmethod def open( cls, *, config: UnifiedConfig, workspace_root: Path, agent_config: AgentConfig, request: ManagedAgentSessionRequest, deps: ManagedAgentSessionDeps | None = None, agents_policy: AgentsPolicy | None = None, ) -> ManagedAgentSessionRuntime: """Construct a managed Ralph session ready for :meth:`invoke_prompt_file`. ``open`` allocates the session id, starts the MCP bridge that the agent will talk to, and (optionally) materializes a system-prompt file. It is the only sanctioned way to build a :class:`ManagedAgentSessionRuntime`; the regular ``__init__`` is reserved for the runtime's internal use so it can be reasoned about as a pure dependency-injected bundle. Keyword Args: config: The fully-merged :class:`ralph.config.models.UnifiedConfig` that drives general settings (verbosity, retry limits, JSON parser). Reused for every :meth:`invoke_prompt_file` call made through the runtime. workspace_root: Filesystem location the agent session will treat as its working directory. Forwarded to the workspace factory in ``deps`` to produce a :class:`Workspace` that the MCP bridge will hand to tools. agent_config: Selected :class:`ralph.config.models.AgentConfig` that names which agent CLI to invoke (e.g. Claude, Codex, OpenCode), which transport to use, and which optional model flag to pass through. request: Caller-supplied :class:`ManagedAgentSessionRequest` that names the session id prefix, ``drain``, capabilities, system prompt, and any pre-resolved session plan. deps: Optional :class:`ManagedAgentSessionDeps` bundle overriding one or more collaborator boundaries (workspace factory, MCP server starter, agent invoker, system-prompt materializer, bridge shutdown). Pass a stubbed bundle in tests to avoid real subprocesses and filesystem access. When ``None`` the production defaults from :class:`ManagedAgentSessionDeps` are used. agents_policy: Optional :class:`ralph.policy.models.AgentsPolicy` used to resolve MCP capabilities and access modes when ``request.capabilities`` and ``request.session_mcp_plan`` are both ``None``. Falls back to the policy embedded in ``config`` when omitted. Returns: A fully wired :class:`ManagedAgentSessionRuntime` whose MCP bridge is already listening on a private endpoint. The caller owns the returned runtime and must invoke :meth:`close` (or use it as a context manager) so the bridge shuts down on exit. Raises: Exception: Any failure during bridge start or system-prompt materialization is re-raised **after** any partially-started bridge has been shut down via ``deps.shutdown_bridge``, so the caller never inherits a half-initialized MCP listener. Side Effects: - Starts an MCP server subprocess (or in-memory bridge, if the ``deps.start_mcp_server`` override returns one) bound to a private endpoint. - Captures a fresh ``run_id`` (UUID4) that is exposed through ``MCP_RUN_ID_ENV`` and ``AGENT_LABEL_SCOPE_ENV`` to the agent subprocess. - May write a system-prompt file under ``workspace_root`` via ``deps.materialize_system_prompt``. """ runtime_deps = deps or ManagedAgentSessionDeps() session_plan = _resolve_session_plan( request=request, deps=runtime_deps, agent_config=agent_config, workspace_root=workspace_root, agents_policy=agents_policy, ) agent_session = AgentSession( session_id=f"{request.session_id_prefix}-{uuid4().hex[:8]}", run_id=str(uuid4()), drain=request.drain, capabilities=set(session_plan.capabilities), model_identity=session_plan.model_identity, stored_capability_profile=session_plan.capability_profile, ) workspace = runtime_deps.workspace_factory(workspace_root) bridge = runtime_deps.start_mcp_server( agent_session, workspace, McpServerExtras(phase=request.drain, extra_env=session_plan.server_env), ) try: system_prompt_file = None if request.system_prompt_name is not None: system_prompt_file = runtime_deps.materialize_system_prompt( workspace_root, request.system_prompt_name, request.default_current_prompt, ) return cls( config=config, workspace_root=workspace_root, agent_config=agent_config, request=request, bridge=bridge, agent_session=agent_session, system_prompt_file=system_prompt_file, deps=runtime_deps, ) except Exception: runtime_deps.shutdown_bridge(bridge) raise
def __enter__(self) -> ManagedAgentSessionRuntime: return self def __exit__(self, exc_type: object, exc: object, tb: object) -> None: del exc_type, exc, tb self.close()
[docs] def close(self) -> None: """Shut down the MCP bridge owned by this session.""" self._deps.shutdown_bridge(self._bridge)
[docs] def invoke_prompt_file( self, prompt_file: str | Path, *, session_id: str | None = None, session_id_sink: Callable[[str], None] | None = None, required_artifact: RequiredArtifact | None = None, waiting_listener: WaitingStatusListener | None = None, permission_prompt_listener: Callable[[str], None] | None = None, extra_env: dict[str, str] | None = None, ) -> Iterable[str]: """Drive one host-owned agent turn and stream the agent's output. ``invoke_prompt_file`` resolves the per-turn ``InvokeOptions`` for the configured agent, injects the MCP endpoint / run-id / agent scope environment variables into the agent subprocess, and yields the agent's output line-by-line (logging through :func:`ralph.agents.invoke.invoke_agent`). The MCP bridge started by :meth:`open` is reachable as the ``MCP_ENDPOINT`` env value, so tools that call back into Ralph resolve back to the same bridge for the lifetime of the turn. Args: prompt_file: Path to the prompt file the agent will be asked to read. Resolved relative to the runtime's ``workspace_root`` by the underlying invocation mechanism; absolute paths are honored as-is. Keyword Args: session_id: Optional explicit session id forwarded to the agent. When ``None`` the agent runtime generates one; the actual id is reported through ``session_id_sink``. session_id_sink: Optional callback invoked as soon as the agent makes its session id observable (i.e. after the first handshake). Receives the resolved session id so the host can store it for resumption, logging, or checkpoint writes. required_artifact: Optional :class:`ralph.phases.required_artifacts.RequiredArtifact` declaration used by the agent runtime to gate completion; the turn fails fast if no matching artifact is produced. Most host loops leave this ``None``. waiting_listener: Optional callback invoked when the agent reports it is waiting on a tool call. Used by progress UIs to surface the wait state without depending on stdout parsing. permission_prompt_listener: Optional callback invoked when the agent prompts for permission (e.g. before an action that requires operator approval). Implementations should return the agent's answer or raise to abort the turn. extra_env: Optional additional environment variables merged into the agent subprocess environment, **excluding** the three reserved names ``MCP_ENDPOINT``, ``MCP_RUN_ID``, and ``AGENT_LABEL_SCOPE`` (these are owned by the runtime and always set by ``open``). Returns: Iterable[str]: A lazy iterator over the agent's streamed output lines. The iterator is wrapped by :func:`ralph.agents.invoke._direct_mcp_recovery.iter_with_direct_mcp_recovery`, which transparently retries on direct-MCP failures up to ``config.general.max_same_agent_retries`` attempts. Retry events are emitted through :func:`loguru.logger.warning`. Raises: Exception: Propagated from the underlying agent invocation or from the recovery iterator once retries are exhausted. The MCP bridge started by :meth:`open` is **not** shut down on failure; callers are expected to invoke :meth:`close` (or use the runtime as a context manager) regardless of outcome. Side Effects: - Launches the configured agent CLI as a subprocess. - Injects ``MCP_ENDPOINT``, ``MCP_RUN_ID``, and ``AGENT_LABEL_SCOPE`` into the subprocess environment so the agent can reach the MCP bridge owned by this runtime. - May invoke ``session_id_sink`` and the listener callbacks as the turn progresses. - On retryable failure, may re-launch the agent subprocess and may reset the bridge's tool registry (when one is exposed via ``bridge.reset_tool_registry``). Example: >>> with ManagedAgentSessionRuntime.open( ... config=config, ... workspace_root=repo_root, ... agent_config=agent_config, ... request=ManagedAgentSessionRequest( ... session_id_prefix="plan", drain="planning" ... ), ... ) as runtime: ... for line in runtime.invoke_prompt_file("your-prompt-file.md"): ... print(line) """ runtime_env = { str(MCP_ENDPOINT_ENV): self._bridge.agent_endpoint_uri(), str(MCP_RUN_ID_ENV): self._agent_session.run_id, str(AGENT_LABEL_SCOPE_ENV): self._agent_session.run_id, } if extra_env is not None: reserved_env = { str(MCP_ENDPOINT_ENV), str(MCP_RUN_ID_ENV), str(AGENT_LABEL_SCOPE_ENV), } runtime_env.update( {key: value for key, value in extra_env.items() if key not in reserved_env} ) options = build_invoke_options_from_config( self._config.general, InvokeRuntimeOptions( verbose=False, show_progress=False, workspace_path=self._workspace_root, extra_env=runtime_env, pure=self._agent_config.transport == AgentTransport.OPENCODE, session_id=session_id, system_prompt_file=self._system_prompt_file, waiting_listener=waiting_listener, permission_prompt_listener=permission_prompt_listener, required_artifact=required_artifact, ), ) max_retries = default_direct_mcp_retry_limit(self._config.general.max_same_agent_retries) reset_tool_registry = _reset_tool_registry_callback(self._bridge) base_session_id = session_id def _invoke_with_retry_session(retry_session_id: str | None) -> Iterable[str]: return self._deps.invoke_agent( self._agent_config, str(prompt_file), _with_session_id(options, retry_session_id or base_session_id), ) return iter_with_direct_mcp_recovery( _invoke_with_retry_session, max_retries=max_retries, reset_tool_registry=reset_tool_registry, on_session_observed=session_id_sink, on_retry_failure=lambda lines: logger.warning( "Retrying managed agent session after retryable failure: {}", summarize_retry_failure_evidence(lines), ), )
def _with_session_id(options: InvokeOptions, session_id: str | None) -> InvokeOptions: return replace(options, session_id=session_id) def _reset_tool_registry_callback(bridge: object) -> Callable[[], object] | None: reset_tool_registry_obj: object = getattr(bridge, "reset_tool_registry", None) if not callable(reset_tool_registry_obj): return None return cast("Callable[[], object]", reset_tool_registry_obj) def _resolve_session_plan( *, request: ManagedAgentSessionRequest, deps: ManagedAgentSessionDeps, agent_config: AgentConfig, workspace_root: Path, agents_policy: AgentsPolicy | None, ) -> SessionMcpPlan: if request.session_mcp_plan is not None: return request.session_mcp_plan if request.capabilities is not None: return SessionMcpPlan( capabilities=request.capabilities, server_env=request.server_env, ) return deps.build_session_mcp_plan( agent_config.transport, request.drain, workspace_root, agents_policy, SessionModelOpts(model_flag=agent_config.model_flag), None, ) __all__ = ["ManagedAgentSessionRuntime"]