fix: trivial fix for pro function (#201)

## What type of PR is this?

- [ ] feat (new feature)
- [ ] fix (bug fix)
- [ ] docs (documentation)
- [ ] style (formatting, no code change)
- [x] refactor (code change that neither fixes a bug nor adds a feature)
- [ ] perf (performance improvement)
- [ ] test (adding or updating tests)
- [ ] chore (maintenance, tooling)
- [ ] build / ci (build or CI changes)

## Which issue(s) this PR fixes

Fixes # matrixorigin/memoria-website#98

## What this PR does / why we need it


### Scope

These edits extend Memoria’s storage/API/MCP layers for **group
collaboration**, **branch metadata**, **list filtering**, **MatrixOne
compatibility**, and **operational observability**. They also align
tests with updated function signatures.

---

### 1. `memoria-storage` (`store.rs` + tests)

**Schema / migrations**

- **`CURRENT_USER_SCHEMA_VERSION`**: bumped from `1` → **`2`** so
existing user databases run compat migrations on next startup.
- **`author_id` on `mem_memories`**: migration is **idempotent** — adds
column only if missing (information_schema check); **`AFTER user_id`
removed** for broader MatrixOne compatibility.
- **`idx_author`**: added only if missing (handles “column exists but
index missing”).
- **Branch tables** (`mem_branches.table_name` → physical branch
tables): **same `author_id` + `idx_author` migration** applied per
active branch table. Rationale: upstream group mode added `author_id` to
inserts; **legacy branch tables** created before that column may not
have it; zero-copy branches don’t automatically pick up a later `ALTER`
on `mem_memories`.

**Behavior**

- **`active_table`**: when `mem_branches` has metadata for the active
branch, **trust it** and **skip** `information_schema.tables` existence
checks — MatrixOne zero-copy branch tables may not appear reliably
there; failures surface on actual DML instead.
- **`list_branches`**: return type extended to **`(name, table_name,
created_at)`**; query adds **`ORDER BY created_at ASC`** for stable
ordering.
- **`list_active_lite`**: new parameter **`trust_tier: Option<&str>`**
with SQL filtering when set.

**Tests**

- `branch_ops.rs`: destructuring updated for 3-tuple `list_branches`.
- `store_crud.rs`: `list_active_lite(..., None)` extended with extra
**`None`** for `trust_tier`.

---

### 2. `memoria-service` (`service.rs`)

- **`list_active_filtered`**: accepts **`trust_tier: Option<&str>`** and
filters in-memory when provided (consistent with REST/MCP).

---

### 3. `memoria-api`

**`routes/memory.rs`**

- **`ListQuery`**: optional **`trust_tier`** query param wired through
to **`list_active_filtered`**.

**`routes/snapshots.rs`**

- **`list_branches` JSON**: includes **`created_at`** where available
(tuple destructuring updated for new storage return type).

**`routes/admin.rs`**

- **`user_stats`**: counting memories for **group scopes** (`grp_*`)
adjusted so counts are not incorrectly filtered by personal `user_id`
only.
- **New handler `user_branch_stats`**: **`GET
/admin/users/:user_id/branch-stats`** — returns per-branch memory counts
(uses main count + iterates branches via `list_branches`), with
defensive error handling per branch.

**`lib.rs`**

- Registers the new admin route above.

---

### 4. `memoria-mcp`

**`server.rs`**

- **`ACTOR_USER_ID`** (task-local): captured before **`spawn_blocking`**
and re-entered inside the blocking closure so **active branch context
matches** between async REST/MCP paths and blocking Git/storage work
(fixes Dashboard vs MCP branch mismatch when using the same API key).

**`tools.rs`**

- **`memory_list` / list path**: passes **`trust_tier`** from tool args
into **`list_active_filtered`**.

**`git_tools.rs`**

- All **`list_branches`** call sites updated for **3-tuple** return
(third element ignored where unused).

---

### 5. `memoria-git` (`service.rs` + `Cargo.toml`)

**MatrixOne DDL reliability**

- **`exec_ddl`**: retries up to **3 attempts** with backoff when the DB
returns **20631** / **`txn need retry`** (“def changed” under RC mode).
Typical trigger: concurrent **`data branch merge`** vs **`ALTER TABLE`**
on branch tables during migration (transient; retry is the intended
client-side mitigation).

**Diff / compatibility**

- **`data branch diff`**: **`columns (...)` clause removed** from SQL —
filtered in Rust instead — for **MatrixOne version differences** (local
vs cloud).
- **`FIXME(memoria-team)`** comment documenting **duplicate `memory_id`
rows** in diff results / UI linkage — left for upstream; not “fixed”
here.

**Dependencies**

- **`tokio`** moved from **dev-dependencies only** to **normal
dependency** (needed for **`tokio::time::sleep`** in `exec_ddl` retry
loop).

---

### 6. Integration tests (`memoria-mcp`)

- **`branch_e2e.rs`**, **`integration_full.rs`**: tuple destructuring
updated for **`list_branches`** 3-tuple.

---

### Risks / review focus

1. **Migration + DDL concurrency**: `author_id` ALTERs on branch tables
can overlap with **`data branch merge`** — mitigated by **`exec_ddl`
retry**; reviewers may want to confirm whether long-term
**serialization** or **migration ordering** is preferable.
2. **`CURRENT_USER_SCHEMA_VERSION = 2`**: ensures one migration pass;
coordinate with upstream versioning if they already use `2` differently.
3. **Admin `user_branch_stats`**: new surface area — auth / master-key
expectations should match existing admin routes.
4. **Known open issue**: **`FIXME`** in `memoria-git` for duplicate diff
rows — document for Memoria team.

---

### How to verify locally

- **`cargo check`** / **`cargo test`** with a running MatrixOne and
**`DATABASE_URL`** pointing at the correct port (e.g. **`6001`** if
that’s where MO listens). Note: **`group_collab_api`** defaults to port
**`6666`** if **`DATABASE_URL`** is unset — set env explicitly if tests
fail with DB timeouts.

---

### Files changed (from your status)

| Area | Files |
|------|--------|
| API | `memoria-api/src/lib.rs`, `routes/admin.rs`, `routes/memory.rs`,
`routes/snapshots.rs` |
| Git / DDL | `memoria-git/Cargo.toml`, `memoria-git/src/service.rs` |
| MCP | `memoria-mcp/src/git_tools.rs`, `server.rs`, `tools.rs`, tests
`branch_e2e.rs`, `integration_full.rs` |
| Service | `memoria-service/src/service.rs` |
| Storage | `memoria-storage/src/store.rs`, tests `branch_ops.rs`,
`store_crud.rs` |

---

You can paste this block into a PR description or share it as an
internal review brief. If you want a shorter “changelog” paragraph only,
say so and I’ll trim it.

Author: loveRhythm1990

memoria/crates/memoria-api/src/auth.rs

@@ -83,7 +83,7 @@ async fn cached_or_db_principal(token: &str, state: &AppState) -> Option<CachedA
     Some(principal)
 }
 
