feat: Deep Research Slack integration (#365)

frontend/src/types/connectors.ts

@@ -114,7 +114,7 @@ export const SOURCE_CATALOG: ConnectorMeta[] = [
     category: 'communication',
     icon: 'Hash',
     color: 'text-purple-400',
-    description: 'Read messages from channels, DMs, and threads',
+    description: 'Read messages from every channel, private channel, DM, and group DM you have access to',
     unitLabel: 'messages',
     steps: [
       {
@@ -123,16 +123,16 @@ export const SOURCE_CATALOG: ConnectorMeta[] = [
         urlLabel: 'Open Slack Apps',
       },
       {
-        label: 'In the left sidebar, click "OAuth & Permissions". Scroll down to "Bot Token Scopes" and click "Add an OAuth Scope" to add EACH of these scopes one by one:',
+        label: 'In the left sidebar, click "OAuth & Permissions". Scroll down to "User Token Scopes" (NOT "Bot Token Scopes"). Click "Add an OAuth Scope" and add EACH of these scopes one by one:',
       },
       {
-        label: 'channels:read • channels:history • channels:join • groups:read • groups:history • im:read • im:history • mpim:read • mpim:history • chat:write • users:read • app_mentions:read',
+        label: 'channels:history • channels:read • groups:history • groups:read • im:history • im:read • mpim:history • mpim:read • users:read',
       },
       {
-        label: 'In the left sidebar, click "Install App" → click "Install to Workspace" → click "Allow". After installing, copy the "Bot User OAuth Token" that appears (starts with xoxb-)',
+        label: 'In the left sidebar, click "Install App" → click "Install to Workspace" → click "Allow". After installing, copy the "User OAuth Token" that appears (starts with xoxp-, NOT xoxb-)',
       },
       {
-        label: 'Paste the bot token below. After connecting, invite the bot to channels you want indexed by typing /invite @OpenJarvis in each channel',
+        label: 'Paste the user token below. Sync indexes every channel, private channel, DM, and group DM you have access to — no need to invite anything to channels',
       },
       {
         label: '(Optional) Set the app icon: in the left sidebar click "Basic Information" → scroll to "Display Information" → upload the OpenJarvis logo',
@@ -141,7 +141,7 @@ export const SOURCE_CATALOG: ConnectorMeta[] = [
       },
     ],
     inputFields: [
-      { name: 'token', placeholder: 'xoxb-...', type: 'password' },
+      { name: 'token', placeholder: 'xoxp-...', type: 'password' },
     ],
   },
   {

src/openjarvis/agents/research_loop.py

@@ -225,11 +225,11 @@ def _bare_doc_id(source: str, document_id: str) -> str:
 def _hit_url(source: str, document_id: str) -> str:
     """Build a clickable URL for a hit, when we know how to link it.
 
-    Gmail is the only source we currently link out for; everything else
-    returns an empty string and the client falls back to non-clickable
-    citation chips.
+    Gmail and Slack are the linkable sources today; anything else falls
+    back to an empty string and the client renders a non-clickable
+    citation chip.
 
-    Two id flavors land here:
+    Gmail ids land here in two flavors:
 
     - **Hex message id** (``19dfa2ccbeff78b0``) — what the OAuth Gmail
       connector stores. Resolves directly via ``#all/<id>`` permalink.
@@ -238,6 +238,13 @@ def _hit_url(source: str, document_id: str) -> str:
       id. The permalink form would 404; instead route through Gmail's
       search URL with the ``rfc822msgid:`` operator, which lands the user
       on the specific message.
+
+    Slack doc_ids encode workspace + channel + timestamp as
+    ``slack:{team_domain}:{channel_id}:{ts}`` so the permalink
+    ``https://{team_domain}.slack.com/archives/{channel_id}/p{ts}`` can
+    be reconstructed without a side lookup. Legacy two-segment ids
+    (``slack:{channel_id}:{ts}`` from earlier ingests) fall back to the
+    workspace-less ``slack.com/archives/...`` form.
     """
     if source == "gmail" and document_id:
         msg_id = _bare_doc_id(source, document_id)
@@ -247,6 +254,24 @@ def _hit_url(source: str, document_id: str) -> str:
             rfc_id = msg_id.strip("<>")
             return f"https://mail.google.com/mail/u/0/#search/rfc822msgid:{rfc_id}"
         return f"https://mail.google.com/mail/u/0/#all/{msg_id}"
+    if source == "slack" and document_id:
+        bare = _bare_doc_id(source, document_id)
+        if not bare:
+            return ""
+        parts = bare.split(":")
+        if len(parts) >= 3:
+            team_domain, channel_id, ts = parts[0], parts[1], ":".join(parts[2:])
+        elif len(parts) == 2:
+            team_domain = ""
+            channel_id, ts = parts
+        else:
+            return ""
+        if not channel_id or not ts:
+            return ""
+        ts_clean = ts.replace(".", "")
+        if team_domain:
+            return f"https://{team_domain}.slack.com/archives/{channel_id}/p{ts_clean}"
+        return f"https://slack.com/archives/{channel_id}/p{ts_clean}"
     return ""
 
 

src/openjarvis/connectors/slack_connector.py

@@ -1,12 +1,18 @@
 """Slack connector — bulk channel message sync via the Slack Web API.
 
-Uses OAuth tokens stored locally (see :mod:`openjarvis.connectors.oauth`).
-All network calls are isolated in module-level functions (``_slack_api_*``)
-to make them trivially mockable in tests.
+Uses a Slack **user** OAuth token (``xoxp-...``) stored locally so the
+sync sees everything the user can see — including their 1:1 DMs and
+multi-person DMs with other humans. A bot token (``xoxb-``) cannot see
+user-to-user DMs (Slack platform constraint), so this connector
+explicitly rejects bot tokens with a clear error at connect time.
+
+All network calls are isolated in module-level functions
+(``_slack_api_*``) to make them trivially mockable in tests.
 """
 
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timezone
 from typing import Any, Dict, Iterator, List, Optional
 from urllib.parse import urlencode
@@ -19,13 +25,18 @@ from openjarvis.core.config import DEFAULT_CONFIG_DIR
 from openjarvis.core.registry import ConnectorRegistry
 from openjarvis.tools._stubs import ToolSpec
 
+logger = logging.getLogger(__name__)
+
 # ---------------------------------------------------------------------------
 # Constants
 # ---------------------------------------------------------------------------
 
 _SLACK_API_BASE = "https://slack.com/api"
 _SLACK_AUTH_ENDPOINT = "https://slack.com/oauth/v2/authorize"
-_SLACK_SCOPES = (
+# User Token Scopes — Slack distinguishes user_scope (xoxp-) from scope
+# (xoxb-) on the OAuth authorize URL. We only request user scopes; passed
+# via ``user_scope`` so the install grants a User OAuth Token.
+_SLACK_USER_SCOPES = (
     "channels:read,channels:history,groups:read,groups:history,"
     "im:read,im:history,mpim:read,mpim:history,users:read"
 )
@@ -55,8 +66,12 @@ def _slack_api_conversations_list(
     dict
         Raw API response containing ``channels`` list and ``response_metadata``.
     """
+    # Include every conversation type the bot can list — public + private
+    # channels, multi-person DMs, and 1:1 DMs — so a "connect and sync"
+    # flow indexes everything the token has access to without the user
+    # picking channels (matches Gmail's connect-and-go behavior).
     params: Dict[str, str] = {
-        "types": "public_channel,private_channel",
+        "types": "public_channel,private_channel,mpim,im",
         "exclude_archived": "true",
     }
     if cursor:
@@ -110,6 +125,16 @@ def _slack_api_users_list(token: str) -> Dict[str, Any]:
     return _slack_api_with_retry("users.list", token)
 
 
+def _slack_api_auth_test(token: str) -> Dict[str, Any]:
+    """Call the Slack ``auth.test`` endpoint.
+
+    Returns workspace context (``team_id``, ``team``, ``url``) used to
+    construct message permalinks — the workspace subdomain isn't carried
+    by any other endpoint we already call.
+    """
+    return _slack_api_with_retry("auth.test", token)
+
+
 def _slack_api_with_retry(
     method: str,
     token: str,
@@ -155,6 +180,45 @@ def _slack_api_with_retry(
 # ---------------------------------------------------------------------------
 
 
+class SlackTokenError(ValueError):
+    """Raised when the stored Slack token isn't a usable user token.
+
+    Surfaced via :class:`SyncStatus.error` so the UI can render the actual
+    reason (e.g. ``xoxb-`` bot token rejected) instead of an empty sync.
+    """
+
+
+_USER_TOKEN_PREFIX = "xoxp-"
+_BOT_TOKEN_PREFIX = "xoxb-"
+
+
+def _validate_user_token(token: str) -> None:
+    """Raise :class:`SlackTokenError` unless *token* looks like a user token.
+
+    Slack user tokens start with ``xoxp-``. We refuse ``xoxb-`` bot tokens
+    explicitly because a bot can only see DMs *to/from itself* — paste a
+    bot token and your sync silently misses every human-to-human DM. The
+    error message tells the user exactly which token type to provide and
+    where to find it.
+    """
+    if not token:
+        raise SlackTokenError("Slack token is empty.")
+    if token.startswith(_BOT_TOKEN_PREFIX):
+        raise SlackTokenError(
+            "Slack bot tokens (xoxb-) can't see user-to-user DMs. "
+            "Use a User OAuth Token (xoxp-) from "
+            "api.slack.com/apps → OAuth & Permissions, "
+            "with User Token Scopes channels:history, channels:read, "
+            "groups:history, groups:read, im:history, im:read, "
+            "mpim:history, mpim:read, users:read."
+        )
+    if not token.startswith(_USER_TOKEN_PREFIX):
+        raise SlackTokenError(
+            "Slack token must be a User OAuth Token (starts with 'xoxp-'). "
+            "Got a token with an unexpected prefix."
+        )
+
+
 def _build_user_map(members: List[Dict[str, Any]]) -> Dict[str, Dict[str, str]]:
     """Build a user_id → {name, email} map from a ``users.list`` members list."""
     user_map: Dict[str, Dict[str, str]] = {}
@@ -180,9 +244,32 @@ def _ts_to_datetime(ts: str) -> datetime:
         return datetime.now()
 
 
-def _slack_archive_url(team_id: str, channel_id: str, ts: str) -> str:
-    """Build a Slack message archive URL from team, channel, and timestamp."""
+def _team_domain_from_auth(auth_resp: Dict[str, Any]) -> str:
+    """Derive the workspace subdomain ('acme' from 'https://acme.slack.com/').
+
+    Falls back to the ``team_id`` so doc_ids stay non-empty when the
+    workspace ``url`` is missing — losing the workspace breaks permalinks
+    but lets ingestion continue.
+    """
+    workspace_url: str = (auth_resp.get("url") or "").rstrip("/")
+    if workspace_url:
+        host = workspace_url.split("//", 1)[-1].split("/", 1)[0]
+        suffix = ".slack.com"
+        if host.endswith(suffix):
+            return host[: -len(suffix)]
+    return auth_resp.get("team_id", "") or ""
+
+
+def _slack_archive_url(team_domain: str, channel_id: str, ts: str) -> str:
+    """Build a Slack message permalink for ``team_domain``/``channel``/``ts``.
+
+    With a workspace subdomain the link resolves directly; without one we
+    fall back to ``slack.com/archives/...`` which Slack redirects only for
+    logged-in members of that workspace.
+    """
     ts_clean = ts.replace(".", "")
+    if team_domain:
+        return f"https://{team_domain}.slack.com/archives/{channel_id}/p{ts_clean}"
     return f"https://slack.com/archives/{channel_id}/p{ts_clean}"
 
 
@@ -215,6 +302,10 @@ class SlackConnector(BaseConnector):
         self._items_total: int = 0
         self._last_sync: Optional[datetime] = None
         self._last_cursor: Optional[str] = None
+        # Surfaced via sync_status().error so the UI can render the actual
+        # failure reason (typically a bot-token-rejection) instead of a
+        # silent "synced 0 messages".
+        self._last_error: Optional[str] = None
 
     # ------------------------------------------------------------------
     # BaseConnector interface
@@ -232,21 +323,30 @@ class SlackConnector(BaseConnector):
         delete_tokens(self._credentials_path)
 
     def auth_url(self) -> str:
-        """Return a Slack OAuth consent URL requesting channel history scopes."""
+        """Return a Slack OAuth consent URL requesting user-token scopes.
+
+        Uses ``user_scope`` (not ``scope``) so the install grants a User
+        OAuth Token (``xoxp-``) — bot tokens (``xoxb-``) can't see human-
+        to-human DMs and are rejected by :func:`handle_callback`.
+        """
         params = {
             "client_id": "",  # placeholder — real client_id from config
-            "scope": _SLACK_SCOPES,
+            "user_scope": _SLACK_USER_SCOPES,
             "redirect_uri": "http://localhost:8789/callback",
         }
         return f"{_SLACK_AUTH_ENDPOINT}?{urlencode(params)}"
 
     def handle_callback(self, code: str) -> None:
-        """Handle the OAuth callback by persisting the authorization code.
+        """Persist the supplied User OAuth Token after validating its shape.
 
-        In a full implementation this would exchange the code for tokens.
-        For now the code is saved directly as the token value.
+        The connector ``/connect`` endpoint funnels manually-pasted tokens
+        through this method (the parameter is named ``code`` for OAuth-flow
+        compatibility). Bot tokens (``xoxb-``) are rejected here so the
+        invalid credential never lands on disk.
         """
+        _validate_user_token(code)
         save_tokens(self._credentials_path, {"token": code})
+        self._last_error = None
 
     def sync(
         self,
@@ -254,10 +354,13 @@ class SlackConnector(BaseConnector):
         since: Optional[datetime] = None,  # noqa: ARG002 — reserved for future use
         cursor: Optional[str] = None,  # noqa: ARG002 — reserved for future use
     ) -> Iterator[Document]:
-        """Yield :class:`Document` objects for Slack channel messages.
+        """Yield :class:`Document` objects for every accessible Slack message.
 
-        Builds a user map, then paginates through channels and retrieves
-        message history for each channel.
+        With a user OAuth token (``xoxp-``) the listing returned by
+        ``conversations.list`` already reflects what the user can see —
+        no ``conversations.join`` step, no membership filtering. We
+        enumerate every conversation first (so the per-type count can
+        be logged up front), then stream history for each one.
 
         Parameters
         ----------
@@ -274,118 +377,225 @@ class SlackConnector(BaseConnector):
         if not token:
             return
 
-        # Step 1: build user map
+        # Reject bot tokens up front so the user sees the actual reason
+        # for an empty sync instead of every API call coming back
+        # missing_scope / 0 messages.
+        try:
+            _validate_user_token(token)
+        except SlackTokenError as exc:
+            self._last_error = str(exc)
+            logger.warning("Slack sync rejected token: %s", exc)
+            return
+
+        # Step 0: resolve workspace context — the subdomain is needed for
+        # message permalinks and is the only piece of state that doesn't
+        # come back from conversations.history. Done once per sync.
+        try:
+            auth_resp = _slack_api_auth_test(token)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Slack auth.test failed: %s", exc)
+            auth_resp = {}
+        if not auth_resp.get("ok", True):
+            err = str(auth_resp.get("error", "auth_failed"))
+            self._last_error = f"Slack auth.test failed: {err}"
+            logger.warning("Slack auth.test returned not-ok: %s", err)
+            return
+
+        team_domain: str = _team_domain_from_auth(auth_resp)
+        team_id: str = auth_resp.get("team_id", "") or ""
+        workspace_name: str = auth_resp.get("team", "") or ""
+        workspace_url: str = auth_resp.get("url", "") or ""
+
+        # Step 1: build user map (so DMs render with peer names, not IDs)
         users_resp = _slack_api_users_list(token)
         members: List[Dict[str, Any]] = users_resp.get("members", [])
         user_map = _build_user_map(members)
 
-        synced = 0
+        # Step 2: enumerate every conversation up front so we can log a
+        # per-type summary before fetching history. Pagination over
+        # conversations.list is cheap relative to history fetch and gives
+        # the user (and the logs) immediate signal about what the token
+        # can actually see.
+        all_channels: List[Dict[str, Any]] = []
         channels_cursor = ""
-
-        # Step 2: paginate through channels
         while True:
-            channels_resp = _slack_api_conversations_list(token, cursor=channels_cursor)
-            channels: List[Dict[str, Any]] = channels_resp.get("channels", [])
+            channels_resp = _slack_api_conversations_list(
+                token, cursor=channels_cursor
+            )
+            if not channels_resp.get("ok", True):
+                err = str(channels_resp.get("error", "list_failed"))
+                self._last_error = f"Slack conversations.list failed: {err}"
+                logger.warning("Slack conversations.list returned not-ok: %s", err)
+                return
+            all_channels.extend(channels_resp.get("channels", []))
+            channels_cursor = (
+                channels_resp.get("response_metadata", {}).get("next_cursor", "")
+                or ""
+            )
+            if not channels_cursor:
+                break
 
-            for channel in channels:
-                chan_id: str = channel.get("id", "")
-                chan_name: str = channel.get("name", chan_id)
-                is_member: bool = channel.get("is_member", False)
-                is_private: bool = channel.get("is_private", False)
-                if not chan_id:
-                    continue
+        counts = {"public_channel": 0, "private_channel": 0, "im": 0, "mpim": 0}
+        for c in all_channels:
+            if c.get("is_im"):
+                counts["im"] += 1
+            elif c.get("is_mpim"):
+                counts["mpim"] += 1
+            elif c.get("is_private"):
+                counts["private_channel"] += 1
+            else:
+                counts["public_channel"] += 1
+        logger.info(
+            "Slack: Found %d public channels, %d private channels, "
+            "%d DMs, %d group DMs",
+            counts["public_channel"],
+            counts["private_channel"],
+            counts["im"],
+            counts["mpim"],
+        )
 
-                # Auto-join public channels; skip private channels the bot isn't in
-                if not is_member:
-                    if is_private:
-                        continue  # Can't join private channels without invite
-                    # Try to join the public channel
-                    try:
-                        join_resp = _slack_api_with_retry(
-                            "conversations.join",
-                            token,
-                            {"channel": chan_id},
-                            http_method="POST",
-                        )
-                        if not join_resp.get("ok"):
-                            continue
-                    except Exception:
+        # Step 3: fetch history per channel and yield Documents.
+        synced = 0
+        for channel in all_channels:
+            chan_id: str = channel.get("id", "")
+            is_private: bool = channel.get("is_private", False)
+            is_im: bool = channel.get("is_im", False)
+            is_mpim: bool = channel.get("is_mpim", False)
+            if not chan_id:
+                continue
+
+            # Display name: IMs have no ``name`` field — render as
+            # ``dm-<peer-name>`` so result chips show something readable.
+            raw_name: str = channel.get("name", "") or ""
+            if is_im:
+                peer_id: str = channel.get("user", "") or ""
+                peer_info = user_map.get(peer_id, {})
+                peer_label = peer_info.get("name") or peer_id or "user"
+                chan_name = f"dm-{peer_label}"
+            else:
+                chan_name = raw_name or chan_id
+
+            channel_type = (
+                "im"
+                if is_im
+                else "mpim"
+                if is_mpim
+                else "private_channel"
+                if is_private
+                else "public_channel"
+            )
+
+            history_cursor = ""
+            while True:
+                try:
+                    history_resp = _slack_api_conversations_history(
+                        token, chan_id, cursor=history_cursor
+                    )
+                except Exception as exc:  # noqa: BLE001
+                    logger.debug(
+                        "Slack history fetch failed for %s (%s): %s",
+                        chan_name,
+                        chan_id,
+                        exc,
+                    )
+                    break
+                if not history_resp.get("ok", True):
+                    # User token shouldn't hit not_in_channel, but if a
+                    # scope was revoked mid-sync we skip the channel rather
+                    # than abort the whole sync.
+                    logger.debug(
+                        "Slack history not-ok for %s (%s): %s",
+                        chan_name,
+                        chan_id,
+                        history_resp.get("error"),
+                    )
+                    break
+                messages: List[Dict[str, Any]] = history_resp.get("messages", [])
+
+                for msg in messages:
+                    # Skip bot messages and non-content subtypes
+                    if msg.get("bot_id") or msg.get("subtype") in (
+                        "message_changed",
+                        "message_deleted",
+                        "bot_message",
+                        "channel_join",
+                        "channel_leave",
+                    ):
                         continue
 
-                # Step 3: paginate through message history
-                history_cursor = ""
-                while True:
-                    try:
-                        history_resp = _slack_api_conversations_history(
-                            token, chan_id, cursor=history_cursor
-                        )
-                    except Exception:
-                        break  # Skip channels we can't read
-                    if not history_resp.get("ok", True):
-                        break  # not_in_channel or other error
-                    messages: List[Dict[str, Any]] = history_resp.get("messages", [])
+                    ts: str = msg.get("ts", "")
+                    user_id: str = msg.get("user", "")
+                    text: str = msg.get("text", "")
+                    thread_ts: Optional[str] = msg.get("thread_ts")
 
-                    for msg in messages:
-                        # Skip bot messages and non-content subtypes
-                        if msg.get("bot_id") or msg.get("subtype") in (
-                            "message_changed",
-                            "message_deleted",
-                            "bot_message",
-                            "channel_join",
-                            "channel_leave",
-                        ):
-                            continue
+                    user_info = user_map.get(user_id, {})
+                    author_name: str = user_info.get("name", user_id)
+                    author_email: str = user_info.get("email", "")
 
-                        ts: str = msg.get("ts", "")
-                        user_id: str = msg.get("user", "")
-                        text: str = msg.get("text", "")
-                        thread_ts: Optional[str] = msg.get("thread_ts")
+                    timestamp = _ts_to_datetime(ts)
+                    url = _slack_archive_url(team_domain, chan_id, ts)
 
-                        user_info = user_map.get(user_id, {})
-                        author = user_info.get("name", user_id)
+                    # v1 schema participants: lowercase email when we have
+                    # one, else the display name — matches the Gmail
+                    # connector's contract (one identity per participant)
+                    # so cross-source queries work.
+                    canonical = (author_email or author_name).lower()
+                    participants = [canonical] if canonical else []
+                    participants_raw = [user_id] if user_id else []
 
-                        timestamp = _ts_to_datetime(ts)
-                        url = _slack_archive_url("", chan_id, ts)
+                    # Encode workspace into doc_id so research_loop can
+                    # rebuild a workspace-qualified permalink from
+                    # source + document_id alone.
+                    doc_id = f"slack:{team_domain}:{chan_id}:{ts}"
 
-                        doc = Document(
-                            doc_id=f"slack:{chan_id}:{ts}",
-                            source="slack",
-                            doc_type="message",
-                            content=text,
-                            title=f"#{chan_name}",
-                            author=author,
-                            timestamp=timestamp,
-                            thread_id=thread_ts,
-                            url=url,
-                            metadata={
-                                "channel_id": chan_id,
-                                "channel_name": chan_name,
-                                "user_id": user_id,
-                                "ts": ts,
-                            },
-                        )
-                        synced += 1
-                        yield doc
+                    # Channel rows get ``#name``; DM rows get
+                    # ``DM with <peer>`` (more useful than ``#dm-alice``
+                    # in result chips).
+                    if is_im:
+                        title = f"DM with {chan_name.removeprefix('dm-')}"
+                    else:
+                        title = f"#{chan_name}"
 
-                    next_history_cursor: str = (
-                        history_resp.get("response_metadata", {}).get("next_cursor", "")
-                        or ""
+                    doc = Document(
+                        doc_id=doc_id,
+                        source="slack",
+                        doc_type="message",
+                        content=text,
+                        title=title,
+                        author=author_email or author_name,
+                        participants=participants,
+                        participants_raw=participants_raw,
+                        timestamp=timestamp,
+                        thread_id=thread_ts,
+                        channel=chan_name,
+                        url=url,
+                        metadata={
+                            "channel_id": chan_id,
+                            "channel_name": chan_name,
+                            "channel_type": channel_type,
+                            "user_id": user_id,
+                            "ts": ts,
+                            "team_id": team_id,
+                            "team_domain": team_domain,
+                            "workspace_name": workspace_name,
+                            "workspace_url": workspace_url,
+                        },
                     )
-                    if not history_resp.get("has_more") or not next_history_cursor:
-                        break
-                    history_cursor = next_history_cursor
+                    synced += 1
+                    yield doc
 
-            next_channels_cursor: str = (
-                channels_resp.get("response_metadata", {}).get("next_cursor", "") or ""
-            )
-            if not next_channels_cursor:
-                self._last_cursor = None
-                break
-            channels_cursor = next_channels_cursor
-            self._last_cursor = channels_cursor
+                next_history_cursor: str = (
+                    history_resp.get("response_metadata", {}).get("next_cursor", "")
+                    or ""
+                )
+                if not history_resp.get("has_more") or not next_history_cursor:
+                    break
+                history_cursor = next_history_cursor
 
         self._items_synced = synced
         self._last_sync = datetime.now()
+        self._last_cursor = None
+        self._last_error = None
 
     def sync_status(self) -> SyncStatus:
         """Return sync progress from the most recent :meth:`sync` call."""
@@ -394,6 +604,7 @@ class SlackConnector(BaseConnector):
             items_synced=self._items_synced,
             last_sync=self._last_sync,
             cursor=self._last_cursor,
+            error=self._last_error,
         )
 
     # ------------------------------------------------------------------

tests/agents/test_research_loop.py

@@ -12,7 +12,7 @@ from unittest.mock import MagicMock
 
 import pytest
 
-from openjarvis.agents.research_loop import ResearchAgent
+from openjarvis.agents.research_loop import ResearchAgent, _hit_url
 
 
 class _MockEngine:
@@ -184,3 +184,32 @@ def test_clarify_before_any_search_is_rejected(stub_search: MagicMock) -> None:
     # No clarify invocation recorded.
     assert all(t.tool_name != "clarify" for t in result.tool_calls)
     assert result.answer == "Final."
+
+
+# ---------------------------------------------------------------------------
+# _hit_url — Slack permalink reconstruction
+# ---------------------------------------------------------------------------
+
+
+def test_hit_url_slack_full_workspace() -> None:
+    """A workspace-qualified doc_id produces an ``{team}.slack.com`` permalink."""
+    url = _hit_url("slack", "slack:acme:C123:1710500000.000100")
+    assert url == "https://acme.slack.com/archives/C123/p1710500000000100"
+
+
+def test_hit_url_slack_legacy_two_segment_doc_id() -> None:
+    """Legacy ``slack:{channel}:{ts}`` ids fall back to the docless form."""
+    url = _hit_url("slack", "slack:C123:1710500000.000100")
+    assert url == "https://slack.com/archives/C123/p1710500000000100"
+
+
+def test_hit_url_slack_empty_team_domain() -> None:
+    """Empty workspace segment still produces a usable docless permalink."""
+    url = _hit_url("slack", "slack::C123:1710500000.000100")
+    assert url == "https://slack.com/archives/C123/p1710500000000100"
+
+
+def test_hit_url_unknown_source_returns_empty() -> None:
+    """Unsupported sources don't get a guessed URL."""
+    assert _hit_url("notion", "notion:abc") == ""
+    assert _hit_url("", "") == ""

tests/connectors/test_slack_connector.py

@@ -51,6 +51,15 @@ _USERS_RESPONSE = {
     ],
 }
 
+_AUTH_TEST_RESPONSE = {
+    "ok": True,
+    "team_id": "T0ACME",
+    "team": "Acme",
+    "url": "https://acme.slack.com/",
+    "user": "bot",
+    "user_id": "UBOT",
+}
+
 
 # ---------------------------------------------------------------------------
 # Fixtures
@@ -92,10 +101,19 @@ def test_auth_type_is_oauth(connector) -> None:
 
 
 def test_auth_url(connector) -> None:
-    """auth_url() returns a URL pointing to Slack's OAuth endpoint."""
+    """auth_url() returns a URL requesting user-token scopes (not bot scopes).
+
+    The migration to user OAuth tokens means scopes go in ``user_scope``
+    (not ``scope``), and DM/MPIM history scopes are mandatory.
+    """
     url = connector.auth_url()
     assert isinstance(url, str)
     assert "slack.com" in url
+    # User-token install: scopes carried by ``user_scope``, not ``scope``.
+    assert "user_scope=" in url
+    # DM and MPIM history are required for Deep Research over personal DMs.
+    assert "im:history" in url or "im%3Ahistory" in url
+    assert "mpim:history" in url or "mpim%3Ahistory" in url
     assert "channels:history" in url or "channels%3Ahistory" in url
 
 
@@ -104,6 +122,7 @@ def test_auth_url(connector) -> None:
 # ---------------------------------------------------------------------------
 
 
+@patch("openjarvis.connectors.slack_connector._slack_api_auth_test")
 @patch("openjarvis.connectors.slack_connector._slack_api_conversations_list")
 @patch("openjarvis.connectors.slack_connector._slack_api_conversations_history")
 @patch("openjarvis.connectors.slack_connector._slack_api_users_list")
@@ -111,6 +130,7 @@ def test_sync_yields_documents(
     mock_users,
     mock_history,
     mock_channels,
+    mock_auth,
     connector,
     tmp_path: Path,
 ) -> None:
@@ -118,11 +138,15 @@ def test_sync_yields_documents(
 
     With 2 channels each having 2 messages, we expect exactly 4 documents.
     """
-    # Set up fake credentials so is_connected() returns True
+    # Set up fake credentials so is_connected() returns True. User tokens
+    # (``xoxp-``) are the only shape the connector accepts post-migration.
     creds_path = Path(connector._credentials_path)
-    creds_path.write_text(json.dumps({"token": "fake-access-token"}), encoding="utf-8")
+    creds_path.write_text(
+        json.dumps({"token": "xoxp-fake-user-token"}), encoding="utf-8"
+    )
 
     # Configure mocks
+    mock_auth.return_value = _AUTH_TEST_RESPONSE
     mock_users.return_value = _USERS_RESPONSE
     mock_channels.return_value = _CHANNELS_RESPONSE
     mock_history.return_value = _HISTORY_RESPONSE
@@ -137,35 +161,178 @@ def test_sync_yields_documents(
         assert doc.source == "slack"
         assert doc.doc_type == "message"
 
-    # Check a specific document from #general
+    # Check a specific document from #general — doc_id encodes the
+    # workspace subdomain so research_loop can rebuild the permalink.
     doc_c001 = next(
-        (d for d in docs if d.doc_id == "slack:C001:1710500000.000100"), None
+        (d for d in docs if d.doc_id == "slack:acme:C001:1710500000.000100"),
+        None,
     )
     assert doc_c001 is not None
     assert doc_c001.title == "#general"
-    assert doc_c001.author == "Alice"
+    assert doc_c001.author == "alice@co.com"
     assert doc_c001.content == "Let's discuss the API redesign."
     assert doc_c001.thread_id == "1710500000.000100"
+    # v1 schema fields
+    assert doc_c001.participants == ["alice@co.com"]
+    assert doc_c001.participants_raw == ["U001"]
+    assert doc_c001.channel == "general"
+    assert doc_c001.url == (
+        "https://acme.slack.com/archives/C001/p1710500000000100"
+    )
     assert doc_c001.metadata["channel_id"] == "C001"
     assert doc_c001.metadata["channel_name"] == "general"
+    assert doc_c001.metadata["team_id"] == "T0ACME"
+    assert doc_c001.metadata["team_domain"] == "acme"
 
     # Check a specific document from #engineering
     doc_c002 = next(
-        (d for d in docs if d.doc_id == "slack:C002:1710500060.000200"), None
+        (d for d in docs if d.doc_id == "slack:acme:C002:1710500060.000200"),
+        None,
     )
     assert doc_c002 is not None
     assert doc_c002.title == "#engineering"
-    assert doc_c002.author == "Bob"
+    assert doc_c002.author == "bob@co.com"
     assert doc_c002.content == "Sounds good, I'll prepare a doc."
     assert doc_c002.thread_id is None
+    assert doc_c002.channel == "engineering"
 
     # Verify the API was called correctly
+    mock_auth.assert_called_once()
     mock_users.assert_called_once()
     assert mock_channels.call_count == 1
     # conversations.history called once per channel (2 channels)
     assert mock_history.call_count == 2
 
 
+# ---------------------------------------------------------------------------
+# Test — conversations.list is called with every conversation type so
+# DMs and group DMs auto-sync alongside public/private channels.
+# ---------------------------------------------------------------------------
+
+
+@patch("openjarvis.connectors.slack_connector._slack_api_with_retry")
+def test_conversations_list_requests_all_conversation_types(mock_retry) -> None:
+    """``_slack_api_conversations_list`` widens ``types`` to cover IMs + MPIMs.
+
+    A user connecting Slack expects ``everything I can see is searchable``
+    without a per-channel opt-in — same as Gmail. The proxy for that here
+    is the API request shape: ``types`` must include im + mpim so the bot
+    token's DMs and group DMs come back in the listing.
+    """
+    from openjarvis.connectors.slack_connector import (  # noqa: PLC0415
+        _slack_api_conversations_list,
+    )
+
+    mock_retry.return_value = {"channels": [], "response_metadata": {}}
+    _slack_api_conversations_list("fake-token")
+
+    method, _token, params = mock_retry.call_args.args[:3]
+    assert method == "conversations.list"
+    types = params["types"].split(",")
+    assert set(types) == {"public_channel", "private_channel", "mpim", "im"}
+
+
+# ---------------------------------------------------------------------------
+# Test — sync() yields documents for DMs (im) and group DMs (mpim) too,
+# without requiring conversations.join (no join concept on those types).
+# ---------------------------------------------------------------------------
+
+
+@patch("openjarvis.connectors.slack_connector._slack_api_with_retry")
+@patch("openjarvis.connectors.slack_connector._slack_api_auth_test")
+@patch("openjarvis.connectors.slack_connector._slack_api_conversations_list")
+@patch("openjarvis.connectors.slack_connector._slack_api_conversations_history")
+@patch("openjarvis.connectors.slack_connector._slack_api_users_list")
+def test_sync_includes_dms_and_group_dms(
+    mock_users,
+    mock_history,
+    mock_channels,
+    mock_auth,
+    mock_retry,
+    connector,
+    tmp_path: Path,
+) -> None:
+    """IMs and MPIMs are synced with sensible labels — no join, no membership filter.
+
+    With a user token, ``conversations.list`` already reflects what the
+    user can see. The connector must NOT call ``conversations.join`` for
+    any conversation type, and must NOT skip non-``is_member`` channels.
+    """
+    creds_path = Path(connector._credentials_path)
+    creds_path.write_text(
+        json.dumps({"token": "xoxp-fake-user-token"}), encoding="utf-8"
+    )
+
+    mock_auth.return_value = _AUTH_TEST_RESPONSE
+    mock_users.return_value = _USERS_RESPONSE
+    mock_channels.return_value = {
+        "channels": [
+            {
+                "id": "C001",
+                "name": "general",
+                # No is_member field — user tokens shouldn't depend on it.
+            },
+            {
+                "id": "D001",
+                "is_im": True,
+                "user": "U001",  # 1:1 DM with Alice
+            },
+            {
+                "id": "G001",
+                "name": "mpdm-alice--bob-1",
+                "is_mpim": True,
+            },
+        ],
+        "response_metadata": {"next_cursor": ""},
+    }
+    mock_history.return_value = {
+        "messages": [
+            {
+                "ts": "1710500000.000100",
+                "user": "U001",
+                "text": "context-specific message",
+            },
+        ],
+        "has_more": False,
+    }
+    # _slack_api_with_retry covers any endpoint not individually mocked
+    # (auth.test, users.list, conversations.list, conversations.history
+    # are intercepted above). A call to ``conversations.join`` here would
+    # mean the connector is back to bot-token behavior.
+    mock_retry.return_value = {"ok": False}
+
+    docs: List[Document] = list(connector.sync())
+
+    # One message per conversation × 3 conversations = 3 documents.
+    assert len(docs) == 3
+
+    by_chan_id = {d.metadata["channel_id"]: d for d in docs}
+
+    public_doc = by_chan_id["C001"]
+    assert public_doc.title == "#general"
+    assert public_doc.channel == "general"
+    assert public_doc.metadata["channel_type"] == "public_channel"
+
+    im_doc = by_chan_id["D001"]
+    assert im_doc.title == "DM with Alice"
+    assert im_doc.channel == "dm-Alice"
+    assert im_doc.metadata["channel_type"] == "im"
+
+    mpim_doc = by_chan_id["G001"]
+    assert mpim_doc.title == "#mpdm-alice--bob-1"
+    assert mpim_doc.channel == "mpdm-alice--bob-1"
+    assert mpim_doc.metadata["channel_type"] == "mpim"
+
+    # User-token sync must NEVER call conversations.join (the user is
+    # already in the conversations the listing returns).
+    join_calls = [
+        c
+        for c in mock_retry.call_args_list
+        if c.args and c.args[0] == "conversations.join"
+    ]
+    assert join_calls == []
+
+
 # ---------------------------------------------------------------------------
 # Test 5 — disconnect removes the credentials file
 # ---------------------------------------------------------------------------
@@ -174,7 +341,9 @@ def test_sync_yields_documents(
 def test_disconnect(connector, tmp_path: Path) -> None:
     """disconnect() deletes the credentials file."""
     creds_path = Path(connector._credentials_path)
-    creds_path.write_text(json.dumps({"token": "fake-access-token"}), encoding="utf-8")
+    creds_path.write_text(
+        json.dumps({"token": "xoxp-fake-user-token"}), encoding="utf-8"
+    )
     assert connector.is_connected() is True
 
     connector.disconnect()
@@ -213,3 +382,181 @@ def test_registry() -> None:
     assert ConnectorRegistry.contains("slack")
     cls = ConnectorRegistry.get("slack")
     assert cls.connector_id == "slack"
+
+
+# ---------------------------------------------------------------------------
+# Test 8 — end-to-end: connector → pipeline → KnowledgeStore → HybridSearch
+# ---------------------------------------------------------------------------
+
+
+@patch("openjarvis.connectors.slack_connector._slack_api_auth_test")
+@patch("openjarvis.connectors.slack_connector._slack_api_conversations_list")
+@patch("openjarvis.connectors.slack_connector._slack_api_conversations_history")
+@patch("openjarvis.connectors.slack_connector._slack_api_users_list")
+def test_end_to_end_ingest_and_search(
+    mock_users,
+    mock_history,
+    mock_channels,
+    mock_auth,
+    connector,
+    tmp_path: Path,
+) -> None:
+    """Synced Slack messages are searchable via HybridSearch with v1 fields.
+
+    Lexical-only path (no embedder) so this stays a pure unit test — no
+    Ollama daemon needed. Confirms the v1 contract end-to-end: source,
+    namespaced thread_id, channel, participants, and a workspace-qualified
+    permalink all survive ingest → store → hit.
+    """
+    from openjarvis.connectors.hybrid_search import HybridSearch  # noqa: PLC0415
+    from openjarvis.connectors.pipeline import IngestionPipeline  # noqa: PLC0415
+    from openjarvis.connectors.store import KnowledgeStore  # noqa: PLC0415
+
+    creds_path = Path(connector._credentials_path)
+    creds_path.write_text(
+        json.dumps({"token": "xoxp-fake-user-token"}), encoding="utf-8"
+    )
+    mock_auth.return_value = _AUTH_TEST_RESPONSE
+    mock_users.return_value = _USERS_RESPONSE
+    mock_channels.return_value = _CHANNELS_RESPONSE
+    mock_history.return_value = _HISTORY_RESPONSE
+
+    store = KnowledgeStore(db_path=tmp_path / "slack_e2e.db")
+    pipeline = IngestionPipeline(store)
+    chunks_stored = pipeline.ingest(connector.sync())
+
+    # 4 short messages → 4 chunks (no chunk splitting at this length).
+    assert chunks_stored == 4
+
+    hybrid = HybridSearch(store)
+    hits = hybrid.search("API redesign", limit=5)
+    assert len(hits) >= 1
+
+    target = next(
+        (h for h in hits if "API redesign" in h.content_snippet),
+        None,
+    )
+    assert target is not None
+    assert target.source == "slack"
+    assert target.title == "#general"
+    # Thread id is namespaced by the pipeline.
+    assert target.thread_id == "slack:1710500000.000100"
+    assert target.participants == ["alice@co.com"]
+    # The stored doc_id flows through the hit and is the format the
+    # research-loop URL builder expects.
+    assert target.document_id == "slack:acme:C001:1710500000.000100"
+
+    # And the research-loop builder reconstructs the workspace permalink.
+    from openjarvis.agents.research_loop import _hit_url  # noqa: PLC0415
+
+    assert _hit_url(target.source, target.document_id) == (
+        "https://acme.slack.com/archives/C001/p1710500000000100"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Test — bot tokens (xoxb-) are rejected with a clear error at connect time.
+# ---------------------------------------------------------------------------
+
+
+def test_handle_callback_rejects_xoxb(connector) -> None:
+    """handle_callback() refuses bot tokens before writing them to disk."""
+    from openjarvis.connectors.slack_connector import (  # noqa: PLC0415
+        SlackTokenError,
+    )
+
+    with pytest.raises(SlackTokenError) as excinfo:
+        connector.handle_callback("xoxb-some-bot-token")
+
+    msg = str(excinfo.value).lower()
+    assert "xoxb" in msg
+    assert "user oauth token" in msg or "xoxp" in msg
+    # Token must not have been persisted on rejection.
+    assert not Path(connector._credentials_path).exists()
+
+
+def test_handle_callback_rejects_unknown_prefix(connector) -> None:
+    """handle_callback() refuses tokens that don't match the user-token shape."""
+    from openjarvis.connectors.slack_connector import (  # noqa: PLC0415
+        SlackTokenError,
+    )
+
+    with pytest.raises(SlackTokenError):
+        connector.handle_callback("xapp-app-level-token")
+    assert not Path(connector._credentials_path).exists()
+
+
+# ---------------------------------------------------------------------------
+# Test — a stored bot token surfaces an error in sync_status and yields
+# zero documents (defensive guard for credentials written before this
+# migration).
+# ---------------------------------------------------------------------------
+
+
+def test_sync_with_xoxb_token_surfaces_error(connector, tmp_path: Path) -> None:
+    """A leftover bot token must produce an explanatory sync error."""
+    creds_path = Path(connector._credentials_path)
+    creds_path.write_text(json.dumps({"token": "xoxb-bot-token"}), encoding="utf-8")
+
+    docs = list(connector.sync())
+
+    assert docs == []
+    status = connector.sync_status()
+    assert status.error is not None
+    assert "xoxb" in status.error.lower() or "user oauth token" in status.error.lower()
+    assert "xoxp" in status.error.lower()
+
+
+# ---------------------------------------------------------------------------
+# Test — sync logs a per-type channel count before fetching history so
+# operators can see at a glance what the token has access to.
+# ---------------------------------------------------------------------------
+
+
+@patch("openjarvis.connectors.slack_connector._slack_api_auth_test")
+@patch("openjarvis.connectors.slack_connector._slack_api_conversations_list")
+@patch("openjarvis.connectors.slack_connector._slack_api_conversations_history")
+@patch("openjarvis.connectors.slack_connector._slack_api_users_list")
+def test_sync_logs_per_type_channel_counts(
+    mock_users,
+    mock_history,
+    mock_channels,
+    mock_auth,
+    connector,
+    tmp_path: Path,
+    caplog,
+) -> None:
+    """Each sync logs ``Found X public, Y private, Z DMs, W group DMs``.
+
+    The diagnostic is the single fastest way to tell whether a token's
+    scopes are correct without grepping per-channel logs.
+    """
+    creds_path = Path(connector._credentials_path)
+    creds_path.write_text(
+        json.dumps({"token": "xoxp-fake-user-token"}), encoding="utf-8"
+    )
+
+    mock_auth.return_value = _AUTH_TEST_RESPONSE
+    mock_users.return_value = _USERS_RESPONSE
+    mock_channels.return_value = {
+        "channels": [
+            {"id": "C001", "name": "public-1"},
+            {"id": "C002", "name": "public-2"},
+            {"id": "C003", "name": "public-3"},
+            {"id": "P001", "name": "private-1", "is_private": True},
+            {"id": "D001", "is_im": True, "user": "U001"},
+            {"id": "D002", "is_im": True, "user": "U002"},
+            {"id": "G001", "name": "mpdm-x", "is_mpim": True},
+        ],
+        "response_metadata": {"next_cursor": ""},
+    }
+    mock_history.return_value = {"messages": [], "has_more": False}
+
+    with caplog.at_level("INFO", logger="openjarvis.connectors.slack_connector"):
+        list(connector.sync())
+
+    summary = " ".join(r.message for r in caplog.records)
+    assert "Found 3 public channels" in summary
+    assert "1 private channels" in summary
+    assert "2 DMs" in summary
+    assert "1 group DMs" in summary