Beaver/beaver_project
6
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
c3a0aef104871b72d563678b8936a21a7a07d4c2
beaver_project/docs/superpowers/plans/2026-06-02-channel-connectors-foundation.md

63 KiB
Raw Blame History

Channel Connectors Foundation Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build the first connector slice: durable channel connections, pairing/credential primitives, connector registry, Telegram token connector, and backend APIs that materialize a runtime channel without manual JSON editing.

Architecture: Add a connector layer under beaver/interfaces/channels/connections/ while keeping ChannelRuntime, MessageBus, and existing adapters as the message path. A ChannelConnectionStore persists setup state, a small credential vault stores secrets by reference, and ChannelConnectorRegistry materializes enabled connections into ChannelConfig objects during app startup. The first concrete connector is Telegram because token validation and runtime materialization are simple and testable with fake clients.

Tech Stack: Python dataclasses, FastAPI, Pydantic v2, local JSON stores, pytest, existing Beaver channel runtime.

Scope

This plan implements phase 1 of docs/superpowers/specs/2026-06-02-channel-connectors-and-pairing-design.md.

Included:

  • ChannelConnection data model and persistent JSON store.
  • Restricted local credential store with secret redaction.
  • One-time pairing token store, used now by tests and future terminal/QR connectors. It is implemented in this phase but not exposed through APIs; future terminal and QR connectors will consume it.
  • Connector protocol and registry.
  • Telegram connector with fake-client test hooks.
  • Connection control APIs.
  • App startup materialization from connections into ChannelRuntime.

Excluded from this plan:

  • Terminal authenticated pairing.
  • Feishu/Lark official SDK integration.
  • Weixin external connector process.
  • QQBot connector.
  • Frontend connection wizard.
  • Hot starting/stopping adapters without backend restart.
  • Multi-process-safe storage. The JSON stores use threading.Lock plus atomic file replace for the single backend process used in phase 1. Production multi-worker deployment needs a file lock or database-backed store.
  • Credential garbage collection. Updating secrets writes a new credential reference and leaves the old reference in the local credential file until a later cleanup pass.

File Structure

  • Create app-instance/backend/beaver/interfaces/channels/connections/__init__.py
    • Exports connection models, stores, connector registry, and Telegram connector.
  • Create app-instance/backend/beaver/interfaces/channels/connections/models.py
    • Dataclasses and constants for ChannelConnection, PairingSession, ChannelRuntimeSpec, ValidationResult.
  • Create app-instance/backend/beaver/interfaces/channels/connections/store.py
    • JSON-backed ChannelConnectionStore, CredentialStore, and PairingTokenStore.
  • Create app-instance/backend/beaver/interfaces/channels/connections/connectors.py
    • ChannelConnector protocol and ChannelConnectorRegistry.
  • Create app-instance/backend/beaver/interfaces/channels/connections/telegram.py
    • Telegram token connector that validates via injected client factory and materializes a runtime spec.
  • Modify app-instance/backend/beaver/interfaces/web/schemas/chat.py
    • Add Pydantic request/response models for connection APIs.
  • Modify app-instance/backend/beaver/interfaces/web/schemas/__init__.py
    • Export the new schemas.
  • Modify app-instance/backend/beaver/interfaces/web/app.py
    • Instantiate connection stores/registry, expose /api/channel-connectors and /api/channel-connections APIs, and merge materialized connection configs into runtime startup.
  • Create app-instance/backend/tests/unit/test_channel_connection_store.py
    • Store, credential redaction, and pairing token tests.
  • Create app-instance/backend/tests/unit/test_channel_connector_registry.py
    • Registry dispatch and runtime materialization tests.
  • Create app-instance/backend/tests/unit/test_telegram_channel_connector.py
    • Telegram validation/materialization tests with fake client.
  • Create app-instance/backend/tests/unit/test_channel_connection_api.py
    • FastAPI endpoint tests with fake service/app context.

Task 1: Connection Models And Store

Files:

  • Create: app-instance/backend/beaver/interfaces/channels/connections/__init__.py

  • Create: app-instance/backend/beaver/interfaces/channels/connections/models.py

  • Create: app-instance/backend/beaver/interfaces/channels/connections/store.py

  • Test: app-instance/backend/tests/unit/test_channel_connection_store.py

  • Step 1: Write failing store tests

Create app-instance/backend/tests/unit/test_channel_connection_store.py:

from __future__ import annotations

from beaver.interfaces.channels.connections import (
    ChannelConnectionStore,
    CredentialStore,
    PairingTokenStore,
)


def test_channel_connection_store_creates_updates_lists_and_revokes(tmp_path) -> None:
    store = ChannelConnectionStore(tmp_path / "connections.json")

    created = store.create(
        kind="telegram",
        mode="polling",
        display_name="Telegram Main",
        account_id="telegram:bot-main",
        owner_user_id="user-1",
        auth_type="token",
        runtime_config={"max_message_chars": 4096},
        capabilities=["receive_text", "send_text"],
    )
    updated = store.update_status(created.connection_id, status="connected", last_error=None)
    revoked = store.revoke(created.connection_id)

    assert created.connection_id
    assert created.channel_id.startswith("telegram-")
    assert created.status == "draft"
    assert updated.status == "connected"
    assert revoked.status == "revoked"
    assert store.get(created.connection_id).status == "revoked"
    assert [item.connection_id for item in store.list()] == [created.connection_id]


def test_credential_store_saves_values_by_reference_and_redacts_views(tmp_path) -> None:
    store = CredentialStore(tmp_path / "credentials.json")

    ref = store.put(kind="telegram", values={"botToken": "secret-token", "empty": ""})

    assert ref.startswith("cred_")
    assert store.get(ref) == {"botToken": "secret-token"}
    assert store.redacted(ref) == {"botToken": "***"}


def test_pairing_token_store_uses_one_time_expiring_tokens(tmp_path) -> None:
    store = PairingTokenStore(tmp_path / "pairing.json")

    session = store.create(kind="terminal", ttl_seconds=60, scope="channel:pair")
    consumed = store.consume(session.token, expected_kind="terminal")
    reused = store.consume(session.token, expected_kind="terminal")

    assert session.status == "pending"
    assert consumed is not None
    assert consumed.status == "consumed"
    assert reused is None


def test_pairing_token_store_rejects_expired_tokens(tmp_path) -> None:
    store = PairingTokenStore(tmp_path / "pairing.json")

    session = store.create(kind="weixin", ttl_seconds=-1, scope="channel:pair")

    assert store.consume(session.token, expected_kind="weixin") is None
  • Step 2: Run tests to verify they fail

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connection_store.py -q