-async fn group_main_write_allowed_for_solo_owner(
+pub(crate) async fn group_main_write_allowed_for_solo_owner(
     state: &AppState,
     group_id: &str,
     user_id: &str,

memoria/crates/memoria-api/src/lib.rs

@@ -358,6 +358,10 @@ pub fn build_router(state: AppState) -> Router {
             "/admin/users/:user_id/stats",
             get(routes::admin::user_stats),
         )
+        .route(
+            "/admin/users/:user_id/branch-stats",
+            get(routes::admin::user_branch_stats),
+        )
         .route(
             "/admin/users/:user_id/call-stats",
             get(routes::admin::user_call_stats),

memoria/crates/memoria-api/src/routes/admin.rs

@@ -165,13 +165,25 @@ pub async fn user_stats(
     auth.require_master()?;
     let user_store = get_user_store(&state, &user_id).await?;
     let memories_table = user_store.t("mem_memories");
-    let memory_count = sqlx::query_scalar::<_, i64>(&format!(
-        "SELECT COUNT(*) FROM {memories_table} WHERE user_id = ? AND is_active > 0"
-    ))
-    .bind(&user_id)
-    .fetch_one(user_store.pool())
-    .await
-    .map_err(db_err)?;
+    // For group scopes (user_id starts with "grp_") count ALL active memories in the
+    // group database — the individual user_id filter would always return 0 since
+    // memories are attributed to each member's uid, not the group id itself.
+    let memory_count = if user_id.starts_with("grp_") {
+        sqlx::query_scalar::<_, i64>(&format!(
+            "SELECT COUNT(*) FROM {memories_table} WHERE is_active > 0"
+        ))
+        .fetch_one(user_store.pool())
+        .await
+        .map_err(db_err)?
+    } else {
+        sqlx::query_scalar::<_, i64>(&format!(
+            "SELECT COUNT(*) FROM {memories_table} WHERE user_id = ? AND is_active > 0"
+        ))
+        .bind(&user_id)
+        .fetch_one(user_store.pool())
+        .await
+        .map_err(db_err)?
+    };
     let snapshot_count = user_store
         .list_snapshot_registrations(&user_id)
         .await
@@ -788,6 +800,108 @@ pub async fn user_call_stats(
     })))
 }
 
