feat: add slack multi-workspace oauth support

connector/config.py

@@ -102,6 +102,23 @@ class ConnectorConfig:
     slack_reply_in_thread: bool = True
     slack_mode: str = "events"  # F57: events | socket
     slack_app_token: Optional[str] = None  # F57: required in socket mode (xapp-...)
+    slack_client_id: Optional[str] = None
+    slack_client_secret: Optional[str] = None
+    slack_oauth_redirect_uri: Optional[str] = None
+    slack_oauth_install_path: str = "/slack/install"
+    slack_oauth_callback_path: str = "/slack/oauth/callback"
+    slack_oauth_scopes: List[str] = field(
+        default_factory=lambda: [
+            "app_mentions:read",
+            "channels:history",
+            "chat:write",
+            "files:write",
+            "groups:history",
+            "im:history",
+            "mpim:history",
+        ]
+    )
+    slack_oauth_state_ttl_sec: int = 600
 
     # Privileged Access (ID match across platforms; Telegram Int vs Discord Str handled by router)
     admin_users: List[str] = field(default_factory=list)
@@ -280,6 +297,28 @@ def load_config() -> ConnectorConfig:
         cfg.slack_reply_in_thread = False
     cfg.slack_mode = os.environ.get("OPENCLAW_CONNECTOR_SLACK_MODE", "events").lower()
     cfg.slack_app_token = os.environ.get("OPENCLAW_CONNECTOR_SLACK_APP_TOKEN")
+    cfg.slack_client_id = os.environ.get("OPENCLAW_CONNECTOR_SLACK_CLIENT_ID")
+    cfg.slack_client_secret = os.environ.get("OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET")
+    cfg.slack_oauth_redirect_uri = os.environ.get(
+        "OPENCLAW_CONNECTOR_SLACK_OAUTH_REDIRECT_URI"
+    )
+    cfg.slack_oauth_install_path = os.environ.get(
+        "OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH", "/slack/install"
+    )
+    cfg.slack_oauth_callback_path = os.environ.get(
+        "OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH", "/slack/oauth/callback"
+    )
+    if slack_scopes := os.environ.get("OPENCLAW_CONNECTOR_SLACK_OAUTH_SCOPES"):
+        parsed_scopes = [
+            scope.strip() for scope in slack_scopes.split(",") if scope.strip()
+        ]
+        if parsed_scopes:
+            cfg.slack_oauth_scopes = parsed_scopes
+    if slack_oauth_ttl := os.environ.get(
+        "OPENCLAW_CONNECTOR_SLACK_OAUTH_STATE_TTL_SEC"
+    ):
+        if slack_oauth_ttl.isdigit():
+            cfg.slack_oauth_state_ttl_sec = max(60, int(slack_oauth_ttl))
 
     # Admin
     if admins := os.environ.get("OPENCLAW_CONNECTOR_ADMIN_USERS"):

connector/contract.py

@@ -4,7 +4,7 @@ Shared data models for request/response.
 """
 
 from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 
 
 @dataclass
@@ -16,6 +16,9 @@ class CommandRequest:
     message_id: str
     text: str
     timestamp: float
+    workspace_id: str = ""
+    thread_id: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
@@ -42,10 +45,16 @@ class Platform:
         image_data: bytes,
         filename: str = "image.png",
         caption: Optional[str] = None,
+        delivery_context: Optional[Dict[str, Any]] = None,
     ):
         """Send an image to the channel."""
         pass
 
-    async def send_message(self, channel_id: str, text: str):
+    async def send_message(
+        self,
+        channel_id: str,
+        text: str,
+        delivery_context: Optional[Dict[str, Any]] = None,
+    ):
         """Send a text message to the channel."""
         pass

connector/platforms/discord_gateway.py

@@ -230,6 +230,7 @@ class DiscordGateway:
         image_data: bytes,
         filename: str = "image.png",
         caption: Optional[str] = None,
+        delivery_context: Optional[dict] = None,
     ):
         """Send image via Discord API."""
         if not self.session:
@@ -279,7 +280,12 @@ class DiscordGateway:
             logger.error(f"Discord send_image error: {e}")
             raise
 
-    async def send_message(self, channel_id: str, text: str):
+    async def send_message(
+        self,
+        channel_id: str,
+        text: str,
+        delivery_context: Optional[dict] = None,
+    ):
         """Send text message."""
         if not self.session:
             return

connector/platforms/line_webhook.py

@@ -286,6 +286,7 @@ class LINEWebhookServer:
         image_data: bytes,
         filename: str = "image.png",
         caption: Optional[str] = None,
+        delivery_context: Optional[dict] = None,
     ):
         """
         Send image via LINE using public URL.
@@ -368,7 +369,12 @@ class LINEWebhookServer:
                 err = await resp.text()
                 logger.error(f"LINE image push failed: {resp.status} {err}")
 
-    async def send_message(self, channel_id: str, text: str):
+    async def send_message(
+        self,
+        channel_id: str,
+        text: str,
+        delivery_context: Optional[dict] = None,
+    ):
         """Send push message."""
         if self._session_invalid:
             logger.warning("R93: Connector session invalid - blocking outbound message")

connector/platforms/slack_installation_manager.py

