diff --git a/client/DESIGN.md b/client/DESIGN.md index cac491cd..38bfb646 100644 --- a/client/DESIGN.md +++ b/client/DESIGN.md @@ -217,8 +217,8 @@ CLI 层不自行定义“平行配置模型”，只对 `Config` 做覆盖合并 ### 8.3 宿主编排协议基线（planned） -> 本节是 Slice A 的目标设计输入，**尚未实现**。 -> 当前仓库事实仍以 `8.1`/`8.2` 描述的 landed 行为为准。 +> 本节记录 Issue #135 宿主编排协议的当前设计基线。 +> 其中 `8.3.3` 已在 Slice B 落地，`8.3.4`/`8.3.5` 的最小 control baseline 已在 Slice C 落地；capability discovery 仍保留给后续 Slice D。 #### 8.3.1 模式分层 @@ -226,7 +226,7 @@ CLI 层不自行定义“平行配置模型”，只对 `Config` 做覆盖合并 |---|---|---|---| | interactive | landed | `dare chat` | 允许 `dare>` prompt、内联审批提示、人类可读输出。 | | automation-json | landed / legacy | `run/script --output json` | 允许脚本消费 `log/event/result` 行输出，但不承诺宿主级稳定 envelope。 | -| headless | landed / partial | `run/script --headless` | 禁止 prompt、禁止内联审批提示，输出 versioned event envelope；外部 control plane 仍待后续 Slice。 | +| headless | landed | `run/script --headless` | 禁止 prompt、禁止内联审批提示，输出 versioned event envelope，并可选开启 `--control-stdin` 宿主控制面。 | #### 8.3.2 核心流程 @@ -279,9 +279,9 @@ headless 目标流程要求： 1. 当前 `--output json` 行结构视为 legacy automation schema。 2. headless envelope 使用独立 schema version，不直接复用现有 `type=log|event|result` 结构。 -3. 当前 landed 行为仅覆盖结构化事件流；`control-stdin`、capability handshake、动态 MCP 事件仍属于后续 Slice。 +3. 当前 landed 行为覆盖结构化事件流与 `control-stdin` 最小控制面；`actions:list` / capability handshake 与动态 MCP 事件仍属于后续 Slice。 -#### 8.3.4 control-stdin v1（planned） +#### 8.3.4 control-stdin v1（Slice C landed baseline） v1 设计选择：优先支持 `--control-stdin`，即 stdin 一行一个 JSON 命令帧。 @@ -300,15 +300,25 @@ v1 设计选择：优先支持 `--control-stdin`，即 stdin 一行一个 JSON 4. `result` 5. `error` -首批 action 基线： +协议约束： + +1. `schema_version` 固定为 `client-control-stdin.v1` +2. control result/error 与 headless event 一样走 `stdout` 多路复用，由 `schema_version` 区分 +3. `status:get` 的最小返回字段包含 `mode`、`status`、`running`、`active_task`、`pending_approvals` + +当前 landed action 基线： + +1. `approvals:list/poll/grant/deny/revoke` +2. `mcp:list/reload/show-tool` +3. `skills:list` +4. `status:get` + +当前仍未纳入 Slice C 基线： 1. `actions:list` -2. `approvals:list/poll/grant/deny/revoke` -3. `mcp:list/reload/show-tool` -4. `skills:list` -5. `status:get` +2. capability discovery / startup handshake -#### 8.3.5 错误处理与安全边界（planned） +#### 8.3.5 错误处理与安全边界（Slice C landed baseline） 1. `run/script --headless` 已禁止回落到 `input("dare> ")` 或内联 `approve>` 提示；审批等待超时会返回结构化 `task.failed` 事件。 2. control plane 的失败必须返回结构化错误对象，而不是只写 stdout 文案。 diff --git a/client/README.md b/client/README.md index 49b932f8..033d249e 100644 --- a/client/README.md +++ b/client/README.md @@ -339,18 +339,21 @@ Issue #135 对应的宿主编排能力目前分成“已落地”和“未落地 - 已接通的运行时事件：`approval.pending`、`approval.resolved`、`tool.invoke`、`tool.result`、`tool.error`、`model.response` - `chat` 不支持 `--headless` - `--headless` 不能与 legacy `--output json` 混用 +- `run` / `script --headless` 支持可选 `--control-stdin` + - 控制响应使用独立 schema：`client-control-stdin.v1` + - 当前已接通：`status:get`、`approvals:list/poll/grant/deny/revoke`、`mcp:list/reload/show-tool`、`skills:list` + - `mcp:unload` 仍然不是宿主协议 action；宿主发送时会得到结构化 `UNSUPPORTED_ACTION` + - 未支持或未完成的 action 会返回结构化 error，而不是回落到 prompt 文案 仍未落地： -- `--control-stdin` 结构化控制面 - `actions:list` / 启动握手式能力发现 -- MCP 动态控制对应的宿主协议面 当前推荐边界是： 1. 自动化脚本仍使用 `run/script --output json`。 2. 宿主事件流接入使用 `run/script --headless`。 -3. 审批、MCP、skills 等运行时控制当前仍以显式 CLI 命令或当前 transport/action 能力为主。 +3. 运行中控制当前优先使用 `--control-stdin` 做 `status:get`、approvals、MCP 与 `skills:list`；`actions:list` / 启动握手仍在后续 Slice 中。 4. 不要把当前 `log/event/result` 三类 JSON 行当作长期稳定的宿主协议。 补充说明： diff --git a/client/main.py b/client/main.py index 5baf3eb8..c947a587 100644 --- a/client/main.py +++ b/client/main.py @@ -4,9 +4,12 @@ import argparse import asyncio +import contextlib import dataclasses import json import logging +import sys +import threading import time from pathlib import Path from typing import Any, Awaitable, Callable, Iterable @@ -22,6 +25,7 @@ ) from client.commands.mcp import format_mcp_inspection, handle_mcp_tokens from client.parser.command import Command, CommandType, parse_command +from client.render.control import ControlStdinRenderer from client.render.headless import HeadlessRenderer from client.render.human import HumanRenderer from client.render.json import JsonRenderer @@ -202,6 +206,199 @@ def _is_execution_running(state: CLISessionState) -> bool: return task is not None and not task.done() +class _ControlStdinReader: + """Read control frames without pinning the asyncio default executor. + + A daemon thread is used here because cancellation cannot interrupt a + blocking `stdin.readline()` call. Leaving the read inside `to_thread()` + can keep the loop waiting on default-executor shutdown even after the + control task itself has been cancelled. + """ + + def __init__(self, *, stdin: Any | None = None) -> None: + self._stdin = sys.stdin if stdin is None else stdin + self._closed = False + self._loop: asyncio.AbstractEventLoop | None = None + self._queue: asyncio.Queue[str | None] | None = None + self._thread: threading.Thread | None = None + + def _ensure_started(self) -> None: + if self._thread is not None: + return + self._loop = asyncio.get_running_loop() + self._queue = asyncio.Queue() + self._thread = threading.Thread( + target=self._pump_lines, + name="dare-control-stdin", + daemon=True, + ) + self._thread.start() + + def _publish(self, line: str | None) -> None: + loop = self._loop + queue = self._queue + if loop is None or queue is None or loop.is_closed(): + return + with contextlib.suppress(RuntimeError): + loop.call_soon_threadsafe(queue.put_nowait, line) + + def _pump_lines(self) -> None: + while True: + raw = self._stdin.readline() + if self._closed: + return + if raw == "": + self._publish(None) + return + self._publish(raw.rstrip("\n")) + + async def read_line(self) -> str | None: + self._ensure_started() + assert self._queue is not None + return await self._queue.get() + + def close(self) -> None: + self._closed = True + + +_control_stdin_reader: _ControlStdinReader | None = None + + +def _get_control_stdin_reader() -> _ControlStdinReader: + global _control_stdin_reader + if _control_stdin_reader is None: + _control_stdin_reader = _ControlStdinReader() + return _control_stdin_reader + + +def _close_control_stdin_reader() -> None: + global _control_stdin_reader + if _control_stdin_reader is None: + return + _control_stdin_reader.close() + _control_stdin_reader = None + + +async def _read_control_stdin_line() -> str | None: + """Read one control frame line from stdin without pinning loop shutdown.""" + return await _get_control_stdin_reader().read_line() + + +def _status_snapshot(state: CLISessionState) -> dict[str, Any]: + """Project CLI session state into a stable host-control snapshot.""" + running = state.status == SessionStatus.RUNNING or _is_execution_running(state) + return { + "mode": state.mode.value, + "status": state.status.value, + "running": running, + "active_task": state.active_execution_description, + "pending_approvals": sorted(state.pending_runtime_approvals), + } + + +async def _dispatch_control_action( + *, + action_id: str, + params: dict[str, Any], + state: CLISessionState, + runtime: Any, + action_client: TransportActionClient, +) -> Any: + """Bridge host control actions onto the current CLI/runtime surface.""" + if action_id == "status:get": + return _status_snapshot(state) + resolved = ResourceAction.value_of(action_id) + if resolved in { + ResourceAction.APPROVALS_LIST, + ResourceAction.APPROVALS_POLL, + ResourceAction.APPROVALS_GRANT, + ResourceAction.APPROVALS_DENY, + ResourceAction.APPROVALS_REVOKE, + ResourceAction.MCP_LIST, + ResourceAction.MCP_RELOAD, + ResourceAction.MCP_SHOW_TOOL, + ResourceAction.SKILLS_LIST, + }: + return await action_client.invoke_action(resolved, **params) + _ = runtime + raise ActionClientError( + code="UNSUPPORTED_ACTION", + reason=f"unsupported control action: {action_id}", + target=action_id, + ) + + +async def _run_control_stdin_loop( + *, + state: CLISessionState, + runtime: Any, + action_client: TransportActionClient, +) -> None: + """Process structured host control commands from stdin.""" + renderer = ControlStdinRenderer() + try: + while True: + line = await _read_control_stdin_line() + if line is None: + return + if not line.strip(): + continue + + request_id = "?" + action_id = "?" + try: + payload = json.loads(line) + if not isinstance(payload, dict): + raise ValueError("control frame must be a JSON object") + request_id = str(payload.get("id", "?")).strip() or "?" + schema_version = str(payload.get("schema_version", "")).strip() + if schema_version != ControlStdinRenderer.schema_version: + raise ValueError( + "unsupported control schema_version: " + f"{schema_version or ''}" + ) + action_id = str(payload.get("action", "")).strip() + if not action_id: + raise ValueError("control action is required") + params = payload.get("params", {}) + if params is None: + params = {} + if not isinstance(params, dict): + raise ValueError("control params must be a JSON object") + result = await _dispatch_control_action( + action_id=action_id, + params=params, + state=state, + runtime=runtime, + action_client=action_client, + ) + except json.JSONDecodeError as exc: + renderer.emit( + request_id=request_id, + ok=False, + error={"code": "INVALID_JSON", "message": str(exc), "target": "control-stdin"}, + ) + continue + except ActionClientError as exc: + renderer.emit( + request_id=request_id, + ok=False, + error={"code": exc.code, "message": exc.reason, "target": exc.target}, + ) + continue + except Exception as exc: # noqa: BLE001 + renderer.emit( + request_id=request_id, + ok=False, + error={"code": "INVALID_CONTROL_FRAME", "message": str(exc), "target": action_id}, + ) + continue + + renderer.emit(request_id=request_id, ok=True, result=_serialize(result), error=None) + finally: + _close_control_stdin_reader() + + async def _execute_task_and_report( *, runtime: Any, @@ -679,22 +876,26 @@ async def _handle_shell_command( return False state.status = SessionStatus.RUNNING - if approval_watch is not None: - await _execute_task_with_approval_timeout( - runtime=runtime, - output=output, - state=state, - task_text=task_text, - approval_watch=approval_watch, - approval_timeout_seconds=approval_timeout_seconds, - ) - else: - await _execute_task_and_report( - runtime=runtime, - output=output, - state=state, - task_text=task_text, - ) + state.active_execution_description = task_text + try: + if approval_watch is not None: + await _execute_task_with_approval_timeout( + runtime=runtime, + output=output, + state=state, + task_text=task_text, + approval_watch=approval_watch, + approval_timeout_seconds=approval_timeout_seconds, + ) + else: + await _execute_task_and_report( + runtime=runtime, + output=output, + state=state, + task_text=task_text, + ) + finally: + state.active_execution_description = None state.status = SessionStatus.IDLE return False @@ -849,22 +1050,26 @@ async def _run_cli_loop( continue state.status = SessionStatus.RUNNING - if approval_watch is not None: - await _execute_task_with_approval_timeout( - runtime=runtime, - output=output, - state=state, - task_text=task_text, - approval_watch=approval_watch, - approval_timeout_seconds=approval_timeout_seconds, - ) - else: - await _execute_task_and_report( - runtime=runtime, - output=output, - state=state, - task_text=task_text, - ) + state.active_execution_description = task_text + try: + if approval_watch is not None: + await _execute_task_with_approval_timeout( + runtime=runtime, + output=output, + state=state, + task_text=task_text, + approval_watch=approval_watch, + approval_timeout_seconds=approval_timeout_seconds, + ) + else: + await _execute_task_and_report( + runtime=runtime, + output=output, + state=state, + task_text=task_text, + ) + finally: + state.active_execution_description = None state.status = SessionStatus.IDLE await _finalize_background_task_if_done(state, output=output) @@ -972,6 +1177,7 @@ async def _run_chat( mode: str, script_lines: list[str] | None, approval_timeout_seconds: float | None = None, + control_stdin: bool = False, ) -> int: state = CLISessionState(mode=_normalize_mode(mode)) if output.is_headless: @@ -1012,6 +1218,15 @@ def _handle_chat_approval_resolved(request_id: str) -> None: ), ) pump.start() + control_task: asyncio.Task[None] | None = None + if output.is_headless and control_stdin: + control_task = asyncio.create_task( + _run_control_stdin_loop( + state=state, + runtime=runtime, + action_client=action_client, + ) + ) try: if script_lines is not None: await _run_cli_loop( @@ -1062,6 +1277,10 @@ def _handle_chat_approval_resolved(request_id: str) -> None: await _wait_for_background_task(state, output=output) return 0 finally: + if control_task is not None: + control_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await control_task await pump.stop() @@ -1110,6 +1329,11 @@ def _build_parser() -> argparse.ArgumentParser: action="store_true", help="planned host-orchestrated headless mode for non-interactive execution", ) + run.add_argument( + "--control-stdin", + action="store_true", + help="enable structured control commands from stdin in headless mode", + ) script = sub.add_parser("script", help="run script and exit") script.add_argument("--file", required=True) @@ -1119,6 +1343,11 @@ def _build_parser() -> argparse.ArgumentParser: action="store_true", help="planned host-orchestrated headless mode for non-interactive execution", ) + script.add_argument( + "--control-stdin", + action="store_true", + help="enable structured control commands from stdin in headless mode", + ) script.add_argument( "--approval-timeout-seconds", type=float, @@ -1206,6 +1435,12 @@ def _validate_cli_args(args: argparse.Namespace, *, output: OutputFacade) -> int level="error", ) return 2 + if getattr(args, "control_stdin", False) and not getattr(args, "headless", False): + output.display( + "--control-stdin requires --headless; interactive and legacy modes do not expose the host control plane", + level="error", + ) + return 2 return None @@ -1275,6 +1510,7 @@ async def main(argv: list[str] | None = None) -> int: mode=args.mode, script_lines=lines, approval_timeout_seconds=None, + control_stdin=False, ) if command == "run": @@ -1325,21 +1561,41 @@ async def main(argv: list[str] | None = None) -> int: watch=approval_watch, auto_approve_tools=auto_tools, ) + + async def _handle_run_approval_pending( + request: dict[str, Any], + tool_name: str, + capability_id: str, + ) -> None: + request_id = str(request.get("request_id", "?")).strip() or "?" + state.mark_runtime_approval_pending(request_id) + await approval_policy.on_pending(request_id, tool_name, capability_id) + + def _handle_run_approval_resolved(request_id: str) -> None: + state.mark_runtime_approval_resolved(request_id) + approval_watch.mark_resolved(request_id) + pump = EventPump( client_channel=runtime.client_channel, on_event=lambda payload: _on_transport_event( payload, output=output, - on_approval_pending=lambda request, tool_name, capability_id: approval_policy.on_pending( - str(request.get("request_id", "?")).strip() or "?", - tool_name, - capability_id, - ), - on_approval_resolved=approval_watch.mark_resolved, + on_approval_pending=_handle_run_approval_pending, + on_approval_resolved=_handle_run_approval_resolved, ), ) pump.start() + control_task: asyncio.Task[None] | None = None + if args.control_stdin: + control_task = asyncio.create_task( + _run_control_stdin_loop( + state=state, + runtime=runtime, + action_client=action_client, + ) + ) state.status = SessionStatus.RUNNING + state.active_execution_description = args.task try: success = await _execute_task_with_approval_timeout( runtime=runtime, @@ -1350,6 +1606,11 @@ async def main(argv: list[str] | None = None) -> int: approval_timeout_seconds=args.approval_timeout_seconds, ) finally: + state.active_execution_description = None + if control_task is not None: + control_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await control_task await pump.stop() return 0 if success else 1 @@ -1367,6 +1628,7 @@ async def main(argv: list[str] | None = None) -> int: mode=args.mode, script_lines=lines, approval_timeout_seconds=script_approval_timeout_seconds, + control_stdin=args.control_stdin, ) if command == "approvals": diff --git a/client/render/control.py b/client/render/control.py new file mode 100644 index 00000000..fcfd9708 --- /dev/null +++ b/client/render/control.py @@ -0,0 +1,36 @@ +"""Renderer for structured control-stdin responses.""" + +from __future__ import annotations + +import json +import time +from typing import Any + + +class ControlStdinRenderer: + """Emit versioned responses for stdin-driven host control.""" + + schema_version = "client-control-stdin.v1" + + def emit( + self, + *, + request_id: str, + ok: bool, + result: Any = None, + error: Any = None, + ) -> None: + print( + json.dumps( + { + "schema_version": self.schema_version, + "ts": time.time(), + "id": request_id, + "ok": ok, + "result": result, + "error": error, + }, + ensure_ascii=False, + ), + flush=True, + ) diff --git a/docs/features/client-external-control-plane-v1.md b/docs/features/client-external-control-plane-v1.md index 256cc8ab..d93bb89b 100644 --- a/docs/features/client-external-control-plane-v1.md +++ b/docs/features/client-external-control-plane-v1.md @@ -34,6 +34,16 @@ mode: openspec - `git fetch origin` - `git worktree add .worktrees/client-external-control-plane-v1 -b codex/client-external-control-plane-v1 origin/main` +- `../../.venv/bin/python -m pytest tests/unit/test_client_cli.py -q -k 'control_stdin or control-stdin or chat_parser_rejects_control_stdin_flag or run_and_script_parser_accept_control_stdin_flag'` +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'control_stdin_status_get_emits_structured_result'` +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'bridges_approvals_list'` +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'bridges_additional_host_actions'` +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'script_headless_control_stdin_status_get_reports_active_task'` +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'status_get_reports_pending_approvals'` +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'surfaces_action_handler_failure or rejects_unsupported_action or bridges_approvals_list or control_stdin_status_get_emits_structured_result or script_headless_control_stdin_status_get_reports_active_task'` +- `../../.venv/bin/python -m pytest tests/unit/test_client_cli.py -q -k 'cancellation_does_not_block_default_executor_shutdown'` +- `../../.venv/bin/python -m pytest tests/unit/test_client_cli.py -q` +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q` - `openspec list` - `openspec validate client-external-control-plane-v1 --type change --strict --json --no-interactive` - `./scripts/ci/check_governance_evidence_truth.sh` @@ -42,14 +52,26 @@ mode: openspec - `git fetch origin`: confirmed `origin/main` includes merged Slice B via PR `#145`. - `git worktree add .worktrees/client-external-control-plane-v1 -b codex/client-external-control-plane-v1 origin/main`: created an isolated Slice C workspace from `origin/main` commit `bc39bc0`. -- `openspec list`: confirms Slice A / Slice B have been archived out of the active change list, and the new Slice C kickoff change is recognized as `0/7 tasks`. +- `../../.venv/bin/python -m pytest tests/unit/test_client_cli.py -q -k 'control_stdin or control-stdin or chat_parser_rejects_control_stdin_flag or run_and_script_parser_accept_control_stdin_flag'`: passed (`4` tests) after adding `--control-stdin` parser support and rejecting non-headless usage. +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'control_stdin_status_get_emits_structured_result'`: failed before the first control-loop implementation because no `client-control-stdin.v1` response frame was emitted; passed after landing the control stdin loop and `status:get` snapshot. +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'bridges_approvals_list'`: failed before approvals were bridged through canonical action dispatch; passed after exposing `approvals:list` via the control plane. +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'bridges_additional_host_actions'`: failed (`3` MCP cases red, `skills:list` already green) before `mcp:list/reload/show-tool` were admitted to the host bridge; passed (`4` tests) after extending the canonical action allow-list while keeping `mcp:unload` structurally rejected. +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'script_headless_control_stdin_status_get_reports_active_task'`: failed before script foreground execution tracked `active_task`; passed after aligning script/session state with run-mode snapshots. +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'status_get_reports_pending_approvals'`: failed before the review fix because `run --headless --control-stdin` only updated the approval timeout watch, leaving `status:get` snapshots blind to pending runtime approvals; passed after wiring run-mode approval pending/resolved events into `CLISessionState.pending_runtime_approvals`. +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q -k 'surfaces_action_handler_failure or rejects_unsupported_action or bridges_approvals_list or control_stdin_status_get_emits_structured_result or script_headless_control_stdin_status_get_reports_active_task'`: passed (`5` tests), covering happy path plus unsupported-action and handler-failure branches. +- `../../.venv/bin/python -m pytest tests/unit/test_client_cli.py -q -k 'cancellation_does_not_block_default_executor_shutdown'`: failed before the review fix because cancelling the control task still left a blocking `stdin.readline()` worker pinned in the default executor; passed after moving `control-stdin` reads onto a daemon thread + asyncio queue bridge. +- `../../.venv/bin/python -m pytest tests/unit/test_client_cli.py -q`: passed (`44` tests, `0` failures). +- `../../.venv/bin/python -m pytest tests/integration/test_client_cli_flow.py -q`: passed (`20` tests, `0` failures). +- `openspec list`: confirms Slice A / Slice B have been archived out of the active change list, and the active Slice C change now shows `✓ Complete`. - `openspec validate client-external-control-plane-v1 --type change --strict --json --no-interactive`: passed (`1/1` change valid, `0` issues). - `./scripts/ci/check_governance_evidence_truth.sh`: passed after the Slice C feature evidence block and prior-slice archive moves were synchronized. ### Behavior Verification -- Happy path: the planned Slice C contract narrows v1 external control to `--control-stdin`, preserving the landed headless event envelope from Slice B as the read-only observation channel. -- Error branch: Slice C keeps unknown action ids and unsupported MCP operations on the structured error path; it does not permit fallback to prompt text or undocumented CLI-only verbs such as `mcp:unload`. +- Happy path: `run/script --headless --control-stdin` now multiplexes `client-control-stdin.v1` responses on stdout alongside the existing headless event envelope, and `status:get` returns a structured session snapshot including the current active task. +- Happy path: approvals are now reachable from the host control plane through the canonical `approvals:*` action ids, without falling back to slash-command parsing. +- Happy path: `skills:list` and canonical MCP actions `mcp:list/reload/show-tool` now return structured control results with request correlation, and `mcp:show-tool` preserves `mcp_name/tool_name` params across the bridge. +- Error branch: unsupported actions such as `mcp:unload` return structured `UNSUPPORTED_ACTION` errors, and transport/action handler failures are surfaced as structured error responses instead of prompt text. ### Risks and Rollback @@ -60,5 +82,8 @@ mode: openspec - Slice A intent gate (merged): `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/141` - Slice B implementation gate (merged): `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/145` -- Slice C docs-only intent PR: `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/148` +- Slice C docs-only intent PR (merged): `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/148` +- Slice C implementation PR (open): `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/151` - Slice C spec-fold review thread: `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/148#discussion_r2872038646` +- Slice C implementation review thread: `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/151#discussion_r2872629793` +- Slice C implementation review thread: `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/151#discussion_r2872876273` diff --git a/docs/todos/2026-03-02_client_cli_host_orchestration_master_todo.md b/docs/todos/2026-03-02_client_cli_host_orchestration_master_todo.md index 8d8b7513..f740efa3 100644 --- a/docs/todos/2026-03-02_client_cli_host_orchestration_master_todo.md +++ b/docs/todos/2026-03-02_client_cli_host_orchestration_master_todo.md @@ -16,14 +16,14 @@ mode: openspec ## 认领声明（Claim Ledger） -> 当前状态：Slice A / Slice B 已于 2026-03-02 完成并合入 `main`，当前进入 Slice C kickoff。 +> 当前状态：Slice A / Slice B 已于 2026-03-02 完成并合入 `main`；Slice C 的 docs-only intent PR `#148` 已合入，当前实现已覆盖 approvals / MCP / skills / status 的 `control-stdin` 基线，待进入实现 PR / review gate。 > Slice C 负责外部 control plane v1；Slice D 继续承担 capability discovery 与宿主级回归测试。 | Claim ID | TODO Scope | Owner | Status | Declared At | Expires At | OpenSpec Change | Notes | |---|---|---|---|---|---|---|---| | CLM-20260302-CCLI-A | CCLI-001~CCLI-002 | bouillipx | done | 2026-03-02 | 2026-03-09 | `client-host-orchestration-doc-baseline` | Slice A: docs baseline 已随 PR `#141` 合入 `main`，待归档到 `openspec/changes/archive/2026-03-02-client-host-orchestration-doc-baseline/`。 | | CLM-20260302-CCLI-B | CCLI-003~CCLI-004 | bouillipx | done | 2026-03-02 | 2026-03-09 | `client-headless-event-envelope-v1` | Slice B: headless event envelope v1 已随 PR `#145` 合入 `main`，待归档到 `openspec/changes/archive/2026-03-02-client-headless-event-envelope-v1/`。 | -| CLM-20260302-CCLI-C | CCLI-005~CCLI-006 | bouillipx | active | 2026-03-02 | 2026-03-09 | `client-external-control-plane-v1` | Slice C: 建立 `--control-stdin` v1 基线、MCP/approval/status/skills 外部控制 contract 与 docs-only intent PR payload。 | +| CLM-20260302-CCLI-C | CCLI-005~CCLI-006 | bouillipx | active | 2026-03-02 | 2026-03-09 | `client-external-control-plane-v1` | Slice C: docs-only intent PR `#148` 已合入；当前已落地 `--control-stdin` 最小基线，以及 `status:get`、approvals、`mcp:list/reload/show-tool`、`skills:list` host bridge；`actions:list` 仍留给 Slice D。 | ## 切片规划 @@ -42,8 +42,8 @@ mode: openspec | CCLI-002 | P0 | done | CCLI-GAP-006 | `client-host-orchestration-doc-baseline` | 更新 `client/README.md`，明确当前 `--output json` 是 legacy automation schema，补充后续宿主协议模式的兼容说明。 | bouillipx | `client/README.md`；`docs/features/archive/client-host-orchestration-doc-baseline.md` | 2026-03-02 | | CCLI-003 | P1 | done | CCLI-GAP-001 | `client-headless-event-envelope-v1` | 为 `client` 设计并实现显式 headless 模式，定义禁止 prompt / 禁止内联审批 / 只输出协议帧的行为边界。 | bouillipx | `client/main.py`；`client/session.py`；`tests/integration/test_client_cli_flow.py`；`docs/features/archive/client-headless-event-envelope-v1.md` | 2026-03-02 | | CCLI-004 | P1 | done | CCLI-GAP-002 | `client-headless-event-envelope-v1` | 设计并实现 versioned event envelope（至少含 `schema_version`、`run_id`、`seq`、`event`、`data`），并定义与现有 JSON 输出的兼容策略。 | bouillipx | `client/render/headless.py`；`tests/unit/test_client_cli.py`；`tests/integration/test_client_cli_flow.py`；`docs/features/archive/client-headless-event-envelope-v1.md` | 2026-03-02 | -| CCLI-005 | P1 | todo | CCLI-GAP-003 | `client-external-control-plane-v1` | 设计外部控制协议入口（如 `control-stdin` 或 loopback RPC），覆盖 approvals / MCP / skills / status 的结构化控制。 | bouillipx | `client/main.py`；`client/runtime/action_client.py`；相关 OpenSpec design/specs/tasks | 2026-03-02 | -| CCLI-006 | P2 | todo | CCLI-GAP-005 | `client-external-control-plane-v1` | 将当前 canonical MCP actions（首批为 `mcp:list/reload/show-tool`）接入外部 control plane，并明确运行中生效与错误处理语义。CLI 层 `unload` 待后续补 canonical action 后再纳入宿主协议面。 | bouillipx | `client/commands/mcp.py`；相关集成测试 | 2026-03-02 | +| CCLI-005 | P1 | doing | CCLI-GAP-003 | `client-external-control-plane-v1` | 设计外部控制协议入口（如 `control-stdin` 或 loopback RPC），覆盖 approvals / MCP / skills / status 的结构化控制。 | bouillipx | `client/main.py`；`client/render/control.py`；`tests/unit/test_client_cli.py`；`tests/integration/test_client_cli_flow.py`；相关 OpenSpec design/specs/tasks | 2026-03-02 | +| CCLI-006 | P2 | doing | CCLI-GAP-005 | `client-external-control-plane-v1` | 将当前 canonical MCP actions（首批为 `mcp:list/reload/show-tool`）接入外部 control plane，并明确运行中生效与错误处理语义。CLI 层 `unload` 待后续补 canonical action 后再纳入宿主协议面。 | bouillipx | `client/main.py`；`tests/integration/test_client_cli_flow.py`；`docs/features/client-external-control-plane-v1.md` | 2026-03-02 | | CCLI-007 | P2 | todo | CCLI-GAP-004 | `client-capability-discovery-and-host-tests` | 将 `actions:list` 提升到 CLI 宿主协议面，并定义启动握手或显式查询命令。 | TBD | `dare_framework/transport/interaction/resource_action.py`；`client/main.py`；相关文档 | 2026-03-02 | | CCLI-008 | P1 | todo | CCLI-GAP-006 | `client-capability-discovery-and-host-tests` | 新增 headless 协议稳定性、外部控制、能力发现三组集成测试，并回写 README / 设计文档中的验证锚点。 | TBD | `tests/integration/test_client_cli_flow.py`；新增协议测试文件 | 2026-03-02 | diff --git a/openspec/changes/client-external-control-plane-v1/design.md b/openspec/changes/client-external-control-plane-v1/design.md index d3e70583..f0e77a2a 100644 --- a/openspec/changes/client-external-control-plane-v1/design.md +++ b/openspec/changes/client-external-control-plane-v1/design.md @@ -49,7 +49,9 @@ Issue #135 的 Slice C 需要补足“写路径”，但不应同时引入网络 - `ok` - `result` - `error` +- v1 `schema_version` 固定为 `client-control-stdin.v1`。 - 结果帧不复用 `event` 字段，避免把控制往返混入只读事件流语义。 +- result/error 与 headless event 一样走 `stdout` 多路复用，由 `schema_version` 区分。 ### Decision 3: v1 action 范围收敛到现有 canonical surface @@ -65,6 +67,7 @@ Issue #135 的 Slice C 需要补足“写路径”，但不应同时引入网络 - `skills:list` - `status:get` - `status:get` 由 CLI session state 提供结构化快照；其余 action 复用现有运行时 handler。 +- `status:get` 的最小返回字段为 `mode`、`status`、`running`、`active_task`、`pending_approvals`。 - `mcp:unload` 继续留在 CLI 命令面，不进入 v1 协议基线。 ### Decision 4: 错误必须结构化且不可回落为 prompt UX @@ -93,6 +96,4 @@ Issue #135 的 Slice C 需要补足“写路径”，但不应同时引入网络 ## Open Questions -- control result/error 是否输出到 stdout 还是 stderr，才能兼容宿主同时读取事件流与控制响应？ -- `status:get` 的最小返回字段是否应包含 `mode/status/running/active_task/pending_approvals`？ - `mcp:show-tool` 与现有 `/mcp inspect` 的输出投影是否需要完全一致，还是先只保证结构化字段稳定？ diff --git a/openspec/changes/client-external-control-plane-v1/tasks.md b/openspec/changes/client-external-control-plane-v1/tasks.md index e84e1d0b..cd1df0dc 100644 --- a/openspec/changes/client-external-control-plane-v1/tasks.md +++ b/openspec/changes/client-external-control-plane-v1/tasks.md @@ -1,15 +1,15 @@ ## 1. Control-Stdin Entry -- [ ] 1.1 为 headless `run/script` 增加 `--control-stdin` 参数，并定义与 interactive / legacy 输入模式的兼容边界。 -- [ ] 1.2 建立 command/result/error frame 解析与响应逻辑，保证请求 `id` 相关联且错误结构化返回。 +- [x] 1.1 为 headless `run/script` 增加 `--control-stdin` 参数，并定义与 interactive / legacy 输入模式的兼容边界。 +- [x] 1.2 建立 command/result/error frame 解析与响应逻辑，保证请求 `id` 相关联且错误结构化返回。 ## 2. Action Bridging -- [ ] 2.1 将 `approvals:list/poll/grant/deny/revoke` 暴露到外部 control plane。 -- [ ] 2.2 将 `mcp:list/reload/show-tool` 与 `skills:list` 暴露到外部 control plane，并显式排除 `mcp:unload`。 -- [ ] 2.3 提供 `status:get` 的结构化会话快照返回。 +- [x] 2.1 将 `approvals:list/poll/grant/deny/revoke` 暴露到外部 control plane。 +- [x] 2.2 将 `mcp:list/reload/show-tool` 与 `skills:list` 暴露到外部 control plane，并显式排除 `mcp:unload`。 +- [x] 2.3 提供 `status:get` 的结构化会话快照返回。 ## 3. Verification And Evidence -- [ ] 3.1 增加 `control-stdin` happy path、unknown action、handler failure、session edge case 的测试。 -- [ ] 3.2 验证 Slice B headless event envelope 未被控制面引入回归破坏，并回写 `docs/features/client-external-control-plane-v1.md` 的 Evidence 区块。 +- [x] 3.1 增加 `control-stdin` happy path、unknown action、handler failure、session edge case 的测试。 +- [x] 3.2 验证 Slice B headless event envelope 未被控制面引入回归破坏，并回写 `docs/features/client-external-control-plane-v1.md` 的 Evidence 区块。 diff --git a/tests/integration/test_client_cli_flow.py b/tests/integration/test_client_cli_flow.py index d8ece546..a9ad36eb 100644 --- a/tests/integration/test_client_cli_flow.py +++ b/tests/integration/test_client_cli_flow.py @@ -48,6 +48,20 @@ async def invoke_action(self, action: Any, **params: Any) -> dict[str, Any]: type(self).calls.append((action_id, dict(params))) if action_id == "approvals:list": return {"pending": [], "rules": []} + if action_id == "skills:list": + return {"skills": [{"name": "development-workflow"}]} + if action_id == "mcp:list": + return {"mcps": ["demo"], "mcp_paths": ["/tmp/demo"], "tools": []} + if action_id == "mcp:reload": + return {"ok": True, "reloaded": params.get("mcp_name") or "all"} + if action_id == "mcp:show-tool": + return { + "found": True, + "tool": { + "name": params.get("tool_name", "?"), + "mcp_name": params.get("mcp_name", "?"), + }, + } return {"ok": True} async def invoke_control(self, control: Any, **params: Any) -> dict[str, Any]: @@ -226,6 +240,644 @@ async def _fake_run_task(*, agent, task_text, conversation_id=None, transport=No assert runtime.closed is True +@pytest.mark.asyncio +async def test_main_run_headless_control_stdin_status_get_emits_structured_result( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + client_main = importlib.import_module("client.main") + config = _config_for_tests(tmp_path) + runtime = _FakeRuntime(config=config) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + return runtime + + class _OkResult: + success = True + output = {"content": "assistant says hi"} + errors: list[str] = [] + + async def _slow_run_task(*, agent, task_text, conversation_id=None, transport=None): # noqa: ANN001 + _ = (agent, task_text, conversation_id, transport) + await asyncio.sleep(0.2) + return _OkResult() + + control_lines = iter( + [ + json.dumps( + { + "schema_version": "client-control-stdin.v1", + "id": "ctl-1", + "action": "status:get", + "params": {}, + } + ), + None, + ] + ) + + async def _fake_read_control_stdin_line() -> str | None: + await asyncio.sleep(0) + return next(control_lines) + + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + monkeypatch.setattr(client_main, "run_task", _slow_run_task) + monkeypatch.setattr( + client_main, + "_read_control_stdin_line", + _fake_read_control_stdin_line, + raising=False, + ) + + rc = await client_main.main( + [ + "--workspace", + config.workspace_dir, + "--user-dir", + config.user_dir, + "run", + "--task", + "do one task", + "--headless", + "--control-stdin", + ] + ) + + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()] + headless_events = [line for line in lines if line["schema_version"] == "client-headless-event-envelope.v1"] + control_frames = [line for line in lines if line["schema_version"] == "client-control-stdin.v1"] + + assert rc == 0 + assert [line["event"] for line in headless_events[:3]] == [ + "session.started", + "task.started", + "task.completed", + ] + assert len(control_frames) == 1 + assert control_frames[0]["id"] == "ctl-1" + assert control_frames[0]["ok"] is True + assert control_frames[0]["result"]["mode"] == "execute" + assert control_frames[0]["result"]["status"] == "running" + assert control_frames[0]["result"]["running"] is True + assert control_frames[0]["result"]["active_task"] == "do one task" + assert runtime.closed is True + + +@pytest.mark.asyncio +async def test_main_run_headless_control_stdin_status_get_reports_pending_approvals( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + client_main = importlib.import_module("client.main") + config = _config_for_tests(tmp_path) + pending_event = { + "type": "approval_pending", + "resp": { + "request": {"request_id": "req-status-pending-1"}, + "capability_id": "run_command", + }, + } + runtime = _FakeRuntime(config=config, events=[pending_event]) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + return runtime + + class _OkResult: + success = True + output = {"content": "assistant says hi"} + errors: list[str] = [] + + async def _slow_run_task(*, agent, task_text, conversation_id=None, transport=None): # noqa: ANN001 + _ = (agent, task_text, conversation_id, transport) + await asyncio.sleep(0.4) + return _OkResult() + + approval_seen = asyncio.Event() + original_on_transport_event = client_main._on_transport_event + + async def _observed_on_transport_event(payload, *, output, **kwargs): # noqa: ANN001 + await original_on_transport_event(payload, output=output, **kwargs) + if payload.get("type") == "approval_pending": + approval_seen.set() + + control_sent = False + + async def _fake_read_control_stdin_line() -> str | None: + nonlocal control_sent + if control_sent: + await asyncio.sleep(0) + return None + await approval_seen.wait() + control_sent = True + return json.dumps( + { + "schema_version": "client-control-stdin.v1", + "id": "ctl-status-pending-1", + "action": "status:get", + "params": {}, + } + ) + + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + monkeypatch.setattr(client_main, "run_task", _slow_run_task) + monkeypatch.setattr(client_main, "_on_transport_event", _observed_on_transport_event) + monkeypatch.setattr( + client_main, + "_read_control_stdin_line", + _fake_read_control_stdin_line, + raising=False, + ) + + rc = await client_main.main( + [ + "--workspace", + config.workspace_dir, + "--user-dir", + config.user_dir, + "run", + "--task", + "do one task", + "--headless", + "--control-stdin", + "--approval-timeout-seconds", + "2.0", + ] + ) + + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()] + control_frames = [line for line in lines if line["schema_version"] == "client-control-stdin.v1"] + + assert rc == 0 + assert len(control_frames) == 1 + assert control_frames[0]["id"] == "ctl-status-pending-1" + assert control_frames[0]["ok"] is True + assert control_frames[0]["result"]["pending_approvals"] == ["req-status-pending-1"] + assert runtime.closed is True + + +@pytest.mark.asyncio +async def test_main_run_headless_control_stdin_bridges_approvals_list( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + client_main = importlib.import_module("client.main") + config = _config_for_tests(tmp_path) + runtime = _FakeRuntime(config=config) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + return runtime + + class _OkResult: + success = True + output = {"content": "assistant says hi"} + errors: list[str] = [] + + async def _slow_run_task(*, agent, task_text, conversation_id=None, transport=None): # noqa: ANN001 + _ = (agent, task_text, conversation_id, transport) + await asyncio.sleep(0.2) + return _OkResult() + + control_lines = iter( + [ + json.dumps( + { + "schema_version": "client-control-stdin.v1", + "id": "ctl-approvals-1", + "action": "approvals:list", + "params": {}, + } + ), + None, + ] + ) + + async def _fake_read_control_stdin_line() -> str | None: + await asyncio.sleep(0) + return next(control_lines) + + _FakeActionClient.calls = [] + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + monkeypatch.setattr(client_main, "TransportActionClient", _FakeActionClient) + monkeypatch.setattr(client_main, "run_task", _slow_run_task) + monkeypatch.setattr( + client_main, + "_read_control_stdin_line", + _fake_read_control_stdin_line, + raising=False, + ) + + rc = await client_main.main( + [ + "--workspace", + config.workspace_dir, + "--user-dir", + config.user_dir, + "run", + "--task", + "do one task", + "--headless", + "--control-stdin", + ] + ) + + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()] + control_frames = [line for line in lines if line["schema_version"] == "client-control-stdin.v1"] + + assert rc == 0 + assert len(control_frames) == 1 + assert control_frames[0]["id"] == "ctl-approvals-1" + assert control_frames[0]["ok"] is True + assert control_frames[0]["result"] == {"pending": [], "rules": []} + assert any(action_id == "approvals:list" for action_id, _ in _FakeActionClient.calls) + assert runtime.closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("action_id", "params", "expected_result", "expected_call"), + [ + ( + "skills:list", + {}, + {"skills": [{"name": "development-workflow"}]}, + ("skills:list", {}), + ), + ( + "mcp:list", + {"mcp_name": "demo"}, + {"mcps": ["demo"], "mcp_paths": ["/tmp/demo"], "tools": []}, + ("mcp:list", {"mcp_name": "demo"}), + ), + ( + "mcp:reload", + {"mcp_name": "demo"}, + {"ok": True, "reloaded": "demo"}, + ("mcp:reload", {"mcp_name": "demo"}), + ), + ( + "mcp:show-tool", + {"mcp_name": "demo", "tool_name": "search"}, + {"found": True, "tool": {"name": "search", "mcp_name": "demo"}}, + ("mcp:show-tool", {"mcp_name": "demo", "tool_name": "search"}), + ), + ], +) +async def test_main_run_headless_control_stdin_bridges_additional_host_actions( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + action_id: str, + params: dict[str, Any], + expected_result: dict[str, Any], + expected_call: tuple[str, dict[str, Any]], +) -> None: + client_main = importlib.import_module("client.main") + config = _config_for_tests(tmp_path) + runtime = _FakeRuntime(config=config) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + return runtime + + class _OkResult: + success = True + output = {"content": "assistant says hi"} + errors: list[str] = [] + + async def _slow_run_task(*, agent, task_text, conversation_id=None, transport=None): # noqa: ANN001 + _ = (agent, task_text, conversation_id, transport) + await asyncio.sleep(0.2) + return _OkResult() + + control_lines = iter( + [ + json.dumps( + { + "schema_version": "client-control-stdin.v1", + "id": f"ctl-{action_id}-1", + "action": action_id, + "params": params, + } + ), + None, + ] + ) + + async def _fake_read_control_stdin_line() -> str | None: + await asyncio.sleep(0) + return next(control_lines) + + _FakeActionClient.calls = [] + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + monkeypatch.setattr(client_main, "TransportActionClient", _FakeActionClient) + monkeypatch.setattr(client_main, "run_task", _slow_run_task) + monkeypatch.setattr( + client_main, + "_read_control_stdin_line", + _fake_read_control_stdin_line, + raising=False, + ) + + rc = await client_main.main( + [ + "--workspace", + config.workspace_dir, + "--user-dir", + config.user_dir, + "run", + "--task", + "do one task", + "--headless", + "--control-stdin", + ] + ) + + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()] + control_frames = [line for line in lines if line["schema_version"] == "client-control-stdin.v1"] + + assert rc == 0 + assert len(control_frames) == 1 + assert control_frames[0]["id"] == f"ctl-{action_id}-1" + assert control_frames[0]["ok"] is True + assert control_frames[0]["result"] == expected_result + assert expected_call in _FakeActionClient.calls + assert runtime.closed is True + + +@pytest.mark.asyncio +async def test_main_script_headless_control_stdin_status_get_reports_active_task( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + client_main = importlib.import_module("client.main") + config = _config_for_tests(tmp_path) + runtime = _FakeRuntime(config=config) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + return runtime + + class _OkResult: + success = True + output = {"content": "assistant says hi"} + errors: list[str] = [] + + async def _slow_run_task(*, agent, task_text, conversation_id=None, transport=None): # noqa: ANN001 + _ = (agent, task_text, conversation_id, transport) + await asyncio.sleep(0.2) + return _OkResult() + + control_lines = iter( + [ + json.dumps( + { + "schema_version": "client-control-stdin.v1", + "id": "ctl-script-status-1", + "action": "status:get", + "params": {}, + } + ), + None, + ] + ) + + async def _fake_read_control_stdin_line() -> str | None: + await asyncio.sleep(0) + return next(control_lines) + + script_path = tmp_path / "headless-control.script.txt" + script_path.write_text("task one\n", encoding="utf-8") + + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + monkeypatch.setattr(client_main, "run_task", _slow_run_task) + monkeypatch.setattr( + client_main, + "_read_control_stdin_line", + _fake_read_control_stdin_line, + raising=False, + ) + + rc = await client_main.main( + [ + "--workspace", + config.workspace_dir, + "--user-dir", + config.user_dir, + "script", + "--file", + str(script_path), + "--headless", + "--control-stdin", + ] + ) + + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()] + control_frames = [line for line in lines if line["schema_version"] == "client-control-stdin.v1"] + + assert rc == 0 + assert len(control_frames) == 1 + assert control_frames[0]["id"] == "ctl-script-status-1" + assert control_frames[0]["ok"] is True + assert control_frames[0]["result"]["status"] == "running" + assert control_frames[0]["result"]["running"] is True + assert control_frames[0]["result"]["active_task"] == "task one" + assert runtime.closed is True + + +@pytest.mark.asyncio +async def test_main_run_headless_control_stdin_rejects_unsupported_action( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + client_main = importlib.import_module("client.main") + config = _config_for_tests(tmp_path) + runtime = _FakeRuntime(config=config) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + return runtime + + class _OkResult: + success = True + output = {"content": "assistant says hi"} + errors: list[str] = [] + + async def _slow_run_task(*, agent, task_text, conversation_id=None, transport=None): # noqa: ANN001 + _ = (agent, task_text, conversation_id, transport) + await asyncio.sleep(0.2) + return _OkResult() + + control_lines = iter( + [ + json.dumps( + { + "schema_version": "client-control-stdin.v1", + "id": "ctl-unsupported-1", + "action": "mcp:unload", + "params": {}, + } + ), + None, + ] + ) + + async def _fake_read_control_stdin_line() -> str | None: + await asyncio.sleep(0) + return next(control_lines) + + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + monkeypatch.setattr(client_main, "run_task", _slow_run_task) + monkeypatch.setattr( + client_main, + "_read_control_stdin_line", + _fake_read_control_stdin_line, + raising=False, + ) + + rc = await client_main.main( + [ + "--workspace", + config.workspace_dir, + "--user-dir", + config.user_dir, + "run", + "--task", + "do one task", + "--headless", + "--control-stdin", + ] + ) + + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()] + control_frames = [line for line in lines if line["schema_version"] == "client-control-stdin.v1"] + + assert rc == 0 + assert len(control_frames) == 1 + assert control_frames[0]["id"] == "ctl-unsupported-1" + assert control_frames[0]["ok"] is False + assert control_frames[0]["error"]["code"] == "UNSUPPORTED_ACTION" + assert control_frames[0]["error"]["target"] == "mcp:unload" + assert runtime.closed is True + + +@pytest.mark.asyncio +async def test_main_run_headless_control_stdin_surfaces_action_handler_failure( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + client_main = importlib.import_module("client.main") + config = _config_for_tests(tmp_path) + runtime = _FakeRuntime(config=config) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + return runtime + + class _OkResult: + success = True + output = {"content": "assistant says hi"} + errors: list[str] = [] + + class _FailingActionClient(_FakeActionClient): + async def invoke_action(self, action: Any, **params: Any) -> dict[str, Any]: + action_id = action.value if hasattr(action, "value") else str(action) + type(self).calls.append((action_id, dict(params))) + raise client_main.ActionClientError( + code="ACTION_HANDLER_FAILED", + reason="approval backend unavailable", + target=action_id, + ) + + async def _slow_run_task(*, agent, task_text, conversation_id=None, transport=None): # noqa: ANN001 + _ = (agent, task_text, conversation_id, transport) + await asyncio.sleep(0.2) + return _OkResult() + + control_lines = iter( + [ + json.dumps( + { + "schema_version": "client-control-stdin.v1", + "id": "ctl-approvals-fail-1", + "action": "approvals:list", + "params": {}, + } + ), + None, + ] + ) + + async def _fake_read_control_stdin_line() -> str | None: + await asyncio.sleep(0) + return next(control_lines) + + _FailingActionClient.calls = [] + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + monkeypatch.setattr(client_main, "TransportActionClient", _FailingActionClient) + monkeypatch.setattr(client_main, "run_task", _slow_run_task) + monkeypatch.setattr( + client_main, + "_read_control_stdin_line", + _fake_read_control_stdin_line, + raising=False, + ) + + rc = await client_main.main( + [ + "--workspace", + config.workspace_dir, + "--user-dir", + config.user_dir, + "run", + "--task", + "do one task", + "--headless", + "--control-stdin", + ] + ) + + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()] + control_frames = [line for line in lines if line["schema_version"] == "client-control-stdin.v1"] + + assert rc == 0 + assert len(control_frames) == 1 + assert control_frames[0]["id"] == "ctl-approvals-fail-1" + assert control_frames[0]["ok"] is False + assert control_frames[0]["error"]["code"] == "ACTION_HANDLER_FAILED" + assert control_frames[0]["error"]["target"] == "approvals:list" + assert runtime.closed is True + + @pytest.mark.asyncio async def test_main_doctor_human_uses_configured_log_path( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/test_client_cli.py b/tests/unit/test_client_cli.py index 2e35d611..2714c94b 100644 --- a/tests/unit/test_client_cli.py +++ b/tests/unit/test_client_cli.py @@ -1488,6 +1488,65 @@ def test_chat_parser_rejects_headless_flag() -> None: assert excinfo.value.code == 2 +def test_run_and_script_parser_accept_control_stdin_flag() -> None: + client_main = importlib.import_module("client.main") + parser = client_main._build_parser() + + run_args = parser.parse_args(["run", "--task", "summarize readme", "--headless", "--control-stdin"]) + script_args = parser.parse_args( + ["script", "--file", "tasks.txt", "--headless", "--control-stdin"] + ) + + assert run_args.control_stdin is True + assert script_args.control_stdin is True + + +def test_chat_parser_rejects_control_stdin_flag() -> None: + client_main = importlib.import_module("client.main") + parser = client_main._build_parser() + + with pytest.raises(SystemExit) as excinfo: + parser.parse_args(["chat", "--control-stdin"]) + + assert excinfo.value.code == 2 + + +def test_read_control_stdin_line_cancellation_does_not_block_default_executor_shutdown( + monkeypatch, +) -> None: + client_main = importlib.import_module("client.main") + entered = threading.Event() + release = threading.Event() + + class _BlockingStdin: + def readline(self) -> str: + entered.set() + release.wait(timeout=5.0) + return "" + + async def _exercise() -> None: + task = asyncio.create_task(client_main._read_control_stdin_line()) + deadline = time.time() + 1.0 + while not entered.is_set(): + assert time.time() < deadline + await asyncio.sleep(0.01) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + monkeypatch.setattr(client_main.sys, "stdin", _BlockingStdin()) + + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(asyncio.wait_for(_exercise(), timeout=1.0)) + loop.run_until_complete(asyncio.wait_for(loop.shutdown_default_executor(), timeout=0.2)) + finally: + release.set() + with contextlib.suppress(Exception): + loop.run_until_complete(asyncio.wait_for(loop.shutdown_default_executor(), timeout=1.0)) + loop.close() + + @pytest.mark.asyncio async def test_main_run_headless_rejects_legacy_output(monkeypatch, tmp_path, capsys) -> None: client_main = importlib.import_module("client.main") @@ -1539,3 +1598,99 @@ async def _fake_bootstrap_runtime(_options): # noqa: ANN001 assert payload["schema_version"] == "client-headless-event-envelope.v1" assert payload["event"] == "log.error" assert "cannot combine --headless with legacy --output" in payload["data"]["message"] + + +@pytest.mark.asyncio +async def test_main_run_control_stdin_requires_headless(monkeypatch, tmp_path, capsys) -> None: + client_main = importlib.import_module("client.main") + workspace = tmp_path / "workspace" + user_dir = tmp_path / "user" + workspace.mkdir(parents=True, exist_ok=True) + user_dir.mkdir(parents=True, exist_ok=True) + + config = Config.from_dict( + { + "workspace_dir": str(workspace), + "user_dir": str(user_dir), + "llm": { + "adapter": "openai", + "model": "gpt-4o-mini", + "api_key": "dummy", + }, + } + ) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + raise AssertionError("bootstrap_runtime should not run for invalid control-stdin args") + + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + + rc = await client_main.main( + [ + "--workspace", + str(workspace), + "--user-dir", + str(user_dir), + "run", + "--task", + "summarize readme", + "--control-stdin", + ] + ) + + assert rc == 2 + output = capsys.readouterr().out + assert "--control-stdin requires --headless" in output + + +@pytest.mark.asyncio +async def test_main_script_control_stdin_requires_headless(monkeypatch, tmp_path, capsys) -> None: + client_main = importlib.import_module("client.main") + workspace = tmp_path / "workspace" + user_dir = tmp_path / "user" + workspace.mkdir(parents=True, exist_ok=True) + user_dir.mkdir(parents=True, exist_ok=True) + script_path = tmp_path / "tasks.txt" + script_path.write_text("task one\n", encoding="utf-8") + + config = Config.from_dict( + { + "workspace_dir": str(workspace), + "user_dir": str(user_dir), + "llm": { + "adapter": "openai", + "model": "gpt-4o-mini", + "api_key": "dummy", + }, + } + ) + + def _fake_load_effective_config(_options): # noqa: ANN001 + return object(), config + + async def _fake_bootstrap_runtime(_options): # noqa: ANN001 + raise AssertionError("bootstrap_runtime should not run for invalid control-stdin args") + + monkeypatch.setattr(client_main, "load_effective_config", _fake_load_effective_config) + monkeypatch.setattr(client_main, "bootstrap_runtime", _fake_bootstrap_runtime) + + rc = await client_main.main( + [ + "--workspace", + str(workspace), + "--user-dir", + str(user_dir), + "script", + "--file", + str(script_path), + "--control-stdin", + ] + ) + + assert rc == 2 + output = capsys.readouterr().out + assert "--control-stdin requires --headless" in output