feat(channels): add channels enable wechat for one-command, no-portal setup

coding_bridge/channels_cli.py

@@ -140,6 +140,165 @@ def cmd_channels_init(settings: Settings) -> int:
     return 0
 
 
+# ---------- enable ------------------------------------------------------------
+
+
+def _derive_instance_id(base_url: str) -> str:
+    """A stable, collision-resistant instance id from the gateway host.
+
+    ``https://ebcdd3c54eb2.wisdom.acedata.cloud`` -> ``wisdom-ebcdd3c54eb2`` so
+    two AceData appliances stay human-readable and distinct. For any other host
+    use the FULL host (dots -> dashes) so unrelated gateways never collapse to
+    the same id, e.g. ``foo.example.com`` -> ``foo-example-com`` and
+    ``http://1.2.3.4:8000`` -> ``1-2-3-4``.
+    """
+    from urllib.parse import urlparse
+
+    host = (urlparse(base_url).hostname or "wechat").lower()
+    labels = [p for p in host.split(".") if p]
+    if len(labels) >= 3 and labels[1] == "wisdom":
+        return f"wisdom-{labels[0]}"
+    return "-".join(labels) if labels else "wechat"
+
+
+def cmd_channels_enable(
+    settings: Settings,
+    *,
+    base_url: str | None,
+    instance_id: str | None,
+    token_env: str | None,
+    sender: list[str] | None,
+) -> int:
+    """Non-interactively add/enable a ``[[channels.wechat]]`` instance.
+
+    Fills the gap the portal used to require a human for: given a gateway URL
+    (flag or ``WECHAT_BASE_URL``), it writes a valid, enabled block into
+    ``channels.toml`` - creating the file (0600) if missing, or merging into an
+    existing one by ``instance_id`` without touching other channels. Idempotent:
+    re-running with the same values just rewrites the same block.
+    """
+    from dataclasses import replace as _dc_replace
+    from urllib.parse import urlparse
+
+    from .channels.config import WeChatInstanceConfig
+    from .channels.portal import dump_channels_toml
+
+    resolved_url = base_url or os.environ.get("WECHAT_BASE_URL")
+    if not resolved_url:
+        print(
+            "No gateway URL. Pass --base-url <https://...> or set WECHAT_BASE_URL.",
+            file=sys.stderr,
+        )
+        return 1
+    resolved_url = resolved_url.rstrip("/")
+    # Validate up front (same rules the loader enforces) so we never persist a
+    # base_url that `doctor`/`start` will later reject - and never write a URL
+    # with embedded credentials into the file.
+    if not (resolved_url.startswith("http://") or resolved_url.startswith("https://")):
+        print(
+            f"Invalid --base-url {resolved_url!r}: must start with http:// or https://",
+            file=sys.stderr,
+        )
+        return 1
+    parsed = urlparse(resolved_url)
+    if not parsed.hostname:
+        print(f"Invalid --base-url {resolved_url!r}: no host", file=sys.stderr)
+        return 1
+    if "@" in parsed.netloc:
+        print(
+            f"Invalid --base-url {resolved_url!r}: must not contain embedded "
+            "credentials (user:pass@host); the token goes in WECHAT_TOKEN.",
+            file=sys.stderr,
+        )
+        return 1
+    resolved_id = instance_id or os.environ.get("WECHAT_INSTANCE_ID") or _derive_instance_id(
+        resolved_url
+    )
+    resolved_token_env = token_env or os.environ.get("WECHAT_TOKEN_ENV") or "WECHAT_TOKEN"
+
+    path = settings.channels_config_path
+    existing = None
+    if path.exists():
+        try:
+            existing = load_channels_config(path)
+        except ConfigError as exc:
+            print(f"Existing {path} is invalid, refusing to edit: {exc}", file=sys.stderr)
+            print(
+                "Fix or delete it, then re-run `coding-bridge channels enable`.",
+                file=sys.stderr,
+            )
+            return 1
+
+    # Merge: update the instance with the same id (PRESERVING its other fields -
+    # allowed_senders, rate limits, approval, etc. - so a re-run can't silently
+    # widen a previously restricted bot), keep every other channel untouched.
+    prior = None
+    if existing is not None:
+        prior = next((i for i in existing.wechat if i.instance_id == resolved_id), None)
+
+    if prior is not None:
+        # Only override what the caller actually specified.
+        new_inst = _dc_replace(
+            prior,
+            base_url=resolved_url,
+            token_env=resolved_token_env,
+            enabled=True,
+        )
+        if sender is not None:
+            new_inst = _dc_replace(new_inst, allowed_senders=tuple(sender))
+    else:
+        new_inst = WeChatInstanceConfig(
+            instance_id=resolved_id,
+            base_url=resolved_url,
+            token_env=resolved_token_env,
+            enabled=True,
+            # Free-form by default (reply to every message) - a personal bridge
+            # to your own gateway; add --sender to restrict who can drive it.
+            trigger_prefix="",
+            allowed_senders=tuple(sender or ()),
+        )
+    senders = list(new_inst.allowed_senders)
+
+    wechat: list[WeChatInstanceConfig] = []
+    replaced = False
+    if existing is not None:
+        for inst in existing.wechat:
+            if inst.instance_id == resolved_id:
+                wechat.append(new_inst)
+                replaced = True
+            else:
+                wechat.append(inst)
+    if not replaced:
+        wechat.append(new_inst)
+    telegram = list(existing.telegram) if existing is not None else []
+
+    from .channels.config import ChannelsConfig
+
+    body = dump_channels_toml(ChannelsConfig(wechat=tuple(wechat), telegram=tuple(telegram)))
+    if path.exists():
+        path.write_text(body, encoding="utf-8")
+    else:
+        _write_secure_file(path, body)
+    if sys.platform != "win32":
+        with contextlib.suppress(OSError):
+            path.chmod(0o600)
+
+    action = "Updated" if replaced else "Enabled"
+    print(f"{action} wechat instance {resolved_id!r} in {path}")
+    print(f"  base_url  = {resolved_url}")
+    print(f"  token_env = {resolved_token_env}")
+    if not senders:
+        print("  allowed_senders = []  (everyone - restrict with --sender <wxid> if shared)")
+    token_set = bool(os.environ.get(resolved_token_env))
+    if not token_set:
+        print(f"\nNext: export {resolved_token_env}=\"<gateway token>\" in this shell,")
+        print("      then `coding-bridge channels doctor` and `channels start`.")
+    else:
+        print("\nNext: `coding-bridge channels doctor` then `coding-bridge channels start`.")
+    return 0
+
+
+
 # ---------- doctor ------------------------------------------------------------
 
 
@@ -638,6 +797,41 @@ def register_subparsers(
     )
     p_init.set_defaults(func=_dispatch_init)
 
+    p_enable = sub.add_parser(
+        "enable",
+        help="Add/enable a WeChat instance from a URL or env vars (no manual editing)",
+        parents=[common],
+    )
+    p_enable.add_argument(
+        "channel",
+        choices=["wechat"],
+        help="Channel to enable (only 'wechat' today; Telegram needs no gateway URL).",
+    )
+    p_enable.add_argument(
+        "--base-url",
+        default=None,
+        help="Gateway URL, e.g. https://<id>.wisdom.acedata.cloud. "
+        "Falls back to the WECHAT_BASE_URL env var.",
+    )
+    p_enable.add_argument(
+        "--id",
+        dest="instance_id",
+        default=None,
+        help="Instance id (unique). Default: derived from the gateway host.",
+    )
+    p_enable.add_argument(
+        "--token-env",
+        default=None,
+        help="Env var name holding the gateway token. Default: WECHAT_TOKEN.",
+    )
+    p_enable.add_argument(
+        "--sender",
+        action="append",
+        default=None,
+        help="Restrict to this sender wxid (repeatable). Default: allow all.",
+    )
+    p_enable.set_defaults(func=_dispatch_enable)
+
     p_start = sub.add_parser(
         "start",
         help="Run every enabled [[channels.wechat]] instance until Ctrl-C",
@@ -706,6 +900,20 @@ def _dispatch_init(args: argparse.Namespace) -> None:
     raise SystemExit(cmd_channels_init(_build_settings(args)))
 
 
+def _dispatch_enable(args: argparse.Namespace) -> None:
+    from .cli import _build_settings
+
+    raise SystemExit(
+        cmd_channels_enable(
+            _build_settings(args),
+            base_url=args.base_url,
+            instance_id=args.instance_id,
+            token_env=args.token_env,
+            sender=args.sender,
+        )
+    )
+
+
 def _dispatch_start(args: argparse.Namespace) -> None:
     from .cli import _build_settings
 

tests/test_channels_cli.py

@@ -1,6 +1,6 @@
 """Tests for the ``coding-bridge channels`` CLI subcommand group.
 
-Focus: pure surface — arg parsing, init file write behavior, doctor output
+Focus: pure surface - arg parsing, init file write behavior, doctor output
 against mocked httpx. `start` is thin glue and is smoke-tested by the E2E.
 """
 
@@ -46,7 +46,7 @@ def test_creates_file_with_safe_defaults(self, tmp_path: Path) -> None:
         assert rc == 0
         assert s.channels_config_path.exists()
         body = s.channels_config_path.read_text(encoding="utf-8")
-        # Every example is commented out — nothing enabled by default
+        # Every example is commented out - nothing enabled by default
         assert "# [[channels.wechat]]" in body
         # `enabled = false` appears in the template
         assert "enabled = false" in body
@@ -73,6 +73,120 @@ def test_posix_permissions_are_0600(self, tmp_path: Path) -> None:
         assert mode == 0o600, f"expected 0o600, got {oct(mode)}"
 
 
+# ---------- enable ------------------------------------------------------------
+
+
+class TestChannelsEnable:
+    def _enable(self, s, **kw):
+        kw.setdefault("base_url", None)
+        kw.setdefault("instance_id", None)
+        kw.setdefault("token_env", None)
+        kw.setdefault("sender", None)
+        return _capture(lambda: channels_cli.cmd_channels_enable(s, **kw))
+
+    def test_creates_enabled_block_from_flag(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        rc, out, err = self._enable(
+            s, base_url="https://ebcdd3c54eb2.wisdom.acedata.cloud"
+        )
+        assert rc == 0
+        assert s.channels_config_path.exists()
+        cfg = channels_cli.load_channels_config(s.channels_config_path)
+        assert len(cfg.wechat) == 1
+        inst = cfg.wechat[0]
+        assert inst.enabled is True
+        assert inst.base_url == "https://ebcdd3c54eb2.wisdom.acedata.cloud"
+        # id derived from a *.wisdom.acedata.cloud host
+        assert inst.instance_id == "wisdom-ebcdd3c54eb2"
+        assert inst.token_env == "WECHAT_TOKEN"
+
+    def test_reads_base_url_from_env(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("WECHAT_BASE_URL", "https://host.example.com")
+        s = _settings(tmp_path)
+        rc, out, err = self._enable(s)
+        assert rc == 0
+        cfg = channels_cli.load_channels_config(s.channels_config_path)
+        assert cfg.wechat[0].base_url == "https://host.example.com"
+
+    def test_missing_url_errors(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        rc, out, err = self._enable(s)
+        assert rc == 1
+        assert "WECHAT_BASE_URL" in err
+        assert not s.channels_config_path.exists()
+
+    def test_rerun_updates_in_place_no_duplicate(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        self._enable(s, base_url="https://host.example.com", instance_id="w1")
+        self._enable(
+            s, base_url="https://host.example.com", instance_id="w1", sender=["Msg_x"]
+        )
+        cfg = channels_cli.load_channels_config(s.channels_config_path)
+        assert len(cfg.wechat) == 1
+        assert cfg.wechat[0].allowed_senders == ("Msg_x",)
+
+    def test_rerun_without_sender_preserves_restrictions(self, tmp_path: Path) -> None:
+        # A re-run that omits --sender must NOT widen a previously restricted bot.
+        s = _settings(tmp_path)
+        self._enable(
+            s, base_url="https://host.example.com", instance_id="w1", sender=["Msg_x"]
+        )
+        self._enable(s, base_url="https://host.example.com", instance_id="w1")
+        cfg = channels_cli.load_channels_config(s.channels_config_path)
+        assert cfg.wechat[0].allowed_senders == ("Msg_x",)
+
+    def test_rejects_non_http_url(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        rc, out, err = self._enable(s, base_url="ftp://x.example.com")
+        assert rc == 1
+        assert "http://" in err
+        assert not s.channels_config_path.exists()
+
+    def test_rejects_embedded_credentials(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        rc, out, err = self._enable(s, base_url="https://user:pass@host.example.com")
+        assert rc == 1
+        assert "credentials" in err.lower()
+        assert not s.channels_config_path.exists()
+
+    def test_derived_id_uses_full_host_for_non_wisdom(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        self._enable(s, base_url="https://foo.example.com")
+        cfg = channels_cli.load_channels_config(s.channels_config_path)
+        assert cfg.wechat[0].instance_id == "foo-example-com"
+
+    def test_preserves_other_instances(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        self._enable(s, base_url="https://a.example.com", instance_id="a")
+        self._enable(s, base_url="https://b.example.com", instance_id="b")
+        cfg = channels_cli.load_channels_config(s.channels_config_path)
+        ids = sorted(i.instance_id for i in cfg.wechat)
+        assert ids == ["a", "b"]
+
+    def test_refuses_to_edit_invalid_existing(self, tmp_path: Path) -> None:
+        s = _settings(tmp_path)
+        s.channels_config_path.parent.mkdir(parents=True, exist_ok=True)
+        # A wechat block missing instance_id -> load fails -> enable must refuse.
+        s.channels_config_path.write_text(
+            "[[channels.wechat]]\nbase_url = 'https://x.example.com'\nenabled = true\n",
+            encoding="utf-8",
+        )
+        rc, out, err = self._enable(s, base_url="https://y.example.com")
+        assert rc == 1
+        assert "refusing to edit" in err.lower()
+
+    @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits don't apply on Windows")
+    def test_created_file_is_0600(self, tmp_path: Path) -> None:
+        import stat as _stat
+
+        s = _settings(tmp_path)
+        self._enable(s, base_url="https://host.example.com")
+        mode = _stat.S_IMODE(s.channels_config_path.stat().st_mode)
+        assert mode == 0o600, f"expected 0o600, got {oct(mode)}"
+
+
 # ---------- doctor ------------------------------------------------------------
 
 
@@ -118,7 +232,7 @@ def test_enabled_reachable_endpoint_marked_ok(
         self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         # Patch WeChatClient to use a MockTransport where the tasks endpoint
-        # returns 404 (token accepted, probe id unknown) — the happy path.
+        # returns 404 (token accepted, probe id unknown) - the happy path.
         monkeypatch.setenv("MY_TOKEN", "abc")
         s = _settings(tmp_path)
         s.config_dir.mkdir(parents=True, exist_ok=True)
@@ -380,7 +494,7 @@ def script(sid):
     def test_smoke_timeout_no_text_exits_1(self, tmp_path: Path, monkeypatch) -> None:
         # Provider emits NOTHING and never signals completion. With a tiny
         # turn_timeout the dispatcher synthesises "(provider timed out; no
-        # reply)" — smoke must report failure (rc=1), not a false-healthy 0.
+        # reply)" - smoke must report failure (rc=1), not a false-healthy 0.
         # (This is the BLOCKER the adversarial review caught: the old exit-code
         # check ignored the timeout marker.)
         def script(_sid):