feat: runtime config switching across REPL, ACP, and HTTP

coding-agent/src/coding_agent/acp/__init__.py

@@ -1,5 +1,5 @@
 """Agent Client Protocol stdio adapter for Coding Agent."""
 
-from coding_agent.acp.server import AcpServer, run_stdio
+from coding_agent.acp.server import AcpMode, AcpServer, run_stdio
 
-__all__ = ["AcpServer", "run_stdio"]
+__all__ = ["AcpMode", "AcpServer", "run_stdio"]

coding-agent/src/coding_agent/acp/server.py

@@ -2,6 +2,7 @@
 
 import asyncio
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
+from dataclasses import dataclass
 import json
 from pathlib import Path
 import sys
@@ -26,6 +27,32 @@
 ACP_DEFAULT_MODE_ID = "default"
 
 
+@dataclass(frozen=True)
+class AcpMode:
+    """A named provider/model preset exposed as an ACP session mode."""
+
+    id: str
+    name: str
+    description: str = ""
+    provider: str | None = None
+    model: str | None = None
+    base_url: str | None = None
+
+
+def default_acp_modes() -> list[AcpMode]:
+    """Return the default single-mode configuration.
+
+    Override by passing custom modes to AcpServer or via ACP_TOML config.
+    """
+    return [
+        AcpMode(
+            id=ACP_DEFAULT_MODE_ID,
+            name="Default",
+            description="Standard Coding Agent behavior.",
+        ),
+    ]
+
+
 class AcpSessionManager(Protocol):
     async def create_session(self, **kwargs: Any) -> str: ...
 
@@ -72,6 +99,15 @@ async def update_session_additional_directories(
         additional_directories: list[str],
     ) -> None: ...
 
+    async def replace_session_runtime_config(
+        self,
+        session_id: str,
+        *,
+        model_name: str | None = None,
+        provider_name: str | None = None,
+        base_url: str | None = None,
+    ) -> Any: ...
+
 
 class JsonRpcError(Exception):
     def __init__(self, code: int, message: str) -> None:
@@ -92,6 +128,7 @@ def __init__(
         model_name: str | None = None,
         base_url: str | None = None,
         max_steps: int = 30,
+        modes: list[AcpMode] | None = None,
     ) -> None:
         self._session_manager = session_manager
         self._emit = emit
@@ -101,6 +138,9 @@ def __init__(
         self._model_name = model_name
         self._base_url = base_url
         self._max_steps = max_steps
+        self._modes = (
+            modes if modes is not None else default_acp_modes()
+        )
         self._active_prompt_sessions: set[str] = set()
         self._cancelled_prompt_sessions: set[str] = set()
 
@@ -173,10 +213,10 @@ async def _dispatch(self, method: str, params: JSONObject) -> object:
             return await self._session_new(params)
         if method == "session/load":
             await self._session_load(params)
-            return {"modes": _session_mode_state()}
+            return {"modes": _session_mode_state(self._modes)}
         if method == "session/resume":
             await self._session_resume(params)
-            return {"modes": _session_mode_state()}
+            return {"modes": _session_mode_state(self._modes)}
         if method == "session/list":
             return await self._session_list(params)
         if method == "session/close":
@@ -213,6 +253,7 @@ def _initialize(self, params: JSONObject) -> JSONObject:
                     "additionalDirectories": {},
                 },
             },
+            "modes": _session_mode_state(self._modes),
             "agentInfo": {
                 "name": "coding-agent",
                 "title": "Coding Agent",
@@ -260,7 +301,7 @@ async def _session_new(self, params: JSONObject) -> JSONObject:
             mcp_servers=mcp_servers,
             additional_directories=additional_directories,
         )
-        return {"sessionId": session_id, "modes": _session_mode_state()}
+        return {"sessionId": session_id, "modes": _session_mode_state(self._modes)}
 
     async def _session_list(self, params: JSONObject) -> JSONObject:
         cursor = params.get("cursor")
@@ -399,12 +440,32 @@ async def _session_set_mode(self, params: JSONObject) -> None:
                 -32602, "session/set_mode params.sessionId must be a string"
             )
         mode_id = params.get("modeId")
-        if mode_id != ACP_DEFAULT_MODE_ID:
+        if not isinstance(mode_id, str) or not mode_id:
+            raise JsonRpcError(
+                -32602, "session/set_mode params.modeId must be a string"
+            )
+
+        mode = next((m for m in self._modes if m.id == mode_id), None)
+        if mode is None:
             raise JsonRpcError(
                 -32602,
-                f"session/set_mode params.modeId must be {ACP_DEFAULT_MODE_ID}",
+                f"session/set_mode unknown modeId: {mode_id}",
             )
