feat!: remove standalone env configuration (ACP contract only)

src/rlm/api.py

@@ -1,13 +1,15 @@
 """Public Python API for running rlm agents."""
 
 from rlm import broker
-from rlm.engine import RLMEngine
 from rlm.types import RLMResult
 
 
 async def run(prompt: str) -> RLMResult:
-    """Run a single rlm agent."""
-    if broker.is_configured():
-        return await broker.run(prompt)
-    engine = RLMEngine()
-    return await engine.run(prompt)
+    """Run a recursive sub-agent through the session's broker."""
+    if not broker.is_configured():
+        raise RuntimeError(
+            "rlm.run() requires the recursion broker (available inside a "
+            "running rlm session). Standalone execution was removed: rlm is "
+            "consumed via the ACP runtime contract."
+        )
+    return await broker.run(prompt)

src/rlm/cli.py

@@ -1,70 +1,28 @@
-"""CLI entry point."""
+"""CLI entry point: an Agent Client Protocol agent over stdio."""
 
 from __future__ import annotations
 
 import asyncio
-import os
-import sys
-
-import rlm
 
 
 def main():
     import argparse
 
     parser = argparse.ArgumentParser(
         prog="rlm",
-        description="A minimalistic CLI agent for true recursion.",
-    )
-    parser.add_argument(
-        "prompt",
-        nargs="?",
-        help="Task prompt (omit for interactive mode)",
-    )
-    parser.add_argument(
-        "--model", default=None, help="Model name (overrides RLM_MODEL)"
-    )
-    parser.add_argument(
-        "--system-prompt-path",
-        default=None,
-        help="Path to a file whose contents replace the generated system prompt",
-    )
-    parser.add_argument(
-        "--append-to-system-prompt",
-        default=None,
-        help="Extra instructions appended to the generated system prompt",
+        description="A minimalistic recursive agent, served over the Agent Client Protocol.",
     )
     parser.add_argument(
         "--acp",
         action="store_true",
-        help="Serve as an Agent Client Protocol agent over stdio",
+        help="Serve as an Agent Client Protocol agent over stdio (the only mode)",
     )
     args = parser.parse_args()
+    if not args.acp:
+        parser.error("rlm runs only as an ACP agent: use `rlm --acp`")
+    from rlm.acp import serve_acp
 
-    # Apply CLI overrides to env
-    if args.model:
-        os.environ["RLM_MODEL"] = args.model
-    if args.system_prompt_path:
-        os.environ["RLM_SYSTEM_PROMPT_PATH"] = args.system_prompt_path
-    if args.append_to_system_prompt:
-        os.environ["RLM_APPEND_TO_SYSTEM_PROMPT"] = args.append_to_system_prompt
-
-    if args.acp:
-        if args.prompt:
-            parser.error("a prompt cannot be supplied with --acp")
-        from rlm.acp import serve_acp
-
-        asyncio.run(serve_acp())
-    elif args.prompt:
-        print(asyncio.run(rlm.run(args.prompt)).answer)
-    else:
-        _run_interactive()
-
-
-def _run_interactive():
-    print("rlm interactive mode")
-    print('TUI not yet implemented. Use: rlm "your prompt" for headless mode.')
-    sys.exit(0)
+    asyncio.run(serve_acp())
 
 
 if __name__ == "__main__":

src/rlm/client.py

@@ -39,31 +39,8 @@
 _RETRY_DELAYS: tuple[int, ...] = (15, 30, 60, 90, 120)
 
 
