feat(connectors): index Teams meeting transcripts as their own documents (#14863)

backend/onyx/connectors/teams/connector.py

@@ -38,10 +38,16 @@ from onyx.connectors.models import (
 from onyx.connectors.teams import listing
 from onyx.connectors.teams.files import FileSource
 from onyx.connectors.teams.models import ChannelRef
+from onyx.connectors.teams.organizers import (
+    Organizer,
+    OrganizerSource,
+    OrganizerStage,
+)
 from onyx.connectors.teams.refusals import channel_failure, is_permanent, status
 from onyx.connectors.teams.session import TeamsSession
 from onyx.connectors.teams.sources import SlimWalk
 from onyx.connectors.teams.threads import ThreadSource
+from onyx.connectors.teams.transcripts import TranscriptSource
 from onyx.connectors.teams.utils import ChannelFilesUnavailable
 from onyx.indexing.indexing_heartbeat import IndexingHeartbeatInterface
 from onyx.utils.batching import batch_generator
@@ -52,9 +58,6 @@ logger = setup_logger()
 
 _SLIM_DOC_BATCH_SIZE = 5000
 
-# Pruning and permission sync both run the slim walk under this label.
-_SLIM_WALK = "retrieve_all_slim_docs_perm_sync"
-
 
 class TeamsCheckpoint(ConnectorCheckpoint):
     # None until the teams are listed.
@@ -64,6 +67,12 @@ class TeamsCheckpoint(ConnectorCheckpoint):
     # a page instead of a whole team. No page url means the channel's first page.
     current_channel: ChannelRef | None = None
     next_messages_url: str | None = None
+    # The meeting organizers follow the channels. None until their first page is
+    # listed, then a batch of organizers per step and a page at a time. Whole
+    # organizers ride along, not ids: the documents name the organizer, and a
+    # resumed attempt would otherwise read every user again. A page is 100.
+    todo_organizers: list[Organizer] | None = None
+    next_organizers_url: str | None = None
 
 
 class TeamsConnector(
@@ -74,7 +83,8 @@ class TeamsConnector(
 ):
     """Walks the teams, then each team's channels a page at a time. What a
     channel holds is read by one source per content type: its threads, and its
-    files when attachments are on."""
+    files when attachments are on. What a scheduled meeting leaves behind is
+    listed per organizer, so those sources follow the channels."""
 
     MAX_WORKERS = 10
 
@@ -91,12 +101,19 @@ class TeamsConnector(
         include_attachments: bool = False,
         # Off by default: every pasted image is a download and a vision call.
         include_inline_images: bool = False,
+        # Off by default: transcripts need three more grants, a tenant setting
+        # and an application access policy. Empty organizers means every
+        # enabled user with a Teams license.
+        include_meeting_transcripts: bool = False,
+        transcript_organizers: list[str] | None = None,
     ) -> None:
         TeamsSession.__init__(self, graph_api_host, authority_host)
         self.max_workers = max_workers
         self.requested_team_list: list[str] = teams or []
         self.include_attachments = include_attachments
         self.include_inline_images = include_inline_images
+        self.include_meeting_transcripts = include_meeting_transcripts
+        self.transcript_organizers: list[str] = transcript_organizers or []
         # Channels walked again from their first page in this attempt: a saved
         # page url Graph rejects recovers once per attempt and can never loop.
         self._restarted_channel_ids: set[str] = set()
@@ -106,6 +123,16 @@ class TeamsConnector(
             if include_attachments
             else None
         )
+        organizer_sources: list[OrganizerSource] = []
+        if include_meeting_transcripts:
+            organizer_sources.append(
+                TranscriptSource(self, covers_every_user=not self.transcript_organizers)
+            )
+        self._organizers: OrganizerStage | None = (
+            OrganizerStage(self, self.transcript_organizers, organizer_sources)
+            if organizer_sources
+            else None
+        )
 
     # impls for BaseConnector
 
@@ -195,6 +222,8 @@ class TeamsConnector(
 
         if self._files is not None:
             self._files.validate(list(validation_query))
+        if self._organizers is not None:
+            self._organizers.validate()
 
     # impls for CheckpointedConnector
 
@@ -209,7 +238,7 @@ class TeamsConnector(
     def load_from_checkpoint(
         self,
         start: SecondsSinceUnixEpoch,
-        end: SecondsSinceUnixEpoch,  # noqa: ARG002
+        end: SecondsSinceUnixEpoch,
         checkpoint: TeamsCheckpoint,
     ) -> CheckpointOutput[TeamsCheckpoint]:
         graph_client = self.graph()
@@ -222,8 +251,11 @@ class TeamsConnector(
                 requested=self.requested_team_list,
             )
             checkpoint.todo_team_ids = [team.id for team in teams if team.id]
-        elif checkpoint.current_channel is None and not checkpoint.todo_channels:
-            # A team is left, or has_more would have ended the walk.
+        elif checkpoint.current_channel is not None or checkpoint.todo_channels:
+            if checkpoint.current_channel is None:
+                checkpoint.current_channel = checkpoint.todo_channels.pop()
+            yield from self._walk_channel_page(checkpoint, start)
+        elif checkpoint.todo_team_ids:
             team_id = checkpoint.todo_team_ids.pop()
             team = listing.get_team_by_id(graph_client=graph_client, team_id=team_id)
             checkpoint.todo_channels = [
@@ -236,15 +268,29 @@ class TeamsConnector(
                 team_id,
                 len(checkpoint.todo_team_ids),
             )
-        else:
-            if checkpoint.current_channel is None:
-                checkpoint.current_channel = checkpoint.todo_channels.pop()
-            yield from self._walk_channel_page(checkpoint, start)
+        elif self._organizers is not None and checkpoint.todo_organizers:
+            yield from self._organizers.index_batch(
+                checkpoint.todo_organizers, start, end
+            )
+        elif self._organizers is not None and (
+            checkpoint.todo_organizers is None or checkpoint.next_organizers_url
+        ):
+            checkpoint.todo_organizers, checkpoint.next_organizers_url = (
+                self._organizers.next_page(checkpoint.next_organizers_url)
+            )
 
         checkpoint.has_more = bool(
             checkpoint.current_channel
             or checkpoint.todo_channels
             or checkpoint.todo_team_ids
+            or (
+                self._organizers is not None
+                and (
+                    checkpoint.todo_organizers is None
+                    or checkpoint.todo_organizers
+                    or checkpoint.next_organizers_url
+                )
+            )
         )
         return checkpoint
 
@@ -400,17 +446,14 @@ class TeamsConnector(
             yield from batch_generator(
                 slim_docs,
                 _SLIM_DOC_BATCH_SIZE,
-                pre_batch_yield=lambda _: _slim_batch_signals(callback),
+                pre_batch_yield=lambda _: walk.batch_signals(),
+            )
+        if self._organizers is not None:
+            yield from batch_generator(
+                self._organizers.slim(walk),
+                _SLIM_DOC_BATCH_SIZE,
+                pre_batch_yield=lambda _: walk.batch_signals(),
             )
-
-
-def _slim_batch_signals(callback: IndexingHeartbeatInterface | None) -> None:
-    """The stop and progress signals the runner gets before every batch."""
-    if callback is None:
-        return
-    if callback.should_stop():
-        raise RuntimeError(f"{_SLIM_WALK}: Stop signal detected")
-    callback.progress(_SLIM_WALK, 1)
 
 
 def _rejects_saved_cursor(

backend/onyx/connectors/teams/listing.py

@@ -11,24 +11,12 @@ from office365.teams.channels.channel import Channel
 from office365.teams.team import Team
 
 from onyx.connectors.teams.models import ChannelRef
-from onyx.connectors.teams.utils import execute_query_with_retry
+from onyx.connectors.teams.utils import escape_odata_string, execute_query_with_retry
 from onyx.utils.logger import setup_logger
 
 logger = setup_logger()
 
 
-def _escape_odata_string(name: str) -> str:
-    """Escape special characters for OData string literals.
-
-    Uses proper OData v4 string literal escaping:
-    - Single quotes: ' becomes ''
-    - Other characters are handled by using contains() instead of eq for problematic cases
-    """
-    # Escape single quotes for OData syntax (replace ' with '')
-    escaped = name.replace("'", "''")
-    return escaped
-
-
 def has_odata_incompatible_chars(team_names: list[str] | None) -> bool:
     """Check if any team name contains characters that break Microsoft Graph OData filters.
 
@@ -75,7 +63,7 @@ def _build_simple_odata_filter(safe_names: list[str]) -> str | None:
 
     filter_parts = []
     for name in safe_names:
-        escaped_name = _escape_odata_string(name)
+        escaped_name = escape_odata_string(name)
         filter_parts.append(f"displayName eq '{escaped_name}'")
 
     return " or ".join(filter_parts)

backend/onyx/connectors/teams/organizers.py

@@ -0,0 +1,280 @@
+"""Meeting organizers: Graph lists what a scheduled meeting leaves behind (its
+transcript, its chat) per user, so those content types are walked organizer by
+organizer, after the channels and several organizers at a time."""
+
+import abc
+from collections.abc import Callable, Generator, Iterator
+from typing import Any
+from urllib.parse import quote
+
+import requests
+from office365.graph_client import GraphClient
+from pydantic import BaseModel
+
+from onyx.connectors.exceptions import (
+    ConnectorValidationError,
+    InsufficientPermissionsError,
+    UnexpectedValidationError,
+)
+from onyx.connectors.interfaces import SecondsSinceUnixEpoch
+from onyx.connectors.models import (
+    BasicExpertInfo,
+    ConnectorFailure,
+    Document,
+    SlimDocument,
+)
+from onyx.connectors.teams.refusals import graph_said, status
+from onyx.connectors.teams.session import TeamsSession
+from onyx.connectors.teams.sources import SLIM_WALK, SlimWalk
+from onyx.connectors.teams.utils import (
+    GraphRetriesExhausted,
+    escape_odata_string,
+    get_json_with_retry,
+    iter_values,
+    next_page_url,
+)
+from onyx.utils.batching import batch_generator
+from onyx.utils.logger import setup_logger
+from onyx.utils.threadpool_concurrency import parallel_yield
+
+logger = setup_logger()
+
+# Graph lists per organizer and a listing costs the same with or without
+# content, so a tenant is walked several organizers at a time. Eight measured 7
+# times faster with no throttling, sixteen only 9 with slower calls.
+ORGANIZER_WORKERS = 8
+
+# A page of organizers rides in the indexing checkpoint, so pages stay small.
+USER_PAGE_SIZE = 100
+# Only a user with a Teams plan can organize a meeting, and a tenant can hold
+# thousands of accounts without one. Graph filters on assigned plans only as an
+# advanced query, which is a count and this header on every page.
+ADVANCED_QUERY = {"ConsistencyLevel": "eventual"}
+ORGANIZERS_URL = (
+    "users?$select=id,userPrincipalName,mail,displayName"
+    "&$filter=accountEnabled eq true and assignedPlans/any("
+    "p:p/service eq 'TeamspaceAPI' and p/capabilityStatus eq 'Enabled')"
+    f"&$count=true&$top={USER_PAGE_SIZE}"
+)
+
+
+class Organizer(BaseModel):
+    """A user whose scheduled meetings are exported."""
+
+    id: str
+    email: str | None
+    display_name: str | None
+
+    @classmethod
+    def from_graph(cls, row: dict[str, Any]) -> "Organizer":
+        return cls(
+            id=row["id"],
+            email=(row.get("mail") or row.get("userPrincipalName") or None),
+            display_name=row.get("displayName"),
+        )
+
+
+def iter_organizers(
+    graph_client: GraphClient,
+    principal_names: list[str],
+    before_page: Callable[[], None] | None = None,
+) -> Generator[Organizer]:
+    """The configured users, or every enabled user with a Teams plan when none
+    are configured. A configured name that resolves to nothing raises.
+    ``before_page`` runs ahead of each user page request."""
+    if principal_names:
+        yield from _resolve_organizers(graph_client, principal_names)
+        return
+    for row in iter_values(graph_client, ORGANIZERS_URL, before_page, ADVANCED_QUERY):
+        yield Organizer.from_graph(row)
+
+
+def fetch_organizer_page(
+    graph_client: GraphClient, principal_names: list[str], page_url: str | None
+) -> tuple[list[Organizer], str | None]:
+    """One page of enabled users with a Teams plan from ``page_url`` (the first
+    page when None) and the link to the next. Configured names are one page of
+    their own."""
+    if principal_names:
+        return _resolve_organizers(graph_client, principal_names), None
+    page_url = page_url or ORGANIZERS_URL
+    json_response = get_json_with_retry(graph_client, page_url, ADVANCED_QUERY)
+    organizers = [
+        Organizer.from_graph(row)
+        for row in json_response.get("value", [])
+        if isinstance(row, dict)
+    ]
+    return organizers, next_page_url(graph_client, json_response, page_url)
+
+
+def _resolve_organizers(
+    graph_client: GraphClient, principal_names: list[str]
+) -> list[Organizer]:
+    # Eager on purpose: every configured name is resolved even when the caller
+    # wants one, so a misspelled name fails at setup and not at index time. A
+    # name goes into an OData string literal, so its apostrophes double.
+    return [
+        Organizer.from_graph(
+            get_json_with_retry(
+                graph_client,
+                f"users('{quote(escape_odata_string(name), safe='@.')}')"
+                "?$select=id,userPrincipalName,mail,displayName",
+            )
+        )
+        for name in principal_names
+    ]
+
+
+def organizer_expert(organizer: Organizer) -> list[BasicExpertInfo]:
+    if not organizer.email:
+        return []
+    return [BasicExpertInfo(display_name=organizer.display_name, email=organizer.email)]
+
+
+class OrganizerSource(abc.ABC):
+    """One content type Graph lists per meeting organizer."""
+
+    # The form option that turns the source on, for setup messages.
+    option: str
+
+    @abc.abstractmethod
+    def validate(self, organizer: Organizer) -> None:
+        """Reads what indexing reads for one organizer and names what is
+        missing: a grant, a tenant setting, a policy."""
+
+    @abc.abstractmethod
+    def index(
+        self,
+        organizer: Organizer,
+        start: SecondsSinceUnixEpoch,
+        end: SecondsSinceUnixEpoch,
+    ) -> Iterator[Document | ConnectorFailure]:
+        """The organizer's documents that changed inside the window. A refusal
+        that stays is recorded, anything else raises."""
+
+    @abc.abstractmethod
+    def slim(self, organizer: Organizer, walk: SlimWalk) -> Iterator[SlimDocument]:
+        """Every document id the indexing walk writes for the organizer. An
+        organizer the app may no longer read lists nothing, so their documents
+        are pruned like any document the credential lost."""
+
+
+class OrganizerStage:
+    """Walks the organizers for every source that is on."""
+
+    def __init__(
+        self,
+        session: TeamsSession,
+        principal_names: list[str],
+        sources: list[OrganizerSource],
+    ) -> None:
+        self._session = session
+        self._principal_names = principal_names
+        self._sources = sources
+        # A saved page url Graph rejects recovers once per attempt, or a token
+        # it always rejects would list the first page for ever.
+        self._restarted_listing = False
+
+    def validate(self) -> None:
+        """One organizer is enough to probe with. Configured names are all
+        resolved on the way, since resolving them is not lazy."""
+        options = " and ".join(source.option for source in self._sources)
+        try:
+            organizer = next(
+                iter_organizers(self._session.graph(), self._principal_names), None
+            )
+        except requests.HTTPError as e:
+            if status(e) == 404 and self._principal_names:
+                raise ConnectorValidationError(
+                    "No user matches a configured organizer: Graph answered "
+                    f"{status(e)}. {graph_said(e)}"
+                )
+            if status(e) in (401, 403):
+                raise InsufficientPermissionsError(
+                    f"{options} needs the User.Read.All application permission to "
+                    f"list organizers ({status(e)})."
+                )
+            raise UnexpectedValidationError(f"Could not list organizers: {e}")
+        except (GraphRetriesExhausted, requests.RequestException) as e:
+            raise UnexpectedValidationError(f"Could not list organizers: {e}")
+        if organizer is None:
+            raise UnexpectedValidationError(
+                f"No enabled user with a Teams license to check {options} on. "
+                "Configure an organizer."
+            )
+        for source in self._sources:
+            source.validate(organizer)
+
+    def next_page(self, page_url: str | None) -> tuple[list[Organizer], str | None]:
+        """The next page of organizers for the checkpoint, which is saved after
+        every step and so never carries a whole tenant's directory."""
+        graph_client = self._session.graph()
+        try:
+            organizers, next_url = fetch_organizer_page(
+                graph_client, self._principal_names, page_url
+            )
+        except requests.HTTPError as e:
+            # Graph answers a skip token it no longer honors with 400 or 410.
+            if (
+                page_url is None
+                or self._restarted_listing
+                or status(e) not in (400, 410)
+            ):
+                raise
+            self._restarted_listing = True
+            logger.warning(
+                "The saved organizer page is no longer honored, listing the "
+                "organizers from the start: %s",
+                e,
+            )
+            organizers, next_url = fetch_organizer_page(
+                graph_client, self._principal_names, None
+            )
+        logger.info("Listed %s meeting organizer(s)", len(organizers))
+        return organizers, next_url
+
+    def index_batch(
+        self,
+        todo: list[Organizer],
+        start: SecondsSinceUnixEpoch,
+        end: SecondsSinceUnixEpoch,
+    ) -> Iterator[Document | ConnectorFailure]:
+        """Takes one batch of organizers off ``todo`` and indexes them side by
+        side. The Graph client is shared: a direct request builds its own
+        options, so threads do not collide."""
+        batch = todo[-ORGANIZER_WORKERS:]
+        del todo[-ORGANIZER_WORKERS:]
+        yield from parallel_yield(
+            [self._index_one(organizer, start, end) for organizer in batch],
+            max_workers=ORGANIZER_WORKERS,
+        )
+
+    def slim(self, walk: SlimWalk) -> Iterator[SlimDocument]:
+        organizers = iter_organizers(
+            self._session.graph(),
+            self._principal_names,
+            before_page=walk.raise_if_stopped,
+        )
+        for batch in batch_generator(organizers, ORGANIZER_WORKERS):
+            # One stop check and one progress report per batch. Each source
+            # honors a stop before every page of its own.
+            walk.raise_if_stopped()
+            if walk.callback:
+                walk.callback.progress(SLIM_WALK, len(batch))
+            yield from parallel_yield(
+                [self._slim_one(organizer, walk) for organizer in batch],
+                max_workers=ORGANIZER_WORKERS,
+            )
+
+    def _index_one(
+        self,
+        organizer: Organizer,
+        start: SecondsSinceUnixEpoch,
+        end: SecondsSinceUnixEpoch,
+    ) -> Iterator[Document | ConnectorFailure]:
+        for source in self._sources:
+            yield from source.index(organizer, start, end)
+
+    def _slim_one(self, organizer: Organizer, walk: SlimWalk) -> Iterator[SlimDocument]:
+        for source in self._sources:
+            yield from source.slim(organizer, walk)

backend/onyx/connectors/teams/refusals.py

@@ -4,6 +4,7 @@ attempt so it is retried."""
 
 from collections.abc import Iterator
 from contextlib import contextmanager
+from typing import Any
 
 import requests
 from office365.runtime.client_request_exception import ClientRequestException
@@ -80,3 +81,39 @@ def channel_failure(
         ),
         exception=error,
     )
+
+
+def _error_body(error: requests.HTTPError) -> dict[str, Any]:
+    """Graph's error object, empty when the body is not one."""
+    if error.response is None:
+        return {}
+    try:
+        payload = error.response.json()
+    except ValueError:
+        return {}
+    body = payload.get("error") if isinstance(payload, dict) else None
+    return body if isinstance(body, dict) else {}
+
+
+def graph_inner_error_code(error: requests.HTTPError) -> str:
+    """Graph's inner error code, or its outer code, or the empty string. The
+    inner one comes first: it names the cause, where the outer one repeats the
+    status."""
+    body = _error_body(error)
+    inner = body.get("innerError")
+    if isinstance(inner, dict) and inner.get("code"):
+        return str(inner["code"])
+    return str(body.get("code") or "")
+
+
+def graph_error_message(error: requests.HTTPError) -> str:
+    return str(_error_body(error).get("message") or "")
+
+
+def graph_said(error: requests.HTTPError) -> str:
+    """Graph's own code and message. A refusal the connector cannot name is
+    otherwise unreadable, and a missing grant is only its most common cause."""
+    return (
+        f"Graph said: {graph_inner_error_code(error) or 'no code'}, "
+        f"{graph_error_message(error) or 'no message'}"
+    )

backend/onyx/connectors/teams/session.py

@@ -1,6 +1,7 @@
 """What every Teams content source reads the tenant through: the app's
 credential, the Graph client built from it, and the endpoints of its cloud."""
 
+import threading
 from collections.abc import Callable
 from typing import Any
 
@@ -15,6 +16,7 @@ from onyx.connectors.microsoft_utils.graph_auth import (
 from onyx.connectors.microsoft_utils.graph_client import GraphApiClient
 from onyx.connectors.microsoft_utils.graph_env import resolve_microsoft_environment
 from onyx.connectors.models import ConnectorMissingCredentialError
+from onyx.connectors.teams.utils import UserDirectory
 
 CREDENTIAL_AUTH_METHOD = "authentication_method"
 CREDENTIAL_PRIVATE_KEY = "teams_private_key"
@@ -29,6 +31,8 @@ class TeamsSession:
         self._auth_method = MicrosoftAuthMethod.CLIENT_SECRET
         # Granted by the factory from the image analysis setting.
         self.allow_images = False
+        self._user_directory: UserDirectory | None = None
+        self._directory_lock = threading.Lock()
 
         resolved_env = resolve_microsoft_environment(graph_api_host, authority_host)
         self._azure_environment = resolved_env.environment
@@ -84,6 +88,14 @@ class TeamsSession:
     def graph_root(self) -> str:
         return f"{self.graph_api_host}/v1.0"
 
+    def directory(self) -> UserDirectory:
+        """The run's one user directory, also when workers ask for it at once. It
+        lists nothing until a member without an email asks for a name."""
+        with self._directory_lock:
+            if self._user_directory is None:
+                self._user_directory = UserDirectory(self.graph())
+            return self._user_directory
+
     def graph(self) -> GraphClient:
         if self.graph_client is None:
             raise ConnectorMissingCredentialError("Teams")

backend/onyx/connectors/teams/sources.py

@@ -2,9 +2,16 @@
 
 from dataclasses import dataclass
 
+import requests
+
 from onyx.connectors.interfaces import SecondsSinceUnixEpoch
+from onyx.connectors.teams.refusals import is_permanent
 from onyx.indexing.indexing_heartbeat import IndexingHeartbeatInterface
 
+# Pruning and permission sync both run the slim walk, so its signals name the
+# walk and not either caller.
+SLIM_WALK = "teams_slim_walk"
+
 
 @dataclass
 class SlimWalk:
@@ -14,3 +21,32 @@ class SlimWalk:
     start: SecondsSinceUnixEpoch
     callback: IndexingHeartbeatInterface | None
     with_readers: bool
+
+    def raise_if_stopped(self) -> None:
+        if self.callback and self.callback.should_stop():
+            raise RuntimeError(f"{SLIM_WALK}: Stop signal detected")
+
+    def batch_signals(self) -> None:
+        """The stop and progress signals the runner gets before every batch."""
+        self.raise_if_stopped()
+        if self.callback:
+            self.callback.progress(SLIM_WALK, 1)
+
+
+@dataclass
+class PagedListing:
+    """One paged listing of a walk. Refused at its first request, the app lost
+    access: the listing is empty, so pruning removes what it held and indexing
+    records nothing. Refused after it answered, the listing broke and the rest
+    still exists, so the attempt fails and the next one decides."""
+
+    walk: SlimWalk | None = None
+    pages: int = 0
+
+    def before_page(self) -> None:
+        if self.walk is not None:
+            self.walk.raise_if_stopped()
+        self.pages += 1
+
+    def lost_access(self, error: requests.RequestException) -> bool:
+        return self.pages <= 1 and is_permanent(error)

backend/onyx/connectors/teams/threads.py

@@ -165,6 +165,7 @@ class ThreadSource:
                 graph_client=self._session.graph(),
                 team_id=channel.team_id,
                 channel_id=channel.id,
+                directory=self._session.directory(),
             )
         return self._readers[channel.id]
 
@@ -233,6 +234,7 @@ class ThreadSource:
                     graph_client=self._session.graph(),
                     team_id=channel.team_id,
                     channel_id=channel.id,
+                    directory=self._session.directory(),
                 )
         messages = fetch_messages(
             graph_client=self._session.graph(),

backend/onyx/connectors/teams/transcripts.py

@@ -0,0 +1,462 @@
+"""Meeting transcripts: Graph exports them per organizer for scheduled meetings
+(never channel meetings), behind the OnlineMeetingTranscript.Read.All grant, the
+tenant's transcript API access setting and an application access policy that
+names the app for the organizer. Each transcript is a document of its own,
+readable by the organizer and the people the meeting record lists."""
+
+import re
+import time
+from collections.abc import Callable, Generator, Iterator
+from datetime import datetime, timezone
+from typing import Any
+
+import requests
+from office365.graph_client import GraphClient
+from pydantic import BaseModel
+
+from onyx.access.models import ExternalAccess
+from onyx.configs.constants import DocumentSource
+from onyx.connectors.exceptions import (
+    ConnectorValidationError,
+    InsufficientPermissionsError,
+    UnexpectedValidationError,
+)
+from onyx.connectors.interfaces import SecondsSinceUnixEpoch
+from onyx.connectors.models import (
+    ConnectorFailure,
+    Document,
+    DocumentFailure,
+    EntityFailure,
+    SlimDocument,
+    TextSection,
+)
+from onyx.connectors.teams.organizers import (
+    Organizer,
+    OrganizerSource,
+    organizer_expert,
+)
+from onyx.connectors.teams.refusals import (
+    graph_error_message,
+    graph_inner_error_code,
+    graph_said,
+    is_permanent,
+    status,
+)
+from onyx.connectors.teams.session import TeamsSession
+from onyx.connectors.teams.sources import PagedListing, SlimWalk
+from onyx.connectors.teams.utils import (
+    GraphRetriesExhausted,
+    get_json_with_retry,
+    iter_values,
+    request_with_retry,
+)
+from onyx.file_processing.webvtt import is_timing_line, parse_vtt_transcript
+from onyx.utils.logger import setup_logger
+
+logger = setup_logger()
+
+TRANSCRIPT_DOCUMENT_ID_PREFIX = "teams-transcript:"
+
+ATTRIBUTED_FORMAT = "text/vtt"
+# The only way to ask for a transcript without speakers: the $format parameter
+# does not accept this type.
+UNATTRIBUTED_FORMAT = "application/vnd.microsoft.graph.transcript+text"
+
+# Graph's inner error codes. Branch on these, the messages change.
+TRANSCRIPT_ACCESS_DISABLED_CODE = "GraphAccessToTranscriptsDisabled"
+SPEAKER_ATTRIBUTION_DISABLED_CODE = "SpeakerAttributionNotAllowed"
+# A missing application access policy has no code of its own, only its text:
+# the first is how a meeting record is refused, the second a transcript's content.
+ACCESS_POLICY_MESSAGES = (
+    "application access policy",
+    "not allowed to perform operations on the user",
+)
+
+TRANSCRIPT_PAGE_SIZE = 50
+# Graph walks the listing newest first in slices of about 35 days, about 2
+# seconds each with or without transcripts, and answers 404 past the 13th from
+# now. Six months keeps a walk short, and every walk shares it so pruning agrees.
+TRANSCRIPT_LOOKBACK_S = 183 * 24 * 60 * 60
+
+_BLANK_LINE_RE = re.compile(r"\n[ \t]*\n")
+
+
+def transcript_document_id(organizer_id: str, transcript_id: str) -> str:
+    """Transcripts are listed per organizer, so an id says which listing it
+    came from."""
+    return f"{TRANSCRIPT_DOCUMENT_ID_PREFIX}{organizer_id}:{transcript_id}"
+
+
+class Transcript(BaseModel):
+    id: str
+    meeting_id: str
+    created: datetime
+    content_url: str
+
+    @classmethod
+    def from_graph(cls, row: dict[str, Any]) -> "Transcript":
+        return cls(
+            id=row["id"],
+            meeting_id=row["meetingId"],
+            created=datetime.fromisoformat(row["createdDateTime"]),
+            content_url=row["transcriptContentUrl"],
+        )
+
+
+class MeetingRecord(BaseModel):
+    """What the meeting itself tells: its name and who was in it."""
+
+    subject: str | None
+    start: datetime | None
+    join_web_url: str | None
+    participant_emails: set[str]
+
+    @classmethod
+    def from_graph(cls, row: dict[str, Any]) -> "MeetingRecord":
+        participants = row.get("participants") or {}
+        people = [
+            participants.get("organizer") or {},
+            *(participants.get("attendees") or []),
+        ]
+        emails = {upn.lower() for person in people if (upn := person.get("upn"))}
+        start = row.get("startDateTime")
+        return cls(
+            subject=row.get("subject") or None,
+            start=datetime.fromisoformat(start) if start else None,
+            join_web_url=row.get("joinWebUrl") or None,
+            participant_emails=emails,
+        )
+
+
+def transcripts_disabled(error: requests.HTTPError) -> bool:
+    """The tenant setting that turns transcript export off for every app.
+    Nothing on our side cures it, so callers stop rather than move on."""
+    return graph_inner_error_code(error) == TRANSCRIPT_ACCESS_DISABLED_CODE
+
+
+def access_policy_missing(error: requests.HTTPError) -> bool:
+    """The organizer is outside the application access policy naming this app."""
+    message = graph_error_message(error).lower()
+    return any(marker in message for marker in ACCESS_POLICY_MESSAGES)
+
+
+def _graph_timestamp(moment: SecondsSinceUnixEpoch) -> str:
+    return datetime.fromtimestamp(moment, tz=timezone.utc).strftime(
+        "%Y-%m-%dT%H:%M:%SZ"
+    )
+
+
+def fetch_transcripts(
+    graph_client: GraphClient,
+    organizer_id: str,
+    start: SecondsSinceUnixEpoch | None,
+    end: SecondsSinceUnixEpoch | None,
+    page_size: int = TRANSCRIPT_PAGE_SIZE,
+    before_page: Callable[[], None] | None = None,
+) -> Generator[Transcript]:
+    """The transcripts of the meetings this user organized, created inside the
+    window when one is given and never further back than the lookback. Graph
+    pages the listing itself, ``before_page`` runs ahead of each page request."""
+    floor = time.time() - TRANSCRIPT_LOOKBACK_S
+    if end is not None and end <= floor:
+        return
+    parameters = [
+        f"meetingOrganizerUserId='{organizer_id}'",
+        f"startDateTime={_graph_timestamp(max(start or 0, floor))}",
+    ]
+    if end is not None:
+        parameters.append(f"endDateTime={_graph_timestamp(end)}")
+    url = (
+        f"users/{organizer_id}/onlineMeetings/getAllTranscripts"
+        f"({','.join(parameters)})?$top={page_size}"
+    )
+    for row in iter_values(graph_client, url, before_page):
+        yield Transcript.from_graph(row)
+
+
+def fetch_meeting(
+    graph_client: GraphClient, organizer_id: str, meeting_id: str
+) -> MeetingRecord:
+    """Needs OnlineMeetings.Read.All and the same access policy as the transcripts."""
+    return MeetingRecord.from_graph(
+        get_json_with_retry(
+            graph_client,
+            f"users/{organizer_id}/onlineMeetings/{meeting_id}"
+            "?$select=subject,startDateTime,joinWebUrl,participants",
+        )
+    )
+
+
+def fetch_transcript_text(
+    graph_client: GraphClient, content_url: str
+) -> tuple[str, bool]:
+    """The transcript as text and whether it names its speakers. A tenant that
+    disallows speaker attribution refuses the attributed format and serves the
+    plain one, which carries the same speech without the names."""
+    request_url = content_url.removeprefix(
+        graph_client.service_root_url()
+    ).removeprefix("/")
+    try:
+        return request_with_retry(
+            graph_client, request_url, {"Accept": ATTRIBUTED_FORMAT}
+        ).text, True
+    except requests.HTTPError as e:
+        if graph_inner_error_code(e) != SPEAKER_ATTRIBUTION_DISABLED_CODE:
+            raise
+    return request_with_retry(
+        graph_client, request_url, {"Accept": UNATTRIBUTED_FORMAT}
+    ).text, False
+
+
+def transcript_text(content: str, attributed: bool) -> str:
+    """The spoken text. The attributed format is WebVTT with voice spans. The
+    unattributed one puts a blank line between a cue's timing and its speech,
+    which the WebVTT parser reads as a cue with no text and a block with no
+    timing and keeps neither, so its blocks are read in turn."""
+    if attributed:
+        return parse_vtt_transcript(content, keep_speakers=True)
+    return "\n\n".join(_plain_speech(content))
+
+
+def _plain_speech(content: str) -> list[str]:
+    """Blocks alternate between timing and speech. Only a block in timing
+    position is read as timing, so speech shaped like a timestamp is kept."""
+    blocks = [
+        lines
+        for block in _BLANK_LINE_RE.split(content.replace("\r\n", "\n"))
+        if (lines := [line.strip() for line in block.split("\n") if line.strip()])
+    ]
+    if blocks and blocks[0][0].startswith("WEBVTT"):
+        blocks = blocks[1:]
+    speech: list[str] = []
+    expect_timing = True
+    for lines in blocks:
+        if expect_timing and is_timing_line(lines[0]):
+            lines = lines[1:]
+        speech.extend(lines)
+        expect_timing = bool(lines)
+    return speech
+
+
+def transcript_access(
+    organizer: Organizer, meeting: MeetingRecord | None
+) -> ExternalAccess:
+    """The organizer and everyone the meeting record names as organizer or
+    attendee: the invited people, since Graph exposes neither who joined nor an
+    organizer's narrower viewing setting. Without the record, the organizer."""
+    emails = {organizer.email.lower()} if organizer.email else set()
+    if meeting is not None:
+        emails |= meeting.participant_emails
+    return ExternalAccess(
+        external_user_emails=emails, external_user_group_ids=set(), is_public=False
+    )
+
+
+# A listing costs about 2 seconds for every month it reaches back, and the
+# connector form waits 10. A refusal answers the same inside a window, so the
+# setup check asks for recent transcripts only.
+_TRANSCRIPT_PROBE_WINDOW_S = 30 * 24 * 60 * 60
+
+_TRANSCRIPTS_DISABLED = (
+    "Include Meeting Transcripts needs the tenant setting that allows Graph API "
+    "access to transcripts, which a Teams administrator has turned off."
+)
+
+
+def _transcript_refusal(error: requests.HTTPError) -> str:
+    """The admin-facing cause of a refused transcript call, from Graph's code,
+    message or status."""
+    if transcripts_disabled(error):
+        return _TRANSCRIPTS_DISABLED
+    if access_policy_missing(error):
+        return (
+            "An application access policy naming this app must be granted to the "
+            "organizer (or the whole tenant) for meeting transcripts."
+        )
+    if status(error) == 403:
+        return (
+            "Include Meeting Transcripts needs the OnlineMeetingTranscript.Read.All "
+            f"and OnlineMeetings.Read.All application permissions. {graph_said(error)}"
+        )
+    return f"Graph answered {status(error)}. {graph_said(error)}"
+
+
+class TranscriptSource(OrganizerSource):
+    option = "Include Meeting Transcripts"
+
+    def __init__(self, session: TeamsSession, covers_every_user: bool) -> None:
+        self._session = session
+        self._covers_every_user = covers_every_user
+
+    def validate(self, organizer: Organizer) -> None:
+        """Lists the organizer's last 30 days and, when there is a transcript,
+        reads its meeting and its content: the grants, the tenant setting and the
+        access policy all answer on these calls. An empty listing proves the
+        listing grant alone, which is enough to run."""
+        if self._covers_every_user:
+            # The check probes one user, so a policy that covers a group rather
+            # than the tenant passes here and refuses users at index time.
+            logger.warning(
+                "Include Meeting Transcripts has no organizers configured, so it "
+                "reads every enabled user with a Teams license. The application "
+                "access policy has to cover them all, or list the organizers it "
+                "covers."
+            )
+        graph_client = self._session.graph()
+        try:
+            probe_start = time.time() - _TRANSCRIPT_PROBE_WINDOW_S
+            transcript = next(
+                fetch_transcripts(graph_client, organizer.id, probe_start, None, 1),
+                None,
+            )
+            if transcript is None:
+                return
+            fetch_meeting(graph_client, organizer.id, transcript.meeting_id)
+            fetch_transcript_text(graph_client, transcript.content_url)
+        except requests.HTTPError as e:
+            if transcripts_disabled(e) or access_policy_missing(e):
+                raise ConnectorValidationError(_transcript_refusal(e))
+            if status(e) in (401, 403):
+                raise InsufficientPermissionsError(_transcript_refusal(e))
+            raise UnexpectedValidationError(f"Could not read a transcript: {e}")
+        except (GraphRetriesExhausted, requests.RequestException) as e:
+            raise UnexpectedValidationError(f"Could not read a transcript: {e}")
+
+    def index(
+        self,
+        organizer: Organizer,
+        start: SecondsSinceUnixEpoch,
+        end: SecondsSinceUnixEpoch,
+    ) -> Iterator[Document | ConnectorFailure]:
+        """The transcripts one organizer's meetings produced inside the window.
+        A refused listing is one recorded failure for the organizer, since the
+        access policy is granted per user, except the tenant-wide setting that
+        turns transcript export off, which no other organizer escapes."""
+        # Documents flow page by page. A document read records its own 403 or
+        # 404, so what reaches this handler is a refused listing or an error
+        # too general to blame on the organizer.
+        try:
+            for transcript in fetch_transcripts(
+                self._session.graph(), organizer.id, start, end
+            ):
+                yield self._document(organizer, transcript)
+        except requests.HTTPError as e:
+            if transcripts_disabled(e):
+                raise ConnectorValidationError(_TRANSCRIPTS_DISABLED) from e
+            if not is_permanent(e):
+                raise
+            yield ConnectorFailure(
+                failed_entity=EntityFailure(entity_id=organizer.id),
+                failure_message=(
+                    f"Could not list the meeting transcripts of {organizer.email}: "
+                    f"{_transcript_refusal(e)}"
+                ),
+                exception=e,
+            )
+
+    def _meeting_record(
+        self, organizer: Organizer, transcript: Transcript
+    ) -> MeetingRecord | None:
+        """The meeting's name and people. A refused record leaves the transcript
+        readable by its organizer alone, anything else fails the attempt."""
+        try:
+            return fetch_meeting(
+                self._session.graph(), organizer.id, transcript.meeting_id
+            )
+        except requests.HTTPError as e:
+            if not is_permanent(e):
+                raise
+            logger.warning(
+                "Meeting %s of %s is not readable, its transcript is shared with "
+                "the organizer alone: %s",
+                transcript.meeting_id,
+                organizer.email,
+                _transcript_refusal(e),
+            )
+            return None
+
+    def _document(
+        self, organizer: Organizer, transcript: Transcript
+    ) -> Document | ConnectorFailure:
+        meeting = self._meeting_record(organizer, transcript)
+        link = meeting.join_web_url if meeting else None
+        try:
+            content, attributed = fetch_transcript_text(
+                self._session.graph(), transcript.content_url
+            )
+        except requests.HTTPError as e:
+            if not is_permanent(e):
+                raise
+            return ConnectorFailure(
+                failed_document=DocumentFailure(
+                    document_id=transcript_document_id(organizer.id, transcript.id),
+                    document_link=link,
+                ),
+                failure_message=(
+                    f"Transcript of meeting {transcript.meeting_id} organized by "
+                    f"{organizer.email}: {_transcript_refusal(e)}"
+                ),
+                exception=e,
+            )
+        when = (
+            meeting.start if meeting and meeting.start else transcript.created
+        ).date()
+        title = (meeting.subject if meeting else None) or f"Teams meeting on {when}"
+        # The slim walk lists every transcript, so an empty one still needs a
+        # document or its old text would outlive it.
+        text = transcript_text(content, attributed) or title
+        return Document(
+            id=transcript_document_id(organizer.id, transcript.id),
+            sections=[TextSection(link=link, text=text)],
+            source=DocumentSource.TEAMS,
+            semantic_identifier=f"{title} ({when})",
+            title=title,
+            doc_created_at=transcript.created,
+            doc_updated_at=transcript.created,
+            primary_owners=organizer_expert(organizer),
+            metadata={
+                "organizer": organizer.email or "",
+                "meeting_start": (
+                    meeting.start.isoformat() if meeting and meeting.start else ""
+                ),
+                "speakers": "attributed" if attributed else "unattributed",
+            },
+            external_access=transcript_access(organizer, meeting),
+        )
+
+    def slim(self, organizer: Organizer, walk: SlimWalk) -> Iterator[SlimDocument]:
+        """One organizer's transcripts. A refused organizer lists nothing and
+        the walk goes on: the app lost those transcripts, so pruning removes
+        them, and one organizer never costs the rest of the connector its sync."""
+        listing = PagedListing(walk)
+        try:
+            for transcript in fetch_transcripts(
+                self._session.graph(),
+                organizer.id,
+                None,
+                None,
+                before_page=listing.before_page,
+            ):
+                walk.raise_if_stopped()
+                yield SlimDocument(
+                    id=transcript_document_id(organizer.id, transcript.id),
+                    external_access=(
+                        transcript_access(
+                            organizer, self._meeting_record(organizer, transcript)
+                        )
+                        if walk.with_readers
+                        else None
+                    ),
+                    doc_created_at=transcript.created,
+                )
+        except requests.HTTPError as e:
+            if transcripts_disabled(e):
+                raise ConnectorValidationError(_TRANSCRIPTS_DISABLED) from e
+            if not listing.lost_access(e):
+                raise
+            logger.warning(
+                "Could not list the transcripts of %s, so their indexed "
+                "transcripts are pruned: %s",
+                organizer.email,
+                _transcript_refusal(e),
+            )

backend/onyx/connectors/teams/utils.py

@@ -1,4 +1,5 @@
 import re
+import threading
 import time
 from collections.abc import Callable, Generator
 from datetime import datetime, timezone
@@ -7,6 +8,8 @@ from typing import Any
 
 import requests
 from office365.graph_client import GraphClient
+from office365.runtime.http.http_method import HttpMethod
+from office365.runtime.http.request_options import RequestOptions
 from office365.runtime.queries.client_query import ClientQuery
 
 from onyx.access.models import ExternalAccess
@@ -24,6 +27,12 @@ from onyx.utils.logger import setup_logger
 logger = setup_logger()
 
 
+def escape_odata_string(name: str) -> str:
+    """An OData string literal doubles its apostrophes. Other characters that
+    break Graph's OData parser are handled by filtering on the client instead."""
+    return name.replace("'", "''")
+
+
 def execute_query_with_retry(
     build_query: Callable[[], ClientQuery],
     method_name: str,
@@ -70,10 +79,45 @@ class GraphRetriesExhausted(RuntimeError):
     """Graph kept answering with a retryable status for every attempt."""
 
 
-def _retry(
+def get_json_with_retry(
     graph_client: GraphClient,
     request_url: str,
+    headers: dict[str, str] | None = None,
+    json_body: dict[str, Any] | None = None,
 ) -> dict:
+    json = request_with_retry(graph_client, request_url, headers, json_body).json()
+    if not isinstance(json, dict):
+        raise RuntimeError(f"Expected a JSON object, instead got {json=}")
+    return json
+
+
+def _execute(
+    graph_client: GraphClient,
+    request_url: str,
+    headers: dict[str, str] | None,
+    json_body: dict[str, Any] | None,
+) -> requests.Response:
+    """The SDK's direct request. Headers ride along when the route needs them,
+    a format to serve or an advanced query, and a body makes it a POST. The SDK
+    adds the bearer token to every shape."""
+    if not headers and json_body is None:
+        return graph_client.execute_request_direct(request_url)
+    request = RequestOptions(f"{graph_client.service_root_url()}/{request_url}")
+    request.headers.update(headers or {})
+    if json_body is not None:
+        request.method = HttpMethod.Post
+        request.data = json_body
+    return graph_client.pending_request().execute_request_direct(request)
+
+
+def request_with_retry(
+    graph_client: GraphClient,
+    request_url: str,
+    headers: dict[str, str] | None = None,
+    json_body: dict[str, Any] | None = None,
+) -> requests.Response:
+    """One Graph request under Teams' retry policy, as a raw response rather
+    than parsed JSON."""
     MAX_RETRIES = 10
     retry_number = 0
 
@@ -81,17 +125,13 @@ def _retry(
         # The SDK raises on every non-2xx status, so the response is taken from
         # the exception to apply one retry policy to raised and returned errors.
         try:
-            response = graph_client.execute_request_direct(request_url)
+            response = _execute(graph_client, request_url, headers, json_body)
         except requests.HTTPError as e:
             if e.response is None:
                 raise
             response = e.response
         if response.ok:
-            json = response.json()
-            if not isinstance(json, dict):
-                raise RuntimeError(f"Expected a JSON object, instead got {json=}")
-
-            return json
+            return response
 
         # Transient Graph errors (rate limits + 5xx gateway/server hiccups) are
         # retried with backoff; any other status is surfaced immediately.
@@ -125,10 +165,12 @@ def _retry(
     )
 
 
-def _get_next_url(
+def next_page_url(
     graph_client: GraphClient,
     json_response: dict,
+    page_url: str,
 ) -> str | None:
+    """The link to the page after ``page_url``, relative to the service root."""
     next_url = json_response.get("@odata.nextLink")
 
     if not next_url:
@@ -139,54 +181,87 @@ def _get_next_url(
             f"Expected a string for the `@odata.nextUrl`, instead got {next_url=}"
         )
 
-    return next_url.removeprefix(graph_client.service_root_url()).removeprefix("/")
+    next_url = next_url.removeprefix(graph_client.service_root_url()).removeprefix("/")
+    # A listing has no name to stop on, so a page that points back at itself
+    # would be read for ever, by a walk or by a checkpoint that saves the link.
+    if next_url == page_url:
+        raise RuntimeError(f"Graph repeated a page: {page_url}")
+    return next_url
 
 
-def _iter_values(
-    graph_client: GraphClient, request_url: str
+def iter_values(
+    graph_client: GraphClient,
+    request_url: str,
+    before_page: Callable[[], None] | None = None,
+    headers: dict[str, str] | None = None,
 ) -> Generator[dict[str, Any]]:
-    """Every row of a paged Graph collection."""
+    """Every row of a paged Graph collection. ``before_page`` runs ahead of
+    each page request, so a walk can honor a stop between pages. ``headers``
+    ride on every page."""
     url: str | None = request_url
     while url:
-        json_response = _retry(graph_client=graph_client, request_url=url)
+        if before_page is not None:
+            before_page()
+        json_response = get_json_with_retry(graph_client, url, headers)
         for value in json_response.get("value", []):
             if isinstance(value, dict):
                 yield value
-        url = _get_next_url(graph_client=graph_client, json_response=json_response)
+        url = next_page_url(graph_client, json_response, url)
 
 
-def _member_email(graph_client: GraphClient, member: ChannelMember) -> str | None:
-    """A member row carries its email for users of any tenant. A row without
-    one is looked up by user id, which only resolves users of this tenant."""
-    if member.email:
-        return member.email
+# The most ids Graph names in one getByIds request.
+USER_LOOKUP_BATCH_SIZE = 1000
+USER_LOOKUP_URL = "directoryObjects/getByIds?$select=id,userPrincipalName"
 
-    if not member.user_id:
-        logger.warning("Channel member %r has no user id; skipping", member)
-        return None
 
-    # Only a missing user is skipped: a user of another tenant is not in this
-    # directory. Any other refusal propagates, or a partial list would revoke access.
-    try:
-        json_data = _retry(
-            graph_client=graph_client, request_url=f"users/{member.user_id}"
+class UserDirectory:
+    """User id to principal name for the members a run meets, once per run. A
+    channel can hold thousands of members whose rows carry no email, and Graph
+    names 1000 ids per request, so the cost follows the members seen."""
+
+    def __init__(self, graph_client: GraphClient) -> None:
+        self._graph_client = graph_client
+        # None records an id Graph did not name, so it is not asked for again.
+        self._principal_names: dict[str, str | None] = {}
+        # Workers share one directory, and two asking for the same unknown ids
+        # would each pay for the lookup.
+        self._lock = threading.Lock()
+
+    def principal_names(self, user_ids: list[str]) -> dict[str, str]:
+        """The names Graph has for these ids. One it omits is not in this
+        directory, such as a user of another tenant. A refusal raises, or a
+        partial list would revoke access."""
+        with self._lock:
+            unknown = [
+                uid
+                for uid in dict.fromkeys(user_ids)
+                if uid not in self._principal_names
+            ]
+            for start in range(0, len(unknown), USER_LOOKUP_BATCH_SIZE):
+                batch = unknown[start : start + USER_LOOKUP_BATCH_SIZE]
+                self._principal_names.update(self._lookup(batch))
+            return {
+                uid: name
+                for uid in user_ids
+                if (name := self._principal_names.get(uid)) is not None
+            }
+
+    def _lookup(self, batch: list[str]) -> dict[str, str | None]:
+        rows = get_json_with_retry(
+            self._graph_client,
+            USER_LOOKUP_URL,
+            json_body={"ids": batch, "types": ["user"]},
         )
-    except requests.HTTPError as e:
-        if e.response is not None and e.response.status_code == 404:
-            logger.warning(
-                "Channel member %s is not in this directory; skipping",
-                member.display_name,
-            )
-            return None
-        raise
-
-    email = json_data.get("userPrincipalName")
-    if not isinstance(email, str) or not email:
-        logger.warning(
-            "Channel member %s has no principal name; skipping", member.user_id
-        )
-        return None
-    return email
+        # A batch answers in one page. A second page would hold names this
+        # reads as missing, and a missing name takes a reader's access away.
+        if rows.get("@odata.nextLink"):
+            raise RuntimeError("Graph paged a user lookup of one batch")
+        named = {
+            row["id"]: row["userPrincipalName"]
+            for row in rows.get("value", [])
+            if row.get("id") and row.get("userPrincipalName")
+        }
+        return {uid: named.get(uid) for uid in batch}
 
 
 def fetch_channel_members(
@@ -197,7 +272,7 @@ def fetch_channel_members(
     all-members call serves every channel type."""
     return [
         ChannelMember(**row)
-        for row in _iter_values(
+        for row in iter_values(
             graph_client, f"teams/{team_id}/channels/{channel_id}/allMembers"
         )
     ]
@@ -218,18 +293,35 @@ def channel_access(expert_infos: list[BasicExpertInfo]) -> ExternalAccess:
 
 
 def fetch_channel_readers(
-    graph_client: GraphClient, team_id: str, channel_id: str
+    graph_client: GraphClient,
+    team_id: str,
+    channel_id: str,
+    directory: UserDirectory,
 ) -> tuple[list[BasicExpertInfo], ExternalAccess]:
-    """The channel's members as document owners and as its access list."""
+    """The channel's members as document owners and as its access list. A row
+    carries its email for users of any tenant, and one without is named through
+    ``directory``, which only knows users of this tenant."""
+    members = fetch_channel_members(graph_client, team_id, channel_id)
+    names = directory.principal_names(
+        [m.user_id for m in members if not m.email and m.user_id]
+    )
     expert_infos: list[BasicExpertInfo] = []
-    for member in fetch_channel_members(graph_client, team_id, channel_id):
-        email = _member_email(graph_client, member)
+    unnamed = 0
+    for member in members:
+        email = member.email or names.get(member.user_id or "")
         if email is None:
+            unnamed += 1
             continue
         # The email is what grants access, so a member without a name still reads.
         expert_infos.append(
             BasicExpertInfo(display_name=member.display_name, email=email)
         )
+    if unnamed:
+        logger.warning(
+            "%s member(s) of channel %s are not in this directory; skipping",
+            unnamed,
+            channel_id,
+        )
     return expert_infos, channel_access(expert_infos)
 
 
@@ -254,15 +346,15 @@ def fetch_message_page(
 ) -> tuple[list[Message], str | None]:
     """One page of root messages and the link to the next, so a checkpoint can
     resume mid-channel."""
-    json_response = _retry(graph_client=graph_client, request_url=request_url)
+    json_response = get_json_with_retry(
+        graph_client=graph_client, request_url=request_url
+    )
     messages = [
         Message(**_sanitize_message_user_display_name(value))
         for value in json_response.get("value", [])
         if isinstance(value, dict)
     ]
-    return messages, _get_next_url(
-        graph_client=graph_client, json_response=json_response
-    )
+    return messages, next_page_url(graph_client, json_response, request_url)
 
 
 def fetch_messages(
@@ -271,7 +363,7 @@ def fetch_messages(
     channel_id: str,
     start: SecondsSinceUnixEpoch,
 ) -> Generator[Message]:
-    for value in _iter_values(
+    for value in iter_values(
         graph_client, message_delta_url(team_id, channel_id, start)
     ):
         yield Message(**_sanitize_message_user_display_name(value))
@@ -281,7 +373,7 @@ def fetch_channel_files_folder(
     graph_client: GraphClient, team_id: str, channel_id: str
 ) -> ChannelFilesFolder:
     """Needs Files.Read.All or Sites.Read.All on Graph."""
-    json_data = _retry(
+    json_data = get_json_with_retry(
         graph_client=graph_client,
         request_url=f"teams/{team_id}/channels/{channel_id}/filesFolder",
     )
@@ -297,7 +389,7 @@ def fetch_drive_library(graph_client: GraphClient, drive_id: str) -> tuple[str,
     """The document library's name, which SharePoint REST looks the list up by,
     and the url of its site. The site comes from the drive because the files
     folder can leave its own site id empty."""
-    drive = _retry(
+    drive = get_json_with_retry(
         graph_client=graph_client,
         request_url=f"drives/{drive_id}?$select=name,sharePointIds",
     )
@@ -355,5 +447,5 @@ def fetch_replies(
         f"teams/{team_id}/channels/{channel_id}/messages/{root_message_id}/replies"
     )
 
-    for value in _iter_values(graph_client, request_url):
+    for value in iter_values(graph_client, request_url):
         yield Message(**_sanitize_message_user_display_name(value))

backend/onyx/file_processing/webvtt.py

@@ -29,6 +29,11 @@ _VOICE_TAG_RE = re.compile(r"<v(?:\.[^ \t>]*)?[ \t]+([^>]+)>")
 _VOICE_BOUNDARY_RE = re.compile(rf"{_VOICE_TAG_RE.pattern}|</v>")
 
 
+def is_timing_line(line: str) -> bool:
+    """A cue's timing line, "00:00:01.000 --> 00:00:02.000" plus settings."""
+    return _TIMING_LINE_RE.match(line) is not None
+
+
 def _clean_cue_line(line: str) -> str:
     """WebVTT forbids a literal "&", so "R&D" arrives as "R&amp;D". Strip markup
     before decoding, or an escaped "&lt;v Jane&gt;" becomes a tag and gets

backend/onyx/utils/threadpool_concurrency.py

@@ -589,7 +589,11 @@ def parallel_yield(gens: list[Iterator[R]], max_workers: int = 10) -> Iterator[R
     with ThreadPoolExecutor(max_workers=max_workers) as executor:
         future_to_index: dict[Future[tuple[int, R | None]], int] = (  # ty: ignore[invalid-assignment]
             {
-                executor.submit(_next_or_none, ind, gen): ind
+                # The caller's context rides along, as in the rest of this
+                # module: the tenant id and the log prefix live in contextvars.
+                executor.submit(
+                    contextvars.copy_context().run, _next_or_none, ind, gen
+                ): ind
                 for ind, gen in enumerate(gens)
             }
         )
@@ -601,9 +605,14 @@ def parallel_yield(gens: list[Iterator[R]], max_workers: int = 10) -> Iterator[R
                 ind, result = future.result()
                 if result is not None:
                     yield result
-                    future_to_index[executor.submit(_next_or_none, ind, gens[ind])] = (
-                        next_ind  # ty: ignore[invalid-assignment]
-                    )
+                    future_to_index[
+                        executor.submit(
+                            contextvars.copy_context().run,
+                            _next_or_none,
+                            ind,
+                            gens[ind],
+                        )
+                    ] = next_ind  # ty: ignore[invalid-assignment]
                     next_ind += 1
                 del future_to_index[future]
 

backend/tests/unit/onyx/connectors/teams/helpers.py

@@ -3,10 +3,12 @@
 Lives beside the tests rather than in conftest.py because pytest imports
 conftest itself and test modules are not meant to import it back."""
 
+from collections.abc import Mapping
 from typing import Any
 from unittest.mock import MagicMock
 
 import requests
+from office365.runtime.http.http_method import HttpMethod
 
 from onyx.connectors.interfaces import SecondsSinceUnixEpoch
 from onyx.connectors.microsoft_utils.graph_auth import MicrosoftAuthMethod
@@ -40,26 +42,89 @@ def response(status: int, payload: dict[str, Any]) -> MagicMock:
     return resp
 
 
+Refusal = int | tuple[int, str, str]
+"""A status alone, or a status with Graph's inner error code and message."""
+
+
+def _refuse(refusal: Refusal) -> requests.HTTPError:
+    status, code, message = (
+        (refusal, str(refusal), str(refusal)) if isinstance(refusal, int) else refusal
+    )
+    # Graph puts the general code outside and the cause inside, so the outer
+    # one is never the code a caller branches on.
+    resp = response(
+        status,
+        {
+            "error": {
+                "code": "Forbidden",
+                "message": message,
+                "innerError": {"code": code},
+            }
+        },
+    )
+    return requests.HTTPError(str(status), response=resp)
+
+
 def graph_client(
-    routes: dict[str, dict[str, Any]], refused: dict[str, int] | None = None
+    routes: dict[str, dict[str, Any]],
+    refused: Mapping[str, Refusal] | None = None,
+    contents: Mapping[tuple[str, str], str | Refusal] | None = None,
 ) -> MagicMock:
     """A client whose direct requests answer from ``routes``, fail with the
-    status in ``refused``, and 404 elsewhere. The SDK raises on every non-2xx
-    status, so failures arrive as exceptions the way they do in production."""
+    refusal in ``refused``, and 404 elsewhere. ``contents`` answers requests
+    that name an Accept header, keyed by (url, accept), with text or a refusal.
+    The SDK raises on every non-2xx status, so failures arrive as exceptions
+    the way they do in production."""
     client = MagicMock()
     client.service_root_url.return_value = SERVICE_ROOT
 
     def execute(url: str) -> MagicMock:
         if url in routes:
             return response(200, routes[url])
-        status = (refused or {}).get(url, 404)
-        resp = response(status, {"error": {"code": str(status)}})
-        raise requests.HTTPError(str(status), response=resp)
+        raise _refuse((refused or {}).get(url, 404))
+
+    def execute_with_headers(request: Any) -> MagicMock:
+        url = request.url.removeprefix(SERVICE_ROOT + "/")
+        if request.method == HttpMethod.Post:
+            # Kept with its body, then answered by the users it names.
+            client.posted.append((url, request.data))
+            if url in (refused or {}):
+                raise _refuse((refused or {})[url])
+            return response(200, _named_users(routes.get(url, {}), request.data))
+        if "Accept" not in request.headers:
+            # A JSON route asked with headers, such as an advanced query: kept
+            # with its headers, then answered and recorded like a plain request.
+            client.headers_by_url[url] = dict(request.headers)
+            return client.execute_request_direct(url)
+        key = (url, request.headers["Accept"])
+        answer = (contents or {}).get(key, 404)
+        if not isinstance(answer, str):
+            raise _refuse(answer)
+        resp = response(200, {})
+        resp.text = answer
+        return resp
 
     client.execute_request_direct.side_effect = execute
+    client.headers_by_url = {}
+    client.posted = []
+    client.pending_request.return_value.execute_request_direct.side_effect = (
+        execute_with_headers
+    )
     return client
 
 
+def _named_users(known: dict[str, str], body: dict[str, Any]) -> dict[str, Any]:
+    """Graph's answer to a getByIds request over ``known`` (id to principal
+    name): a row for each id it has and nothing for the rest."""
+    return {
+        "value": [
+            {"id": uid, "userPrincipalName": known[uid]}
+            for uid in body["ids"]
+            if uid in known
+        ]
+    }
+
+
 def member(name: str | None, email: str | None, user_id: str) -> dict[str, Any]:
     return {"displayName": name, "email": email, "userId": user_id}
 
@@ -96,10 +161,14 @@ def connector(
     client: MagicMock,
     include_attachments: bool = False,
     include_inline_images: bool = False,
+    include_meeting_transcripts: bool = False,
+    transcript_organizers: list[str] | None = None,
 ) -> TeamsConnector:
     teams_connector = TeamsConnector(
         include_attachments=include_attachments,
         include_inline_images=include_inline_images,
+        include_meeting_transcripts=include_meeting_transcripts,
+        transcript_organizers=transcript_organizers,
     )
     teams_connector.graph_client = client
     # The factory grants this from the image analysis setting.

backend/tests/unit/onyx/connectors/teams/test_collect_teams.py

@@ -81,13 +81,13 @@ def test_helper_functions() -> None:
     """Test the helper functions for team name processing."""
     from onyx.connectors.teams.listing import (
         _can_use_odata_filter,
-        _escape_odata_string,
         has_odata_incompatible_chars,
     )
+    from onyx.connectors.teams.utils import escape_odata_string
 
     # Test OData string escaping
-    assert _escape_odata_string("Team's Group") == "Team''s Group"
-    assert _escape_odata_string("Normal Team") == "Normal Team"
+    assert escape_odata_string("Team's Group") == "Team''s Group"
+    assert escape_odata_string("Normal Team") == "Normal Team"
 
     # Test special character detection
     assert has_odata_incompatible_chars(["R&D Team"])

backend/tests/unit/onyx/connectors/teams/test_retry_request.py

@@ -1,7 +1,7 @@
-"""Tests for `_retry`, the raw Microsoft Graph request path in the Teams
+"""Tests for `get_json_with_retry`, the raw Microsoft Graph request path in the Teams
 connector (`execute_request_direct`), used by `fetch_messages`/`fetch_replies`.
 
-`_retry` retries transient Graph errors (rate limits + 5xx gateway/server
+`get_json_with_retry` retries transient Graph errors (rate limits + 5xx gateway/server
 hiccups) with backoff and surfaces everything else immediately. A live
 daily-connector test cannot reliably induce a 5xx, so the behaviour is verified
 here in isolation.
@@ -13,7 +13,10 @@ from unittest.mock import MagicMock, patch
 import pytest
 from requests.exceptions import HTTPError
 
-from onyx.connectors.teams.utils import GRAPH_API_RETRYABLE_STATUSES, _retry
+from onyx.connectors.teams.utils import (
+    GRAPH_API_RETRYABLE_STATUSES,
+    get_json_with_retry,
+)
 
 
 def _response(
@@ -45,7 +48,7 @@ def test_returns_json_without_retry_on_success() -> None:
     graph_client = _graph_client(_response(ok=True, json_value=payload))
 
     with patch("onyx.connectors.teams.utils.time.sleep") as mock_sleep:
-        result = _retry(graph_client=graph_client, request_url="teams")
+        result = get_json_with_retry(graph_client=graph_client, request_url="teams")
 
     assert result == payload
     assert graph_client.execute_request_direct.call_count == 1
@@ -61,7 +64,7 @@ def test_retries_transient_status_then_succeeds(status: int) -> None:
     )
 
     with patch("onyx.connectors.teams.utils.time.sleep") as mock_sleep:
-        result = _retry(graph_client=graph_client, request_url="teams")
+        result = get_json_with_retry(graph_client=graph_client, request_url="teams")
 
     assert result == payload
     assert graph_client.execute_request_direct.call_count == 2
@@ -74,7 +77,7 @@ def test_does_not_retry_non_retryable_status(status: int) -> None:
 
     with patch("onyx.connectors.teams.utils.time.sleep") as mock_sleep:
         with pytest.raises(HTTPError):
-            _retry(graph_client=graph_client, request_url="teams")
+            get_json_with_retry(graph_client=graph_client, request_url="teams")
 
     assert graph_client.execute_request_direct.call_count == 1
     mock_sleep.assert_not_called()
@@ -86,11 +89,11 @@ def test_raises_when_json_is_not_an_object() -> None:
 
     with patch("onyx.connectors.teams.utils.time.sleep"):
         with pytest.raises(RuntimeError):
-            _retry(graph_client=graph_client, request_url="teams")
+            get_json_with_retry(graph_client=graph_client, request_url="teams")
 
 
 def test_raises_runtime_error_after_exhausting_retries() -> None:
-    # Always returns a retryable status; _retry caps at MAX_RETRIES (10) attempts
+    # Always returns a retryable status; get_json_with_retry caps at MAX_RETRIES (10) attempts
     # and then raises rather than looping forever.
     graph_client = MagicMock()
     graph_client.execute_request_direct = MagicMock(
@@ -99,7 +102,7 @@ def test_raises_runtime_error_after_exhausting_retries() -> None:
 
     with patch("onyx.connectors.teams.utils.time.sleep") as mock_sleep:
         with pytest.raises(RuntimeError, match="Max number of retries"):
-            _retry(graph_client=graph_client, request_url="teams")
+            get_json_with_retry(graph_client=graph_client, request_url="teams")
 
     # 10 attempts are made, but the final (exhausted) attempt raises immediately
     # instead of sleeping, so there are only 9 backoff sleeps.

backend/tests/unit/onyx/connectors/teams/test_teams_channel_access.py

@@ -7,8 +7,11 @@ import pytest
 import requests
 
 from onyx.connectors.models import ConnectorFailure, Document
+from onyx.connectors.teams import utils as utils_module
 from onyx.connectors.teams.utils import (
+    USER_LOOKUP_URL,
     GraphRetriesExhausted,
+    UserDirectory,
     channel_access,
     fetch_channel_members,
     fetch_channel_readers,
@@ -52,7 +55,9 @@ def test_a_standard_channel_is_shared_with_its_members_not_everyone() -> None:
         {MEMBERS_URL: {"value": [member("Ada", "Ada@Example.com", "u1")]}}
     )
 
-    experts, access = fetch_channel_readers(client, TEAM_ID, CHANNEL_ID)
+    experts, access = fetch_channel_readers(
+        client, TEAM_ID, CHANNEL_ID, UserDirectory(client)
+    )
 
     assert [(e.display_name, e.email) for e in experts] == [("Ada", "Ada@Example.com")]
     assert access.is_public is False
@@ -64,11 +69,13 @@ def test_a_member_without_an_email_is_resolved_by_user_id() -> None:
     client = graph_client(
         {
             MEMBERS_URL: {"value": [member("Raunak", None, "u-raunak")]},
-            "users/u-raunak": {"userPrincipalName": "raunak@example.com"},
+            USER_LOOKUP_URL: {"u-raunak": "raunak@example.com"},
         }
     )
 
-    experts, _ = fetch_channel_readers(client, TEAM_ID, CHANNEL_ID)
+    experts, _ = fetch_channel_readers(
+        client, TEAM_ID, CHANNEL_ID, UserDirectory(client)
+    )
 
     assert [(e.display_name, e.email) for e in experts] == [
         ("Raunak", "raunak@example.com")
@@ -90,20 +97,23 @@ def test_a_member_from_another_tenant_keeps_the_email_on_the_row() -> None:
         }
     )
 
-    _, access = fetch_channel_readers(client, TEAM_ID, CHANNEL_ID)
+    _, access = fetch_channel_readers(
+        client, TEAM_ID, CHANNEL_ID, UserDirectory(client)
+    )
 
     assert access.external_user_emails == {"guest@other.example"}
-    assert client.execute_request_direct.call_args_list[-1].args == ("users/u-gone",)
+    # Only the row without an email is asked about, and Graph does not name it.
+    assert client.posted == [(USER_LOOKUP_URL, {"ids": ["u-gone"], "types": ["user"]})]
 
 
 def test_a_refused_user_lookup_is_not_a_missing_member() -> None:
     client = graph_client(
         {MEMBERS_URL: {"value": [member("Ada", None, "u1")]}},
-        refused={"users/u1": 403},
+        refused={USER_LOOKUP_URL: 403},
     )
 
     with pytest.raises(requests.HTTPError):
-        fetch_channel_readers(client, TEAM_ID, CHANNEL_ID)
+        fetch_channel_readers(client, TEAM_ID, CHANNEL_ID, UserDirectory(client))
 
 
 def test_a_member_without_a_display_name_still_reads() -> None:
@@ -111,7 +121,9 @@ def test_a_member_without_a_display_name_still_reads() -> None:
         {MEMBERS_URL: {"value": [member(None, "x@example.com", "u1")]}}
     )
 
-    experts, access = fetch_channel_readers(client, TEAM_ID, CHANNEL_ID)
+    experts, access = fetch_channel_readers(
+        client, TEAM_ID, CHANNEL_ID, UserDirectory(client)
+    )
 
     assert [(e.display_name, e.email) for e in experts] == [(None, "x@example.com")]
     assert access.external_user_emails == {"x@example.com"}
@@ -120,7 +132,9 @@ def test_a_member_without_a_display_name_still_reads() -> None:
 def test_a_channel_with_no_resolvable_members_is_shared_with_no_one() -> None:
     client = graph_client({MEMBERS_URL: {"value": [member(None, None, "")]}})
 
-    _, access = fetch_channel_readers(client, TEAM_ID, CHANNEL_ID)
+    _, access = fetch_channel_readers(
+        client, TEAM_ID, CHANNEL_ID, UserDirectory(client)
+    )
 
     assert access.is_public is False
     assert access.external_user_emails == set()
@@ -205,3 +219,76 @@ def test_a_channel_whose_members_stay_throttled_fails_the_attempt_not_the_channe
 
     with pytest.raises(GraphRetriesExhausted):
         walk_channel(connector(client))
+
+
+def _requested(client: MagicMock) -> list[str]:
+    return [call.args[0] for call in client.execute_request_direct.call_args_list]
+
+
+def test_members_without_an_email_are_named_a_batch_at_a_time_once_a_run(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # A channel can hold thousands of mail-less members, so names come in
+    # batches and once a run, not a request per member per channel.
+    monkeypatch.setattr(utils_module, "USER_LOOKUP_BATCH_SIZE", 2)
+    known = {"u1": "ada@example.com", "u2": "bob@example.com", "u3": "cy@example.com"}
+    client = graph_client(
+        {
+            MEMBERS_URL: {"value": [member(uid, None, uid) for uid in known]},
+            USER_LOOKUP_URL: known,
+        }
+    )
+    directory = UserDirectory(client)
+
+    for _ in range(2):
+        _, access = fetch_channel_readers(client, TEAM_ID, CHANNEL_ID, directory)
+        assert access.external_user_emails == set(known.values())
+
+    # Two channels of the same three members: two batches, asked for once.
+    assert [body["ids"] for _, body in client.posted] == [["u1", "u2"], ["u3"]]
+    assert not [url for url in _requested(client) if url.startswith("users/")]
+
+
+def test_an_id_graph_does_not_name_is_not_asked_for_again() -> None:
+    client = graph_client(
+        {
+            MEMBERS_URL: {"value": [member("Ghost", None, "u-gone")]},
+            USER_LOOKUP_URL: {},
+        }
+    )
+    directory = UserDirectory(client)
+
+    for _ in range(2):
+        _, access = fetch_channel_readers(client, TEAM_ID, CHANNEL_ID, directory)
+        assert access.external_user_emails == set()
+
+    assert len(client.posted) == 1
+
+
+def test_a_channel_whose_members_all_carry_an_email_asks_for_no_names() -> None:
+    client = graph_client(
+        {MEMBERS_URL: {"value": [member("Ada", "ada@example.com", "u1")]}}
+    )
+
+    fetch_channel_readers(client, TEAM_ID, CHANNEL_ID, UserDirectory(client))
+
+    assert client.posted == []
+
+
+def test_the_channel_walk_names_members_through_one_directory() -> None:
+    client = graph_client(
+        {
+            MEMBERS_URL: {"value": [member("Ada", None, "u1")]},
+            USER_LOOKUP_URL: {"u1": "ada@example.com"},
+            DELTA_URL: {"value": [message("m1", "Plan")]},
+            replies_url("m1"): {"value": []},
+        }
+    )
+    teams_connector = connector(client)
+
+    items = walk_channel(teams_connector)
+
+    documents = [item for item in items if isinstance(item, Document)]
+    assert documents[0].external_access is not None
+    assert documents[0].external_access.external_user_emails == {"ada@example.com"}
+    assert teams_connector.directory() is teams_connector.directory()

web/src/lib/connectors/connectors.tsx

@@ -1059,6 +1059,33 @@ export const connectorConfigs: Record<
           "extraction and analysis is off in the search settings.",
         default: false,
       },
+      {
+        type: "checkbox",
+        query: "Include meeting transcripts?",
+        label: "Include Meeting Transcripts",
+        name: "include_meeting_transcripts",
+        description:
+          "Index the transcripts of scheduled meetings as their own documents, " +
+          "readable by the organizer and the attendees. Needs the " +
+          "OnlineMeetingTranscript.Read.All, OnlineMeetings.Read.All and " +
+          "User.Read.All application permissions, the tenant setting that " +
+          "allows Graph API access to transcripts, and an application access " +
+          "policy granted to the organizers. Channel meetings are not covered. " +
+          "Transcripts from the last six months are indexed, and a transcript " +
+          "leaves the index once it is older than that.",
+        default: false,
+      },
+      {
+        type: "list",
+        query: "Enter meeting organizers to include:",
+        label: "Transcript Organizers",
+        name: "transcript_organizers",
+        optional: true,
+        description:
+          "User principal names of the organizers whose meeting transcripts " +
+          "to index. Leave empty to include every enabled user with a Teams " +
+          "license.",
+      },
     ],
     advanced_values: [
       {
@@ -2323,6 +2350,8 @@ export interface TeamsConfig {
   teams?: string[];
   include_attachments?: boolean;
   include_inline_images?: boolean;
+  include_meeting_transcripts?: boolean;
+  transcript_organizers?: string[];
   authority_host?: string;
   graph_api_host?: string;
 }