use std::collections::HashMap;
use std::sync::Arc;
use axum::{
extract::{Path, Query, State},
http::StatusCode,
Json,
};
use serde::Deserialize;
use serde_json::{json, Value};
use tracing::info;
use crate::auth::{AuthenticatedUser, ForwardedToken, InstanceContext, WorkspaceAdmins};
use crate::entity_config::{slug_to_cedar_type, EntityConfig};
use crate::pdt::{
AuthContext, CreateAssetRequest, CreateRelationRequest, CreateTagRequest, PdtAsset,
PdtSearchResult,
};
use crate::AppState;
fn extract_field_value(
asset: &PdtAsset,
field_name: &str,
field_map: &Option<String>,
config: &EntityConfig,
) -> Option<Value> {
match field_map.as_deref() {
Some("title_suffix") => {
let title = asset
.title
.strip_prefix(&config.title_prefix)
.unwrap_or(&asset.title);
if title.is_empty() {
None
} else {
Some(Value::String(title.to_string()))
}
}
Some("content") => {
if asset.content.is_empty() {
None
} else {
Some(Value::String(asset.content.clone()))
}
}
Some(m) if m.starts_with("tag:") => {
let tag_name = &m[4..];
asset
.get_tag(tag_name)
.map(|s| Value::String(s.to_string()))
}
Some(m) if m.starts_with("metadata:") => {
let key = &m[9..];
asset.metadata.get(key).cloned()
}
Some("created_at") => Some(Value::String(asset.created_at.clone())),
Some("updated_at") => Some(Value::String(asset.updated_at.clone())),
_ => {
asset
.get_tag(field_name)
.map(|s| Value::String(s.to_string()))
}
}
}
fn extract_field_value_compact(
result: &PdtSearchResult,
field_name: &str,
field_map: &Option<String>,
config: &EntityConfig,
) -> Option<Value> {
match field_map.as_deref() {
Some("title_suffix") => {
let title = result
.title
.strip_prefix(&config.title_prefix)
.unwrap_or(&result.title);
if title.is_empty() {
None
} else {
Some(Value::String(title.to_string()))
}
}
Some("content") => None, Some(m) if m.starts_with("tag:") => {
let tag_name = &m[4..];
result
.get_tag(tag_name)
.map(|s| Value::String(s.to_string()))
}
Some(m) if m.starts_with("metadata:") => None, Some("created_at") | Some("updated_at") => Some(Value::String(result.updated_at.clone())),
_ => result
.get_tag(field_name)
.map(|s| Value::String(s.to_string())),
}
}
async fn resolve_relations(
state: &AppState,
instance_id: Option<&str>,
asset_id: &str,
config: &EntityConfig,
entity: &mut serde_json::Map<String, Value>,
token: Option<&str>,
) {
let pdt = state.pdt.for_instance(instance_id);
for (field_name, field_config) in config.relation_fields() {
let rel_type = field_config
.relation_type
.as_deref()
.unwrap_or("related_to");
if let Ok(relations) = pdt.get_relations(asset_id, None, token).await {
let related_id = relations.iter().find_map(|r| {
if r.from_asset_id == asset_id && r.relation_type == rel_type {
Some(r.to_asset_id.clone())
} else {
None
}
});
if let Some(id) = related_id {
entity.insert(field_name.clone(), Value::String(id));
}
}
}
}
pub fn map_asset_to_entity(asset: &PdtAsset, config: &EntityConfig) -> Value {
let mut entity = serde_json::Map::new();
entity.insert("id".to_string(), Value::String(asset.id.clone()));
for (field_name, field_config) in &config.fields {
if let Some(v) = extract_field_value(asset, field_name, &field_config.map, config) {
entity.insert(field_name.clone(), v);
}
}
Value::Object(entity)
}
pub fn map_compact_to_entity(result: &PdtSearchResult, config: &EntityConfig) -> Value {
let mut entity = serde_json::Map::new();
entity.insert("id".to_string(), Value::String(result.id.clone()));
for (field_name, field_config) in &config.fields {
if let Some(v) = extract_field_value_compact(result, field_name, &field_config.map, config)
{
entity.insert(field_name.clone(), v);
}
}
Value::Object(entity)
}
pub async fn map_asset_to_entity_with_relations(
state: &AppState,
instance_id: Option<&str>,
asset: &PdtAsset,
config: &EntityConfig,
token: Option<&str>,
) -> Value {
let mut entity_json = match map_asset_to_entity(asset, config) {
Value::Object(m) => m,
_ => return Value::Null,
};
resolve_relations(
state,
instance_id,
&asset.id,
config,
&mut entity_json,
token,
)
.await;
Value::Object(entity_json)
}
#[derive(Debug, Deserialize, Clone)]
pub struct ListEntitiesQuery {
pub status: Option<String>,
pub attached_to: Option<String>,
}
fn get_entity_config(state: &AppState, slug: &str) -> Option<Arc<EntityConfig>> {
state
.entity_configs
.iter()
.find(|c| c.slug == slug || c.slug == format!("{}s", slug))
.cloned()
}
async fn list_entities_inner(
state: &AppState,
token: Option<&str>,
instance_id: Option<&str>,
slug: &str,
query: &ListEntitiesQuery,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let pdt = state.pdt.for_instance(instance_id);
let config = get_entity_config(state, &slug).ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(json!({"error": format!("Unknown entity type: {}", slug)})),
)
})?;
let type_tag = config.type_tag().unwrap_or(&config.slug);
let matches_entity = |r: &PdtSearchResult| config.tags_match(&r.tags);
let mut entities: Vec<Value> = if let Some(ref attached_to_id) = query.attached_to {
let results = pdt
.search_by_tag("type", type_tag, token)
.await
.unwrap_or_default();
let results: Vec<PdtSearchResult> = if let Some(ref agent_id) = instance_id {
results
.into_iter()
.filter(|r| r.get_tag("agent") == Some(agent_id))
.collect()
} else {
results
};
let mut filtered = Vec::new();
for r in &results {
if !matches_entity(r) {
continue;
}
if let Ok(relations) = pdt.get_relations(&r.id, None, token).await {
let has_relation = relations.iter().any(|rel| {
rel.from_asset_id == r.id
&& rel.to_asset_id == *attached_to_id
&& config.relation_fields().iter().any(|(_, fc)| {
fc.relation_type.as_deref() == Some(rel.relation_type.as_str())
})
});
if has_relation {
let entity = map_compact_to_entity(r, &config);
filtered.push(entity);
}
}
}
filtered
} else {
let results = pdt
.search_by_tag("type", type_tag, token)
.await
.unwrap_or_default();
let results: Vec<PdtSearchResult> = if let Some(ref agent_id) = instance_id {
results
.into_iter()
.filter(|r| r.get_tag("agent") == Some(agent_id))
.collect()
} else {
results
};
let has_relation_fields = !config.relation_fields().is_empty();
let mut filtered: Vec<Value> = Vec::new();
for r in &results {
if !matches_entity(r) {
continue;
}
if has_relation_fields {
let has_outgoing = match pdt.get_relations(&r.id, None, token).await {
Ok(relations) => relations.iter().any(|rel| rel.from_asset_id == r.id),
Err(_) => false,
};
if !has_outgoing {
continue;
}
}
let entity = map_compact_to_entity(r, &config);
if let Some(ref status_filter) = query.status {
let entity_status = entity.get("status").and_then(|s| s.as_str());
if entity_status != Some(status_filter.as_str()) {
continue;
}
}
filtered.push(entity);
}
filtered
};
entities.sort_by(|a, b| {
let a_time = a.get("updated_at").and_then(|t| t.as_str()).unwrap_or("");
let b_time = b.get("updated_at").and_then(|t| t.as_str()).unwrap_or("");
b_time.cmp(a_time)
});
let total = entities.len();
Ok(Json(json!({
"entities": entities,
"total": total,
})))
}
async fn get_entity_inner(
state: &AppState,
token: Option<&str>,
instance_id: Option<&str>,
slug: &str,
id: &str,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let pdt = state.pdt.for_instance(instance_id);
let config = get_entity_config(state, &slug).ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(json!({"error": format!("Unknown entity type: {}", slug)})),
)
})?;
let asset = pdt
.get_asset(id, token)
.await
.map_err(|e| (StatusCode::NOT_FOUND, Json(json!({"error": e.to_string()}))))?;
if !config.tags_match(&asset.tags) {
return Err((
StatusCode::NOT_FOUND,
Json(json!({
"error": format!("Asset {} is not a {}", id, config.name)
})),
));
}
let entity =
map_asset_to_entity_with_relations(state, instance_id, &asset, &config, token).await;
Ok(Json(entity))
}
fn user_groups(user: &AuthenticatedUser) -> Vec<String> {
user.claims_extra
.as_ref()
.and_then(|c| c.get("groups"))
.and_then(|g| g.as_array())
.map(|a| {
a.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect()
})
.unwrap_or_default()
}
fn pick_owner_group(groups: &[String]) -> Option<String> {
fn bare(g: &str) -> &str {
g.split('@').next().unwrap_or(g)
}
groups
.iter()
.find(|g| bare(g).starts_with("mem-"))
.cloned()
.or_else(|| {
groups
.iter()
.find(|g| {
let b = bare(g);
b.starts_with("ws-") && b.ends_with("-admins")
})
.cloned()
})
}
async fn create_entity_inner(
state: &AppState,
user: &AuthenticatedUser,
token: Option<&str>,
instance_id: Option<&str>,
slug: &str,
body: Value,
workspace_admins: WorkspaceAdmins,
) -> Result<(StatusCode, Json<Value>), (StatusCode, Json<Value>)> {
let pdt = state.pdt.for_instance(instance_id);
let config = get_entity_config(state, &slug).ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(json!({"error": format!("Unknown entity type: {}", slug)})),
)
})?;
let mut workspace_admin_group: Option<String> = None;
if let Some(ref authorizer) = state.authorizer {
if let Some(ref action) = config.cedar.action_create {
let claims = user.to_cedar_claims();
let resource_attrs: Vec<(String, String)> = if slug == "agent-identity" {
workspace_admin_group = workspace_admins.0.clone();
workspace_admin_group
.clone()
.map(|g| {
vec![(
crate::cedar::enforcement::ADMIN_GROUP_METADATA_KEY.to_string(),
g,
)]
})
.unwrap_or_default()
} else {
Vec::new()
};
crate::cedar::enforcement::check_permission_scoped(
authorizer,
&claims,
action,
&slug_to_cedar_type(&config.slug),
"<_>",
&resource_attrs,
)
.map_err(|status| (status, Json(json!({"error": "Access denied"}))))?;
}
}
let body_obj = body.as_object().ok_or_else(|| {
(
StatusCode::BAD_REQUEST,
Json(json!({"error": "Request body must be a JSON object"})),
)
})?;
for (field_name, field_config) in &config.fields {
if field_config.required && !body_obj.contains_key(field_name) {
return Err((
StatusCode::BAD_REQUEST,
Json(json!({
"error": "Validation failed",
"details": format!("Missing required field: {}", field_name)
})),
));
}
}
let title_value = config
.fields
.iter()
.find(|(_, fc)| fc.map.as_deref() == Some("title_suffix"))
.and_then(|(name, _)| body_obj.get(name))
.and_then(|v| v.as_str())
.unwrap_or("Untitled");
let title = format!("{}{}", config.title_prefix, title_value);
let content = config
.fields
.iter()
.find(|(_, fc)| fc.map.as_deref() == Some("content"))
.and_then(|(name, _)| body_obj.get(name))
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let mut tags: Vec<CreateTagRequest> = Vec::new();
for (category, value) in &config.default_tags {
tags.push(CreateTagRequest {
category: category.clone(),
value: value.clone(),
});
}
if let Some(ref agent_id) = instance_id {
tags.push(CreateTagRequest {
category: "agent".to_string(),
value: agent_id.to_string(),
});
}
for (field_name, field_config) in &config.fields {
if let Some(ref m) = field_config.map {
if let Some(tag_name) = m.strip_prefix("tag:") {
let value = body_obj
.get(field_name)
.and_then(|v| v.as_str())
.or(field_config.default.as_deref());
if let Some(v) = value {
tags.push(CreateTagRequest {
category: tag_name.to_string(),
value: v.to_string(),
});
}
}
}
}
let mut metadata: HashMap<String, Value> = HashMap::new();
for (field_name, field_config) in &config.fields {
if let Some(ref m) = field_config.map {
if let Some(key) = m.strip_prefix("metadata:") {
if let Some(v) = body_obj.get(field_name) {
metadata.insert(key.to_string(), v.clone());
}
}
}
}
if let Some(group) = workspace_admin_group.as_ref() {
metadata.insert(
crate::cedar::enforcement::ADMIN_GROUP_METADATA_KEY.to_string(),
Value::String(group.clone()),
);
}
let mut parent_auth_ctx = None;
for (field_name, _field_config) in config.relation_fields() {
if let Some(target_id) = body_obj.get(field_name).and_then(|v| v.as_str()) {
parent_auth_ctx = pdt.get_auth_context(target_id, token).await.ok().flatten();
if parent_auth_ctx.is_some() {
break;
}
}
}
let identity_admin_group = if slug == "agent-identity" {
workspace_admins.0.clone()
} else {
None
};
let auth_context = parent_auth_ctx.or_else(|| {
if slug == "agent-identity" {
return Some(AuthContext {
visibility: "team".to_string(),
owner_groups: identity_admin_group.map(|g| vec![g]).unwrap_or_default(),
confidentiality: String::new(),
});
}
pick_owner_group(&user_groups(user)).map(|group| AuthContext {
visibility: "team".to_string(),
owner_groups: vec![group],
confidentiality: String::new(),
})
});
if auth_context.is_none() {
tracing::warn!(
"memory created with EMPTY auth_context: creator '{}' has no mem-* or ws-*-admins group — admin re-stamp required",
user.user_id
);
}
let asset = pdt
.create_asset(
CreateAssetRequest {
title,
content,
tags: Some(tags),
auth_context,
},
token,
)
.await
.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": e.to_string()})),
)
})?;
if !metadata.is_empty() {
let mut metadata_req = pdt
.http_client()
.put(format!("{}/api/assets/{}", pdt.base_url(), asset.id))
.bearer_auth(token.unwrap_or(""));
if let Some(id) = instance_id {
metadata_req = metadata_req.header("X-Instance-Id", id);
}
let _ = metadata_req
.json(&json!({"metadata": metadata}))
.send()
.await;
}
info!(
"Created {} entity: {} ({})",
config.name, title_value, asset.id
);
for (field_name, field_config) in config.relation_fields() {
if let Some(target_id) = body_obj.get(field_name).and_then(|v| v.as_str()) {
let rel_type = field_config
.relation_type
.as_deref()
.unwrap_or("related_to");
let _ = pdt
.create_relation(
CreateRelationRequest {
from_asset_id: asset.id.clone(),
to_asset_id: target_id.to_string(),
relation_type: rel_type.to_string(),
metadata: None,
},
token,
)
.await;
info!("Linked {} → {} ({})", asset.id, target_id, rel_type);
}
}
let mut entity = map_asset_to_entity(&asset, &config);
if let Value::Object(ref mut map) = entity {
for (field_name, _) in config.relation_fields() {
if let Some(target_id) = body_obj.get(field_name).and_then(|v| v.as_str()) {
map.insert(field_name.clone(), Value::String(target_id.to_string()));
}
}
}
Ok((StatusCode::CREATED, Json(entity)))
}
#[cfg(test)]
mod auth_context_stamp_tests {
use super::pick_owner_group;
#[test]
fn agent_mem_group_wins() {
let groups = vec![
"pdt-api-agents".to_string(),
"mem-e69605b1@idp.example.com".to_string(),
];
assert_eq!(
pick_owner_group(&groups).as_deref(),
Some("mem-e69605b1@idp.example.com")
);
}
#[test]
fn human_ws_admins_group_when_no_mem_group() {
let groups = vec![
"pdt-api-users@idp.example.com".to_string(),
"ws-9f2e7d01-4c5b-6a7d-8e9f-001122334455-admins".to_string(),
];
assert_eq!(
pick_owner_group(&groups).as_deref(),
Some("ws-9f2e7d01-4c5b-6a7d-8e9f-001122334455-admins")
);
}
#[test]
fn spn_form_ws_admins_still_matches() {
let groups = vec!["ws-abc123-admins@idp.example.com".to_string()];
assert_eq!(
pick_owner_group(&groups).as_deref(),
Some("ws-abc123-admins@idp.example.com")
);
}
#[test]
fn plain_roles_never_own_memories() {
let groups = vec![
"pdt-api-users@idp.example.com".to_string(),
"pdt-api-agents".to_string(),
];
assert_eq!(pick_owner_group(&groups), None);
}
#[test]
fn mem_group_preferred_over_ws_admins() {
let groups = vec!["ws-abc-admins".to_string(), "mem-ff766ee2".to_string()];
assert_eq!(pick_owner_group(&groups).as_deref(), Some("mem-ff766ee2"));
}
}
async fn update_entity_status_inner(
state: &AppState,
user: &AuthenticatedUser,
token: Option<&str>,
instance_id: Option<&str>,
slug: &str,
id: &str,
body: Value,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let pdt = state.pdt.for_instance(instance_id);
let config = get_entity_config(state, &slug).ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(json!({"error": format!("Unknown entity type: {}", slug)})),
)
})?;
if let Some(ref authorizer) = state.authorizer {
if let Some(ref action) = config.cedar.action_edit {
let claims = user.to_cedar_claims();
let resource_attrs: Vec<(String, String)> = if slug == "agent-identity" {
pdt.get_asset(id, token)
.await
.ok()
.and_then(|asset| {
asset
.metadata
.get(crate::cedar::enforcement::ADMIN_GROUP_METADATA_KEY)
.and_then(|v| v.as_str())
.map(|g| {
vec![(
crate::cedar::enforcement::ADMIN_GROUP_METADATA_KEY.to_string(),
g.to_string(),
)]
})
})
.unwrap_or_default()
} else {
Vec::new()
};
crate::cedar::enforcement::check_permission_scoped(
authorizer,
&claims,
action,
&slug_to_cedar_type(&config.slug),
id,
&resource_attrs,
)
.map_err(|status| (status, Json(json!({"error": "Access denied"}))))?;
}
}
let new_status = body.get("status").and_then(|s| s.as_str()).ok_or_else(|| {
(
StatusCode::BAD_REQUEST,
Json(json!({"error": "Missing 'status' field"})),
)
})?;
if let Some(allowed) = config.status_values() {
if !allowed.iter().any(|v| v == new_status) {
return Err((
StatusCode::BAD_REQUEST,
Json(json!({
"error": "Invalid status",
"allowed_values": allowed,
"got": new_status
})),
));
}
}
let asset = pdt
.update_asset_tag(id, "status", new_status, token)
.await
.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": e.to_string()})),
)
})?;
let entity =
map_asset_to_entity_with_relations(state, instance_id, &asset, &config, token).await;
Ok(Json(entity))
}
async fn update_identity_content_inner(
state: &AppState,
user: &AuthenticatedUser,
token: Option<&str>,
instance_id: Option<&str>,
slug: &str,
id: &str,
body: Value,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let instance_id = instance_id.ok_or_else(|| {
(
StatusCode::BAD_REQUEST,
Json(json!({"error": "X-Instance-Id header required: agent identity content updates are instance-scoped (the identity lives in the agent's nested DB)"})),
)
})?;
let pdt = state.pdt.for_instance(Some(instance_id));
let config = get_entity_config(state, slug).ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(json!({"error": format!("Unknown entity type: {}", slug)})),
)
})?;
let asset = pdt.get_asset(id, token).await.map_err(|e| {
(
StatusCode::NOT_FOUND,
Json(json!({"error": format!("Agent identity not found: {} ({})", id, e)})),
)
})?;
if let Some(ref authorizer) = state.authorizer {
if let Some(ref action) = config.cedar.action_edit {
let claims = user.to_cedar_claims();
let resource_attrs: Vec<(String, String)> = asset
.metadata
.get(crate::cedar::enforcement::ADMIN_GROUP_METADATA_KEY)
.and_then(|v| v.as_str())
.map(|g| {
vec![(
crate::cedar::enforcement::ADMIN_GROUP_METADATA_KEY.to_string(),
g.to_string(),
)]
})
.unwrap_or_default();
crate::cedar::enforcement::check_permission_scoped(
authorizer,
&claims,
action,
&slug_to_cedar_type(&config.slug),
id,
&resource_attrs,
)
.map_err(|status| (status, Json(json!({"error": "Access denied"}))))?;
}
}
let new_content = body
.get("content")
.and_then(|c| c.as_str())
.ok_or_else(|| {
(
StatusCode::BAD_REQUEST,
Json(json!({"error": "Missing or non-string 'content' field"})),
)
})?
.to_string();
if new_content.trim().is_empty() {
tracing::warn!(
"EXPLICIT CONTENT WIPE: user {} wiped content of agent identity {}",
user.user_id,
id
);
}
use sha2::Digest;
let hash = |s: &str| format!("sha256:{:x}", sha2::Sha256::digest(s.as_bytes()));
let old_hash = if asset.content.is_empty() {
None
} else {
Some(hash(&asset.content))
};
let asset = pdt
.update_asset_content(id, &new_content, token)
.await
.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": format!("Failed to update identity content: {}", e)})),
)
})?;
tracing::info!(
"Agent identity {} content updated by user {} ({} chars)",
id,
user.user_id,
new_content.len()
);
let mut entity =
map_asset_to_entity_with_relations(state, Some(instance_id), &asset, &config, token).await;
if let Some(obj) = entity.as_object_mut() {
obj.insert(
"content_hash".to_string(),
json!({
"old": old_hash,
"new": hash(&new_content),
}),
);
}
Ok(Json(entity))
}
pub fn register_entity_routes(
router: axum::Router<AppState>,
configs: &[Arc<EntityConfig>],
) -> axum::Router<AppState> {
use axum::routing::{get, patch};
let mut router = router;
for config in configs {
if config.handler != "generic" {
continue;
}
let base = format!("/api/v1/{}s", config.slug);
let slug = config.slug.clone();
tracing::info!("Registering entity routes for {} at {}", config.name, base);
let slug_for_list = slug.clone();
router = router.route(
&base,
get(
move |State(state): State<AppState>,
ForwardedToken(token): ForwardedToken,
instance: InstanceContext,
Query(query): Query<ListEntitiesQuery>| {
let slug = slug_for_list.clone();
async move {
list_entities_inner(
&state,
token.as_deref(),
instance.as_deref(),
&slug,
&query,
)
.await
}
},
)
.post(
move |State(state): State<AppState>,
user: AuthenticatedUser,
ForwardedToken(token): ForwardedToken,
instance: InstanceContext,
workspace_admins: WorkspaceAdmins,
Json(body): Json<Value>| {
let slug = slug.clone();
async move {
create_entity_inner(
&state,
&user,
token.as_deref(),
instance.as_deref(),
&slug,
body,
workspace_admins,
)
.await
}
},
),
);
let slug_for_get = config.slug.clone();
router = router.route(
&format!("{}/{{id}}", base),
get(
move |State(state): State<AppState>,
ForwardedToken(token): ForwardedToken,
instance: InstanceContext,
Path(id): Path<String>| {
let slug = slug_for_get.clone();
async move {
get_entity_inner(&state, token.as_deref(), instance.as_deref(), &slug, &id)
.await
}
},
),
);
if config.slug == "agent-identity" {
let slug_for_content = config.slug.clone();
router = router.route(
&format!("{}/{{id}}/content", base),
patch(
move |State(state): State<AppState>,
user: AuthenticatedUser,
ForwardedToken(token): ForwardedToken,
instance: InstanceContext,
Path(id): Path<String>,
Json(body): Json<Value>| {
let slug = slug_for_content.clone();
async move {
update_identity_content_inner(
&state,
&user,
token.as_deref(),
instance.as_deref(),
&slug,
&id,
body,
)
.await
}
},
),
);
}
let slug_for_patch = config.slug.clone();
router = router.route(
&format!("{}/{{id}}/status", base),
patch(
move |State(state): State<AppState>,
user: AuthenticatedUser,
ForwardedToken(token): ForwardedToken,
instance: InstanceContext,
Path(id): Path<String>,
Json(body): Json<Value>| {
let slug = slug_for_patch.clone();
async move {
update_entity_status_inner(
&state,
&user,
token.as_deref(),
instance.as_deref(),
&slug,
&id,
body,
)
.await
}
},
),
);
}
router
}
pub async fn list_entity_configs(State(state): State<AppState>) -> Json<Value> {
let configs: Vec<Value> = state
.entity_configs
.iter()
.map(|c| {
json!({
"name": c.name,
"slug": c.slug,
"title_prefix": c.title_prefix,
"fields": c.fields.iter().map(|(name, fc)| {
json!({
"name": name,
"type": format!("{:?}", fc.field_type).to_lowercase(),
"required": fc.required,
"label": fc.label,
"values": fc.values,
"badge": fc.badge,
"in_views": fc.in_views,
})
}).collect::<Vec<_>>(),
"views": c.views.iter().map(|(_name, vc)| {
json!({
"layout": vc.layout,
"fields": vc.fields,
})
}).collect::<Vec<_>>(),
})
})
.collect();
Json(json!({
"entities": configs,
"total": configs.len(),
}))
}
#[cfg(test)]
mod identity_content_tests {
use super::*;
use crate::pdt::{CreateAssetRequest, CreateTagRequest, PdtClient};
use axum::response::IntoResponse;
use axum::routing::{get, post, put};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Mutex;
const ADMIN_GROUP: &str = "ws-1-admins";
const OLD: &str = "old charter";
const NEW: &str = "new charter";
fn user_with(role: &str, groups: &[&str]) -> AuthenticatedUser {
AuthenticatedUser {
user_id: "user-test".to_string(),
username: Some("tester".to_string()),
email: None,
claims_extra: Some(std::collections::HashMap::from([
(
"role".to_string(),
serde_json::Value::String(role.to_string()),
),
(
"groups".to_string(),
serde_json::Value::Array(
groups
.iter()
.map(|g| serde_json::Value::String(g.to_string()))
.collect(),
),
),
])),
}
}
fn sha_hex(s: &str) -> String {
use sha2::Digest;
format!("sha256:{:x}", sha2::Sha256::digest(s.as_bytes()))
}
fn prebuilt_pdt_bin() -> Option<std::path::PathBuf> {
if let Ok(p) = std::env::var("FAME_IT_PDT_BIN") {
let b = std::path::PathBuf::from(p);
if b.exists() {
return Some(b);
}
}
let b = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.parent()?
.join("pdt/target/debug/pdt");
b.exists().then_some(b)
}
struct PdtProc(Child);
impl Drop for PdtProc {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
async fn free_port() -> u16 {
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = l.local_addr().unwrap().port();
drop(l);
port
}
async fn wait_healthy(base: &str) {
let http = reqwest::Client::new();
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
loop {
if tokio::time::Instant::now() > deadline {
panic!("pdt did not become healthy at {base}");
}
if let Ok(resp) = http.get(format!("{base}/health")).send().await {
if resp.status().is_success() {
return;
}
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
}
async fn spawn_real_pdt(bin: std::path::PathBuf) -> (String, PdtProc) {
let n = SEQ.fetch_add(1, Ordering::SeqCst);
let tmp = std::env::temp_dir().join(format!("fame-pdt-it-{}-{n}", std::process::id()));
std::fs::create_dir_all(tmp.join("instances")).expect("tmp instances dir");
let port = free_port().await;
let child = Command::new(&bin)
.env("PDT_HOST", "127.0.0.1")
.env("PDT_PORT", port.to_string())
.env("PDT_DB_BACKEND", "sqlite")
.env("SQLITE_PATH", tmp.join("global.db"))
.env("PDT_INSTANCES_DIR", tmp.join("instances"))
.env("AUTH_ENABLED", "false")
.env("AUTH_DEV_MODE", "true")
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.expect("failed to spawn prebuilt pdt binary");
let base = format!("http://127.0.0.1:{port}");
wait_healthy(&base).await;
(base, PdtProc(child))
}
#[derive(Default)]
struct MockDb {
assets: Vec<serde_json::Value>,
next_id: u32,
}
struct MockState {
db: Mutex<MockDb>,
}
fn mock_tag(db: &mut MockDb, category: &str, value: &str) -> serde_json::Value {
db.next_id += 1;
serde_json::json!({
"id": format!("tag-{}", db.next_id),
"category": category,
"value": value,
"added_by": "mock",
"added_at": "2026-08-29T00:00:00Z"
})
}
async fn spawn_mock_pdt() -> String {
let state = std::sync::Arc::new(MockState {
db: Mutex::new(MockDb::default()),
});
let st = state.clone();
let create_route = post(
move |headers: axum::http::HeaderMap, Json(body): Json<serde_json::Value>| {
let st = st.clone();
async move {
let mut db = st.db.lock().unwrap();
db.next_id += 1;
let id = format!("ident-{:04}", db.next_id);
let mut tags: Vec<serde_json::Value> = Vec::new();
if let Some(list) = body.get("tags").and_then(|t| t.as_array()) {
for t in list {
tags.push(mock_tag(
&mut db,
t["category"].as_str().unwrap_or_default(),
t["value"].as_str().unwrap_or_default(),
));
}
}
let _ = headers; let auth_context = body
.get("auth_context")
.cloned()
.filter(|v| v.is_object())
.unwrap_or_else(|| {
serde_json::json!({
"visibility": "",
"owner_groups": [],
"confidentiality": ""
})
});
let asset = serde_json::json!({
"_id": id,
"title": body.get("title").cloned().unwrap_or(serde_json::Value::Null),
"content": body.get("content").cloned().unwrap_or(serde_json::Value::String(String::new())),
"tags": tags,
"metadata": {},
"auth_context": auth_context,
"created_at": "2026-08-29T00:00:00Z",
"updated_at": "2026-08-29T00:00:00Z"
});
db.assets.push(asset.clone());
axum::Json(asset).into_response()
}
},
);
let st_get = state.clone();
let st_put = state.clone();
let asset_route = get(move |Path(id): Path<String>| {
let st = st_get.clone();
async move {
use axum::response::IntoResponse;
let db = st.db.lock().unwrap();
match db.assets.iter().find(|a| a["_id"] == id) {
Some(a) => axum::Json(a.clone()).into_response(),
None => (
axum::http::StatusCode::NOT_FOUND,
axum::Json(serde_json::json!({"error": "not found"})),
)
.into_response(),
}
}
})
.put(
move |Path(id): Path<String>, Json(body): Json<serde_json::Value>| {
let st = st_put.clone();
async move {
use axum::response::IntoResponse;
let mut db = st.db.lock().unwrap();
match db.assets.iter_mut().find(|a| a["_id"] == id) {
Some(a) => {
if let Some(c) = body.get("content") {
a["content"] = c.clone();
}
if let Some(t) = body.get("title") {
a["title"] = t.clone();
}
if let Some(m) = body.get("metadata").and_then(|m| m.as_object()) {
let md = a["metadata"].as_object_mut().unwrap();
for (k, v) in m {
md.insert(k.clone(), v.clone());
}
}
a["updated_at"] =
serde_json::Value::String("2026-08-31T12:00:00Z".into());
axum::Json(a.clone()).into_response()
}
None => (
axum::http::StatusCode::NOT_FOUND,
axum::Json(serde_json::json!({"error": "not found"})),
)
.into_response(),
}
}
},
);
let relations_route =
get(move |Path(_id): Path<String>| async move { axum::Json(serde_json::json!([])) });
let st = state.clone();
let search_route = get(
move |Query(q): Query<std::collections::HashMap<String, String>>| {
let st = st.clone();
async move {
use axum::response::IntoResponse;
let tag = q.get("tag").cloned().unwrap_or_default();
let (cat, val) = match tag.split_once(':') {
Some((c, v)) => (c.to_string(), v.to_string()),
None => (tag.clone(), String::new()),
};
let db = st.db.lock().unwrap();
let data: Vec<serde_json::Value> = db
.assets
.iter()
.filter(|a| {
a["tags"].as_array().is_some_and(|tags| {
tags.iter().any(|t| {
t["category"] == serde_json::Value::String(cat.clone())
&& t["value"] == serde_json::Value::String(val.clone())
})
})
})
.map(|a| {
serde_json::json!({
"_id": a["_id"],
"title": a["title"],
"tags": a["tags"],
"updated_at": a["updated_at"]
})
})
.collect();
axum::Json(serde_json::json!({ "data": data })).into_response()
}
},
);
let app = axum::Router::new()
.route("/api/assets", create_route)
.route("/api/assets/{id}", asset_route)
.route("/api/assets/{id}/relations", relations_route)
.route("/api/search", search_route);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
format!("http://{addr}")
}
use axum::extract::Query;
static SEQ: AtomicU32 = AtomicU32::new(0);
struct TestEnv {
state: AppState,
base: String,
ws: String,
agent: String,
identity_id: String,
_proc: Option<PdtProc>,
}
async fn test_env() -> TestEnv {
let (tier, base, proc) = match prebuilt_pdt_bin() {
Some(bin) => {
let (base, proc) = spawn_real_pdt(bin).await;
("real-pdt", base, Some(proc))
}
None => {
eprintln!(
"identity_content_tests: backend = mock-pdt (prebuilt pdt binary not found; \
build ../pdt with `cargo build --no-default-features --features \
sqlite-backend` to run the real tier)"
);
("mock-pdt", spawn_mock_pdt().await, None)
}
};
let _ = tier;
let http = reqwest::Client::new();
let ws = uuid::Uuid::new_v4().to_string();
let agent = uuid::Uuid::new_v4().to_string();
if proc.is_some() {
for (id, parent) in [(&ws, None), (&agent, Some(&ws))] {
let resp = http
.post(format!("{base}/api/instances/{id}/provision"))
.query(&[("parent", parent.map(|p| p.to_string()))])
.send()
.await
.expect("provision request");
assert!(
resp.status().is_success(),
"provision {id} failed: {}",
resp.status()
);
}
}
let cedar_config: pep::cedar::CedarConfig = crate::config::CedarConfig {
enabled: true,
policy_path: "./policies".to_string(),
schema_path: "./policies/schema.cedarschema".to_string(),
validate_on_load: true,
policy_store_url: None,
policy_store_token: None,
}
.into();
let authorizer = pep::cedar::CedarAuthorizer::new_with_policy_store(cedar_config)
.await
.expect("authorizer loads from embedded policies");
let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("entities");
let entity_configs: Vec<_> =
crate::entity_config::load_entity_configs(dir.to_str().unwrap())
.into_iter()
.map(Arc::new)
.collect();
assert!(entity_configs.iter().any(|c| c.slug == "agent-identity"));
let config: crate::config::Config =
toml::from_str(&format!("host = '127.0.0.1'\nport = 0\npdt_url = '{base}'"))
.expect("minimal test config parses");
let state = AppState {
config: Arc::new(config),
pdt: Arc::new(PdtClient::new(&base)),
authorizer: Some(Arc::new(authorizer)),
entity_configs,
};
let identity = state
.pdt
.for_instance(Some(&agent))
.create_asset(
CreateAssetRequest {
title: "Identity: Test Agent".to_string(),
content: Some(OLD.to_string()),
tags: Some(vec![
CreateTagRequest {
category: "type".to_string(),
value: "agent-identity".to_string(),
},
CreateTagRequest {
category: "agent".to_string(),
value: agent.clone(),
},
]),
auth_context: None,
},
None,
)
.await
.expect("seed identity via pdt client");
let resp = http
.put(format!("{base}/api/assets/{}", identity.id))
.header("X-Instance-Id", &agent)
.json(&serde_json::json!({ "metadata": { "admin_group": ADMIN_GROUP } }))
.send()
.await
.expect("metadata patch");
assert!(resp.status().is_success(), "metadata patch failed");
TestEnv {
state,
base,
ws,
agent,
identity_id: identity.id,
_proc: proc,
}
}
fn body(content: &str) -> Value {
serde_json::json!({ "content": content })
}
async fn call(
env: &TestEnv,
user: &AuthenticatedUser,
instance: Option<&str>,
id: &str,
body: Value,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
update_identity_content_inner(&env.state, user, None, instance, "agent-identity", id, body)
.await
}
#[tokio::test]
async fn roundtrip_updates_in_place_with_hash_echo() {
let env = test_env().await;
let resp = call(
&env,
&user_with("user", &[ADMIN_GROUP]),
Some(&env.agent),
&env.identity_id,
body(NEW),
)
.await
.expect("scoped user may edit identity content");
let resp = resp.0;
assert_eq!(resp["content"], NEW, "content lands");
assert_eq!(resp["content_hash"]["old"], sha_hex(OLD), "old hash echoes");
assert_eq!(resp["content_hash"]["new"], sha_hex(NEW), "new hash echoes");
let inst = env.state.pdt.for_instance(Some(&env.agent));
let read = inst
.get_asset(&env.identity_id, None)
.await
.expect("identity readable");
assert_eq!(read.content, NEW);
let idents = inst
.search_by_tag("type", "agent-identity", None)
.await
.expect("identity search");
assert_eq!(idents.len(), 1, "zero duplicate identity records");
}
#[tokio::test]
async fn unknown_identity_is_loud_404() {
let env = test_env().await;
let err = call(
&env,
&user_with("admin", &[]),
Some(&env.agent),
"ident-missing",
body(NEW),
)
.await
.expect_err("unknown id must 404");
assert_eq!(err.0, StatusCode::NOT_FOUND);
assert!(
err.1 .0["error"]
.as_str()
.unwrap()
.contains("Agent identity not found"),
"loud message: {}",
err.1 .0
);
}
#[tokio::test]
async fn cedar_denies_principal_outside_admin_group() {
let env = test_env().await;
let err = call(
&env,
&user_with("user", &["some-other-group"]),
Some(&env.agent),
&env.identity_id,
body(NEW),
)
.await
.expect_err("out-of-group user must be denied");
assert_eq!(err.0, StatusCode::FORBIDDEN);
assert_eq!(err.1 .0["error"], "Access denied");
let asset = env
.state
.pdt
.for_instance(Some(&env.agent))
.get_asset(&env.identity_id, None)
.await
.unwrap();
assert_eq!(asset.content, OLD, "denied write must not land");
}
#[tokio::test]
async fn admin_role_edits_without_group() {
let env = test_env().await;
let resp = call(
&env,
&user_with("admin", &[]),
Some(&env.agent),
&env.identity_id,
body(NEW),
)
.await
.expect("admin may edit");
assert_eq!(resp.0["content"], NEW);
}
#[tokio::test]
async fn agent_role_edits_own_identity() {
let env = test_env().await;
let resp = call(
&env,
&user_with("agent", &[ADMIN_GROUP]),
Some(&env.agent),
&env.identity_id,
body("self-written charter"),
)
.await
.expect("agent may edit own identity — admin_group matches its group");
assert_eq!(resp.0["content"], "self-written charter");
}
#[tokio::test]
async fn identity_create_birth_stamps_auth_context() {
let env = test_env().await;
let resp = create_entity_inner(
&env.state,
&user_with("service", &[]),
None,
Some(&env.agent),
"agent-identity",
serde_json::json!({"title": "Fresh Charter", "content": NEW}),
WorkspaceAdmins(Some("mem-fresh@idp.example.com".to_string())),
)
.await
.expect("bootstrap create");
assert_eq!(resp.0, StatusCode::CREATED);
let id = resp.1 .0["id"].as_str().expect("entity id");
let asset = env
.state
.pdt
.for_instance(Some(&env.agent))
.get_asset(id, None)
.await
.expect("created identity readable");
let ac = asset.auth_context.expect("auth_context present at birth");
assert_eq!(ac.visibility, "team");
assert_eq!(ac.owner_groups, vec!["mem-fresh@idp.example.com".to_string()]);
assert_eq!(ac.confidentiality, "");
}
#[tokio::test]
async fn new_agent_self_patch_after_birth_zero_hand_stamps() {
let env = test_env().await;
let resp = create_entity_inner(
&env.state,
&user_with("service", &[]),
None,
Some(&env.agent),
"agent-identity",
serde_json::json!({"title": "Fresh Charter", "content": OLD}),
WorkspaceAdmins(Some("mem-fresh@idp.example.com".to_string())),
)
.await
.expect("bootstrap create");
let id = resp.1 .0["id"].as_str().expect("entity id").to_string();
let patched = update_identity_content_inner(
&env.state,
&user_with("agent", &["mem-fresh@idp.example.com"]),
None,
Some(&env.agent),
"agent-identity",
&id,
serde_json::json!({ "content": NEW }),
)
.await
.expect("self-PATCH allowed on birth-stamped identity");
assert_eq!(patched.0["content"], NEW);
assert_eq!(patched.0["content_hash"]["new"], {
use sha2::Digest;
format!("sha256:{:x}", sha2::Sha256::digest(NEW.as_bytes()))
});
}
#[tokio::test]
async fn agent_edit_cross_identity_denied_via_cedar() {
let env = test_env().await;
let err = call(
&env,
&user_with("agent", &["mem-someoneelse"]),
Some(&env.agent),
&env.identity_id,
body("hijack attempt"),
)
.await
.expect_err("cross-agent edit must be denied by policy");
assert_eq!(err.0, StatusCode::FORBIDDEN);
assert_eq!(
err.1 .0["error"], "Access denied",
"Cedar decision, not resolution error"
);
let asset = env
.state
.pdt
.for_instance(Some(&env.agent))
.get_asset(&env.identity_id, None)
.await
.unwrap();
assert_eq!(asset.content, OLD, "denied write must not land");
}
#[tokio::test]
async fn owner_stamped_identities_outside_agent_write_scope() {
let env = test_env().await;
let err = call(
&env,
&user_with("agent", &["mem-agentown"]),
Some(&env.agent),
&env.identity_id,
body("still denied"),
)
.await
.expect_err("unstamped identity must deny agents");
assert_eq!(err.0, StatusCode::FORBIDDEN);
assert_eq!(err.1 .0["error"], "Access denied");
let owner_asset = env
.state
.pdt
.for_instance(Some(&env.agent))
.create_asset(
crate::pdt::CreateAssetRequest {
title: "Identity: Owner-scope".to_string(),
content: Some(OLD.to_string()),
tags: Some(vec![crate::pdt::CreateTagRequest {
category: "type".to_string(),
value: "agent-identity".to_string(),
}]),
auth_context: None,
},
None,
)
.await
.expect("owner identity seeded");
let http = reqwest::Client::new();
let resp = http
.put(format!("{}/api/assets/{}", env.base, owner_asset.id))
.header("X-Instance-Id", &env.agent)
.json(&serde_json::json!({ "metadata": { "admin_group": "farzan-owner-scope" } }))
.send()
.await
.expect("owner stamp");
assert!(resp.status().is_success());
let err = call(
&env,
&user_with("agent", &["mem-agentown"]),
Some(&env.agent),
&owner_asset.id,
body("hijack"),
)
.await
.expect_err("foreign admin_group must deny agents");
assert_eq!(err.0, StatusCode::FORBIDDEN);
assert_eq!(err.1 .0["error"], "Access denied");
}
#[tokio::test]
async fn empty_content_is_explicit_wipe() {
let env = test_env().await;
let resp = call(
&env,
&user_with("user", &[ADMIN_GROUP]),
Some(&env.agent),
&env.identity_id,
body(""),
)
.await
.expect("explicit wipe allowed");
assert_eq!(resp.0["content_hash"]["new"], sha_hex(""));
let asset = env
.state
.pdt
.for_instance(Some(&env.agent))
.get_asset(&env.identity_id, None)
.await
.unwrap();
assert_eq!(asset.content, "", "wipe stored as empty string");
}
#[tokio::test]
async fn missing_instance_header_is_loud_400() {
let env = test_env().await;
let err = call(
&env,
&user_with("admin", &[]),
None,
&env.identity_id,
body(NEW),
)
.await
.expect_err("missing instance header must 400");
assert_eq!(err.0, StatusCode::BAD_REQUEST);
assert!(
err.1 .0["error"]
.as_str()
.unwrap()
.contains("X-Instance-Id header required"),
"loud message: {}",
err.1 .0
);
}
#[tokio::test]
async fn agent_role_creates_missing_identity_allowed() {
let env = test_env().await;
let report_instance = uuid::Uuid::new_v4().to_string();
let new_id = uuid::Uuid::new_v4().to_string();
let resp = create_entity_inner(
&env.state,
&user_with("agent", &[]),
None,
Some(&report_instance),
"agent-identity",
serde_json::json!({"title": "Provisioned identity (bootstrap)", "content": NEW}),
WorkspaceAdmins(None),
)
.await
.expect("manager agent may bootstrap a report identity");
assert_eq!(resp.0, StatusCode::CREATED);
assert_eq!(resp.1 .0["content"], NEW);
}
#[tokio::test]
async fn service_role_creates_missing_identity_allowed() {
let env = test_env().await;
let report_instance = uuid::Uuid::new_v4().to_string();
let new_id = uuid::Uuid::new_v4().to_string();
let resp = create_entity_inner(
&env.state,
&user_with("service", &[]),
None,
Some(&report_instance),
"agent-identity",
serde_json::json!({"title": "Provisioned identity (bootstrap)", "content": NEW}),
WorkspaceAdmins(None),
)
.await
.expect("service principal may bootstrap identity");
assert_eq!(resp.0, StatusCode::CREATED);
assert_eq!(resp.1 .0["content"], NEW);
}
}