-def resolve_provider() -> tuple[str | None, str | None, dict[str, str]]:
-    """Pick the first provider whose key is set: ``(base_url, api_key, headers)``.
-
-    Each provider is a self-contained pair so a key never reaches a base
-    URL it wasn't issued for:
-
-    1. **Explicit** — ``RLM_API_KEY`` (pairs with ``RLM_BASE_URL`` if set,
-       otherwise SDK default = ``api.openai.com``). Set both for a
-       non-OpenAI custom endpoint.
-    2. **PI Inference** — ``PRIME_API_KEY`` at PI's base, with
-       ``PRIME_TEAM_ID`` forwarded as ``X-Prime-Team-ID``.
-    3. **OpenAI** — ``OPENAI_API_KEY`` set: capture ``OPENAI_API_KEY`` and
-       ``OPENAI_BASE_URL`` into the trusted provider configuration. Covers
-       OpenAI direct and verifiers' rollout tunnel both.
-
-    Falls back to PI + ``"EMPTY"`` so the SDK can't silently inherit
-    ``OPENAI_API_KEY`` and ship it to the PI default base.
-    """
-    provider = ProviderConfig.from_env()
-    return provider.base_url, provider.api_key, provider.headers.copy()
-
-
-def make_client(provider: ProviderConfig | None = None) -> AsyncOpenAI:
-    """Create an AsyncOpenAI client from explicit or environment configuration."""
-    provider = provider or ProviderConfig.from_env()
+def make_client(provider: ProviderConfig) -> AsyncOpenAI:
+    """Create an AsyncOpenAI client from an explicit provider configuration."""
     reserved = sorted(
         name
         for name in provider.headers

src/rlm/config.py

@@ -1,62 +1,16 @@
-"""Validated runtime configuration for RLM engines."""
+"""Validated runtime configuration for RLM engines.
 
-from __future__ import annotations
+Configuration enters an rlm process exactly once, through the versioned ACP
+runtime contract (``ai.prime.rlm/runtime-v1``); recursive children inherit it
+in-memory via ``model_copy``. There is no environment-variable resolution.
+"""
 
-import json
-import os
-from typing import Mapping
+from __future__ import annotations
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from typing_extensions import Self
 
 from rlm.semantic import ACP_EXTENSION_HEADER_NAMES
-from rlm.tools.registry import preset_skills
-
-
-PI_INFERENCE_BASE_URL = "https://api.pinference.ai/api/v1"
-KERNEL_ENV_CONFIG_ENV = "RLM_KERNEL_ENV"
-
-
-def _optional_positive_int(value: str | int | None, name: str) -> int | None:
-    if value is None or value == "":
-        return None
-    if isinstance(value, bool):
-        raise ValueError(f"{name} must be an int")
-    try:
-        parsed = int(value)
-    except (TypeError, ValueError) as exc:
-        raise ValueError(f"{name} must be an int (got {value!r})") from exc
-    return parsed if parsed > 0 else None
-
-
-def _positive_int(value: str | int, name: str) -> int:
-    parsed = _optional_positive_int(value, name)
-    if parsed is None:
-        raise ValueError(f"{name} must be positive")
-    return parsed
-
-
-def _summarize_at_tokens(value: str | int | None) -> int | None:
-    """Unset -> the 256k default; "" or "0" -> disabled; else a positive threshold."""
-    if value is None:
-        return 256_000
-    if value in ("", "0", 0):
-        return None
-    parsed = _optional_positive_int(value, "summarize_at_tokens")
-    if parsed is None:
-        raise ValueError(f"summarize_at_tokens must be positive (got {value})")
-    return parsed
-
-
-def _kernel_env(value: str | None) -> tuple[tuple[str, str], ...]:
-    if not value:
-        return ()
-    parsed = json.loads(value)
-    if not isinstance(parsed, dict) or not all(
-        isinstance(key, str) and isinstance(item, str) for key, item in parsed.items()
-    ):
-        raise ValueError(f"{KERNEL_ENV_CONFIG_ENV} must be a JSON object of strings")
-    return tuple(parsed.items())
 
 
 class _ConfigModel(BaseModel):
@@ -84,49 +38,12 @@ def _reserve_transport_headers(cls, headers: dict[str, str]) -> dict[str, str]:
             raise ValueError(f"provider headers contain reserved names: {reserved}")
         return headers
 
-    @classmethod
-    def from_env(cls, environ: Mapping[str, str] | None = None) -> ProviderConfig:
-        env = os.environ if environ is None else environ
-        max_retries = int(env.get("RLM_SDK_MAX_RETRIES", "5"))
-        if api_key := env.get("RLM_API_KEY"):
-            return cls(
-                base_url=env.get("RLM_BASE_URL"),
-                api_key=api_key,
-                max_retries=max_retries,
-            )
-        if api_key := env.get("PRIME_API_KEY"):
-            headers = {}
-            if team_id := env.get("PRIME_TEAM_ID"):
-                headers["X-Prime-Team-ID"] = team_id
-            return cls(
-                base_url=PI_INFERENCE_BASE_URL,
-                api_key=api_key,
-                headers=headers,
-                max_retries=max_retries,
-            )
-        if env.get("OPENAI_API_KEY"):
-            return cls(
-                base_url=env.get("OPENAI_BASE_URL"),
-                api_key=env["OPENAI_API_KEY"],
-                max_retries=max_retries,
-            )
-        return cls(
-            base_url=PI_INFERENCE_BASE_URL,
-            api_key="EMPTY",
-            max_retries=max_retries,
-        )
-
 
 class InvocationContext(_ConfigModel):
     """Trusted identity of one engine within a recursive session tree."""
 
     depth: int = Field(default=0, ge=0)
 
-    @classmethod
-    def from_env(cls, environ: Mapping[str, str] | None = None) -> InvocationContext:
-        env = os.environ if environ is None else environ
-        return cls(depth=int(env.get("RLM_DEPTH", "0")))
-
     def child(self) -> InvocationContext:
         return InvocationContext(depth=self.depth + 1)
 
@@ -162,55 +79,3 @@ class RuntimeConfig(_ConfigModel):
     skills: tuple[str, ...] = ()
     kernel_env: tuple[tuple[str, str], ...] = Field(default=(), repr=False)
     search_api_key: str | None = Field(default=None, repr=False)
-
-    @classmethod
-    def from_env(
-        cls,
-        *,
-        environ: Mapping[str, str] | None = None,
-    ) -> RuntimeConfig:
-        env = os.environ if environ is None else environ
-        raw_skills = env.get("RLM_SKILLS")
-        max_depth = int(env.get("RLM_MAX_DEPTH", "1"))
-        default_concurrency = max(4, max_depth)
-        max_concurrent_subagents = _positive_int(
-            env.get("RLM_MAX_CONCURRENT_SUBAGENTS", str(default_concurrency)),
-            "RLM_MAX_CONCURRENT_SUBAGENTS",
-        )
-        if max_depth > max_concurrent_subagents:
-            raise ValueError(
-                "RLM_MAX_CONCURRENT_SUBAGENTS must be at least RLM_MAX_DEPTH"
-            )
-        return cls(
-            model=env.get("RLM_MODEL", "openai/gpt-5-mini"),
-            provider=ProviderConfig.from_env(env),
-            invocation=InvocationContext.from_env(env),
-            policy=ExecutionPolicy(
-                max_depth=max_depth,
-                exec_timeout=int(env.get("RLM_EXEC_TIMEOUT", "300")),
-                max_tokens=_optional_positive_int(
-                    env.get("RLM_MAX_TOKENS"), "RLM_MAX_TOKENS"
-                ),
-                summarize_at_tokens=_summarize_at_tokens(
-                    env.get("RLM_SUMMARIZE_AT_TOKENS")
-                ),
-                max_compactions=_optional_positive_int(
-                    env.get("RLM_MAX_COMPACTIONS"), "RLM_MAX_COMPACTIONS"
-                ),
-                max_concurrent_subagents=max_concurrent_subagents,
-                max_subagent_calls=_positive_int(
-                    env.get("RLM_MAX_SUBAGENT_CALLS", "64"),
-                    "RLM_MAX_SUBAGENT_CALLS",
-                ),
-                allow_git=env.get("RLM_ALLOW_GIT") == "1",
-            ),
-            system_prompt_path=env.get("RLM_SYSTEM_PROMPT_PATH"),
-            append_to_system_prompt=env.get("RLM_APPEND_TO_SYSTEM_PROMPT"),
-            skills=(
-                tuple(s.strip() for s in raw_skills.split(",") if s.strip())
-                if raw_skills is not None
-                else preset_skills()
-            ),
-            kernel_env=_kernel_env(env.get(KERNEL_ENV_CONFIG_ENV)),
-            search_api_key=env.get("SERPER_API_KEY"),
-        )

src/rlm/engine.py

@@ -22,7 +22,7 @@
 )
 from rlm.config import RuntimeConfig
 from rlm.semantic import SemanticEdgeTracker
