A kind-word decision records the inputs it considered, and a reply whose inputs moved is not accepted

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 = 79;
+const CURRENT_SCHEMA_VERSION: u32 = 80;
 
 fn main() -> anyhow::Result<()> {
     dotenvy::dotenv().ok();

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

@@ -1361,7 +1361,8 @@ pub struct DecideAlignmentKindWordReq {
 }
 
 /// 人定一个类别词绑到哪个类（#725 对齐队列）。绑上的类写到它名下的实体上；两端的类变了，
-/// 短语签名跟着变，所以再排一次短语对齐。
+/// 短语签名跟着变，所以再排一次短语对齐——与判定同一事务排下（`decide_and_apply_human`），
+/// 提交之后再排的话，进程在两步之间退出就只剩判定。
 pub async fn decide_alignment_kind_word(
     State(state): State<AppState>,
     AuthUser(user): AuthUser,
@@ -1399,12 +1400,6 @@ pub async fn decide_alignment_kind_word(
     if !written {
         return Err(utopia_core::AppError::NotFound.into());
     }
-    utopia_store::jobs::enqueue_unless_queued(
-        &state.pool,
-        "align_phrases",
-        json!({ "kb_id": kb_id }),
-    )
-    .await?;
     let _ = utopia_store::audit::record(
         &state.pool,
         Some(kb_id),

crates/utopia-server/src/type_alignment_tests.rs

@@ -33,6 +33,13 @@ async fn reply(State(m): State<Model>, Json(body): Json<Value>) -> impl IntoResp
         format!("data: {frame}\n\ndata: [DONE]\n\n"),
     )
 }
+/// 嵌入端点一直坏着：只有配了嵌入模型的测试会走到这里
+async fn embeddings_down() -> impl IntoResponse {
+    (
+        axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+        "embedding backend down",
+    )
+}
 struct Fx {
     pool: sqlx::PgPool,
     state: AppState,
@@ -87,6 +94,7 @@ impl Fx {
         let endpoint = format!("http://{}", listener.local_addr()?);
         let router = Router::new()
             .route("/chat/completions", post(reply))
+            .route("/embeddings", post(embeddings_down))
             .with_state(model.clone());
         let server = tokio::spawn(async move {
             axum::serve(listener, router).await.unwrap();
@@ -250,11 +258,16 @@ async fn human_decision_during_disagreement_survives() -> anyhow::Result<()> {
         .bind(f.entity)
         .fetch_one(&f.pool)
         .await?;
+    let requeued = align_types_jobs(&f).await?;
     let class = f.class;
     f.cleanup().await?;
     assert_eq!(binding.decided_by, "person");
     assert_eq!(binding.type_id, Some(class));
     assert_eq!(projected, Some(class));
+    assert!(
+        requeued.is_empty(),
+        "a person's decision carries no basis and is never asked about again"
+    );
     Ok(())
 }
 
@@ -732,3 +745,247 @@ async fn an_unreadable_reply_is_asked_again_a_bounded_number_of_times() -> anyho
     assert_eq!(phrases_queued_late, 1, "最后一轮排短语对齐");
     Ok(())
 }
+
+/// #795：模型还在答的时候类的定义改了。两票读的都是旧定义，它们的答案不能当成对新定义
+/// 的判定留下来；下一轮要拿新定义再问，问完之后输入没变就不再问
+#[tokio::test]
+async fn an_edit_during_the_model_request_is_asked_again() -> anyhow::Result<()> {
+    let Some(f) = Fx::new(vec![
+        vote(None),
+        vote(None),
+        vote(Some("organization")),
+        vote(Some("organization")),
+    ])
+    .await?
+    else {
+        return Ok(());
+    };
+    let run = async {
+        f.model
+            .hold
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+        let state = f.state.clone();
+        let kb = f.kb;
+        let worker = tokio::spawn(async move { align_types_reasking(&state, kb, 0).await });
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            f.model.entered.notified(),
+        )
+        .await?;
+        sqlx::query(
+            "UPDATE entity_types SET description='NEW definition', updated_at=clock_timestamp()
+              WHERE id=$1",
+        )
+        .bind(f.class)
+        .execute(&f.pool)
+        .await?;
+        f.model.release.notify_one();
+        worker.await??;
+        let first = f.requests();
+        anyhow::ensure!(first.len() == 2, "two votes, got {}", first.len());
+        anyhow::ensure!(
+            first
+                .iter()
+                .all(|r| r.to_string().contains("OLD definition")),
+            "both votes were built before the edit"
+        );
+        sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1")
+            .bind(f.kb.to_string())
+            .execute(&f.pool)
+            .await?;
+        f.run().await?;
+        let second = f.requests();
+        anyhow::ensure!(
+            second.len() == 4,
+            "the edit made during the request was taken as seen: the next run asked {} more",
+            second.len() - 2
+        );
+        anyhow::ensure!(
+            second[2..]
+                .iter()
+                .all(|r| r.to_string().contains("NEW definition")),
+            "the next run asks with the new definition"
+        );
+        let binding = type_bindings::bindings(&f.pool, f.kb).await?.remove(0);
+        anyhow::ensure!(binding.status == "bound" && binding.type_id == Some(f.class));
+        f.run().await?;
+        anyhow::ensure!(f.requests().len() == 4, "unchanged inputs ask nothing");
+        anyhow::Ok(())
+    }
+    .await;
+    f.cleanup().await?;
+    run
+}
+
+async fn align_types_jobs(f: &Fx) -> anyhow::Result<Vec<Option<i64>>> {
+    Ok(sqlx::query_scalar(
+        "SELECT (payload->>'reask')::bigint FROM jobs
+          WHERE kind='align_types' AND payload->>'kb_id'=$1 ORDER BY id",
+    )
+    .bind(f.kb.to_string())
+    .fetch_all(&f.pool)
+    .await?)
+}
+
+async fn clear_jobs(f: &Fx) -> anyhow::Result<()> {
+    sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1")
+        .bind(f.kb.to_string())
+        .execute(&f.pool)
+        .await?;
+    Ok(())
+}
+
+/// 父边的增删不碰 `updated_at`，时间戳看不见它；指纹里有祖先闭包，看得见
+#[tokio::test]
+async fn a_parent_edge_makes_an_agent_binding_stale() -> anyhow::Result<()> {
+    let Some(f) = Fx::new(vec![
+        vote(Some("organization")),
+        vote(Some("organization")),
+        vote(Some("organization")),
+        vote(Some("organization")),
+    ])
+    .await?
+    else {
+        return Ok(());
+    };
+    let run = async {
+        let legal = Uuid::now_v7();
+        sqlx::query("INSERT INTO entity_types(id,kb_id,key,label,description) VALUES($1,$2,'legal_entity','Legal entity','A body the law treats as a person')")
+            .bind(legal).bind(f.kb).execute(&f.pool).await?;
+        f.run().await?;
+        anyhow::ensure!(f.requests().len() == 2);
+        clear_jobs(&f).await?;
+        let versions = "SELECT id, updated_at FROM entity_types WHERE kb_id=$1 ORDER BY id";
+        let before: Vec<(Uuid, chrono::DateTime<chrono::Utc>)> = sqlx::query_as(versions)
+            .bind(f.kb)
+            .fetch_all(&f.pool)
+            .await?;
+        sqlx::query(
+            "INSERT INTO entity_type_parents(child_id,parent_id,is_primary) VALUES($1,$2,true)",
+        )
+        .bind(f.class)
+        .bind(legal)
+        .execute(&f.pool)
+        .await?;
+        let after: Vec<(Uuid, chrono::DateTime<chrono::Utc>)> = sqlx::query_as(versions)
+            .bind(f.kb)
+            .fetch_all(&f.pool)
+            .await?;
+        anyhow::ensure!(before == after, "a parent edge leaves updated_at alone");
+        f.run().await?;
+        anyhow::ensure!(
+            f.requests().len() == 4,
+            "the ancestor closure is part of the basis, so the word is asked again"
+        );
+        anyhow::Ok(())
+    }
+    .await;
+    f.cleanup().await?;
+    run
+}
+
+/// 这一列出现之前代理判的行没有指纹：各重判一次，之后不再问
+#[tokio::test]
+async fn rows_without_a_basis_are_decided_again_once() -> anyhow::Result<()> {
+    let Some(f) = Fx::new(vec![vote(Some("organization")), vote(Some("organization"))]).await?
+    else {
+        return Ok(());
+    };
+    let run = async {
+        f.seed_bound().await?;
+        f.run().await?;
+        anyhow::ensure!(f.requests().len() == 2);
+        let binding = type_bindings::bindings(&f.pool, f.kb).await?.remove(0);
+        anyhow::ensure!(binding.basis.is_some() && binding.decided_by == "agent");
+        f.run().await?;
+        anyhow::ensure!(
+            f.requests().len() == 2,
+            "a recorded basis that still matches asks nothing"
+        );
+        anyhow::Ok(())
+    }
+    .await;
+    f.cleanup().await?;
+    run
+}
+
+/// 过期的词这一轮问了却没问出结论（端点一直报错、回复读不出）：走有上限的延时再问，
+/// 不能因为「还过期」立刻再排——那样一个永久报错的端点会让任务一轮接一轮地跑
+#[tokio::test]
+async fn a_stale_word_whose_batch_fails_takes_the_bounded_reask() -> anyhow::Result<()> {
+    let Some(f) = Fx::new(vec![
+        json!({"answer": "organization"}),
+        json!({"answer": "organization"}),
+    ])
+    .await?
+    else {
+        return Ok(());
+    };
+    let run = async {
+        f.seed_bound().await?;
+        f.run().await?;
+        anyhow::ensure!(f.requests().len() == 2);
+        let jobs = align_types_jobs(&f).await?;
+        anyhow::ensure!(
+            jobs == vec![Some(1)],
+            "only the delayed re-ask is queued, got {jobs:?}"
+        );
+        let binding = type_bindings::bindings(&f.pool, f.kb).await?.remove(0);
+        anyhow::ensure!(binding.basis.is_none(), "nothing was decided this round");
+        anyhow::Ok(())
+    }
+    .await;
+    f.cleanup().await?;
+    run
+}
+
+/// 嵌入端点一直坏着：检索退回整表，小库照样绑得上（#894 修过的「整库没有类型」不能回来）；
+/// 一直坏下去候选不变、指纹不变，也不会每轮再问一遍
+#[tokio::test]
+async fn a_retrieval_error_falls_back_to_the_whole_class_list() -> anyhow::Result<()> {
+    let Some(f) = Fx::new(vec![vote(Some("organization")), vote(Some("organization"))]).await?
+    else {
+        return Ok(());
+    };
+    let run = async {
+        let ws: Uuid = sqlx::query_scalar("SELECT workspace_id FROM knowledge_bases WHERE id=$1")
+            .bind(f.kb)
+            .fetch_one(&f.pool)
+            .await?;
+        let chat = utopia_store::settings::get(&f.pool, ws)
+            .await?
+            .and_then(|s| s.chat_base_url)
+            .expect("fixture endpoint");
+        // upsert 整行覆盖（只有密钥传 None 才保留旧值），对话端点得原样再给一次
+        utopia_store::settings::upsert(
+            &f.pool,
+            ws,
+            Some(&chat),
+            None,
+            Some("scripted"),
+            Some(&chat),
+            None,
+            Some("scripted-embedding"),
+            Some(4),
+        )
+        .await?;
+        f.run().await?;
+        anyhow::ensure!(f.requests().len() == 2, "the fallback still asks");
+        let binding = type_bindings::bindings(&f.pool, f.kb).await?.remove(0);
+        anyhow::ensure!(binding.status == "bound" && binding.type_id == Some(f.class));
+        let jobs = align_types_jobs(&f).await?;
+        anyhow::ensure!(
+            jobs.is_empty(),
+            "a settled word queues nothing, got {jobs:?}"
+        );
+        f.run().await?;
+        anyhow::ensure!(
+            f.requests().len() == 2,
+            "while retrieval keeps failing the candidates, and so the basis, stay the same"
+        );
+        anyhow::Ok(())
+    }
+    .await;
+    f.cleanup().await?;
+    run
+}

crates/utopia-store/src/graph.rs

@@ -39,7 +39,11 @@ const ADOPT_MERGED: &str = "merged";
 // 剩下的那个函数于是只是在遍历一张空表。**本体从建库第一天起就只有
 // 用户自己导入的词表**——与 0009 删掉内置实体类是同一件事的下半段。
 
-pub async fn entity_types(pool: &PgPool, kb_id: Uuid) -> AppResult<Vec<EntityType>> {
+/// 库里全部类。执行器取泛型：类别词对齐要在调模型之前的同一个快照事务里读它（#795）。
+pub async fn entity_types<'e>(
+    pool: impl sqlx::Executor<'e, Database = sqlx::Postgres>,
+    kb_id: Uuid,
+) -> AppResult<Vec<EntityType>> {
     Ok(
         // 又一次 SELECT *：parents 在关联表里，`*` 取不到。
         // 这是同一个陷阱的第三次——SQL 在字符串里，cargo check 全绿，

crates/utopia-store/src/jobs.rs

@@ -69,6 +69,19 @@ pub async fn enqueue_unless_queued(
     payload: serde_json::Value,
 ) -> AppResult<Option<i64>> {
     let mut tx = pool.begin().await?;
+    let id = enqueue_unless_queued_tx(&mut tx, kind, payload).await?;
+    tx.commit().await?;
+    Ok(id)
+}
+
+/// 同 [`enqueue_unless_queued`]，但排在调用方的事务里：任务与事务里别的写一起提交、
+/// 一起回滚。人定一个类别词时，判定和它的短语对齐任务要么都在、要么都不在——先提交
+/// 判定再排，进程在两步之间退出就只剩判定，它改了的签名没人去重判
+pub async fn enqueue_unless_queued_tx(
+    tx: &mut Transaction<'_, Postgres>,
+    kind: &str,
+    payload: serde_json::Value,
+) -> AppResult<Option<i64>> {
     let row: Option<(i64,)> = sqlx::query_as(
         "INSERT INTO jobs (kind, payload)
          SELECT $1, $2
@@ -77,12 +90,11 @@ pub async fn enqueue_unless_queued(
     )
     .bind(kind)
     .bind(payload)
-    .fetch_optional(&mut *tx)
+    .fetch_optional(&mut **tx)
     .await?;
     if row.is_some() {
-        notify_worker_tx(&mut tx).await?;
+        notify_worker_tx(tx).await?;
     }
-    tx.commit().await?;
     Ok(row.map(|(id,)| id))
 }