+/// GET /admin/users/:user_id/branch-stats
+///
+/// Returns memory counts per branch for a given user (or group).
+/// Response:
+/// ```json
+/// {
+///   "branches": [
+///     {"name": "main",    "count": 170},
+///     {"name": "another", "count": 320}
+///   ],
+///   "total": 490
+/// }
+/// ```
+pub async fn user_branch_stats(
+    auth: AuthUser,
+    State(state): State<AppState>,
+    Path(user_id): Path<String>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    auth.require_master()?;
+    let user_store = get_user_store(&state, &user_id).await?;
+
+    // --- main branch ---
+    // Personal users have rows from multiple users in mem_memories (shared table), so
+    // we filter by user_id. Group users own an isolated per-group database where
+    // mem_memories contains only group data, so no user_id filter is needed there.
+    let memories_table = user_store.t("mem_memories");
+    let main_count: i64 = if user_id.starts_with("grp_") {
+        sqlx::query_scalar::<_, i64>(&format!(
+            "SELECT COUNT(*) FROM {memories_table} WHERE is_active > 0"
+        ))
+        .fetch_one(user_store.pool())
+        .await
+        .map_err(db_err)?
+    } else {
+        sqlx::query_scalar::<_, i64>(&format!(
+            "SELECT COUNT(*) FROM {memories_table} WHERE user_id = ? AND is_active > 0"
+        ))
+        .bind(&user_id)
+        .fetch_one(user_store.pool())
+        .await
+        .map_err(db_err)?
+    };
+
+    let mut branches_stats = vec![serde_json::json!({"name": "main", "count": main_count})];
+    let mut total = main_count;
+    let mut degraded_branches: Vec<String> = Vec::new();
+
+    // --- non-main branches from mem_branches ---
+    // Branch tables are per-user physical tables (created via `data branch create table`),
+    // so they only contain data belonging to the owning user/group. No user_id filter is
+    // required — this is consistent with the group-scoped main count above.
+    let branch_rows = user_store.list_branches(&user_id).await.map_err(api_err)?;
+    for (name, raw_table_name, _created_at) in &branch_rows {
+        if name == "main" || raw_table_name.is_empty() {
+            continue;
+        }
+        if !raw_table_name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+            tracing::warn!(
+                user_id = %user_id,
+                branch = %name,
+                table = %raw_table_name,
+                "user_branch_stats: skipping branch with invalid table identifier"
+            );
+            degraded_branches.push(name.clone());
+            continue;
+        }
+        let bt = user_store.t(raw_table_name);
+        let count_result = sqlx::query_scalar::<_, i64>(&format!(
+            "SELECT COUNT(*) FROM {bt} WHERE is_active > 0"
+        ))
+        .fetch_one(user_store.pool())
+        .await;
+        let count = match count_result {
+            Ok(n) => n,
+            Err(e) => {
+                tracing::warn!(
+                    user_id = %user_id,
+                    branch = %name,
+                    table = %raw_table_name,
+                    error = %e,
+                    "user_branch_stats: failed to count memories for branch"
+                );
+                // Record degraded branch so the caller knows the total is unreliable.
+                degraded_branches.push(name.clone());
+                0
+            }
+        };
+        branches_stats.push(serde_json::json!({"name": name, "count": count}));
+        total += count;
+    }
+
+    Ok(Json(serde_json::json!({
+        "branches": branches_stats,
+        "total": total,
+        // `degraded` is true when at least one branch table could not be counted.
+        // In that case `total` is a lower bound and `degraded_branches` lists the
+        // affected branches — callers should treat the numbers as approximate.
+        "degraded": !degraded_branches.is_empty(),
+        "degraded_branches": degraded_branches,
+    })))
+}
+
 /// Redact password from database URL for safe display.
 fn redact_url(url: &str) -> String {
     // mysql://user:pass@host:port/db → mysql://user:***@host:port/db

memoria/crates/memoria-api/src/routes/mcp.rs

@@ -76,6 +76,25 @@ fn tracking_path(method: &str, params: Option<&serde_json::Value>) -> String {
     format!("/mcp/{sanitized}")
 }
 
+/// MCP write tools that mutate the active memory table.
+/// These are blocked when the caller is group-scoped and checked out on `main`,
+/// mirroring the REST `group_main_write_guard` middleware.
+///
+/// Read-only tools, branch-management tools (`memory_checkout`, `memory_branch`,
+/// `memory_branch_delete`), and snapshot ops are intentionally excluded.
+const GROUP_MAIN_WRITE_BLOCKED: &[&str] = &[
+    "memory_store",
+    "memory_correct",
+    "memory_purge",
+    "memory_observe",
+    "memory_governance",
+    "memory_consolidate",
+    "memory_reflect",
+    "memory_extract_entities",
+    "memory_link_entities",
+    "memory_rollback",
+];
+
 fn mcp_tool_dirty_mask(tool: &str) -> Option<crate::metrics_summary::DirtyMask> {
     use crate::metrics_summary::DirtyMask;
     match tool {
@@ -226,9 +245,75 @@ pub async fn mcp_handler(
         }
     };
 
+    // ── Group main-write guard (computed once, shared by both code paths) ─────
+    // Resolved here — before the Notification early-return — so that
+    // Notification-form write calls to `main` are also blocked instead of
+    // silently bypassing the restriction.
+    //
+    // Returns Some(tool_name) when the call must be blocked, None otherwise.
+    let blocked_tool: Option<String> = if let Some(gid) = &auth.group_id {
+        if let Some(tool) = tracked_tool.as_deref() {
+            if GROUP_MAIN_WRITE_BLOCKED.contains(&tool) {
+                // ACTOR_USER_ID is already set by the global actor_scope_layer.
+                let maybe_blocked = state
+                    .service
+                    .user_sql_store(gid)
+                    .await
+                    .ok()
+                    .map(|sql| {
+                        let gid_owned = gid.clone();
+                        memoria_storage::ACTOR_USER_ID
+                            .scope(auth.user_id.clone(), async move {
+                                sql.active_branch_name(&gid_owned).await
+                            })
+                    });
+                if let Some(fut) = maybe_blocked {
+                    if let Ok(branch) = fut.await {
+                        let is_solo_owner =
+                            crate::auth::group_main_write_allowed_for_solo_owner(
+                                &state,
+                                gid,
+                                &auth.user_id,
+                            )
+                            .await;
+                        if branch == "main" && !is_solo_owner {
+                            Some(tool.to_string())
+                        } else {
+                            None
+                        }
+                    } else {
+                        None
+                    }
+                } else {
+                    None
+                }
+            } else {
+                None
+            }
+        } else {
+            None
+        }
+    } else {
+        None
+    };
+
     // JSON-RPC 2.0: a Notification is a *valid* Request without an "id" member.
     // The server MUST NOT reply to Notifications.
     if req.get("id").is_none() {
+        // Write guard: per JSON-RPC 2.0 the server MUST NOT reply to Notifications,
+        // so we silently drop blocked writes without dispatching.
+        if blocked_tool.is_some() {
+            report_stats(&track_path, false);
+            state.call_log_batcher.record_rpc(
+                user_id,
+                "POST".to_string(),
+                track_path,
+                204,
+                t.elapsed().as_millis() as u32,
+                RpcMeta::err(-32001),
+            );
+            return StatusCode::NO_CONTENT.into_response();
+        }
         let dispatch_result = memoria_mcp::dispatch_http(
             method.clone(),
             params,
@@ -262,6 +347,36 @@ pub async fn mcp_handler(
 
     let id = req["id"].clone();
 
+    // Use the pre-computed write-guard decision (see above).
+    if let Some(tool) = &blocked_tool {
+        let err_body = Json(json!({
+            "jsonrpc": "2.0",
+            "id": id,
+            "error": {
+                "code": -32001,
+                "message": format!(
+                    "Cannot call '{}' on 'main' in a shared space — \
+                     main is read-only in collaboration mode. \
+                     Your active branch is currently 'main'. \
+                     Call memory_checkout with a branch name \
+                     (e.g. {{\"name\": \"my-branch\"}}) to switch \
+                     to your own branch, then retry.",
+                    tool
+                )
+            }
+        }));
+        report_stats(&track_path, false);
+        state.call_log_batcher.record_rpc(
+            user_id,
+            "POST".to_string(),
+            track_path,
+            200,
+            t.elapsed().as_millis() as u32,
+            RpcMeta::err(-32001),
+        );
+        return err_body.into_response();
+    }
+
     // JSON-RPC spec: the HTTP response is always 200 OK, even for RPC errors.
     // Business-level error tracking uses rpc_success / rpc_error_code in the call log.
     let (response, rpc) = match memoria_mcp::dispatch_http(

memoria/crates/memoria-api/src/routes/memory.rs

@@ -67,6 +67,7 @@ async fn find_memory_any_user(
 pub struct ListQuery {
     pub memory_type: Option<String>,
     pub session_id: Option<String>,
+    pub trust_tier: Option<String>,
     #[serde(default = "default_limit")]
     pub limit: i64,
     pub cursor: Option<String>,
@@ -114,6 +115,9 @@ pub async fn list_memories(
         .cursor
         .as_deref()
         .filter(|c| c.len() == 32 && c.chars().all(|ch| ch.is_ascii_hexdigit()));
+    if let Some(tier) = q.trust_tier.as_deref() {
+        parse_trust_tier(tier).map_err(|e| (StatusCode::UNPROCESSABLE_ENTITY, e))?;
+    }
     let fetch_limit = limit + 1;
     let mut memories = state
         .service
@@ -122,6 +126,7 @@ pub async fn list_memories(
             fetch_limit,
             q.memory_type.as_deref(),
             q.session_id.as_deref(),
+            q.trust_tier.as_deref(),
             cursor,
         )
         .await

memoria/crates/memoria-api/src/routes/snapshots.rs

@@ -214,7 +214,7 @@ async fn branch_table_name_raw(
     if branch_name == "main" {
         return Ok("mem_memories".to_string());
     }
-    for (name, table_name) in sql.list_branches(scope_id).await.map_err(api_err)? {
+    for (name, table_name, _created_at) in sql.list_branches(scope_id).await.map_err(api_err)? {
         if name == branch_name {
             return Ok(table_name);
         }
@@ -241,7 +241,26 @@ async fn branch_diff_items_payload(
         .diff_branch_rows(&branch_table_raw, "mem_memories", scope_id, limit * 3)
         .await
         .map_err(api_err)?;
-    let classified = memoria_git::classify_diff_rows(raw_rows, &branch_table_raw);
+    let mut classified = memoria_git::classify_diff_rows(raw_rows, &branch_table_raw);
+
+    // classify_diff_rows cannot distinguish "deleted from main on branch" from
+    // "created-then-deleted on branch" without querying the main table (because
+    // data branch diff only returns the branch-side row for deletions). Resolve
+    // now: ghost removes (never in main) move to classified.ghost_removes.
+    git.resolve_ghost_removes(&mut classified, "mem_memories", scope_id)
+        .await
+        .map_err(api_err)?;
+
+    tracing::debug!(
+        branch = %branch_name,
+        added = classified.added.len(),
+        updated = classified.updated.len(),
+        removed = classified.removed.len(),
+        ghost_removes = classified.ghost_removes.len(),
+        conflicts = classified.conflicts.len(),
+        behind_main = classified.behind_main.len(),
+        "branch_diff_items_payload: classified result"
+    );
 
     let item_to_json = |item: &memoria_git::DiffItem| -> Value {
         json!({
@@ -749,10 +768,12 @@ pub async fn list_branches(
         "name": "main",
         "active": active_branch == "main",
     })];
-    for (name, _table_name) in sql.list_branches(auth.scope_id()).await.map_err(api_err)? {
+    for (name, _table_name, created_at) in sql.list_branches(auth.scope_id()).await.map_err(api_err)? {
+        let created_at_str = format_snapshot_timestamp(created_at);
         branches.push(json!({
             "name": name,
             "active": name == active_branch,
+            "created_at": created_at_str,
         }));
     }
     if !branches

memoria/crates/memoria-git/Cargo.toml

@@ -11,8 +11,8 @@ serde_json = { workspace = true }
 chrono = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
+tokio = { workspace = true }
 
 [dev-dependencies]
 memoria-storage = { path = "../memoria-storage" }
-tokio = { workspace = true }
 uuid = { workspace = true }