@@ -0,0 +1,379 @@
+from __future__ import annotations
+
+import json
+import logging
+import os
+import secrets
+import threading
+import time
+from pathlib import Path
+from typing import Any, Dict, Optional, Tuple
+from urllib.parse import urlencode
+
+from connector.config import ConnectorConfig
+
+try:
+    from services.audit import emit_audit_event
+    from services.connector_installation_registry import (
+        ConnectorInstallation,
+        ConnectorInstallationRegistry,
+        InstallationResolution,
+        InstallationStatus,
+        get_connector_installation_registry,
+    )
+    from services.secret_store import SecretStore, get_secret_store
+    from services.state_dir import get_state_dir
+except ImportError:  # pragma: no cover
+    from services.audit import emit_audit_event  # type: ignore
+    from services.connector_installation_registry import (  # type: ignore
+        ConnectorInstallation,
+        ConnectorInstallationRegistry,
+        InstallationResolution,
+        InstallationStatus,
+        get_connector_installation_registry,
+    )
+    from services.secret_store import SecretStore, get_secret_store  # type: ignore
+    from services.state_dir import get_state_dir  # type: ignore
+
+logger = logging.getLogger(__name__)
+
+SLACK_AUTHORIZE_URL = "https://slack.com/oauth/v2/authorize"
+SLACK_OAUTH_ACCESS_URL = "https://slack.com/api/oauth.v2.access"
+SLACK_OAUTH_STATE_FILE = "slack_oauth_states.json"
+
+_INVALID_TOKEN_ERRORS = frozenset(
+    {"account_inactive", "invalid_auth", "not_authed", "token_revoked"}
+)
+_DEGRADED_TOKEN_ERRORS = frozenset({"ratelimited", "request_timeout", "fatal_error"})
+
+
+def _load_aiohttp():
+    try:
+        import aiohttp  # type: ignore
+    except ModuleNotFoundError:
+        return None
+    return aiohttp
+
+
+class SlackInstallationManager:
+    def __init__(
+        self,
+        config: ConnectorConfig,
+        *,
+        registry: Optional[ConnectorInstallationRegistry] = None,
+        secret_store: Optional[SecretStore] = None,
+        state_dir: Optional[str] = None,
+    ):
+        self.config = config
+        self._state_dir = Path(state_dir or get_state_dir())
+        self._state_path = self._state_dir / SLACK_OAUTH_STATE_FILE
+        self._registry = registry or get_connector_installation_registry(
+            state_dir=str(self._state_dir)
+        )
+        self._secret_store = secret_store or get_secret_store(str(self._state_dir))
+        self._lock = threading.RLock()
+        self._states: Dict[str, Dict[str, Any]] = {}
+        self._load_states()
+
+    @property
+    def oauth_enabled(self) -> bool:
+        return bool(self.config.slack_client_id and self.config.slack_client_secret)
+
+    def resolve_redirect_uri(self) -> str:
+        if self.config.slack_oauth_redirect_uri:
+            return str(self.config.slack_oauth_redirect_uri).strip()
+        if self.config.public_base_url:
+            base = self.config.public_base_url.rstrip("/")
+            path = self.config.slack_oauth_callback_path or "/slack/oauth/callback"
+            return f"{base}{path}"
+        return ""
+
+    def can_handle_oauth(self) -> bool:
+        return self.oauth_enabled and bool(self.resolve_redirect_uri())
+
+    def _save_states(self) -> None:
+        self._state_dir.mkdir(parents=True, exist_ok=True)
+        temp_path = self._state_path.with_suffix(".tmp")
+        temp_path.write_text(json.dumps(self._states, indent=2), encoding="utf-8")
+        os.replace(temp_path, self._state_path)
+
+    def _load_states(self) -> None:
+        if not self._state_path.exists():
+            return
+        try:
+            data = json.loads(self._state_path.read_text(encoding="utf-8"))
+            if isinstance(data, dict):
+                self._states = data
+        except Exception as exc:
+            logger.warning("Failed to load Slack OAuth state store: %s", exc)
+            self._states = {}
+        self._prune_expired_states()
+
+    def _prune_expired_states(self) -> None:
+        now = time.time()
+        ttl = max(60, int(self.config.slack_oauth_state_ttl_sec or 600))
+        changed = False
+        for key, payload in list(self._states.items()):
+            created_at = float(payload.get("created_at", 0) or 0)
+            if not created_at or (now - created_at) > ttl:
+                self._states.pop(key, None)
+                changed = True
+        if changed:
+            self._save_states()
+
+    def issue_install_state(self) -> str:
+        if not self.can_handle_oauth():
+            raise RuntimeError("Slack OAuth flow not configured")
+        with self._lock:
+            self._prune_expired_states()
+            state = secrets.token_urlsafe(32)
+            self._states[state] = {"created_at": time.time()}
+            self._save_states()
+            return state
+
+    def consume_install_state(self, state: str) -> bool:
+        with self._lock:
+            self._prune_expired_states()
+            payload = self._states.pop(str(state or "").strip(), None)
+            if payload is None:
+                return False
+            self._save_states()
+            return True
+
+    def build_install_url(self, state: str) -> str:
+        params = {
+            "client_id": self.config.slack_client_id or "",
+            "scope": ",".join(self.config.slack_oauth_scopes or []),
+            "redirect_uri": self.resolve_redirect_uri(),
+            "state": state,
+        }
+        return f"{SLACK_AUTHORIZE_URL}?{urlencode(params)}"
+
+    async def exchange_code(self, code: str) -> Dict[str, Any]:
+        aiohttp = _load_aiohttp()
+        if aiohttp is None:
+            raise RuntimeError("aiohttp required for Slack OAuth exchange")
+        payload = {
+            "client_id": self.config.slack_client_id or "",
+            "client_secret": self.config.slack_client_secret or "",
+            "code": str(code or "").strip(),
+            "redirect_uri": self.resolve_redirect_uri(),
+        }
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                SLACK_OAUTH_ACCESS_URL,
+                data=payload,
+                timeout=aiohttp.ClientTimeout(total=15),
+            ) as resp:
+                data = await resp.json(content_type=None)
+                if resp.status != 200 or not data.get("ok"):
+                    raise RuntimeError(
+                        f"slack_oauth_exchange_failed:{resp.status}:{data.get('error', 'unknown')}"
+                    )
+                return data
+
+    def _normalize_workspace_id(self, payload: Dict[str, Any]) -> str:
+        workspace_id = (
+            (payload.get("team") or {}).get("id")
+            or payload.get("team_id")
+            or (
+                (payload.get("enterprise") or {}).get("id")
+                if payload.get("enterprise")
+                else ""
+            )
+        )
+        workspace_id = str(workspace_id or "").strip()
+        if not workspace_id:
+            raise ValueError("workspace_id_missing")
+        return workspace_id
+
+    def installation_id_for_workspace(self, workspace_id: str) -> str:
+        return f"slack:{str(workspace_id or '').strip()}"
+
+    def metadata_from_oauth_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        team = dict(payload.get("team", {}) or {})
+        enterprise = dict(payload.get("enterprise", {}) or {})
+        authed_user = dict(payload.get("authed_user", {}) or {})
+        metadata = {
+            "workspace_name": str(team.get("name", "") or "").strip(),
+            "enterprise_id": str(enterprise.get("id", "") or "").strip(),
+            "enterprise_name": str(enterprise.get("name", "") or "").strip(),
+            "bot_user_id": str(payload.get("bot_user_id", "") or "").strip(),
+            "app_id": str(payload.get("app_id", "") or "").strip(),
+            "scope": str(payload.get("scope", "") or "").strip(),
+            "authed_user_id": str(authed_user.get("id", "") or "").strip(),
+            "token_type": str(payload.get("token_type", "") or "").strip(),
+            "transport_mode": self.config.slack_mode,
+        }
+        return {key: value for key, value in metadata.items() if value}
+
+    def upsert_from_oauth_payload(
+        self, payload: Dict[str, Any]
+    ) -> ConnectorInstallation:
+        workspace_id = self._normalize_workspace_id(payload)
+        installation_id = self.installation_id_for_workspace(workspace_id)
+        token_values = {"bot_token": str(payload.get("access_token", "") or "").strip()}
+        if self.config.slack_app_token:
+            token_values["app_token"] = self.config.slack_app_token
+        if not token_values["bot_token"]:
+            raise ValueError("bot_token_missing")
+
+        metadata = self.metadata_from_oauth_payload(payload)
+        existing = self._registry.get_installation(installation_id)
+        if existing is not None:
+            rotated = self._registry.rotate_installation_tokens(
+                installation_id,
+                token_values,
+                reason="slack_oauth_reinstall",
+            )
+            inst = self._registry.upsert_installation(
+                platform="slack",
+                workspace_id=workspace_id,
+                installation_id=installation_id,
+                token_refs=rotated.token_refs,
+                status=rotated.status,
+                metadata=metadata,
+                status_reason="slack_oauth_reinstall",
+            )
+        else:
+            inst = self._registry.upsert_installation(
+                platform="slack",
+                workspace_id=workspace_id,
+                installation_id=installation_id,
+                token_values=token_values,
+                status=InstallationStatus.CREATED.value,
+                metadata=metadata,
+                status_reason="slack_oauth_install",
+            )
+        inst = self._registry.activate_installation(
+            installation_id, reason="slack_oauth_complete"
+        )
+        inst = self._registry.update_installation_health(
+            installation_id,
+            health_code="ok",
+            reason="slack_oauth_complete",
+            details={"workspace_id": workspace_id},
+        )
+        emit_audit_event(
+            action="connector.slack.oauth.install",
+            target=installation_id,
+            outcome="allow",
+            status_code=200,
+            details={
+                "workspace_id": workspace_id,
+                "workspace_name": metadata.get("workspace_name", ""),
+                "transport_mode": self.config.slack_mode,
+            },
+        )
+        return inst
+
+    def extract_workspace_id(self, payload: Dict[str, Any]) -> str:
+        if isinstance(payload.get("team_id"), str) and payload.get("team_id"):
+            return str(payload["team_id"]).strip()
+        team = payload.get("team") or {}
+        if isinstance(team, dict) and team.get("id"):
+            return str(team.get("id")).strip()
+        authorizations = payload.get("authorizations") or []
+        if isinstance(authorizations, list) and authorizations:
+            workspace_id = str((authorizations[0] or {}).get("team_id", "")).strip()
+            if workspace_id:
+                return workspace_id
+        event = payload.get("event") or {}
+        workspace_id = str(event.get("team", "") or "").strip()
+        return workspace_id
+
+    def resolve_workspace_tokens(
+        self, workspace_id: str
+    ) -> Tuple[InstallationResolution, Dict[str, str]]:
+        resolution = self._registry.resolve_installation("slack", workspace_id)
+        if not resolution.ok or resolution.installation is None:
+            emit_audit_event(
+                action="connector.slack.resolve",
+                target=workspace_id or "unknown_workspace",
+                outcome="deny",
+                status_code=409,
+                details={
+                    "workspace_id": workspace_id,
+                    "reject_reason": resolution.reject_reason,
+                    "health_code": resolution.health_code,
+                },
+            )
+            return resolution, {}
+
+        tokens: Dict[str, str] = {}
+        for token_name, ref in resolution.installation.token_refs.items():
+            secret = self._secret_store.get_secret(
+                ref, tenant_id=resolution.installation.tenant_id
+            )
+            if secret:
+                tokens[token_name] = secret
+        return resolution, tokens
+
+    def bot_user_id_for_installation(
+        self, installation: Optional[ConnectorInstallation]
+    ) -> str:
+        if installation is None:
+            return ""
+        return str(
+            (
+                installation.metadata.get("bot_user_id", "")
+                if installation.metadata
+                else ""
+            )
+            or ""
+        ).strip()
+
+    def mark_installation_health(
+        self,
+        installation_id: str,
+        *,
+        health_code: str,
+        reason: str,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._registry.update_installation_health(
+            installation_id,
+            health_code=health_code,
+            reason=reason,
+            details=details,
+        )
+
+    def uninstall_installation(self, installation_id: str, *, reason: str) -> None:
+        self._registry.uninstall_installation(installation_id, reason=reason)
+
+    def mark_resolution_success(self, installation_id: str, workspace_id: str) -> None:
+        self._registry.update_installation_health(
+            installation_id,
+            health_code="ok",
+            reason="workspace_resolved",
+            details={"workspace_id": workspace_id},
+        )
+
+    def classify_error_health(self, error_code: str, status_code: int = 0) -> str:
+        normalized = str(error_code or "").strip().lower()
+        if normalized in _INVALID_TOKEN_ERRORS or status_code in (401, 403):
+            return "invalid_token"
+        if (
+            normalized in _DEGRADED_TOKEN_ERRORS
+            or status_code == 429
+            or status_code >= 500
+        ):
+            return "degraded"
+        return "degraded"
+
+    def mark_api_error(
+        self,
+        installation_id: str,
+        *,
+        error_code: str,
+        status_code: int = 0,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> str:
+        health_code = self.classify_error_health(error_code, status_code=status_code)
+        self.mark_installation_health(
+            installation_id,
+            health_code=health_code,
+            reason=error_code or f"http_{status_code}",
+            details=details,
+        )
+        return health_code

connector/platforms/slack_webhook.py

@@ -33,12 +33,13 @@ import hmac
 import json
 import logging
 import time
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Tuple
 
 from ..config import ConnectorConfig
 from ..contract import CommandRequest, CommandResponse
 from ..router import CommandRouter
 from ..security_profile import AllowlistPolicy, ReplayGuard
+from .slack_installation_manager import SlackInstallationManager
 
 logger = logging.getLogger(__name__)
 
@@ -90,6 +91,12 @@ def _make_json_response(web_mod, data: dict, *, status: int = 200):
     )
 
 
+def _make_redirect_response(web_mod, url: str):
+    if web_mod is not None:
+        raise web_mod.HTTPFound(location=url)
+    return _CompatResponse(status=302, text=url)
+
+
 # -- Slack signature verification -------------------------------------------
 
 # Maximum acceptable clock skew for timestamp validation (5 minutes).
@@ -172,9 +179,11 @@ class SlackWebhookServer:
         self._channel_allowlist = AllowlistPolicy(
             config.slack_allowed_channels, strict=False
         )
+        self._installation_manager = SlackInstallationManager(config)
 
         # Bot user ID (resolved on first event or set from config)
         self._bot_user_id: Optional[str] = None
+        self._bot_user_ids: Dict[str, str] = {}
 
     # ------------------------------------------------------------------
     # Lifecycle
@@ -186,11 +195,19 @@ class SlackWebhookServer:
             logger.warning("aiohttp not installed. Skipping Slack adapter.")
             return
 
-        if not self.config.slack_bot_token or not self.config.slack_signing_secret:
+        if not self.config.slack_signing_secret:
             logger.info(
                 "Slack adapter disabled "
-                "(OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN or "
-                "OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET missing)"
+                "(OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET missing)"
+            )
+            return
+        if (
+            not self.config.slack_bot_token
+            and not self._installation_manager.can_handle_oauth()
+        ):
+            logger.info(
+                "Slack adapter disabled "
+                "(legacy bot token missing and Slack OAuth flow not configured)"
             )
             return
 
@@ -202,6 +219,13 @@ class SlackWebhookServer:
 
         self.app = web.Application()
         self.app.router.add_post(self.config.slack_webhook_path, self.handle_event)
+        if self._installation_manager.can_handle_oauth():
+            self.app.router.add_get(
+                self.config.slack_oauth_install_path, self.handle_oauth_install
+            )
+            self.app.router.add_get(
+                self.config.slack_oauth_callback_path, self.handle_oauth_callback
+            )
 
         self.runner = web.AppRunner(self.app)
         await self.runner.setup()
@@ -220,6 +244,155 @@ class SlackWebhookServer:
     # Event handler
     # ------------------------------------------------------------------
 
+    async def handle_oauth_install(self, request):
+        _, web = _import_aiohttp_web()
+        if not self._installation_manager.can_handle_oauth():
+            return _make_response(web, status=503, text="Slack OAuth not configured")
+        state = self._installation_manager.issue_install_state()
+        return _make_redirect_response(
+            web, self._installation_manager.build_install_url(state)
+        )
+
+    async def handle_oauth_callback(self, request):
+        _, web = _import_aiohttp_web()
+        if not self._installation_manager.can_handle_oauth():
+            return _make_response(web, status=503, text="Slack OAuth not configured")
+        query = getattr(request, "query", {}) or {}
+        if query.get("error"):
+            return _make_response(
+                web,
+                status=400,
+                text=f"Slack OAuth rejected: {query.get('error')}",
+            )
+        state = str(query.get("state", "") or "").strip()
+        code = str(query.get("code", "") or "").strip()
+        if not state or not code:
+            return _make_response(web, status=400, text="Missing OAuth callback fields")
+        if not self._installation_manager.consume_install_state(state):
+            return _make_response(
+                web, status=400, text="Invalid or replayed OAuth state"
+            )
+        try:
+            payload = await self._installation_manager.exchange_code(code)
+            installation = self._installation_manager.upsert_from_oauth_payload(payload)
+            return _make_response(
+                web,
+                status=200,
+                text=(
+                    "Slack installation complete for "
+                    f"{installation.workspace_id} ({installation.installation_id})."
+                ),
+            )
+        except Exception as exc:
+            logger.warning("Slack OAuth callback failed: %s", exc)
+            return _make_response(web, status=502, text=str(exc))
+
+    def _get_bot_user_id(self, payload: Dict[str, Any], workspace_id: str) -> str:
+        candidate = ""
+        if workspace_id and workspace_id in self._bot_user_ids:
+            return self._bot_user_ids[workspace_id]
+        if self._bot_user_id:
+            return self._bot_user_id
+        authorizations = payload.get("authorizations", [])
+        if authorizations and isinstance(authorizations, list):
+            candidate = str((authorizations[0] or {}).get("user_id", "") or "").strip()
+            if candidate:
+                self._bot_user_id = candidate
+                if workspace_id:
+                    self._bot_user_ids[workspace_id] = candidate
+                return candidate
+        if workspace_id:
+            workspace_resolution, _ = (
+                self._installation_manager.resolve_workspace_tokens(workspace_id)
+            )
+            candidate = self._installation_manager.bot_user_id_for_installation(
+                workspace_resolution.installation if workspace_resolution.ok else None
+            )
+            if candidate:
+                self._bot_user_ids[workspace_id] = candidate
+                if self._bot_user_id is None:
+                    self._bot_user_id = candidate
+        return candidate
+
+    def _resolve_workspace_credentials(
+        self, workspace_id: str
+    ) -> Tuple[Optional[str], Optional[str], Optional[str]]:
+        workspace_id = str(workspace_id or "").strip()
+        if workspace_id:
+            resolution, tokens = self._installation_manager.resolve_workspace_tokens(
+                workspace_id
+            )
+            if resolution.ok and resolution.installation is not None:
+                bot_token = tokens.get("bot_token")
+                if bot_token:
+                    self._installation_manager.mark_resolution_success(
+                        resolution.installation.installation_id, workspace_id
+                    )
+                    return (
+                        resolution.installation.installation_id,
+                        bot_token,
+                        workspace_id,
+                    )
+                logger.warning(
+                    "Slack workspace %s resolved without bot token secret", workspace_id
+                )
+                return (
+                    resolution.installation.installation_id,
+                    None,
+                    workspace_id,
+                )
+            if (
+                not self._installation_manager.oauth_enabled
+                and self.config.slack_bot_token
+            ):
+                return (None, self.config.slack_bot_token, workspace_id)
+            logger.warning(
+                "Slack workspace resolution failed for %s: %s (%s)",
+                workspace_id,
+                resolution.reject_reason,
+                resolution.health_code,
+            )
+            return (None, None, workspace_id)
+        if self.config.slack_bot_token:
+            return (None, self.config.slack_bot_token, "")
+        return (None, None, workspace_id)
+
+    def _handle_lifecycle_event(self, workspace_id: str, event_type: str) -> None:
+        installation_id = self._installation_manager.installation_id_for_workspace(
+            workspace_id
+        )
+        try:
+            if event_type == "app_uninstalled":
+                self._installation_manager.mark_installation_health(
+                    installation_id,
+                    health_code="revoked",
+                    reason="slack_app_uninstalled",
+                    details={"workspace_id": workspace_id},
+                )
+                self._installation_manager.uninstall_installation(
+                    installation_id, reason="slack_app_uninstalled"
+                )
+            elif event_type == "tokens_revoked":
+                self._installation_manager.mark_installation_health(
+                    installation_id,
+                    health_code="invalid_token",
+                    reason="slack_tokens_revoked",
+                    details={"workspace_id": workspace_id},
+                )
+            elif event_type == "app_rate_limited":
+                self._installation_manager.mark_installation_health(
+                    installation_id,
+                    health_code="degraded",
+                    reason="slack_app_rate_limited",
+                    details={"workspace_id": workspace_id},
+                )
+        except ValueError:
+            logger.warning(
+                "Slack lifecycle event for unbound workspace %s (%s)",
+                workspace_id,
+                event_type,
+            )
+
     async def handle_event(self, request):
         """POST handler for Slack Events API."""
         _, web = _import_aiohttp_web()
@@ -273,6 +446,12 @@ class SlackWebhookServer:
         event = payload.get("event", {})
         event_id = payload.get("event_id", "")
         event_type = event.get("type", "")
+        workspace_id = self._installation_manager.extract_workspace_id(payload)
+
+        if event_type in ("app_uninstalled", "tokens_revoked", "app_rate_limited"):
+            if workspace_id:
+                self._handle_lifecycle_event(workspace_id, event_type)
+            return
 
         # -- Step 5: Replay / dedupe guard --
         if not event_id:
@@ -285,13 +464,10 @@ class SlackWebhookServer:
 
         # -- Step 6: Bot-loop prevention --
         # Resolve bot user ID from authorizations or cache.
-        if self._bot_user_id is None:
-            auths = payload.get("authorizations", [])
-            if auths and isinstance(auths, list):
-                self._bot_user_id = auths[0].get("user_id", "")
+        bot_user_id = self._get_bot_user_id(payload, workspace_id)
 
         sender_id = event.get("user", "")
-        if sender_id and sender_id == self._bot_user_id:
+        if sender_id and bot_user_id and sender_id == bot_user_id:
             return
 
         if event.get("bot_id"):
@@ -317,11 +493,11 @@ class SlackWebhookServer:
         is_dm = channel_id.startswith("D")
         if not is_dm and self.config.slack_require_mention:
             if event_type != "app_mention":
-                if self._bot_user_id and f"<@{self._bot_user_id}>" not in text:
+                if bot_user_id and f"<@{bot_user_id}>" not in text:
                     return
 
-        if self._bot_user_id:
-            text = text.replace(f"<@{self._bot_user_id}>", "").strip()
+        if bot_user_id:
+            text = text.replace(f"<@{bot_user_id}>", "").strip()
 
         # -- Step 8: Allowlist checks (S67) --
         if self._user_allowlist.entries:
@@ -345,6 +521,9 @@ class SlackWebhookServer:
             message_id=event_id,
             text=text,
             timestamp=float(message_ts) if message_ts else time.time(),
+            workspace_id=workspace_id,
+            thread_id=thread_ts
+            or (message_ts if self.config.slack_reply_in_thread else ""),
         )
 
         try:
