A definition can be written by hand

crates/utopia-core/src/models.rs

@@ -1030,6 +1030,9 @@ pub struct ConceptMapping {
     /// **状态而不是置信度。** 从前借事实的 confidence 表达「提议 0.6 / 确认 1.0」，
     /// 那是把二值状态编码成浮点数，还顺带让它落进「低置信事实」那一档
     pub status: String,
+    /// 人从零写的口径记写的人；探索提的为空（#562）。`decided_by` 分不出这件事——
+    /// 探索提的经人确认之后同样有 decided_by
+    pub written_by: Option<Uuid>,
 }
 
 /// 一处公理违规，配好展示所需的三元组文本（见 `axiom_violations`）。

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

@@ -146,3 +146,249 @@ pub async fn revisions(
     let rows = utopia_store::mappings::revisions(&state.pool, kb_id, mapping_id).await?;
     Ok(Json(json!({ "revisions": rows })))
 }
+
+/// 空白等于没填：前端清空一个输入框传来的是 ""，落库该是 NULL 而不是空串。
+fn clean(s: &Option<String>) -> Option<String> {
+    s.as_deref()
+        .map(str::trim)
+        .filter(|x| !x.is_empty())
+        .map(str::to_string)
+}
+
+/// 一条口径至少得有表、表达式、SQL 之一，否则它不是一个可执行的定义。
+/// `create`、`preview` 与 `revise` 判的是同一条。
+fn definition_shape(
+    table_name: &Option<String>,
+    expr: &Option<String>,
+    sql: &Option<String>,
+) -> Result<(), AppError> {
+    if table_name.is_none() && expr.is_none() && sql.is_none() {
+        return Err(AppError::invalid(
+            "empty_mapping",
+            "A mapping needs at least one of table, expression or SQL",
+        ));
+    }
+    Ok(())
+}
+
+/// 一条口径跑起来是哪句 SQL：给了 `sql` 用 `sql`，否则 `expr` + `table` 拼一句。
+/// 测量台（`scripts/bench/mappings.mjs` 的 `proposalSql`）判分用的是同一个拼法，
+/// 所以页面上预览到的数就是测量台会打分的数。
+fn render_sql(
+    table_name: &Option<String>,
+    expr: &Option<String>,
+    sql: &Option<String>,
+) -> Option<String> {
+    if let Some(s) = sql {
+        return Some(s.trim_end_matches(';').trim().to_string());
+    }
+    match (expr, table_name) {
+        (Some(e), Some(t)) => Some(format!("SELECT {e} FROM {t}")),
+        _ => None,
+    }
+}
+
+#[derive(Deserialize)]
+pub struct CreateReq {
+    concept: String,
+    /// metric | dimension；缺省 metric
+    #[serde(default = "default_kind")]
+    kind: String,
+    source: String,
+    table_name: Option<String>,
+    expr: Option<String>,
+    sql: Option<String>,
+    unit: Option<String>,
+    summary: Option<String>,
+    #[serde(default)]
+    derived: bool,
+}
+fn default_kind() -> String {
+    "metric".into()
+}
+
+/// 人从零写一条口径（#562）。
+///
+/// 从前这张表只有探索一条来路。一个数据团队手上有自己的指标口径文档，却没有地方
+/// 把它填进去——而口径进了问数的提示词，宽表语料从 1/18 到 17/18（#520）。缺的
+/// 不是结构，是这扇门。
+///
+/// 概念仍落成 Metric / Dimension 类的实体——那是这张表的现状（`concept_id` 非空
+/// 指向实体）。0035 要退役这个形状（#556），但保留这张表并往里渲染，人写的 SQL
+/// 搬得过去。
+pub async fn create(
+    State(state): State<AppState>,
+    AuthUser(user): AuthUser,
+    Path(kb_id): Path<Uuid>,
+    Json(req): Json<CreateReq>,
+) -> ApiResult<Json<serde_json::Value>> {
+    require_kb(&state, &user, kb_id, Role::Editor).await?;
+    let concept = req.concept.trim();
+    if concept.is_empty() {
+        return Err(AppError::invalid("empty_concept", "A definition needs a concept name").into());
+    }
+    if !matches!(req.kind.as_str(), "metric" | "dimension") {
+        return Err(AppError::invalid("bad_kind", "kind must be metric or dimension").into());
+    }
+    let (table_name, expr, sql, unit, summary) = (
+        clean(&req.table_name),
+        clean(&req.expr),
+        clean(&req.sql),
+        clean(&req.unit),
+        clean(&req.summary),
+    );
+    definition_shape(&table_name, &expr, &sql)?;
+    // 源按名字点单，且必须是本库挂载的——与 `query_data` 同一条安全边界
+    let mounted = utopia_store::datasources::mounted(&state.pool, kb_id).await?;
+    let Some(ds) = mounted
+        .iter()
+        .find(|d| d.name.eq_ignore_ascii_case(req.source.trim()))
+    else {
+        return Err(AppError::invalid(
+            "source_not_mounted",
+            "That data source is not mounted on this knowledge base",
+        )
+        .into());
+    };
+
+    crate::mappings::ensure_concept_types(&state.pool, kb_id)
+        .await
+        .map_err(AppError::Other)?;
+    let (type_id,): (Uuid,) =
+        sqlx::query_as("SELECT id FROM entity_types WHERE kb_id = $1 AND key = $2")
+            .bind(kb_id)
+            .bind(&req.kind)
+            .fetch_one(&state.pool)
+            .await?;
+    // 概念实体走消解：同名归并，跟探索提的那些落在一处
+    let resolved = utopia_store::resolution::resolve_mention(
+        &state.pool,
+        kb_id,
+        Some(type_id),
+        concept,
+        None,
+        None,
+        &[],
+    )
+    .await?;
+    let id = utopia_store::mappings::create(
+        &state.pool,
+        kb_id,
+        resolved.entity_id,
+        &ds.name,
+        table_name.as_deref(),
+        expr.as_deref(),
+        sql.as_deref(),
+        unit.as_deref(),
+        summary.as_deref(),
+        req.derived,
+        user.id,
+    )
+    .await?;
+    let _ = utopia_store::audit::record(
+        &state.pool,
+        Some(kb_id),
+        user.id,
+        "mapping.written",
+        "concept_mapping",
+        Some(id),
+        json!({ "concept": concept, "source": ds.name }),
+    )
+    .await;
+    Ok(Json(json!({ "id": id, "concept_id": resolved.entity_id })))
+}
+
+#[derive(Deserialize)]
+pub struct PreviewReq {
+    source: String,
+    table_name: Option<String>,
+    expr: Option<String>,
+    sql: Option<String>,
+}
+
+/// 把一条口径过只读闸跑一遍，回第一行——写口径的人当场看到它算出来的数。
+///
+/// 这正是测量台判分的那个动作（#501 / #520：把定义跑一遍，跟 gold 比数），
+/// 搬到页面上来。跑不通的定义无害，它失败得很响；跑得通而算错的才是全部风险，
+/// 而看一眼数是人能做的唯一一道检查。
+pub async fn preview(
+    State(state): State<AppState>,
+    AuthUser(user): AuthUser,
+    Path(kb_id): Path<Uuid>,
+    Json(req): Json<PreviewReq>,
+) -> ApiResult<Json<serde_json::Value>> {
+    require_kb(&state, &user, kb_id, Role::Editor).await?;
+    let (table_name, expr, sql) = (clean(&req.table_name), clean(&req.expr), clean(&req.sql));
+    definition_shape(&table_name, &expr, &sql)?;
+    let Some(rendered) = render_sql(&table_name, &expr, &sql) else {
+        return Err(AppError::invalid(
+            "empty_mapping",
+            "A preview needs SQL, or an expression together with a table",
+        )
+        .into());
+    };
+    let mounted = utopia_store::datasources::mounted(&state.pool, kb_id).await?;
+    let Some(ds) = mounted
+        .iter()
+        .find(|d| d.name.eq_ignore_ascii_case(req.source.trim()))
+    else {
+        return Err(AppError::invalid(
+            "source_not_mounted",
+            "That data source is not mounted on this knowledge base",
+        )
+        .into());
+    };
+    // 引擎的报错原样带回：列名拼错、语法错、超时是三件不同的事
+    let out = super::tools::run_query(&state, ds.id, &rendered)
+        .await
+        .map_err(|e| AppError::invalid("preview_failed", e.to_string()))?;
+    // 结果是 JSON Lines；第一行就是这条口径算出来的那个数（或那一组数）
+    let rows: Vec<serde_json::Value> = out
+        .lines()
+        .filter_map(|l| serde_json::from_str::<serde_json::Value>(l.trim()).ok())
+        .take(5)
+        .collect();
+    Ok(Json(
+        json!({ "sql": rendered, "row": rows.first(), "rows": rows }),
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{definition_shape, render_sql};
+
+    #[test]
+    fn a_definition_needs_something_to_run() {
+        let none: Option<String> = None;
+        assert!(definition_shape(&none, &none, &none).is_err());
+        assert!(definition_shape(&Some("t".into()), &none, &none).is_ok());
+        assert!(definition_shape(&none, &Some("sum(x)".into()), &none).is_ok());
+        assert!(definition_shape(&none, &none, &Some("SELECT 1".into())).is_ok());
+    }
+
+    #[test]
+    fn preview_runs_what_the_bench_scores() {
+        let none: Option<String> = None;
+        // sql 优先，且去掉结尾的分号——闸门只放行单条语句
+        assert_eq!(
+            render_sql(
+                &Some("t".into()),
+                &Some("sum(x)".into()),
+                &Some("SELECT 2;".into())
+            )
+            .as_deref(),
+            Some("SELECT 2")
+        );
+        assert_eq!(
+            render_sql(
+                &Some("dw.orders".into()),
+                &Some("sum(amt) / 100.0".into()),
+                &none
+            )
+            .as_deref(),
+            Some("SELECT sum(amt) / 100.0 FROM dw.orders")
+        );
+        // 只有表名不是一个可执行的口径
+        assert_eq!(render_sql(&Some("t".into()), &none, &none), None);
+    }
+}

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

@@ -148,7 +148,12 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router {
             "/me/tokens/{token_id}",
             axum::routing::delete(token_routes::revoke),
         )
-        .route("/kbs/{id}/mappings", get(mapping_routes::list))
+        .route(
+            "/kbs/{id}/mappings",
+            get(mapping_routes::list).post(mapping_routes::create),
+        )
+        // 静态段排在 `{mapping_id}` 前面：先跑一遍看数，不落库
+        .route("/kbs/{id}/mappings/preview", post(mapping_routes::preview))
         .route(
             "/kbs/{id}/mappings/{mapping_id}",
             axum::routing::patch(mapping_routes::revise),

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

@@ -708,7 +708,7 @@ pub(super) fn charter_source_json(n: usize, h: &utopia_search::DocsSection) -> s
 }
 
 /// 问数执行：安全闸（解析白名单）→ 引擎执行（只读会话 + 强制 LIMIT + 超时）→ JSON 行。
-async fn run_query(state: &AppState, ds_id: Uuid, sql: &str) -> anyhow::Result<String> {
+pub(crate) async fn run_query(state: &AppState, ds_id: Uuid, sql: &str) -> anyhow::Result<String> {
     let (engine, conn) = utopia_store::datasources::engine_and_conn(&state.pool, ds_id).await?;
     // 闸门按引擎选方言：Databricks 的反引号、Snowflake 的 :: 转型都得先过得了解析
     let guarded = crate::query_engine::guard_sql_for(&engine, sql)?;

crates/utopia-server/src/mappings.rs

@@ -19,7 +19,7 @@ const MAX_SCHEMA_CHARS: usize = 12_000;
 /// 内置本体包里——0009 之后建库不再自带类。没有它们，下面的 `type_id` 查不到，
 /// 每条提议都被 `continue` 吞掉，页面只说"已排队"就再无下文（#223）。
 /// 所以探索前把两个类补上：builtin，描述给抽取提示词，本体页可以改
-async fn ensure_concept_types(pool: &sqlx::PgPool, kb_id: Uuid) -> anyhow::Result<()> {
+pub(crate) async fn ensure_concept_types(pool: &sqlx::PgPool, kb_id: Uuid) -> anyhow::Result<()> {
     for (key, label, description) in [
         (
             "metric",

crates/utopia-store/src/mappings.rs

@@ -76,6 +76,58 @@ pub async fn propose(
     Ok(id)
 }
 
+/// 人从零写一条口径（#562）。**落下来就是确认的**：写的人就是表态的人，
+/// 不需要再过一遍审。
+///
+/// 从前这张表只有探索一条来路。一个数据团队手上有自己的指标口径文档，却没有
+/// 地方把它填进去——而口径进了问数的提示词，宽表语料从 1/18 到 17/18（#520）。
+///
+/// 同一个 (概念, 源) 已经有一条时报冲突，而不是悄悄覆盖：那一条可能是探索提的、
+/// 人已经确认过的，覆盖等于抹掉一次表态。要改用 `revise`。
+/// 探索反过来也盖不掉这条：`propose` 只刷新 `proposed` 的行。
+#[allow(clippy::too_many_arguments)]
+pub async fn create(
+    pool: &PgPool,
+    kb_id: Uuid,
+    concept_id: Uuid,
+    source: &str,
+    table_name: Option<&str>,
+    expr: Option<&str>,
+    sql: Option<&str>,
+    unit: Option<&str>,
+    summary: Option<&str>,
+    derived: bool,
+    actor: Uuid,
+) -> AppResult<Uuid> {
+    let id = Uuid::now_v7();
+    sqlx::query(
+        "INSERT INTO concept_mappings
+             (id, kb_id, concept_id, source, table_name, expr, sql, unit, summary, derived,
+              status, decided_by, decided_at, written_by)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, 'confirmed', $11, now(), $11)",
+    )
+    .bind(id)
+    .bind(kb_id)
+    .bind(concept_id)
+    .bind(source)
+    .bind(table_name)
+    .bind(expr)
+    .bind(sql)
+    .bind(unit)
+    .bind(summary)
+    .bind(derived)
+    .bind(actor)
+    .execute(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db) if db.is_unique_violation() => AppError::Conflict(
+            "A definition for this concept on this source already exists; revise it instead".into(),
+        ),
+        _ => AppError::Db(e),
+    })?;
+    Ok(id)
+}
+
 /// 还等着人表态的。Review 页读它。
 pub async fn proposed(
     pool: &PgPool,
@@ -85,7 +137,7 @@ pub async fn proposed(
 ) -> AppResult<Vec<ConceptMapping>> {
     Ok(sqlx::query_as(
         "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source,
-                m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status
+                m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by
          FROM concept_mappings m
          JOIN entities e ON e.id = m.concept_id
          WHERE m.kb_id = $1 AND m.status = 'proposed'
@@ -104,7 +156,7 @@ pub async fn proposed(
 pub async fn confirmed(pool: &PgPool, kb_id: Uuid, limit: i64) -> AppResult<Vec<ConceptMapping>> {
     Ok(sqlx::query_as(
         "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source,
-                m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status
+                m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by
          FROM concept_mappings m
          JOIN entities e ON e.id = m.concept_id
          WHERE m.kb_id = $1 AND m.status = 'confirmed'
@@ -229,7 +281,7 @@ pub async fn page(
                 OR m.table_name ILIKE '%' || $3 || '%')";
     let rows: Vec<ConceptMapping> = sqlx::query_as(&format!(
         "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source,
-                m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status
+                m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by
          FROM concept_mappings m
          JOIN entities e ON e.id = m.concept_id
          {WHERE}