Expected: fail during import with ModuleNotFoundError: No module named 'beaver.interfaces.channels.connections'.

  • Step 3: Implement connection dataclasses

Create app-instance/backend/beaver/interfaces/channels/connections/models.py:

"""Channel connection setup models."""

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from typing import Any


CONNECTION_STATUSES = {"draft", "pairing", "connected", "running", "degraded", "error", "revoked"}


def iso_now() -> str:
    return datetime.now(timezone.utc).isoformat()


@dataclass(slots=True)
class ChannelConnection:
    connection_id: str
    owner_user_id: str | None
    channel_id: str
    kind: str
    mode: str
    display_name: str
    account_id: str
    status: str
    auth_type: str
    credentials_ref: str | None = None
    connector_ref: str | None = None
    pairing_session_id: str | None = None
    runtime_config: dict[str, Any] = field(default_factory=dict)
    capabilities: list[str] = field(default_factory=list)
    created_at: str = field(default_factory=iso_now)
    updated_at: str = field(default_factory=iso_now)
    last_seen_at: str | None = None
    last_error: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "ChannelConnection":
        return cls(
            connection_id=str(data.get("connection_id") or ""),
            owner_user_id=_optional_string(data.get("owner_user_id")),
            channel_id=str(data.get("channel_id") or ""),
            kind=str(data.get("kind") or ""),
            mode=str(data.get("mode") or ""),
            display_name=str(data.get("display_name") or ""),
            account_id=str(data.get("account_id") or ""),
            status=str(data.get("status") or "draft"),
            auth_type=str(data.get("auth_type") or ""),
            credentials_ref=_optional_string(data.get("credentials_ref")),
            connector_ref=_optional_string(data.get("connector_ref")),
            pairing_session_id=_optional_string(data.get("pairing_session_id")),
            runtime_config=dict(data.get("runtime_config") or {}),
            capabilities=[str(item) for item in data.get("capabilities") or []],
            created_at=str(data.get("created_at") or iso_now()),
            updated_at=str(data.get("updated_at") or iso_now()),
            last_seen_at=_optional_string(data.get("last_seen_at")),
            last_error=_optional_string(data.get("last_error")),
        )


@dataclass(slots=True)
class PairingSession:
    pairing_session_id: str
    kind: str
    scope: str
    token: str
    status: str
    expires_at_ms: int
    created_at_ms: int

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "PairingSession":
        return cls(
            pairing_session_id=str(data.get("pairing_session_id") or ""),
            kind=str(data.get("kind") or ""),
            scope=str(data.get("scope") or ""),
            token=str(data.get("token") or ""),
            status=str(data.get("status") or "pending"),
            expires_at_ms=int(data.get("expires_at_ms") or 0),
            created_at_ms=int(data.get("created_at_ms") or 0),
        )


@dataclass(slots=True)
class ChannelRuntimeSpec:
    channel_id: str
    kind: str
    mode: str
    account_id: str
    display_name: str
    config: dict[str, Any] = field(default_factory=dict)
    secrets_ref: str | None = None
    external_endpoint: str | None = None


@dataclass(slots=True)
class ValidationResult:
    ok: bool
    status: str
    account_id: str | None = None
    display_name: str | None = None
    error: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)