-from rlm.mcp import MCPServer, load_mcp_servers, validate_mcp_servers
+from rlm.mcp import MCPServer, validate_mcp_servers
 from rlm.prompt import build_system_prompt
 from rlm.session import Session
 from rlm.skills import enable_builtin_skills
@@ -174,7 +174,13 @@ def __init__(
         parent_session_id: str | None = None,
         spawned_by_request_id: str | None = None,
     ):
-        self.runtime_config = runtime_config or RuntimeConfig.from_env()
+        if runtime_config is None:
+            raise ValueError(
+                "RLMEngine requires an explicit runtime_config: standalone "
+                "environment configuration was removed (rlm is consumed via "
+                "the ACP runtime contract; children inherit in-memory)."
+            )
+        self.runtime_config = runtime_config
         config = self.runtime_config
         self.model = config.model
         self.cwd = cwd or os.getcwd()
@@ -189,9 +195,7 @@ def __init__(
 
         # Task MCP tool servers to expose as IPython skills; kwarg wins, otherwise
         # parse RLM_MCP_CONFIG (a standard mcpServers config).
-        self.mcp_servers = validate_mcp_servers(
-            mcp_servers if mcp_servers is not None else load_mcp_servers()
-        )
+        self.mcp_servers = validate_mcp_servers(mcp_servers or {})
 
         # Built-in skills (rlm.skills) to enable for this run, from RLM_SKILLS (comma-separated).
         self.skills = list(config.skills)