Add relay authority for private managed agents

Introduce an authenticated aggregate endpoint and serializable authority
transaction for encrypted private managed-agent heads and their exact public
projections. Preserve immutable signed recovery bytes, enforce generation and
definition revision CAS, reactivate submitted projections, retire prior heads,
and return a writer-consistent signed aggregate snapshot.

Keep kind 30179 author-only and out of full-text search, reject generic writes,
and advertise the aggregate capability through NIP-11.

Co-authored-by: Princess Donut <68157ebd23b3897c1991015c3038658ea916200c67d3a54620b0754d1b92f6e0@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>

crates/buzz-db/src/error.rs

@@ -41,6 +41,10 @@ pub enum DbError {
     #[error("serialization error: {0}")]
     Serde(#[from] serde_json::Error),
 
+    /// A private managed-agent compare-and-swap conflicted with current authority.
+    #[error("managed-agent conflict: {0}")]
+    ManagedAgentConflict(String),
+
     /// A value in the database is malformed or unexpected.
     #[error("invalid data: {0}")]
     InvalidData(String),

crates/buzz-db/src/event.rs

@@ -1164,7 +1164,7 @@ pub struct ThreadMetadataParams<'a> {
     pub broadcast: bool,
 }
 
-async fn insert_event_with_thread_metadata_tx(
+pub(crate) async fn insert_event_with_thread_metadata_tx(
     tx: &mut Transaction<'_, Postgres>,
     community_id: CommunityId,
     event: &Event,

crates/buzz-db/src/lib.rs

@@ -27,6 +27,8 @@ pub mod event;
 pub mod feed;
 /// Git repository name registry (NIP-34 kind:30617).
 pub mod git_repo;
+/// Transactional private managed-agent authority.
+pub mod managed_agent;
 /// Embedded database migrations.
 pub mod migration;
 /// Community moderation: reports, bans/timeouts, audit actions.
@@ -4777,6 +4779,25 @@ impl Db {
 
     /// Atomically replace a NIP-33 parameterized replaceable event (kind 30000–39999).
     ///
+    /// Atomically commit a relay-verifiable private managed-agent aggregate.
+    pub async fn commit_managed_agent_aggregate(
+        &self,
+        community_id: CommunityId,
+        request: &managed_agent::AggregateRequest,
+    ) -> Result<managed_agent::AggregateCommit> {
+        managed_agent::commit_aggregate(&self.pool, community_id, request).await
+    }
+
+    /// Whether an ordinary projection targets PMA-authoritative state.
+    pub async fn managed_agent_projection_is_authoritative(
+        &self,
+        community_id: CommunityId,
+        event: &nostr::Event,
+        d_tag: &str,
+    ) -> Result<bool> {
+        managed_agent::projection_is_authoritative(&self.pool, community_id, event, d_tag).await
+    }
+
     /// Keeps only the event with the highest `created_at` per `(kind, pubkey, d_tag)`.
     /// Same-second ties are broken by lowest event `id` (deterministic ordering).
     /// The entire check → retire old payload → insert runs in a single transaction

crates/buzz-db/src/managed_agent.rs

@@ -0,0 +1,668 @@
+//! Transactional relay authority for NIP-PMA private managed agents.
+
+use buzz_core::private_managed_agent::{self, Envelope, State};
+use buzz_core::{CommunityId, StoredEvent};
+use nostr::Event;
+use sqlx::{PgPool, Postgres, Row, Transaction};
+
+use crate::{DbError, Result};
+
+/// Relay-verifiable aggregate submission. Plaintext and agent secrets never
+/// cross this boundary: Desktop validates ciphertext/binding consistency after
+/// writer-consistent read-back.
+#[derive(Debug, Clone)]
+pub struct AggregateRequest {
+    /// Signed encrypted kind:30179 head.
+    pub private_event: Event,
+    /// Signed kind:30175 candidate, required for active heads.
+    pub definition_event: Option<Event>,
+    /// Signed kind:30177 candidate, required for active heads.
+    pub instance_event: Option<Event>,
+    /// Revision already embedded in the encrypted active payload. Required for
+    /// active heads and rejected unless it matches the locked definition head.
+    pub expected_definition_revision: Option<u64>,
+}
+
+/// Events committed by an aggregate submission, suitable for post-commit fan-out.
+#[derive(Debug, Clone)]
+pub struct AggregateCommit {
+    /// Writer-consistent committed head metadata.
+    pub head: ManagedAgentHead,
+    /// Newly committed events. Empty for a byte-identical retry.
+    pub events: Vec<StoredEvent>,
+    /// Whether this call committed a new generation.
+    pub inserted: bool,
+    /// Transactionally confirmed definition revision for active heads.
+    pub definition_revision: Option<u64>,
+    /// Writer-consistent signed private head.
+    pub private_event: Event,
+    /// Writer-consistent signed definition binding.
+    pub definition_event: Option<Event>,
+    /// Writer-consistent signed instance binding.
+    pub instance_event: Option<Event>,
+}
+
+/// Writer-consistent PMA authority head.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ManagedAgentHead {
+    /// Owner pubkey.
+    pub owner_pubkey: Vec<u8>,
+    /// Agent pubkey.
+    pub agent_pubkey: Vec<u8>,
+    /// Current generation floor.
+    pub generation: u64,
+    /// Exact current private event ID.
+    pub event_id: Vec<u8>,
+    /// Whether the current head is active.
+    pub active: bool,
+}
+
+fn invalid(message: impl Into<String>) -> DbError {
+    DbError::InvalidData(message.into())
+}
+
+fn exact_d(event: &Event) -> Result<&str> {
+    let values: Vec<_> = event
+        .tags
+        .iter()
+        .filter_map(|tag| {
+            let parts = tag.as_slice();
+            (parts.first().is_some_and(|part| part == "d")).then_some(parts)
+        })
+        .collect();
+    if values.len() != 1 || values[0].len() != 2 || values[0][1].is_empty() {
+        return Err(invalid(
+            "projection requires exactly one two-element non-empty d tag",
+        ));
+    }
+    Ok(&values[0][1])
+}
+
+fn validate_projection(event: &Event, owner: &nostr::PublicKey, kind: u32) -> Result<String> {
+    if event.kind.as_u16() as u32 != kind
+        || &event.pubkey != owner
+        || !event.verify_id()
+        || !event.verify_signature()
+    {
+        return Err(invalid(format!(
+            "invalid signed kind:{kind} owner projection"
+        )));
+    }
+    Ok(exact_d(event)?.to_owned())
+}
+
+fn validate_request(request: &AggregateRequest) -> Result<(Envelope, Vec<Event>, Option<String>)> {
+    let owner = request.private_event.pubkey;
+    let envelope = private_managed_agent::validate_envelope(&request.private_event, &owner)
+        .map_err(|e| invalid(e.to_string()))?;
+    match envelope.state {
+        State::Active => {
+            request
+                .expected_definition_revision
+                .filter(|revision| *revision > 0 && *revision <= i64::MAX as u64)
+                .ok_or_else(|| {
+                    invalid("active aggregate requires a valid expected definition revision")
+                })?;
+            let definition = request
+                .definition_event
+                .as_ref()
+                .ok_or_else(|| invalid("active aggregate missing definition projection"))?;
+            let instance = request
+                .instance_event
+                .as_ref()
+                .ok_or_else(|| invalid("active aggregate missing instance projection"))?;
+            let definition_d = validate_projection(definition, &owner, 30175)?;
+            let instance_d = validate_projection(instance, &owner, 30177)?;
+            if instance_d != envelope.agent_pubkey.to_hex() {
+                return Err(invalid(
+                    "instance projection d does not match agent coordinate",
+                ));
+            }
+            Ok((
+                envelope,
+                vec![definition.clone(), instance.clone()],
+                Some(definition_d),
+            ))
+        }
+        State::Deleted => {
+            if request.definition_event.is_some()
+                || request.instance_event.is_some()
+                || request.expected_definition_revision.is_some()
+            {
+                return Err(invalid(
+                    "deleted aggregate must not carry projections or a definition revision",
+                ));
+            }
+            Ok((envelope, vec![], None))
+        }
+    }
+}
+
+fn decode_event(value: serde_json::Value, label: &str) -> Result<Event> {
+    serde_json::from_value(value)
+        .map_err(|error| invalid(format!("invalid stored {label}: {error}")))
+}
+
+async fn read_snapshot_tx(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    owner: &[u8],
+    agent: &[u8],
+    generation: u64,
+    inserted: bool,
+    events: Vec<StoredEvent>,
+) -> Result<AggregateCommit> {
+    let row = sqlx::query(
+        "SELECT h.owner_pubkey,h.agent_pubkey,h.generation,h.event_id,h.state,\
+                r.definition_revision,r.private_event,r.definition_event,r.instance_event \
+           FROM managed_agent_heads h \
+           JOIN managed_agent_revisions r ON r.community_id=h.community_id \
+            AND r.owner_pubkey=h.owner_pubkey AND r.agent_pubkey=h.agent_pubkey \
+            AND r.generation=h.generation \
+          WHERE h.community_id=$1 AND h.owner_pubkey=$2 AND h.agent_pubkey=$3 \
+            AND h.generation=$4",
+    )
+    .bind(community.as_uuid())
+    .bind(owner)
+    .bind(agent)
+    .bind(generation as i64)
+    .fetch_optional(&mut **tx)
+    .await?
+    .ok_or_else(|| invalid("committed aggregate snapshot missing"))?;
+    let definition = row
+        .try_get::<Option<serde_json::Value>, _>("definition_event")?
+        .map(|value| decode_event(value, "definition event"))
+        .transpose()?;
+    let instance = row
+        .try_get::<Option<serde_json::Value>, _>("instance_event")?
+        .map(|value| decode_event(value, "instance event"))
+        .transpose()?;
+    Ok(AggregateCommit {
+        head: ManagedAgentHead {
+            owner_pubkey: row.try_get("owner_pubkey")?,
+            agent_pubkey: row.try_get("agent_pubkey")?,
+            generation: row.try_get::<i64, _>("generation")? as u64,
+            event_id: row.try_get("event_id")?,
+            active: row.try_get::<String, _>("state")? == "active",
+        },
+        events,
+        inserted,
+        definition_revision: row
+            .try_get::<Option<i64>, _>("definition_revision")?
+            .map(|value| value as u64),
+        private_event: decode_event(row.try_get("private_event")?, "private event")?,
+        definition_event: definition,
+        instance_event: instance,
+    })
+}
+
+/// Atomically CAS and persist a PMA head and its exact projections.
+/// Serialization failures are retried at most twice with the same already-owned
+/// request object. Validation is deterministic and no CAS conflict is retried.
+pub async fn commit_aggregate(
+    pool: &PgPool,
+    community: CommunityId,
+    request: &AggregateRequest,
+) -> Result<AggregateCommit> {
+    validate_request(request)?;
+    for attempt in 0..=2 {
+        match commit_aggregate_once(pool, community, request).await {
+            Err(DbError::Sqlx(sqlx::Error::Database(error)))
+                if error.code().as_deref() == Some("40001") && attempt < 2 => {}
+            result => return result,
+        }
+    }
+    unreachable!("bounded serialization retry always returns")
+}
+
+async fn commit_aggregate_once(
+    pool: &PgPool,
+    community: CommunityId,
+    request: &AggregateRequest,
+) -> Result<AggregateCommit> {
+    let (envelope, projections, definition_d) = validate_request(request)?;
+    let owner = envelope.owner_pubkey.to_bytes().to_vec();
+    let agent = envelope.agent_pubkey.to_bytes().to_vec();
+    let event_id = request.private_event.id.to_bytes().to_vec();
+    let previous = envelope.previous_event_id.map(|id| id.to_bytes().to_vec());
+    let definition_event_id = projections
+        .first()
+        .map(|event| event.id.to_bytes().to_vec());
+    let definition_hash = projections
+        .first()
+        .map(|event| private_managed_agent::content_sha256(event.content.as_bytes()))
+        .map(|hash| hex::decode(hash).expect("sha256 hex"));
+    let instance_event_id = projections.get(1).map(|event| event.id.to_bytes().to_vec());
+    let instance_hash = projections
+        .get(1)
+        .map(|event| private_managed_agent::content_sha256(event.content.as_bytes()))
+        .map(|hash| hex::decode(hash).expect("sha256 hex"));
+
+    let mut tx = pool.begin().await?;
+    sqlx::query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
+        .execute(&mut *tx)
+        .await?;
+    let lock = super::event_replacement_lock_key(community, 30179, &owner, Some(&agent));
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(lock)
+        .execute(&mut *tx)
+        .await?;
+    let current = sqlx::query("SELECT generation,event_id,state,definition_revision,definition_event_id,definition_content_sha256,instance_event_id,instance_content_sha256 FROM managed_agent_heads WHERE community_id=$1 AND owner_pubkey=$2 AND agent_pubkey=$3 FOR UPDATE")
+        .bind(community.as_uuid()).bind(&owner).bind(&agent).fetch_optional(&mut *tx).await?;
+    if let Some(row) = &current {
+        let same_head = row.try_get::<i64, _>("generation")? as u64 == envelope.generation
+            && row.try_get::<Vec<u8>, _>("event_id")? == event_id;
+        if same_head {
+            let same_bindings = row.try_get::<String, _>("state")?
+                == match envelope.state {
+                    State::Active => "active",
+                    State::Deleted => "deleted",
+                }
+                && row.try_get::<Option<Vec<u8>>, _>("definition_event_id")? == definition_event_id
+                && row.try_get::<Option<Vec<u8>>, _>("definition_content_sha256")?
+                    == definition_hash
+                && row.try_get::<Option<Vec<u8>>, _>("instance_event_id")? == instance_event_id
+                && row.try_get::<Option<Vec<u8>>, _>("instance_content_sha256")? == instance_hash
+                && row
+                    .try_get::<Option<i64>, _>("definition_revision")?
+                    .map(|revision| revision as u64)
+                    == request.expected_definition_revision;
+            if !same_bindings {
+                return Err(DbError::ManagedAgentConflict(
+                    "idempotent retry changed projection bindings".into(),
+                ));
+            }
+            let snapshot = read_snapshot_tx(
+                &mut tx,
+                community,
+                &owner,
+                &agent,
+                envelope.generation,
+                false,
+                vec![],
+            )
+            .await?;
+            tx.commit().await?;
+            return Ok(snapshot);
+        }
+    }
+    match &current {
+        None if envelope.generation == 1 && previous.is_none() => {}
+        Some(row)
+            if envelope.generation == row.try_get::<i64, _>("generation")? as u64 + 1
+                && previous.as_deref()
+                    == Some(row.try_get::<Vec<u8>, _>("event_id")?.as_slice()) => {}
+        None => {
+            return Err(DbError::ManagedAgentConflict(
+                "genesis requires generation 1 without predecessor".into(),
+            ))
+        }
+        Some(_) => {
+            return Err(DbError::ManagedAgentConflict(
+                "predecessor or generation conflict".into(),
+            ))
+        }
+    }
+
+    let definition_revision = if let (Some(d), Some(definition_id), Some(hash)) =
+        (&definition_d, &definition_event_id, &definition_hash)
+    {
+        let definition_lock =
+            super::event_replacement_lock_key(community, 30175, &owner, Some(d.as_bytes()));
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(definition_lock)
+            .execute(&mut *tx)
+            .await?;
+        let definition_head: Option<(i64, Vec<u8>, Vec<u8>)> = sqlx::query_as("SELECT revision,event_id,content_sha256 FROM managed_agent_definition_heads WHERE community_id=$1 AND owner_pubkey=$2 AND definition_d=$3 FOR UPDATE")
+            .bind(community.as_uuid()).bind(&owner).bind(d).fetch_optional(&mut *tx).await?;
+        let revision = match definition_head {
+            Some((revision, old_id, old_hash)) if old_id == *definition_id && old_hash == *hash => {
+                revision
+            }
+            Some((revision, _, _)) => revision + 1,
+            None => 1,
+        };
+        if request.expected_definition_revision != Some(revision as u64) {
+            return Err(DbError::ManagedAgentConflict(format!(
+                "expected definition revision does not match current definition head; expected {revision}"
+            )));
+        }
+        sqlx::query("INSERT INTO managed_agent_definition_heads (community_id,owner_pubkey,definition_d,revision,event_id,content_sha256) VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (community_id,owner_pubkey,definition_d) DO UPDATE SET revision=EXCLUDED.revision,event_id=EXCLUDED.event_id,content_sha256=EXCLUDED.content_sha256")
+            .bind(community.as_uuid()).bind(&owner).bind(d).bind(revision).bind(definition_id).bind(hash).execute(&mut *tx).await?;
+        Some(revision)
+    } else {
+        None
+    };
+
+    // Insert first, then atomically reactivate the exact submitted IDs while
+    // retiring every competing live row at each coordinate.
+    let mut stored = Vec::with_capacity(1 + projections.len());
+    for event in projections
+        .iter()
+        .chain(std::iter::once(&request.private_event))
+    {
+        let was_deleted: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2 AND deleted_at IS NOT NULL)")
+            .bind(community.as_uuid()).bind(event.id.to_bytes()).fetch_one(&mut *tx).await?;
+        let (row, inserted) = crate::event::insert_event_with_thread_metadata_tx(
+            &mut tx, community, event, None, None,
+        )
+        .await?;
+        let d = exact_d(event)?;
+        let activated = sqlx::query("UPDATE events SET deleted_at=CASE WHEN id=$5 THEN NULL ELSE now() END WHERE community_id=$1 AND kind=$2 AND pubkey=$3 AND d_tag=$4 AND (deleted_at IS NULL OR id=$5)")
+            .bind(community.as_uuid()).bind(event.kind.as_u16() as i32).bind(event.pubkey.to_bytes()).bind(d).bind(event.id.to_bytes()).execute(&mut *tx).await?;
+        if activated.rows_affected() == 0 {
+            return Err(DbError::ManagedAgentConflict(
+                "submitted event ID is not stored at its signed coordinate".into(),
+            ));
+        }
+        if inserted || was_deleted {
+            stored.push(row);
+        }
+    }
+
+    // Tombstones carry no projections, but their prior public instance must not
+    // remain discoverable as an active managed agent. Definitions are shared
+    // and revision-pinned, so deleting one agent must not retire that row.
+    if envelope.state == State::Deleted {
+        sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1 AND kind=30177 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL")
+            .bind(community.as_uuid())
+            .bind(&owner)
+            .bind(envelope.agent_pubkey.to_hex())
+            .execute(&mut *tx)
+            .await?;
+    }
+
+    let state = match envelope.state {
+        State::Active => "active",
+        State::Deleted => "deleted",
+    };
+    sqlx::query("INSERT INTO managed_agent_heads (community_id,owner_pubkey,agent_pubkey,generation,event_id,state,definition_d,definition_revision,definition_event_id,definition_content_sha256,instance_event_id,instance_content_sha256) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT (community_id,owner_pubkey,agent_pubkey) DO UPDATE SET generation=EXCLUDED.generation,event_id=EXCLUDED.event_id,state=EXCLUDED.state,definition_d=EXCLUDED.definition_d,definition_revision=EXCLUDED.definition_revision,definition_event_id=EXCLUDED.definition_event_id,definition_content_sha256=EXCLUDED.definition_content_sha256,instance_event_id=EXCLUDED.instance_event_id,instance_content_sha256=EXCLUDED.instance_content_sha256,updated_at=now()")
+        .bind(community.as_uuid()).bind(&owner).bind(&agent).bind(envelope.generation as i64).bind(&event_id).bind(state).bind(&definition_d).bind(definition_revision).bind(&definition_event_id).bind(&definition_hash).bind(&instance_event_id).bind(&instance_hash).execute(&mut *tx).await?;
+    sqlx::query("INSERT INTO managed_agent_revisions (community_id,owner_pubkey,agent_pubkey,generation,event_id,previous_event_id,state,definition_d,definition_revision,definition_event_id,definition_content_sha256,instance_event_id,instance_content_sha256,private_event,definition_event,instance_event) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)")
+        .bind(community.as_uuid()).bind(&owner).bind(&agent).bind(envelope.generation as i64).bind(&event_id).bind(&previous).bind(state).bind(&definition_d).bind(definition_revision).bind(&definition_event_id).bind(&definition_hash).bind(&instance_event_id).bind(&instance_hash)
+        .bind(serde_json::to_value(&request.private_event)?).bind(request.definition_event.as_ref().map(serde_json::to_value).transpose()?).bind(request.instance_event.as_ref().map(serde_json::to_value).transpose()?).execute(&mut *tx).await?;
+    let snapshot = read_snapshot_tx(
+        &mut tx,
+        community,
+        &owner,
+        &agent,
+        envelope.generation,
+        true,
+        stored,
+    )
+    .await?;
+    tx.commit().await?;
+    Ok(snapshot)
+}
+
+/// Read a head from the writer pool, never a lagging replica.
+pub async fn read_head(
+    pool: &PgPool,
+    community: CommunityId,
+    owner: &[u8],
+    agent: &[u8],
+) -> Result<Option<ManagedAgentHead>> {
+    let row = sqlx::query("SELECT owner_pubkey,agent_pubkey,generation,event_id,state FROM managed_agent_heads WHERE community_id=$1 AND owner_pubkey=$2 AND agent_pubkey=$3")
+        .bind(community.as_uuid()).bind(owner).bind(agent).fetch_optional(pool).await?;
+    row.map(|r| {
+        Ok(ManagedAgentHead {
+            owner_pubkey: r.try_get("owner_pubkey")?,
+            agent_pubkey: r.try_get("agent_pubkey")?,
+            generation: r.try_get::<i64, _>("generation")? as u64,
+            event_id: r.try_get("event_id")?,
+            active: r.try_get::<String, _>("state")? == "active",
+        })
+    })
+    .transpose()
+}
+
+/// Whether an ordinary projection write targets PMA-authoritative state.
+pub async fn projection_is_authoritative(
+    pool: &PgPool,
+    community: CommunityId,
+    event: &Event,
+    d_tag: &str,
+) -> Result<bool> {
+    let kind = event.kind.as_u16() as u32;
+    let exists = match kind {
+        30177 => sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM managed_agent_heads WHERE community_id=$1 AND owner_pubkey=$2 AND agent_pubkey=$3)").bind(community.as_uuid()).bind(event.pubkey.to_bytes()).bind(hex::decode(d_tag).map_err(|_| invalid("invalid managed-agent d tag"))?).fetch_one(pool).await?,
+        30175 => sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM managed_agent_heads WHERE community_id=$1 AND owner_pubkey=$2 AND definition_d=$3)").bind(community.as_uuid()).bind(event.pubkey.to_bytes()).bind(d_tag).fetch_one(pool).await?,
+        _ => false,
+    };
+    Ok(exists)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nostr::{EventBuilder, Keys, Kind, Tag};
+
+    fn projection(keys: &Keys, kind: u16, d: &str) -> Event {
+        EventBuilder::new(Kind::Custom(kind), "projection")
+            .tag(Tag::parse(["d", d]).expect("d tag"))
+            .sign_with_keys(keys)
+            .expect("sign projection")
+    }
+
+    #[test]
+    fn active_projection_validation_binds_owner_kind_signature_and_agent_coordinate() {
+        let owner = Keys::generate();
+        let agent = Keys::generate();
+        let definition = projection(&owner, 30175, "definition");
+        let instance = projection(&owner, 30177, &agent.public_key().to_hex());
+        assert_eq!(
+            validate_projection(&definition, &owner.public_key(), 30175).unwrap(),
+            "definition"
+        );
+        assert_eq!(
+            validate_projection(&instance, &owner.public_key(), 30177).unwrap(),
+            agent.public_key().to_hex()
+        );
+        assert!(validate_projection(&instance, &owner.public_key(), 30175).is_err());
+        assert!(validate_projection(&instance, &Keys::generate().public_key(), 30177).is_err());
+    }
+
+    fn private_head(
+        owner: &Keys,
+        agent: &Keys,
+        generation: u64,
+        previous: Option<&nostr::EventId>,
+        state: &str,
+    ) -> Event {
+        let mut tags = vec![
+            Tag::parse(["d", agent.public_key().to_hex().as_str()]).unwrap(),
+            Tag::parse(["g", generation.to_string().as_str()]).unwrap(),
+            Tag::parse(["state", state]).unwrap(),
+        ];
+        if let Some(previous) = previous {
+            tags.push(Tag::parse(["prev", previous.to_hex().as_str()]).unwrap());
+        }
+        EventBuilder::new(Kind::Custom(30179), "encrypted-ciphertext")
+            .tags(tags)
+            .sign_with_keys(owner)
+            .unwrap()
+    }
+
+    async fn postgres_fixture() -> Option<(PgPool, CommunityId)> {
+        let url = std::env::var("BUZZ_TEST_DATABASE_URL")
+            .or_else(|_| std::env::var("DATABASE_URL"))
+            .ok()?;
+        let pool = PgPool::connect(&url).await.ok()?;
+        let id = uuid::Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id,host) VALUES ($1,$2)")
+            .bind(id)
+            .bind(format!("pma-test-{}.example", id.simple()))
+            .execute(&pool)
+            .await
+            .ok()?;
+        Some((pool, CommunityId::from_uuid(id)))
+    }
+
+    #[tokio::test]
+    #[ignore = "requires migrated Postgres via BUZZ_TEST_DATABASE_URL"]
+    async fn aggregate_readback_retry_reactivation_retirement_and_shared_pinning() {
+        let (pool, community) = postgres_fixture().await.expect("Postgres fixture");
+        let owner = Keys::generate();
+        let first_agent = Keys::generate();
+        let second_agent = Keys::generate();
+        let definition_v1 = projection(&owner, 30175, "shared");
+        let first_instance = projection(&owner, 30177, &first_agent.public_key().to_hex());
+        let first_private = private_head(&owner, &first_agent, 1, None, "active");
+        let first = AggregateRequest {
+            private_event: first_private.clone(),
+            definition_event: Some(definition_v1.clone()),
+            instance_event: Some(first_instance.clone()),
+            expected_definition_revision: Some(1),
+        };
+        let committed = commit_aggregate(&pool, community, &first).await.unwrap();
+        assert!(committed.inserted);
+        assert_eq!(committed.definition_revision, Some(1));
+        assert_eq!(committed.private_event, first_private);
+        assert_eq!(committed.definition_event, Some(definition_v1.clone()));
+
+        let retried = commit_aggregate(&pool, community, &first).await.unwrap();
+        assert!(!retried.inserted);
+        assert_eq!(retried.private_event, committed.private_event);
+        assert_eq!(retried.definition_event, committed.definition_event);
+        assert_eq!(retried.instance_event, committed.instance_event);
+        let changed_binding = AggregateRequest {
+            private_event: first.private_event.clone(),
+            definition_event: Some(projection(&owner, 30175, "different-definition")),
+            instance_event: first.instance_event.clone(),
+            expected_definition_revision: first.expected_definition_revision,
+        };
+        assert!(matches!(
+            commit_aggregate(&pool, community, &changed_binding).await,
+            Err(DbError::ManagedAgentConflict(message))
+                if message.contains("changed projection bindings")
+        ));
+
+        // A previously soft-deleted exact projection is made live again.
+        sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1 AND id=$2")
+            .bind(community.as_uuid())
+            .bind(definition_v1.id.to_bytes())
+            .execute(&pool)
+            .await
+            .unwrap();
+        let first_v2 = AggregateRequest {
+            private_event: private_head(&owner, &first_agent, 2, Some(&first_private.id), "active"),
+            definition_event: Some(definition_v1.clone()),
+            instance_event: Some(first_instance),
+            expected_definition_revision: Some(1),
+        };
+        commit_aggregate(&pool, community, &first_v2).await.unwrap();
+        let definition_live: bool = sqlx::query_scalar(
+            "SELECT deleted_at IS NULL FROM events WHERE community_id=$1 AND id=$2",
+        )
+        .bind(community.as_uuid())
+        .bind(definition_v1.id.to_bytes())
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        assert!(definition_live);
+        let first_private_live: bool = sqlx::query_scalar(
+            "SELECT deleted_at IS NULL FROM events WHERE community_id=$1 AND id=$2",
+        )
+        .bind(community.as_uuid())
+        .bind(first_private.id.to_bytes())
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        assert!(!first_private_live);
+
+        // Advancing a shared definition for another agent does not invalidate
+        // the first agent's pinned immutable revision bytes.
+        let definition_v2 = EventBuilder::new(Kind::Custom(30175), "projection-v2")
+            .tag(Tag::parse(["d", "shared"]).unwrap())
+            .sign_with_keys(&owner)
+            .unwrap();
+        let second = AggregateRequest {
+            private_event: private_head(&owner, &second_agent, 1, None, "active"),
+            definition_event: Some(definition_v2.clone()),
+            instance_event: Some(projection(
+                &owner,
+                30177,
+                &second_agent.public_key().to_hex(),
+            )),
+            expected_definition_revision: Some(2),
+        };
+        let second_commit = commit_aggregate(&pool, community, &second).await.unwrap();
+        assert_eq!(second_commit.definition_revision, Some(2));
+        let pinned: serde_json::Value = sqlx::query_scalar("SELECT definition_event FROM managed_agent_revisions WHERE community_id=$1 AND owner_pubkey=$2 AND agent_pubkey=$3 AND generation=2")
+            .bind(community.as_uuid()).bind(owner.public_key().to_bytes()).bind(first_agent.public_key().to_bytes()).fetch_one(&pool).await.unwrap();
+        assert_eq!(
+            decode_event(pinned, "pinned definition").unwrap(),
+            definition_v1
+        );
+
+        // Two distinct candidates that both predict current+1 cannot both
+        // commit. The loser observes the advanced definition under the lock and
+        // gets a conflict before its own managed-agent generation is written.
+        let race_agent_a = Keys::generate();
+        let race_agent_b = Keys::generate();
+        let race_a = AggregateRequest {
+            private_event: private_head(&owner, &race_agent_a, 1, None, "active"),
+            definition_event: Some(
+                EventBuilder::new(Kind::Custom(30175), "race-a")
+                    .tag(Tag::parse(["d", "shared"]).unwrap())
+                    .sign_with_keys(&owner)
+                    .unwrap(),
+            ),
+            instance_event: Some(projection(
+                &owner,
+                30177,
+                &race_agent_a.public_key().to_hex(),
+            )),
+            expected_definition_revision: Some(3),
+        };
+        let race_b = AggregateRequest {
+            private_event: private_head(&owner, &race_agent_b, 1, None, "active"),
+            definition_event: Some(
+                EventBuilder::new(Kind::Custom(30175), "race-b")
+                    .tag(Tag::parse(["d", "shared"]).unwrap())
+                    .sign_with_keys(&owner)
+                    .unwrap(),
+            ),
+            instance_event: Some(projection(
+                &owner,
+                30177,
+                &race_agent_b.public_key().to_hex(),
+            )),
+            expected_definition_revision: Some(3),
+        };
+        let (result_a, result_b) = tokio::join!(
+            commit_aggregate(&pool, community, &race_a),
+            commit_aggregate(&pool, community, &race_b)
+        );
+        assert_eq!(
+            usize::from(result_a.is_ok()) + usize::from(result_b.is_ok()),
+            1
+        );
+        let conflict = if let Err(error) = result_a {
+            error
+        } else {
+            result_b.unwrap_err()
+        };
+        assert!(matches!(
+            conflict,
+            DbError::ManagedAgentConflict(message)
+                if message.contains("expected definition revision")
+        ));
+        let race_heads: i64 = sqlx::query_scalar("SELECT count(*) FROM managed_agent_heads WHERE community_id=$1 AND owner_pubkey=$2 AND agent_pubkey IN ($3,$4)")
+            .bind(community.as_uuid()).bind(owner.public_key().to_bytes()).bind(race_agent_a.public_key().to_bytes()).bind(race_agent_b.public_key().to_bytes()).fetch_one(&pool).await.unwrap();
+        assert_eq!(race_heads, 1);
+    }
+
+    #[test]
+    fn projection_rejects_ambiguous_d_tags() {
+        let owner = Keys::generate();
+        let event = EventBuilder::new(Kind::Custom(30175), "projection")
+            .tags([
+                Tag::parse(["d", "one"]).unwrap(),
+                Tag::parse(["d", "two"]).unwrap(),
+            ])
+            .sign_with_keys(&owner)
+            .unwrap();
+        assert!(validate_projection(&event, &owner.public_key(), 30175).is_err());
+    }
+}

crates/buzz-db/src/migration.rs

@@ -943,12 +943,17 @@ mod tests {
 
         // Private managed-agent ciphertext is author-only and must remain
         // unsearchable on brownfield databases without changing 0001's sqlx
-        // checksum. The additive wrapper preserves the existing expression.
+        // checksum. Migration 0028 also installs the relay authority tables.
         assert_eq!(migrations[27].version, 28);
         let private_managed_agent = migrations[27].sql.as_str();
+        assert!(private_managed_agent.contains("CREATE TABLE managed_agent_heads"));
+        assert!(private_managed_agent.contains("CREATE TABLE managed_agent_revisions"));
         assert!(private_managed_agent.contains("kind = 30179"));
         assert!(private_managed_agent.contains("pg_get_expr"));
         assert!(private_managed_agent.contains("search_tsv"));
+        assert!(desired_schema.contains("CREATE TABLE managed_agent_heads"));
+        assert!(desired_schema.contains("CREATE TABLE managed_agent_revisions"));
+        assert!(desired_schema.contains("30179"));
         assert!(!migrations[0].sql.as_str().contains("30179"));
     }
 

crates/buzz-relay/src/api/bridge.rs

@@ -21,7 +21,7 @@ use crate::state::AppState;
 
 use super::{api_error, internal_error, not_found};
 
-async fn enforce_http_admission(
+pub(crate) async fn enforce_http_admission(
     state: &AppState,
     tenant: &TenantContext,
     pubkey: &nostr::PublicKey,

crates/buzz-relay/src/api/managed_agents.rs

@@ -0,0 +1,126 @@
+//! Dedicated NIP-98 submission boundary for NIP-PMA aggregates.
+
+use std::sync::Arc;
+
+use axum::{
+    extract::State,
+    http::{HeaderMap, StatusCode},
+    response::Json,
+};
+use serde::Deserialize;
+use serde_json::Value;
+
+use super::{api_error, internal_error};
+use crate::state::AppState;
+
+/// Maximum aggregate request body. The encrypted payload and two signed
+/// projections fit below this cap while preventing unbounded JSON allocation.
+const MAX_AGGREGATE_BODY_BYTES: usize = 256 * 1024;
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct AggregateBody {
+    private_event: nostr::Event,
+    #[serde(default)]
+    definition_event: Option<nostr::Event>,
+    #[serde(default)]
+    instance_event: Option<nostr::Event>,
+    #[serde(default)]
+    expected_definition_revision: Option<u64>,
+}
+
+/// Atomically install an encrypted PMA head and exact public projections.
+pub async fn submit_aggregate(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    body: axum::body::Bytes,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    if body.len() > MAX_AGGREGATE_BODY_BYTES {
+        return Err(api_error(
+            StatusCode::PAYLOAD_TOO_LARGE,
+            "aggregate body too large",
+        ));
+    }
+    let raw_host = headers
+        .get(axum::http::header::HOST)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("");
+    let tenant = crate::tenant::bind_community(&state.db, raw_host)
+        .await
+        .map_err(|_| {
+            api_error(
+                StatusCode::NOT_FOUND,
+                "relay: no community is configured for this host",
+            )
+        })?;
+    let url = super::bridge::nip98_expected_url(
+        &state.config.relay_url,
+        &tenant,
+        "/api/managed-agents/aggregate",
+    );
+    let (owner, replay_id) = super::bridge::verify_bridge_auth_with_options(
+        &headers,
+        "POST",
+        &url,
+        Some(&body),
+        state.config.require_auth_token,
+        true,
+    )?;
+    super::bridge::enforce_http_admission(&state, &tenant, &owner).await?;
+    super::bridge::check_nip98_replay(&state, &tenant, replay_id).await?;
+    super::relay_members::enforce_relay_membership(
+        &state,
+        tenant.community(),
+        &owner.to_bytes(),
+        None,
+    )
+    .await?;
+    let parsed: AggregateBody = serde_json::from_slice(&body)
+        .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid aggregate JSON"))?;
+    if parsed.private_event.pubkey != owner {
+        return Err(api_error(
+            StatusCode::FORBIDDEN,
+            "aggregate author must be authenticated owner",
+        ));
+    }
+    let request = buzz_db::managed_agent::AggregateRequest {
+        private_event: parsed.private_event,
+        definition_event: parsed.definition_event,
+        instance_event: parsed.instance_event,
+        expected_definition_revision: parsed.expected_definition_revision,
+    };
+    let commit = state
+        .db
+        .commit_managed_agent_aggregate(tenant.community(), &request)
+        .await
+        .map_err(|e| match e {
+            buzz_db::DbError::InvalidData(message) => api_error(StatusCode::BAD_REQUEST, &message),
+            buzz_db::DbError::ManagedAgentConflict(message) => {
+                api_error(StatusCode::CONFLICT, &message)
+            }
+            _ => internal_error(&format!("managed-agent aggregate commit failed: {e}")),
+        })?;
+    let owner_hex = owner.to_hex();
+    for stored in &commit.events {
+        crate::handlers::event::dispatch_persistent_event(
+            &tenant,
+            &state,
+            stored,
+            buzz_core::kind::event_kind_u32(&stored.event),
+            &owner_hex,
+            None,
+        )
+        .await;
+    }
+    Ok(Json(serde_json::json!({
+        "event_id": hex::encode(&commit.head.event_id),
+        "generation": commit.head.generation,
+        "state": if commit.head.active { "active" } else { "deleted" },
+        "accepted": true,
+        "inserted": commit.inserted,
+        "definition_revision": commit.definition_revision,
+        "private_event": commit.private_event,
+        "definition_event": commit.definition_event,
+        "instance_event": commit.instance_event
+    })))
+}

crates/buzz-relay/src/api/mod.rs

@@ -5,6 +5,7 @@ pub mod bridge;
 pub mod events;
 pub mod git;
 pub mod invites;
+pub mod managed_agents;
 pub mod media;
 pub mod mesh_demo;
 pub mod nip05;

crates/buzz-relay/src/handlers/ingest.rs

@@ -2759,6 +2759,20 @@ async fn ingest_event_inner(
         });
     }
 
