diff --git a/coding-agent/src/coding_agent/__main__.py b/coding-agent/src/coding_agent/__main__.py index f8580e76d..511f4654c 100644 --- a/coding-agent/src/coding_agent/__main__.py +++ b/coding-agent/src/coding_agent/__main__.py @@ -418,6 +418,113 @@ def serve(port: int, host: str): uvicorn.run(app, host=host, port=port) +@main.group() +def remote() -> None: + """Manage and use remote Coding Agent servers.""" + + +@remote.command("add") +@click.argument("name") +@click.argument("url") +@click.option("--token", default=None, help="Bearer token for the remote server") +def remote_add(name: str, url: str, token: str | None) -> None: + """Add or update a named remote endpoint.""" + from coding_agent.remote.client import add_remote + + endpoint = add_remote(name, url, token) + click.echo(f"Added remote {endpoint.name}: {endpoint.url}") + + +@remote.command("list") +def remote_list() -> None: + """List named remote endpoints.""" + from coding_agent.remote.client import load_remotes + + remotes = load_remotes() + if not remotes: + click.echo("No remotes configured.") + return + for endpoint in remotes.values(): + auth = "token" if endpoint.token is not None else "no-token" + click.echo(f"{endpoint.name}\t{endpoint.url}\t{auth}") + + +@remote.command("remove") +@click.argument("name") +def remote_remove(name: str) -> None: + """Remove a named remote endpoint.""" + from coding_agent.remote.client import remove_remote + + remove_remote(name) + click.echo(f"Removed remote {name}") + + +@remote.command("repl") +@click.argument("name") +@click.option( + "--repo", + default=None, + help="Reserved for future workspace upload; currently unsupported.", +) +@click.option( + "--empty-workspace", + is_flag=True, + help="Create an empty server-side Docker workspace.", +) +@click.option("--goal", required=True, help="Initial prompt to send to the remote session") +def remote_repl(name: str, repo: str | None, empty_workspace: bool, goal: str) -> None: + """Create a remote cloud session and stream one prompt.""" + from coding_agent.remote.client import ( + auth_headers, + create_remote_session, + get_remote, + stream_prompt, + ) + + if repo is not None: + raise click.ClickException( + "--repo is not supported for remote Docker workspaces yet; " + "workspace upload/download is planned for the next slice. " + "Use --empty-workspace to create an empty server-side workspace." + ) + if not empty_workspace: + raise click.ClickException( + "Remote repl currently creates an empty server-side Docker workspace. " + "Pass --empty-workspace to acknowledge that no local files are uploaded." + ) + + endpoint = get_remote(name) + session_id = create_remote_session(endpoint) + click.echo(f"Created remote session {session_id}") + raise SystemExit( + stream_prompt( + base_url=endpoint.url, + session_id=session_id, + prompt=goal, + headers=auth_headers(endpoint), + ) + ) + + +@main.command() +@click.argument("name") +@click.option("--session", "session_id", required=True, help="Remote session ID") +@click.option("--goal", required=True, help="Prompt to send to the remote session") +def attach(name: str, session_id: str, goal: str) -> None: + """Attach to an existing remote session and stream one prompt.""" + from coding_agent.remote.client import auth_headers, get_remote, stream_prompt + + endpoint = get_remote(name) + raise SystemExit( + stream_prompt( + base_url=endpoint.url, + session_id=session_id, + prompt=goal, + headers=auth_headers(endpoint), + ) + ) + + @main.command() @click.option( "--task-packet", diff --git a/coding-agent/src/coding_agent/remote/__init__.py b/coding-agent/src/coding_agent/remote/__init__.py new file mode 100644 index 000000000..7ec26f570 --- /dev/null +++ b/coding-agent/src/coding_agent/remote/__init__.py @@ -0,0 +1 @@ +"""Remote session client support.""" diff --git a/coding-agent/src/coding_agent/remote/client.py b/coding-agent/src/coding_agent/remote/client.py new file mode 100644 index 000000000..2fafacbb5 --- /dev/null +++ b/coding-agent/src/coding_agent/remote/client.py @@ -0,0 +1,366 @@ +"""Client helpers for remote Coding Agent HTTP sessions.""" + +from __future__ import annotations + +import json +import os +import tempfile +from dataclasses import dataclass +from collections.abc import Mapping +from pathlib import Path +from typing import cast + +import click +import httpx +from httpx_sse import connect_sse + + +@dataclass(frozen=True) +class RemoteEndpoint: + name: str + url: str + token: str | None = None + + +def remotes_file_path() -> Path: + override = os.environ.get("CODING_AGENT_REMOTES_FILE") + if override: + return Path(override).expanduser() + return Path.home() / ".config" / "coding-agent" / "remotes.json" + + +def load_remotes() -> dict[str, RemoteEndpoint]: + path = remotes_file_path() + if not path.exists(): + return {} + try: + raw = cast(object, json.loads(path.read_text(encoding="utf-8"))) + except json.JSONDecodeError as exc: + raise click.ClickException(f"Invalid remotes file: {path}: {exc}") from exc + if not isinstance(raw, Mapping): + raise click.ClickException(f"Invalid remotes file: {path}") + raw = cast(Mapping[str, object], raw) + remotes = raw.get("remotes") + if not isinstance(remotes, Mapping): + raise click.ClickException(f"Invalid remotes file: {path}") + remotes = cast(Mapping[str, object], remotes) + loaded: dict[str, RemoteEndpoint] = {} + for name, value in remotes.items(): + if not isinstance(value, Mapping): + raise click.ClickException(f"Invalid remote entry in {path}") + value = cast(Mapping[str, object], value) + url = value.get("url") + token = value.get("token") + if not isinstance(url, str) or not url.strip(): + raise click.ClickException(f"Remote {name} is missing url") + if token is not None and not isinstance(token, str): + raise click.ClickException(f"Remote {name} has invalid token") + loaded[name] = RemoteEndpoint(name=name, url=url, token=token) + return loaded + + +def save_remotes(remotes: dict[str, RemoteEndpoint]) -> None: + path = remotes_file_path() + parent_existed = path.parent.exists() + _ = path.parent.mkdir(parents=True, exist_ok=True) + if not parent_existed: + _ = path.parent.chmod(0o700) + payload = { + "remotes": { + name: _remote_payload(endpoint) for name, endpoint in sorted(remotes.items()) + } + } + contents = json.dumps(payload, indent=2, sort_keys=True) + "\n" + fd, temp_name = tempfile.mkstemp( + prefix=f".{path.name}.", suffix=".tmp", dir=path.parent, text=True + ) + temp_path = Path(temp_name) + try: + with os.fdopen(fd, "w", encoding="utf-8") as temp_file: + _ = temp_file.write(contents) + _ = temp_path.chmod(0o600) + _ = temp_path.replace(path) + _ = path.chmod(0o600) + finally: + if temp_path.exists(): + temp_path.unlink() + + +def add_remote(name: str, url: str, token: str | None) -> RemoteEndpoint: + normalized_url = url.rstrip("/") + if not normalized_url: + raise click.ClickException("Remote URL must not be empty") + remotes = load_remotes() + endpoint = RemoteEndpoint(name=name, url=normalized_url, token=token) + remotes[name] = endpoint + save_remotes(remotes) + return endpoint + + +def remove_remote(name: str) -> None: + remotes = load_remotes() + if name not in remotes: + raise click.ClickException(f"Remote not found: {name}") + del remotes[name] + save_remotes(remotes) + + +def get_remote(name: str) -> RemoteEndpoint: + remotes = load_remotes() + try: + return remotes[name] + except KeyError as exc: + raise click.ClickException(f"Remote not found: {name}") from exc + + +def auth_headers(endpoint: RemoteEndpoint) -> dict[str, str]: + if endpoint.token is None: + return {} + return {"Authorization": f"Bearer {endpoint.token}"} + + +def create_remote_session(endpoint: RemoteEndpoint) -> str: + payload: dict[str, object] = { + "workspace_source": {"kind": "docker"}, + "approval_policy": "auto", + } + try: + with httpx.Client( + base_url=endpoint.url, + headers=auth_headers(endpoint), + timeout=60.0, + ) as client: + response = client.post("/sessions", json=payload) + _raise_remote_http_error(response, "create remote session") + data = cast(object, response.json()) + except httpx.RequestError as exc: + raise click.ClickException( + f"Failed to create remote session: {exc}" + ) from exc + if not isinstance(data, Mapping): + raise click.ClickException("Remote session response must be a JSON object") + data = cast(Mapping[str, object], data) + session_id = data.get("session_id") + if not isinstance(session_id, str) or not session_id: + raise click.ClickException("Remote session response missing session_id") + return session_id + + +def stream_prompt( + *, base_url: str, session_id: str, prompt: str, headers: dict[str, str] +) -> int: + timeout = httpx.Timeout(connect=10.0, write=30.0, pool=30.0, read=None) + line_open = False + try: + with httpx.Client(base_url=base_url, headers=headers, timeout=timeout) as client: + with connect_sse( + client, + "POST", + f"/sessions/{session_id}/prompt", + json={"prompt": prompt}, + ) as event_source: + _raise_remote_http_error( + event_source.response, "stream remote prompt" + ) + for sse in event_source.iter_sse(): + status, line_open = handle_sse_event( + base_url=base_url, + session_id=session_id, + headers=headers, + event=sse.event, + data=sse.data, + line_open=line_open, + ) + if status is not None: + if line_open: + click.echo() + return status + except httpx.RequestError as exc: + if line_open: + click.echo() + raise click.ClickException(f"Failed to stream remote prompt: {exc}") from exc + if line_open: + click.echo() + raise click.ClickException("Remote prompt stream ended without TurnEnd") + + +def handle_sse_event( + *, + base_url: str, + session_id: str, + headers: dict[str, str], + event: str, + data: str, + line_open: bool = False, +) -> tuple[int | None, bool]: + try: + payload = _parse_sse_payload(data) + except click.ClickException: + if line_open: + click.echo() + raise + if event == "StreamDelta": + content = payload.get("content") + if isinstance(content, str) and content: + click.echo(content, nl=False) + line_open = True + return None, line_open + if event == "ThinkingDelta": + text = payload.get("text") + if isinstance(text, str) and text: + click.echo(text, nl=False) + line_open = True + return None, line_open + if event == "ToolCallDelta": + line_open = _end_inline_stream_line(line_open) + tool_name = payload.get("tool_name") + if isinstance(tool_name, str): + click.echo(f"[tool] {tool_name}") + return None, line_open + if event == "ToolResultDelta": + line_open = _end_inline_stream_line(line_open) + display_result = payload.get("display_result") + if isinstance(display_result, str) and display_result: + click.echo(display_result) + return None, line_open + if event == "ApprovalRequest": + line_open = _end_inline_stream_line(line_open) + request_id = payload.get("request_id") + if isinstance(request_id, str): + decision = _prompt_for_approval(payload) + _submit_approval( + base_url=base_url, + session_id=session_id, + request_id=request_id, + approved=decision.approved, + feedback=decision.feedback, + scope=decision.scope, + headers=headers, + ) + return None, line_open + if event == "Error": + line_open = _end_inline_stream_line(line_open) + error = payload.get("error") + raise click.ClickException(str(error) if error is not None else "Remote error") + if event == "TurnEnd": + status = payload.get("completion_status") + return (0 if status == "completed" else 1), line_open + return None, line_open + + +@dataclass(frozen=True) +class _ApprovalDecision: + approved: bool + feedback: str | None = None + scope: str = "once" + + +def _prompt_for_approval(payload: dict[str, object]) -> _ApprovalDecision: + tool_call = payload.get("tool_call") + tool_name = "unknown" + if isinstance(tool_call, Mapping): + tool_call = cast(Mapping[str, object], tool_call) + raw_tool_name = tool_call.get("tool_name") + if isinstance(raw_tool_name, str) and raw_tool_name: + tool_name = raw_tool_name + click.echo(f"[approval] Remote tool request {tool_name}") + arguments = tool_call.get("arguments") + if isinstance(arguments, Mapping): + click.echo( + json.dumps( + dict(cast(Mapping[str, object], arguments)), + indent=2, + sort_keys=True, + ) + ) + # TODO(adr-0019): Extend the remote approval prompt to capture session-scope + # approvals and user feedback instead of the current yes/no confirmation. + approved = click.confirm( + f"Approve remote tool request {tool_name}?", + default=False, + ) + return _ApprovalDecision(approved=approved) + + +def _end_inline_stream_line(line_open: bool) -> bool: + if line_open: + click.echo() + return False + + +def _submit_approval( + *, + base_url: str, + session_id: str, + request_id: str, + approved: bool, + feedback: str | None, + scope: str, + headers: dict[str, str], +) -> None: + payload: dict[str, object] = { + "request_id": request_id, + "approved": approved, + "scope": scope, + } + if feedback is not None: + payload["feedback"] = feedback + try: + with httpx.Client(base_url=base_url, headers=headers, timeout=30.0) as client: + response = client.post( + f"/sessions/{session_id}/approve", + json=payload, + ) + _raise_remote_http_error(response, "approve remote tool request") + except httpx.RequestError as exc: + raise click.ClickException( + f"Failed to approve remote tool request: {exc}" + ) from exc + + +def _parse_sse_payload(data: str) -> dict[str, object]: + if not data: + return {} + try: + payload = cast(object, json.loads(data)) + except json.JSONDecodeError as exc: + raise click.ClickException( + f"Remote SSE event payload must be valid JSON: {exc}" + ) from exc + if not isinstance(payload, Mapping): + raise click.ClickException("Remote SSE event payload must be a JSON object") + return dict(cast(Mapping[str, object], payload)) + + +def _raise_remote_http_error(response: httpx.Response, action: str) -> None: + try: + _ = response.raise_for_status() + except httpx.HTTPStatusError as exc: + detail = _response_error_detail(response) + raise click.ClickException(f"Failed to {action}: {detail}") from exc + + +def _response_error_detail(response: httpx.Response) -> str: + if not hasattr(response, "json"): + status_code = getattr(response, "status_code", None) + return f"HTTP {status_code}" if isinstance(status_code, int) else "HTTP error" + try: + payload = cast(object, response.json()) + except ValueError: + text = response.text.strip() + if text: + return text + return f"HTTP {response.status_code}" + if isinstance(payload, Mapping): + payload = cast(Mapping[str, object], payload) + detail = payload.get("detail") + if isinstance(detail, str) and detail: + return detail + return f"HTTP {response.status_code}" + + +def _remote_payload(endpoint: RemoteEndpoint) -> dict[str, str]: + payload = {"url": endpoint.url} + if endpoint.token is not None: + payload["token"] = endpoint.token + return payload diff --git a/coding-agent/tests/cli/test_remote_client.py b/coding-agent/tests/cli/test_remote_client.py new file mode 100644 index 000000000..75e6cf0de --- /dev/null +++ b/coding-agent/tests/cli/test_remote_client.py @@ -0,0 +1,664 @@ +from __future__ import annotations + +import json +import stat +from pathlib import Path + +import httpx +import click +import pytest +from click.testing import CliRunner + +from coding_agent.__main__ import main + + +def test_remote_add_list_remove_manage_named_endpoint( + tmp_path: Path, monkeypatch +) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + + add = runner.invoke( + main, + [ + "remote", + "add", + "dev", + "http://127.0.0.1:8080/", + "--token", + "secret-token", + ], + catch_exceptions=False, + ) + + assert add.exit_code == 0 + assert "Added remote dev" in add.output + saved = json.loads(config_path.read_text(encoding="utf-8")) + assert saved == { + "remotes": { + "dev": { + "url": "http://127.0.0.1:8080", + "token": "secret-token", + } + } + } + + listed = runner.invoke(main, ["remote", "list"], catch_exceptions=False) + + assert listed.exit_code == 0 + assert "dev" in listed.output + assert "http://127.0.0.1:8080" in listed.output + assert "secret-token" not in listed.output + + removed = runner.invoke(main, ["remote", "remove", "dev"], catch_exceptions=False) + + assert removed.exit_code == 0 + assert "Removed remote dev" in removed.output + assert json.loads(config_path.read_text(encoding="utf-8")) == {"remotes": {}} + + +def test_remote_repl_creates_cloud_session_and_streams_prompt_events( + tmp_path: Path, monkeypatch +) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example", "--token", "secret-token"], + catch_exceptions=False, + ) + calls: list[tuple[str, str, dict[str, object] | None, dict[str, str]]] = [] + + class FakeResponse: + def __init__(self, payload: dict[str, object]) -> None: + self._payload = payload + + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, object]: + return self._payload + + class FakeClient: + def __init__( + self, + *, + base_url: str, + headers: dict[str, str] | None = None, + timeout: float, + ) -> None: + self.base_url = base_url + self.headers = headers or {} + self.timeout = timeout + + def __enter__(self) -> FakeClient: + return self + + def __exit__(self, exc_type, exc, tb) -> None: + return None + + def post( + self, path: str, json: dict[str, object] | None = None + ) -> FakeResponse: + calls.append(("post", path, json, self.headers)) + if path == "/sessions": + return FakeResponse({"session_id": "sess-123"}) + raise AssertionError(f"unexpected post {path}") + + def fake_stream_prompt( + *, base_url: str, session_id: str, prompt: str, headers: dict[str, str] + ) -> int: + calls.append( + ( + "stream", + f"/sessions/{session_id}/prompt", + {"prompt": prompt, "base_url": base_url}, + headers, + ) + ) + return 0 + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FakeClient) + monkeypatch.setattr("coding_agent.remote.client.stream_prompt", fake_stream_prompt) + + result = runner.invoke( + main, + ["remote", "repl", "dev", "--empty-workspace", "--goal", "hello"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert "Created remote session sess-123" in result.output + assert calls == [ + ( + "post", + "/sessions", + {"workspace_source": {"kind": "docker"}, "approval_policy": "auto"}, + {"Authorization": "Bearer secret-token"}, + ), + ( + "stream", + "/sessions/sess-123/prompt", + {"prompt": "hello", "base_url": "http://agent.example"}, + {"Authorization": "Bearer secret-token"}, + ), + ] + + +def test_attach_streams_prompt_to_existing_session( + tmp_path: Path, monkeypatch +) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example"], + catch_exceptions=False, + ) + calls: list[dict[str, object]] = [] + + def fake_stream_prompt( + *, base_url: str, session_id: str, prompt: str, headers: dict[str, str] + ) -> int: + calls.append( + { + "base_url": base_url, + "session_id": session_id, + "prompt": prompt, + "headers": headers, + } + ) + return 0 + + monkeypatch.setattr("coding_agent.remote.client.stream_prompt", fake_stream_prompt) + + result = runner.invoke( + main, + ["attach", "dev", "--session", "sess-456", "--goal", "continue"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert calls == [ + { + "base_url": "http://agent.example", + "session_id": "sess-456", + "prompt": "continue", + "headers": {}, + } + ] + + +def test_remote_repl_reports_http_session_create_error( + tmp_path: Path, monkeypatch +) -> None: + import httpx + + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example"], + catch_exceptions=False, + ) + + class FailingClient: + def __init__(self, **kwargs: object) -> None: + del kwargs + + def __enter__(self) -> FailingClient: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + def post( + self, path: str, json: dict[str, object] | None = None + ) -> httpx.Response: + del path, json + request = httpx.Request("POST", "http://agent.example/sessions") + return httpx.Response(400, json={"detail": "cloud workspace disabled"}, request=request) + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FailingClient) + + result = runner.invoke( + main, + ["remote", "repl", "dev", "--empty-workspace", "--goal", "hello"], + ) + + assert result.exit_code != 0 + assert "Failed to create remote session" in result.output + assert "cloud workspace disabled" in result.output + + +def test_remote_repl_does_not_accept_repo_until_workspace_upload_exists( + tmp_path: Path, monkeypatch +) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example"], + catch_exceptions=False, + ) + + result = runner.invoke( + main, + ["remote", "repl", "dev", "--repo", str(tmp_path), "--goal", "hello"], + ) + + assert result.exit_code != 0 + assert "--repo is not supported for remote Docker workspaces yet" in result.output + + +def test_remote_repl_can_explicitly_create_empty_docker_workspace( + tmp_path: Path, monkeypatch +) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example"], + catch_exceptions=False, + ) + calls: list[tuple[str, str, dict[str, object] | None, dict[str, str]]] = [] + + class FakeResponse: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, object]: + return {"session_id": "sess-empty"} + + class FakeClient: + def __init__(self, **kwargs: object) -> None: + headers = kwargs.get("headers") + self.headers = headers if isinstance(headers, dict) else {} + + def __enter__(self) -> FakeClient: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + def post( + self, path: str, json: dict[str, object] | None = None + ) -> FakeResponse: + calls.append(("post", path, json, self.headers)) + return FakeResponse() + + def fake_stream_prompt( + *, base_url: str, session_id: str, prompt: str, headers: dict[str, str] + ) -> int: + calls.append( + ( + "stream", + f"/sessions/{session_id}/prompt", + {"prompt": prompt, "base_url": base_url}, + headers, + ) + ) + return 0 + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FakeClient) + monkeypatch.setattr("coding_agent.remote.client.stream_prompt", fake_stream_prompt) + + result = runner.invoke( + main, + ["remote", "repl", "dev", "--empty-workspace", "--goal", "hello"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert calls == [ + ( + "post", + "/sessions", + {"workspace_source": {"kind": "docker"}, "approval_policy": "auto"}, + {}, + ), + ( + "stream", + "/sessions/sess-empty/prompt", + {"prompt": "hello", "base_url": "http://agent.example"}, + {}, + ), + ] + + +def test_remote_approval_request_prompts_before_submitting_decision(monkeypatch) -> None: + approvals: list[dict[str, object] | None] = [] + prompts: list[tuple[str, bool]] = [] + + class FakeResponse: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, object]: + return {"status": "ok"} + + class FakeClient: + def __init__(self, **kwargs: object) -> None: + del kwargs + + def __enter__(self) -> FakeClient: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + def post( + self, path: str, json: dict[str, object] | None = None + ) -> FakeResponse: + del path + approvals.append(json) + return FakeResponse() + + def fake_confirm(text: str, default: bool) -> bool: + prompts.append((text, default)) + return False + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FakeClient) + monkeypatch.setattr("coding_agent.remote.client.click.confirm", fake_confirm) + + from coding_agent.remote.client import handle_sse_event + + status, line_open = handle_sse_event( + base_url="http://agent.example", + session_id="sess-approval", + headers={}, + event="ApprovalRequest", + data=json.dumps( + { + "request_id": "req-1", + "tool_call": { + "tool_name": "bash_run", + "arguments": {"command": "rm -rf scratch"}, + "call_id": "call-1", + }, + } + ), + ) + + assert status is None + assert line_open is False + assert prompts == [("Approve remote tool request bash_run?", False)] + assert approvals == [ + {"request_id": "req-1", "approved": False, "scope": "once"} + ] + + +def test_handle_sse_event_formats_approval_after_inline_stream_output( + monkeypatch, capsys +) -> None: + approvals: list[dict[str, object] | None] = [] + + class FakeResponse: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, object]: + return {"status": "ok"} + + class FakeClient: + def __init__(self, **kwargs: object) -> None: + del kwargs + + def __enter__(self) -> FakeClient: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + def post( + self, path: str, json: dict[str, object] | None = None + ) -> FakeResponse: + del path + approvals.append(json) + return FakeResponse() + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FakeClient) + monkeypatch.setattr( + "coding_agent.remote.client.click.confirm", lambda text, default: False + ) + + from coding_agent.remote.client import handle_sse_event + + status, line_open = handle_sse_event( + base_url="http://agent.example", + session_id="sess-approval", + headers={}, + event="StreamDelta", + data=json.dumps({"content": "partial"}), + ) + + assert status is None + assert line_open is True + + status, line_open = handle_sse_event( + base_url="http://agent.example", + session_id="sess-approval", + headers={}, + event="ApprovalRequest", + data=json.dumps( + { + "request_id": "req-1", + "tool_call": { + "tool_name": "bash_run", + "arguments": {"command": "rm -rf scratch"}, + "call_id": "call-1", + }, + } + ), + line_open=line_open, + ) + + assert status is None + assert line_open is False + assert approvals == [ + {"request_id": "req-1", "approved": False, "scope": "once"} + ] + assert capsys.readouterr().out == ( + "partial\n" + "[approval] Remote tool request bash_run\n" + '{\n "command": "rm -rf scratch"\n}\n' + ) + + +def test_remote_config_file_is_written_private(tmp_path: Path, monkeypatch) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + + result = runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example", "--token", "secret-token"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert stat.S_IMODE(config_path.stat().st_mode) == 0o600 + + +def test_remote_add_reports_invalid_remotes_json(tmp_path: Path, monkeypatch) -> None: + config_path = tmp_path / "remotes.json" + _ = config_path.write_text("{not-json", encoding="utf-8") + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + + result = runner.invoke(main, ["remote", "list"]) + + assert result.exit_code != 0 + assert f"Invalid remotes file: {config_path}" in result.output + + +def test_save_remotes_does_not_chmod_existing_override_parent( + tmp_path: Path, monkeypatch +) -> None: + config_path = tmp_path / "nested" / "remotes.json" + config_path.parent.mkdir(parents=True) + original_mode = stat.S_IMODE(config_path.parent.stat().st_mode) + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + + result = runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example", "--token", "secret-token"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert stat.S_IMODE(config_path.parent.stat().st_mode) == original_mode + + +def test_remote_repl_reports_request_error_on_session_create( + tmp_path: Path, monkeypatch +) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + runner = CliRunner() + runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example"], + catch_exceptions=False, + ) + + class FailingClient: + def __init__(self, **kwargs: object) -> None: + del kwargs + + def __enter__(self) -> FailingClient: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + def post( + self, path: str, json: dict[str, object] | None = None + ) -> httpx.Response: + del path, json + request = httpx.Request("POST", "http://agent.example/sessions") + raise httpx.ConnectError("connection refused", request=request) + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FailingClient) + + result = runner.invoke( + main, + ["remote", "repl", "dev", "--empty-workspace", "--goal", "hello"], + ) + + assert result.exit_code != 0 + assert "Failed to create remote session" in result.output + assert "connection refused" in result.output + + +def test_stream_prompt_reports_non_200_sse_response(monkeypatch) -> None: + class FakeResponse: + status_code = 503 + + def raise_for_status(self) -> None: + request = httpx.Request("POST", "http://agent.example/sessions/sess-1/prompt") + raise httpx.HTTPStatusError( + "bad status", + request=request, + response=httpx.Response(503, request=request, json={"detail": "server busy"}), + ) + + def json(self) -> dict[str, object]: + return {"detail": "server busy"} + + class FakeEventSource: + response = FakeResponse() + + def __enter__(self) -> FakeEventSource: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + def iter_sse(self): + if False: + yield None + + class FakeClient: + def __init__(self, **kwargs: object) -> None: + self.timeout = kwargs.get("timeout") + + def __enter__(self) -> FakeClient: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FakeClient) + monkeypatch.setattr("coding_agent.remote.client.connect_sse", lambda *args, **kwargs: FakeEventSource()) + + from coding_agent.remote.client import stream_prompt + + with pytest.raises(click.ClickException, match="Failed to stream remote prompt: server busy"): + stream_prompt( + base_url="http://agent.example", + session_id="sess-1", + prompt="hello", + headers={}, + ) + + +def test_stream_prompt_rejects_truncated_stream_without_turn_end(monkeypatch) -> None: + class FakeResponse: + def raise_for_status(self) -> None: + return None + + class FakeEventSource: + response = FakeResponse() + + def __enter__(self) -> FakeEventSource: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + def iter_sse(self): + yield type("SSE", (), {"event": "StreamDelta", "data": json.dumps({"content": "partial"})})() + + class FakeClient: + def __init__(self, **kwargs: object) -> None: + self.timeout = kwargs.get("timeout") + + def __enter__(self) -> FakeClient: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + monkeypatch.setattr("coding_agent.remote.client.httpx.Client", FakeClient) + monkeypatch.setattr("coding_agent.remote.client.connect_sse", lambda *args, **kwargs: FakeEventSource()) + + from coding_agent.remote.client import stream_prompt + + with pytest.raises(click.ClickException, match="Remote prompt stream ended without TurnEnd"): + stream_prompt( + base_url="http://agent.example", + session_id="sess-1", + prompt="hello", + headers={}, + ) + + +def test_handle_sse_event_reports_invalid_json(monkeypatch) -> None: + from coding_agent.remote.client import handle_sse_event + + with pytest.raises(click.ClickException, match="Remote SSE event payload must be valid JSON"): + handle_sse_event( + base_url="http://agent.example", + session_id="sess-1", + headers={}, + event="StreamDelta", + data="{not-json", + )