fix(start-goal): reject desktop host in SSH sessions

loopx/cli_commands/start_goal.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+import os
 import re
 from collections.abc import Callable
 from pathlib import Path
@@ -31,6 +32,8 @@
     r"(?P<remainder>[\s\S]*)\Z"
 )
 _FINE_GRAINED_PREFIX = re.compile(r"\A--fine-grained(?P<remainder>(?:\s[\s\S]*)?)\Z")
+_CODEX_APP_SSH_HOST_SURFACE = "codex-app-ssh"
+_SSH_TRANSPORT_ENV_VARS = ("SSH_CONNECTION", "SSH_CLIENT")
 
 _EFFECT_RUNTIME_STARTUP_REMEDIATION_BY_CODE = {
     "startup_lock_timeout": (
@@ -104,7 +107,12 @@ def _render_start_goal_markdown(payload: dict[str, object]) -> str:
     if payload.get("ok") is not False:
         return render_start_goal_guided_markdown(payload)
     lines = ["# Guided Start Goal", "", f"- error: `{payload.get('error')}`"]
-    for field in ("diagnostic_code", "recommended_action", "suggested_command"):
+    for field in (
+        "error_code",
+        "diagnostic_code",
+        "recommended_action",
+        "suggested_command",
+    ):
         value = payload.get(field)
         if not value:
             continue
@@ -113,6 +121,47 @@ def _render_start_goal_markdown(payload: dict[str, object]) -> str:
     return "\n".join(lines) + "\n"
 
 
+def _host_surface_environment_mismatch_payload(
+    *,
+    host_surface: str | None,
+    cli_bin: str,
+) -> dict[str, object] | None:
+    if host_surface != "codex-app" or not os.environ.get("CODEX_THREAD_ID"):
+        return None
+    ssh_evidence = [
+        variable for variable in _SSH_TRANSPORT_ENV_VARS if os.environ.get(variable)
+    ]
+    if not ssh_evidence:
+        return None
+    return {
+        "ok": False,
+        "schema_version": "loopx_start_goal_guided_v0",
+        "error_code": "host_surface_environment_mismatch",
+        "error": "Selected Codex App host surface conflicts with the detected SSH runtime",
+        "blocked_by": "host_surface_environment_mismatch",
+        "host_surface_environment_gate": {
+            "schema_version": "loopx_host_surface_environment_gate_v0",
+            "state": "blocked",
+            "requested_host_surface": host_surface,
+            "detected_runtime": "codex-app-over-ssh",
+            "required_host_surface": _CODEX_APP_SSH_HOST_SURFACE,
+            "evidence_env_vars": ["CODEX_THREAD_ID", *ssh_evidence],
+        },
+        "recommended_action": (
+            "Rerun the same command with `--host-surface codex-app-ssh`. "
+            "That profile emits the visible Goal task body through "
+            "`codex_app_ssh_goal` instead of requesting desktop heartbeat automation."
+        ),
+        "suggested_command": (
+            f"{cli_bin} start-goal --guided --host-surface codex-app-ssh ..."
+        ),
+        "safety_contract": {
+            "writes_registry": False,
+            "activates_host_loop": False,
+        },
+    }
+
+
 def add_capability_route_argument(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--capability-route",
@@ -312,6 +361,13 @@ def handle_start_goal_command(
         }
         print_payload(payload, args.format, _render_start_goal_markdown)
         return 2
+    host_surface_mismatch = _host_surface_environment_mismatch_payload(
+        host_surface=args.host_surface,
+        cli_bin=args.cli_bin,
+    )
+    if host_surface_mismatch is not None:
+        print_payload(host_surface_mismatch, args.format, _render_start_goal_markdown)
+        return 2
     display_name = args.display_name
     if not args.host_surface:
         try:

tests/control_plane/test_start_goal_compact_projection.py

@@ -34,6 +34,12 @@
 GOAL_TEXT = "Ship a bounded public issue triage workflow."
 
 
+@pytest.fixture(autouse=True)
+def _isolate_ambient_ssh_transport(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("SSH_CONNECTION", raising=False)
+    monkeypatch.delenv("SSH_CLIENT", raising=False)
+
+
 def _write_connected_project(root: Path) -> Path:
     project = root / "project"
     state_file = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md"
@@ -122,6 +128,38 @@ def _invoke_detail_command(command: str) -> dict[str, Any]:
     return payload
 
 
+def _invoke_start_goal_for_host(
+    project: Path,
+    host_surface: str,
+    *,
+    include_detail: bool = False,
+) -> tuple[int, dict[str, Any]]:
+    args = [
+        "--format",
+        "json",
+        "start-goal",
+        "--guided",
+        "--project",
+        str(project),
+        "--goal-id",
+        GOAL_ID,
+        "--agent-id",
+        AGENT_ID,
+        "--host-surface",
+        host_surface,
+        "--goal-text",
+        GOAL_TEXT,
+    ]
+    if include_detail:
+        args.append("--include-command-pack-detail")
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        exit_code = cli_main(args)
+    payload = json.loads(output.getvalue())
+    assert isinstance(payload, dict)
+    return exit_code, payload
+
+
 def _invoke_ark_start_goal_cli(
     project: Path,
     *goal_input_args: str,
@@ -1505,6 +1543,46 @@ def test_cli_codex_app_ssh_reuses_ambient_thread_binding(
     assert payload["guided_transaction"].get("blocked_by") is None
 
 
+def test_cli_codex_app_fails_closed_when_codex_thread_runs_over_ssh(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    project = _write_connected_project(tmp_path)
+    monkeypatch.setenv("CODEX_THREAD_ID", "thread-ssh-runtime")
+    monkeypatch.setenv("SSH_CONNECTION", "client 1000 server 22")
+    monkeypatch.setenv("SSH_CLIENT", "client 1000 22")
+
+    wrong_exit_code, rejected = _invoke_start_goal_for_host(project, "codex-app")
+    assert wrong_exit_code == 2
+    assert rejected["ok"] is False
+    assert rejected["error_code"] == "host_surface_environment_mismatch"
+    assert rejected["blocked_by"] == "host_surface_environment_mismatch"
+    assert rejected["safety_contract"] == {
+        "writes_registry": False,
+        "activates_host_loop": False,
+    }
+    assert rejected["host_surface_environment_gate"] == {
+        "schema_version": "loopx_host_surface_environment_gate_v0",
+        "state": "blocked",
+        "requested_host_surface": "codex-app",
+        "detected_runtime": "codex-app-over-ssh",
+        "required_host_surface": "codex-app-ssh",
+        "evidence_env_vars": ["CODEX_THREAD_ID", "SSH_CONNECTION", "SSH_CLIENT"],
+    }
+
+    correct_exit_code, accepted = _invoke_start_goal_for_host(
+        project,
+        "codex-app-ssh",
+        include_detail=True,
+    )
+    assert correct_exit_code == 0
+    activation = accepted["command_pack"]["host_loop_activation"]
+    assert activation["activation_method"] == "set_visible_goal"
+    thin_prompt_command = activation["commands"]["heartbeat_prompt"]
+    assert "--thin" in thin_prompt_command
+    assert "--runtime-profile codex_app_ssh_goal" in thin_prompt_command
+    assert "--codex-app" not in thin_prompt_command
+
+
 def test_cli_codex_app_ssh_unbound_ambient_thread_requires_lane_selection(
     tmp_path: Path, monkeypatch
 ) -> None: