feat(channels): install-service — one command to run channels start at login/boot

coding_bridge/channels/service.py

@@ -0,0 +1,158 @@
+"""Render an OS service unit for ``coding-bridge channels start``.
+
+Pure rendering (no filesystem side effects) so it's unit-testable; the CLI
+(:func:`coding_bridge.channels_cli.cmd_channels_install_service`) writes the file
+and prints the activation command. We only ever generate a **user-scoped** unit
+(no sudo / admin) and we never enable or start it — the operator runs the printed
+activation command themselves, so nothing touches the system unasked.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from xml.sax.saxutils import escape
+
+_LABEL = "coding-bridge-channels"
+_MAC_LABEL = "cloud.acedata.coding-bridge-channels"
+
+_TOKEN_NOTE = (
+    "If any instance uses token_env, a background service won't see your shell "
+    "exports — switch that instance to token_file in channels.toml, or add the "
+    "token(s) to the unit's environment before enabling."
+)
+
+
+@dataclass(frozen=True)
+class ServicePlan:
+    """A ready-to-write service unit + the commands to activate it."""
+
+    kind: str  # "systemd" | "launchd" | "schtasks"
+    path: Path
+    content: str
+    activate: list[str]
+    notes: list[str] = field(default_factory=list)
+
+
+def _validate(python_exe: str, config_dir: str) -> None:
+    """Reject control chars so a path can't inject a second unit directive.
+
+    ``python_exe``/``config_dir`` are the operator's own local paths, but a
+    newline in one would let it smuggle an extra ``systemd`` line (or break the
+    ``.cmd``), so we fail fast rather than emit a dangerous unit.
+    """
+    for label, value in (("python path", python_exe), ("config dir", config_dir)):
+        if any(ord(c) < 0x20 for c in value):
+            raise ValueError(f"{label} contains a control character; refusing to write a unit")
+
+
+def _systemd(python_exe: str, config_dir: str) -> str:
+    return (
+        "[Unit]\n"
+        "Description=Coding Bridge channels (WeChat / Telegram)\n"
+        "After=network-online.target\n"
+        "Wants=network-online.target\n"
+        "\n"
+        "[Service]\n"
+        "Type=simple\n"
+        f'Environment="CODING_BRIDGE_CONFIG_DIR={config_dir}"\n'
+        f'ExecStart="{python_exe}" -m coding_bridge channels start\n'
+        "Restart=on-failure\n"
+        "RestartSec=5\n"
+        "\n"
+        "[Install]\n"
+        "WantedBy=default.target\n"
+    )
+
+
+def _launchd(python_exe: str, config_dir: str) -> str:
+    py = escape(python_exe)
+    cfg = escape(config_dir)
+    return (
+        '<?xml version="1.0" encoding="UTF-8"?>\n'
+        '<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" '
+        '"http://www.apple.com/DTDs/PropertyList-1.0.dtd">\n'
+        '<plist version="1.0">\n'
+        "<dict>\n"
+        f"  <key>Label</key><string>{_MAC_LABEL}</string>\n"
+        "  <key>ProgramArguments</key>\n"
+        "  <array>\n"
+        f"    <string>{py}</string>\n"
+        "    <string>-m</string>\n"
+        "    <string>coding_bridge</string>\n"
+        "    <string>channels</string>\n"
+        "    <string>start</string>\n"
+        "  </array>\n"
+        "  <key>EnvironmentVariables</key>\n"
+        "  <dict>\n"
+        f"    <key>CODING_BRIDGE_CONFIG_DIR</key><string>{cfg}</string>\n"
+        "  </dict>\n"
+        "  <key>RunAtLoad</key><true/>\n"
+        "  <key>KeepAlive</key><true/>\n"
+        "</dict>\n"
+        "</plist>\n"
+    )
+
+
+def _win_cmd(python_exe: str, config_dir: str) -> str:
+    # CRLF + quoted paths so a space in the python path / config dir is safe.
+    return (
+        "@echo off\r\n"
+        f'set "CODING_BRIDGE_CONFIG_DIR={config_dir}"\r\n'
+        f'"{python_exe}" -m coding_bridge channels start\r\n'
+    )
+
+
+def build_service_plan(
+    system: str, python_exe: str, config_dir: Path | str, home: Path | str
+) -> ServicePlan:
+    """Render the service unit for ``system`` ("linux" / "darwin" / "windows")."""
+    cfg = str(config_dir)
+    py = str(python_exe)
+    home = Path(home)
+    _validate(py, cfg)
+    if system == "linux":
+        path = home / ".config" / "systemd" / "user" / f"{_LABEL}.service"
+        return ServicePlan(
+            "systemd",
+            path,
+            _systemd(py, cfg),
+            [
+                "systemctl --user daemon-reload",
+                f"systemctl --user enable --now {_LABEL}.service",
+            ],
+            [
+                _TOKEN_NOTE,
+                "To keep it running without an active login: "
+                "`sudo loginctl enable-linger $USER`.",
+            ],
+        )
+    if system == "darwin":
+        path = home / "Library" / "LaunchAgents" / f"{_MAC_LABEL}.plist"
+        return ServicePlan(
+            "launchd",
+            path,
+            _launchd(py, cfg),
+            [f"launchctl load {path}"],
+            [_TOKEN_NOTE],
+        )
+    if system == "windows":
+        path = Path(cfg) / "run-channels.cmd"
+        return ServicePlan(
+            "schtasks",
+            path,
+            _win_cmd(py, cfg),
+            [
+                f'schtasks /create /tn "CodingBridgeChannels" /tr "{path}" '
+                "/sc onlogon /rl limited /f",
+            ],
+            [
+                _TOKEN_NOTE,
+                "Runs at next logon. Start it now with "
+                "`schtasks /run /tn CodingBridgeChannels`.",
+            ],
+        )
+    raise ValueError(f"unsupported platform {system!r}")
+
+
+__all__ = ["ServicePlan", "build_service_plan"]

coding_bridge/channels_cli.py

@@ -21,6 +21,7 @@
 import asyncio
 import contextlib
 import logging
+import os
 import stat
 import sys
 import time
@@ -572,6 +573,50 @@ def cmd_channels_portal(
     return serve(settings, host=host, port=port, open_browser=open_browser)
 
 
+# ---------- install-service ---------------------------------------------------
+
+
+def cmd_channels_install_service(settings: Settings, *, force: bool) -> int:
+    """Write a user-scoped OS service that runs `channels start` at login/boot.
+
+    Generates the unit file only + prints the single command that enables it —
+    it never enables/starts the service or touches anything system-wide itself.
+    """
+    import platform as _platform
+
+    from .channels.service import build_service_plan
+
+    system = _platform.system().lower()
+    try:
+        plan = build_service_plan(system, sys.executable, settings.config_dir, Path.home())
+    except ValueError:
+        print(
+            f"install-service isn't supported on this platform ({system or 'unknown'}). "
+            "See docs/deploy/ for manual templates.",
+            file=sys.stderr,
+        )
+        return 2
+    if plan.path.exists() and not force:
+        print(f"Refusing to overwrite existing {plan.path} (pass --force).", file=sys.stderr)
+        return 1
+    try:
+        plan.path.parent.mkdir(parents=True, exist_ok=True)
+        plan.path.write_text(plan.content, encoding="utf-8")
+        if os.name == "posix":
+            with contextlib.suppress(OSError):
+                plan.path.chmod(0o700 if plan.kind == "schtasks" else 0o600)
+    except OSError as exc:
+        print(f"could not write {plan.path}: {exc.__class__.__name__}", file=sys.stderr)
+        return 1
+    print(f"Wrote {plan.kind} unit → {plan.path}")
+    print("\nActivate it (starts `coding-bridge channels start` in the background):")
+    for cmd in plan.activate:
+        print(f"  {cmd}")
+    for note in plan.notes:
+        print(f"\nnote: {note}")
+    return 0
+
+
 # ---------- argparse wiring ---------------------------------------------------
 
 
@@ -640,6 +685,16 @@ def register_subparsers(
     )
     p_portal.set_defaults(func=_dispatch_portal)
 
+    p_svc = sub.add_parser(
+        "install-service",
+        help="Write an OS service unit that runs `channels start` at login/boot",
+        parents=[common],
+    )
+    p_svc.add_argument(
+        "--force", action="store_true", help="Overwrite an existing unit file."
+    )
+    p_svc.set_defaults(func=_dispatch_install_service)
+
 
 def _dispatch_init(args: argparse.Namespace) -> None:
     from .cli import _build_settings  # local import to avoid circular
@@ -685,9 +740,16 @@ def _dispatch_portal(args: argparse.Namespace) -> None:
     )
 
 
+def _dispatch_install_service(args: argparse.Namespace) -> None:
+    from .cli import _build_settings
+
+    raise SystemExit(cmd_channels_install_service(_build_settings(args), force=args.force))
+
+
 __all__ = [
     "cmd_channels_doctor",
     "cmd_channels_init",
+    "cmd_channels_install_service",
     "cmd_channels_portal",
     "cmd_channels_smoke",
     "cmd_channels_start",

tests/test_channels_service.py

@@ -0,0 +1,70 @@
+"""Tests for `channels install-service` unit rendering (pure, no filesystem)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from coding_bridge.channels.service import build_service_plan
+
+
+def test_systemd_plan(tmp_path):
+    plan = build_service_plan("linux", "/usr/bin/python3", "/home/u/.ace-bridge", tmp_path)
+    assert plan.kind == "systemd"
+    assert plan.path == tmp_path / ".config/systemd/user/coding-bridge-channels.service"
+    assert 'ExecStart="/usr/bin/python3" -m coding_bridge channels start' in plan.content
+    assert 'Environment="CODING_BRIDGE_CONFIG_DIR=/home/u/.ace-bridge"' in plan.content
+    assert "WantedBy=default.target" in plan.content
+    assert any("systemctl --user enable --now" in c for c in plan.activate)
+
+
+def test_launchd_plan(tmp_path):
+    plan = build_service_plan("darwin", "/usr/bin/python3", "/Users/u/.ace-bridge", tmp_path)
+    assert plan.kind == "launchd"
+    assert (
+        plan.path
+        == tmp_path / "Library/LaunchAgents/cloud.acedata.coding-bridge-channels.plist"
+    )
+    assert "<string>coding_bridge</string>" in plan.content
+    assert "<string>/usr/bin/python3</string>" in plan.content
+    assert "<string>/Users/u/.ace-bridge</string>" in plan.content
+    assert any("launchctl load" in c for c in plan.activate)
+
+
+def test_windows_plan(tmp_path):
+    plan = build_service_plan(
+        "windows", r"C:\Py\python.exe", r"C:\Users\u\.ace-bridge", tmp_path
+    )
+    assert plan.kind == "schtasks"
+    assert plan.path == Path(r"C:\Users\u\.ace-bridge") / "run-channels.cmd"
+    assert '"C:\\Py\\python.exe" -m coding_bridge channels start' in plan.content
+    assert 'set "CODING_BRIDGE_CONFIG_DIR=C:\\Users\\u\\.ace-bridge"' in plan.content
+    assert any(
+        "schtasks /create" in c and "CodingBridgeChannels" in c for c in plan.activate
+    )
+
+
+def test_unsupported_platform_raises(tmp_path):
+    with pytest.raises(ValueError):
+        build_service_plan("plan9", "/py", "/cfg", tmp_path)
+
+
+def test_launchd_escapes_xml_special_chars(tmp_path):
+    plan = build_service_plan("darwin", "/opt/py&x/python", "/cfg", tmp_path)
+    assert "<string>/opt/py&amp;x/python</string>" in plan.content
+    assert "/opt/py&x/python</string>" not in plan.content  # raw & would break the plist
+
+
+def test_all_plans_carry_token_note(tmp_path):
+    for sysname in ("linux", "darwin", "windows"):
+        plan = build_service_plan(sysname, "/py", "/cfg", tmp_path)
+        assert any("token_env" in n for n in plan.notes)
+
+
+def test_rejects_control_char_in_path(tmp_path):
+    # a newline could smuggle a second systemd directive / break the .cmd
+    with pytest.raises(ValueError):
+        build_service_plan("linux", "/usr/bin/py\nExecStart=evil", "/cfg", tmp_path)
+    with pytest.raises(ValueError):
+        build_service_plan("linux", "/py", "/cfg\nRestart=no", tmp_path)