+
         await self._session_manager.get_session_async(session_id)
+        try:
+            await self._session_manager.replace_session_runtime_config(
+                session_id,
+                model_name=mode.model,
+                provider_name=mode.provider,
+                base_url=mode.base_url,
+            )
+        except RuntimeError as exc:
+            if str(exc) == "turn already in progress":
+                raise JsonRpcError(-32000, "Turn already in progress") from exc
+            raise JsonRpcError(-32603, str(exc)) from exc
+        except ValueError as exc:
+            raise JsonRpcError(-32603, str(exc)) from exc
 
     async def _session_set_config_option(self, params: JSONObject) -> JSONObject:
         session_id = params.get("sessionId")
@@ -770,16 +831,19 @@ def _session_additional_directories(session: object) -> list[str]:
     return list(value)
 
 
-def _session_mode_state() -> JSONObject:
+def _session_mode_state(modes: Iterable[AcpMode] | None = None) -> JSONObject:
+    available = [
+        {
+            "id": m.id,
+            "name": m.name,
+            "description": m.description or f"Provider: {m.provider or 'any'}, Model: {m.model or 'any'}",
+        }
+        for m in (modes or default_acp_modes())
+    ]
+    default_id = available[0]["id"] if available else ACP_DEFAULT_MODE_ID
     return {
-        "currentModeId": ACP_DEFAULT_MODE_ID,
-        "availableModes": [
-            {
-                "id": ACP_DEFAULT_MODE_ID,
-                "name": "Default",
-                "description": "Standard Coding Agent behavior.",
-            }
-        ],
+        "currentModeId": default_id,
+        "availableModes": available,
     }
 
 

coding-agent/src/coding_agent/cli/acp_command.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import time
 import uuid
 
 import click
 
 from agentkit.tracing import configure_tracing
