diff --git a/nanobot/channels/weixin/runtime.py b/nanobot/channels/weixin/runtime.py index 366ecd895..018e9bb90 100644 --- a/nanobot/channels/weixin/runtime.py +++ b/nanobot/channels/weixin/runtime.py @@ -486,6 +486,35 @@ class WeixinChannel(BaseChannel): if base_url: self.config.base_url = base_url self._save_state(force=True) + self._persist_connect_credentials(token=token, base_url=base_url) + + def _persist_connect_credentials(self, *, token: str, base_url: str) -> None: + """Write the QR-login token and base_url back to config.json. + + The connect flow saves account state to ``account.json`` (via + ``_save_state``), but the WebUI's post-connect ``enable`` step calls + ``set_channel_config_enabled`` which reads config.json. Without + persisting the token here, that step would overwrite it with the + default empty value, losing the freshly obtained credential. + """ + from nanobot.config.loader import get_config_path, load_config, save_config + + try: + full_config = load_config() + section = getattr(full_config.channels, "weixin", None) + if section is not None and hasattr(section, "model_dump"): + values = section.model_dump(mode="json", by_alias=True) + elif isinstance(section, dict): + values = dict(cast(dict[str, Any], section)) + else: + values = {} + values["token"] = token + if base_url: + values["baseUrl"] = base_url + setattr(full_config.channels, "weixin", values) + save_config(full_config, get_config_path()) + except Exception: + self.logger.exception("Failed to persist WeChat credentials to config.json") # ------------------------------------------------------------------ # HTTP helpers (matches api.ts buildHeaders / apiFetch) diff --git a/nanobot/channels/weixin/tests/test_connect.py b/nanobot/channels/weixin/tests/test_connect.py index 511c1881c..aec484780 100644 --- a/nanobot/channels/weixin/tests/test_connect.py +++ b/nanobot/channels/weixin/tests/test_connect.py @@ -66,6 +66,63 @@ async def test_weixin_connect_store_saves_confirmed_qr_login( assert saved["token"] == "wx-token" assert saved["base_url"] == "https://weixin.example" + # Token and base_url must also be persisted to config.json so the + # post-connect enable step does not overwrite them with empty defaults. + config_data = json.loads(config_path.read_text(encoding="utf-8")) + weixin_cfg = config_data.get("channels", {}).get("weixin", {}) + assert weixin_cfg.get("token") == "wx-token" + assert weixin_cfg.get("baseUrl") == "https://weixin.example" + assert weixin_cfg.get("stateDir") == str(state_dir) + + +@pytest.mark.asyncio +async def test_weixin_connect_persists_credentials_without_channels_config( + tmp_path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """When config.json has no channels key at all, connect must still write + the obtained token and base_url back to config.json.""" + config_path = tmp_path / "config.json" + # config.json with NO channels key — the bug scenario + config_path.write_text( + json.dumps({"agents": {"defaults": {"model": "test"}}}), + encoding="utf-8", + ) + monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path) + + async def fake_fetch_qr_code( + self: WeixinChannel, **_kwargs: Any + ) -> tuple[str, str]: + return "qr-1", "https://qr.example/1" + + async def fake_api_get_with_base( + self: WeixinChannel, + *, + base_url: str, + endpoint: str, + params: dict[str, Any], + auth: bool, + ) -> dict[str, str]: + return { + "status": "confirmed", + "bot_token": "wx-token", + "baseurl": "https://weixin.example", + "ilink_user_id": "wx-user", + } + + monkeypatch.setattr(WeixinChannel, "_fetch_qr_code", fake_fetch_qr_code) + monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base) + + store = WeixinConnectStore() + started = await store.start() + completed = await store.poll(started["session_id"]) + assert completed["status"] == "succeeded" + + config_data = json.loads(config_path.read_text(encoding="utf-8")) + weixin_cfg = config_data.get("channels", {}).get("weixin", {}) + assert weixin_cfg.get("token") == "wx-token" + assert weixin_cfg.get("baseUrl") == "https://weixin.example" + @pytest.mark.asyncio async def test_weixin_reconnect_keeps_existing_account_until_scan_succeeds(