diff --git a/_primitives/_rust/kei-entity-store/src/ddl.rs b/_primitives/_rust/kei-entity-store/src/ddl.rs
new file mode 100644
index 0000000..e457232
--- /dev/null
+++ b/_primitives/_rust/kei-entity-store/src/ddl.rs
@@ -0,0 +1,157 @@
+//! DDL-string generators split out of `engine.rs` to keep that file
+//! under the Constructor-Pattern 200-LOC cap. One function per emitted
+//! `CREATE` statement; the engine's `run_migrations` orchestrates the
+//! calls and stamps `user_version`.
+
+use crate::schema::{EdgeKeyKind, EntitySchema, FieldDef, FieldKind};
+
+pub fn primary_table(schema: &EntitySchema) -> String {
+    let cols: Vec<String> = schema.fields.iter().map(column).collect();
+    format!(
+        "CREATE TABLE IF NOT EXISTS {} (\n  {}\n);",
+        schema.table,
+        cols.join(",\n  ")
+    )
+}
+
+fn column(f: &FieldDef) -> String {
+    match f.kind {
+        FieldKind::IntegerPk => format!("{} INTEGER PRIMARY KEY", f.name),
+        FieldKind::TextPk => format!("{} TEXT PRIMARY KEY", f.name),
+        FieldKind::IntegerNotNull => format!("{} INTEGER NOT NULL", f.name),
+        FieldKind::Integer => format!("{} INTEGER DEFAULT 0", f.name),
+        FieldKind::TextNotNull => format!("{} TEXT NOT NULL", f.name),
+        FieldKind::Text => format!("{} TEXT DEFAULT ''", f.name),
+        FieldKind::TextDefault => text_default_column(f),
+        FieldKind::TextArchiveEnum => archive_enum_column(f),
+        FieldKind::Real => format!("{} REAL NOT NULL DEFAULT 0.0", f.name),
+        FieldKind::RealDefault => real_default_column(f),
+        FieldKind::TimestampCreated => format!("{} INTEGER NOT NULL", f.name),
+        FieldKind::TimestampUpdated => format!("{} INTEGER NOT NULL", f.name),
+    }
+}
+
+fn text_default_column(f: &FieldDef) -> String {
+    let d = f.default.unwrap_or("");
+    // SQL-escape embedded single quotes (per SQL standard: `'` → `''`)
+    // so `text_default("status", "don't know")` does not inject.
+    let escaped = d.replace('\'', "''");
+    format!("{} TEXT NOT NULL DEFAULT '{}'", f.name, escaped)
+}
+
+fn archive_enum_column(f: &FieldDef) -> String {
+    let (active, _archived) = f.archive_enum.unwrap_or(("active", "archived"));
+    let escaped = active.replace('\'', "''");
+    format!("{} TEXT NOT NULL DEFAULT '{}'", f.name, escaped)
+}
+
+fn real_default_column(f: &FieldDef) -> String {
+    let d = f.real_default.unwrap_or(0.0);
+    format!("{} REAL NOT NULL DEFAULT {}", f.name, format_real(d))
+}
+
+/// Deterministic SQL literal for an f64 — always has a decimal point,
+/// no exponent for finite values. Non-finite values fall back to 0.0.
+fn format_real(v: f64) -> String {
+    if !v.is_finite() {
+        return "0.0".to_string();
+    }
+    if v.fract() == 0.0 {
+        format!("{:.1}", v)
+    } else {
+        format!("{}", v)
+    }
+}
+
+pub fn indexes(schema: &EntitySchema) -> String {
+    let mut out = String::new();
+    for f in schema.fields.iter().filter(|f| f.indexed) {
+        out.push_str(&format!(
+            "CREATE INDEX IF NOT EXISTS idx_{t}_{c} ON {t}({c});\n",
+            t = schema.table,
+            c = f.name
+        ));
+    }
+    out
+}
+
+pub fn fts_table(table: &str, cols: &[&str]) -> String {
+    let col_list = cols.join(", ");
+    format!(
+        "CREATE VIRTUAL TABLE IF NOT EXISTS fts_{table} \
+         USING fts5({table}_id UNINDEXED, {col_list}, tokenize='porter unicode61');"
+    )
+}
+
+/// Dispatcher — picks edge-table DDL for a given `EdgeKeyKind`. Added
+/// for kei-sage migration; `IntegerPair` branch preserves legacy body.
+pub fn edge_table_for(edge: &str, kind: EdgeKeyKind) -> String {
+    match kind {
+        EdgeKeyKind::IntegerPair => edge_integer(edge),
+        EdgeKeyKind::TextPair => edge_text(edge),
+        EdgeKeyKind::TextPairWithMetadata {
+            has_id,
+            has_weight,
+            has_created_at,
+        } => edge_text_meta(edge, has_id, has_weight, has_created_at),
+    }
+}
+
+fn edge_integer(edge: &str) -> String {
+    format!(
+        "CREATE TABLE IF NOT EXISTS {edge} (\n  \
+            from_id INTEGER NOT NULL,\n  \
+            to_id INTEGER NOT NULL,\n  \
+            edge_type TEXT NOT NULL DEFAULT 'links',\n  \
+            PRIMARY KEY(from_id, to_id, edge_type)\n\
+         );\n\
+         CREATE INDEX IF NOT EXISTS idx_{edge}_to ON {edge}(to_id);"
+    )
+}
+
+/// Text-keyed edge DDL: `(src_path TEXT, dst_path TEXT, edge_type TEXT)`.
+fn edge_text(edge: &str) -> String {
+    format!(
+        "CREATE TABLE IF NOT EXISTS {edge} (\n  \
+            src_path TEXT NOT NULL,\n  \
+            dst_path TEXT NOT NULL,\n  \
+            edge_type TEXT NOT NULL DEFAULT 'links',\n  \
+            PRIMARY KEY(src_path, dst_path, edge_type)\n\
+         );\n\
+         CREATE INDEX IF NOT EXISTS idx_{edge}_dst ON {edge}(dst_path);"
+    )
+}
+
+/// Text-keyed edge DDL with optional metadata columns.
+fn edge_text_meta(
+    edge: &str,
+    has_id: bool,
+    has_weight: bool,
+    has_created_at: bool,
+) -> String {
+    let mut cols: Vec<String> = Vec::new();
+    if has_id {
+        cols.push("edge_id INTEGER PRIMARY KEY AUTOINCREMENT".to_string());
+    }
+    cols.push("src_path TEXT NOT NULL".to_string());
+    cols.push("dst_path TEXT NOT NULL".to_string());
+    cols.push("edge_type TEXT NOT NULL DEFAULT 'links'".to_string());
+    if has_weight {
+        cols.push("weight REAL NOT NULL DEFAULT 1.0".to_string());
+    }
+    if has_created_at {
+        cols.push("created_at INTEGER NOT NULL".to_string());
+    }
+    // Without an autoincrement PK we still want `INSERT OR IGNORE`
+    // idempotent over the triple; with one we emit a UNIQUE instead.
+    if has_id {
+        cols.push("UNIQUE(src_path, dst_path, edge_type)".to_string());
+    } else {
+        cols.push("PRIMARY KEY(src_path, dst_path, edge_type)".to_string());
+    }
+    let body = cols.join(",\n  ");
+    format!(
+        "CREATE TABLE IF NOT EXISTS {edge} (\n  {body}\n);\n\
+         CREATE INDEX IF NOT EXISTS idx_{edge}_dst ON {edge}(dst_path);"
+    )
+}
diff --git a/_primitives/_rust/kei-entity-store/src/engine.rs b/_primitives/_rust/kei-entity-store/src/engine.rs
index 63c4445..cd155df 100644
--- a/_primitives/_rust/kei-entity-store/src/engine.rs
+++ b/_primitives/_rust/kei-entity-store/src/engine.rs
@@ -6,8 +6,9 @@
 //! &SCHEMA, input)`). This keeps the engine a passive provider of
 //! connection + schema-aware DDL.
 
+use crate::ddl;
 use crate::error::VerbError;
-use crate::schema::{EdgeKeyKind, EntitySchema, FieldDef, FieldKind};
+use crate::schema::EntitySchema;
 use anyhow::{Context, Result};
 use rusqlite::Connection;
 use std::path::Path;
@@ -16,10 +17,6 @@ use std::path::Path;
 /// first open. Future migrations bump this constant and gate their DDL
 /// on the pragma's current value — idempotent `CREATE TABLE IF NOT
 /// EXISTS` is not enough once column shapes diverge.