@@ -357,8 +536,11 @@ class SlackWebhookServer:
                 await self._send_reply(
                     channel_id=channel_id,
                     text=resp_text,
-                    thread_ts=thread_ts
-                    or (message_ts if self.config.slack_reply_in_thread else ""),
+                    thread_ts=req.thread_id,
+                    delivery_context={
+                        "workspace_id": workspace_id,
+                        "thread_id": req.thread_id,
+                    },
                 )
         except Exception as e:
             logger.exception(f"Error handling Slack event: {e}")
@@ -372,6 +554,7 @@ class SlackWebhookServer:
         channel_id: str,
         text: str,
         thread_ts: str = "",
+        delivery_context: Optional[Dict[str, Any]] = None,
     ) -> None:
         """Send a message via Slack Web API (chat.postMessage)."""
         try:
@@ -380,9 +563,22 @@ class SlackWebhookServer:
             logger.warning("aiohttp not available; cannot send Slack reply")
             return
 
+        ctx = dict(delivery_context or {})
+        if not thread_ts:
+            thread_ts = str(ctx.get("thread_id", "") or "").strip()
+        installation_id, bot_token, workspace_id = self._resolve_workspace_credentials(
+            str(ctx.get("workspace_id", "") or "").strip()
+        )
+        if not bot_token:
+            logger.warning(
+                "Slack reply dropped: no workspace token available (workspace=%s)",
+                workspace_id or "legacy",
+            )
+            return
+
         url = "https://slack.com/api/chat.postMessage"
         headers = {
-            "Authorization": f"Bearer {self.config.slack_bot_token}",
+            "Authorization": f"Bearer {bot_token}",
             "Content-Type": "application/json; charset=utf-8",
         }
         payload: Dict[str, Any] = {
@@ -402,15 +598,41 @@ class SlackWebhookServer:
                 ) as resp:
                     if resp.status != 200:
                         body = await resp.text()