-from coding_agent.acp import AcpServer, run_stdio
+from coding_agent.acp import AcpServer, AcpMode, run_stdio
 from coding_agent.approval import ApprovalPolicy
 from coding_agent.cli.local_runtime import create_local_cli_session_manager
 from coding_agent.local_storage import (
@@ -29,6 +30,30 @@ def _shared_cli_arg(ctx: click.Context, name: str) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _parse_acp_mode(mode_str: str) -> AcpMode:
+    """Parse a mode string in format 'id:name[:provider[:model[:base_url]]]'."""
+    if not mode_str:
+        raise click.BadParameter("--acp-mode value must not be empty")
+    parts = mode_str.split(":", 4)
+    mode_id = parts[0]
+    if not mode_id:
+        raise click.BadParameter(
+            f"--acp-mode '{mode_str}' has empty mode id"
+        )
+    name = parts[1] if len(parts) > 1 else mode_id
+    provider = parts[2] if len(parts) > 2 else None
+    model = parts[3] if len(parts) > 3 else None
+    base_url = parts[4] if len(parts) > 4 else None
+    return AcpMode(
+        id=mode_id,
+        name=name,
+        description=f"Provider: {provider or 'any'}, Model: {model or 'any'}",
+        provider=provider,
+        model=model,
+        base_url=base_url,
+    )
+
+
 @click.command("acp")
 @click.option(
     "--approval",
@@ -38,16 +63,46 @@ def _shared_cli_arg(ctx: click.Context, name: str) -> str | None:
     help="Tool approval policy for ACP sessions.",
 )
 @click.option("--max-steps", default=30, type=int, show_default=True)
-def acp(approval_policy: str, max_steps: int) -> None:
+@click.option(
+    "--acp-mode",
+    "acp_modes",
+    multiple=True,
+    default=None,
+    help=(
+        "Define an ACP mode preset (repeatable). Format: "
+        "id:name[:provider[:model[:base_url]]]. "
+        "When set, these replace the single Default mode."
+    ),
+)
+@click.option(
+    "--acp-modes-json",
+    default=None,
+    help="JSON string defining ACP modes array (alternative to --acp-mode).",
+)
+def acp(
+    approval_policy: str,
+    max_steps: int,
+    acp_modes: tuple[str, ...],
+    acp_modes_json: str | None,
+) -> None:
     """Run Coding Agent as an ACP stdio agent."""
-
-    asyncio.run(_run_acp_stdio(click.get_current_context(), approval_policy, max_steps))
+    asyncio.run(
+        _run_acp_stdio(
+            click.get_current_context(),
+            approval_policy,
+            max_steps,
+            acp_modes,
+            acp_modes_json,
+        )
+    )
 
 
 async def _run_acp_stdio(
     ctx: click.Context,
     approval_policy: str,
     max_steps: int,
+    acp_modes: tuple[str, ...],
+    acp_modes_json: str | None,
 ) -> None:
     configure_tracing(enabled=False)
     data_dir = local_data_dir()
@@ -57,6 +112,19 @@ async def _run_acp_stdio(
         owner_id=f"acp-stdio:{uuid.uuid4().hex}",
         fencing_token=time.time_ns(),
     )
+
+    modes: list[AcpMode] | None = None
+    if acp_modes:
+        modes = [_parse_acp_mode(m) for m in acp_modes]
+    elif acp_modes_json:
+        try:
+            raw = json.loads(acp_modes_json)
+            modes = [AcpMode(**m) for m in raw]
+        except (json.JSONDecodeError, TypeError, KeyError) as exc:
+            raise click.BadParameter(
+                f"Invalid --acp-modes-json: {exc}"
+            ) from exc
+
     try:
         await manager.start_owner_lease_renewal()
         server = AcpServer(
@@ -66,6 +134,7 @@ async def _run_acp_stdio(
             model_name=_shared_cli_arg(ctx, "model"),
             base_url=_shared_cli_arg(ctx, "base_url"),
             max_steps=max_steps,
+            modes=modes,
         )
         await run_stdio(server)
     finally:

coding-agent/src/coding_agent/cli/commands.py

@@ -107,27 +107,40 @@ async def cmd_plan(args: list[str], context: dict[str, Any]) -> None:
         print_pt("No active plan. Use todo_write to create one.", output=_out())
 
 
-@command("model", "Show or change model")
+@command("model", "Show or change model. Syntax: [provider:]model")
 async def cmd_model(args: list[str], context: dict[str, Any]) -> None:
     if args:
-        new_model = args[0]
-        if len(new_model) < 2 or len(new_model) > 100:
+        raw = args[0]
+        if len(raw) < 2 or len(raw) > 100:
             print_html(
-                f"<ansired>Invalid model name: {_h(new_model)}</ansired>",
+                f"<ansired>Invalid model name: {_h(raw)}</ansired>",
                 output=_out(),
             )
             return
+        # Parse optional "provider:model" syntax
+        provider_override: str | None = None
+        model_name = raw
+        if ":" in raw:
+            provider_override, _, model_name = raw.partition(":")
+            if not provider_override or not model_name:
+                print_html(
+                    f"<ansired>Invalid syntax. Use [provider:]model. Got: {_h(raw)}</ansired>",
+                    output=_out(),
+                )
+                return
+            context["provider_override"] = provider_override
+
+        context["model"] = model_name
         on_model_change = cast(
-            Callable[[str], Coroutine[Any, Any, None]] | None,
+            Callable[[str, str | None], Coroutine[Any, Any, None]] | None,
             context.get("on_model_change"),
         )
         if callable(on_model_change):
-            await on_model_change(new_model)
-        else:
-            context["model"] = new_model
+            await on_model_change(model_name, provider_override)
         output = _out()
+        label = f"{provider_override}:{model_name}" if provider_override else model_name
         print_html(
-            f"Model changed to: <ansicyan><b>{_h(new_model)}</b></ansicyan>",
+            f"Model changed to: <ansicyan><b>{_h(label)}</b></ansicyan>",
             output=output,
         )
         print_html(
@@ -136,8 +149,10 @@ async def cmd_model(args: list[str], context: dict[str, Any]) -> None:
         )
     else:
         current = context.get("model", "unknown")
+        provider_current = context.get("provider", None)
+        label = f"{provider_current}:{current}" if provider_current else current
         print_html(
-            f"Current model: <ansicyan><b>{_h(current)}</b></ansicyan>",
+            f"Current model: <ansicyan><b>{_h(label)}</b></ansicyan>",
             output=_out(),
         )
 

coding-agent/src/coding_agent/cli/local_runtime.py

@@ -98,6 +98,8 @@ async def replace_session_runtime_config(
         session_id: str,
         *,
         model_name: str,
+        provider_name: str | None = None,
+        base_url: str | None = None,
     ) -> Any: ...
 
     async def restore_checkpoint(self, session_id: str, checkpoint_id: str) -> None: ...
@@ -252,10 +254,14 @@ async def replace_session_runtime_config(
         session_id: str,
         *,
         model_name: str,
+        provider_name: str | None = None,
+        base_url: str | None = None,
     ) -> Any:
         return await self._delegate.replace_session_runtime_config(
             session_id,
             model_name=model_name,
+            provider_name=provider_name,
+            base_url=base_url,
         )
 
     async def restore_checkpoint(self, session_id: str, checkpoint_id: str) -> None: