完善 ACPAgentRunner 会话管理与权限边界 (#2822)

* feat(acp-agent-runner): add ACP session commands

* fix(acp-agent-runner): 收紧会话管理契约

* fix(acp-agent-runner): 修复 event-first 会话身份与切换能力校验

plugins/Runner/acp-agent-runner/README.md

@@ -11,6 +11,7 @@ It is a thin runtime adapter:
 - ACP `session/update` text chunks are streamed back to LangBot.
 - LangBot multimodal input is mapped to ACP prompt content blocks when the selected ACP runtime advertises the matching prompt capability.
 - LangBot tools, knowledge bases, and history are exposed through the SDK-owned run-scoped MCP bridge.
+- `!new` resets the current ACP binding for any location. Daemon mode additionally provides `!list`, `!sessions`, `!switch`, and `!delete` when the ACP client advertises the required session capabilities. `!stop` uses the Host's generic Runner cancellation contract.
 
 ## Runner ID
 
@@ -129,6 +130,22 @@ Useful options:
 Headless ACP permission requests are answered with `allow_once` by default.
 LangBot does not yet expose an interactive approval UI for these requests.
 
+## Session management commands
+
+`!list`, `!sessions`, `!switch`, and `!delete` target `location=daemon` and require the daemon-side ACP runtime to advertise the corresponding session capability. `!new` only resets the ACPAgentRunner binding and works with every location. These are ACPAgentRunner features, not a generic Runner command contract. Command state uses the reset-stable IM session identity, so Host context resets do not lose the selected ACP session. If `env-json` is required, configure those variables in the daemon process instead; daemon session-control commands do not persist environment values in plugin storage.
+
+| Command | Behavior |
+| --- | --- |
+| `!new` | Clears the current binding; the next message creates a new ACP session while preserving the old one |
+| `!list` / `!sessions` | Lists ACP sessions in the configured daemon workspace |
+| `!switch <number or session ID>` | Selects an existing ACP session |
+| `!delete` | Deletes the current ACP session, then clears the LangBot binding and visible context |
+| `!stop` | Stops the active run without deleting the ACP session |
+
+`!delete` preserves LangBot run-audit records. Immediate `!stop` cancellation requires Host support for control-command cancellation before the per-session concurrency slot.
+
+In group chats, `!new`, `!switch`, and `!delete` require bot administrator privilege. `!stop` may stop a run started by the same member; bot administrators may stop any active run in the group.
+
 ## Steering (follow-up input)
 
 This runner declares `capabilities.steering: true`. When a run is still in

plugins/Runner/acp-agent-runner/components/commands/__init__.py

@@ -0,0 +1 @@
+"""ACP session-management command components."""

plugins/Runner/acp-agent-runner/components/commands/delete.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Command
+metadata:
+  name: delete
+  label:
+    en_US: Delete ACP session
+    zh_Hans: 删除 ACP 会话
+  description:
+    en_US: Delete the current session for an ACP client with session management support
+    zh_Hans: 删除支持会话管理的 ACP 客户端当前会话
+spec:
+execution:
+  python:
+    path: session_commands.py
+    attr: DeleteSession

plugins/Runner/acp-agent-runner/components/commands/list.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Command
+metadata:
+  name: list
+  label:
+    en_US: List ACP sessions
+    zh_Hans: 列出 ACP 会话
+  description:
+    en_US: List sessions for an ACP client with session management support
+    zh_Hans: 列出支持会话管理的 ACP 客户端会话
+spec:
+execution:
+  python:
+    path: session_commands.py
+    attr: ListSessions

plugins/Runner/acp-agent-runner/components/commands/new.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Command
+metadata:
+  name: new
+  label:
+    en_US: New ACP session
+    zh_Hans: 新建 ACP 会话
+  description:
+    en_US: Start a new ACP session without deleting the previous one
+    zh_Hans: 新建 ACP 会话并保留旧会话
+spec:
+execution:
+  python:
+    path: session_commands.py
+    attr: NewSession

plugins/Runner/acp-agent-runner/components/commands/session_commands.py

@@ -0,0 +1,324 @@
+"""Commands for managing ACP sessions owned by the current IM conversation."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+
+from langbot_plugin.api.definition.components.command.command import Command
+from langbot_plugin.api.entities.builtin.command.context import CommandReturn, ExecuteContext
+from pkg.session_management import (
+    conversation_id_from_command,
+    delete_value,
+    normalize_sessions,
+    read_json,
+    resolve_session_target,
+    run_daemon_control,
+    storage_key,
+    write_json,
+)
+
+
+def _language(component: Command) -> str:
+    return str(component.get_plugin_config().get("language") or "zh_Hans")
+
+
+def _text(component: Command, zh: str, en: str) -> str:
+    return zh if _language(component).startswith("zh") else en
+
+
+def _group_admin_required(context: ExecuteContext) -> bool:
+    launcher_type = getattr(getattr(context, "session", None), "launcher_type", None)
+    launcher_type = getattr(launcher_type, "value", launcher_type)
+    return launcher_type == "group" and context.privilege < 2
+
+
+def _group_admin_message(component: Command) -> str:
+    return _text(
+        component,
+        "群聊中的会话变更仅允许机器人管理员执行。",
+        "Only bot administrators can change ACP sessions in group chats.",
+    )
+
+
+async def _clear_session_values(component: Command, conversation_id: str) -> None:
+    for kind in ("selected", "last", "list"):
+        await delete_value(component.plugin, storage_key(kind, conversation_id))
+
+
+async def _session_values(component: Command, conversation_id: str) -> dict[str, object]:
+    return {
+        kind: await read_json(component.plugin, storage_key(kind, conversation_id), None)
+        for kind in ("selected", "last", "list")
+    }
+
+
+async def _restore_session_values(
+    component: Command,
+    conversation_id: str,
+    values: dict[str, object],
+) -> None:
+    for kind, value in values.items():
+        key = storage_key(kind, conversation_id)
+        if value is None:
+            await delete_value(component.plugin, key)
+        else:
+            await write_json(component.plugin, key, value)
+
+
+async def _conversation_state(component: Command, context: ExecuteContext):
+    conversation_id = conversation_id_from_command(context)
+    control = await read_json(component.plugin, storage_key("control", conversation_id), {})
+    selected = await read_json(component.plugin, storage_key("selected", conversation_id), {})
+    last = await read_json(component.plugin, storage_key("last", conversation_id), {})
+    current = selected if isinstance(selected, dict) and selected.get("session_id") else last
+    return conversation_id, control if isinstance(control, dict) else {}, current if isinstance(current, dict) else {}
+
+
+class NewSession(Command):
+    async def initialize(self) -> None:
+        await super().initialize()
+
+        @self.subcommand(name="", help="Create a new ACP session", usage="new")
+        async def _(self: NewSession, context: ExecuteContext) -> AsyncGenerator[CommandReturn, None]:
+            if _group_admin_required(context):
+                yield CommandReturn(text=_group_admin_message(self))
+                return
+            conversation_id = conversation_id_from_command(context)
+            try:
+                previous_values = await _session_values(self, conversation_id)
+                await _clear_session_values(self, conversation_id)
+                await context.create_new_conversation()
+            except Exception as exc:
+                try:
+                    if "previous_values" in locals():
+                        await _restore_session_values(self, conversation_id, previous_values)
+                except Exception as rollback_exc:
+                    exc = RuntimeError(f"{exc}; local state rollback also failed: {rollback_exc}")
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        f"新建会话失败，原会话状态已保留或需要管理员协调：{exc}",
+                        f"Failed to create a new session; the previous state was preserved or needs reconciliation: {exc}",
+                    )
+                )
+                return
+            yield CommandReturn(
+                text=_text(
+                    self,
+                    "✅ 下一条普通消息将创建新的 ACP 会话。旧会话仍然保留，可通过 !list 查看并切换。",
+                    "✅ The next normal message will create a new ACP session. Previous sessions remain available in !list.",
+                )
+            )
+
+
+class ListSessions(Command):
+    async def initialize(self) -> None:
+        await super().initialize()
+
+        @self.subcommand(name="", help="List ACP sessions", usage="list")
+        async def _(self: ListSessions, context: ExecuteContext) -> AsyncGenerator[CommandReturn, None]:
+            conversation_id, control, current = await _conversation_state(self, context)
+            if not control:
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        "当前会话尚未初始化 ACP 配置，请先发送一条普通消息。",
+                        "ACP session configuration is not initialized. Send one normal message first.",
+                    )
+                )
+                return
+            try:
+                result = await run_daemon_control(
+                    self.plugin,
+                    control,
+                    "session.list",
+                    cwd=control.get("session_cwd") or control.get("workspace"),
+                )
+            except Exception as exc:
+                yield CommandReturn(text=_text(self, f"查询会话失败：{exc}", f"Failed to list sessions: {exc}"))
+                return
+            sessions = normalize_sessions(result)
+            await write_json(self.plugin, storage_key("list", conversation_id), sessions)
+            if not sessions:
+                yield CommandReturn(
+                    text=_text(self, "当前没有可管理的 ACP 会话。", "No manageable ACP sessions found.")
+                )
+                return
+            current_id = str(current.get("session_id") or "")
+            lines = [_text(self, "ACP 会话：", "ACP sessions:")]
+            for index, item in enumerate(sessions[:20], start=1):
+                marker = " ★" if item["session_id"] == current_id else ""
+                updated = item["updated_at"].replace("T", " ").replace("Z", "")[:16]
+                lines.append(f"{index}. {updated}  {item['title']}{marker}")
+            lines.append(
+                _text(
+                    self,
+                    "\n使用 !switch <序号> 切换，!new 新建，!delete 删除当前会话。",
+                    "\nUse !switch <number>, !new, or !delete.",
+                )
+            )
+            yield CommandReturn(text="\n".join(lines))
+
+
+class SwitchSession(Command):
+    async def initialize(self) -> None:
+        await super().initialize()
+
+        @self.subcommand(name="*", help="Switch ACP session", usage="switch <number|session_id>")
+        async def _(self: SwitchSession, context: ExecuteContext) -> AsyncGenerator[CommandReturn, None]:
+            if _group_admin_required(context):
+                yield CommandReturn(text=_group_admin_message(self))
+                return
+            reference = context.crt_params[0] if context.crt_params else ""
+            if not reference:
+                yield CommandReturn(
+                    text=_text(self, "用法：!switch <序号或 ses_xxx>", "Usage: !switch <number or ses_xxx>")
+                )
+                return
+            conversation_id, control, _current = await _conversation_state(self, context)
+            if not control:
+                yield CommandReturn(
+                    text=_text(self, "请先发送一条普通消息初始化会话。", "Send one normal message first.")
+                )
+                return
+            capabilities = control.get("session_capabilities")
+            if not isinstance(capabilities, dict) or not any(
+                capabilities.get(method) is True for method in ("session/load", "session/resume")
+            ):
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        "当前 ACP 客户端不支持跨进程恢复 session，无法切换会话。",
+                        "The ACP client does not support cross-process session recovery; switching is unavailable.",
+                    )
+                )
+                return
+            try:
+                result = await run_daemon_control(
+                    self.plugin,
+                    control,
+                    "session.list",
+                    cwd=control.get("session_cwd") or control.get("workspace"),
+                )
+            except Exception as exc:
+                yield CommandReturn(text=_text(self, f"查询会话失败：{exc}", f"Failed to list sessions: {exc}"))
+                return
+            sessions = normalize_sessions(result)
+            await write_json(self.plugin, storage_key("list", conversation_id), sessions)
+            target = resolve_session_target(reference, sessions)
+            if not target:
+                yield CommandReturn(
+                    text=_text(self, "找不到指定会话，请先使用 !list。", "Session not found. Run !list first.")
+                )
+                return
+            selected_key = storage_key("selected", conversation_id)
+            try:
+                previous_selected = await read_json(self.plugin, selected_key, None)
+                await write_json(self.plugin, selected_key, {"session_id": target})
+                await context.create_new_conversation()
+            except Exception as exc:
+                try:
+                    if "previous_selected" in locals():
+                        if previous_selected is None:
+                            await delete_value(self.plugin, selected_key)
+                        else:
+                            await write_json(self.plugin, selected_key, previous_selected)
+                except Exception as rollback_exc:
+                    exc = RuntimeError(f"{exc}; selected-session rollback also failed: {rollback_exc}")
+                yield CommandReturn(text=_text(self, f"切换会话失败：{exc}", f"Failed to switch ACP session: {exc}"))
+                return
+            yield CommandReturn(text=_text(self, f"✅ 已切换到会话：{target}", f"✅ Switched to session: {target}"))
+
+
+class DeleteSession(Command):
+    async def initialize(self) -> None:
+        await super().initialize()
+
+        @self.subcommand(name="", help="Delete current ACP session", usage="delete")
+        async def _(self: DeleteSession, context: ExecuteContext) -> AsyncGenerator[CommandReturn, None]:
+            if _group_admin_required(context):
+                yield CommandReturn(text=_group_admin_message(self))
+                return
+            conversation_id, control, current = await _conversation_state(self, context)
+            session_id = str(current.get("session_id") or "").strip()
+            if not session_id:
+                yield CommandReturn(
+                    text=_text(self, "当前没有可删除的 ACP 会话。", "There is no current ACP session to delete.")
+                )
+                return
+            try:
+                await run_daemon_control(
+                    self.plugin,
+                    control,
+                    "session.delete",
+                    session_id=session_id,
+                )
+            except Exception as exc:
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        f"删除会话失败，当前绑定未清理：{exc}",
+                        f"Delete failed; current binding was preserved: {exc}",
+                    )
+                )
+                return
+            try:
+                await _clear_session_values(self, conversation_id)
+            except Exception as exc:
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        f"远端 ACP 会话已删除，但本地绑定清理失败；请勿继续使用该会话并联系管理员处理：{exc}",
+                        "The remote ACP session was deleted, but local binding cleanup failed. "
+                        f"Do not continue using this session; contact an administrator: {exc}",
+                    )
+                )
+                return
+            try:
+                await context.create_new_conversation()
+            except Exception as exc:
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        f"远端 ACP 会话及本地绑定已删除，但 Host 上下文重置失败；请联系管理员处理：{exc}",
+                        "The remote ACP session and local binding were deleted, but the Host context reset failed. "
+                        f"Contact an administrator: {exc}",
+                    )
+                )
+                return
+            yield CommandReturn(
+                text=_text(
+                    self,
+                    f"✅ 已删除当前 ACP 会话：{session_id}",
+                    f"✅ Deleted current ACP session: {session_id}",
+                )
+            )
+
+
+class StopRun(Command):
+    async def initialize(self) -> None:
+        await super().initialize()
+
+        @self.subcommand(name="", help="Stop the active run", usage="stop")
+        async def _(self: StopRun, context: ExecuteContext) -> AsyncGenerator[CommandReturn, None]:
+            query_vars = await context.get_query_vars()
+            if query_vars.get("_stop_command_denied"):
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        "当前活动任务由其他群成员发起，仅机器人管理员可以停止。",
+                        "The active run belongs to another group member; only a bot administrator can stop it.",
+                    )
+                )
+                return
+            run_id = query_vars.get("_stop_command_run_id")
+            if run_id:
+                yield CommandReturn(
+                    text=_text(
+                        self,
+                        "⏹️ 已停止当前任务，ACP 会话已保留。",
+                        "⏹️ The active task was stopped. The ACP session was preserved.",
+                    )
+                )
+            else:
+                yield CommandReturn(text=_text(self, "当前没有正在执行的任务。", "There is no active task."))

plugins/Runner/acp-agent-runner/components/commands/sessions.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Command
+metadata:
+  name: sessions
+  label:
+    en_US: List ACP sessions
+    zh_Hans: 列出 ACP 会话
+  description:
+    en_US: Alias of the list command
+    zh_Hans: list 命令的别名
+spec:
+execution:
+  python:
+    path: session_commands.py
+    attr: ListSessions

plugins/Runner/acp-agent-runner/components/commands/stop.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Command
+metadata:
+  name: stop
+  label:
+    en_US: Stop active run
+    zh_Hans: 停止当前任务
+  description:
+    en_US: Stop the active run without deleting its ACP session
+    zh_Hans: 停止当前任务但保留 ACP 会话
+spec:
+execution:
+  python:
+    path: session_commands.py
+    attr: StopRun

plugins/Runner/acp-agent-runner/components/commands/switch.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Command
+metadata:
+  name: switch
+  label:
+    en_US: Switch ACP session
+    zh_Hans: 切换 ACP 会话
+  description:
+    en_US: Switch to an existing session for an ACP client with session management support
+    zh_Hans: 切换到支持会话管理的 ACP 客户端已有会话
+spec:
+execution:
+  python:
+    path: session_commands.py
+    attr: SwitchSession

plugins/Runner/acp-agent-runner/components/runner/default.py

@@ -29,6 +29,7 @@ from pkg.daemon_relay import (
 from pkg.prompt import acp_prompt_blocks, has_acp_prompt_input, prompt_capabilities
 from pkg.runtime_support import local_workspace
 from pkg.session import CLAUDE_ENV_VARS_TO_UNSET, build_session_params
+from pkg.session_management import read_json, runner_session_identity, storage_key, write_json
 from pkg.steering import run_with_steering
 
 logger = logging.getLogger(__name__)
@@ -506,6 +507,75 @@ class DefaultRunner(Runner):
     def _stored_session_id(self, ctx: RunnerContext) -> str:
         return str(ctx.state.conversation.get(ACP_SESSION_STATE_KEY) or "").strip()
 
+    def _conversation_id(self, ctx: RunnerContext) -> str:
+        return runner_session_identity(ctx)
+
+    async def _managed_session_id(self, ctx: RunnerContext, config: dict[str, typing.Any]) -> str:
+        """Persist command control data and return the command-selected session."""
+
+        conversation_id = self._conversation_id(ctx)
+        if not conversation_id:
+            return ""
+        try:
+            previous_control = await read_json(self.plugin, storage_key("control", conversation_id), {})
+            control = {
+                "provider": config["provider"],
+                "location": config["location"],
+                "acp_command": config["acp_command"],
+                "workspace": config["workspace"],
+                "session_cwd": config["session_cwd"],
+                "daemon_id": config["daemon_id"],
+                "timeout": min(float(config["timeout"]), 120.0),
+                "startup_timeout": config["startup_timeout"],
+                "initialize_timeout": config["initialize_timeout"],
+                "permission_decision": config["permission_decision"],
+                "requires_env_json": bool(config["env"]),
+            }
+            if isinstance(previous_control, dict) and isinstance(previous_control.get("session_capabilities"), dict):
+                control["session_capabilities"] = previous_control["session_capabilities"]
+            await write_json(self.plugin, storage_key("control", conversation_id), control)
+            selected = await read_json(self.plugin, storage_key("selected", conversation_id), {})
+        except Exception as exc:
+            logger.warning("Failed to load ACP managed-session state: %s", exc)
+            return ""
+        if not isinstance(selected, dict):
+            return ""
+        return str(selected.get("session_id") or "").strip()
+
+    async def _record_managed_session(self, ctx: RunnerContext, session_id: str) -> None:
+        conversation_id = self._conversation_id(ctx)
+        if not conversation_id or not session_id:
+            return
+        try:
+            await write_json(
+                self.plugin,
+                storage_key("last", conversation_id),
+                {"session_id": session_id},
+            )
+        except Exception as exc:
+            logger.warning("Failed to persist ACP managed-session state: %s", exc)
+
+    async def _record_session_capabilities(
+        self,
+        ctx: RunnerContext,
+        capabilities: dict[str, typing.Any],
+    ) -> None:
+        conversation_id = self._conversation_id(ctx)
+        if not conversation_id:
+            return
+        try:
+            key = storage_key("control", conversation_id)
+            control = await read_json(self.plugin, key, {})
+            if not isinstance(control, dict):
+                return
+            control["session_capabilities"] = {
+                method: _runtime_has_method(capabilities, method)
+                for method in ("session/list", "session/delete", "session/load", "session/resume")
+            }
+            await write_json(self.plugin, key, control)
+        except Exception as exc:
+            logger.warning("Failed to persist ACP session capabilities: %s", exc)
+
     def _mcp_server_to_acp(self, server: AgentMCPServerConfig) -> dict[str, typing.Any]:
         if server.transport == "http":
             return {
@@ -812,6 +882,16 @@ class DefaultRunner(Runner):
             )
         ) as event_stream:
             async for event in event_stream:
+                if event.get("type") == "control.capabilities":
+                    data = event.get("data") if isinstance(event.get("data"), dict) else {}
+                    capabilities = data.get("agent_capabilities")
+                    if isinstance(capabilities, dict):
+                        await self._record_session_capabilities(ctx, capabilities)
+                    continue
+                if event.get("type") == "state.updated":
+                    data = event.get("data") if isinstance(event.get("data"), dict) else {}
+                    if data.get("key") == ACP_SESSION_STATE_KEY:
+                        await self._record_managed_session(ctx, str(data.get("value") or ""))
                 event.setdefault("run_id", ctx.run_id)
                 yield RunnerResult.model_validate(event)
 
@@ -840,7 +920,8 @@ class DefaultRunner(Runner):
                 return self._run_acp_turn(ctx, config, turn_prompt, resume_session_id, ctx.input)
             return self._run_acp_turn(ctx, config, turn_prompt, resume_session_id, AgentInput(text=turn_prompt))
 
-        initial_resume = self._stored_session_id(ctx) if config["reuse_session"] else ""
+        managed_session_id = await self._managed_session_id(ctx, config)
+        initial_resume = managed_session_id or (self._stored_session_id(ctx) if config["reuse_session"] else "")
         async for result in run_with_steering(
             ctx,
             lambda: self.get_run_api(ctx),
@@ -889,7 +970,7 @@ class DefaultRunner(Runner):
             )
             async with client:
                 initialize_result = await client.initialize(timeout=config["initialize_timeout"])
-                session_id, created = await self._create_or_resume_session(
+                session_id, _created = await self._create_or_resume_session(
                     client,
                     initialize_result,
                     ctx,
@@ -898,13 +979,13 @@ class DefaultRunner(Runner):
                     stored_session_id,
                 )
 
-                if created or stored_session_id != session_id:
-                    yield RunnerResult.state_updated(
-                        ctx.run_id,
-                        ACP_SESSION_STATE_KEY,
-                        session_id,
-                        scope="conversation",
-                    )
+                yield RunnerResult.state_updated(
+                    ctx.run_id,
+                    ACP_SESSION_STATE_KEY,
+                    session_id,
+                    scope="conversation",
+                )
+                await self._record_managed_session(ctx, session_id)
 
                 prompt_blocks = acp_prompt_blocks(prompt_text, agent_input, prompt_capabilities(initialize_result))
                 async for result in self._stream_prompt_results(

plugins/Runner/acp-agent-runner/daemon.py

@@ -133,10 +133,56 @@ def _result_event(
 class RunnerDaemon(AgentRuntimeDaemonClient):
     """Outbound WebSocket daemon that runs ACP jobs locally."""
 
+    async def _run_control_operation(
+        self,
+        job_id: str,
+        operation: str,
+        payload: dict[str, typing.Any],
+    ) -> None:
+        config = payload.get("config") if isinstance(payload.get("config"), dict) else {}
+        arguments = payload.get("arguments") if isinstance(payload.get("arguments"), dict) else {}
+        command_args = _parse_args(config.get("acp_command"))
+        if not command_args:
+            raise AcpError("acp_command is required", code="acp.daemon_config_invalid")
+        client = AcpStdioClient(
+            command=command_args[0],
+            args=command_args[1:],
+            cwd=str(config.get("workspace") or os.getcwd()),
+            env={str(k): str(v) for k, v in dict(config.get("env") or {}).items()},
+            unset_env=(CLAUDE_ENV_VARS_TO_UNSET if str(config.get("provider") or "custom") == "claude-code" else ()),
+            permission_decision=str(config.get("permission_decision") or "allow_once"),
+            startup_timeout=float(config.get("startup_timeout") or 30.0),
+        )
+        async with client:
+            initialize_result = await client.initialize(timeout=float(config.get("initialize_timeout") or 30.0))
+            capabilities = initialize_result.get("agentCapabilities")
+            if not isinstance(capabilities, dict):
+                capabilities = {}
+            timeout = float(config.get("timeout") or 60.0)
+            if operation == "session.list":
+                if not _runtime_has_method(capabilities, "session/list"):
+                    raise AcpError("ACP agent does not support session/list", code="acp.session_list_unsupported")
+                cwd = str(arguments.get("cwd") or config.get("session_cwd") or config.get("workspace") or "")
+                result = await client.request("session/list", {"cwd": cwd} if cwd else {}, timeout=timeout)
+            elif operation == "session.delete":
+                if not _runtime_has_method(capabilities, "session/delete"):
+                    raise AcpError("ACP agent does not support session/delete", code="acp.session_delete_unsupported")
+                session_id = str(arguments.get("session_id") or "").strip()
+                if not session_id:
+                    raise AcpError("session_id is required", code="acp.session_id_required")
+                result = await client.request("session/delete", {"sessionId": session_id}, timeout=timeout)
+            else:
+                raise AcpError(f"unsupported control operation: {operation}", code="acp.control_unsupported")
+            await self.emit_event(job_id, _result_event("control.result", {"operation": operation, "result": result}))
+
     async def run_job(self, job_id: str, payload: dict[str, typing.Any]) -> None:
         proxy: LocalMCPProxy | None = None
         client: AcpStdioClient | None = None
         try:
+            operation = str(payload.get("operation") or "").strip()
+            if operation:
+                await self._run_control_operation(job_id, operation, payload)
+                return
             config = payload.get("config") if isinstance(payload.get("config"), dict) else {}
             prompt_text = str(payload.get("prompt_text") or "")
             input_data = payload.get("input") if isinstance(payload.get("input"), dict) else {}
@@ -167,22 +213,27 @@ class RunnerDaemon(AgentRuntimeDaemonClient):
 
             async with client:
                 initialize_result = await client.initialize(timeout=float(config.get("initialize_timeout") or 30.0))
-                session_id, created = await self._create_or_resume_session(
+                capabilities = initialize_result.get("agentCapabilities")
+                if not isinstance(capabilities, dict):
+                    capabilities = {}
+                await self.emit_event(
+                    job_id,
+                    _result_event("control.capabilities", {"agent_capabilities": capabilities}),
+                )
+                session_id, _created = await self._create_or_resume_session(
                     client, initialize_result, config, mcp_servers
                 )
-                stored_session_id = str(config.get("stored_session_id") or "")
-                if created or stored_session_id != session_id:
-                    await self.emit_event(
-                        job_id,
-                        _result_event(
-                            "state.updated",
-                            {
-                                "key": "external.acp_session_id",
-                                "value": session_id,
-                                "scope": "conversation",
-                            },
-                        ),
-                    )
+                await self.emit_event(
+                    job_id,
+                    _result_event(
+                        "state.updated",
+                        {
+                            "key": "external.acp_session_id",
+                            "value": session_id,
+                            "scope": "conversation",
+                        },
+                    ),
+                )
                 prompt_blocks = acp_prompt_blocks(prompt_text, input_data, prompt_capabilities(initialize_result))
                 await self._stream_prompt_results(client, job_id, session_id, prompt_blocks, config)
         except asyncio.CancelledError:

plugins/Runner/acp-agent-runner/manifest.yaml

@@ -3,7 +3,7 @@ kind: Plugin
 metadata:
   author: langbot-team
   name: ACPAgentRunner
-  version: 0.1.16
+  version: 0.1.18
   repository: https://github.com/langbot-app/LangBot/tree/master/plugins/Runner/acp-agent-runner
   label:
     en_US: ACP Agent Runner
@@ -15,8 +15,8 @@ metadata:
     es_ES: ACP Agent Runner
     ru_RU: ACP Agent Runner
   description:
-    en_US: Run any Agent Client Protocol compatible coding agent as a LangBot Runner.
-    zh_Hans: 将任意兼容 Agent Client Protocol 的编码智能体作为 LangBot 运行器运行。
+    en_US: Run Agent Client Protocol compatible coding agents as a LangBot Runner; session commands require the advertised ACP session capabilities.
+    zh_Hans: 将兼容 Agent Client Protocol 的编码智能体作为 LangBot 运行器运行；会话命令需要客户端声明对应的 ACP session capability。
     zh_Hant: 將任何相容 Agent Client Protocol 的程式設計代理作為 LangBot 運行器執行。
     ja_JP: Agent Client Protocol 対応のコーディングエージェントを LangBot Runner として実行します。
     th_TH: เรียกใช้เอเจนต์เขียนโค้ดที่รองรับ Agent Client Protocol เป็น LangBot Runner
@@ -26,6 +26,21 @@ metadata:
   icon: assets/icon.svg
 spec:
   config:
+    - name: language
+      label:
+        en_US: Session command language
+        zh_Hans: 会话命令语言
+      type: select
+      options:
+        - name: en_US
+          label:
+            en_US: English
+            zh_Hans: 英文
+        - name: zh_Hans
+          label:
+            en_US: Simplified Chinese
+            zh_Hans: 简体中文
+      default: en_US
     - name: daemon-enabled
       label:
         en_US: Enable Runner Daemon Hub
@@ -64,6 +79,9 @@ spec:
       required: false
       default: ''
   components:
+    Command:
+      fromDirs:
+        - path: components/commands/
     Runner:
       fromDirs:
         - path: components/runner/

plugins/Runner/acp-agent-runner/pkg/session_management.py

@@ -0,0 +1,216 @@
+"""Shared session-management helpers for ACP commands and runners."""
+
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import json
+import typing
+
+from pkg.daemon_relay import (
+    agent_runtime_daemon_config_from_plugin_config,
+    get_agent_runtime_daemon_hub,
+)
+
+STORAGE_PREFIX = "acp-session:v1"
+
+
+def _storage_key_missing(exc: Exception) -> bool:
+    code = str(getattr(exc, "code", "") or getattr(exc, "error_code", "")).lower()
+    message = str(exc)
+    explicit_action_error = (
+        type(exc).__name__ == "ActionCallError"
+        and message.startswith("Storage with key ")
+        and message.endswith(" not found")
+    )
+    return (
+        isinstance(exc, (FileNotFoundError, KeyError))
+        or explicit_action_error
+        or code
+        in {
+            "not_found",
+            "storage_not_found",
+            "plugin_storage_not_found",
+        }
+    )
+
+
+def stable_session_identity(session: typing.Any) -> str:
+    """Return the IM session identity that survives Host conversation resets."""
+
+    launcher_type = getattr(session, "launcher_type", None)
+    launcher_type = getattr(launcher_type, "value", launcher_type)
+    launcher_id = getattr(session, "launcher_id", None)
+    if not launcher_type or launcher_id in (None, ""):
+        raise ValueError("session has no stable conversation identity")
+    return f"{launcher_type}_{launcher_id}"
+
+
+def runner_session_identity(context: typing.Any) -> str:
+    """Resolve the same reset-stable identity from a Runner context."""
+
+    conversation = getattr(context, "conversation", None)
+    session_id = str(getattr(conversation, "session_id", "") or "").strip()
+    if session_id:
+        return session_id
+    launcher_type = getattr(conversation, "launcher_type", None)
+    launcher_id = getattr(conversation, "launcher_id", None)
+    if launcher_type in {"group", "person"} and launcher_id not in (None, ""):
+        return f"{launcher_type}_{launcher_id}"
+    return str(getattr(conversation, "conversation_id", "") or "").strip()
+
+
+def conversation_id_from_command(context: typing.Any) -> str:
+    """Build the Host conversation id used by IM-backed Runner state."""
+
+    return stable_session_identity(getattr(context, "session", None))
+
+
+def storage_key(kind: str, conversation_id: str) -> str:
+    digest = hashlib.sha256(conversation_id.encode("utf-8")).hexdigest()
+    return f"{STORAGE_PREFIX}:{kind}:{digest}"
+
+
+async def read_json(plugin: typing.Any, key: str, default: typing.Any = None) -> typing.Any:
+    try:
+        raw = await plugin.get_plugin_storage(key)
+    except Exception as exc:
+        if _storage_key_missing(exc):
+            return default
+        raise
+    if not raw:
+        return default
+    try:
+        return json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError, TypeError):
+        return default
+
+
+async def write_json(plugin: typing.Any, key: str, value: typing.Any) -> None:
+    await plugin.set_plugin_storage(
+        key,
+        json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
+    )
+
+
+async def delete_value(plugin: typing.Any, key: str) -> None:
+    try:
+        current = await plugin.get_plugin_storage(key)
+    except Exception as exc:
+        if _storage_key_missing(exc):
+            return
+        raise
+    if not current:
+        return
+    try:
+        await plugin.delete_plugin_storage(key)
+    except Exception as exc:
+        if _storage_key_missing(exc):
+            return
+        raise
+
+
+def normalize_sessions(result: typing.Any) -> list[dict[str, str]]:
+    """Return a bounded, newest-first list from an ACP session/list result."""
+
+    if isinstance(result, dict):
+        raw_sessions = result.get("sessions")
+    else:
+        raw_sessions = None
+    if not isinstance(raw_sessions, list):
+        return []
+
+    sessions: list[dict[str, str]] = []
+    for item in raw_sessions:
+        if not isinstance(item, dict):
+            continue
+        session_id = str(item.get("sessionId") or item.get("session_id") or item.get("id") or "").strip()
+        if not session_id:
+            continue
+        sessions.append(
+            {
+                "session_id": session_id,
+                "title": str(item.get("title") or "未命名会话").strip() or "未命名会话",
+                "cwd": str(item.get("cwd") or item.get("directory") or "").strip(),
+                "updated_at": str(item.get("updatedAt") or item.get("updated_at") or "").strip(),
+            }
+        )
+    sessions.sort(key=lambda item: item["updated_at"], reverse=True)
+    return sessions[:100]
+
+
+def resolve_session_target(reference: str, sessions: list[dict[str, str]]) -> str | None:
+    reference = str(reference or "").strip()
+    if not reference:
+        return None
+    if reference.isdigit():
+        index = int(reference) - 1
+        if 0 <= index < len(sessions):
+            return sessions[index]["session_id"]
+        return None
+    for item in sessions:
+        if item["session_id"] == reference:
+            return reference
+    return None
+
+
+async def run_daemon_control(
+    plugin: typing.Any,
+    control: dict[str, typing.Any],
+    operation: str,
+    **arguments: typing.Any,
+) -> typing.Any:
+    """Run one non-prompt ACP control operation through the configured daemon."""
+
+    if str(control.get("location") or "") != "daemon":
+        raise RuntimeError("ACP session management commands require location=daemon")
+    if control.get("requires_env_json"):
+        raise RuntimeError(
+            "ACP session management cannot reuse env-json securely; configure required variables in the daemon process"
+        )
+    capabilities = control.get("session_capabilities")
+    capability_method = operation.replace(".", "/", 1)
+    if isinstance(capabilities, dict) and capabilities.get(capability_method) is False:
+        raise RuntimeError(f"ACP agent does not support {operation}")
+
+    daemon_config = agent_runtime_daemon_config_from_plugin_config(
+        plugin.get_config(),
+        env_prefix="LANGBOT_ACP_DAEMON",
+    )
+    hub = get_agent_runtime_daemon_hub("acp", error_code_prefix="acp")
+    await hub.start(
+        host=daemon_config["host"],
+        port=daemon_config["port"],
+        token=daemon_config["token"],
+    )
+    daemon_id = str(control.get("daemon_id") or "").strip()
+    if not daemon_id:
+        raise RuntimeError("ACP daemon id is unavailable; send one normal message first")
+    timeout = float(control.get("timeout") or 60.0)
+    await hub.wait_for_daemon(daemon_id, min(timeout, 60.0))
+    payload = {
+        "operation": operation,
+        "arguments": arguments,
+        "config": dict(control),
+    }
+    result: typing.Any = None
+    received_result = False
+    async with contextlib.aclosing(
+        hub.run_job(
+            daemon_id=daemon_id,
+            payload=payload,
+            tools=None,
+            timeout=timeout,
+        )
+    ) as events:
+        async for event in events:
+            if event.get("type") == "control.result":
+                data = event.get("data") if isinstance(event.get("data"), dict) else {}
+                result = data.get("result")
+                received_result = True
+            elif event.get("type") == "run.failed":
+                data = event.get("data") if isinstance(event.get("data"), dict) else {}
+                raise RuntimeError(str(data.get("error") or data.get("code") or "ACP control operation failed"))
+    if not received_result:
+        raise RuntimeError(f"ACP daemon returned no result for {operation}")
+    return result

plugins/Runner/acp-agent-runner/readme/README_en_US.md

@@ -11,6 +11,7 @@ It is a thin runtime adapter:
 - ACP `session/update` text chunks are streamed back to LangBot.
 - LangBot multimodal input is mapped to ACP prompt content blocks when the selected ACP runtime advertises the matching prompt capability.
 - LangBot tools, knowledge bases, and history are exposed through the SDK-owned run-scoped MCP bridge.
+- `!new` resets the current ACP binding for any location. Daemon mode additionally provides `!list`, `!sessions`, `!switch`, and `!delete` when the ACP client advertises the required session capabilities. `!stop` uses the Host's generic Runner cancellation contract.
 
 ## Runner ID
 
@@ -129,6 +130,24 @@ Useful options:
 Headless ACP permission requests are answered with `allow_once` by default.
 LangBot does not yet expose an interactive approval UI for these requests.
 
+## Session management commands
+
+`!list`, `!sessions`, `!switch`, and `!delete` target `location=daemon` and require the daemon-side ACP runtime to advertise the corresponding session capability. `!new` only resets the ACPAgentRunner binding and works with every location. These are ACPAgentRunner features, not a generic Runner command contract. Command state uses the reset-stable IM session identity, so private and group conversations remain isolated and Host context resets do not lose the selected session. If `env-json` is required, configure those variables in the daemon process instead; daemon session-control commands do not persist environment values in plugin storage.
+
+| Command | Behavior |
+| --- | --- |
+| `!new` | Clears the current binding. The next normal message creates a new ACP session while the old session remains available |
+| `!list` / `!sessions` | Lists ACP sessions in the configured daemon workspace |
+| `!switch <number or session ID>` | Selects an existing session for the next `session/load` or `session/resume` |
+| `!delete` | Calls ACP `session/delete`; only after success does it clear the LangBot binding and visible context |
+| `!stop` | Stops the active run without deleting its ACP session |
+
+`!delete` does not physically remove LangBot run-audit records. It deletes the ACP session and starts a new LangBot context generation, so old transcript entries are no longer included in future context. If ACP deletion fails, the current binding is preserved to avoid divergent LangBot and ACP state.
+
+Immediate `!stop` cancellation requires a LangBot Host that recognizes the control command before the per-session concurrency slot. Older Hosts queue it as a normal command and cannot stop an already running task immediately.
+
+In group chats, `!new`, `!switch`, and `!delete` require bot administrator privilege. `!stop` may stop a run started by the same member; bot administrators may stop any active run in the group.
+
 ## Steering (follow-up input)
 
 This runner declares `capabilities.steering: true`. When a run is still in

plugins/Runner/acp-agent-runner/readme/README_zh_Hans.md

@@ -15,6 +15,7 @@ ACP Runner 用于把兼容 Agent Client Protocol（ACP）的编码 Agent 接入
 - 根据 ACP runtime 公告的能力转发图片或嵌入文本资源。
 - 通过 SDK MCP bridge 提供授权的 LangBot 工具、知识库和历史。
 - 支持运行中的 steering 跟进消息。
+- `!new` 可在任意运行位置重置当前 ACP 绑定；daemon 模式下仅在客户端声明对应 session capability 时额外提供 `!list`、`!sessions`、`!switch` 和 `!delete`；`!stop` 使用 Host 的通用 Runner 取消契约。
 
 ## 支持的预设
 
@@ -81,6 +82,24 @@ ACP Runner 用于把兼容 Agent Client Protocol（ACP）的编码 Agent 接入
 
 启用 `reuse-session` 后，runner 会保存 ACP session ID，并在后续 LangBot 会话运行中优先调用 ACP session load/resume 能力。若 provider 声称支持恢复但实际恢复失败，runner 会创建新 session，并记录可诊断错误。不同 ACP provider 的恢复质量取决于其自身实现。
 
+## 会话管理命令
+
+`!list`、`!sessions`、`!switch` 和 `!delete` 用于 `location=daemon`，需要 daemon 端 ACP runtime 支持相应的 session capability；`!new` 只重置 ACPAgentRunner 绑定，可用于任意运行位置。它们属于 ACPAgentRunner 专属能力，不是通用 Runner Command 契约。命令状态使用可跨 Host context reset 保持稳定的 IM session identity，群聊和私聊不会共享当前会话指针。若 ACP 进程依赖 `env-json`，请改为在 daemon 进程环境中配置；daemon 会话控制命令不会把环境变量值持久化到插件存储。
+
+| 命令 | 说明 |
+| --- | --- |
+| `!new` | 清除当前绑定；下一条普通消息创建新 ACP 会话。旧会话保留，可稍后切回 |
+| `!list` / `!sessions` | 列出当前 daemon 工作目录中的 ACP 会话 |
+| `!switch <序号或 session ID>` | 切换到已有 ACP 会话，下一条消息通过 `session/load` 或 `session/resume` 继续 |
+| `!delete` | 调用 ACP `session/delete` 删除当前会话；成功后再清理 LangBot 绑定和可见上下文 |
+| `!stop` | 停止当前正在运行的任务，但保留 ACP 会话 |
+
+`!delete` 不会物理删除 LangBot 的运行审计记录；它删除 ACP 会话并创建新的 LangBot context generation，因此旧 transcript 不再进入后续上下文。若 ACP 删除失败，当前绑定会保留，避免 LangBot 与 ACP 状态不一致。
+
+`!stop` 的即时取消需要 LangBot Host 在 session 并发槽之前识别控制命令。较旧的 Host 会把它作为普通命令排队，不能立即停止运行中的任务。
+
+群聊中，`!new`、`!switch` 和 `!delete` 仅允许机器人管理员执行；普通成员只能停止自己发起的活动任务，机器人管理员可以停止群内任意活动任务。
+
 ## 多模态输入
 
 runner 会根据 ACP 初始化响应中声明的 prompt capability 决定如何发送输入：

plugins/Runner/acp-agent-runner/tests/test_session_management.py

@@ -0,0 +1,417 @@
+from __future__ import annotations
+
+import types
+from unittest.mock import AsyncMock
+
+import pytest
+from components.commands.session_commands import NewSession, StopRun, SwitchSession
+from components.runner.default import DefaultRunner
+from daemon import RunnerDaemon
+from langbot_plugin.api.entities.builtin.runner.event import ConversationContext
+from pkg.session_management import (
+    conversation_id_from_command,
+    delete_value,
+    normalize_sessions,
+    resolve_session_target,
+    run_daemon_control,
+    runner_session_identity,
+    storage_key,
+)
+
+
+def test_normalize_sessions_sorts_and_filters_invalid_items():
+    result = {
+        "sessions": [
+            {
+                "sessionId": "ses_old",
+                "title": "Old",
+                "cwd": "/workspace",
+                "updatedAt": "2026-01-01T00:00:00Z",
+            },
+            {"title": "Missing id"},
+            {
+                "sessionId": "ses_new",
+                "title": "New",
+                "cwd": "/workspace",
+                "updatedAt": "2026-01-02T00:00:00Z",
+            },
+        ]
+    }
+
+    sessions = normalize_sessions(result)
+
+    assert [item["session_id"] for item in sessions] == ["ses_new", "ses_old"]
+
+
+def test_resolve_session_target_accepts_index_or_exact_id():
+    sessions = [
+        {"session_id": "ses_one", "title": "One", "cwd": "", "updated_at": ""},
+        {"session_id": "ses_two", "title": "Two", "cwd": "", "updated_at": ""},
+    ]
+
+    assert resolve_session_target("2", sessions) == "ses_two"
+    assert resolve_session_target("ses_one", sessions) == "ses_one"
+    assert resolve_session_target("3", sessions) is None
+    assert resolve_session_target("ses_missing", sessions) is None
+
+
+def test_storage_key_does_not_expose_conversation_id():
+    key = storage_key("current", "person_sensitive-platform-id")
+
+    assert key.startswith("acp-session:v1:current:")
+    assert "sensitive-platform-id" not in key
+
+
+def test_runner_and_command_share_reset_stable_session_identity():
+    session = types.SimpleNamespace(
+        launcher_type=types.SimpleNamespace(value="person"),
+        launcher_id="user-1",
+    )
+    command_context = types.SimpleNamespace(session=session)
+    runner_context = types.SimpleNamespace(
+        conversation=types.SimpleNamespace(
+            conversation_id="conversation-uuid-123",
+            session_id=None,
+            launcher_type="person",
+            launcher_id="user-1",
+        )
+    )
+
+    assert conversation_id_from_command(command_context) == "person_user-1"
+    assert runner_session_identity(runner_context) == "person_user-1"
+    assert DefaultRunner()._conversation_id(runner_context) == "person_user-1"
+
+
+def test_event_first_context_preserves_reset_stable_session_identity():
+    context = types.SimpleNamespace(
+        conversation=ConversationContext(
+            conversation_id="conversation-uuid-123",
+            session_id=None,
+            launcher_type="person",
+            launcher_id="user-1",
+        )
+    )
+
+    assert DefaultRunner()._conversation_id(context) == "person_user-1"
+
+
+@pytest.mark.asyncio
+async def test_delete_value_only_ignores_explicit_missing_key_errors():
+    missing_plugin = types.SimpleNamespace(get_plugin_storage=AsyncMock(side_effect=FileNotFoundError()))
+    await delete_value(missing_plugin, "missing")
+
+    failing_plugin = types.SimpleNamespace(get_plugin_storage=AsyncMock(side_effect=RuntimeError("storage offline")))
+    with pytest.raises(RuntimeError, match="storage offline"):
+        await delete_value(failing_plugin, "current")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("control", "error"),
+    [
+        ({"location": "local"}, "location=daemon"),
+        ({"location": "remote-ssh"}, "location=daemon"),
+        ({"location": "daemon", "requires_env_json": True}, "env-json securely"),
+        (
+            {"location": "daemon", "session_capabilities": {"session/list": False}},
+            "does not support session.list",
+        ),
+    ],
+)
+async def test_daemon_control_rejects_unsupported_configuration_without_waiting(control, error):
+    plugin = types.SimpleNamespace(get_config=lambda: {})
+
+    with pytest.raises(RuntimeError, match=error):
+        await run_daemon_control(plugin, control, "session.list")
+
+
+class FakeCommandPlugin:
+    def __init__(self):
+        self.values = {}
+        self.deleted = []
+
+    def get_config(self):
+        return {"language": "en_US"}
+
+    async def get_plugin_storage(self, key):
+        return self.values.get(key, b"")
+
+    async def delete_plugin_storage(self, key):
+        self.deleted.append(key)
+
+    async def set_plugin_storage(self, key, value):
+        self.values[key] = value
+
+
+async def _command_results(command, context):
+    await command.initialize()
+    subcommand = command.registered_subcommands[""].subcommand
+    return [item async for item in subcommand(command, context)]
+
+
+@pytest.mark.asyncio
+async def test_group_session_changes_require_admin_privilege():
+    plugin = FakeCommandPlugin()
+    command = NewSession()
+    command.plugin = plugin
+    context = types.SimpleNamespace(
+        session=types.SimpleNamespace(launcher_type=types.SimpleNamespace(value="group"), launcher_id="group-1"),
+        privilege=1,
+        create_new_conversation=AsyncMock(),
+    )
+
+    results = await _command_results(command, context)
+
+    assert "Only bot administrators" in results[0].text
+    assert plugin.deleted == []
+    context.create_new_conversation.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_group_admin_can_create_new_session():
+    plugin = FakeCommandPlugin()
+    for kind in ("selected", "last", "list"):
+        plugin.values[storage_key(kind, "group_group-1")] = b"{}"
+    command = NewSession()
+    command.plugin = plugin
+    context = types.SimpleNamespace(
+        session=types.SimpleNamespace(launcher_type=types.SimpleNamespace(value="group"), launcher_id="group-1"),
+        privilege=2,
+        create_new_conversation=AsyncMock(),
+    )
+
+    results = await _command_results(command, context)
+
+    assert "next normal message" in results[0].text
+    assert len(plugin.deleted) == 3
+    context.create_new_conversation.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_storage_failure_blocks_new_session_reset():
+    plugin = FakeCommandPlugin()
+    plugin.get_plugin_storage = AsyncMock(side_effect=RuntimeError("storage offline"))
+    command = NewSession()
+    command.plugin = plugin
+    context = types.SimpleNamespace(
+        session=types.SimpleNamespace(launcher_type=types.SimpleNamespace(value="person"), launcher_id="user-1"),
+        privilege=1,
+        create_new_conversation=AsyncMock(),
+    )
+
+    results = await _command_results(command, context)
+
+    assert "storage offline" in results[0].text
+    context.create_new_conversation.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_switch_reset_failure_restores_previous_selection(monkeypatch):
+    plugin = FakeCommandPlugin()
+    conversation_id = "person_user-1"
+    selected_key = storage_key("selected", conversation_id)
+    plugin.values[selected_key] = b'{"session_id":"ses_previous"}'
+    control_key = storage_key("control", conversation_id)
+    plugin.values[control_key] = (
+        b'{"location":"daemon","daemon_id":"daemon-1","session_capabilities":{"session/list":true,"session/load":true}}'
+    )
+    monkeypatch.setattr(
+        "components.commands.session_commands.run_daemon_control",
+        AsyncMock(
+            return_value={"sessions": [{"sessionId": "ses_next", "title": "Next", "updatedAt": "2026-01-02T00:00:00Z"}]}
+        ),
+    )
+    command = SwitchSession()
+    command.plugin = plugin
+    await command.initialize()
+    context = types.SimpleNamespace(
+        session=types.SimpleNamespace(launcher_type=types.SimpleNamespace(value="person"), launcher_id="user-1"),
+        privilege=1,
+        crt_params=["1"],
+        create_new_conversation=AsyncMock(side_effect=RuntimeError("reset failed")),
+    )
+    subcommand = command.registered_subcommands["*"].subcommand
+
+    results = [item async for item in subcommand(command, context)]
+
+    assert "reset failed" in results[0].text
+    assert plugin.values[selected_key] == b'{"session_id":"ses_previous"}'
+
+
+@pytest.mark.asyncio
+async def test_switch_rejects_session_list_without_cross_process_recovery(monkeypatch):
+    plugin = FakeCommandPlugin()
+    conversation_id = "person_user-1"
+    plugin.values[storage_key("control", conversation_id)] = (
+        b'{"location":"daemon","daemon_id":"daemon-1",'
+        b'"session_capabilities":{"session/list":true,"session/delete":true}}'
+    )
+    monkeypatch.setattr(
+        "components.commands.session_commands.run_daemon_control",
+        AsyncMock(
+            return_value={"sessions": [{"sessionId": "ses_next", "title": "Next", "updatedAt": "2026-01-02T00:00:00Z"}]}
+        ),
+    )
+    command = SwitchSession()
+    command.plugin = plugin
+    await command.initialize()
+    context = types.SimpleNamespace(
+        session=types.SimpleNamespace(launcher_type=types.SimpleNamespace(value="person"), launcher_id="user-1"),
+        privilege=1,
+        crt_params=["1"],
+        create_new_conversation=AsyncMock(),
+    )
+    subcommand = command.registered_subcommands["*"].subcommand
+
+    results = [item async for item in subcommand(command, context)]
+
+    assert "cross-process session recovery" in results[0].text
+    context.create_new_conversation.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_switch_survives_host_conversation_reset(monkeypatch):
+    plugin = FakeCommandPlugin()
+    runner = DefaultRunner()
+    runner.plugin = plugin
+    first_runner_context = types.SimpleNamespace(
+        conversation=types.SimpleNamespace(
+            conversation_id="conversation-before-reset",
+            session_id="person_user-1",
+        )
+    )
+    config = {
+        "provider": "opencode",
+        "location": "daemon",
+        "acp_command": "opencode acp",
+        "workspace": "/workspace",
+        "session_cwd": "/workspace",
+        "daemon_id": "daemon-1",
+        "timeout": 60,
+        "startup_timeout": 30,
+        "initialize_timeout": 30,
+        "permission_decision": "allow_once",
+        "env": {},
+        "session_capabilities": {"session/load": True},
+    }
+    assert await runner._managed_session_id(first_runner_context, config) == ""
+    plugin.values[storage_key("control", "person_user-1")] = (
+        b'{"location":"daemon","daemon_id":"daemon-1","session_capabilities":{"session/list":true,"session/load":true}}'
+    )
+
+    monkeypatch.setattr(
+        "components.commands.session_commands.run_daemon_control",
+        AsyncMock(
+            return_value={
+                "sessions": [
+                    {
+                        "sessionId": "ses_selected",
+                        "title": "Selected",
+                        "updatedAt": "2026-01-02T00:00:00Z",
+                    }
+                ]
+            }
+        ),
+    )
+    command = SwitchSession()
+    command.plugin = plugin
+    await command.initialize()
+    context = types.SimpleNamespace(
+        session=types.SimpleNamespace(launcher_type=types.SimpleNamespace(value="person"), launcher_id="user-1"),
+        privilege=1,
+        crt_params=["1"],
+        create_new_conversation=AsyncMock(),
+    )
+    subcommand = command.registered_subcommands["*"].subcommand
+    results = [item async for item in subcommand(command, context)]
+
+    assert "ses_selected" in results[0].text
+    context.create_new_conversation.assert_awaited_once()
+    next_runner_context = types.SimpleNamespace(
+        conversation=types.SimpleNamespace(
+            conversation_id="conversation-after-reset",
+            session_id=None,
+            launcher_type="person",
+            launcher_id="user-1",
+        )
+    )
+    assert await runner._managed_session_id(next_runner_context, config) == "ses_selected"
+
+
+@pytest.mark.asyncio
+async def test_stop_reports_group_ownership_denial():
+    plugin = FakeCommandPlugin()
+    command = StopRun()
+    command.plugin = plugin
+    context = types.SimpleNamespace(get_query_vars=AsyncMock(return_value={"_stop_command_denied": True}))
+
+    results = await _command_results(command, context)
+
+    assert "belongs to another group member" in results[0].text
+
+
+@pytest.mark.asyncio
+async def test_daemon_control_operation_uses_supplied_environment(monkeypatch):
+    seen = {}
+
+    class FakeClient:
+        def __init__(self, **kwargs):
+            seen.update(kwargs)
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *_args):
+            return None
+
+        async def initialize(self, timeout):
+            return {"agentCapabilities": {"methods": ["session/list"]}}
+
+        async def request(self, method, params, timeout):
+            return {"sessions": []}
+
+    monkeypatch.setattr("daemon.AcpStdioClient", FakeClient)
+    daemon = RunnerDaemon(url="ws://example.test", daemon_id="daemon-1")
+    daemon.emit_event = AsyncMock()
+
+    await daemon._run_control_operation(
+        "job-1",
+        "session.list",
+        {
+            "config": {
+                "acp_command": "opencode acp",
+                "workspace": "/workspace",
+                "env": {"ACP_PROFILE": "team"},
+            },
+            "arguments": {},
+        },
+    )
+
+    assert seen["env"] == {"ACP_PROFILE": "team"}
+
+
+@pytest.mark.asyncio
+async def test_prompt_only_acp_client_rejects_session_management(monkeypatch):
+    class PromptOnlyClient:
+        def __init__(self, **_kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *_args):
+            return None
+
+        async def initialize(self, timeout):
+            return {"agentCapabilities": {"methods": ["session/new", "session/prompt"]}}
+
+    monkeypatch.setattr("daemon.AcpStdioClient", PromptOnlyClient)
+    daemon = RunnerDaemon(url="ws://example.test", daemon_id="daemon-1")
+    daemon.emit_event = AsyncMock()
+
+    with pytest.raises(Exception, match="does not support session/list"):
+        await daemon._run_control_operation(
+            "job-1",
+            "session.list",
+            {"config": {"acp_command": "prompt-only-acp"}, "arguments": {}},
+        )

plugins/Runner/tests/fixtures/real_sdk_runtime_probe.py

@@ -132,10 +132,12 @@ async def probe(destination):
     discovery = ComponentDiscoveryEngine()
     manifest = discovery.load_component_manifest("manifest.yaml")
     assert manifest is not None
-    assert set(manifest.spec["components"]) == {"Runner"}
+    expected_kinds = {"Runner", "Command"} if PLUGIN == "acp-agent-runner" else {"Runner"}
+    assert set(manifest.spec["components"]) == expected_kinds
     components = discover_plugin_components(manifest, discovery)
-    assert len(components) == 1
-    component = components[0]
+    runner_components = [item for item in components if item.kind == "Runner"]
+    assert len(runner_components) == 1
+    component = runner_components[0]
     assert component.kind == "Runner"
     identity = f"{manifest.metadata.author}/{manifest.metadata.name}"
     descriptor = RunnerManifest(
@@ -193,7 +195,13 @@ async def probe(destination):
                     1,
                 )
                 container = (await rpc(conn, RuntimeToPluginAction.GET_PLUGIN_CONTAINER, {}, 2))[0]
-                instance = controller.plugin_container.components[0].component_instance
+                runner_instances = [
+                    item.component_instance
+                    for item in controller.plugin_container.components
+                    if isinstance(item.component_instance, Runner)
+                ]
+                assert len(runner_instances) == 1
+                instance = runner_instances[0]
                 assert isinstance(instance, Runner)
                 assert instance.plugin_identity == identity
                 invalid_results = await rpc(

plugins/catalog.json

@@ -50,7 +50,7 @@
   },
   {
     "id": "langbot-team/ACPAgentRunner",
-    "version": "0.1.16",
+    "version": "0.1.18",
     "category": "Runner",
     "path": "Runner/acp-agent-runner",
     "repository": "https://github.com/langbot-app/LangBot/tree/master/plugins/Runner/acp-agent-runner"

src/langbot/pkg/agent/runner/context_builder.py

@@ -278,6 +278,21 @@ class RunnerContextBuilder:
             }
 
             identity = event.legacy_identity
+            if (
+                identity is not None
+                and identity.workspace_id == event.workspace_id
+                and identity.bot_id == event.bot_id
+                and identity.pipeline_id == (binding.processor_id or binding.agent_id)
+            ):
+                # Query-backed platform events carry a Host-owned launcher
+                # identity even when the event-first conversation UUID changes
+                # after create_new_conversation(). Keep it separate from the
+                # protocol conversation UUID so reset-sensitive plugins can use
+                # a stable session key.
+                stable_launcher_type = identity.session_launcher_type or identity.query_launcher_type
+                stable_launcher_id = identity.session_launcher_id or identity.query_launcher_id
+                if stable_launcher_type in ('group', 'person') and stable_launcher_id:
+                    conversation['session_id'] = f'{stable_launcher_type}_{stable_launcher_id}'
             if (
                 binding.runner_config.get('user-id-source') == 'legacy-session'
                 and identity is not None