refactor(chat): share startup options with trusted launchers

docs/reference/protocols/host-integration-surface-v0.md

@@ -354,6 +354,44 @@ commands and preserve the same state transitions. If a host offers an operation
 without a CLI fallback, that operation is experimental and must not be used as
 the default project control path.
 
+## Native Chat startup from a trusted launcher
+
+The native CLI and locally reviewed Chat launchers share
+`loopx.cli_commands.support_control_chat.chat_server_options`. Use the parser
+and builder from the same installed LoopX release instead of copying option
+conversion into a launcher:
+
+```python
+from pathlib import Path
+import sys
+
+from loopx.cli import build_parser
+from loopx.cli_commands.support_control_chat import chat_server_options
+from loopx.chat_server import serve_chat
+
+args = build_parser().parse_args(sys.argv[1:])
+options = chat_server_options(args, registry_path=Path(args.registry).expanduser())
+serve_chat(**options)
+```
+
+Invoke the launcher with the ordinary `chat` arguments, including `--no-open`
+when it runs under a supervisor. A trusted launcher can pass its locally
+reviewed `external_conversation_factories` tuple separately to `serve_chat`;
+HTTP requests and IM messages cannot select factory code.
+
+The builder accepts only the `chat` command and has no server-start, replacement,
+listener or grant effects. The CLI retains ownership of its explicit
+replacement operation. It preserves workspace grants, filesystem restrictions,
+reaction settings, registry selection and scan paths. An omitted startup timeout
+remains `None` for the native adapter defaults; an explicit timeout retains its
+existing lower clamp and upper-bound meaning.
+
+After an upgrade, check the supervised process and `/healthz`, then verify the
+configured transport's listener, original Session and actual message return.
+HTTP health alone does not establish message delivery. This startup helper
+does not admit missed messages or replay existing Turns; outage catch-up retains
+the original source, audience and idempotency requirements of its owning transport.
+
 ## Public/Private Boundary
 
 Host integrations must preserve these invariants:

loopx/cli_commands/support_control.py

@@ -46,7 +46,7 @@
     handle_backup_state_command,
     register_backup_state_command,
 )
-from .support_control_chat import register_chat_and_dashboard_commands
+from .support_control_chat import chat_server_options, register_chat_and_dashboard_commands
 from .support_control_chat_endpoint import (
     handle_chat_endpoint_command,
 )
