Sync upstream dev into re-dev (2026-10-23): upstream's containment and identity names replace the fork's

.env.example

@@ -25,6 +25,10 @@ UTOPIA_OPEN_REGISTRATION=true
 # UTOPIA_OIDC_CLIENT_ID=
 # UTOPIA_OIDC_CLIENT_SECRET=
 # UTOPIA_OIDC_REDIRECT_URI=https://your-deployment/api/v1/auth/oidc/callback
+# 授权请求要的 scope，空格分隔，默认 openid profile email；openid 总会带上。身份提供方的管理员
+# 只批了 openid profile 时写这两个，免得每个人登录都卡在同意页。设了 UTOPIA_OIDC_ALLOWED_DOMAINS
+# 就要留着 email：按域名关联靠 ID token 里的 email。
+# UTOPIA_OIDC_SCOPES=openid profile email
 # 受信的另一个应用可以拿它用户在同一个身份提供方刚登录到的 ID token，到
 # POST /api/v1/auth/oidc/exchange 换一个 Utopia 会话（0066）。这里列它的 client id，逗号分隔；
 # 与 Utopia 共用一个 client 的就写 Utopia 自己的 client id。留空 = 不开放这个端点。

crates/utopia-cli/src/main.rs

@@ -82,7 +82,7 @@ struct ManifestDataDir {
 /// not a side effect of a code change.
 // 是迁移文件的**个数**，不是最大的编号（守卫 `schema_version_policy_compares_against_current`
 // 按个数比）：编号有空缺时两者不同——0071 由一个开放 PR 占着，0072 先落，个数是 71
-const CURRENT_SCHEMA_VERSION: u32 = 111;
+const CURRENT_SCHEMA_VERSION: u32 = 112;
 
 fn main() -> anyhow::Result<()> {
     dotenvy::dotenv().ok();

crates/utopia-core/src/models.rs

@@ -183,6 +183,13 @@ impl Source {
             .and_then(serde_json::Value::as_bool)
             .unwrap_or(true)
     }
+
+    /// statements 来源的 config 写了 `{"names": "identity"}`：它推来的每个名字各指一个
+    /// 确定的东西（0071），按字面归并、不算名字向量。其余来源照旧。
+    pub fn names_are_identities(&self) -> bool {
+        self.kind == "statements"
+            && self.config.get("names").and_then(serde_json::Value::as_str) == Some("identity")
+    }
 }
 
 #[cfg(test)]
@@ -220,6 +227,25 @@ mod source_extracts_tests {
         assert!(with(serde_json::json!({ "extract": 0 })).extracts());
         assert!(!with(serde_json::json!({ "extract": false })).extracts());
     }
+
+    fn of_kind(kind: &str, config: serde_json::Value) -> Source {
+        Source {
+            kind: kind.into(),
+            ..with(config)
+        }
+    }
+
+    #[test]
+    fn only_a_statements_source_with_the_string_identity_has_identity_names() {
+        let identity = serde_json::json!({ "names": "identity" });
+        assert!(of_kind("statements", identity.clone()).names_are_identities());
+        assert!(!of_kind("api", identity).names_are_identities());
+        assert!(!of_kind("statements", serde_json::json!({})).names_are_identities());
+        assert!(!of_kind("statements", serde_json::json!({ "names": true })).names_are_identities());
+        assert!(
+            !of_kind("statements", serde_json::json!({ "names": "prose" })).names_are_identities()
+        );
+    }
 }
 
 /// 来源配置里**用来鉴权**的那几个键。凭据只进不出：列表与创建 / 更新的响应都剔掉，
