Merge main (roles and access revamp) into connectors

Main's access model (team editors and viewers, edit_credentials, owner-only
OAuth servers, per-user tool preferences, resource sponsors) now applies to
connection-backed tools and sources. Our migrations are renumbered to
0040_connections and 0041_connection_account_name, after main's
0038_resource_access_settings and 0039_resource_sponsors.

Where the two sides met: MCP tools save and load their connections as the
tool owner, editors may add a key (a new connection on the owner's account)
but never rewrite an existing connection's secret, fixed values stay
owner-only, and a member's own connection is used in member mode.

docsgpt/api/connector/routes.py

@@ -16,6 +16,8 @@ from flask_restx import fields, Namespace, Resource
 
 
 from docsgpt.api import api
+from docsgpt.api.user.resource_access import AccessDenied
+from docsgpt.api.user.sources.access import load_source
 from docsgpt.api.user.tasks import (
     ingest_connector_task,
 )
@@ -24,8 +26,8 @@ from docsgpt.core.settings import settings
 from docsgpt.parser.connectors.connector_creator import ConnectorCreator
 from docsgpt.storage.db.repositories.connector_sessions import (
     ConnectorSessionsRepository,
+    owns_connector_session,
 )
-from docsgpt.storage.db.repositories.sources import SourcesRepository
 from docsgpt.storage.db.session import db_readonly, db_session
 
 
@@ -528,6 +530,53 @@ class ConnectorDisconnect(Resource):
             return make_response(jsonify({"success": False, "error": "Failed to disconnect session"}), 500)
 
 
+def _owner_connector_session(
+    conn, owner_id: str, provider: str, connection_id: Optional[str] = None,
+) -> Optional[dict]:
+    """The owner's usable connection for ``provider``, or None.
+
+    Used when a team editor syncs a shared connector source: the sync runs
+    with the owner's account, the source's own connection when it has one,
+    else the owner's first connection for the provider. A connection that is
+    not connected (signed out, flagged for reconnect, holding no
+    credentials), cannot be decrypted, or has an expired access token that
+    can't be refreshed counts as missing.
+
+    Args:
+        conn: Open database connection.
+        owner_id: The source owner's ``sub``.
+        provider: The source's connector provider.
+        connection_id: The source's own connection, if it names one.
+
+    Returns:
+        Optional[dict]: The connection row, or None when the owner must reconnect.
+    """
+    from docsgpt.security.encryption import CredentialDecryptionError
+
+    repo = ConnectorSessionsRepository(conn)
+    if connection_id:
+        row = repo.get_for_user(str(connection_id), owner_id)
+        candidates = [row] if owns_connector_session(row, owner_id, provider) else []
+    else:
+        candidates = [s for s in repo.list_for_user(owner_id) if owns_connector_session(s, owner_id, provider)]
+    for session in candidates:
+        if service.normalize_status(session) != service.STATUS_CONNECTED:
+            continue
+        try:
+            token_info = service.read_secrets(session).get("token_info") or {}
+        except CredentialDecryptionError:
+            continue
+        if isinstance(token_info, dict) and token_info and not token_info.get("refresh_token"):
+            try:
+                if ConnectorCreator.create_auth(provider).is_token_expired(token_info):
+                    continue
+            except Exception:
+                # Providers without an expiry check leave the verdict to the sync.
+                pass
+        return session
+    return None
+
+
 @connectors_ns.route("/api/connectors/sync")
 class ConnectorSync(Resource):
     @api.expect(
@@ -536,7 +585,9 @@ class ConnectorSync(Resource):
             {
                 "source_id": fields.String(required=True, description="Source ID to sync"),
                 "connection_id": fields.String(
-                    required=False, description="Connection to sync with; defaults to the source's own",
+                    required=False,
+                    description="Connection to sync with; defaults to the source's own (ignored for "
+                    "team editors, whose sync uses the owner's connection)",
                 ),
                 "session_token": fields.String(required=False, description="Legacy; use connection_id")
             },
@@ -561,19 +612,19 @@ class ConnectorSync(Resource):
                     400
                 )
             user_id = decoded_token.get('sub')
-            with db_readonly() as conn:
-                source = SourcesRepository(conn).get_any(source_id, user_id)
-            if not source:
+            # Owner or team editor. The sync always runs AS the owner, with the
+            # owner's connector account: a grantee can't point the source at
+            # their own account (that is ``reconnect``, owner-only).
+            try:
+                with db_readonly() as conn:
+                    source, ra = load_source(conn, source_id, user_id, "edit")
+            except AccessDenied as err:
                 return make_response(
-                    jsonify({
-                        "success": False,
-                        "error": "Source not found"
-                    }),
-                    404
+                    jsonify({"success": False, "error": err.message, "message": err.message}),
+                    err.status,
                 )
-
-            # ``get_any`` already scopes by ``user_id``; an extra guard
-            # here would be dead code.
+            owner_id = ra.owner_id
+            is_owner = ra.access == "owner"
 
             remote_data = source.get('remote_data') or {}
             if isinstance(remote_data, str):
@@ -589,19 +640,40 @@ class ConnectorSync(Resource):
                     jsonify({
                         "success": False,
                         "error": "Source provider not found in remote_data"
-                    }), 
+                    }),
                     400
                 )
 
-            lookup = dict(data)
-            if not (lookup.get('connection_id') or lookup.get('session_token')) and source.get('connection_id'):
-                lookup['connection_id'] = str(source['connection_id'])
-            session = service.resolve_request_connection(user_id, source_type, lookup)
-            if session is None:
-                return make_response(
-                    jsonify({"success": False, "error": "Invalid or unauthorized session"}),
-                    401,
-                )
+            source_connection = str(source['connection_id']) if source.get('connection_id') else None
+            if is_owner:
+                lookup = dict(data)
+                if not (lookup.get('connection_id') or lookup.get('session_token')):
+                    if not source_connection:
+                        return make_response(
+                            jsonify({"success": False, "error": "connection_id is required"}),
+                            400,
+                        )
+                    lookup['connection_id'] = source_connection
+                session = service.resolve_request_connection(user_id, source_type, lookup)
+                if session is None:
+                    return make_response(
+                        jsonify({"success": False, "error": "Invalid or unauthorized session"}),
+                        401,
+                    )
+            else:
+                # A grantee can't name a connection: theirs would read the
+                # source's files with another account.
+                with db_readonly() as conn:
+                    session = _owner_connector_session(conn, owner_id, source_type, source_connection)
+                if session is None:
+                    message = (
+                        "The owner needs to reconnect this source's account "
+                        "before it can be synced."
+                    )
+                    return make_response(
+                        jsonify({"success": False, "error": message, "message": message}),
+                        409,
+                    )
 
             # Extract configuration from remote_data
             file_ids = remote_data.get('file_ids', [])
@@ -611,7 +683,7 @@ class ConnectorSync(Resource):
             # Start the sync task
             task = ingest_connector_task.delay(
                 job_name=source.get('name'),
-                user=decoded_token.get('sub'),
+                user=owner_id,
                 source_type=source_type,
                 connection_id=str(session["id"]),
                 file_ids=file_ids,