+                        if installation_id:
+                            self._installation_manager.mark_api_error(
+                                installation_id,
+                                error_code=f"http_{resp.status}",
+                                status_code=resp.status,
+                                details={
+                                    "workspace_id": workspace_id,
+                                    "path": "chat.postMessage",
+                                },
+                            )
                         logger.warning(
                             f"Slack API error: status={resp.status} body={body[:200]}"
                         )
                     else:
                         data = await resp.json()
                         if not data.get("ok"):
+                            if installation_id:
+                                self._installation_manager.mark_api_error(
+                                    installation_id,
+                                    error_code=str(data.get("error", "unknown")),
+                                    details={
+                                        "workspace_id": workspace_id,
+                                        "path": "chat.postMessage",
+                                    },
+                                )
                             logger.warning(
                                 f"Slack API error: {data.get('error', 'unknown')}"
                             )
+                        elif installation_id:
+                            self._installation_manager.mark_installation_health(
+                                installation_id,
+                                health_code="ok",
+                                reason="chat_post_message_ok",
+                                details={"workspace_id": workspace_id},
+                            )
         except Exception as e:
             logger.warning(f"Slack reply failed: {e}")
 
@@ -418,9 +640,18 @@ class SlackWebhookServer:
     # Platform contract: send_message / send_image
     # ------------------------------------------------------------------
 
-    async def send_message(self, channel_id: str, text: str):
+    async def send_message(
+        self,
+        channel_id: str,
+        text: str,
+        delivery_context: Optional[Dict[str, Any]] = None,
+    ):
         """Platform contract: send text message."""
-        await self._send_reply(channel_id=channel_id, text=text)
+        await self._send_reply(
+            channel_id=channel_id,
+            text=text,
+            delivery_context=delivery_context,
+        )
 
     async def send_image(
         self,
@@ -428,6 +659,7 @@ class SlackWebhookServer:
         image_data: bytes,
         filename: str = "image.png",
         caption: Optional[str] = None,
+        delivery_context: Optional[Dict[str, Any]] = None,
     ):
         """Platform contract: send image (Slack files.upload)."""
         try:
@@ -436,15 +668,29 @@ class SlackWebhookServer:
             logger.warning("aiohttp not available; cannot upload Slack image")
             return
 
+        ctx = dict(delivery_context or {})
+        thread_ts = str(ctx.get("thread_id", "") or "").strip()
+        installation_id, bot_token, workspace_id = self._resolve_workspace_credentials(
+            str(ctx.get("workspace_id", "") or "").strip()
+        )
+        if not bot_token:
+            logger.warning(
+                "Slack image dropped: no workspace token available (workspace=%s)",
+                workspace_id or "legacy",
+            )
+            return
+
         url = "https://slack.com/api/files.upload"
         headers = {
-            "Authorization": f"Bearer {self.config.slack_bot_token}",
+            "Authorization": f"Bearer {bot_token}",
         }
         data = _aiohttp.FormData()
         data.add_field("file", image_data, filename=filename, content_type="image/png")
         data.add_field("channels", channel_id)
         if caption:
             data.add_field("initial_comment", caption)
+        if thread_ts:
+            data.add_field("thread_ts", thread_ts)
 
         try:
             async with _aiohttp.ClientSession() as session:
@@ -455,12 +701,38 @@ class SlackWebhookServer:
                     timeout=_aiohttp.ClientTimeout(total=30),
                 ) as resp:
                     if resp.status != 200:
+                        if installation_id:
+                            self._installation_manager.mark_api_error(
+                                installation_id,
+                                error_code=f"http_{resp.status}",
+                                status_code=resp.status,
+                                details={
+                                    "workspace_id": workspace_id,
+                                    "path": "files.upload",
+                                },
+                            )
                         logger.warning(f"Slack file upload error: status={resp.status}")
                     else:
                         resp_data = await resp.json()
                         if not resp_data.get("ok"):
+                            if installation_id:
+                                self._installation_manager.mark_api_error(
+                                    installation_id,
+                                    error_code=str(resp_data.get("error", "unknown")),
+                                    details={
+                                        "workspace_id": workspace_id,
+                                        "path": "files.upload",
+                                    },
+                                )
                             logger.warning(
                                 f"Slack file upload error: {resp_data.get('error')}"
                             )
+                        elif installation_id:
+                            self._installation_manager.mark_installation_health(
+                                installation_id,
+                                health_code="ok",
+                                reason="files_upload_ok",
+                                details={"workspace_id": workspace_id},
+                            )
         except Exception as e:
             logger.warning(f"Slack image upload failed: {e}")

connector/platforms/telegram_polling.py

@@ -182,6 +182,7 @@ class TelegramPolling:
         image_data: bytes,
         filename: str = "image.png",
         caption: Optional[str] = None,
+        delivery_context: Optional[dict] = None,
     ):
         """Send photo via Telegram sendPhoto."""
         if not self.session:
@@ -205,7 +206,12 @@ class TelegramPolling:
         except Exception as e:
             logger.error(f"Telegram send_image error: {e}")
 
-    async def send_message(self, channel_id: str, text: str):
+    async def send_message(
+        self,
+        channel_id: str,
+        text: str,
+        delivery_context: Optional[dict] = None,
+    ):
         """Send text message."""
         if not self.session:
             return

connector/platforms/wechat_webhook.py

@@ -680,7 +680,12 @@ class WeChatWebhookServer:
     # Outbound: Text (Customer Service Message API)
     # ------------------------------------------------------------------
 
-    async def send_message(self, recipient_openid: str, text: str):
+    async def send_message(
+        self,
+        recipient_openid: str,
+        text: str,
+        delivery_context: Optional[dict] = None,
+    ):
         """
         Send text via WeChat Customer Service Message API.
 
@@ -736,6 +741,7 @@ class WeChatWebhookServer:
         image_data: bytes,
         filename: str = "image.png",
         caption: Optional[str] = None,
+        delivery_context: Optional[dict] = None,
     ):
         """
         Send image via WeChat.

connector/platforms/whatsapp_webhook.py

@@ -296,7 +296,12 @@ class WhatsAppWebhookServer:
     # Outbound: Text
     # ------------------------------------------------------------------
 
-    async def send_message(self, recipient_id: str, text: str):
+    async def send_message(
+        self,
+        recipient_id: str,
+        text: str,
+        delivery_context: Optional[dict] = None,
+    ):
         """Send text message via WhatsApp Cloud API."""
         if self._session_invalid:
             logger.warning("R93: Connector session invalid - blocking outbound")
@@ -350,6 +355,7 @@ class WhatsAppWebhookServer:
         image_data: bytes,
         filename: str = "image.png",
         caption: Optional[str] = None,