+    if matches!(kind_u32, KIND_PERSONA | KIND_MANAGED_AGENT) {
+        let d_tag = buzz_db::event::extract_d_tag(&event).unwrap_or_default();
+        if state
+            .db
+            .managed_agent_projection_is_authoritative(tenant.community(), &event, &d_tag)
+            .await
+            .map_err(|e| IngestError::Internal(format!("error: PMA authority check failed: {e}")))?
+        {
+            return Err(IngestError::Rejected(
+                "restricted: projection is controlled by a private managed-agent aggregate".into(),
+            ));
+        }
+    }
+
     let (stored_event, was_inserted) = if buzz_core::kind::is_replaceable(kind_u32) {
         // NIP-16 replaceable event — atomic replace with stale-write protection.
         // channel_id is None for global kinds (0, 1, 3) due to step 5b above.

crates/buzz-relay/src/nip11.rs

@@ -162,7 +162,10 @@ impl RelayInfo {
             pubkey: None,
             contact: None,
             supported_nips,
-            supported_extensions: Some(vec!["nip-er".to_string()]),
+            supported_extensions: Some(vec![
+                "nip-er".to_string(),
+                "nip-pma-aggregate-v1".to_string(),
+            ]),
             push: None,
             software: "https://github.com/block/buzz".to_string(),
             version: env!("CARGO_PKG_VERSION").to_string(),
@@ -389,6 +392,17 @@ mod tests {
         );
     }
 
+    #[test]
+    fn build_advertises_private_managed_agent_aggregate_v1() {
+        let info = RelayInfo::build(None, None, false, DEFAULT_MAX_FRAME_BYTES, None);
+        assert!(info
+            .supported_extensions
+            .as_ref()
+            .is_some_and(|extensions| extensions
+                .iter()
+                .any(|extension| extension == "nip-pma-aggregate-v1")));
+    }
+
     #[test]
     fn build_advertises_buzz_repository_url() {
         let info = RelayInfo::build(None, None, false, DEFAULT_MAX_FRAME_BYTES, None);

crates/buzz-relay/src/router.rs

@@ -70,6 +70,10 @@ pub fn build_router(state: Arc<AppState>) -> Router {
         .route("/_readiness", get(readiness_handler))
         // Nostr HTTP bridge (NIP-98 auth)
         .route("/events", post(api::bridge::submit_event))
+        .route(
+            "/api/managed-agents/aggregate",
+            post(api::managed_agents::submit_aggregate),
+        )
         .route("/query", post(api::bridge::query_events))
         .route("/count", post(api::bridge::count_events))
         .route(

crates/buzz-search/tests/fts_integration.rs

@@ -28,6 +28,8 @@ const MIGRATION_0007_SQL: &str = include_str!("../../../migrations/0007_nip_rs_r
 const MIGRATION_0008_SQL: &str =
     include_str!("../../../migrations/0008_fresh_install_search_allowlist.sql");
 const MIGRATION_0014_SQL: &str = include_str!("../../../migrations/0014_push_lease_fts.sql");
+const MIGRATION_0028_SQL: &str =
+    include_str!("../../../migrations/0028_private_managed_agent_authority.sql");
 
 async fn setup() -> (PgPool, String) {
     let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string());
@@ -81,6 +83,9 @@ async fn setup() -> (PgPool, String) {
     pool.execute(MIGRATION_0014_SQL)
         .await
         .expect("apply 0014 migration");
+    pool.execute(MIGRATION_0028_SQL)
+        .await
+        .expect("apply 0028 migration");
     (pool, schema)
 }
 
@@ -1229,6 +1234,7 @@ async fn excluded_kinds_are_storage_level_unsearchable() {
     // Negative (load-bearing): each excluded kind MUST NOT surface.
     for forbidden in [
         1059,
+        30179,
         30300,
         30622,
         KIND_MEMBER_ADDED_NOTIFICATION as i32,

migrations/0028_private_managed_agent_foundation.sql

@@ -1,3 +1,69 @@
+-- Relay-canonical authority for NIP-PMA kind:30179. The head retains the
+-- generation floor after deletion; immutable revisions provide the audit/CAS
+-- chain. Public projections are bound by exact signed event IDs and hashes.
+CREATE TABLE managed_agent_definition_heads (
+    community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE,
+    owner_pubkey BYTEA NOT NULL CHECK (length(owner_pubkey) = 32),
+    definition_d TEXT NOT NULL,
+    revision BIGINT NOT NULL CHECK (revision > 0),
+    event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
+    content_sha256 BYTEA NOT NULL CHECK (length(content_sha256) = 32),
+    PRIMARY KEY (community_id, owner_pubkey, definition_d)
+);
+
+CREATE TABLE managed_agent_heads (
+    community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE,
+    owner_pubkey BYTEA NOT NULL CHECK (length(owner_pubkey) = 32),
+    agent_pubkey BYTEA NOT NULL CHECK (length(agent_pubkey) = 32),
+    generation BIGINT NOT NULL CHECK (generation > 0 AND generation <= 9007199254740991),
+    event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
+    state TEXT NOT NULL CHECK (state IN ('active', 'deleted')),
+    definition_d TEXT,
+    definition_revision BIGINT,
+    definition_event_id BYTEA,
+    definition_content_sha256 BYTEA,
+    instance_event_id BYTEA,
+    instance_content_sha256 BYTEA,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (community_id, owner_pubkey, agent_pubkey),
+    UNIQUE (community_id, event_id),
+    CHECK ((state = 'active') =
+      (definition_d IS NOT NULL AND definition_revision IS NOT NULL AND
+       definition_event_id IS NOT NULL AND definition_content_sha256 IS NOT NULL AND
+       instance_event_id IS NOT NULL AND instance_content_sha256 IS NOT NULL)),
+    CHECK (definition_event_id IS NULL OR length(definition_event_id) = 32),
+    CHECK (definition_content_sha256 IS NULL OR length(definition_content_sha256) = 32),
+    CHECK (instance_event_id IS NULL OR length(instance_event_id) = 32),
+    CHECK (instance_content_sha256 IS NULL OR length(instance_content_sha256) = 32)
+);
+CREATE INDEX managed_agent_heads_definition_binding
+    ON managed_agent_heads (community_id, owner_pubkey, definition_d);
+
+CREATE TABLE managed_agent_revisions (
+    community_id UUID NOT NULL,
+    owner_pubkey BYTEA NOT NULL,
+    agent_pubkey BYTEA NOT NULL,
+    generation BIGINT NOT NULL CHECK (generation > 0 AND generation <= 9007199254740991),
+    event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
+    previous_event_id BYTEA CHECK (previous_event_id IS NULL OR length(previous_event_id) = 32),
+    state TEXT NOT NULL CHECK (state IN ('active', 'deleted')),
+    definition_d TEXT,
+    definition_revision BIGINT,
+    definition_event_id BYTEA CHECK (definition_event_id IS NULL OR length(definition_event_id) = 32),
+    definition_content_sha256 BYTEA CHECK (definition_content_sha256 IS NULL OR length(definition_content_sha256) = 32),
+    instance_event_id BYTEA CHECK (instance_event_id IS NULL OR length(instance_event_id) = 32),
+    instance_content_sha256 BYTEA CHECK (instance_content_sha256 IS NULL OR length(instance_content_sha256) = 32),
+    private_event JSONB NOT NULL,
+    definition_event JSONB,
+    instance_event JSONB,
+    committed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (community_id, owner_pubkey, agent_pubkey, generation),
+    UNIQUE (community_id, event_id),
+    FOREIGN KEY (community_id, owner_pubkey, agent_pubkey)
+      REFERENCES managed_agent_heads (community_id, owner_pubkey, agent_pubkey)
+      DEFERRABLE INITIALLY DEFERRED
+);
+
 -- NIP-PMA kind:30179 contains owner-encrypted runnable agent configuration and
 -- is author-only. Preserve every installation's current FTS expression while
 -- making existing and future private aggregate rows storage-level unsearchable.

schema/schema.sql

@@ -218,7 +218,7 @@ CREATE TABLE events (
     -- never matches `@@`.
     -- Keep in sync with migrations (final state: 0001 + 0005 + 0009).
     search_tsv  TSVECTOR GENERATED ALWAYS AS (
-        CASE WHEN kind IN (1059, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector
+        CASE WHEN kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector
              ELSE to_tsvector('simple', content)
         END
     ) STORED,
@@ -275,6 +275,57 @@ CREATE INDEX idx_events_not_before ON events (community_id, not_before)
 -- EXPLAIN before its work lands (Quinn option A; Max's index-spelling caveat).
 CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv);
 
+-- ── Private managed-agent authority (NIP-PMA) ────────────────────────────────
+CREATE TABLE managed_agent_definition_heads (
+    community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE,
+    owner_pubkey BYTEA NOT NULL CHECK (length(owner_pubkey) = 32),
+    definition_d TEXT NOT NULL,
+    revision BIGINT NOT NULL CHECK (revision > 0),
+    event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
+    content_sha256 BYTEA NOT NULL CHECK (length(content_sha256) = 32),
+    PRIMARY KEY (community_id, owner_pubkey, definition_d)
+);
+
+CREATE TABLE managed_agent_heads (
+    community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE,
+    owner_pubkey BYTEA NOT NULL CHECK (length(owner_pubkey) = 32),
+    agent_pubkey BYTEA NOT NULL CHECK (length(agent_pubkey) = 32),
+    generation BIGINT NOT NULL CHECK (generation > 0 AND generation <= 9007199254740991),
+    event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
+    state TEXT NOT NULL CHECK (state IN ('active', 'deleted')),
+    definition_d TEXT, definition_revision BIGINT, definition_event_id BYTEA,
+    definition_content_sha256 BYTEA, instance_event_id BYTEA, instance_content_sha256 BYTEA,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (community_id, owner_pubkey, agent_pubkey),
+    UNIQUE (community_id, event_id),
+    CHECK ((state = 'active') = (definition_d IS NOT NULL AND definition_revision IS NOT NULL AND definition_event_id IS NOT NULL AND definition_content_sha256 IS NOT NULL AND instance_event_id IS NOT NULL AND instance_content_sha256 IS NOT NULL)),
+    CHECK (definition_event_id IS NULL OR length(definition_event_id) = 32),
+    CHECK (definition_content_sha256 IS NULL OR length(definition_content_sha256) = 32),
+    CHECK (instance_event_id IS NULL OR length(instance_event_id) = 32),
+    CHECK (instance_content_sha256 IS NULL OR length(instance_content_sha256) = 32)
+);
+CREATE INDEX managed_agent_heads_definition_binding ON managed_agent_heads (community_id, owner_pubkey, definition_d);
+CREATE TABLE managed_agent_revisions (
+    community_id UUID NOT NULL, owner_pubkey BYTEA NOT NULL, agent_pubkey BYTEA NOT NULL,
+    generation BIGINT NOT NULL CHECK (generation > 0 AND generation <= 9007199254740991),
+    event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
+    previous_event_id BYTEA CHECK (previous_event_id IS NULL OR length(previous_event_id) = 32),
+    state TEXT NOT NULL CHECK (state IN ('active', 'deleted')),
+    definition_d TEXT,
+    definition_revision BIGINT,
+    definition_event_id BYTEA CHECK (definition_event_id IS NULL OR length(definition_event_id) = 32),
+    definition_content_sha256 BYTEA CHECK (definition_content_sha256 IS NULL OR length(definition_content_sha256) = 32),
+    instance_event_id BYTEA CHECK (instance_event_id IS NULL OR length(instance_event_id) = 32),
+    instance_content_sha256 BYTEA CHECK (instance_content_sha256 IS NULL OR length(instance_content_sha256) = 32),
+    private_event JSONB NOT NULL,
+    definition_event JSONB,
+    instance_event JSONB,
+    committed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (community_id, owner_pubkey, agent_pubkey, generation),
+    UNIQUE (community_id, event_id),
+    FOREIGN KEY (community_id, owner_pubkey, agent_pubkey) REFERENCES managed_agent_heads (community_id, owner_pubkey, agent_pubkey) DEFERRABLE INITIALLY DEFERRED
+);
+
 -- ── Event mentions ────────────────────────────────────────────────────────────
 -- Conformance: "Channel-less global events and DMs" (#p fan-out). The join to
 -- events MUST carry the community tuple (e.community_id = m.community_id AND