A human phrase decision commits with its own recomputation job and answers 202

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

@@ -51,6 +51,21 @@ pub async fn failed_in_kb(
     Ok(Json(json!({ "failed": failed })))
 }
 
+/// 一个任务跑完了没（0051）。人定一条短语签名时拿到的是 job id 而不是结果，
+/// 结果要么从 `review` / `graph` 事件里等到，要么来这里问。Viewer 就能问：
+/// 任务属于这个库才答，否则 404，与库里看不见的东西一个口径
+pub async fn job_in_kb(
+    State(state): State<AppState>,
+    AuthUser(user): AuthUser,
+    Path((kb_id, job_id)): Path<(Uuid, i64)>,
+) -> ApiResult<Json<serde_json::Value>> {
+    require_kb(&state, &user, kb_id, Role::Viewer).await?;
+    let job = utopia_store::jobs::status_in_kb(&state.pool, kb_id, job_id)
+        .await?
+        .ok_or(utopia_core::AppError::NotFound)?;
+    Ok(Json(json!({ "job": job })))
+}
+
 pub async fn requeue_in_kb(
     State(state): State<AppState>,
     AuthUser(user): AuthUser,

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

@@ -174,6 +174,8 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router {
         .route("/kbs/{id}/jobs/failed", get(jobs_routes::failed_in_kb))
         .route("/kbs/{id}/jobs/requeue", post(jobs_routes::requeue_in_kb))
         .route("/jobs/requeue", post(jobs_routes::requeue_all))
+        // 一个任务的状态（0051）：人定完短语签名拿到 job id 后来这里问跑完没
+        .route("/kbs/{id}/jobs/{job_id}", get(jobs_routes::job_in_kb))
         .route(
             "/kbs/{id}/members/{user_id}",
             axum::routing::put(kbs::set_member).delete(kbs::remove_member),

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

@@ -12,6 +12,10 @@ use crate::auth::AuthUser;
 use crate::error::ApiResult;
 use crate::state::AppState;
 
+#[cfg(test)]
+#[path = "review_routes_phrase_tests.rs"]
+mod phrase_tests;
+
 /// 一页多少条。**服务端的默认，不是上限**——前端可以要更少，多则被 clamp 挡住
 const REVIEW_PAGE: i64 = 10;
 
@@ -1228,7 +1232,7 @@ pub async fn decide_alignment_phrase(
     AuthUser(user): AuthUser,
     Path((kb_id, binding_id)): Path<(Uuid, Uuid)>,
     Json(req): Json<DecideAlignmentPhraseReq>,
-) -> ApiResult<Json<serde_json::Value>> {
+) -> ApiResult<(axum::http::StatusCode, Json<serde_json::Value>)> {
     require_kb(&state, &user, kb_id, Role::Editor).await?;
     let sig = utopia_store::phrase_bindings::signature_of(&state.pool, kb_id, binding_id)
         .await?
@@ -1264,7 +1268,10 @@ pub async fn decide_alignment_phrase(
         }
     };
     let votes = json!({ "person": { "property": req.property, "direction": direction } });
-    utopia_store::phrase_bindings::decide(
+    // 判定和它的重算任务一次提交（0051）。这里**不再**同步重算：等物化锁占的是池里的
+    // 连接，而正在跑的那次对齐可能已经读完最后一遍，谁也不替这条判定投影。一个 job
+    // 只在判定提交后可见，worker 读的是当前绑定；屏幕上等的是 `review` / `graph` 事件
+    let job_id = utopia_store::phrase_bindings::decide_with_delivery(
         &state.pool,
         kb_id,
         &sig,
@@ -1276,8 +1283,8 @@ pub async fn decide_alignment_phrase(
             decided_by: "person",
         },
     )
-    .await?;
-    let typed = utopia_store::materialize::materialize_human(&state.pool, kb_id).await?;
+    .await?
+    .ok_or_else(|| utopia_core::AppError::Conflict("the decision was not written".into()))?;
     let _ = utopia_store::audit::record(
         &state.pool,
         Some(kb_id),
@@ -1286,13 +1293,14 @@ pub async fn decide_alignment_phrase(
         "phrase_binding",
         Some(binding_id),
         json!({ "phrase": sig.phrase, "property": req.property, "direction": direction,
-                "typed_added": typed.added, "typed_retired": typed.retired }),
+                "job_id": job_id }),
     )
     .await;
     state.emit_review(kb_id);
-    state.emit_graph(kb_id);
-    Ok(Json(
-        json!({ "ok": true, "typed": { "added": typed.added, "merged": typed.merged, "retired": typed.retired } }),
+    // 202：收下了，投影在路上。不编一个 typed: {added: 0} 出来——那不是这次请求知道的事
+    Ok((
+        axum::http::StatusCode::ACCEPTED,
+        Json(json!({ "ok": true, "job_id": job_id, "status": "accepted" })),
     ))
 }
 

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

@@ -0,0 +1,246 @@
+//! 人定一条短语签名：判定和它的重算任务一次提交，请求答 202 和 job id，不编数字（0051）。
+//!
+//! 三件事：路由答 202、job 排着、绑定写成人判的；`GET /kbs/{id}/jobs/{job_id}` 在本库能读、
+//! 换个库答 404；Viewer 能读状态但不能判。没有 `UTOPIA_DATABASE_URL` 时跳过。
+use axum::body::{to_bytes, Body};
+use axum::http::{Request, StatusCode};
+use serde_json::{json, Value};
+use std::sync::Arc;
+use tower::ServiceExt;
+use utopia_store::phrase_bindings;
+use uuid::Uuid;
+
+struct Fx {
+    pool: sqlx::PgPool,
+    app: axum::Router,
+    org: Uuid,
+    kb: Uuid,
+    other_kb: Uuid,
+    editor: String,
+    viewer: String,
+    binding: Uuid,
+    _dir: tempfile::TempDir,
+}
+
+impl Fx {
+    async fn new() -> anyhow::Result<Option<Self>> {
+        let Some(url) = utopia_store::test_db::url() else {
+            return Ok(None);
+        };
+        let pool = sqlx::PgPool::connect(&url).await?;
+        utopia_store::db::migrate(&pool).await?;
+        let (org, ws, kb, other_kb, editor, viewer, subject, object, property, statement) = (
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+        );
+        // Only locally generated UUIDs are interpolated into fixture SQL.
+        sqlx::raw_sql(&format!(
+            "INSERT INTO organizations(id,name) VALUES ('{org}','phrase-route-test');
+             INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','phrase-route-test');
+             INSERT INTO users(id,org_id,email,display_name,password_hash) VALUES
+                 ('{editor}','{org}','{editor}@phrase.test','editor','unused'),
+                 ('{viewer}','{org}','{viewer}@phrase.test','viewer','unused');
+             INSERT INTO knowledge_bases(id,workspace_id,name) VALUES
+                 ('{kb}','{ws}','phrases'), ('{other_kb}','{ws}','other');
+             INSERT INTO kb_members(kb_id,user_id,role) VALUES
+                 ('{kb}','{editor}','editor'), ('{kb}','{viewer}','viewer'),
+                 ('{other_kb}','{editor}','editor');
+             INSERT INTO entities(id,kb_id,canonical_name) VALUES
+                 ('{subject}','{kb}','Acme'), ('{object}','{kb}','London');
+             INSERT INTO relation_types(id,kb_id,key,label,temporal) VALUES
+                 ('{property}','{kb}','based_in','based in','state');
+             INSERT INTO facts(id,kb_id,subject_id,object_id,layer,phrase) VALUES
+                 ('{statement}','{kb}','{subject}','{object}','open','based in');"
+        ))
+        .execute(&pool)
+        .await?;
+        // 队列里的一条：代理判成 undecided，人来定
+        let sig = phrase_bindings::signatures(&pool, kb).await?.remove(0);
+        phrase_bindings::decide(
+            &pool,
+            kb,
+            &sig,
+            phrase_bindings::Decision {
+                relation_type_id: None,
+                direction: None,
+                status: "undecided",
+                votes: &json!({}),
+                decided_by: "agent",
+            },
+        )
+        .await?;
+        let binding: Uuid = sqlx::query_scalar("SELECT id FROM phrase_bindings WHERE kb_id=$1")
+            .bind(kb)
+            .fetch_one(&pool)
+            .await?;
+        let dir = tempfile::tempdir()?;
+        let cfg = utopia_core::config::AppConfig {
+            data_dir: dir.path().to_string_lossy().into_owned(),
+            ..Default::default()
+        };
+        let search = Arc::new(utopia_search::SearchIndex::open(
+            &dir.path().join("search"),
+        )?);
+        let state = crate::state::AppState::new(pool.clone(), &cfg, search, "test-only".into());
+        let editor_token = crate::auth::issue_token(&state, editor)?;
+        let viewer_token = crate::auth::issue_token(&state, viewer)?;
+        let app = super::super::router(state, &cfg);
+        Ok(Some(Self {
+            pool,
+            app,
+            org,
+            kb,
+            other_kb,
+            editor: editor_token,
+            viewer: viewer_token,
+            binding,
+            _dir: dir,
+        }))
+    }
+
+    async fn call(
+        &self,
+        token: &str,
+        method: &str,
+        path: &str,
+        body: Option<Value>,
+    ) -> anyhow::Result<(StatusCode, Value)> {
+        let request = Request::builder()
+            .method(method)
+            .uri(path)
+            .header("Authorization", format!("Bearer {token}"));
+        let request = match body {
+            Some(b) => request
+                .header("Content-Type", "application/json")
+                .body(Body::from(b.to_string()))?,
+            None => request.body(Body::empty())?,
+        };
+        let response = self.app.clone().oneshot(request).await?;
+        let status = response.status();
+        let bytes = to_bytes(response.into_body(), usize::MAX).await?;
+        let value = if bytes.is_empty() {
+            Value::Null
+        } else {
+            serde_json::from_slice(&bytes)?
+        };
+        Ok((status, value))
+    }
+
+    async fn cleanup(self) -> anyhow::Result<()> {
+        sqlx::query("DELETE FROM jobs WHERE payload->>'kb_id'=$1")
+            .bind(self.kb.to_string())
+            .execute(&self.pool)
+            .await?;
+        sqlx::query("DELETE FROM organizations WHERE id=$1")
+            .bind(self.org)
+            .execute(&self.pool)
+            .await?;
+        Ok(())
+    }
+}
+
+#[tokio::test]
+async fn a_phrase_decision_is_accepted_with_its_job() -> anyhow::Result<()> {
+    let Some(f) = Fx::new().await? else {
+        return Ok(());
+    };
+    let run = async {
+        let path = format!(
+            "/api/v1/kbs/{}/review/alignment/phrases/{}",
+            f.kb, f.binding
+        );
+        let (status, body) = f
+            .call(
+                &f.editor,
+                "POST",
+                &path,
+                Some(json!({ "property": "based_in", "direction": "forward" })),
+            )
+            .await?;
+        assert_eq!(status, StatusCode::ACCEPTED, "{body}");
+        assert_eq!(body["ok"], json!(true));
+        assert_eq!(body["status"], json!("accepted"));
+        assert!(body.get("typed").is_none(), "no invented counts: {body}");
+        let job_id = body["job_id"].as_i64().expect("job id");
+
+        // job 排着，载荷是这个库；绑定是人判的、bound
+        let (kind, job_status, payload): (String, String, Value) =
+            sqlx::query_as("SELECT kind, status, payload FROM jobs WHERE id=$1")
+                .bind(job_id)
+                .fetch_one(&f.pool)
+                .await?;
+        assert_eq!(kind, phrase_bindings::MATERIALIZE_KIND);
+        assert_eq!(job_status, "queued");
+        assert_eq!(payload["kb_id"], json!(f.kb));
+        let b = phrase_bindings::bindings(&f.pool, f.kb).await?.remove(0);
+        assert_eq!(
+            (b.status.as_str(), b.decided_by.as_str()),
+            ("bound", "person")
+        );
+        // 这次请求没有重算：投影要等 job
+        assert_eq!(utopia_store::materialize::count(&f.pool, f.kb).await?, 0);
+
+        // 状态读：本库 200，换库 404，Viewer 也能读
+        let (status, body) = f
+            .call(
+                &f.editor,
+                "GET",
+                &format!("/api/v1/kbs/{}/jobs/{job_id}", f.kb),
+                None,
+            )
+            .await?;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        assert_eq!(body["job"]["status"], json!("queued"));
+        assert_eq!(
+            body["job"]["kind"],
+            json!(phrase_bindings::MATERIALIZE_KIND)
+        );
+        let (status, _) = f
+            .call(
+                &f.editor,
+                "GET",
+                &format!("/api/v1/kbs/{}/jobs/{job_id}", f.other_kb),
+                None,
+            )
+            .await?;
+        assert_eq!(status, StatusCode::NOT_FOUND);
+        let (status, _) = f
+            .call(
+                &f.viewer,
+                "GET",
+                &format!("/api/v1/kbs/{}/jobs/{job_id}", f.kb),
+                None,
+            )
+            .await?;
+        assert_eq!(status, StatusCode::OK);
+
+        // Viewer 不能判
+        let (status, _) = f
+            .call(
+                &f.viewer,
+                "POST",
+                &path,
+                Some(json!({ "property": null, "direction": "forward" })),
+            )
+            .await?;
+        assert_eq!(status, StatusCode::FORBIDDEN);
+
+        // 让 job 该做的事在这里做一遍：读当前绑定，算出一条类型化行
+        let outcome = utopia_store::materialize::try_materialize(&f.pool, f.kb)
+            .await?
+            .expect("nobody holds the lock");
+        assert_eq!(outcome.added, 1);
+        anyhow::Ok(())
+    }
+    .await;
+    f.cleanup().await?;
+    run
+}

crates/utopia-server/src/main.rs

@@ -498,6 +498,45 @@ async fn dispatch(st: &state::AppState, job: &utopia_store::jobs::Job) -> anyhow
         }
         // 时间提及按文档解析（0045）：抽完一篇排一个，重排一次就是重新解析
         // 类别词绑到类（0044 对齐的第一片）：库级任务，抽完一篇排一个，本体改了再排
+        // 人定了一条短语签名，随判定同事务排下的重算（0051）。试锁不等：拿不到就
+        // 挂 `Deferred` 十秒后再来，占着连接排队的是别人的池子；拿到了就是一次完整
+        // 的按绑定重算，读的是当前绑定而不是判定时的载荷
+        utopia_store::phrase_bindings::MATERIALIZE_KIND => {
+            let kb_id: Uuid = job
+                .payload
+                .get("kb_id")
+                .and_then(|v| v.as_str())
+                .and_then(|s| s.parse().ok())
+                .ok_or_else(|| anyhow::anyhow!("payload 缺少 kb_id"))?;
+            match utopia_store::materialize::try_materialize(&st.pool, kb_id).await? {
+                Some(typed) => {
+                    if typed.added > 0 || typed.merged > 0 || typed.retired > 0 {
+                        let _ = utopia_store::audit::record(
+                            &st.pool,
+                            Some(kb_id),
+                            Uuid::nil(),
+                            "alignment.materialized",
+                            "knowledge_base",
+                            Some(kb_id),
+                            serde_json::json!({
+                                "job_id": job.id,
+                                "added": typed.added,
+                                "merged": typed.merged,
+                                "retired": typed.retired,
+                            }),
+                        )
+                        .await;
+                        st.emit_graph(kb_id);
+                    }
+                    // 队列卡片按绑定的状态显示，重算完了才算这条判定「落地」
+                    st.emit_review(kb_id);
+                    Ok(())
+                }
+                None => Err(anyhow::anyhow!("typed projection busy").context(
+                    utopia_core::Deferred::new(std::time::Duration::from_secs(10)),
+                )),
+            }
+        }
         "align_types" => {
             let kb_id: Uuid = job
                 .payload