@@ -618,47 +618,10 @@ def handle_support_control_command(
 
     if args.command == "chat":
         try:
-            chat_registry_path = (
-                explicit_global_registry(args.runtime_root, registry_path=registry_path)
-                if args.global_registry
-                else registry_path
-            )
-            scan_roots = [Path(item).expanduser() for item in args.scan_path]
-            if not scan_roots:
-                scan_roots = [Path(args.scan_root).expanduser()]
+            options = chat_server_options(args, registry_path=registry_path)
             if bool(getattr(args, "replace_existing_loopx_chat", False)):
                 replace_existing_loopx_chat(args.host, args.port)
-            serve_chat(
-                project_workspace_grant=args.project_workspace_grant,
-                project_filesystem_scope=args.project_filesystem_scope,
-                private_reactions=not getattr(args, "no_private_reactions", False),
-                registry_path=chat_registry_path,
-                runtime_root_override=args.runtime_root,
-                scan_roots=scan_roots,
-                limit=max(0, args.limit),
-                host=args.host,
-                port=args.port,
-                goal_id=args.goal_id,
-                codex_bin=args.codex_bin,
-                claude_bin=args.claude_bin,
-                kiro_cli_bin=getattr(args, "kiro_cli_bin", KIRO_CLI_BIN),
-                lark_cli_bin=args.lark_cli_bin,
-                startup_timeout_sec=(
-                    max(0.1, float(args.startup_timeout_seconds))
-                    if args.startup_timeout_seconds is not None
-                    else None
-                ),
-                idle_timeout_sec=max(0.1, float(args.idle_timeout_seconds)),
-                hard_timeout_sec=max(0.1, float(args.hard_timeout_seconds)),
-                assets_dir=Path(args.assets_dir).expanduser()
-                if args.assets_dir
-                else None,
-                open_browser=not bool(args.no_open),
-                verbose=bool(args.verbose),
-                enable_goal_subagent_configuration=bool(
-                    args.enable_goal_subagent_configuration
-                ),
-            )
+            serve_chat(**options)
         except Exception as exc:
             payload = {
                 "ok": False,

loopx/cli_commands/support_control_chat.py

@@ -1,20 +1,66 @@
-"""Owner-local parser registration for the chat and dashboard commands.
+"""Native Chat argument registration and shared startup option composition.
 
-Both commands launch the same local presentation surface with different
-defaults, so they share one cohesive registration unit. Keeping them beside
-the other support commands pushed that module over its size budget.
+Dashboard registration shares the local presentation defaults. The runtime
+retains adapter timeout policy and server/transport lifecycle ownership.
 """
 
 from __future__ import annotations
 
 import argparse
 from collections.abc import Callable
+from pathlib import Path
+from typing import Any
 
 from ..chat_server import DEFAULT_CHAT_HOST, DEFAULT_CHAT_PORT
 from ..kiro_cli_goal_mode import KIRO_CLI_BIN
 from ..paths import default_public_scan_root
 from .support_control_agent_runtime import register_agent_runtime_arguments
 from .support_control_chat_endpoint import register_chat_endpoint_command
+from .support_control_registry import explicit_global_registry
+
+
+def chat_server_options(
+    args: argparse.Namespace, *, registry_path: Path
+) -> dict[str, Any]:
+    """Compose native Chat options for the CLI and trusted local launchers.
+
+    Keep parser-to-server conversion here. In particular, an omitted startup
+    timeout must reach the runtime as None so each adapter owns its default.
+    This function does not start, replace, or authorize a server or transport.
+    """
+    if args.command != "chat":
+        raise ValueError("native Chat options require the chat command")
+    chat_registry_path = (
+        explicit_global_registry(args.runtime_root, registry_path=registry_path)
+        if args.global_registry else registry_path
+    )
+    scan_roots = [Path(item).expanduser() for item in args.scan_path]
+    return {
+        "project_workspace_grant": args.project_workspace_grant,
+        "project_filesystem_scope": args.project_filesystem_scope,
+        "private_reactions": not getattr(args, "no_private_reactions", False),
+        "registry_path": chat_registry_path,
+        "runtime_root_override": args.runtime_root,
+        "scan_roots": scan_roots or [Path(args.scan_root).expanduser()],
+        "limit": max(0, args.limit),
+        "host": args.host,
+        "port": args.port,
+        "goal_id": args.goal_id,
+        "codex_bin": args.codex_bin,
+        "claude_bin": args.claude_bin,
+        "kiro_cli_bin": getattr(args, "kiro_cli_bin", KIRO_CLI_BIN),
+        "lark_cli_bin": args.lark_cli_bin,
+        "startup_timeout_sec": (
+            max(0.1, float(args.startup_timeout_seconds))
+            if args.startup_timeout_seconds is not None else None
+        ),
+        "idle_timeout_sec": max(0.1, float(args.idle_timeout_seconds)),
+        "hard_timeout_sec": max(0.1, float(args.hard_timeout_seconds)),
+        "assets_dir": Path(args.assets_dir).expanduser() if args.assets_dir else None,
+        "open_browser": not bool(args.no_open),
+        "verbose": bool(args.verbose),
+        "enable_goal_subagent_configuration": bool(args.enable_goal_subagent_configuration),
+    }
 
 
 def register_chat_and_dashboard_commands(
@@ -170,4 +216,4 @@ def register_chat_and_dashboard_commands(
     )
 
 
-__all__ = ["register_chat_and_dashboard_commands"]
+__all__ = ["chat_server_options", "register_chat_and_dashboard_commands"]

tests/test_chat_startup_options.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from loopx.chat_server import serve_chat
+from loopx.cli import build_parser
+from loopx.cli_commands import support_control
+from loopx.cli_commands.support_control_chat import chat_server_options
+
+
+@pytest.mark.parametrize("configured, expected", [(None, None), ("0", 0.1), ("12.5", 12.5), ("45", 45.0)])
+def test_shared_startup_options_preserve_adapter_defaults_and_explicit_limits(tmp_path, configured, expected):
+    tail = [] if configured is None else ["--startup-timeout-seconds", configured]
+    args = build_parser().parse_args(["chat", "--no-open", *tail])
+    options = chat_server_options(args, registry_path=tmp_path / "registry.json")
+    assert options["startup_timeout_sec"] == expected
+    assert options["idle_timeout_sec"] == 180.0
+    assert options["hard_timeout_sec"] == 900.0
+    inspect.signature(serve_chat).bind(**options)
+
+
+def test_shared_options_keep_workspace_permissions_and_explicit_timeout_floors(tmp_path):
+    args = build_parser().parse_args([
+        "chat", "--project-workspace-grant", "workspace_read",
+        "--project-filesystem-scope", "workspace_only", "--no-private-reactions",
+        "--idle-timeout-seconds", "0", "--hard-timeout-seconds", "0", "--limit", "-1",
+    ])
+    options = chat_server_options(args, registry_path=tmp_path / "registry.json")
+    assert options["project_workspace_grant"] == "workspace_read"
+    assert options["project_filesystem_scope"] == "workspace_only"
+    assert options["private_reactions"] is False
+    assert options["idle_timeout_sec"] == options["hard_timeout_sec"] == 0.1
+    assert options["limit"] == 0
+
+
+def test_shared_options_resolve_global_registry_and_expand_selected_paths(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    runtime = tmp_path / "runtime"
+    args = build_parser().parse_args([
+        "--runtime-root", str(runtime), "chat", "--global-registry",
+        "--scan-path", "~/first", "--scan-path", "~/second", "--assets-dir", "~/bundle",
+    ])
+    options = chat_server_options(args, registry_path=tmp_path / "absent.json")
+    assert options["registry_path"] == runtime / "registry.global.json"
+    assert options["scan_roots"] == [tmp_path / "first", tmp_path / "second"]
+    assert options["assets_dir"] == tmp_path / "bundle"
+    assert not runtime.exists()
+
+
+def test_shared_options_keep_scan_root_fallback_and_reject_another_command(tmp_path):
+    args = build_parser().parse_args(["chat", "--scan-root", str(tmp_path)])
+    assert chat_server_options(args, registry_path=tmp_path / "registry.json")["scan_roots"] == [tmp_path]
+    args = build_parser().parse_args(["dashboard"])
+    with pytest.raises(ValueError, match="require the chat command"):
+        chat_server_options(args, registry_path=tmp_path / "registry.json")
+
+
+def test_cli_uses_shared_options_before_replacing_or_starting_a_server(tmp_path, monkeypatch):
+    args = build_parser().parse_args(["chat", "--no-open", "--replace-existing-loopx-chat"])
+    registry = tmp_path / "registry.json"
+    calls = []
+
+    def compose(received, *, registry_path):
+        assert received is args and registry_path == registry
+        calls.append("compose")
+        return chat_server_options(received, registry_path=registry_path)
+
+    monkeypatch.setattr(support_control, "chat_server_options", compose)
+    monkeypatch.setattr(support_control, "replace_existing_loopx_chat", lambda *args: calls.append("replace"))
+    monkeypatch.setattr(support_control, "serve_chat", lambda **kwargs: calls.append(kwargs))
+    result = support_control.handle_support_control_command(args, registry_path=registry,
+        registry_was_supplied=True, print_payload=lambda *args: None, output_format=lambda *args: "json")
+    assert result == 0
+    assert calls[:2] == ["compose", "replace"]
+    assert calls[2]["startup_timeout_sec"] is None
+    assert calls[2]["registry_path"] == registry
+    assert calls[2]["open_browser"] is False
+
+
+def test_failed_option_composition_does_not_replace_a_running_server(tmp_path, monkeypatch):
+    args = build_parser().parse_args(["chat", "--replace-existing-loopx-chat"])
+    calls = []
+    payloads = []
+
+    def reject(*args, **kwargs):
+        raise ValueError("invalid fixture options")
+
+    monkeypatch.setattr(support_control, "chat_server_options", reject)
+    monkeypatch.setattr(support_control, "replace_existing_loopx_chat", lambda *args: calls.append("replace"))
+    monkeypatch.setattr(support_control, "serve_chat", lambda **kwargs: calls.append("serve"))
+    result = support_control.handle_support_control_command(args, registry_path=tmp_path / "registry.json",
+        registry_was_supplied=True, print_payload=lambda payload, *args: payloads.append(payload),
+        output_format=lambda *args: "json")
+    assert result == 1 and calls == []
+    assert payloads[0]["ok"] is False
+    assert payloads[0]["schema_version"] == "loopx_chat_start_v0"

tests/test_chat_transport_composition.py

@@ -171,6 +171,8 @@ def test_return_rechecks_actual_typed_binding_after_owner_disconnect(ordinary):
 @pytest.mark.parametrize("installed", [False, True])
 def test_real_chat_entrypoint_composes_one_store_controller_and_return_service(tmp_path, monkeypatch, installed):
     import loopx.chat_server as chat
+    from loopx.cli import build_parser
+    from loopx.cli_commands.support_control_chat import chat_server_options
     from loopx.extensions.lark.cli_resolution import LarkCliResolution
     home = tmp_path / "home"
     registry = home / ".loopx" / "registry.json"
@@ -202,16 +204,21 @@ def __init__(self, *args, **kwargs):
             captured.append(self)
             entered.set()
     monkeypatch.setattr(chat, "ChatHTTPServer", Server)
-    worker = threading.Thread(target=chat.serve_chat, kwargs={"registry_path": registry, "port": 0,
-        "runtime_root_override": tmp_path / "runtime", "assets_dir": assets,
-        "external_conversation_factories": (lambda server: external,) if installed else (),
-        "codex_bin": "unavailable-fixture"}, daemon=True)
+    args = build_parser().parse_args(["--runtime-root", str(tmp_path / "runtime"),
+        "chat", "--port", "0", "--assets-dir", str(assets), "--no-open",
+        "--codex-bin", "unavailable-fixture"])
+    options = chat_server_options(args, registry_path=registry)
+    assert options["startup_timeout_sec"] is None
+    worker = threading.Thread(target=chat.serve_chat, kwargs={**options,
+        "external_conversation_factories": (lambda server: external,) if installed else ()}, daemon=True)
     worker.start()
     try:
         assert entered.wait(8)
         server = captured[0]
         with urlopen(f"http://127.0.0.1:{server.server_port}/healthz", timeout=8) as response:
             assert json.load(response)["ok"] is True
+        assert server.runtime_controller.startup_timeout_sec == 30.0
+        assert server.runtime_controller.acp_startup_timeout_sec is None
         assert server.manager_return_service.args[2] is server.chat_store
         if installed:
             assert server.manager_return_service.args[3] is server.conversation_transports