@@ -277,6 +303,8 @@ pub enum SourceKind {
     Rss,
     GithubIssues,
     GithubRepo,
+    /// gitlab.com 项目默认分支的文件（#1327）；自建 GitLab 没有验证过
+    GitlabRepo,
     JiraIssues,
     S3,
     AzureBlob,

crates/utopia-server/src/api/chat_empty_reply_tests.rs

@@ -301,8 +301,50 @@ impl Fx {
         let _ = std::fs::remove_dir_all(&self.dir);
         Ok(())
     }
+
+    /// From now on an assistant answer in this base fails to save. One trigger, created once
+    /// and never dropped, rejects marked bases: creating and dropping a trigger per test locks
+    /// `conversation_messages` under every other test, and a queued `DROP TRIGGER` deadlocks
+    /// with a test that holds a row lock while it reads messages.
+    pub(super) async fn reject_answer_saves(&self) -> anyhow::Result<()> {
+        static INSTALLED: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();
+        INSTALLED
+            .get_or_try_init(|| async {
+                sqlx::raw_sql(
+                    "SELECT pg_advisory_xact_lock(hashtext('test_reject_answer_save'));
+                     CREATE OR REPLACE FUNCTION test_reject_answer_save() RETURNS trigger
+                     LANGUAGE plpgsql AS $$ BEGIN
+                       IF NEW.role = 'assistant' AND EXISTS(
+                         SELECT 1 FROM conversations c JOIN knowledge_bases k ON k.id = c.kb_id
+                          WHERE c.id = NEW.conversation_id AND k.name = 'reject-answer-saves')
+                       THEN RAISE EXCEPTION 'private persistence diagnostic'; END IF;
+                       RETURN NEW;
+                     END $$;
+                     DO $$ BEGIN
+                       IF NOT EXISTS(SELECT 1 FROM pg_trigger WHERE tgname = 'test_reject_answer_save'
+                                        AND tgrelid = 'conversation_messages'::regclass) THEN
+                         CREATE TRIGGER test_reject_answer_save BEFORE INSERT ON conversation_messages
+                           FOR EACH ROW EXECUTE FUNCTION test_reject_answer_save();
+                       END IF;
+                     END $$;",
+                )
+                .execute(&self.pool)
+                .await
+                .map(drop)
+            })
+            .await?;
+        sqlx::query("UPDATE knowledge_bases SET name = 'reject-answer-saves' WHERE id = $1")
+            .bind(self.kb)
+            .execute(&self.pool)
+            .await?;
+        Ok(())
+    }
 }
 