def _optional_string(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None
  • Step 4: Implement JSON stores

Create app-instance/backend/beaver/interfaces/channels/connections/store.py:

"""Persistent channel connection stores."""

from __future__ import annotations

import json
import time
from pathlib import Path
from threading import Lock
from typing import Any
from uuid import uuid4

from .models import CONNECTION_STATUSES, ChannelConnection, PairingSession, iso_now


class ChannelConnectionStore:
    def __init__(self, path: Path) -> None:
        self.path = Path(path)
        self._lock = Lock()

    def create(
        self,
        *,
        kind: str,
        mode: str,
        display_name: str,
        account_id: str,
        owner_user_id: str | None,
        auth_type: str,
        runtime_config: dict[str, Any] | None = None,
        capabilities: list[str] | None = None,
        credentials_ref: str | None = None,
    ) -> ChannelConnection:
        with self._lock:
            data = self._load()
            connection_id = f"conn_{uuid4().hex}"
            channel_id = f"{_slug(kind)}-{uuid4().hex[:8]}"
            now = iso_now()
            connection = ChannelConnection(
                connection_id=connection_id,
                owner_user_id=owner_user_id,
                channel_id=channel_id,
                kind=kind,
                mode=mode,
                display_name=display_name or channel_id,
                account_id=account_id,
                status="draft",
                auth_type=auth_type,
                credentials_ref=credentials_ref,
                runtime_config=runtime_config or {},
                capabilities=capabilities or [],
                created_at=now,
                updated_at=now,
            )
            data["connections"][connection_id] = connection.to_dict()
            self._save(data)
            return connection

    def get(self, connection_id: str) -> ChannelConnection:
        data = self._load()
        raw = data["connections"].get(connection_id)
        if not isinstance(raw, dict):
            raise KeyError(connection_id)
        return ChannelConnection.from_dict(raw)

    def list(self) -> list[ChannelConnection]:
        data = self._load()
        return [ChannelConnection.from_dict(item) for item in data["connections"].values() if isinstance(item, dict)]

    def update(self, connection: ChannelConnection) -> ChannelConnection:
        with self._lock:
            data = self._load()
            if connection.connection_id not in data["connections"]:
                raise KeyError(connection.connection_id)
            connection.updated_at = iso_now()
            data["connections"][connection.connection_id] = connection.to_dict()
            self._save(data)
            return connection

    def update_status(self, connection_id: str, *, status: str, last_error: str | None) -> ChannelConnection:
        if status not in CONNECTION_STATUSES:
            raise ValueError(f"Unsupported connection status: {status}")
        connection = self.get(connection_id)
        connection.status = status
        connection.last_error = last_error
        if status in {"connected", "running"}:
            connection.last_seen_at = iso_now()
        return self.update(connection)

    def revoke(self, connection_id: str) -> ChannelConnection:
        return self.update_status(connection_id, status="revoked", last_error=None)

    def _load(self) -> dict[str, Any]:
        if not self.path.exists():
            return {"connections": {}}
        try:
            data = json.loads(self.path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return {"connections": {}}
        if not isinstance(data, dict) or not isinstance(data.get("connections"), dict):
            return {"connections": {}}
        return data

    def _save(self, data: dict[str, Any]) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp_path = self.path.with_name(f"{self.path.name}.tmp")
        tmp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        tmp_path.replace(self.path)


class CredentialStore:
    def __init__(self, path: Path) -> None:
        self.path = Path(path)
        self._lock = Lock()

    def put(self, *, kind: str, values: dict[str, Any]) -> str:
        cleaned = {str(key): str(value) for key, value in values.items() if str(key).strip() and str(value).strip()}
        ref = f"cred_{uuid4().hex}"
        with self._lock:
            data = self._load()
            data["credentials"][ref] = {"kind": kind, "values": cleaned, "created_at": iso_now()}
            self._save(data)
        return ref

    def get(self, ref: str) -> dict[str, str]:
        data = self._load()
        item = data["credentials"].get(ref)
        if not isinstance(item, dict):
            raise KeyError(ref)
        values = item.get("values")
        if not isinstance(values, dict):
            return {}
        return {str(key): str(value) for key, value in values.items()}

    def redacted(self, ref: str | None) -> dict[str, str]:
        if not ref:
            return {}
        try:
            values = self.get(ref)
        except KeyError:
            return {}
        return {key: "***" for key in values}

    def _load(self) -> dict[str, Any]:
        if not self.path.exists():
            return {"credentials": {}}
        try:
            data = json.loads(self.path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return {"credentials": {}}
        if not isinstance(data, dict) or not isinstance(data.get("credentials"), dict):
            return {"credentials": {}}
        return data

    def _save(self, data: dict[str, Any]) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp_path = self.path.with_name(f"{self.path.name}.tmp")
        tmp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        tmp_path.replace(self.path)


class PairingTokenStore:
    def __init__(self, path: Path) -> None:
        self.path = Path(path)
        self._lock = Lock()

    def create(self, *, kind: str, ttl_seconds: int, scope: str) -> PairingSession:
        now_ms = _now_ms()
        session = PairingSession(
            pairing_session_id=f"pair_{uuid4().hex}",
            kind=kind,
            scope=scope,
            token=f"pair_{uuid4().hex}",
            status="pending",
            expires_at_ms=now_ms + int(ttl_seconds * 1000),
            created_at_ms=now_ms,
        )
        with self._lock:
            data = self._load()
            data["sessions"][session.pairing_session_id] = session.to_dict()
            self._save(data)
        return session

    def consume(self, token: str, *, expected_kind: str) -> PairingSession | None:
        with self._lock:
            data = self._load()
            for key, raw in data["sessions"].items():
                session = PairingSession.from_dict(raw)
                if session.token != token or session.kind != expected_kind:
                    continue
                if session.status != "pending" or session.expires_at_ms <= _now_ms():
                    return None
                session.status = "consumed"
                data["sessions"][key] = session.to_dict()
                self._save(data)
                return session
        return None

    def _load(self) -> dict[str, Any]:
        if not self.path.exists():
            return {"sessions": {}}
        try:
            data = json.loads(self.path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return {"sessions": {}}
        if not isinstance(data, dict) or not isinstance(data.get("sessions"), dict):
            return {"sessions": {}}
        return data

    def _save(self, data: dict[str, Any]) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp_path = self.path.with_name(f"{self.path.name}.tmp")
        tmp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        tmp_path.replace(self.path)


def _now_ms() -> int:
    return int(time.time() * 1000)


def _slug(value: str) -> str:
    text = "".join(char if char.isalnum() else "-" for char in str(value).strip().lower())
    return "-".join(part for part in text.split("-") if part) or "channel"
  • Step 5: Export the connection package

Create app-instance/backend/beaver/interfaces/channels/connections/__init__.py:

"""Channel connection setup layer."""

from .models import ChannelConnection, ChannelRuntimeSpec, PairingSession, ValidationResult
from .store import ChannelConnectionStore, CredentialStore, PairingTokenStore

__all__ = [
    "ChannelConnection",
    "ChannelRuntimeSpec",
    "PairingSession",
    "ValidationResult",
    "ChannelConnectionStore",
    "CredentialStore",
    "PairingTokenStore",
]
  • Step 6: Run store tests to verify they pass

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connection_store.py -q

Expected: 4 passed.

  • Step 7: Commit Task 1
git add app-instance/backend/beaver/interfaces/channels/connections app-instance/backend/tests/unit/test_channel_connection_store.py
git commit -m "feat: add channel connection store"

Task 2: Connector Registry

Files:

  • Create: app-instance/backend/beaver/interfaces/channels/connections/connectors.py

  • Modify: app-instance/backend/beaver/interfaces/channels/connections/__init__.py

  • Test: app-instance/backend/tests/unit/test_channel_connector_registry.py

  • Step 1: Write failing registry tests

Create app-instance/backend/tests/unit/test_channel_connector_registry.py:

from __future__ import annotations

import asyncio

from beaver.interfaces.channels.connections import (
    ChannelConnectionStore,
    ChannelConnectorRegistry,
    ChannelRuntimeSpec,
    CredentialStore,
    ValidationResult,
)


class FakeConnector:
    kind = "fake"

    def __init__(self) -> None:
        self.validated: list[str] = []
        self.revoked: list[str] = []

    async def validate(self, connection_id: str) -> ValidationResult:
        self.validated.append(connection_id)
        return ValidationResult(ok=True, status="connected", account_id="fake-account", display_name="Fake")

    async def materialize_runtime(self, connection_id: str) -> ChannelRuntimeSpec:
        return ChannelRuntimeSpec(
            channel_id="fake-channel",
            kind="fake",
            mode="webhook",
            account_id="fake-account",
            display_name="Fake",
            config={"enabled": True},
        )

    async def revoke(self, connection_id: str) -> None:
        self.revoked.append(connection_id)
        return None


def test_connector_registry_dispatches_by_kind(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        connector = FakeConnector()
        registry = ChannelConnectorRegistry(connection_store=connection_store, credential_store=credential_store)
        registry.register(connector)

        connection = connection_store.create(
            kind="fake",
            mode="webhook",
            display_name="Fake",
            account_id="fake-account",
            owner_user_id=None,
            auth_type="token",
        )
        result = await registry.validate(connection.connection_id)
        spec = await registry.materialize_runtime(connection.connection_id)

        assert result.ok is True
        assert connector.validated == [connection.connection_id]
        assert spec.channel_id == "fake-channel"

    asyncio.run(run())


def test_connector_registry_materializes_only_connected_connections(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        registry = ChannelConnectorRegistry(connection_store=connection_store, credential_store=credential_store)
        registry.register(FakeConnector())

        draft = connection_store.create(
            kind="fake",
            mode="webhook",
            display_name="Draft",
            account_id="draft",
            owner_user_id=None,
            auth_type="token",
        )
        connected = connection_store.create(
            kind="fake",
            mode="webhook",
            display_name="Connected",
            account_id="connected",
            owner_user_id=None,
            auth_type="token",
        )
        connection_store.update_status(connected.connection_id, status="connected", last_error=None)

        specs = await registry.materialize_connected_runtime_specs()

        assert [spec.channel_id for spec in specs] == ["fake-channel"]
        assert connection_store.get(draft.connection_id).status == "draft"

    asyncio.run(run())


def test_connector_registry_revoke_calls_connector_and_updates_store(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        connector = FakeConnector()
        registry = ChannelConnectorRegistry(connection_store=connection_store, credential_store=credential_store)
        registry.register(connector)

        connection = connection_store.create(
            kind="fake",
            mode="webhook",
            display_name="Fake",
            account_id="fake-account",
            owner_user_id=None,
            auth_type="token",
        )
        connection_store.update_status(connection.connection_id, status="connected", last_error=None)

        await registry.revoke(connection.connection_id)

        assert connector.revoked == [connection.connection_id]
        assert connection_store.get(connection.connection_id).status == "revoked"

    asyncio.run(run())
  • Step 2: Run tests to verify they fail

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connector_registry.py -q

Expected: fail with ImportError: cannot import name 'ChannelConnectorRegistry'.

  • Step 3: Implement connector protocol and registry

Create app-instance/backend/beaver/interfaces/channels/connections/connectors.py:

"""Channel connector registry."""

from __future__ import annotations

from typing import Protocol

from .models import ChannelRuntimeSpec, ValidationResult
from .store import ChannelConnectionStore, CredentialStore


class ChannelConnector(Protocol):
    kind: str

    async def validate(self, connection_id: str) -> ValidationResult:
        ...

    async def materialize_runtime(self, connection_id: str) -> ChannelRuntimeSpec:
        ...

    async def revoke(self, connection_id: str) -> None:
        ...


class ChannelConnectorRegistry:
    def __init__(self, *, connection_store: ChannelConnectionStore, credential_store: CredentialStore) -> None:
        self.connection_store = connection_store
        self.credential_store = credential_store
        self._connectors: dict[str, ChannelConnector] = {}

    def register(self, connector: ChannelConnector) -> None:
        kind = connector.kind.strip()
        if not kind:
            raise ValueError("Connector kind is required")
        if kind in self._connectors:
            raise ValueError(f"Connector already registered: {kind}")
        self._connectors[kind] = connector

    def connectors(self) -> list[dict[str, str]]:
        return [{"kind": kind} for kind in sorted(self._connectors)]

    async def validate(self, connection_id: str) -> ValidationResult:
        connection = self.connection_store.get(connection_id)
        connector = self._connector(connection.kind)
        result = await connector.validate(connection_id)
        self.connection_store.update_status(
            connection_id,
            status=result.status,
            last_error=result.error,
        )
        return result

    async def materialize_runtime(self, connection_id: str) -> ChannelRuntimeSpec:
        connection = self.connection_store.get(connection_id)
        return await self._connector(connection.kind).materialize_runtime(connection_id)

    async def materialize_connected_runtime_specs(self) -> list[ChannelRuntimeSpec]:
        specs: list[ChannelRuntimeSpec] = []
        for connection in self.connection_store.list():
            if connection.status not in {"connected", "running"}:
                continue
            specs.append(await self._connector(connection.kind).materialize_runtime(connection.connection_id))
        return specs

    async def revoke(self, connection_id: str) -> None:
        connection = self.connection_store.get(connection_id)
        await self._connector(connection.kind).revoke(connection_id)
        self.connection_store.revoke(connection_id)

    def _connector(self, kind: str) -> ChannelConnector:
        connector = self._connectors.get(kind)
        if connector is None:
            raise KeyError(f"Connector not registered: {kind}")
        return connector
  • Step 4: Export registry symbols

Modify app-instance/backend/beaver/interfaces/channels/connections/__init__.py:

"""Channel connection setup layer."""

from .connectors import ChannelConnector, ChannelConnectorRegistry
from .models import ChannelConnection, ChannelRuntimeSpec, PairingSession, ValidationResult
from .store import ChannelConnectionStore, CredentialStore, PairingTokenStore

__all__ = [
    "ChannelConnector",
    "ChannelConnectorRegistry",
    "ChannelConnection",
    "ChannelRuntimeSpec",
    "PairingSession",
    "ValidationResult",
    "ChannelConnectionStore",
    "CredentialStore",
    "PairingTokenStore",
]
  • Step 5: Run registry tests

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connector_registry.py -q

Expected: 3 passed.

  • Step 6: Commit Task 2
git add app-instance/backend/beaver/interfaces/channels/connections app-instance/backend/tests/unit/test_channel_connector_registry.py
git commit -m "feat: add channel connector registry"

Task 3: Telegram Connector

Files:

  • Create: app-instance/backend/beaver/interfaces/channels/connections/telegram.py

  • Modify: app-instance/backend/beaver/interfaces/channels/connections/__init__.py

  • Test: app-instance/backend/tests/unit/test_telegram_channel_connector.py

  • Step 1: Write failing Telegram connector tests

Create app-instance/backend/tests/unit/test_telegram_channel_connector.py:

from __future__ import annotations

import asyncio

from beaver.interfaces.channels.connections import (
    ChannelConnectionStore,
    CredentialStore,
    TelegramConnector,
)


class FakeTelegramClient:
    async def get_me(self):
        return {"id": 12345, "username": "beaver_bot", "first_name": "Beaver"}


class BrokenTelegramClient:
    async def get_me(self):
        raise RuntimeError("invalid token")


def test_telegram_connector_validates_token_and_updates_connection(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        credentials_ref = credential_store.put(kind="telegram", values={"botToken": "token-1"})
        connection = connection_store.create(
            kind="telegram",
            mode="polling",
            display_name="Telegram Main",
            account_id="",
            owner_user_id="user-1",
            auth_type="token",
            credentials_ref=credentials_ref,
            runtime_config={"max_message_chars": 4096},
        )
        connector = TelegramConnector(
            connection_store=connection_store,
            credential_store=credential_store,
            client_factory=lambda token: FakeTelegramClient(),
        )

        result = await connector.validate(connection.connection_id)
        updated = connection_store.get(connection.connection_id)

        assert result.ok is True
        assert result.status == "connected"
        assert result.account_id == "telegram:12345"
        assert updated.account_id == "telegram:12345"
        assert updated.display_name == "Beaver (@beaver_bot)"
        assert updated.capabilities == ["receive_text", "send_text", "receive_media", "groups"]

    asyncio.run(run())


def test_telegram_connector_materializes_runtime_spec(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        credentials_ref = credential_store.put(kind="telegram", values={"botToken": "token-1"})
        connection = connection_store.create(
            kind="telegram",
            mode="polling",
            display_name="Telegram Main",
            account_id="telegram:12345",
            owner_user_id=None,
            auth_type="token",
            credentials_ref=credentials_ref,
            runtime_config={"max_message_chars": 4096, "require_mention_in_groups": True},
        )
        connection_store.update_status(connection.connection_id, status="connected", last_error=None)
        connector = TelegramConnector(
            connection_store=connection_store,
            credential_store=credential_store,
            client_factory=lambda token: FakeTelegramClient(),
        )

        spec = await connector.materialize_runtime(connection.connection_id)

        assert spec.channel_id == connection.channel_id
        assert spec.kind == "telegram"
        assert spec.mode == "polling"
        assert spec.account_id == "telegram:12345"
        assert spec.config["max_message_chars"] == 4096
        assert spec.config["require_mention_in_groups"] is True
        assert spec.secrets_ref == credentials_ref

    asyncio.run(run())


def test_telegram_connector_validation_failure_sets_error_status(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        credentials_ref = credential_store.put(kind="telegram", values={"botToken": "bad-token"})
        connection = connection_store.create(
            kind="telegram",
            mode="polling",
            display_name="Telegram Main",
            account_id="",
            owner_user_id=None,
            auth_type="token",
            credentials_ref=credentials_ref,
        )
        connector = TelegramConnector(
            connection_store=connection_store,
            credential_store=credential_store,
            client_factory=lambda token: BrokenTelegramClient(),
        )

        result = await connector.validate(connection.connection_id)

        assert result.ok is False
        assert result.status == "error"
        assert "invalid token" in (result.error or "")

    asyncio.run(run())


def test_telegram_connector_revoke_leaves_store_status_to_registry(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        connection = connection_store.create(
            kind="telegram",
            mode="polling",
            display_name="Telegram Main",
            account_id="telegram:12345",
            owner_user_id=None,
            auth_type="token",
        )
        connection_store.update_status(connection.connection_id, status="connected", last_error=None)
        connector = TelegramConnector(
            connection_store=connection_store,
            credential_store=credential_store,
            client_factory=lambda token: FakeTelegramClient(),
        )

        await connector.revoke(connection.connection_id)

        assert connection_store.get(connection.connection_id).status == "connected"

    asyncio.run(run())
  • Step 2: Run tests to verify they fail

Run:

cd app-instance/backend
uv run pytest tests/unit/test_telegram_channel_connector.py -q

Expected: fail with ImportError: cannot import name 'TelegramConnector'.

  • Step 3: Verify Telegram dependency

Run:

cd app-instance/backend
rg -n "python-telegram-bot" pyproject.toml uv.lock | sed -n '1,20p'

Expected output includes python-telegram-bot>=22.0,<23.0. The default client factory may use from telegram import Bot, and Bot.get_me() is awaitable in this dependency line.

  • Step 4: Implement TelegramConnector

Create app-instance/backend/beaver/interfaces/channels/connections/telegram.py:

"""Telegram channel connector."""

from __future__ import annotations

from collections.abc import Callable
from typing import Any

from .models import ChannelRuntimeSpec, ValidationResult
from .store import ChannelConnectionStore, CredentialStore


class TelegramConnector:
    kind = "telegram"

    def __init__(
        self,
        *,
        connection_store: ChannelConnectionStore,
        credential_store: CredentialStore,
        client_factory: Callable[[str], Any] | None = None,
    ) -> None:
        self.connection_store = connection_store
        self.credential_store = credential_store
        self.client_factory = client_factory or _default_client_factory

    async def validate(self, connection_id: str) -> ValidationResult:
        connection = self.connection_store.get(connection_id)
        token = self._bot_token(connection.credentials_ref)
        try:
            client = self.client_factory(token)
            raw = await client.get_me()
            bot_id = _value(raw, "id")
            username = _value(raw, "username")
            first_name = _value(raw, "first_name") or "Telegram Bot"
            account_id = f"telegram:{bot_id}" if bot_id else connection.account_id
            display_name = f"{first_name} (@{username})" if username else first_name
            connection.account_id = account_id
            connection.display_name = display_name
            connection.capabilities = ["receive_text", "send_text", "receive_media", "groups"]
            self.connection_store.update(connection)
            return ValidationResult(
                ok=True,
                status="connected",
                account_id=account_id,
                display_name=display_name,
                metadata={"username": username} if username else {},
            )
        except Exception as exc:
            return ValidationResult(ok=False, status="error", error=str(exc))

    async def materialize_runtime(self, connection_id: str) -> ChannelRuntimeSpec:
        connection = self.connection_store.get(connection_id)
        if connection.status not in {"connected", "running"}:
            raise ValueError(f"Connection is not connected: {connection.connection_id}")
        return ChannelRuntimeSpec(
            channel_id=connection.channel_id,
            kind=connection.kind,
            mode=connection.mode,
            account_id=connection.account_id,
            display_name=connection.display_name,
            config=dict(connection.runtime_config),
            secrets_ref=connection.credentials_ref,
        )

    async def revoke(self, connection_id: str) -> None:
        # Telegram bot tokens do not have a Beaver-managed platform revoke action.
        # The registry owns local connection state transitions.
        return None

    def _bot_token(self, credentials_ref: str | None) -> str:
        if not credentials_ref:
            raise ValueError("Telegram credentials are missing")
        token = self.credential_store.get(credentials_ref).get("botToken")
        if not token:
            raise ValueError("botToken is required")
        return token


def _value(raw: Any, key: str) -> str:
    if isinstance(raw, dict):
        value = raw.get(key)
    else:
        value = getattr(raw, key, None)
    return str(value).strip() if value is not None else ""


def _default_client_factory(token: str) -> Any:
    try:
        from telegram import Bot
    except ImportError as exc:  # pragma: no cover - optional live dependency
        raise RuntimeError("Install beaver-backend[telegram] to validate Telegram connections") from exc
    return Bot(token=token)
  • Step 5: Export TelegramConnector

Modify app-instance/backend/beaver/interfaces/channels/connections/__init__.py:

"""Channel connection setup layer."""

from .connectors import ChannelConnector, ChannelConnectorRegistry
from .models import ChannelConnection, ChannelRuntimeSpec, PairingSession, ValidationResult
from .store import ChannelConnectionStore, CredentialStore, PairingTokenStore
from .telegram import TelegramConnector

__all__ = [
    "ChannelConnector",
    "ChannelConnectorRegistry",
    "ChannelConnection",
    "ChannelRuntimeSpec",
    "PairingSession",
    "ValidationResult",
    "ChannelConnectionStore",
    "CredentialStore",
    "PairingTokenStore",
    "TelegramConnector",
]
  • Step 6: Run Telegram connector tests

Run:

cd app-instance/backend
uv run pytest tests/unit/test_telegram_channel_connector.py -q

Expected: 4 passed.

  • Step 7: Commit Task 3
git add app-instance/backend/beaver/interfaces/channels/connections app-instance/backend/tests/unit/test_telegram_channel_connector.py
git commit -m "feat: add telegram channel connector"

Task 4: Runtime Materialization From Connections

Files:

  • Modify: app-instance/backend/beaver/interfaces/web/app.py

  • Test: app-instance/backend/tests/unit/test_channel_connector_registry.py

  • Step 1: Verify ChannelConfig fields

Run:

cd app-instance/backend
uv run python - <<'PY'
from dataclasses import fields
from beaver.foundation.config.schema import ChannelConfig
print([field.name for field in fields(ChannelConfig)])
PY

Expected output includes enabled, kind, mode, account_id, display_name, config, and secrets.

  • Step 2: Extend registry tests for ChannelConfig materialization

Append to app-instance/backend/tests/unit/test_channel_connector_registry.py:

from beaver.foundation.config.schema import ChannelConfig


def test_connector_registry_materializes_channel_configs_with_credentials(tmp_path) -> None:
    async def run() -> None:
        connection_store = ChannelConnectionStore(tmp_path / "connections.json")
        credential_store = CredentialStore(tmp_path / "credentials.json")
        credentials_ref = credential_store.put(kind="telegram", values={"botToken": "token-1"})
        connection = connection_store.create(
            kind="fake",
            mode="webhook",
            display_name="Connected",
            account_id="connected",
            owner_user_id=None,
            auth_type="token",
            credentials_ref=credentials_ref,
        )
        connection_store.update_status(connection.connection_id, status="connected", last_error=None)

        class CredentialAwareConnector(FakeConnector):
            async def materialize_runtime(self, connection_id: str) -> ChannelRuntimeSpec:
                stored = connection_store.get(connection_id)
                return ChannelRuntimeSpec(
                    channel_id="fake-channel",
                    kind="fake",
                    mode="webhook",
                    account_id="fake-account",
                    display_name="Fake",
                    config={"enabled": True},
                    secrets_ref=stored.credentials_ref,
                )

        registry = ChannelConnectorRegistry(connection_store=connection_store, credential_store=credential_store)
        registry.register(CredentialAwareConnector())

        configs = await registry.materialize_channel_configs()

        assert isinstance(configs["fake-channel"], ChannelConfig)
        assert configs["fake-channel"].enabled is True
        assert configs["fake-channel"].secrets == {"botToken": "token-1"}

    asyncio.run(run())
  • Step 3: Run registry tests to verify failure

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connector_registry.py::test_connector_registry_materializes_channel_configs_with_credentials -q

Expected: fail with AttributeError: 'ChannelConnectorRegistry' object has no attribute 'materialize_channel_configs'.

  • Step 4: Implement channel config materialization

Modify app-instance/backend/beaver/interfaces/channels/connections/connectors.py:

from beaver.foundation.config.schema import ChannelConfig

Add this method to ChannelConnectorRegistry:

    async def materialize_channel_configs(self) -> dict[str, ChannelConfig]:
        channels: dict[str, ChannelConfig] = {}
        for spec in await self.materialize_connected_runtime_specs():
            secrets = self.credential_store.get(spec.secrets_ref) if spec.secrets_ref else {}
            channels[spec.channel_id] = ChannelConfig(
                enabled=True,
                kind=spec.kind,
                mode=spec.mode,
                account_id=spec.account_id,
                display_name=spec.display_name,
                config=dict(spec.config),
                secrets=secrets,
            )
        return channels
  • Step 5: Add app helpers for connection state paths and registry construction

Modify app-instance/backend/beaver/interfaces/web/app.py imports:

from beaver.interfaces.channels.connections import (
    ChannelConnectionStore,
    ChannelConnectorRegistry,
    CredentialStore,
    TelegramConnector,
)

Add helper functions near get_channel_runtime():

def _connection_state_dir(workspace: Path) -> Path:
    return Path(workspace) / "state" / "channel_connections"


def _build_channel_connector_registry(workspace: Path) -> ChannelConnectorRegistry:
    state_dir = _connection_state_dir(workspace)
    connection_store = ChannelConnectionStore(state_dir / "connections.json")
    credential_store = CredentialStore(state_dir / "credentials.json")
    registry = ChannelConnectorRegistry(connection_store=connection_store, credential_store=credential_store)
    registry.register(
        TelegramConnector(
            connection_store=connection_store,
            credential_store=credential_store,
        )
    )
    return registry
  • Step 6: Merge materialized connections into runtime startup

Modify the lifespan block in app-instance/backend/beaver/interfaces/web/app.py where ChannelRuntime is created:

        loaded = attached_service.create_loop().boot()
        app.state.channel_connection_workspace = loaded.workspace
        connector_registry = _build_channel_connector_registry(loaded.workspace)
        app.state.channel_connector_registry = connector_registry
        connection_channels = await connector_registry.materialize_channel_configs()
        runtime_channels = dict(loaded.config.channels)
        runtime_channels.update(connection_channels)
        channel_runtime = ChannelRuntime(
            service=attached_service,
            workspace=loaded.workspace,
            channels=runtime_channels,
        )

Keep app.state.channel_connector_registry = connector_registry before runtime startup so API handlers can reuse the same stores.

  • Step 7: Run registry tests

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connector_registry.py -q

Expected: all tests pass.

  • Step 8: Commit Task 4
git add app-instance/backend/beaver/interfaces/channels/connections/connectors.py app-instance/backend/beaver/interfaces/web/app.py app-instance/backend/tests/unit/test_channel_connector_registry.py
git commit -m "feat: materialize channel connections into runtime config"

Task 5: Connection Control API

Files:

  • Modify: app-instance/backend/beaver/interfaces/web/schemas/chat.py

  • Modify: app-instance/backend/beaver/interfaces/web/schemas/__init__.py

  • Modify: app-instance/backend/beaver/interfaces/web/app.py

  • Test: app-instance/backend/tests/unit/test_channel_connection_api.py

  • Step 1: Write failing API tests

Create app-instance/backend/tests/unit/test_channel_connection_api.py:

from __future__ import annotations

from fastapi.testclient import TestClient

from beaver.interfaces.web.app import create_app
from beaver.services.agent_service import AgentService


def test_channel_connection_api_creates_updates_lists_and_revokes(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    config_path.write_text('{"agents": {"defaults": {"workspace": "%s"}}, "providers": {}}' % str(tmp_path), encoding="utf-8")
    service = AgentService(config_path=config_path)
    app = create_app(service=service, manage_service_lifecycle=False)

    try:
        with TestClient(app) as client:
            created = client.post(
                "/api/channel-connections",
                json={
                    "kind": "telegram",
                    "mode": "polling",
                    "displayName": "Telegram Main",
                    "authType": "token",
                    "secrets": {"botToken": "token-1"},
                    "config": {"maxMessageChars": 4096, "requireMentionInGroups": True},
                },
            )
            assert created.status_code == 200
            body = created.json()
            connection_id = body["connection"]["connection_id"]
            assert body["connection"]["kind"] == "telegram"
            assert body["connection"]["status"] == "draft"
            assert "credentials_ref" not in body["connection"]
            assert body["connection"]["runtime_config"] == {
                "max_message_chars": 4096,
                "require_mention_in_groups": True,
            }
            assert body["credentials"] == {"botToken": "***"}

            patched = client.patch(
                f"/api/channel-connections/{connection_id}",
                json={
                    "displayName": "Telegram Ops",
                    "config": {"maxMessageChars": 2048},
                    "secrets": {"botToken": "token-2"},
                },
            )
            assert patched.status_code == 200
            assert patched.json()["connection"]["display_name"] == "Telegram Ops"
            assert patched.json()["connection"]["runtime_config"] == {"max_message_chars": 2048}
            assert patched.json()["credentials"] == {"botToken": "***"}

            listed = client.get("/api/channel-connections")
            assert listed.status_code == 200
            assert listed.json()[0]["connection_id"] == connection_id
            assert "credentials_ref" not in listed.json()[0]

            revoked = client.post(f"/api/channel-connections/{connection_id}/revoke")
            assert revoked.status_code == 200
            assert revoked.json()["connection"]["status"] == "revoked"
    finally:
        service.close()


def test_channel_connectors_api_lists_registered_connectors(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    config_path.write_text('{"agents": {"defaults": {"workspace": "%s"}}, "providers": {}}' % str(tmp_path), encoding="utf-8")
    service = AgentService(config_path=config_path)
    app = create_app(service=service, manage_service_lifecycle=False)

    try:
        with TestClient(app) as client:
            response = client.get("/api/channel-connectors")
    finally:
        service.close()

    assert response.status_code == 200
    assert response.json() == [{"kind": "telegram"}]
  • Step 2: Run API tests to verify failure

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connection_api.py -q

Expected: fail with 404 Not Found for /api/channel-connections.

  • Step 3: Add web schemas

Append to app-instance/backend/beaver/interfaces/web/schemas/chat.py after WebChannelConfigResponse:

class WebChannelConnectionCreateRequest(BaseModel):
    """Create a channel connection from the setup UI."""

    kind: str
    mode: str
    display_name: str | None = Field(default=None, alias="displayName")
    owner_user_id: str | None = Field(default=None, alias="ownerUserId")
    auth_type: str = Field(default="token", alias="authType")
    account_id: str | None = Field(default=None, alias="accountId")
    config: dict[str, Any] = Field(default_factory=dict)
    secrets: dict[str, str | None] = Field(default_factory=dict)


class WebChannelConnectionResponse(BaseModel):
    """Channel connection response with redacted credentials."""

    connection: dict[str, Any]
    credentials: dict[str, str] = Field(default_factory=dict)


class WebChannelConnectionUpdateRequest(BaseModel):
    """Update editable channel connection setup fields."""

    display_name: str | None = Field(default=None, alias="displayName")
    account_id: str | None = Field(default=None, alias="accountId")
    config: dict[str, Any] | None = None
    secrets: dict[str, str | None] | None = None


class WebChannelValidationResponse(BaseModel):
    """Connector validation response."""

    ok: bool
    status: str
    account_id: str | None = None
    display_name: str | None = None
    error: str | None = None
    metadata: dict[str, Any] = Field(default_factory=dict)
    connection: dict[str, Any]
  • Step 4: Export web schemas

Modify app-instance/backend/beaver/interfaces/web/schemas/__init__.py imports and __all__ to include:

    WebChannelConnectionCreateRequest,
    WebChannelConnectionResponse,
    WebChannelConnectionUpdateRequest,
    WebChannelValidationResponse,
  • Step 5: Add connector registry accessors to app.py

Modify imports in app-instance/backend/beaver/interfaces/web/app.py:

from beaver.interfaces.web.schemas import (
    WebChannelConnectionCreateRequest,
    WebChannelConnectionResponse,
    WebChannelConnectionUpdateRequest,
    WebChannelValidationResponse,
)

Add helper:

def get_channel_connector_registry(request: Request) -> ChannelConnectorRegistry:
    registry = getattr(request.app.state, "channel_connector_registry", None)
    if not isinstance(registry, ChannelConnectorRegistry):
        workspace = getattr(request.app.state, "channel_connection_workspace", None)
        if workspace is None:
            raise RuntimeError("Channel connector registry unavailable before service boot")
        registry = _build_channel_connector_registry(workspace)
        request.app.state.channel_connector_registry = registry
    return registry


def _normalize_connection_config(config: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(config, dict):
        return {}
    return {
        _camel_to_snake_text(str(key)): value
        for key, value in config.items()
        if str(key).strip()
    }


def _camel_to_snake_text(value: str) -> str:
    result: list[str] = []
    for char in value:
        if char.isupper() and result:
            result.append("_")
        result.append(char.lower())
    return "".join(result)


def _connection_response_view(connection: Any) -> dict[str, Any]:
    view = connection.to_dict()
    view.pop("credentials_ref", None)
    view.pop("connector_ref", None)
    view.pop("pairing_session_id", None)
    return view
  • Step 6: Add connection API routes

Add routes near existing /api/channels routes in app-instance/backend/beaver/interfaces/web/app.py:

    @app.get("/api/channel-connectors")
    async def list_channel_connectors(request: Request) -> list[dict[str, str]]:
        return get_channel_connector_registry(request).connectors()

    @app.get("/api/channel-connections")
    async def list_channel_connections(request: Request) -> list[dict[str, Any]]:
        registry = get_channel_connector_registry(request)
        return [_connection_response_view(connection) for connection in registry.connection_store.list()]

    @app.post("/api/channel-connections", response_model=WebChannelConnectionResponse)
    async def create_channel_connection(
        request: Request,
        payload: WebChannelConnectionCreateRequest,
    ) -> WebChannelConnectionResponse:
        registry = get_channel_connector_registry(request)
        kind = _clean_text(payload.kind)
        mode = _clean_text(payload.mode)
        if not kind:
            raise HTTPException(status_code=400, detail="Connection kind is required")
        if not mode:
            raise HTTPException(status_code=400, detail="Connection mode is required")
        secrets = {key: value for key, value in payload.secrets.items() if value}
        credentials_ref = registry.credential_store.put(kind=kind, values=secrets) if secrets else None
        connection = registry.connection_store.create(
            kind=kind,
            mode=mode,
            display_name=_clean_text(payload.display_name) or kind,
            account_id=_clean_text(payload.account_id) or "",
            owner_user_id=_clean_text(payload.owner_user_id) or None,
            auth_type=_clean_text(payload.auth_type) or "token",
            credentials_ref=credentials_ref,
            runtime_config=_normalize_connection_config(payload.config),
        )
        return WebChannelConnectionResponse(
            connection=_connection_response_view(connection),
            credentials=registry.credential_store.redacted(credentials_ref),
        )

    @app.patch("/api/channel-connections/{connection_id}", response_model=WebChannelConnectionResponse)
    async def update_channel_connection(
        connection_id: str,
        request: Request,
        payload: WebChannelConnectionUpdateRequest,
    ) -> WebChannelConnectionResponse:
        registry = get_channel_connector_registry(request)
        try:
            connection = registry.connection_store.get(connection_id)
        except KeyError:
            raise HTTPException(status_code=404, detail="Channel connection not found")
        if payload.display_name is not None:
            connection.display_name = _clean_text(payload.display_name) or connection.display_name
        if payload.account_id is not None:
            connection.account_id = _clean_text(payload.account_id) or connection.account_id
        if payload.config is not None:
            connection.runtime_config = _normalize_connection_config(payload.config)
        if payload.secrets:
            secrets = {key: value for key, value in payload.secrets.items() if value}
            if secrets:
                # TODO: add credential GC when connection updates credentials.
                connection.credentials_ref = registry.credential_store.put(kind=connection.kind, values=secrets)
        connection = registry.connection_store.update(connection)
        return WebChannelConnectionResponse(
            connection=_connection_response_view(connection),
            credentials=registry.credential_store.redacted(connection.credentials_ref),
        )

    @app.get("/api/channel-connections/{connection_id}", response_model=WebChannelConnectionResponse)
    async def get_channel_connection(connection_id: str, request: Request) -> WebChannelConnectionResponse:
        registry = get_channel_connector_registry(request)
        try:
            connection = registry.connection_store.get(connection_id)
        except KeyError:
            raise HTTPException(status_code=404, detail="Channel connection not found")
        return WebChannelConnectionResponse(
            connection=_connection_response_view(connection),
            credentials=registry.credential_store.redacted(connection.credentials_ref),
        )

    @app.post("/api/channel-connections/{connection_id}/validate", response_model=WebChannelValidationResponse)
    async def validate_channel_connection(connection_id: str, request: Request) -> WebChannelValidationResponse:
        registry = get_channel_connector_registry(request)
        try:
            result = await registry.validate(connection_id)
            connection = registry.connection_store.get(connection_id)
        except KeyError:
            raise HTTPException(status_code=404, detail="Channel connection not found")
        return WebChannelValidationResponse(
            ok=result.ok,
            status=result.status,
            account_id=result.account_id,
            display_name=result.display_name,
            error=result.error,
            metadata=result.metadata,
            connection=_connection_response_view(connection),
        )

    @app.post("/api/channel-connections/{connection_id}/revoke", response_model=WebChannelConnectionResponse)
    async def revoke_channel_connection(connection_id: str, request: Request) -> WebChannelConnectionResponse:
        registry = get_channel_connector_registry(request)
        try:
            await registry.revoke(connection_id)
            connection = registry.connection_store.get(connection_id)
        except KeyError:
            raise HTTPException(status_code=404, detail="Channel connection not found")
        return WebChannelConnectionResponse(connection=_connection_response_view(connection), credentials={})
  • Step 7: Run API tests

Run:

cd app-instance/backend
uv run pytest tests/unit/test_channel_connection_api.py -q

Expected: 2 passed.

  • Step 8: Run focused backend tests

Run:

cd app-instance/backend
uv run pytest \
  tests/unit/test_channel_connection_store.py \
  tests/unit/test_channel_connector_registry.py \
  tests/unit/test_telegram_channel_connector.py \
  tests/unit/test_channel_connection_api.py \
  -q

Expected: all focused connector tests pass.

  • Step 9: Commit Task 5
git add \
  app-instance/backend/beaver/interfaces/web/app.py \
  app-instance/backend/beaver/interfaces/web/schemas/chat.py \
  app-instance/backend/beaver/interfaces/web/schemas/__init__.py \
  app-instance/backend/tests/unit/test_channel_connection_api.py
git commit -m "feat: add channel connection control api"

Task 6: Final Verification And Spec Alignment

Files:

  • Review: docs/superpowers/specs/2026-06-02-channel-connectors-and-pairing-design.md

  • Review: docs/superpowers/specs/2026-06-02-chat-platform-channel-adapters-design.md

  • Review: docs/superpowers/specs/2026-06-01-terminal-websocket-channel-design.md

  • Step 1: Run connector and existing channel tests

Run:

cd app-instance/backend
uv run pytest \
  tests/unit/test_channel_connection_store.py \
  tests/unit/test_channel_connector_registry.py \
  tests/unit/test_telegram_channel_connector.py \
  tests/unit/test_channel_connection_api.py \
  tests/unit/test_channel_runtime.py \
  tests/unit/test_telegram_channel_adapter.py \
  -q

Expected: all listed tests pass.

  • Step 2: Run import tests

Run:

cd app-instance/backend
uv run pytest tests/unit/test_imports.py -q

Expected: all import tests pass.

  • Step 3: Scan for leaked secret values in connector events and responses

Run:

cd app-instance/backend
rg -n "token-1|token-2|bad-token|secret-token" tests/unit beaver || true

Expected: test fixture strings only appear in test files. They must not appear in implementation files or generated event log code.

  • Step 4: Update adapter spec wording if still contradictory

If docs/superpowers/specs/2026-06-02-chat-platform-channel-adapters-design.md still says pairing is out of scope and Node sidecars are disallowed, change only the Non-Goals and Access Control text:

- Use internal adapters by default, but allow external connector processes where platform SDK or login state requires them.

Pairing is owned by the connector layer. Platform adapters assume a materialized `ChannelConnection` and adapter-ready runtime config.
  • Step 5: Commit spec alignment if changed

If Step 4 changed docs:

git add docs/superpowers/specs/2026-06-02-chat-platform-channel-adapters-design.md
git commit -m "docs: align channel adapter spec with connector layer"

If Step 4 made no change, do not create an empty commit.

  • Step 6: Summarize remaining rollout

Record in the final implementation response that this first plan does not implement Terminal pairing, Feishu/Lark connector, Weixin external connector, QQBot connector, frontend wizard, or hot adapter restart. Those are separate plans.