feat(account-management): add child_count to tenant read shape

docs/api/api.json

@@ -3667,6 +3667,12 @@
       "TenantDto": {
         "description": "AM-internal projection. Wider than `tenant_resolver_sdk::TenantInfo` for\nadmin/UI consumers (carries lifecycle timestamps + depth).\n\nInternal columns (raw `tenant_type_uuid`, retention/claim columns) never\ncross this boundary. `tenant_type: None` only on a transient registry\nblip — reads do not block on it.",
         "properties": {
+          "child_count": {
+            "description": "Number of **direct** children visible to the caller, resolved at\nread time: scope-filtered (children behind a self-managed barrier\nthe caller cannot penetrate are not counted), excludes\nAM-internal `provisioning` rows, includes soft-`deleted`\nchildren. `0` for a leaf. Derived field (not a stored column) —\nnot available in `$filter` / `$orderby`.",
+            "format": "int32",
+            "minimum": 0,
+            "type": "integer"
+          },
           "created_at": {
             "format": "date-time",
             "type": "string"
@@ -3721,6 +3727,7 @@
           "status",
           "self_managed",
           "depth",
+          "child_count",
           "created_at",
           "updated_at"
         ],

gears/system/account-management/account-management-sdk/src/tenant.rs

@@ -87,6 +87,16 @@ pub struct Tenant {
     /// Hierarchy depth recorded at insert. `0` for the root, `parent.depth + 1`
     /// for every child.
     pub depth: u32,
+    /// Number of **direct** children of this tenant visible to the
+    /// caller, resolved at read time. Scope-filtered (direct children
+    /// behind a self-managed barrier the caller cannot penetrate are
+    /// not counted), **excludes** AM-internal `Provisioning` rows, and
+    /// **includes** soft-`Deleted` children. `0` for a leaf. This is a
+    /// derived, read-only field — not a stored column — so it is
+    /// unavailable in `$filter` / `$orderby` over the `listChildren`
+    /// surface.
+    #[serde(default)]
+    pub child_count: u32,
     #[serde(with = "rfc3339")]
     pub created_at: OffsetDateTime,
     #[serde(with = "rfc3339")]

gears/system/account-management/account-management/src/api/rest/dto.rs

@@ -296,6 +296,13 @@ pub struct TenantDto {
     pub parent_id: Option<Uuid>,
     pub self_managed: bool,
     pub depth: u32,
+    /// Number of **direct** children visible to the caller, resolved at
+    /// read time: scope-filtered (children behind a self-managed barrier
+    /// the caller cannot penetrate are not counted), excludes
+    /// AM-internal `provisioning` rows, includes soft-`deleted`
+    /// children. `0` for a leaf. Derived field (not a stored column) —
+    /// not available in `$filter` / `$orderby`.
+    pub child_count: u32,
     #[serde(with = "time::serde::rfc3339")]
     pub created_at: OffsetDateTime,
     #[serde(with = "time::serde::rfc3339")]
@@ -323,6 +330,7 @@ impl TenantDto {
             parent_id: tenant.parent_id.map(|t| t.0),
             self_managed: tenant.self_managed,
             depth: tenant.depth,
+            child_count: tenant.child_count,
             created_at: tenant.created_at,
             updated_at: tenant.updated_at,
             deleted_at: tenant.deleted_at,

gears/system/account-management/account-management/src/api/rest/dto_tests.rs

@@ -399,6 +399,7 @@ fn sample_active_tenant() -> Tenant {
         parent_id: Some(sample_parent_id()),
         self_managed: false,
         depth: 2,
+        child_count: 3,
         created_at: sample_created(),
         updated_at: sample_updated_tenant(),
         deleted_at: None,
@@ -419,6 +420,7 @@ fn tenant_dto_active_wire_shape_mirrors_openapi() {
             "parent_id": "44444444-4444-4444-4444-444444444444",
             "self_managed": false,
             "depth": 2,
+            "child_count": 3,
             "created_at": "2026-05-01T09:30:00Z",
             "updated_at": "2026-05-16T12:00:00Z",
         }),

gears/system/account-management/account-management/src/api/rest/handlers/tenants_tests.rs

@@ -97,6 +97,7 @@ fn sample_active_tenant() -> Tenant {
         parent_id: Some(TenantId(sample_parent_id())),
         self_managed: false,
         depth: 2,
+        child_count: 0,
         created_at: sample_created(),
         updated_at: sample_updated(),
         deleted_at: None,

gears/system/account-management/account-management/src/domain/tenant/repo.rs

@@ -19,6 +19,7 @@
 //!   status) and rewrites `tenant_closure.descendant_status` atomically
 //!   when `status` changes.
 
+use std::collections::HashMap;
 use std::time::Duration;
 
 use async_trait::async_trait;
@@ -367,6 +368,21 @@ pub trait TenantRepo: Send + Sync {
         filter: ChildCountFilter,
     ) -> Result<u64, DomainError>;
 
+    /// Direct-child counts for a batch of parent ids, keyed by parent.
+    ///
+    /// Public-surface counterpart to [`count_children`]: **scope-filtered**
+    /// (direct children behind a self-managed barrier the caller cannot
+    /// reach count as `0`) and **excludes `Provisioning`** (no public
+    /// representation) while **including `Deleted`**. One grouped query
+    /// covers the whole batch; parents with no matching child are absent
+    /// from the map (callers default them to `0`). Powers the
+    /// `child_count` field on the public tenant read shape.
+    async fn count_children_grouped(
+        &self,
+        scope: &AccessScope,
+        parent_ids: &[Uuid],
+    ) -> Result<HashMap<Uuid, u64>, DomainError>;
+
     /// Count live tenants grouped by `(status, self_managed)` for the
     /// `am_tenants` inventory gauge. Visibility is bounded by `scope`
     /// (the periodic refresher passes `AccessScope::allow_all` for a

gears/system/account-management/account-management/src/domain/tenant/service/mod.rs

@@ -367,7 +367,11 @@ impl<R: TenantRepo> TenantService<R> {
     /// `TryFrom<TenantStatus>` so a bypass surfaces as
     /// [`DomainError::Internal`] (HTTP 500) instead of a process
     /// panic.
-    async fn lower_to_tenant(&self, model: TenantModel) -> Result<Tenant, DomainError> {
+    async fn lower_to_tenant(
+        &self,
+        scope: &AccessScope,
+        model: TenantModel,
+    ) -> Result<Tenant, DomainError> {
         let tenant_type = self.resolve_tenant_type(model.tenant_type_uuid).await;
         let status =
             account_management_sdk::TenantStatus::try_from(model.status).map_err(|_| {
@@ -380,6 +384,16 @@ impl<R: TenantRepo> TenantService<R> {
                     cause: None,
                 }
             })?;
+        // Derived public read-shape field: count of direct children
+        // visible to the caller — scope-filtered, excludes Provisioning,
+        // includes Deleted. One indexed grouped COUNT for this single id.
+        let child_count = self
+            .repo
+            .count_children_grouped(scope, &[model.id])
+            .await?
+            .get(&model.id)
+            .copied()
+            .map_or(0, |n| u32::try_from(n).unwrap_or(u32::MAX));
         Ok(Tenant {
             id: TenantId(model.id),
             name: model.name,
@@ -388,6 +402,7 @@ impl<R: TenantRepo> TenantService<R> {
             parent_id: model.parent_id.map(TenantId),
             self_managed: model.self_managed,
             depth: model.depth,
+            child_count,
             created_at: model.created_at,
             updated_at: model.updated_at,
             deleted_at: model.deleted_at,
@@ -399,6 +414,7 @@ impl<R: TenantRepo> TenantService<R> {
     /// scales with number of pages, not number of rows.
     async fn lower_to_tenant_page(
         &self,
+        scope: &AccessScope,
         page: Page<TenantModel>,
     ) -> Result<Page<Tenant>, DomainError> {
         let Page { items, page_info } = page;
@@ -423,6 +439,12 @@ impl<R: TenantRepo> TenantService<R> {
             }
         }
 
+        // One grouped COUNT covers the whole page's direct-child tallies
+        // (scope-filtered, excludes Provisioning, includes Deleted) — no
+        // N+1 across rows. Ids absent from the map default to 0.
+        let child_ids: Vec<Uuid> = items.iter().map(|m| m.id).collect();
+        let child_counts = self.repo.count_children_grouped(scope, &child_ids).await?;
+
         let mut mapped: Vec<Tenant> = Vec::with_capacity(items.len());
         for m in items {
             let status =
@@ -444,6 +466,10 @@ impl<R: TenantRepo> TenantService<R> {
                 parent_id: m.parent_id.map(TenantId),
                 self_managed: m.self_managed,
                 depth: m.depth,
+                child_count: child_counts
+                    .get(&m.id)
+                    .copied()
+                    .map_or(0, |n| u32::try_from(n).unwrap_or(u32::MAX)),
                 created_at: m.created_at,
                 updated_at: m.updated_at,
                 deleted_at: m.deleted_at,
@@ -1133,7 +1159,12 @@ impl<R: TenantRepo> TenantService<R> {
         // `is_sdk_visible` invariant is safe.
         // tenant_type forwarded verbatim — chain already validated
         // upstream, no need for a registry RTT.
-        let mut info = self.lower_to_tenant(activated).await?;
+        // A freshly created tenant is always a 0-child leaf; `child_count`
+        // resolves to 0 regardless of scope, so the saga's `allow_all`
+        // posture is fine here.
+        let mut info = self
+            .lower_to_tenant(&AccessScope::allow_all(), activated)
+            .await?;
         if info.tenant_type.is_none() {
             // `Tenant.tenant_type` is `Option<String>` per the
             // tenant-resolver-sdk public shape; lower the typed
@@ -1421,7 +1452,7 @@ impl<R: TenantRepo> TenantService<R> {
                 resource: id.to_string(),
             });
         }
-        self.lower_to_tenant(tenant).await
+        self.lower_to_tenant(&scope, tenant).await
     }
     // @cpt-end:cpt-cf-account-management-dod-tenant-hierarchy-management-tenant-read-scope:p1:inst-dod-read-scope-service
     // @cpt-end:cpt-cf-account-management-flow-tenant-hierarchy-management-read-tenant:p1:inst-flow-read-service
@@ -1487,7 +1518,11 @@ impl<R: TenantRepo> TenantService<R> {
         // through the repo filter before we lower the page to the
         // public shape (where `Provisioning` is unrepresentable).
         page.items.retain(|r| r.status.is_sdk_visible());
-        self.lower_to_tenant_page(page).await
+        // Pass the original (barrier-respecting) scope, not the
+        // `relaxed` one used for the listing query: a self-managed
+        // child's own subtree stays behind its barrier, so its
+        // `child_count` is scope-filtered to what the caller may see.
+        self.lower_to_tenant_page(&scope, page).await
     }
     // @cpt-end:cpt-cf-account-management-dod-tenant-hierarchy-management-children-query-paginated:p1:inst-dod-children-query-service
     // @cpt-end:cpt-cf-account-management-flow-tenant-hierarchy-management-list-children:p1:inst-flow-listch-service
@@ -1601,7 +1636,7 @@ impl<R: TenantRepo> TenantService<R> {
             );
         }
 
-        self.lower_to_tenant(updated).await
+        self.lower_to_tenant(&scope, updated).await
     }
     // @cpt-end:cpt-cf-account-management-dod-tenant-hierarchy-management-update-mutable-only:p1:inst-dod-update-mutable-service
     // @cpt-end:cpt-cf-account-management-flow-tenant-hierarchy-management-update-tenant:p1:inst-flow-update-service
@@ -1720,7 +1755,7 @@ impl<R: TenantRepo> TenantService<R> {
                 "am tenant state changed"
             );
         }
-        self.lower_to_tenant(updated).await
+        self.lower_to_tenant(&scope, updated).await
     }
 
     // -----------------------------------------------------------------
@@ -1796,7 +1831,7 @@ impl<R: TenantRepo> TenantService<R> {
         // downstream filters can distinguish first-write events from
         // idempotent retries.
         if matches!(tenant.status, TenantStatus::Deleted) {
-            return self.lower_to_tenant(tenant).await;
+            return self.lower_to_tenant(&scope, tenant).await;
         }
         // 1. Child-rejection guard. `include_deleted = false` excludes
         // ONLY rows in `Deleted` status; `Provisioning`, `Active` and
@@ -1855,7 +1890,7 @@ impl<R: TenantRepo> TenantService<R> {
             retention_secs = self.cfg.retention.default_window_secs,
             "am tenant state changed"
         );
-        self.lower_to_tenant(updated).await
+        self.lower_to_tenant(&scope, updated).await
     }
     // @cpt-end:cpt-cf-account-management-dod-tenant-hierarchy-management-data-lifecycle:p1:inst-dod-data-lifecycle-soft-delete
     // @cpt-end:cpt-cf-account-management-dod-tenant-hierarchy-management-soft-delete-preconditions:p1:inst-dod-soft-delete-preconditions

gears/system/account-management/account-management/src/domain/tenant/service/service_tests.rs

@@ -645,6 +645,49 @@ async fn list_children_status_filter_applies() {
     assert_eq!(suspended_only.items[0].id.0, c2);
 }
 
+#[tokio::test]
+async fn list_children_reports_direct_child_count_per_child() {
+    let root = Uuid::from_u128(0x100);
+    let repo = Arc::new(FakeTenantRepo::with_root(root));
+    let svc = make_service(repo.clone(), FakeOutcome::Ok);
+
+    // Two direct children of root: `c1` is a branch, `c2` a leaf.
+    let c1 = Uuid::from_u128(0x301);
+    let c2 = Uuid::from_u128(0x302);
+    svc.create_tenant(&ctx_for(root), child_input(c1, root))
+        .await
+        .expect("c1");
+    svc.create_tenant(&ctx_for(root), child_input(c2, root))
+        .await
+        .expect("c2");
+
+    // Two grandchildren under `c1`; `c2` stays a leaf.
+    svc.create_tenant(&ctx_for(root), child_input(Uuid::from_u128(0x401), c1))
+        .await
+        .expect("gc1");
+    svc.create_tenant(&ctx_for(root), child_input(Uuid::from_u128(0x402), c1))
+        .await
+        .expect("gc2");
+
+    let page = svc
+        .list_children(&ctx_for(root), root, &ODataQuery::default())
+        .await
+        .expect("list ok");
+
+    let by_id: std::collections::HashMap<Uuid, u32> =
+        page.items.iter().map(|t| (t.id.0, t.child_count)).collect();
+    assert_eq!(
+        by_id.get(&c1).copied(),
+        Some(2),
+        "c1 surfaces its two direct children"
+    );
+    assert_eq!(
+        by_id.get(&c2).copied(),
+        Some(0),
+        "c2 is a leaf - child_count defaults to 0"
+    );
+}
+
 #[tokio::test]
 async fn update_tenant_accepts_name_then_status_transitions_via_dedicated_methods() {
     let root = Uuid::from_u128(0x100);

gears/system/account-management/account-management/src/domain/tenant/test_support/repo.rs

@@ -1270,6 +1270,25 @@ impl TenantRepo for FakeTenantRepo {
         Ok(u64::try_from(count).unwrap_or(u64::MAX))
     }
 
+    async fn count_children_grouped(
+        &self,
+        _scope: &AccessScope,
+        parent_ids: &[Uuid],
+    ) -> Result<HashMap<Uuid, u64>, DomainError> {
+        let wanted: HashSet<Uuid> = parent_ids.iter().copied().collect();
+        let state = self.state.lock().expect("lock");
+        let mut out: HashMap<Uuid, u64> = HashMap::new();
+        for t in state.tenants.values() {
+            let Some(pid) = t.parent_id else { continue };
+            // Public-surface semantics: scope is ignored by this mock,
+            // `Provisioning` is excluded, `Deleted` is included.
+            if wanted.contains(&pid) && !matches!(t.status, TenantStatus::Provisioning) {
+                *out.entry(pid).or_insert(0) += 1;
+            }
+        }
+        Ok(out)
+    }
+
     async fn count_tenants_by_status(
         &self,
         _scope: &AccessScope,

gears/system/account-management/account-management/src/infra/storage/repo_impl/mod.rs

@@ -19,6 +19,7 @@ pub use conversion::ConversionRepoImpl;
 pub use hierarchy_read::TenantHierarchyReadAdapter;
 pub use metadata::MetadataRepoImpl;
 
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -201,6 +202,14 @@ impl TenantRepo for TenantRepoImpl {
         reads::count_children(self, scope, parent_id, filter).await
     }
 
+    async fn count_children_grouped(
+        &self,
+        scope: &AccessScope,
+        parent_ids: &[Uuid],
+    ) -> Result<HashMap<Uuid, u64>, DomainError> {
+        reads::count_children_grouped(self, scope, parent_ids).await
+    }
+
     async fn count_tenants_by_status(
         &self,
         scope: &AccessScope,