From 345159374369601668a14ef78ec35e773b3eb2b3 Mon Sep 17 00:00:00 2001 From: Wes Date: Wed, 5 Aug 2026 10:01:40 -0600 Subject: [PATCH] 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 --- crates/buzz-db/src/error.rs | 4 + crates/buzz-db/src/event.rs | 2 +- crates/buzz-db/src/lib.rs | 21 + crates/buzz-db/src/managed_agent.rs | 668 ++++++++++++++++++ crates/buzz-db/src/migration.rs | 7 +- crates/buzz-relay/src/api/bridge.rs | 2 +- crates/buzz-relay/src/api/managed_agents.rs | 126 ++++ crates/buzz-relay/src/api/mod.rs | 1 + crates/buzz-relay/src/handlers/ingest.rs | 14 + crates/buzz-relay/src/nip11.rs | 16 +- crates/buzz-relay/src/router.rs | 4 + crates/buzz-search/tests/fts_integration.rs | 6 + .../0028_private_managed_agent_foundation.sql | 66 ++ schema/schema.sql | 53 +- 14 files changed, 985 insertions(+), 5 deletions(-) create mode 100644 crates/buzz-db/src/managed_agent.rs create mode 100644 crates/buzz-relay/src/api/managed_agents.rs diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index f8b8a2eb56..b3613dedc7 100644 --- a/crates/buzz-db/src/error.rs +++ b/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), diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 5998db6757..1adebc1a65 100644 --- a/crates/buzz-db/src/event.rs +++ b/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, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..0ad73d431c 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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::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 { + 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 diff --git a/crates/buzz-db/src/managed_agent.rs b/crates/buzz-db/src/managed_agent.rs new file mode 100644 index 0000000000..d606823b51 --- /dev/null +++ b/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, + /// Signed kind:30177 candidate, required for active heads. + pub instance_event: Option, + /// 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, +} + +/// 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, + /// Whether this call committed a new generation. + pub inserted: bool, + /// Transactionally confirmed definition revision for active heads. + pub definition_revision: Option, + /// Writer-consistent signed private head. + pub private_event: Event, + /// Writer-consistent signed definition binding. + pub definition_event: Option, + /// Writer-consistent signed instance binding. + pub instance_event: Option, +} + +/// Writer-consistent PMA authority head. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ManagedAgentHead { + /// Owner pubkey. + pub owner_pubkey: Vec, + /// Agent pubkey. + pub agent_pubkey: Vec, + /// Current generation floor. + pub generation: u64, + /// Exact current private event ID. + pub event_id: Vec, + /// Whether the current head is active. + pub active: bool, +} + +fn invalid(message: impl Into) -> 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 { + 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, Option)> { + 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 { + 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, +) -> Result { + 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::, _>("definition_event")? + .map(|value| decode_event(value, "definition event")) + .transpose()?; + let instance = row + .try_get::, _>("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::("generation")? as u64, + event_id: row.try_get("event_id")?, + active: row.try_get::("state")? == "active", + }, + events, + inserted, + definition_revision: row + .try_get::, _>("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 { + 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 { + 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) = ¤t { + let same_head = row.try_get::("generation")? as u64 == envelope.generation + && row.try_get::, _>("event_id")? == event_id; + if same_head { + let same_bindings = row.try_get::("state")? + == match envelope.state { + State::Active => "active", + State::Deleted => "deleted", + } + && row.try_get::>, _>("definition_event_id")? == definition_event_id + && row.try_get::>, _>("definition_content_sha256")? + == definition_hash + && row.try_get::>, _>("instance_event_id")? == instance_event_id + && row.try_get::>, _>("instance_content_sha256")? == instance_hash + && row + .try_get::, _>("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 ¤t { + None if envelope.generation == 1 && previous.is_none() => {} + Some(row) + if envelope.generation == row.try_get::("generation")? as u64 + 1 + && previous.as_deref() + == Some(row.try_get::, _>("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, Vec)> = 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> { + 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::("generation")? as u64, + event_id: r.try_get("event_id")?, + active: r.try_get::("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 { + 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()); + } +} diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index e9800c1dbd..4a9cef06be 100644 --- a/crates/buzz-db/src/migration.rs +++ b/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")); } diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index a118ff453f..99b9e5244d 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/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, diff --git a/crates/buzz-relay/src/api/managed_agents.rs b/crates/buzz-relay/src/api/managed_agents.rs new file mode 100644 index 0000000000..a4bc5d1414 --- /dev/null +++ b/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, + #[serde(default)] + instance_event: Option, + #[serde(default)] + expected_definition_revision: Option, +} + +/// Atomically install an encrypted PMA head and exact public projections. +pub async fn submit_aggregate( + State(state): State>, + headers: HeaderMap, + body: axum::body::Bytes, +) -> Result, (StatusCode, Json)> { + 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 + }))) +} diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index d9f829433b..eb04d8920a 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/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; diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index c711538284..2bce8d3fde 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/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. diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index 2575ddd7ba..0d6531045e 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/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); diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 400ed1dfe3..da4ef19977 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -70,6 +70,10 @@ pub fn build_router(state: Arc) -> 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( diff --git a/crates/buzz-search/tests/fts_integration.rs b/crates/buzz-search/tests/fts_integration.rs index 675d15db8b..d91c6602e9 100644 --- a/crates/buzz-search/tests/fts_integration.rs +++ b/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, diff --git a/migrations/0028_private_managed_agent_foundation.sql b/migrations/0028_private_managed_agent_foundation.sql index 66bdc4c595..1948029c44 100644 --- a/migrations/0028_private_managed_agent_foundation.sql +++ b/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. diff --git a/schema/schema.sql b/schema/schema.sql index 9f3449b066..b53aade689 100644 --- a/schema/schema.sql +++ b/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