+/// Held by tests that keep a database lock open while they wait on the chat under test, so
+/// two such tests never queue behind each other's locks.
+pub(super) static HELD_DB_LOCKS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
 /// 发给模型的请求里，最后一条消息的正文
 fn last_message(request: &serde_json::Value) -> String {
     request["messages"]
@@ -867,14 +909,8 @@ async fn save_failure_is_an_error_without_publishing_the_buffered_answer_or_retr
         let Some(f) = fixture(Scripted::new(replies)).await? else {
             return Ok(());
         };
-        let name = format!("reject_assistant_{}", f.kb.simple());
-        sqlx::raw_sql(&format!("CREATE FUNCTION {name}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF NEW.role='assistant' AND EXISTS(SELECT 1 FROM conversations WHERE id=NEW.conversation_id AND kb_id='{}') THEN RAISE EXCEPTION 'injected persistence failure'; END IF; RETURN NEW; END $$; CREATE TRIGGER {name} BEFORE INSERT ON conversation_messages FOR EACH ROW EXECUTE FUNCTION {name}();",f.kb)).execute(&f.pool).await?;
+        f.reject_answer_saves().await?;
         let sse = f.ask("What happened?").await?;
-        sqlx::raw_sql(&format!(
-            "DROP TRIGGER {name} ON conversation_messages; DROP FUNCTION {name}();"
-        ))
-        .execute(&f.pool)
-        .await?;
         assert!(sse.contains("event: error"), "{sse}");
         assert!(!sse.contains("event: done"), "{sse}");
         if budget {

crates/utopia-server/src/api/chat_error_code_tests.rs

@@ -73,24 +73,8 @@ async fn an_answer_that_cannot_be_saved_says_so_with_its_code() -> anyhow::Resul
     let Some(f) = fixture(Scripted::new(vec![TOOL, Reply::Text("Accepted answer.")])).await? else {
         return Ok(());
     };
-    let name = format!("reject_assistant_{}", f.kb.simple());
-    sqlx::raw_sql(&format!(
-        "CREATE FUNCTION {name}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN \
-         IF NEW.role='assistant' AND EXISTS(SELECT 1 FROM conversations \
-         WHERE id=NEW.conversation_id AND kb_id='{}') THEN \
-         RAISE EXCEPTION 'injected persistence failure'; END IF; RETURN NEW; END $$; \
-         CREATE TRIGGER {name} BEFORE INSERT ON conversation_messages \
-         FOR EACH ROW EXECUTE FUNCTION {name}();",
-        f.kb
-    ))
-    .execute(&f.pool)
-    .await?;
+    f.reject_answer_saves().await?;
     let sse = f.ask("What happened?").await?;
-    sqlx::raw_sql(&format!(
-        "DROP TRIGGER {name} ON conversation_messages; DROP FUNCTION {name}();"
-    ))
-    .execute(&f.pool)
-    .await?;
     assert_eq!(failure(&sse).0, "answer_not_saved");
     f.cleanup().await
 }

crates/utopia-server/src/api/chat_persistence_tests.rs

@@ -29,9 +29,8 @@ async fn exercise(deny: bool, disconnect: bool) -> anyhow::Result<()> {
         .respond_with(model.clone())
         .mount(&f._server)
         .await;
-    let trigger = format!("reject_chat_{}", f.kb.simple());
     if deny {
-        sqlx::raw_sql(&format!("CREATE FUNCTION {trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF NEW.role='assistant' AND EXISTS(SELECT 1 FROM conversations WHERE id=NEW.conversation_id AND kb_id='{}') THEN RAISE EXCEPTION 'private persistence diagnostic'; END IF; RETURN NEW; END $$; CREATE TRIGGER {trigger} BEFORE INSERT ON conversation_messages FOR EACH ROW EXECUTE FUNCTION {trigger}();",f.kb)).execute(&f.pool).await?;
+        f.reject_answer_saves().await?;
     }
     let result = tokio::time::timeout(std::time::Duration::from_secs(20), async {
         let sse = if disconnect {
@@ -89,13 +88,6 @@ async fn exercise(deny: bool, disconnect: bool) -> anyhow::Result<()> {
         Ok::<_, anyhow::Error>(())
     })
     .await;
-    if deny {
-        sqlx::raw_sql(&format!(
-            "DROP TRIGGER {trigger} ON conversation_messages; DROP FUNCTION {trigger}();"
-        ))
-        .execute(&f.pool)
-        .await?;
-    }
     f.cleanup().await?;
     result??;
     Ok(())

crates/utopia-server/src/api/chat_registry_tests.rs

@@ -477,6 +477,7 @@ async fn disconnect_keeps_running_busy_does_not_append_and_stale_stop_spares_fol
 #[tokio::test]
 async fn stop_after_save_starts_keeps_one_complete_answer_and_holds_busy_until_commit(
 ) -> anyhow::Result<()> {
+    let _held = HELD_DB_LOCKS.lock().await;
     let Some((f, gates, server)) = setup(Pause::AfterTool).await? else {
         return Ok(());
     };
@@ -552,8 +553,7 @@ async fn stopped_answer_save_failure_emits_error_and_releases_the_conversation(
     let Some((f, gates, server)) = setup(Pause::AfterTool).await? else {
         return Ok(());
     };
-    let trigger = format!("reject_stopped_{}", f.kb.simple());
-    sqlx::raw_sql(&format!("CREATE FUNCTION {trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF NEW.role='assistant' AND EXISTS(SELECT 1 FROM conversations WHERE id=NEW.conversation_id AND kb_id='{}') THEN RAISE EXCEPTION 'private stop persistence diagnostic'; END IF; RETURN NEW; END $$; CREATE TRIGGER {trigger} BEFORE INSERT ON conversation_messages FOR EACH ROW EXECUTE FUNCTION {trigger}();", f.kb)).execute(&f.pool).await?;
+    f.reject_answer_saves().await?;
     let result = tokio::time::timeout(std::time::Duration::from_secs(20), async {
         let id =
             utopia_store::conversations::create(&f.pool, f.kb, f.user.id, "save failure").await?;
@@ -574,9 +574,7 @@ async fn stopped_answer_save_failure_emits_error_and_releases_the_conversation(
         );
         let error: serde_json::Value = serde_json::from_str(&terminals[0].data)?;
         anyhow::ensure!(error["code"] == "answer_not_saved");
-        anyhow::ensure!(!terminals[0]
-            .data
-            .contains("private stop persistence diagnostic"));
+        anyhow::ensure!(!terminals[0].data.contains("private persistence diagnostic"));
         anyhow::ensure!(f.state.live.attach(id).await.is_none());
         anyhow::ensure!(
             utopia_store::conversations::messages(&f.pool, id)
@@ -591,11 +589,6 @@ async fn stopped_answer_save_failure_emits_error_and_releases_the_conversation(
     gates.release[0].notify_one();
     server.abort();
     let _ = server.await;
-    sqlx::raw_sql(&format!(
-        "DROP TRIGGER {trigger} ON conversation_messages; DROP FUNCTION {trigger}();"
-    ))
-    .execute(&f.pool)
-    .await?;
     f.cleanup().await?;
     result??;
     Ok(())

crates/utopia-server/src/api/chat_retry_tests.rs

@@ -287,6 +287,7 @@ async fn every_question_tells_the_interface_its_stored_id() -> anyhow::Result<()
 /// 重试资格尚未读完时，另一标签页不能订阅这次随后会被拒绝的占位。
 #[tokio::test]
 async fn a_rejected_retry_does_not_publish_a_stream_to_another_tab() -> anyhow::Result<()> {
+    let _held = HELD_DB_LOCKS.lock().await;
     let Some(f) = fixture(Scripted::new(vec![])).await? else {
         return Ok(());
     };

crates/utopia-server/src/api/documents_routes.rs

@@ -85,12 +85,13 @@ pub(crate) async fn require_reader_kb(
     state: &AppState,
     reader: &SessionOrToken,
     kb_id: Uuid,
-) -> ApiResult<()> {
-    utopia_store::access::require_kb(&state.pool, &reader.user, kb_id, Role::Viewer).await?;
+) -> ApiResult<utopia_core::models::KnowledgeBase> {
+    let kb =
+        utopia_store::access::require_kb(&state.pool, &reader.user, kb_id, Role::Viewer).await?;
     if !reader.reaches(kb_id) {
         return Err(AppError::NotFound.into());
     }
-    Ok(())
+    Ok(kb)
 }
 
 /// Serve the retained original named by the document ledger.

crates/utopia-server/src/api/export_routes.rs

@@ -41,8 +41,7 @@ pub async fn export(
     Path(kb_id): Path<Uuid>,
     Query(q): Query<ExportQuery>,
 ) -> ApiResult<Response> {
-    require_reader_kb(&state, &reader, kb_id).await?;
-    let kb = utopia_store::kbs::get(&state.pool, kb_id).await?;
+    let kb = require_reader_kb(&state, &reader, kb_id).await?;
     let format = Format::parse(q.format.as_deref()).ok_or_else(|| {
         AppError::Validation("Unknown `format` (expected turtle or jsonld)".into())
     })?;

crates/utopia-server/src/api/mcp_tests.rs

@@ -193,7 +193,17 @@ impl Fixture {
             .await?;
         let dir = self.dir.clone();
         drop(self);
-        std::fs::remove_dir_all(dir)?;
+        // Tantivy can finish a background file operation after the state is dropped.
+        // Retry only that teardown race; persistent or unrelated IO errors still fail.
+        for attempt in 0..10 {
+            match std::fs::remove_dir_all(&dir) {
+                Ok(()) => break,
+                Err(e) if e.kind() == std::io::ErrorKind::DirectoryNotEmpty && attempt < 9 => {
+                    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+                }
+                Err(e) => return Err(e.into()),
+            }
+        }
         Ok(())
     }
 }

crates/utopia-server/src/api/oidc_routes.rs

@@ -69,6 +69,27 @@ struct Config {
     /// 本地开发：允许 issuer 与各端点走回环地址上的明文 HTTP。只认 localhost /
     /// 127.0.0.1 / ::1，公网地址照旧只能 HTTPS
     loopback_http: bool,
+    /// 授权请求的 scope（`UTOPIA_OIDC_SCOPES`，默认 `openid profile email`）。有的身份提供方
+    /// 的管理员只批了 `openid profile`，多要一个 `email` 就让每个人登录时都卡在同意页（#1355）
+    scopes: String,
+}
+
+/// `openid` 必须在：少了它就不是 OIDC，拿不到 ID token。没写就补在最前面，而不是报错
+fn scopes(raw: Option<&str>) -> String {
+    let mut out: Vec<&str> = Vec::new();
+    for s in raw
+        .filter(|r| !r.trim().is_empty())
+        .unwrap_or("openid profile email")
+        .split_whitespace()
+    {
+        if !out.contains(&s) {
+            out.push(s);
+        }
+    }
+    if !out.contains(&"openid") {
+        out.insert(0, "openid");
+    }
+    out.join(" ")
 }
 
 fn config() -> Result<Config, AppError> {
@@ -100,6 +121,7 @@ fn config() -> Result<Config, AppError> {
             .collect(),
         loopback_http: std::env::var("UTOPIA_OIDC_ALLOW_LOOPBACK_HTTP")
             .is_ok_and(|v| matches!(v.trim(), "1" | "true")),
+        scopes: scopes(std::env::var("UTOPIA_OIDC_SCOPES").ok().as_deref()),
     };
     secure_url(&c.issuer, c.loopback_http)?;
     secure_url(&c.redirect, true)?;
@@ -292,6 +314,16 @@ pub fn password_login(s: &AppState) -> bool {
     s.password_login || config().is_err()
 }
 
+/// 启动时说出来：允许名单按 ID token 里核实过的 email 关联账号，scope 里没有 `email`
+/// 时多数身份提供方根本不发这个声明，名单就形同虚设
+pub fn warn_about_scopes() {
+    if let Ok(c) = config() {
+        if !c.allowed_domains.is_empty() && !c.scopes.split_whitespace().any(|s| s == "email") {
+            tracing::warn!("UTOPIA_OIDC_SCOPES 里没有 email，而 UTOPIA_OIDC_ALLOWED_DOMAINS 已设：身份提供方多半不发 email，按域名关联和开户不会发生")
+        }
+    }
+}
+
 /// 启动时把关掉密码登录的两种后果说出来：没配 SSO 时这个开关不生效；
 /// 没有允许名单时只有绑定过身份的人进得来，新人既注册不了也开不了户
 pub fn warn_about_password_login(s: &AppState) {
@@ -387,7 +419,7 @@ async fn begin(
     .await?;
     url.query_pairs_mut().extend_pairs([
         ("response_type", "code"),
-        ("scope", "openid profile email"),
+        ("scope", c.scopes.as_str()),
         ("client_id", &c.client_id),
         ("redirect_uri", &c.redirect),
         ("state", &state),
@@ -1100,9 +1132,25 @@ mod tests {
             exchange_audiences: vec!["sibling-app".into(), "utopia-test".into()],
             allowed_domains: vec![],
             loopback_http: false,
+            scopes: super::scopes(None),
         }
     }
 
+    #[test]
+    fn scopes_default_to_openid_profile_email_and_always_keep_openid() {
+        assert_eq!(super::scopes(None), "openid profile email");
+        assert_eq!(super::scopes(Some("  ")), "openid profile email");
+        assert_eq!(super::scopes(Some("openid profile")), "openid profile");
+        assert_eq!(
+            super::scopes(Some("profile  email")),
+            "openid profile email"
+        );
+        assert_eq!(
+            super::scopes(Some("profile openid profile")),
+            "profile openid"
+        );
+    }
+
     // This key is generated solely for these tests and is never used by the app.
     #[test]
     fn verifies_signature_issuer_audience_nonce_and_expiry() {