+        delivery_context: Optional[dict] = None,
     ):
         """
         Send image via WhatsApp using public media URL.

connector/results_poller.py

@@ -1,7 +1,7 @@
 import asyncio
 import logging
 import time
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 from .config import ConnectorConfig
 from .contract import Platform
@@ -26,10 +26,10 @@ class ResultsPoller:
         self.platforms = platforms  # map "telegram" -> TelegramPolling, etc.
         self.queue = (
             asyncio.Queue()
-        )  # (prompt_id, platform_name, channel_id, sender_id)
+        )  # (prompt_id, platform_name, channel_id, sender_id, delivery_context)
         self.approval_queue = (
             asyncio.Queue()
-        )  # (approval_id, platform_name, channel_id, sender_id)
+        )  # (approval_id, platform_name, channel_id, sender_id, delivery_context)
         self.active_polls = {}  # prompt_id -> task
         self.active_approval_polls = {}  # approval_id -> task
 
@@ -55,17 +55,35 @@ class ResultsPoller:
         logger.info("ResultsPoller stopped.")
 
     def track_job(
-        self, prompt_id: str, platform_name: str, channel_id: str, sender_id: str
+        self,
+        prompt_id: str,
+        platform_name: str,
+        channel_id: str,
+        sender_id: str,
+        delivery_context: Optional[Dict[str, Any]] = None,
     ):
         """Enqueue a job for result monitoring."""
         if not prompt_id:
             return
 
         logger.info(f"Tracking job {prompt_id} for {platform_name} in {channel_id}")
-        self.queue.put_nowait((prompt_id, platform_name, channel_id, sender_id))
+        self.queue.put_nowait(
+            (
+                prompt_id,
+                platform_name,
+                channel_id,
+                sender_id,
+                dict(delivery_context or {}),
+            )
+        )
 
     def track_approval(
-        self, approval_id: str, platform_name: str, channel_id: str, sender_id: str
+        self,
+        approval_id: str,
+        platform_name: str,
+        channel_id: str,
+        sender_id: str,
+        delivery_context: Optional[Dict[str, Any]] = None,
     ):
         if not approval_id:
             return
@@ -85,16 +103,28 @@ class ResultsPoller:
             f"Tracking approval {approval_id} for {platform_name} in {channel_id}"
         )
         self.approval_queue.put_nowait(
-            (approval_id, platform_name, channel_id, sender_id)
+            (
+                approval_id,
+                platform_name,
+                channel_id,
+                sender_id,
+                dict(delivery_context or {}),
+            )
         )
 
     async def _job_consumer(self):
         while True:
             item = await self.queue.get()
             try:
-                prompt_id, platform_name, channel_id, sender_id = item
+                prompt_id, platform_name, channel_id, sender_id, delivery_context = item
                 task = asyncio.create_task(
-                    self._poll_job(prompt_id, platform_name, channel_id, sender_id)
+                    self._poll_job(
+                        prompt_id,
+                        platform_name,
+                        channel_id,
+                        sender_id,
+                        delivery_context,
+                    )
                 )
                 self.active_polls[prompt_id] = task
                 task.add_done_callback(
@@ -107,10 +137,20 @@ class ResultsPoller:
         while True:
             item = await self.approval_queue.get()
             try:
-                approval_id, platform_name, channel_id, sender_id = item
+                (
+                    approval_id,
+                    platform_name,
+                    channel_id,
+                    sender_id,
+                    delivery_context,
+                ) = item
                 task = asyncio.create_task(
                     self._poll_approval(
-                        approval_id, platform_name, channel_id, sender_id
+                        approval_id,
+                        platform_name,
+                        channel_id,
+                        sender_id,
+                        delivery_context,
                     )
                 )
                 self.active_approval_polls[approval_id] = task
@@ -121,7 +161,12 @@ class ResultsPoller:
                 self.approval_queue.task_done()
 
     async def _poll_approval(
-        self, approval_id: str, platform_name: str, channel_id: str, sender_id: str
+        self,
+        approval_id: str,
+        platform_name: str,
+        channel_id: str,
+        sender_id: str,
+        delivery_context: Optional[Dict[str, Any]] = None,
     ):
         start_time = time.time()
         delay = 2.0
@@ -137,6 +182,7 @@ class ResultsPoller:
                             platform_name,
                             channel_id,
                             f"❌ Approval {approval_id} {status}.",
+                            delivery_context=delivery_context,
                         )
                         return
                     if status == "approved":
@@ -150,7 +196,11 @@ class ResultsPoller:
                         )
                         if prompt_id:
                             self.track_job(
-                                prompt_id, platform_name, channel_id, sender_id
+                                prompt_id,
+                                platform_name,
+                                channel_id,
+                                sender_id,
+                                delivery_context=delivery_context,
                             )
                             return
             except Exception as e:
@@ -166,10 +216,16 @@ class ResultsPoller:
             platform_name,
             channel_id,
             f"⚠️ Approval {approval_id} timed out waiting for execution.",
+            delivery_context=delivery_context,
         )
 
     async def _poll_job(
-        self, prompt_id: str, platform_name: str, channel_id: str, sender_id: str
+        self,
+        prompt_id: str,
+        platform_name: str,
+        channel_id: str,
+        sender_id: str,
+        delivery_context: Optional[Dict[str, Any]] = None,
     ):
         """Poll history with backoff until complete or timeout."""
         start_time = time.time()
@@ -185,7 +241,11 @@ class ResultsPoller:
                     if prompt_id in data:
                         job_data = data[prompt_id]
                         await self._deliver_results(
-                            prompt_id, job_data, platform_name, channel_id
+                            prompt_id,
+                            job_data,
+                            platform_name,
+                            channel_id,
+                            delivery_context=delivery_context,
                         )
                         return
             except Exception as e:
@@ -203,10 +263,16 @@ class ResultsPoller:
             platform_name,
             channel_id,
             f"⚠️ Job {prompt_id} timed out waiting for results.",
+            delivery_context=delivery_context,
         )
 
     async def _deliver_results(
-        self, prompt_id: str, job_data: dict, platform_name: str, channel_id: str
+        self,
+        prompt_id: str,
+        job_data: dict,
+        platform_name: str,
+        channel_id: str,
+        delivery_context: Optional[Dict[str, Any]] = None,
     ):
         """Download images and send to platform."""
         outputs = job_data.get("outputs", {})
@@ -216,6 +282,7 @@ class ResultsPoller:
                 platform_name,
                 channel_id,
                 f"✅ Job {prompt_id} finished (No output images).",
+                delivery_context=delivery_context,
             )
             return
 
@@ -231,7 +298,10 @@ class ResultsPoller:
         if not images_to_send:
             logger.info(f"Job {prompt_id} finished but no images found.")
             await self._send_text(
-                platform_name, channel_id, f"✅ Job {prompt_id} finished (No images)."
+                platform_name,
+                channel_id,
+                f"✅ Job {prompt_id} finished (No images).",
+                delivery_context=delivery_context,
             )
             return
 
@@ -263,23 +333,43 @@ class ResultsPoller:
                     platform_name,
                     channel_id,
                     f"⚠️ Image {filename} skipped (too large).",
+                    delivery_context=delivery_context,
                 )
                 continue
 
             # Send with error handling
             try:
-                await platform.send_image(channel_id, content, filename=filename)
+                await platform.send_image(
+                    channel_id,
+                    content,
+                    filename=filename,
+                    delivery_context=delivery_context,
+                )
             except Exception as e:
                 logger.error(f"Failed to deliver image to {platform_name}: {e}")
                 # Fallback text
                 await self._send_text(
-                    platform_name, channel_id, f"⚠️ Failed to send image: {filename}"
+                    platform_name,
+                    channel_id,
+                    f"⚠️ Failed to send image: {filename}",
+                    delivery_context=delivery_context,
                 )
 
-    async def _send_text(self, platform_name: str, channel_id: str, text: str):
+    async def _send_text(
+        self,
+        platform_name: str,
+        channel_id: str,
+        text: str,
+        *,
+        delivery_context: Optional[Dict[str, Any]] = None,
+    ):
         platform = self.platforms.get(platform_name)
         if platform:
             try:
-                await platform.send_message(channel_id, text)
+                await platform.send_message(
+                    channel_id,
+                    text,
+                    delivery_context=dict(delivery_context or {}),
+                )
             except Exception as e:
                 logger.error(f"Failed to send text to {platform_name}: {e}")

connector/router.py

@@ -184,6 +184,14 @@ class CommandRouter:
     def _is_admin(self, user_id: str) -> bool:
         return str(user_id) in self.config.admin_users
 
+    def _delivery_context(self, req: CommandRequest) -> Dict[str, Any]:
+        context: Dict[str, Any] = {}
+        if getattr(req, "workspace_id", ""):
+            context["workspace_id"] = str(req.workspace_id)
+        if getattr(req, "thread_id", ""):
+            context["thread_id"] = str(req.thread_id)
+        return context
+
     def _check_command_authz(
         self, cmd: str, req: CommandRequest, default_class: CommandClass
     ) -> Optional[CommandResponse]:
@@ -415,14 +423,22 @@ class CommandRouter:
                     # We must start tracking the approval_id so we can map
                     # approval_id -> executed_prompt_id later and auto-deliver images.
                     self.poller.track_approval(
-                        approval_id, req.platform, req.channel_id, req.sender_id
+                        approval_id,
+                        req.platform,
+                        req.channel_id,
+                        req.sender_id,
+                        delivery_context=self._delivery_context(req),
                     )
                 return CommandResponse(text=msg)
             else:
                 prompt_id = data.get("prompt_id", "unknown")
                 if self.poller:
                     self.poller.track_job(
-                        prompt_id, req.platform, req.channel_id, req.sender_id
+                        prompt_id,
+                        req.platform,
+                        req.channel_id,
+                        req.sender_id,
+                        delivery_context=self._delivery_context(req),
                     )
 
                 return CommandResponse(
@@ -547,7 +563,13 @@ class CommandRouter:
             if self.poller:
                 # Approval request might have come from different flow, but usually user invoking /approve
                 # wants the result. Using current req context is safest assumption for "ChatOps".
-                self.poller.track_job(pid, req.platform, req.channel_id, req.sender_id)
+                self.poller.track_job(
+                    pid,
+                    req.platform,
+                    req.channel_id,
+                    req.sender_id,
+                    delivery_context=self._delivery_context(req),
+                )
         elif data.get("executed") is False:
             msg += "\n(Not Executed)"
             if err := data.get("execution_error"):

docs/connector.md

@@ -27,6 +27,7 @@ OpenClaw now includes a platform-agnostic baseline for multi-workspace connector
   - `platform`, `workspace_id`, `installation_id`, `token_refs`, `status`, `updated_at`
 - token material is kept in encrypted server-side secret storage; registry and diagnostics expose token references only
 - workspace resolution is fail-closed on missing/ambiguous/inactive/stale bindings
+- installation diagnostics can also surface stable health states such as `ok`, `invalid_token`, `revoked`, `workspace_unbound`, and `degraded`
 - interactive callback contract enforces signed envelope checks, timestamp window, payload-hash validation, replay/idempotency guardrails, and command-policy mapping (`public`/`run`/`admin`) with explicit force-approval outcomes for untrusted `run` callbacks
 
 Admin diagnostics APIs:
@@ -36,6 +37,12 @@ Admin diagnostics APIs:
 - `GET /openclaw/connector/installations/resolve?platform=<platform>&workspace_id=<workspace_id>`
 - `GET /openclaw/connector/installations/audit`
 
+Slack multi-workspace notes:
+
+- Slack OAuth installs bind one workspace per installation record and persist only encrypted token refs.
+- `GET /openclaw/connector/installations/resolve?platform=slack&workspace_id=<team_id>` returns the fail-closed resolution view for a specific Slack workspace.
+- `GET /openclaw/connector/installations` diagnostics may include per-install health metadata plus aggregate `health_counts`.
+
 ### Multi-tenant boundary behavior
 
 When backend multi-tenant mode is enabled (`OPENCLAW_MULTI_TENANT_ENABLED=1`):
@@ -145,8 +152,15 @@ Set the following environment variables (or put them in a `.env` file if you use
 
 *(Requires Inbound Connectivity - see below)*
 
-- `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN`: Bot User OAuth Token (`xoxb-...`).
+- `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN`: Optional legacy single-workspace Bot User OAuth Token (`xoxb-...`). When Slack OAuth is configured, per-workspace tokens are resolved from the installation registry instead.
 - `OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET`: Signing Secret (from App Credentials).
+- `OPENCLAW_CONNECTOR_SLACK_CLIENT_ID`: OAuth client ID for multi-workspace installation flow.
+- `OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET`: OAuth client secret for multi-workspace installation flow.
+- `OPENCLAW_CONNECTOR_SLACK_OAUTH_REDIRECT_URI`: Explicit OAuth callback URL. If omitted, connector derives it from `OPENCLAW_CONNECTOR_PUBLIC_BASE_URL` + callback path.
+- `OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH`: Local install route (default `/slack/install`).
+- `OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH`: Local callback route (default `/slack/oauth/callback`).
+- `OPENCLAW_CONNECTOR_SLACK_OAUTH_SCOPES`: Comma-separated bot scopes used for install URL generation.
+- `OPENCLAW_CONNECTOR_SLACK_OAUTH_STATE_TTL_SEC`: TTL in seconds for single-use OAuth state tokens (default `600`).
 - `OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS`: Comma-separated user IDs (e.g. `U12345, U67890`).
 - `OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS`: Comma-separated channel IDs (e.g. `C12345`).
 - `OPENCLAW_CONNECTOR_SLACK_BIND`: Host to bind (default `127.0.0.1`).
@@ -373,14 +387,18 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint
      - `im:history` (DM support)
      - `channels:history` (public channel messages)
      - `groups:history` (private channel messages)
-   - Click **Install to Workspace**.
-   - Copy the **Bot User OAuth Token** (`xoxb-...`).
+   - For legacy single-workspace mode, click **Install to Workspace** and copy the **Bot User OAuth Token** (`xoxb-...`).
+   - For F58 multi-workspace mode, configure a redirect URL and let OpenClaw handle installs through its OAuth routes.
 
 3. **Configure connector environment variables**
 
    ```bash
-   OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-your-token
    OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET=your-signing-secret
+   OPENCLAW_CONNECTOR_SLACK_CLIENT_ID=1234567890.1234567890
+   OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET=replace-with-client-secret
+   OPENCLAW_CONNECTOR_PUBLIC_BASE_URL=https://your-public-host
+   OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH=/slack/install
+   OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH=/slack/oauth/callback
    OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS=U12345,U67890
    OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS=C12345
    OPENCLAW_CONNECTOR_SLACK_BIND=127.0.0.1
@@ -392,6 +410,7 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint
    ```
 
    Notes:
+   - Legacy single-workspace fallback can still set `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-...`; F58 multi-workspace mode no longer requires that token at startup if OAuth install flow is configured.
    - `OPENCLAW_CONNECTOR_ADMIN_TOKEN` must match server `OPENCLAW_ADMIN_TOKEN` if server-side admin token is enabled.
    - Slack ingress is fail-closed: invalid/missing signature, stale timestamp, and replayed events are rejected.
 
@@ -400,6 +419,8 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint
    - Expose local endpoint to public HTTPS (Cloudflare Tunnel/ngrok/reverse proxy):
      - local upstream: `http://127.0.0.1:8095`
      - public URL: `https://<public-host>/slack/events`
+     - install URL: `https://<public-host>/slack/install`
+     - callback URL: `https://<public-host>/slack/oauth/callback`
 
 5. **Enable Event Subscriptions**
    - Go to **Event Subscriptions** and enable events.
@@ -412,15 +433,17 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint
      - `message.im`
 
 6. **Invite and validate**
+   - Open `https://<public-host>/slack/install` and complete the workspace install.
    - Invite the app to target channels: `/invite @YourBot`.
    - In channel: `@YourBot /status` (when `OPENCLAW_CONNECTOR_SLACK_REQUIRE_MENTION=true`).
    - In DM: `/help`.
    - Verify connector logs show signed ingress accepted and replies delivered.
+   - Verify `GET /openclaw/connector/installations` shows the Slack workspace binding and health state `ok`.
 
 7. **Security checklist before production**
    - Keep `OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS`/`OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS` restricted.
    - Keep `OPENCLAW_CONNECTOR_SLACK_REQUIRE_MENTION=true` unless intentionally running command-style channels.
-   - Rotate Slack bot token/signing secret on incident response.
+   - Rotate Slack signing secret and OAuth client secret on incident response.
    - Do not expose connector without HTTPS termination.
 
 #### Slack Socket Mode Setup (Optional)
@@ -437,9 +460,13 @@ Use Socket Mode when you cannot expose a public HTTPS webhook endpoint.
    ```bash
    OPENCLAW_CONNECTOR_SLACK_MODE=socket
    OPENCLAW_CONNECTOR_SLACK_APP_TOKEN=xapp-your-token
-   # Bot token + signing secret are still required for parity and safety checks
-   OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-your-token
+   # Signing secret remains required for parity/security checks.
    OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET=your-signing-secret
+   # Either configure legacy single-workspace token...
+   OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-your-token
+   # ...or configure multi-workspace OAuth install flow:
+   OPENCLAW_CONNECTOR_SLACK_CLIENT_ID=1234567890.1234567890
+   OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET=replace-with-client-secret
    ```
 
 3. **Start connector**
@@ -449,6 +476,7 @@ Use Socket Mode when you cannot expose a public HTTPS webhook endpoint.
 Notes:
 - Socket Mode uses outbound WebSocket, so `OPENCLAW_CONNECTOR_SLACK_BIND`, `OPENCLAW_CONNECTOR_SLACK_PORT`, and `OPENCLAW_CONNECTOR_SLACK_PATH` are ignored in this mode.
 - Startup is fail-closed if `OPENCLAW_CONNECTOR_SLACK_APP_TOKEN` is missing or does not start with `xapp-`.
+- In multi-workspace mode, outbound replies still resolve the workspace-specific bot token from the installation registry even though the WebSocket connection itself uses the app-level token.
 
 ## Commands
 

docs/release/api_contract.md

@@ -99,6 +99,11 @@ Assist payload redaction contract:
 | `GET` | `/connector/installations/resolve` | `/moltbot/connector/installations/resolve` | Admin | Run fail-closed workspace resolution diagnostics (`platform`, `workspace_id`). |
 | `GET` | `/connector/installations/audit` | `/moltbot/connector/installations/audit` | Admin | List installation lifecycle audit evidence (redacted). |
 
+Connector diagnostics contract notes:
+- installation records may expose operator-safe health metadata under `installation.metadata.health` (for example `ok`, `invalid_token`, `revoked`, `degraded`) without exposing token material
+- `/connector/installations` diagnostics may include aggregate `health_counts` in addition to lifecycle `status_counts`
+- `/connector/installations/resolve` may expose a stable `health_code` alongside the legacy `reject_reason` so clients can distinguish `workspace_unbound` vs token-health failures without parsing status text
+
 ### 1.3C LLM Management & Chat
 
 **LLM Base Path**: `/openclaw/llm/`

docs/release/config_secrets_contract.md

@@ -131,12 +131,20 @@ Controls the `connector` sidecar process and outbound delivery.
 | `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN` | Slack | Bot OAuth token (`xoxb-*`). |
 | `OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET` | Slack | Ingress signature secret. |
 | `OPENCLAW_CONNECTOR_SLACK_APP_TOKEN` | Slack | Optional Socket Mode app token (`xapp-*`). |
+| `OPENCLAW_CONNECTOR_SLACK_CLIENT_ID` | Slack | OAuth client ID for multi-workspace installation flow. |
+| `OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET` | Slack | OAuth client secret for multi-workspace installation flow. |
+| `OPENCLAW_CONNECTOR_SLACK_OAUTH_REDIRECT_URI` | Slack | Explicit OAuth callback URL. Falls back to `OPENCLAW_CONNECTOR_PUBLIC_BASE_URL + OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH` when omitted. |
+| `OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH` | Slack | Local install route path (default `/slack/install`). |
+| `OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH` | Slack | Local OAuth callback route path (default `/slack/oauth/callback`). |
+| `OPENCLAW_CONNECTOR_SLACK_OAUTH_SCOPES` | Slack | Comma-separated bot scopes for install URL generation. |
+| `OPENCLAW_CONNECTOR_SLACK_OAUTH_STATE_TTL_SEC` | Slack | TTL for single-use OAuth state tokens (default `600`). |
 | `OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS` | Slack | Comma-separated trusted user IDs. |
 | `OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS` | Slack | Comma-separated trusted channel IDs. |
 
 Connector posture rules:
 - In strict posture (`OPENCLAW_DEPLOYMENT_PROFILE=public` or `OPENCLAW_RUNTIME_PROFILE=hardened`), active connector platforms without allowlist coverage are fail-closed.
 - Public deployment profile check surfaces this as `DP-PUBLIC-009`.
+- Slack multi-workspace installs persist only encrypted token refs in `connector_installations.json`; raw bot/app tokens remain in encrypted secret storage and must not appear in diagnostics or exported config surfaces.
 
 **Delivery & Media:**
 

services/connector_installation_registry.py

@@ -49,6 +49,7 @@ class InstallationStatus(str, Enum):
 _RESOLVABLE_STATUSES = frozenset(
     {InstallationStatus.ACTIVE.value, InstallationStatus.ROTATING.value}
 )
+_UNRESOLVABLE_HEALTH_CODES = frozenset({"invalid_token", "revoked"})
 
 
 @dataclass
@@ -109,12 +110,14 @@ class InstallationResolution:
     installation: Optional[ConnectorInstallation] = None
     reject_reason: str = ""
     audit_code: str = ""
+    health_code: str = ""
 
     def to_public_dict(self) -> Dict[str, Any]:
         payload: Dict[str, Any] = {
             "ok": self.ok,
             "reject_reason": self.reject_reason,
             "audit_code": self.audit_code,
+            "health_code": self.health_code,
         }
         if self.installation is not None:
             payload["installation"] = self.installation.to_public_dict()
@@ -147,6 +150,15 @@ class ConnectorInstallationRegistry:
     def _normalize_platform(self, platform: str) -> str:
         return self._normalize_identifier(platform, "platform").lower()
 
+    def _health_code_for_installation(self, inst: ConnectorInstallation) -> str:
+        health = dict(inst.metadata.get("health", {}) or {})
+        health_code = str(health.get("state", "")).strip().lower()
+        if health_code:
+            return health_code
+        if inst.status == InstallationStatus.REVOKED.value:
+            return "revoked"
+        return ""
+
     def _store_token_refs(
         self, installation_id: str, token_values: Dict[str, str], tenant_id: str
     ) -> Dict[str, str]:
@@ -369,12 +381,18 @@ class ConnectorInstallationRegistry:
     def activate_installation(
         self, installation_id: str, reason: str = ""
     ) -> ConnectorInstallation:
-        return self._transition(
+        inst = self._transition(
             installation_id,
             InstallationStatus.ACTIVE.value,
             reason=reason,
             action="activate",
         )
+        return self.update_installation_health(
+            installation_id,
+            health_code="ok",
+            reason=reason or "activated",
+            details={"status": inst.status},
+        )
 
     def rotate_installation_tokens(
         self,
@@ -410,12 +428,18 @@ class ConnectorInstallationRegistry:
     def revoke_installation(
         self, installation_id: str, reason: str = ""
     ) -> ConnectorInstallation:
-        return self._transition(
+        self._transition(
             installation_id,
             InstallationStatus.REVOKED.value,
             reason=reason,
             action="revoke",
         )
+        return self.update_installation_health(
+            installation_id,
+            health_code="revoked",
+            reason=reason or "revoked",
+            details={"status": InstallationStatus.REVOKED.value},
+        )
 
     def deactivate_installation(
         self, installation_id: str, reason: str = ""
@@ -444,6 +468,37 @@ class ConnectorInstallationRegistry:
             self._save()
             return ConnectorInstallation(**asdict(inst))
 
+    def update_installation_health(
+        self,
+        installation_id: str,
+        *,
+        health_code: str,
+        reason: str = "",
+        details: Optional[Dict[str, Any]] = None,
+    ) -> ConnectorInstallation:
+        with self._lock:
+            inst = self._installations.get(str(installation_id).strip())
+            if inst is None:
+                raise ValueError(f"Installation not found: {installation_id}")
+            metadata = dict(inst.metadata or {})
+            metadata["health"] = {
+                "state": str(health_code or "ok").strip().lower() or "ok",
+                "reason": str(reason or "").strip(),
+                "updated_at": time.time(),
+                "details": dict(details or {}),
+            }
+            inst.metadata = metadata
+            inst.updated_at = time.time()
+            self._installations[inst.installation_id] = inst
+            self._audit(
+                "health_update",
+                inst,
+                health_code=metadata["health"]["state"],
+                reason=metadata["health"]["reason"],
+            )
+            self._save()
+            return ConnectorInstallation(**asdict(inst))
+
     def resolve_installation(
         self, platform: str, workspace_id: str, tenant_id: Optional[str] = None
     ) -> InstallationResolution:
@@ -470,6 +525,7 @@ class ConnectorInstallationRegistry:
                         ok=False,
                         reject_reason="tenant_mismatch",
                         audit_code="conn_install.resolve_tenant_mismatch",
+                        health_code="degraded",
                     )
                 matches = tenant_matches
             eligible = [inst for inst in matches if inst.status in _RESOLVABLE_STATUSES]
@@ -478,6 +534,7 @@ class ConnectorInstallationRegistry:
                     ok=False,
                     reject_reason="ambiguous_binding",
                     audit_code="conn_install.resolve_ambiguous",
+                    health_code="degraded",
                 )
             if not eligible:
                 if not matches:
@@ -485,24 +542,39 @@ class ConnectorInstallationRegistry:
                         ok=False,
                         reject_reason="missing_binding",
                         audit_code="conn_install.resolve_missing",
+                        health_code="workspace_unbound",
                     )
+                health_code = (
+                    self._health_code_for_installation(matches[0]) or "degraded"
+                )
                 return InstallationResolution(
                     ok=False,
                     reject_reason="inactive_binding",
                     audit_code="conn_install.resolve_inactive",
+                    health_code=health_code,
                 )
             inst = eligible[0]
+            health_code = self._health_code_for_installation(inst)
+            if health_code in _UNRESOLVABLE_HEALTH_CODES:
+                return InstallationResolution(
+                    ok=False,
+                    reject_reason="inactive_binding",
+                    audit_code="conn_install.resolve_unhealthy",
+                    health_code=health_code,
+                )
             for token_name, ref in inst.token_refs.items():
                 if not self._secret_store.get_secret(ref, tenant_id=inst.tenant_id):
                     return InstallationResolution(
                         ok=False,
                         reject_reason=f"stale_token_ref:{token_name}",
                         audit_code="conn_install.resolve_stale_ref",
+                        health_code="degraded",
                     )
             return InstallationResolution(
                 ok=True,
                 installation=ConnectorInstallation(**asdict(inst)),
                 audit_code="conn_install.resolve_ok",
+                health_code=health_code or "ok",
             )
 
     def get_audit_trail(
@@ -530,15 +602,19 @@ class ConnectorInstallationRegistry:
     def diagnostics(self, tenant_id: Optional[str] = None) -> Dict[str, Any]:
         with self._lock:
             counts: Dict[str, int] = {}
+            health_counts: Dict[str, int] = {}
             items = list(self._installations.values())
             if tenant_id:
                 normalized_tenant = normalize_tenant_id(tenant_id)
                 items = [inst for inst in items if inst.tenant_id == normalized_tenant]
             for inst in items:
                 counts[inst.status] = counts.get(inst.status, 0) + 1
+                health_code = self._health_code_for_installation(inst) or "ok"
+                health_counts[health_code] = health_counts.get(health_code, 0) + 1
             return {
                 "installation_count": len(items),
                 "status_counts": counts,
+                "health_counts": health_counts,
                 "audit_events": len(self._audit_trail),
             }
 

tests/connector/test_poller.py

@@ -9,11 +9,16 @@ from connector.results_poller import ResultsPoller
 
 class MockPlatform(Platform):
     async def send_image(
-        self, channel_id, image_data, filename="image.png", caption=None
+        self,
+        channel_id,
+        image_data,
+        filename="image.png",
+        caption=None,
+        delivery_context=None,
     ):
         pass
 
-    async def send_message(self, channel_id, text):
+    async def send_message(self, channel_id, text, delivery_context=None):
         pass
 
 
@@ -37,7 +42,7 @@ class TestResultsPoller(unittest.TestCase):
         self.poller.track_job("p-1", "test_plat", "c-1", "u-1")
         self.assertEqual(self.poller.queue.qsize(), 1)
         item = self.poller.queue.get_nowait()
-        self.assertEqual(item, ("p-1", "test_plat", "c-1", "u-1"))
+        self.assertEqual(item, ("p-1", "test_plat", "c-1", "u-1", {}))
 
     @patch("connector.results_poller.time")
     @patch("connector.results_poller.asyncio.sleep", new_callable=AsyncMock)
@@ -62,12 +67,23 @@ class TestResultsPoller(unittest.TestCase):
 
         self.client.get_view.return_value = b"image_bytes"
 
-        asyncio.run(self.poller._poll_job("p-1", "test_plat", "c-1", "u-1"))
+        asyncio.run(
+            self.poller._poll_job(
+                "p-1",
+                "test_plat",
+                "c-1",
+                "u-1",
+                {"workspace_id": "T1", "thread_id": "123.456"},
+            )
+        )
 
         self.assertEqual(self.client.get_history.call_count, 2)
         self.client.get_view.assert_called_with("f.png", "", "output")
         self.mock_platform.send_image.assert_called_with(
-            "c-1", b"image_bytes", filename="f.png"
+            "c-1",
+            b"image_bytes",
+            filename="f.png",
+            delivery_context={"workspace_id": "T1", "thread_id": "123.456"},
         )
 
     @patch("connector.results_poller.time")

tests/test_connector_installation_registry.py

@@ -67,6 +67,7 @@ class TestConnectorInstallationRegistry(unittest.TestCase):
         res = self.registry.resolve_installation("slack", "missing")
         self.assertFalse(res.ok)
         self.assertEqual(res.reject_reason, "missing_binding")
+        self.assertEqual(res.health_code, "workspace_unbound")
 
     def test_resolution_duplicate_binding_fails_closed(self):
         self.registry.upsert_installation(
@@ -113,6 +114,27 @@ class TestConnectorInstallationRegistry(unittest.TestCase):
         res = self.registry.resolve_installation("slack", "T3")
         self.assertFalse(res.ok)
         self.assertEqual(res.reject_reason, "inactive_binding")
+        self.assertEqual(res.health_code, "revoked")
+
+    def test_invalid_token_health_fails_closed(self):
+        self.registry.upsert_installation(
+            platform="slack",
+            workspace_id="T4",
+            installation_id="inst-invalid",
+            token_values={"bot_token": "xoxb-invalid"},
+            status=InstallationStatus.ACTIVE.value,
+        )
+        self.registry.update_installation_health(
+            "inst-invalid",
+            health_code="invalid_token",
+            reason="provider_401",
+            details={"source": "slack_api"},
+        )
+
+        res = self.registry.resolve_installation("slack", "T4")
+        self.assertFalse(res.ok)
+        self.assertEqual(res.reject_reason, "inactive_binding")
+        self.assertEqual(res.health_code, "invalid_token")
 
     def test_persistence_reload_and_redaction(self):
         self.registry.upsert_installation(
@@ -131,11 +153,12 @@ class TestConnectorInstallationRegistry(unittest.TestCase):
         self.assertEqual(len(listed), 1)
         self.assertEqual(listed[0].installation_id, "inst-persist")
 
-        raw = open(
+        with open(
             os.path.join(self.state_dir, "connector_installations.json"),
             "r",
             encoding="utf-8",
-        ).read()
+        ) as fh:
+            raw = fh.read()
         self.assertNotIn("xoxb-secret", raw)
 
     def test_multi_tenant_resolve_mismatch_fail_closed(self):

tests/test_f57_slack_transport_parity.py

@@ -10,6 +10,7 @@ from unittest.mock import AsyncMock, MagicMock
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
 from connector.config import ConnectorConfig
+from connector.contract import CommandResponse
 from connector.platforms.slack_socket_mode import SlackSocketModeClient
 
 
@@ -21,7 +22,7 @@ class TestF57SlackTransportParity(unittest.IsolatedAsyncioTestCase):
         self.config.slack_signing_secret = "secret"
 
         self.router = MagicMock()
-        self.router.handle = AsyncMock()
+        self.router.handle = AsyncMock(return_value=CommandResponse(text=""))
         self.client = SlackSocketModeClient(self.config, self.router)
 
     async def test_socket_mode_routes_message(self):

tests/test_f58_slack_oauth_installations.py

@@ -0,0 +1,139 @@
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from connector.config import ConnectorConfig
+from connector.contract import CommandResponse
+from connector.platforms.slack_installation_manager import SlackInstallationManager
+from connector.platforms.slack_webhook import SlackWebhookServer
+from services.connector_installation_registry import ConnectorInstallationRegistry
+from services.secret_store import SecretStore
+
+
+class TestF58SlackOAuthInstallations(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.state_dir = self.tmpdir.name
+        self.secret_store = SecretStore(state_dir=self.state_dir)
+        self.registry = ConnectorInstallationRegistry(
+            state_dir=self.state_dir,
+            secret_store=self.secret_store,
+        )
+        self.config = ConnectorConfig()
+        self.config.public_base_url = "https://connector.example.com"
+        self.config.slack_signing_secret = "signing-secret"
+        self.config.slack_client_id = "client-id"
+        self.config.slack_client_secret = "client-secret"
+        self.config.slack_bot_token = "xoxb-legacy"
+        self.manager = SlackInstallationManager(
+            self.config,
+            registry=self.registry,
+            secret_store=self.secret_store,
+            state_dir=self.state_dir,
+        )
+
+    async def asyncTearDown(self):
+        self.tmpdir.cleanup()
+
+    def _oauth_payload(self, token: str, *, workspace_id: str = "T1") -> dict:
+        return {
+            "ok": True,
+            "app_id": "A123",
+            "access_token": token,
+            "scope": "chat:write,files:write",
+            "bot_user_id": "U_BOT_1",
+            "team": {"id": workspace_id, "name": "Workspace One"},
+            "authed_user": {"id": "U_INSTALLER"},
+            "token_type": "bot",
+        }
+
+    async def test_oauth_state_single_use_and_workspace_binding(self):
+        state = self.manager.issue_install_state()
+        self.assertTrue(self.manager.consume_install_state(state))
+        self.assertFalse(self.manager.consume_install_state(state))
+
+        inst = self.manager.upsert_from_oauth_payload(self._oauth_payload("xoxb-first"))
+        self.assertEqual(inst.workspace_id, "T1")
+        self.assertEqual(inst.status, "active")
+
+        rotated = self.manager.upsert_from_oauth_payload(
+            self._oauth_payload("xoxb-rotated")
+        )
+        self.assertEqual(rotated.installation_id, inst.installation_id)
+
+        resolution, tokens = self.manager.resolve_workspace_tokens("T1")
+        self.assertTrue(resolution.ok)
+        self.assertEqual(tokens["bot_token"], "xoxb-rotated")
+
+        with open(
+            os.path.join(self.state_dir, "connector_installations.json"),
+            "r",
+            encoding="utf-8",
+        ) as fh:
+            raw = fh.read()
+        self.assertNotIn("xoxb-rotated", raw)
+
+    async def test_workspace_bound_reply_uses_installation_token(self):
+        self.manager.upsert_from_oauth_payload(self._oauth_payload("xoxb-workspace"))
+
+        router = MagicMock()
+        router.handle = AsyncMock(return_value=CommandResponse(text="Done"))
+        server = SlackWebhookServer(self.config, router)
+        server._installation_manager = self.manager
+
+        with patch("aiohttp.ClientSession") as mock_session_cls:
+            mock_session = mock_session_cls.return_value
+            mock_session.__aenter__.return_value = mock_session
+            mock_session.post.return_value.__aenter__.return_value.status = 200
+            mock_session.post.return_value.__aenter__.return_value.json = AsyncMock(
+                return_value={"ok": True}
+            )
+            mock_session.post.return_value.__aenter__.return_value.text = AsyncMock(
+                return_value="OK"
+            )
+
+            payload = {
+                "type": "event_callback",
+                "team_id": "T1",
+                "authorizations": [{"user_id": "U_BOT_1", "team_id": "T1"}],
+                "event_id": "EvF58-1",
+                "event": {
+                    "type": "message",
+                    "text": "/status",
+                    "user": "U_SENDER",
+                    "channel": "D_DM",
+                    "ts": "1700000000.001",
+                },
+            }
+
+            await server.process_event_payload(payload)
+
+        routed = router.handle.call_args[0][0]
+        self.assertEqual(routed.workspace_id, "T1")
+        self.assertEqual(routed.thread_id, "1700000000.001")
+        headers = mock_session.post.call_args.kwargs["headers"]
+        self.assertEqual(headers["Authorization"], "Bearer xoxb-workspace")
+
+    async def test_tokens_revoked_event_marks_installation_invalid(self):
+        self.manager.upsert_from_oauth_payload(self._oauth_payload("xoxb-workspace"))
+        router = MagicMock()
+        router.handle = AsyncMock(return_value=CommandResponse(text="Done"))
+        server = SlackWebhookServer(self.config, router)
+        server._installation_manager = self.manager
+
+        payload = {
+            "type": "event_callback",
+            "team_id": "T1",
+            "event": {
+                "type": "tokens_revoked",
+            },
+        }
+        await server.process_event_payload(payload)
+
+        resolution = self.registry.resolve_installation("slack", "T1")
+        self.assertFalse(resolution.ok)
+        self.assertEqual(resolution.health_code, "invalid_token")