-///
-/// TODO B5: expose a `version: u32` field on `EntitySchema` and add a
-/// `custom_migrations: &'static [&'static str]` entry indexed by
-/// target version so sibling crates can publish their own bump paths.
 pub const CURRENT_USER_VERSION: u32 = 1;
 
 pub struct Store {
@@ -28,8 +25,7 @@ pub struct Store {
 
 impl Store {
     /// Open (creates parent dirs, enables WAL, runs migrations for this
-    /// schema). Same sequence the 5 original sibling crates ran byte-
-    /// identically.
+    /// schema).
     pub fn open(path: &Path, schema: &EntitySchema) -> Result<Self> {
         if let Some(parent) = path.parent() {
             let _ = std::fs::create_dir_all(parent);
@@ -62,13 +58,13 @@ impl Store {
 /// Also stamps `PRAGMA user_version` on fresh databases so future
 /// schema bumps can detect the target migration set exactly once.
 pub fn run_migrations(conn: &Connection, schema: &EntitySchema) -> Result<(), VerbError> {
-    conn.execute_batch(&ddl_primary_table(schema))?;
-    conn.execute_batch(&ddl_indexes(schema))?;
+    conn.execute_batch(&ddl::primary_table(schema))?;
+    conn.execute_batch(&ddl::indexes(schema))?;
     if let Some(cols) = schema.fts_columns {
-        conn.execute_batch(&ddl_fts_table(schema.table, cols))?;
+        conn.execute_batch(&ddl::fts_table(schema.table, cols))?;
     }
     if let Some(edge) = schema.edge_table {
-        conn.execute_batch(&ddl_edge_table_for(edge, schema.edge_key_kind))?;
+        conn.execute_batch(&ddl::edge_table_for(edge, schema.edge_key_kind))?;
     }
     for stmt in schema.custom_migrations {
         conn.execute_batch(stmt)?;
@@ -86,92 +82,7 @@ fn apply_user_version(conn: &Connection) -> Result<(), VerbError> {
         .pragma_query_value(None, "user_version", |r| r.get(0))
         .unwrap_or(0);
     if current < CURRENT_USER_VERSION {
-        // PRAGMA does not accept parameter binding; value is a constant.
         conn.pragma_update(None, "user_version", CURRENT_USER_VERSION)?;
     }
     Ok(())
 }
-
-fn ddl_primary_table(schema: &EntitySchema) -> String {
-    let cols: Vec<String> = schema.fields.iter().map(ddl_column).collect();
-    format!(
-        "CREATE TABLE IF NOT EXISTS {} (\n  {}\n);",
-        schema.table,
-        cols.join(",\n  ")
-    )
-}
-
-fn ddl_column(f: &FieldDef) -> String {
-    match f.kind {
-        FieldKind::IntegerPk => format!("{} INTEGER PRIMARY KEY", f.name),
-        FieldKind::IntegerNotNull => format!("{} INTEGER NOT NULL", f.name),
-        FieldKind::Integer => format!("{} INTEGER DEFAULT 0", f.name),
-        FieldKind::TextNotNull => format!("{} TEXT NOT NULL", f.name),
-        FieldKind::Text => format!("{} TEXT DEFAULT ''", f.name),
-        FieldKind::TextDefault => {
-            let d = f.default.unwrap_or("");
-            // SQL-escape embedded single quotes (per SQL standard: `'`
-            // → `''`) so `text_default("status", "don't know")` does
-            // not inject. Today all callers pass safe constants; this
-            // defence is for the first dev who doesn't.
-            let escaped = d.replace('\'', "''");
-            format!("{} TEXT NOT NULL DEFAULT '{}'", f.name, escaped)
-        }
-        FieldKind::TimestampCreated => format!("{} INTEGER NOT NULL", f.name),
-        FieldKind::TimestampUpdated => format!("{} INTEGER NOT NULL", f.name),
-    }
-}
-
-fn ddl_indexes(schema: &EntitySchema) -> String {
-    let mut out = String::new();
-    for f in schema.fields.iter().filter(|f| f.indexed) {
-        out.push_str(&format!(
-            "CREATE INDEX IF NOT EXISTS idx_{t}_{c} ON {t}({c});\n",
-            t = schema.table,
-            c = f.name
-        ));
-    }
-    out
-}
-
-fn ddl_fts_table(table: &str, cols: &[&str]) -> String {
-    let col_list = cols.join(", ");
-    format!(
-        "CREATE VIRTUAL TABLE IF NOT EXISTS fts_{table} \
-         USING fts5({table}_id UNINDEXED, {col_list}, tokenize='porter unicode61');"
-    )
-}
-
-fn ddl_edge_table(edge: &str) -> String {
-    format!(
-        "CREATE TABLE IF NOT EXISTS {edge} (\n  \
-            from_id INTEGER NOT NULL,\n  \
-            to_id INTEGER NOT NULL,\n  \
-            edge_type TEXT NOT NULL DEFAULT 'links',\n  \
-            PRIMARY KEY(from_id, to_id, edge_type)\n\
-         );\n\
-         CREATE INDEX IF NOT EXISTS idx_{edge}_to ON {edge}(to_id);"
-    )
-}
-
-/// Dispatcher — picks edge-table DDL for a given `EdgeKeyKind`. Added
-/// for kei-sage migration; `IntegerPair` branch preserves legacy body.
-fn ddl_edge_table_for(edge: &str, kind: EdgeKeyKind) -> String {
-    match kind {
-        EdgeKeyKind::IntegerPair => ddl_edge_table(edge),
-        EdgeKeyKind::TextPair => ddl_edge_table_text(edge),
-    }
-}
-
-/// Text-keyed edge DDL: `(src_path TEXT, dst_path TEXT, edge_type TEXT)`.
-fn ddl_edge_table_text(edge: &str) -> String {
-    format!(
-        "CREATE TABLE IF NOT EXISTS {edge} (\n  \
-            src_path TEXT NOT NULL,\n  \
-            dst_path TEXT NOT NULL,\n  \
-            edge_type TEXT NOT NULL DEFAULT 'links',\n  \
-            PRIMARY KEY(src_path, dst_path, edge_type)\n\
-         );\n\
-         CREATE INDEX IF NOT EXISTS idx_{edge}_dst ON {edge}(dst_path);"
-    )
-}
diff --git a/_primitives/_rust/kei-entity-store/src/error.rs b/_primitives/_rust/kei-entity-store/src/error.rs
index 023f260..6234c4b 100644
--- a/_primitives/_rust/kei-entity-store/src/error.rs
+++ b/_primitives/_rust/kei-entity-store/src/error.rs
@@ -21,8 +21,10 @@ pub enum VerbError {
     #[error("VerbDisabled: {verb} not enabled on schema {schema}")]
     VerbDisabled { verb: String, schema: String },
 
+    /// Generic not-found. `id` is rendered as text so the same variant
+    /// works for integer-PK and text-PK (UUID) schemas.
     #[error("NotFound: {entity} id {id}")]
-    NotFound { entity: String, id: i64 },
+    NotFound { entity: String, id: String },
 
     #[error("Sqlite: {0}")]
     Sqlite(#[from] rusqlite::Error),
@@ -46,4 +48,15 @@ impl VerbError {
             Self::Sqlite(_) | Self::Serde(_) | Self::Storage(_) => 1,
         }
     }
+
+    /// Construct a `NotFound` from an i64 id. Kept as a shim so existing
+    /// call-sites passing integer PKs keep compiling.
+    pub fn not_found_i64(entity: impl Into<String>, id: i64) -> Self {
+        Self::NotFound { entity: entity.into(), id: id.to_string() }
+    }
+
+    /// Construct a `NotFound` from a String id (TextPk schemas).
+    pub fn not_found_text(entity: impl Into<String>, id: impl Into<String>) -> Self {
+        Self::NotFound { entity: entity.into(), id: id.into() }
+    }
 }
diff --git a/_primitives/_rust/kei-entity-store/src/field.rs b/_primitives/_rust/kei-entity-store/src/field.rs
new file mode 100644
index 0000000..2b3a1f6
--- /dev/null
+++ b/_primitives/_rust/kei-entity-store/src/field.rs
@@ -0,0 +1,94 @@
+//! `FieldDef` — one column in an `EntitySchema`. Split out of
+//! `schema.rs` to keep both files under the Constructor-Pattern
+//! 200-LOC cap.
+
+use crate::schema::FieldKind;
+
+/// One column in an EntitySchema.
+#[derive(Debug, Clone, Copy)]
+pub struct FieldDef {
+    pub name: &'static str,
+    pub kind: FieldKind,
+    /// Default literal for TextDefault / IntegerNotNull (as SQL literal
+    /// WITHOUT surrounding quotes — engine quotes TEXT automatically).
+    pub default: Option<&'static str>,
+    /// Emit a single-column index `idx_<table>_<name>`.
+    pub indexed: bool,
+    /// Default for `Real` / `RealDefault` columns. `None` means 0.0.
+    pub real_default: Option<f64>,
+    /// Sentinel pair for `TextArchiveEnum` — `(active, archived)`.
+    /// Ignored for other kinds. `None` falls back to
+    /// `("active", "archived")`.
+    pub archive_enum: Option<(&'static str, &'static str)>,
+}
+
+impl FieldDef {
+    pub const fn pk(name: &'static str) -> Self {
+        Self::base(name, FieldKind::IntegerPk)
+    }
+    pub const fn text_pk(name: &'static str) -> Self {
+        Self::base(name, FieldKind::TextPk)
+    }
+    pub const fn text(name: &'static str) -> Self {
+        Self::base(name, FieldKind::Text)
+    }
+    pub const fn text_nn(name: &'static str) -> Self {
+        Self::base(name, FieldKind::TextNotNull)
+    }
+    pub const fn text_default(name: &'static str, default: &'static str) -> Self {
+        let mut f = Self::base(name, FieldKind::TextDefault);
+        f.default = Some(default);
+        f
+    }
+    pub const fn integer(name: &'static str) -> Self {
+        Self::base(name, FieldKind::Integer)
+    }
+    pub const fn integer_nn(name: &'static str) -> Self {
+        Self::base(name, FieldKind::IntegerNotNull)
+    }
+    pub const fn real(name: &'static str) -> Self {
+        Self::base(name, FieldKind::Real)
+    }
+    pub const fn real_default(name: &'static str, default: f64) -> Self {
+        let mut f = Self::base(name, FieldKind::RealDefault);
+        f.real_default = Some(default);
+        f
+    }
+    pub const fn text_archive_enum(
+        name: &'static str,
+        active: &'static str,
+        archived: &'static str,
+    ) -> Self {
+        let mut f = Self::base(name, FieldKind::TextArchiveEnum);
+        f.archive_enum = Some((active, archived));
+        f
+    }
+    pub const fn created_at() -> Self {
+        Self::base("created_at", FieldKind::TimestampCreated)
+    }
+    pub const fn updated_at() -> Self {
+        Self::base("updated_at", FieldKind::TimestampUpdated)
+    }
+    pub const fn with_index(mut self) -> Self {
+        self.indexed = true;
+        self
+    }
+
+    /// Internal base constructor — zeroes optional fields so the
+    /// per-kind builders above stay one-liners.
+    const fn base(name: &'static str, kind: FieldKind) -> Self {
+        Self {
+            name,
+            kind,
+            default: None,
+            indexed: false,
+            real_default: None,
+            archive_enum: None,
+        }
+    }
+
+    /// True if this FieldDef is a primary key (either integer or text).
+    pub fn is_pk(&self) -> bool {
+        matches!(self.kind, FieldKind::IntegerPk | FieldKind::TextPk)
+    }
+}
diff --git a/_primitives/_rust/kei-entity-store/src/lib.rs b/_primitives/_rust/kei-entity-store/src/lib.rs
index e45b181..4db6341 100644
--- a/_primitives/_rust/kei-entity-store/src/lib.rs
+++ b/_primitives/_rust/kei-entity-store/src/lib.rs
@@ -14,8 +14,10 @@
 //! Per substrate schema v1 this crate stays library-only — no CLI, no
 //! `bin`. Each sibling crate remains the user-facing binary.
 
+pub mod ddl;
 pub mod engine;
 pub mod error;
+pub mod field;
 pub mod schema;
 pub mod verbs;
 
diff --git a/_primitives/_rust/kei-entity-store/src/schema.rs b/_primitives/_rust/kei-entity-store/src/schema.rs
index 6813188..3e3fbc1 100644
--- a/_primitives/_rust/kei-entity-store/src/schema.rs
+++ b/_primitives/_rust/kei-entity-store/src/schema.rs
@@ -5,13 +5,18 @@
 //! structure to know: table name, fields to INSERT/SELECT, FTS columns,
 //! edge table (for link/rank), and which verbs are enabled.
 
+pub use crate::field::FieldDef;
+
 /// Field kinds the engine knows how to bind for INSERT / UPDATE and
 /// how to read in SELECT. A field's `kind` also drives the CREATE TABLE
 /// DDL produced by the engine's migration runner.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum FieldKind {
-    /// INTEGER PRIMARY KEY — exactly one per schema (name = "id").
+    /// INTEGER PRIMARY KEY — exactly one PK per schema. Name = "id".
     IntegerPk,
+    /// TEXT PRIMARY KEY — caller supplies the PK value (e.g. UUID).
+    /// Mutually exclusive with `IntegerPk` within a single schema.
+    TextPk,
     /// INTEGER NOT NULL (with optional DEFAULT 0).
     IntegerNotNull,
     /// INTEGER, default 0.
@@ -22,73 +27,63 @@ pub enum FieldKind {
     Text,
     /// TEXT NOT NULL with explicit default value (held in `default`).
     TextDefault,
+    /// TEXT NOT NULL representing a soft-delete enum with named
+    /// sentinel values (`active` / `archived`). When used as the
+    /// schema's `archived_field`, the `archive` verb writes the
+    /// `archived` sentinel instead of flipping an integer.
+    /// Default at insert = `active` sentinel.
+    TextArchiveEnum,
+    /// REAL (f64) NOT NULL, default 0.0.
+    Real,
+    /// REAL (f64) NOT NULL with an explicit default (held in
+    /// `real_default`).
+    RealDefault,
     /// Unix-timestamp INTEGER auto-stamped on insert (created_at).
     TimestampCreated,
     /// Unix-timestamp INTEGER auto-stamped on insert + update (updated_at).
     TimestampUpdated,
 }
 
-/// One column in an EntitySchema.
-#[derive(Debug, Clone, Copy)]
-pub struct FieldDef {
-    pub name: &'static str,
-    pub kind: FieldKind,
-    /// Default literal for TextDefault / IntegerNotNull (as SQL literal
-    /// WITHOUT surrounding quotes — engine quotes TEXT automatically).
-    pub default: Option<&'static str>,
-    /// Emit a single-column index `idx_<table>_<name>`.
-    pub indexed: bool,
-}
-
-impl FieldDef {
-    pub const fn pk(name: &'static str) -> Self {
-        Self { name, kind: FieldKind::IntegerPk, default: None, indexed: false }
-    }
-    pub const fn text(name: &'static str) -> Self {
-        Self { name, kind: FieldKind::Text, default: None, indexed: false }
-    }
-    pub const fn text_nn(name: &'static str) -> Self {
-        Self { name, kind: FieldKind::TextNotNull, default: None, indexed: false }
-    }
-    pub const fn text_default(name: &'static str, default: &'static str) -> Self {
-        Self { name, kind: FieldKind::TextDefault, default: Some(default), indexed: false }
-    }
-    pub const fn integer(name: &'static str) -> Self {
-        Self { name, kind: FieldKind::Integer, default: None, indexed: false }
-    }
-    pub const fn integer_nn(name: &'static str) -> Self {
-        Self { name, kind: FieldKind::IntegerNotNull, default: None, indexed: false }
-    }
-    pub const fn created_at() -> Self {
-        Self { name: "created_at", kind: FieldKind::TimestampCreated,
-               default: None, indexed: false }
-    }
-    pub const fn updated_at() -> Self {
-        Self { name: "updated_at", kind: FieldKind::TimestampUpdated,
-               default: None, indexed: false }
-    }
-    pub const fn with_index(mut self) -> Self {
-        self.indexed = true;
-        self
-    }
-}
-
 /// Edge-key storage strategy for the schema's `edge_table`.
 ///
 /// - `IntegerPair` (default) — legacy `(from_id INTEGER, to_id INTEGER,
 ///   edge_type TEXT)` — matches kei-task byte-for-byte.
 /// - `TextPair` — `(src_path TEXT, dst_path TEXT, edge_type TEXT)` —
 ///   required by kei-sage (composite text keys, no integer ids).
+/// - `TextPairWithMetadata` — same text key but with optional
+///   `id`/`weight`/`created_at` columns so edges can carry metadata
+///   (kei-chat-store cross-refs, kei-content-store citations).
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EdgeKeyKind {
     IntegerPair,
     TextPair,
+    /// Extended text-pair edge with optional metadata columns.
+    /// Existing `TextPair` stays backward-compat.
+    TextPairWithMetadata {
+        /// Emit `edge_id INTEGER PRIMARY KEY AUTOINCREMENT` column.
+        has_id: bool,
+        /// Emit `weight REAL NOT NULL DEFAULT 1.0` column.
+        has_weight: bool,
+        /// Emit `created_at INTEGER NOT NULL` column auto-stamped on
+        /// insert.
+        has_created_at: bool,
+    },
 }
 
 impl Default for EdgeKeyKind {
     fn default() -> Self { Self::IntegerPair }
 }
 
+impl EdgeKeyKind {
+    /// True if this edge variant uses TEXT keys (any text variant).
+    pub fn is_text(&self) -> bool {
+        matches!(
+            self,
+            EdgeKeyKind::TextPair | EdgeKeyKind::TextPairWithMetadata { .. }
+        )
+    }
+}
+
 /// Declarative schema for one entity.
 #[derive(Debug, Clone, Copy)]
 pub struct EntitySchema {
@@ -112,10 +107,12 @@ pub struct EntitySchema {
     /// `(from_id, to_id)` schema; `TextPair` switches to
     /// `(src_path, dst_path)` for path-keyed graphs (kei-sage).
     pub edge_key_kind: EdgeKeyKind,
-    /// If `Some`, enables the `archive` verb. Names the INTEGER column
-    /// used as the soft-delete flag (0 / 1). The `archive` verb flips
-    /// it to 1 and stamps a sibling `<field>_at` timestamp if such a
-    /// column exists in `fields`.
+    /// If `Some`, enables the `archive` verb. Names the column used as
+    /// the soft-delete marker. If the column's kind is `TextArchiveEnum`
+    /// the verb writes the `archived` sentinel; otherwise (integer
+    /// column) it flips to 1. In both cases a sibling `<field>_at`
+    /// INTEGER column is stamped with the current Unix timestamp if
+    /// present in `fields`.
     pub archived_field: Option<&'static str>,
     /// Arbitrary DDL statements run after the primary table + FTS +
     /// edge table have been created. Used for secondary tables
@@ -125,12 +122,13 @@ pub struct EntitySchema {
 }
 
 impl EntitySchema {
-    /// Returns the PK column (always "id" by convention).
+    /// Returns the PK column (integer or text). Panics if the schema
+    /// has no PK — schema authors must declare exactly one.
     pub fn pk(&self) -> &FieldDef {
         self.fields
             .iter()
-            .find(|f| f.kind == FieldKind::IntegerPk)
-            .expect("EntitySchema MUST have exactly one IntegerPk field")
+            .find(|f| f.is_pk())
+            .expect("EntitySchema MUST have exactly one PK field (IntegerPk or TextPk)")
     }
 
     /// Returns true if `verb` appears in `enabled_verbs`.
@@ -141,6 +139,11 @@ impl EntitySchema {
     /// Returns the list of non-PK field names, in order. Used by the
     /// `create` verb to build the INSERT column-list.
     pub fn writable_fields(&self) -> impl Iterator<Item = &FieldDef> {
-        self.fields.iter().filter(|f| f.kind != FieldKind::IntegerPk)
+        self.fields.iter().filter(|f| !f.is_pk())
+    }
+
+    /// Look up a field by name.
+    pub fn field(&self, name: &str) -> Option<&FieldDef> {
+        self.fields.iter().find(|f| f.name == name)
     }
 }
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/archive.rs b/_primitives/_rust/kei-entity-store/src/verbs/archive.rs
index af047ae..ef657b3 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/archive.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/archive.rs
@@ -1,19 +1,22 @@
-//! `archive` verb — soft-delete. Flips `<archived_field>` to 1, stamps
-//! a sibling `<archived_field>_at` column with current Unix timestamp
-//! if such a column exists in the schema.
+//! `archive` verb — soft-delete. If the configured `archived_field`
+//! column has kind `TextArchiveEnum`, writes the column's
+//! `archived` sentinel string; otherwise flips an INTEGER column to 1.
+//! A sibling `<archived_field>_at` column is stamped with the current
+//! Unix timestamp when present.
 //!
 //! Required schema configuration: `archived_field: Some("<col>")`.
 //! Without it the verb errors with `InvalidInput` — the engine does NOT
 //! fall back to legacy `archived` heuristics (those remain in
 //! `delete.rs` soft-path only).
 //!
-//! Input: `{ id: i64 }`.
+//! Input: `{ id: <int|string> }`.
 //! Output: `{ id, archived_at }` — `archived_at` is the stamped
 //! timestamp when a `<field>_at` column exists, else `null`.
 
 use crate::error::VerbError;
-use crate::schema::EntitySchema;
-use rusqlite::Connection;
+use crate::schema::{EntitySchema, FieldKind};
+use crate::verbs::pk::{self, PkValue};
+use rusqlite::{types::Value as SqlValue, Connection};
 use serde_json::{json, Value};
 
 pub fn run(
@@ -21,44 +24,80 @@ pub fn run(
     schema: &EntitySchema,
     input: Value,
 ) -> Result<Value, VerbError> {
+    guard_enabled(schema)?;
+    let field_name = schema.archived_field.ok_or_else(|| {
+        VerbError::InvalidInput(format!(
+            "archive: schema {} has no archived_field configured",
+            schema.name
+        ))
+    })?;
+    let id = pk::extract(schema, &input, "archive")?;
+    let ts_col = format!("{field_name}_at");
+    let has_ts = schema.fields.iter().any(|f| f.name == ts_col);
+    let now: i64 = chrono::Utc::now().timestamp();
+
+    let rows = execute_archive(conn, schema, field_name, &ts_col, has_ts, &id, now)?;
+    if rows == 0 {
+        return Err(VerbError::not_found_text(schema.name, id.as_string()));
+    }
+    let stamped = if has_ts { json!(now) } else { Value::Null };
+    Ok(json!({ "id": id.as_json(), "archived_at": stamped }))
+}
+
+fn guard_enabled(schema: &EntitySchema) -> Result<(), VerbError> {
     if !schema.verb_enabled("archive") {
         return Err(VerbError::VerbDisabled {
             verb: "archive".into(),
             schema: schema.name.into(),
         });
     }
-    let field = schema.archived_field.ok_or_else(|| {
-        VerbError::InvalidInput(format!(
-            "archive: schema {} has no archived_field configured",
-            schema.name
-        ))
-    })?;
-    let id = input
-        .get("id")
-        .and_then(|v| v.as_i64())
-        .ok_or_else(|| VerbError::InvalidInput("archive: missing `id` integer".into()))?;
-
-    let ts_col = format!("{field}_at");
-    let has_ts = schema.fields.iter().any(|f| f.name == ts_col);
-    let now: i64 = chrono::Utc::now().timestamp();
+    Ok(())
+}
 
+fn execute_archive(
+    conn: &Connection,
+    schema: &EntitySchema,
+    field_name: &str,
+    ts_col: &str,
+    has_ts: bool,
+    id: &PkValue,
+    now: i64,
+) -> Result<usize, VerbError> {
+    let marker = archive_marker(schema, field_name);
+    let pk_name = schema.pk().name;
     let rows = if has_ts {
         conn.execute(
             &format!(
-                "UPDATE {t} SET {field} = 1, {ts_col} = ?1 WHERE id = ?2",
+                "UPDATE {t} SET {field_name} = ?1, {ts_col} = ?2 WHERE {pk_name} = ?3",
                 t = schema.table
             ),
-            rusqlite::params![now, id],
+            rusqlite::params![marker, now, id.as_sql()],
         )?
     } else {
         conn.execute(
-            &format!("UPDATE {t} SET {field} = 1 WHERE id = ?1", t = schema.table),
-            rusqlite::params![id],
+            &format!(
+                "UPDATE {t} SET {field_name} = ?1 WHERE {pk_name} = ?2",
+                t = schema.table
+            ),
+            rusqlite::params![marker, id.as_sql()],
         )?
     };
-    if rows == 0 {
-        return Err(VerbError::NotFound { entity: schema.name.into(), id });
-    }
-    let stamped = if has_ts { json!(now) } else { Value::Null };
-    Ok(json!({ "id": id, "archived_at": stamped }))
+    Ok(rows)
+}
+
+/// Pick the SQL value written to the archived column. `TextArchiveEnum`
+/// columns receive the `archived` sentinel; any other kind receives
+/// the integer flag `1` (legacy behaviour).
+fn archive_marker(schema: &EntitySchema, field_name: &str) -> SqlValue {
+    let Some(field) = schema.field(field_name) else {
+        return SqlValue::Integer(1);
+    };
+    match field.kind {
+        FieldKind::TextArchiveEnum => {
+            let (_active, archived) =
+                field.archive_enum.unwrap_or(("active", "archived"));
+            SqlValue::Text(archived.to_string())
+        }
+        _ => SqlValue::Integer(1),
+    }
 }
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/create.rs b/_primitives/_rust/kei-entity-store/src/verbs/create.rs
index 63bbfbc..1d432c1 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/create.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/create.rs
@@ -1,17 +1,12 @@
 //! `create` verb — INSERT one row using fields declared on the schema.
+//! Per-kind value defaulting lives in `create_defaults`.
 //!
-//! Input JSON shape: `{ "<field_name>": <value>, ... }`. Only fields
-//! declared on the EntitySchema are copied; extras are silently ignored
-//! (the atom layer above is responsible for rejecting them if desired).
-//! Output: `{ "id": <rowid>, "created_at": <unix ts> }`.
-//!
-//! Type discipline: when a key is present its JSON kind MUST match the
-//! field kind (string for Text*, number for Integer*). Mismatch →
-//! `VerbError::InvalidType`. Missing keys default to 0 / "" as before.
+//! TextPk schemas require the caller to supply `id`; IntegerPk schemas
+//! get an auto-assigned rowid. Output `{id, created_at}`.
 
 use crate::error::VerbError;
-use crate::schema::{EntitySchema, FieldDef, FieldKind};
-use crate::verbs::validate;
+use crate::schema::{EntitySchema, FieldKind};
+use crate::verbs::create_defaults::field_value;
 use chrono::Utc;
 use rusqlite::{types::Value as SqlValue, Connection};
 use serde_json::{json, Value};
@@ -26,8 +21,8 @@ pub fn run(
     let now = Utc::now().timestamp();
     let (cols, values) = build_insert(schema, obj, now)?;
     let id = insert_tx(conn, schema, &cols, &values, obj)?;
-    let created_at = read_created_at(conn, schema, id).unwrap_or(now);
-    Ok(json!({ "id": id, "created_at": created_at }))
+    let created_at = read_created_at(conn, schema, &id).unwrap_or(now);
+    Ok(json!({ "id": id_to_json(&id), "created_at": created_at }))
 }
 
 fn guard_enabled(schema: &EntitySchema) -> Result<(), VerbError> {
@@ -40,21 +35,40 @@ fn guard_enabled(schema: &EntitySchema) -> Result<(), VerbError> {
     Ok(())
 }
 
-/// Wrap INSERT + FTS reindex in one transaction so a rusqlite failure
-/// in the FTS leg rolls back the row insert too. `unchecked_transaction`
-/// is used because callers hold `&Connection` — rusqlite permits this
-/// as long as only one tx is in flight.
+/// Stored PK of the inserted row. `Integer` for auto-rowid schemas,
+/// `Text` for caller-supplied TEXT PKs.
+pub(super) enum InsertedPk {
+    Integer(i64),
+    Text(String),
+}
+
+fn id_to_json(pk: &InsertedPk) -> Value {
+    match pk {
+        InsertedPk::Integer(n) => Value::from(*n),
+        InsertedPk::Text(s) => Value::from(s.clone()),
+    }
+}
+
+fn pk_sql(pk: &InsertedPk) -> SqlValue {
+    match pk {
+        InsertedPk::Integer(n) => SqlValue::Integer(*n),
+        InsertedPk::Text(s) => SqlValue::Text(s.clone()),
+    }
+}
+
+/// INSERT + FTS reindex wrapped in one `unchecked_transaction` so a
+/// mid-flight FTS failure rolls back the row insert too.
 fn insert_tx(
     conn: &Connection,
     schema: &EntitySchema,
     cols: &[&'static str],
     values: &[SqlValue],
     obj: &serde_json::Map<String, Value>,
-) -> Result<i64, VerbError> {
+) -> Result<InsertedPk, VerbError> {
     let tx = conn.unchecked_transaction()?;
-    let id = exec_insert_tx(&tx, schema, cols, values)?;
+    let id = exec_insert_tx(&tx, schema, cols, values, obj)?;
     if let Some(fts_cols) = schema.fts_columns {
-        reindex_fts(&tx, schema.table, fts_cols, id, obj)?;
+        reindex_fts(&tx, schema, fts_cols, &id, obj)?;
     }
     tx.commit()?;
     Ok(id)
@@ -65,21 +79,61 @@ fn exec_insert_tx(
     schema: &EntitySchema,
     cols: &[&'static str],
     values: &[SqlValue],
-) -> Result<i64, VerbError> {
+    obj: &serde_json::Map<String, Value>,
+) -> Result<InsertedPk, VerbError> {
+    if schema.pk().kind == FieldKind::TextPk {
+        return exec_text_pk_insert(tx, schema, cols, values, obj);
+    }
+    exec_raw_insert(tx, schema.table, cols, values)?;
+    Ok(InsertedPk::Integer(tx.last_insert_rowid()))
+}
+
+fn exec_text_pk_insert(
+    tx: &rusqlite::Transaction<'_>,
+    schema: &EntitySchema,
+    cols: &[&'static str],
+    values: &[SqlValue],
+    obj: &serde_json::Map<String, Value>,
+) -> Result<InsertedPk, VerbError> {
+    let pk_name = schema.pk().name;
+    let id_str = obj
+        .get("id")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| {
+            VerbError::InvalidInput("create: `id` required for TextPk schemas".into())
+        })?
+        .to_string();
+    let mut all_cols: Vec<&'static str> = vec![pk_name];
+    all_cols.extend_from_slice(cols);
+    let mut all_vals: Vec<SqlValue> = vec![SqlValue::Text(id_str.clone())];
+    all_vals.extend_from_slice(values);
+    exec_raw_insert(tx, schema.table, &all_cols, &all_vals)?;
+    Ok(InsertedPk::Text(id_str))
+}
+
+fn exec_raw_insert(
+    tx: &rusqlite::Transaction<'_>,
+    table: &str,
+    cols: &[&'static str],
+    values: &[SqlValue],
+) -> Result<(), VerbError> {
     let placeholders: Vec<String> = (1..=cols.len()).map(|i| format!("?{i}")).collect();
     let sql = format!(
         "INSERT INTO {} ({}) VALUES ({})",
-        schema.table,
+        table,
         cols.join(","),
         placeholders.join(","),
     );
     let params: Vec<&dyn rusqlite::ToSql> =
         values.iter().map(|v| v as &dyn rusqlite::ToSql).collect();
     tx.execute(&sql, params.as_slice())?;
-    Ok(tx.last_insert_rowid())
+    Ok(())
 }
 
-fn as_object<'a>(v: &'a Value, verb: &str) -> Result<&'a serde_json::Map<String, Value>, VerbError> {
+fn as_object<'a>(
+    v: &'a Value,
+    verb: &str,
+) -> Result<&'a serde_json::Map<String, Value>, VerbError> {
     v.as_object()
         .ok_or_else(|| VerbError::InvalidInput(format!("{verb}: expected JSON object")))
 }
@@ -93,74 +147,23 @@ fn build_insert(
     let mut values: Vec<SqlValue> = Vec::new();
     for f in schema.writable_fields() {
         cols.push(f.name);
-        values.push(field_value_for_insert(f, input, now)?);
+        values.push(field_value(f, input, now)?);
     }
     Ok((cols, values))
 }
 
-fn field_value_for_insert(
-    f: &FieldDef,
-    input: &serde_json::Map<String, Value>,
-    now: i64,
-) -> Result<SqlValue, VerbError> {
-    match f.kind {
-        FieldKind::TimestampCreated | FieldKind::TimestampUpdated => {
-            Ok(match input.get(f.name).and_then(|v| v.as_i64()) {
-                Some(ts) if ts > 0 => SqlValue::Integer(ts),
-                _ => SqlValue::Integer(now),
-            })
-        }
-        FieldKind::TextDefault => insert_text_default(f, input),
-        FieldKind::IntegerPk => Ok(SqlValue::Null),
-        _ => match input.get(f.name) {
-            Some(raw) => validate::coerce(f, raw),
-            None => Ok(default_for_kind(f)),
-        },
-    }
-}
-
-fn insert_text_default(
-    f: &FieldDef,
-    input: &serde_json::Map<String, Value>,
-) -> Result<SqlValue, VerbError> {
-    match input.get(f.name) {
-        Some(raw) => {
-            let coerced = validate::coerce(f, raw)?;
-            if let SqlValue::Text(ref s) = coerced {
-                if s.is_empty() {
-                    let d = f.default.unwrap_or("");
-                    validate::check_text_len(f, d)?;
-                    return Ok(SqlValue::Text(d.to_string()));
-                }
-            }
-            Ok(coerced)
-        }
-        None => {
-            let d = f.default.unwrap_or("");
-            validate::check_text_len(f, d)?;
-            Ok(SqlValue::Text(d.to_string()))
-        }
-    }
-}
-
-fn default_for_kind(f: &FieldDef) -> SqlValue {
-    match f.kind {
-        FieldKind::IntegerNotNull | FieldKind::Integer => SqlValue::Integer(0),
-        FieldKind::TextNotNull | FieldKind::Text => SqlValue::Text(String::new()),
-        _ => SqlValue::Null,
-    }
-}
-
 fn reindex_fts(
     tx: &rusqlite::Transaction<'_>,
-    table: &str,
+    schema: &EntitySchema,
     cols: &[&str],
-    id: i64,
+    id: &InsertedPk,
     input: &serde_json::Map<String, Value>,
 ) -> Result<(), VerbError> {
+    let table = schema.table;
+    let pk_param = pk_sql(id);
     tx.execute(
         &format!("DELETE FROM fts_{table} WHERE {table}_id=?1"),
-        rusqlite::params![id],
+        rusqlite::params![pk_param],
     )?;
     let placeholders: Vec<String> = (2..=(cols.len() + 1)).map(|i| format!("?{i}")).collect();
     let sql = format!(
@@ -168,7 +171,7 @@ fn reindex_fts(
         cols.join(", "),
         placeholders.join(", "),
     );
-    let mut values: Vec<SqlValue> = vec![SqlValue::Integer(id)];
+    let mut values: Vec<SqlValue> = vec![pk_param];
     for c in cols {
         let v = input.get(*c).and_then(|v| v.as_str()).unwrap_or("").to_string();
         values.push(SqlValue::Text(v));
@@ -179,16 +182,10 @@ fn reindex_fts(
     Ok(())
 }
 
-fn read_created_at(conn: &Connection, schema: &EntitySchema, id: i64) -> Option<i64> {
-    let has_created = schema
-        .fields
-        .iter()
-        .any(|f| f.kind == FieldKind::TimestampCreated);
-    if !has_created { return None; }
-    conn.query_row(
-        &format!("SELECT created_at FROM {} WHERE id=?1", schema.table),
-        rusqlite::params![id],
-        |r| r.get::<_, i64>(0),
-    )
-    .ok()
+fn read_created_at(conn: &Connection, schema: &EntitySchema, id: &InsertedPk) -> Option<i64> {
+    if !schema.fields.iter().any(|f| f.kind == FieldKind::TimestampCreated) {
+        return None;
+    }
+    let sql = format!("SELECT created_at FROM {} WHERE {}=?1", schema.table, schema.pk().name);
+    conn.query_row(&sql, rusqlite::params![pk_sql(id)], |r| r.get::<_, i64>(0)).ok()
 }
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/create_defaults.rs b/_primitives/_rust/kei-entity-store/src/verbs/create_defaults.rs
new file mode 100644
index 0000000..f6d2c3d
--- /dev/null
+++ b/_primitives/_rust/kei-entity-store/src/verbs/create_defaults.rs
@@ -0,0 +1,109 @@
+//! Per-kind value-for-insert helpers split out of `create.rs` to keep
+//! that file under the Constructor-Pattern 200-LOC cap. Each function
+//! handles one FieldKind's default / coerce logic.
+
+use crate::error::VerbError;
+use crate::schema::{FieldDef, FieldKind};
+use crate::verbs::validate;
+use rusqlite::types::Value as SqlValue;
+use serde_json::Value;
+
+pub fn field_value(
+    f: &FieldDef,
+    input: &serde_json::Map<String, Value>,
+    now: i64,
+) -> Result<SqlValue, VerbError> {
+    match f.kind {
+        FieldKind::TimestampCreated | FieldKind::TimestampUpdated => Ok(timestamp(input, f, now)),
+        FieldKind::TextDefault => text_default(f, input),
+        FieldKind::TextArchiveEnum => archive_enum(f, input),
+        FieldKind::RealDefault => real_default(f, input),
+        FieldKind::IntegerPk | FieldKind::TextPk => Ok(SqlValue::Null),
+        _ => match input.get(f.name) {
+            Some(raw) => validate::coerce(f, raw),
+            None => Ok(default_for_kind(f)),
+        },
+    }
+}
+
+fn timestamp(
+    input: &serde_json::Map<String, Value>,
+    f: &FieldDef,
+    now: i64,
+) -> SqlValue {
+    match input.get(f.name).and_then(|v| v.as_i64()) {
+        Some(ts) if ts > 0 => SqlValue::Integer(ts),
+        _ => SqlValue::Integer(now),
+    }
+}
+
+fn text_default(
+    f: &FieldDef,
+    input: &serde_json::Map<String, Value>,
+) -> Result<SqlValue, VerbError> {
+    match input.get(f.name) {
+        Some(raw) => coerce_with_text_fallback(f, raw),
+        None => text_literal_default(f),
+    }
+}
+
+fn coerce_with_text_fallback(
+    f: &FieldDef,
+    raw: &Value,
+) -> Result<SqlValue, VerbError> {
+    let coerced = validate::coerce(f, raw)?;
+    if let SqlValue::Text(ref s) = coerced {
+        if s.is_empty() {
+            return text_literal_default(f);
+        }
+    }
+    Ok(coerced)
+}
+
+fn text_literal_default(f: &FieldDef) -> Result<SqlValue, VerbError> {
+    let d = f.default.unwrap_or("");
+    validate::check_text_len(f, d)?;
+    Ok(SqlValue::Text(d.to_string()))
+}
+
+fn archive_enum(
+    f: &FieldDef,
+    input: &serde_json::Map<String, Value>,
+) -> Result<SqlValue, VerbError> {
+    let (active, _archived) = f.archive_enum.unwrap_or(("active", "archived"));
+    match input.get(f.name) {
+        Some(raw) => {
+            let coerced = validate::coerce(f, raw)?;
+            if let SqlValue::Text(ref s) = coerced {
+                if s.is_empty() {
+                    return Ok(SqlValue::Text(active.to_string()));
+                }
+            }
+            Ok(coerced)
+        }
+        None => Ok(SqlValue::Text(active.to_string())),
+    }
+}
+
+fn real_default(
+    f: &FieldDef,
+    input: &serde_json::Map<String, Value>,
+) -> Result<SqlValue, VerbError> {
+    match input.get(f.name) {
+        Some(raw) => validate::coerce(f, raw),
+        None => Ok(SqlValue::Real(f.real_default.unwrap_or(0.0))),
+    }
+}
+
+fn default_for_kind(f: &FieldDef) -> SqlValue {
+    match f.kind {
+        FieldKind::IntegerNotNull | FieldKind::Integer => SqlValue::Integer(0),
+        FieldKind::TextNotNull | FieldKind::Text | FieldKind::TextDefault => {
+            SqlValue::Text(String::new())
+        }
+        FieldKind::Real | FieldKind::RealDefault => {
+            SqlValue::Real(f.real_default.unwrap_or(0.0))
+        }
+        _ => SqlValue::Null,
+    }
+}
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/delete.rs b/_primitives/_rust/kei-entity-store/src/verbs/delete.rs
index 1e43358..92b451b 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/delete.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/delete.rs
@@ -3,6 +3,7 @@
 
 use crate::error::VerbError;
 use crate::schema::EntitySchema;
+use crate::verbs::pk;
 use rusqlite::Connection;
 use serde_json::{json, Value};
 
@@ -17,34 +18,38 @@ pub fn run(
             schema: schema.name.into(),
         });
     }
-    let id = input
-        .get("id")
-        .and_then(|v| v.as_i64())
-        .ok_or_else(|| VerbError::InvalidInput("delete: missing `id` integer".into()))?;
+    let id = pk::extract(schema, &input, "delete")?;
     let soft = input.get("soft").and_then(|v| v.as_bool()).unwrap_or(false);
 
     let rows = if soft && has_archived_field(schema) {
         conn.execute(
-            &format!("UPDATE {} SET archived = 1 WHERE id=?1", schema.table),
-            rusqlite::params![id],
+            &format!(
+                "UPDATE {} SET archived = 1 WHERE {}=?1",
+                schema.table,
+                schema.pk().name
+            ),
+            rusqlite::params![id.as_sql()],
         )?
     } else {
-        if let Some(cols) = schema.fts_columns {
-            let _ = cols; // silence unused warning if fts disabled
+        if schema.fts_columns.is_some() {
             conn.execute(
                 &format!("DELETE FROM fts_{} WHERE {}_id=?1", schema.table, schema.table),
-                rusqlite::params![id],
+                rusqlite::params![id.as_sql()],
             )?;
         }
         conn.execute(
-            &format!("DELETE FROM {} WHERE id=?1", schema.table),
-            rusqlite::params![id],
+            &format!(
+                "DELETE FROM {} WHERE {}=?1",
+                schema.table,
+                schema.pk().name
+            ),
+            rusqlite::params![id.as_sql()],
         )?
     };
     if rows == 0 {
-        return Err(VerbError::NotFound { entity: schema.name.into(), id });
+        return Err(VerbError::not_found_text(schema.name, id.as_string()));
     }
-    Ok(json!({ "ok": true, "id": id }))
+    Ok(json!({ "ok": true, "id": id.as_json() }))
 }
 
 fn has_archived_field(schema: &EntitySchema) -> bool {
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/get.rs b/_primitives/_rust/kei-entity-store/src/verbs/get.rs
index 22def5c..e6b0e4d 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/get.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/get.rs
@@ -3,6 +3,7 @@
 
 use crate::error::VerbError;
 use crate::schema::{EntitySchema, FieldDef, FieldKind};
+use crate::verbs::pk;
 use rusqlite::Connection;
 use serde_json::{Map, Value};
 
@@ -17,25 +18,19 @@ pub fn run(
             schema: schema.name.into(),
         });
     }
-    let id = input
-        .get("id")
-        .and_then(|v| v.as_i64())
-        .ok_or_else(|| VerbError::InvalidInput("get: missing `id` integer".into()))?;
-
+    let id = pk::extract(schema, &input, "get")?;
     let cols: Vec<&str> = schema.fields.iter().map(|f| f.name).collect();
     let sql = format!(
-        "SELECT {} FROM {} WHERE id=?1",
+        "SELECT {} FROM {} WHERE {}=?1",
         cols.join(","),
-        schema.table
+        schema.table,
+        schema.pk().name
     );
     let mut stmt = conn.prepare(&sql)?;
-    let mut rows = stmt.query(rusqlite::params![id])?;
+    let mut rows = stmt.query(rusqlite::params![id.as_sql()])?;
     match rows.next()? {
         Some(r) => Ok(row_to_json(schema, r)?),
-        None => Err(VerbError::NotFound {
-            entity: schema.name.into(),
-            id,
-        }),
+        None => Err(VerbError::not_found_text(schema.name, id.as_string())),
     }
 }
 
@@ -60,9 +55,18 @@ fn field_to_json(f: &FieldDef, row: &rusqlite::Row, idx: usize) -> Result<Value,
             let n: i64 = row.get(idx)?;
             Value::from(n)
         }
-        FieldKind::TextNotNull | FieldKind::Text | FieldKind::TextDefault => {
+        FieldKind::TextPk
+        | FieldKind::TextNotNull
+        | FieldKind::Text
+        | FieldKind::TextDefault
+        | FieldKind::TextArchiveEnum => {
             let s: String = row.get(idx)?;
             Value::from(s)
         }
+        FieldKind::Real | FieldKind::RealDefault => {
+            let n: f64 = row.get(idx)?;
+            Value::from(n)
+        }
     })
 }
+
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/link.rs b/_primitives/_rust/kei-entity-store/src/verbs/link.rs
index 113985f..334ee2b 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/link.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/link.rs
@@ -4,12 +4,15 @@
 //! crate (e.g. kei-task::deps).
 //!
 //! Dispatches on `schema.edge_key_kind`:
-//! - `IntegerPair` — input `{from: i64, to: i64, edge_type?}`
-//! - `TextPair`    — input `{from: str, to: str, edge_type?}`
+//! - `IntegerPair`             — input `{from: i64, to: i64, edge_type?}`
+//! - `TextPair`                — input `{from: str, to: str, edge_type?}`
+//! - `TextPairWithMetadata {…}` — same text keys plus optional
+//!   `weight: f64` input; `edge_id` / `created_at` are engine-managed
+//!   and NEVER taken from the caller.
 
 use crate::error::VerbError;
 use crate::schema::{EdgeKeyKind, EntitySchema};
-use rusqlite::Connection;
+use rusqlite::{types::Value as SqlValue, Connection};
 use serde_json::{json, Value};
 
 pub fn run(
@@ -38,6 +41,19 @@ pub fn run(
     match schema.edge_key_kind {
         EdgeKeyKind::IntegerPair => insert_integer(conn, edge, &input, &edge_type),
         EdgeKeyKind::TextPair => insert_text(conn, edge, &input, &edge_type),
+        EdgeKeyKind::TextPairWithMetadata {
+            has_id,
+            has_weight,
+            has_created_at,
+        } => insert_text_meta(
+            conn,
+            edge,
+            &input,
+            &edge_type,
+            has_id,
+            has_weight,
+            has_created_at,
+        ),
     }
 }
 
@@ -70,14 +86,7 @@ fn insert_text(
     input: &Value,
     edge_type: &str,
 ) -> Result<Value, VerbError> {
-    let from = input
-        .get("from")
-        .and_then(|v| v.as_str())
-        .ok_or_else(|| VerbError::InvalidInput("link: missing `from` string".into()))?;
-    let to = input
-        .get("to")
-        .and_then(|v| v.as_str())
-        .ok_or_else(|| VerbError::InvalidInput("link: missing `to` string".into()))?;
+    let (from, to) = extract_text_pair(input)?;
     conn.execute(
         &format!(
             "INSERT OR IGNORE INTO {edge} (src_path, dst_path, edge_type) VALUES (?1, ?2, ?3)"
@@ -86,3 +95,53 @@ fn insert_text(
     )?;
     Ok(json!({ "ok": true }))
 }
+
+fn insert_text_meta(
+    conn: &Connection,
+    edge: &str,
+    input: &Value,
+    edge_type: &str,
+    _has_id: bool,
+    has_weight: bool,
+    has_created_at: bool,
+) -> Result<Value, VerbError> {
+    let (from, to) = extract_text_pair(input)?;
+    let mut cols: Vec<&str> = vec!["src_path", "dst_path", "edge_type"];
+    let mut values: Vec<SqlValue> = vec![
+        SqlValue::Text(from),
+        SqlValue::Text(to),
+        SqlValue::Text(edge_type.to_string()),
+    ];
+    if has_weight {
+        let weight = input.get("weight").and_then(|v| v.as_f64()).unwrap_or(1.0);
+        cols.push("weight");
+        values.push(SqlValue::Real(weight));
+    }
+    if has_created_at {
+        let now = chrono::Utc::now().timestamp();
+        cols.push("created_at");
+        values.push(SqlValue::Integer(now));
+    }
+    let placeholders: Vec<String> = (1..=cols.len()).map(|i| format!("?{i}")).collect();
+    let sql = format!(
+        "INSERT OR IGNORE INTO {edge} ({}) VALUES ({})",
+        cols.join(","),
+        placeholders.join(",")
+    );
+    let params: Vec<&dyn rusqlite::ToSql> =
+        values.iter().map(|v| v as &dyn rusqlite::ToSql).collect();
+    conn.execute(&sql, params.as_slice())?;
+    Ok(json!({ "ok": true }))
+}
+
+fn extract_text_pair(input: &Value) -> Result<(String, String), VerbError> {
+    let from = input
+        .get("from")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| VerbError::InvalidInput("link: missing `from` string".into()))?;
+    let to = input
+        .get("to")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| VerbError::InvalidInput("link: missing `to` string".into()))?;
+    Ok((from.to_string(), to.to_string()))
+}
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/list.rs b/_primitives/_rust/kei-entity-store/src/verbs/list.rs
index f99f983..b14ce27 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/list.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/list.rs
@@ -1,4 +1,4 @@
-//! `list` verb — paginated SELECT, ordered by id DESC.
+//! `list` verb — paginated SELECT, ordered by pk DESC.
 //!
 //! Input: `{ "limit": <int = 50>, "offset": <int = 0> }`. Both optional.
 
@@ -27,9 +27,10 @@ pub fn run(
 
     let cols: Vec<&str> = schema.fields.iter().map(|f| f.name).collect();
     let sql = format!(
-        "SELECT {} FROM {} ORDER BY id DESC LIMIT ?1 OFFSET ?2",
+        "SELECT {} FROM {} ORDER BY {} DESC LIMIT ?1 OFFSET ?2",
         cols.join(","),
-        schema.table
+        schema.table,
+        schema.pk().name
     );
     let mut stmt = conn.prepare(&sql)?;
     let mut rows = stmt.query(rusqlite::params![limit, offset])?;
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/mod.rs b/_primitives/_rust/kei-entity-store/src/verbs/mod.rs
index 0aa25f5..73b76ed 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/mod.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/mod.rs
@@ -11,10 +11,12 @@
 
 pub mod archive;
 pub mod create;
+pub mod create_defaults;
 pub mod delete;
 pub mod get;
 pub mod link;
 pub mod list;
+pub mod pk;
 pub mod rank;
 pub mod search;
 pub mod update;
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/pk.rs b/_primitives/_rust/kei-entity-store/src/verbs/pk.rs
new file mode 100644
index 0000000..bcdee93
--- /dev/null
+++ b/_primitives/_rust/kei-entity-store/src/verbs/pk.rs
@@ -0,0 +1,71 @@
+//! Shared primary-key extraction helper — bridges IntegerPk / TextPk
+//! schemas so each verb can accept `{"id": <int>}` or `{"id": "<str>"}`
+//! without duplicating the dispatch logic.
+
+use crate::error::VerbError;
+use crate::schema::{EntitySchema, FieldKind};
+use rusqlite::types::Value as SqlValue;
+use serde_json::Value;
+
+/// A primary-key value bound for SQLite. Text PKs carry the raw string;
+/// integer PKs carry an i64.
+#[derive(Debug, Clone)]
+pub enum PkValue {
+    Integer(i64),
+    Text(String),
+}
+
+impl PkValue {
+    pub fn as_sql(&self) -> SqlValue {
+        match self {
+            PkValue::Integer(n) => SqlValue::Integer(*n),
+            PkValue::Text(s) => SqlValue::Text(s.clone()),
+        }
+    }
+
+    pub fn as_json(&self) -> Value {
+        match self {
+            PkValue::Integer(n) => Value::from(*n),
+            PkValue::Text(s) => Value::from(s.clone()),
+        }
+    }
+
+    /// String form — used to render `NotFound` errors uniformly.
+    pub fn as_string(&self) -> String {
+        match self {
+            PkValue::Integer(n) => n.to_string(),
+            PkValue::Text(s) => s.clone(),
+        }
+    }
+}
+
+/// Extract the primary-key value from a verb input JSON object. `verb`
+/// appears in the error message; caller passes its own name.
+pub fn extract(
+    schema: &EntitySchema,
+    input: &Value,
+    verb: &str,
+) -> Result<PkValue, VerbError> {
+    let raw = input.get("id").ok_or_else(|| {
+        VerbError::InvalidInput(format!("{verb}: missing `id`"))
+    })?;
+    match schema.pk().kind {
+        FieldKind::IntegerPk => raw
+            .as_i64()
+            .map(PkValue::Integer)
+            .ok_or_else(|| VerbError::InvalidInput(format!("{verb}: `id` must be integer"))),
+        FieldKind::TextPk => raw
+            .as_str()
+            .map(|s| PkValue::Text(s.to_string()))
+            .ok_or_else(|| VerbError::InvalidInput(format!("{verb}: `id` must be string"))),
+        other => Err(VerbError::InvalidInput(format!(
+            "{verb}: schema `{}` PK kind {:?} is not a primary key",
+            schema.name, other
+        ))),
+    }
+}
+
+/// The PK column name.
+pub fn pk_name(schema: &EntitySchema) -> &'static str {
+    schema.pk().name
+}
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/rank.rs b/_primitives/_rust/kei-entity-store/src/verbs/rank.rs
index 371d68f..8a149b6 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/rank.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/rank.rs
@@ -3,8 +3,10 @@
 //! sorted by score descending.
 //!
 //! Dispatches on `schema.edge_key_kind`: `IntegerPair` emits
-//! `{id: i64, score: f64}` rows; `TextPair` emits `{id: String, score}`
-//! where `id` carries the text node key (src_path / dst_path).
+//! `{id: i64, score: f64}` rows; `TextPair` and `TextPairWithMetadata`
+//! emit `{id: String, score}`. For `TextPairWithMetadata` with
+//! `has_weight: true` the rank propagation is proportional to edge
+//! weight (weighted PageRank); otherwise each edge contributes equally.
 
 use crate::error::VerbError;
 use crate::schema::{EdgeKeyKind, EntitySchema};
@@ -35,7 +37,10 @@ pub fn run(
     })?;
     match schema.edge_key_kind {
         EdgeKeyKind::IntegerPair => rank_integer(conn, edge),
-        EdgeKeyKind::TextPair => rank_text(conn, edge),
+        EdgeKeyKind::TextPair => rank_text(conn, edge, false),
+        EdgeKeyKind::TextPairWithMetadata { has_weight, .. } => {
+            rank_text(conn, edge, has_weight)
+        }
     }
 }
 
@@ -49,8 +54,8 @@ fn rank_integer(conn: &Connection, edge: &str) -> Result<Value, VerbError> {
     Ok(json!({ "results": results }))
 }
 
-fn rank_text(conn: &Connection, edge: &str) -> Result<Value, VerbError> {
-    let (nodes, out_edges) = collect_text(conn, edge)?;
+fn rank_text(conn: &Connection, edge: &str, with_weight: bool) -> Result<Value, VerbError> {
+    let (nodes, out_edges) = collect_text(conn, edge, with_weight)?;
     let rank = pagerank(&nodes, &out_edges);
     let mut out: Vec<(String, f64)> = rank.into_iter().collect();
     out.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
@@ -62,17 +67,17 @@ fn rank_text(conn: &Connection, edge: &str) -> Result<Value, VerbError> {
 fn collect_integer(
     conn: &Connection,
     edge: &str,
-) -> Result<(Vec<i64>, HashMap<i64, Vec<i64>>), VerbError> {
+) -> Result<(Vec<i64>, HashMap<i64, Vec<(i64, f64)>>), VerbError> {
     let sql = format!("SELECT from_id, to_id FROM {edge}");
     let mut stmt = conn.prepare(&sql)?;
     let rows = stmt.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)))?;
     let mut nodes: HashSet<i64> = HashSet::new();
-    let mut out_edges: HashMap<i64, Vec<i64>> = HashMap::new();
+    let mut out_edges: HashMap<i64, Vec<(i64, f64)>> = HashMap::new();
     for row in rows {
         let (src, dst) = row?;
         nodes.insert(src);
         nodes.insert(dst);
-        out_edges.entry(src).or_default().push(dst);
+        out_edges.entry(src).or_default().push((dst, 1.0));
     }
     Ok((nodes.into_iter().collect(), out_edges))
 }
@@ -80,26 +85,36 @@ fn collect_integer(
 fn collect_text(
     conn: &Connection,
     edge: &str,
-) -> Result<(Vec<String>, HashMap<String, Vec<String>>), VerbError> {
-    let sql = format!("SELECT src_path, dst_path FROM {edge}");
+    with_weight: bool,
+) -> Result<(Vec<String>, HashMap<String, Vec<(String, f64)>>), VerbError> {
+    let sql = if with_weight {
+        format!("SELECT src_path, dst_path, weight FROM {edge}")
+    } else {
+        format!("SELECT src_path, dst_path FROM {edge}")
+    };
     let mut stmt = conn.prepare(&sql)?;
-    let rows =
-        stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
+    let rows = stmt.query_map([], |r| {
+        let src: String = r.get(0)?;
+        let dst: String = r.get(1)?;
+        let w: f64 = if with_weight { r.get(2)? } else { 1.0 };
+        Ok((src, dst, w))
+    })?;
     let mut nodes: HashSet<String> = HashSet::new();
-    let mut out_edges: HashMap<String, Vec<String>> = HashMap::new();
+    let mut out_edges: HashMap<String, Vec<(String, f64)>> = HashMap::new();
     for row in rows {
-        let (src, dst) = row?;
+        let (src, dst, w) = row?;
         nodes.insert(src.clone());
         nodes.insert(dst.clone());
-        out_edges.entry(src).or_default().push(dst);
+        out_edges.entry(src).or_default().push((dst, w));
     }
     Ok((nodes.into_iter().collect(), out_edges))
 }
 
-/// Generic PageRank — works on any hashable node type.
+/// Generic weighted PageRank — each edge entry is `(target, weight)`.
+/// Unit weights reduce exactly to vanilla PageRank.
 fn pagerank<K: Eq + Hash + Clone>(
     nodes: &[K],
-    out_edges: &HashMap<K, Vec<K>>,
+    out_edges: &HashMap<K, Vec<(K, f64)>>,
 ) -> HashMap<K, f64> {
     if nodes.is_empty() {
         return HashMap::new();
@@ -114,7 +129,7 @@ fn pagerank<K: Eq + Hash + Clone>(
 
 fn one_iteration<K: Eq + Hash + Clone>(
     nodes: &[K],
-    out_edges: &HashMap<K, Vec<K>>,
+    out_edges: &HashMap<K, Vec<(K, f64)>>,
     prev: &HashMap<K, f64>,
 ) -> HashMap<K, f64> {
     let n = nodes.len() as f64;
@@ -122,8 +137,11 @@ fn one_iteration<K: Eq + Hash + Clone>(
     let mut next: HashMap<K, f64> = nodes.iter().map(|k| (k.clone(), base)).collect();
     for (src, dsts) in out_edges {
         if dsts.is_empty() { continue; }
-        let share = DAMPING * prev.get(src).copied().unwrap_or(0.0) / dsts.len() as f64;
-        for dst in dsts {
+        let total_w: f64 = dsts.iter().map(|(_, w)| *w).sum();
+        if total_w <= 0.0 { continue; }
+        let src_rank = prev.get(src).copied().unwrap_or(0.0);
+        for (dst, w) in dsts {
+            let share = DAMPING * src_rank * (w / total_w);
             if let Some(slot) = next.get_mut(dst) {
                 *slot += share;
             }
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/update.rs b/_primitives/_rust/kei-entity-store/src/verbs/update.rs
index 71db6c2..e736247 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/update.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/update.rs
@@ -8,6 +8,7 @@
 
 use crate::error::VerbError;
 use crate::schema::{EntitySchema, FieldDef, FieldKind};
+use crate::verbs::pk::{self, PkValue};
 use crate::verbs::validate;
 use chrono::Utc;
 use rusqlite::{types::Value as SqlValue, Connection};
@@ -22,17 +23,14 @@ pub fn run(
     let obj = input
         .as_object()
         .ok_or_else(|| VerbError::InvalidInput("update: expected JSON object".into()))?;
-    let id = obj
-        .get("id")
-        .and_then(|v| v.as_i64())
-        .ok_or_else(|| VerbError::InvalidInput("update: missing `id` integer".into()))?;
+    let id = pk::extract(schema, &input, "update")?;
     let now = Utc::now().timestamp();
     let (set_cols, values) = build_set(schema, obj, now)?;
     if set_cols.is_empty() {
         return Err(VerbError::InvalidInput("update: no writable fields supplied".into()));
     }
-    update_tx(conn, schema, id, &set_cols, values, obj)?;
-    Ok(json!({ "ok": true, "id": id }))
+    update_tx(conn, schema, &id, &set_cols, values, obj)?;
+    Ok(json!({ "ok": true, "id": id.as_json() }))
 }
 
 fn guard_enabled(schema: &EntitySchema) -> Result<(), VerbError> {
@@ -48,7 +46,7 @@ fn guard_enabled(schema: &EntitySchema) -> Result<(), VerbError> {
 fn update_tx(
     conn: &Connection,
     schema: &EntitySchema,
-    id: i64,
+    id: &PkValue,
     set_cols: &[&'static str],
     values: Vec<SqlValue>,
     obj: &serde_json::Map<String, Value>,
@@ -56,7 +54,7 @@ fn update_tx(
     let tx = conn.unchecked_transaction()?;
     exec_update_tx(&tx, schema, id, set_cols, values)?;
     if let Some(cols) = schema.fts_columns {
-        reindex_fts(&tx, schema.table, cols, id, obj)?;
+        reindex_fts(&tx, schema, cols, id, obj)?;
     }
     tx.commit()?;
     Ok(())
@@ -65,7 +63,7 @@ fn update_tx(
 fn exec_update_tx(
     tx: &rusqlite::Transaction<'_>,
     schema: &EntitySchema,
-    id: i64,
+    id: &PkValue,
     set_cols: &[&'static str],
     values: Vec<SqlValue>,
 ) -> Result<(), VerbError> {
@@ -73,18 +71,19 @@ fn exec_update_tx(
         (1..=set_cols.len()).map(|i| format!("{} = ?{i}", set_cols[i - 1])).collect();
     let id_idx = set_cols.len() + 1;
     let sql = format!(
-        "UPDATE {} SET {} WHERE id=?{}",
+        "UPDATE {} SET {} WHERE {}=?{}",
         schema.table,
         placeholders.join(", "),
+        schema.pk().name,
         id_idx
     );
     let mut all: Vec<SqlValue> = values;
-    all.push(SqlValue::Integer(id));
+    all.push(id.as_sql());
     let params: Vec<&dyn rusqlite::ToSql> =
         all.iter().map(|v| v as &dyn rusqlite::ToSql).collect();
     let rows = tx.execute(&sql, params.as_slice())?;
     if rows == 0 {
-        return Err(VerbError::NotFound { entity: schema.name.into(), id });
+        return Err(VerbError::not_found_text(schema.name, id.as_string()));
     }
     Ok(())
 }
@@ -117,7 +116,7 @@ fn value_from_input(
     let Some(raw) = input.get(f.name) else {
         return Ok(None);
     };
-    if f.kind == FieldKind::IntegerPk {
+    if f.is_pk() {
         return Ok(None);
     }
     Ok(Some(validate::coerce(f, raw)?))
@@ -125,16 +124,16 @@ fn value_from_input(
 
 fn reindex_fts(
     tx: &rusqlite::Transaction<'_>,
-    table: &str,
+    schema: &EntitySchema,
     cols: &[&str],
-    id: i64,
+    id: &PkValue,
     input: &serde_json::Map<String, Value>,
 ) -> Result<(), VerbError> {
-    // Pull existing values, overlay supplied ones, re-insert.
-    let existing = read_existing_fts(tx, table, cols, id)?;
+    let table = schema.table;
+    let existing = read_existing_fts(tx, schema, cols, id)?;
     tx.execute(
         &format!("DELETE FROM fts_{table} WHERE {table}_id=?1"),
-        rusqlite::params![id],
+        rusqlite::params![id.as_sql()],
     )?;
     let placeholders: Vec<String> = (2..=(cols.len() + 1)).map(|i| format!("?{i}")).collect();
     let sql = format!(
@@ -150,12 +149,12 @@ fn reindex_fts(
 }
 
 fn fts_row_values(
-    id: i64,
+    id: &PkValue,
     cols: &[&str],
     input: &serde_json::Map<String, Value>,
     existing: &serde_json::Map<String, Value>,
 ) -> Vec<SqlValue> {
-    let mut values: Vec<SqlValue> = vec![SqlValue::Integer(id)];
+    let mut values: Vec<SqlValue> = vec![id.as_sql()];
     for c in cols {
         let val = input
             .get(*c)
@@ -170,14 +169,18 @@ fn fts_row_values(
 
 fn read_existing_fts(
     tx: &rusqlite::Transaction<'_>,
-    table: &str,
+    schema: &EntitySchema,
     cols: &[&str],
-    id: i64,
+    id: &PkValue,
 ) -> Result<serde_json::Map<String, Value>, VerbError> {
     let col_list = cols.join(",");
-    let sql = format!("SELECT {col_list} FROM {table} WHERE id=?1");
+    let sql = format!(
+        "SELECT {col_list} FROM {} WHERE {}=?1",
+        schema.table,
+        schema.pk().name
+    );
     let mut stmt = tx.prepare(&sql)?;
-    let mut rows = stmt.query(rusqlite::params![id])?;
+    let mut rows = stmt.query(rusqlite::params![id.as_sql()])?;
     let mut out = serde_json::Map::new();
     if let Some(r) = rows.next()? {
         for (i, c) in cols.iter().enumerate() {
diff --git a/_primitives/_rust/kei-entity-store/src/verbs/validate.rs b/_primitives/_rust/kei-entity-store/src/verbs/validate.rs
index b2830c6..9800d0a 100644
--- a/_primitives/_rust/kei-entity-store/src/verbs/validate.rs
+++ b/_primitives/_rust/kei-entity-store/src/verbs/validate.rs
@@ -1,7 +1,8 @@
 //! Shared input-type validator for create / update.
 //!
 //! Strict typed validation: integer fields require JSON numbers that
-//! fit i64; text fields require JSON strings. Wrong-type input returns
+//! fit i64; text fields require JSON strings; real fields require JSON
+//! numbers convertible to f64. Wrong-type input returns
 //! `VerbError::InvalidType` instead of silent coercion to `0` / `""`.
 //!
 //! TEXT size cap: any text value longer than `MAX_TEXT_BYTES` is
@@ -27,11 +28,16 @@ pub fn coerce(f: &FieldDef, raw: &Value) -> Result<SqlValue, VerbError> {
             "field `{}` is PK and cannot be set directly",
             f.name
         ))),
+        FieldKind::TextPk => coerce_text(f, raw),
         FieldKind::IntegerNotNull
         | FieldKind::Integer
         | FieldKind::TimestampCreated
         | FieldKind::TimestampUpdated => coerce_int(f, raw),
-        FieldKind::TextNotNull | FieldKind::Text | FieldKind::TextDefault => coerce_text(f, raw),
+        FieldKind::TextNotNull
+        | FieldKind::Text
+        | FieldKind::TextDefault
+        | FieldKind::TextArchiveEnum => coerce_text(f, raw),
+        FieldKind::Real | FieldKind::RealDefault => coerce_real(f, raw),
     }
 }
 
@@ -61,6 +67,17 @@ fn coerce_text(f: &FieldDef, raw: &Value) -> Result<SqlValue, VerbError> {
     Ok(SqlValue::Text(s))
 }
 
+fn coerce_real(f: &FieldDef, raw: &Value) -> Result<SqlValue, VerbError> {
+    match raw {
+        Value::Null => Ok(SqlValue::Real(f.real_default.unwrap_or(0.0))),
+        Value::Number(n) => n
+            .as_f64()
+            .map(SqlValue::Real)
+            .ok_or_else(|| type_err(f, "real (f64)", &format!("number {} out of range", n))),
+        other => Err(type_err(f, "real", kind_name(other))),
+    }
+}
+
 fn type_err(f: &FieldDef, expected: &str, got: &str) -> VerbError {
     VerbError::InvalidType {
         field: f.name.to_string(),
diff --git a/_primitives/_rust/kei-entity-store/tests/real_text_pk_smoke.rs b/_primitives/_rust/kei-entity-store/tests/real_text_pk_smoke.rs
new file mode 100644
index 0000000..1b71e78
--- /dev/null
+++ b/_primitives/_rust/kei-entity-store/tests/real_text_pk_smoke.rs
@@ -0,0 +1,264 @@
+//! Smoke tests for the four M1 / M4 / M5 engine improvements:
+//!
+//! 1. `FieldKind::TextPk` — TEXT primary key schemas with caller-
+//!    supplied UUID-style ids.
+//! 2. `FieldKind::Real` / `RealDefault` — REAL columns round-tripped as
+//!    f64 through create + get.
+//! 3. `FieldKind::TextArchiveEnum` — archive verb writes the archived
+//!    sentinel string on schemas that encode status as a TEXT enum.
+//! 4. `EdgeKeyKind::TextPairWithMetadata` — text-keyed edges with
+//!    optional weight / id / created_at columns, used by rank.
+
+use kei_entity_store::schema::{EdgeKeyKind, EntitySchema, FieldDef};
+use kei_entity_store::verbs::{archive, create, delete, get, link, rank, update};
+use kei_entity_store::Store;
+use serde_json::json;
+
+// ---------- 1. TextPk ----------
+
+static SESSION_FIELDS: &[FieldDef] = &[
+    FieldDef::text_pk("id"),
+    FieldDef::text_nn("title"),
+    FieldDef::created_at(),
+];
+
+static SESSION_SCHEMA: EntitySchema = EntitySchema {
+    name: "session",
+    table: "sessions",
+    fields: SESSION_FIELDS,
+    enabled_verbs: &["create", "get", "list", "update", "delete"],
+    fts_columns: None,
+    edge_table: None,
+    edge_key_kind: EdgeKeyKind::IntegerPair,
+    archived_field: None,
+    custom_migrations: &[],
+};
+
+#[test]
+fn text_pk_create_with_string_id_and_get_by_string_id() {
+    let s = Store::open_memory(&SESSION_SCHEMA).unwrap();
+    let uuid = "550e8400-e29b-41d4-a716-446655440000";
+    let out = create::run(
+        s.conn(),
+        &SESSION_SCHEMA,
+        json!({ "id": uuid, "title": "first session" }),
+    )
+    .unwrap();
+    assert_eq!(out["id"], uuid);
+    assert!(out["created_at"].as_i64().unwrap() > 0);
+
+    let got = get::run(s.conn(), &SESSION_SCHEMA, json!({ "id": uuid })).unwrap();
+    assert_eq!(got["id"], uuid);
+    assert_eq!(got["title"], "first session");
+}
+
+#[test]
+fn text_pk_update_and_delete_by_string_id() {
+    let s = Store::open_memory(&SESSION_SCHEMA).unwrap();
+    let uuid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
+    create::run(
+        s.conn(),
+        &SESSION_SCHEMA,
+        json!({ "id": uuid, "title": "orig" }),
+    )
+    .unwrap();
+    update::run(
+        s.conn(),
+        &SESSION_SCHEMA,
+        json!({ "id": uuid, "title": "updated" }),
+    )
+    .unwrap();
+    let got = get::run(s.conn(), &SESSION_SCHEMA, json!({ "id": uuid })).unwrap();
+    assert_eq!(got["title"], "updated");
+
+    delete::run(s.conn(), &SESSION_SCHEMA, json!({ "id": uuid })).unwrap();
+    let err = get::run(s.conn(), &SESSION_SCHEMA, json!({ "id": uuid })).unwrap_err();
+    assert_eq!(err.exit_code(), 2);
+}
+
+#[test]
+fn text_pk_create_rejects_missing_id() {
+    let s = Store::open_memory(&SESSION_SCHEMA).unwrap();
+    let err = create::run(s.conn(), &SESSION_SCHEMA, json!({ "title": "x" })).unwrap_err();
+    assert_eq!(err.exit_code(), 2);
+}
+
+// ---------- 2. Real + RealDefault ----------
+
+static COST_FIELDS: &[FieldDef] = &[
+    FieldDef::pk("id"),
+    FieldDef::text_nn("label"),
+    FieldDef::real("cost"),
+    FieldDef::real_default("multiplier", 1.5),
+    FieldDef::created_at(),
+];
+
+static COST_SCHEMA: EntitySchema = EntitySchema {
+    name: "cost_entry",
+    table: "cost_entries",
+    fields: COST_FIELDS,
+    enabled_verbs: &["create", "get"],
+    fts_columns: None,
+    edge_table: None,
+    edge_key_kind: EdgeKeyKind::IntegerPair,
+    archived_field: None,
+    custom_migrations: &[],
+};
+
+#[test]
+fn real_column_round_trips_f64_unchanged() {
+    let s = Store::open_memory(&COST_SCHEMA).unwrap();
+    let v = create::run(
+        s.conn(),
+        &COST_SCHEMA,
+        json!({ "label": "gpt-4", "cost": 0.03125 }),
+    )
+    .unwrap();
+    let id = v["id"].as_i64().unwrap();
+    let row = get::run(s.conn(), &COST_SCHEMA, json!({ "id": id })).unwrap();
+    assert_eq!(row["cost"].as_f64().unwrap(), 0.03125);
+}
+
+#[test]
+fn real_default_applies_when_missing() {
+    let s = Store::open_memory(&COST_SCHEMA).unwrap();
+    let v = create::run(
+        s.conn(),
+        &COST_SCHEMA,
+        json!({ "label": "claude", "cost": 0.01 }),
+    )
+    .unwrap();
+    let id = v["id"].as_i64().unwrap();
+    let row = get::run(s.conn(), &COST_SCHEMA, json!({ "id": id })).unwrap();
+    assert_eq!(row["multiplier"].as_f64().unwrap(), 1.5);
+}
+
+// ---------- 3. TextArchiveEnum ----------
+
+static CHAT_FIELDS: &[FieldDef] = &[
+    FieldDef::text_pk("id"),
+    FieldDef::text_nn("project"),
+    FieldDef::text_archive_enum("status", "active", "archived"),
+    FieldDef::integer("status_at"),
+    FieldDef::created_at(),
+];
+
+static CHAT_SCHEMA: EntitySchema = EntitySchema {
+    name: "chat_session",
+    table: "chat_sessions",
+    fields: CHAT_FIELDS,
+    enabled_verbs: &["create", "get", "archive"],
+    fts_columns: None,
+    edge_table: None,
+    edge_key_kind: EdgeKeyKind::IntegerPair,
+    archived_field: Some("status"),
+    custom_migrations: &[],
+};
+
+#[test]
+fn archive_textenum_writes_archived_sentinel_string() {
+    let s = Store::open_memory(&CHAT_SCHEMA).unwrap();
+    let uuid = "aaaa-bbbb-cccc";
+    create::run(
+        s.conn(),
+        &CHAT_SCHEMA,
+        json!({ "id": uuid, "project": "test" }),
+    )
+    .unwrap();
+
+    // Before archive: status defaults to "active" sentinel.
+    let before = get::run(s.conn(), &CHAT_SCHEMA, json!({ "id": uuid })).unwrap();
+    assert_eq!(before["status"], "active");
+
+    let out = archive::run(s.conn(), &CHAT_SCHEMA, json!({ "id": uuid })).unwrap();
+    assert_eq!(out["id"], uuid);
+    let stamped = out["archived_at"].as_i64().unwrap();
+    assert!(stamped > 0);
+
+    let after = get::run(s.conn(), &CHAT_SCHEMA, json!({ "id": uuid })).unwrap();
+    assert_eq!(after["status"], "archived");
+    assert_eq!(after["status_at"].as_i64().unwrap(), stamped);
+}
+
+// ---------- 4. TextPairWithMetadata ----------
+
+static NODE_FIELDS: &[FieldDef] = &[
+    FieldDef::pk("id"),
+    FieldDef::text_nn("path"),
+    FieldDef::created_at(),
+];
+
+static META_EDGE_SCHEMA: EntitySchema = EntitySchema {
+    name: "doc",
+    table: "docs_meta",
+    fields: NODE_FIELDS,
+    enabled_verbs: &["link", "rank"],
+    fts_columns: None,
+    edge_table: Some("doc_edges_meta"),
+    edge_key_kind: EdgeKeyKind::TextPairWithMetadata {
+        has_id: true,
+        has_weight: true,
+        has_created_at: true,
+    },
+    archived_field: None,
+    custom_migrations: &[],
+};
+
+#[test]
+fn text_pair_metadata_link_stores_weight_and_timestamp() {
+    let s = Store::open_memory(&META_EDGE_SCHEMA).unwrap();
+    link::run(
+        s.conn(),
+        &META_EDGE_SCHEMA,
+        json!({ "from": "a.md", "to": "b.md", "weight": 3.5 }),
+    )
+    .unwrap();
+    let (weight, created_at): (f64, i64) = s
+        .conn()
+        .query_row(
+            "SELECT weight, created_at FROM doc_edges_meta \
+             WHERE src_path='a.md' AND dst_path='b.md'",
+            [],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .unwrap();
+    assert_eq!(weight, 3.5);
+    assert!(created_at > 0);
+}
+
+#[test]
+fn text_pair_metadata_rank_respects_weight() {
+    // Graph: a → b (weight 10), a → c (weight 1). b and c both sink.
+    // Weighted PageRank should push `b` above `c`; unweighted would
+    // split flow 50/50 (→ tie or identical rank).
+    let s = Store::open_memory(&META_EDGE_SCHEMA).unwrap();
+    link::run(
+        s.conn(),
+        &META_EDGE_SCHEMA,
+        json!({ "from": "a.md", "to": "b.md", "weight": 10.0 }),
+    )
+    .unwrap();
+    link::run(
+        s.conn(),
+        &META_EDGE_SCHEMA,
+        json!({ "from": "a.md", "to": "c.md", "weight": 1.0 }),
+    )
+    .unwrap();
+
+    let v = rank::run(s.conn(), &META_EDGE_SCHEMA, json!({})).unwrap();
+    let results = v["results"].as_array().unwrap();
+    let score_of = |id: &str| -> f64 {
+        results
+            .iter()
+            .find(|r| r["id"].as_str().unwrap() == id)
+            .unwrap()["score"]
+            .as_f64()
+            .unwrap()
+    };
+    let b = score_of("b.md");
+    let c = score_of("c.md");
+    assert!(
+        b > c * 1.5,
+        "expected weighted rank b ({b}) to exceed c ({c}) by > 1.5x"
+    );
+}