feat(graph): rfc 0047 slice 4 — emitter + erasure (RFC0047.10–.11), RFC green

.github/workflows/ci.yml

@@ -384,8 +384,8 @@ jobs:
           set -euo pipefail
           fga model transform --file model.fga | diff - model.json
 
-  # RFC 0047 §5 (RFC0047.1–.9): the OpenFGA resolver, the layer-2
-  # visibility two-step and the MCP tool gate on the served binary against a real
+  # RFC 0047 §5 (RFC0047.1–.11): the OpenFGA resolver, the layer-2
+  # visibility two-step, the MCP tool gate, the graph emitter and erasure against a real
   # `openfga/openfga` container (testcontainers, image pinned by digest in
   # the tests) loaded with the in-tree model — the same posture as
   # `dex-oidc`. A required check (in `ci-success`'s `needs`).
@@ -407,6 +407,7 @@ jobs:
           --ignored --exact
           rfc0047_openfga::rfc0047_1_to_3_resolver_end_to_end
           rfc0047_visibility::rfc0047_4_to_9_visibility_end_to_end
+          rfc0047_emitter::rfc0047_10_11_emitter_and_erasure_end_to_end
 
   # Emission-time semconv conformance: boot the real `ourios-server`,
   # point its OTLP export (metrics + the dogfooded logs signal) at

crates/ourios-core/src/alias.rs

@@ -402,6 +402,7 @@ impl AliasMap {
             | AuditPayload::Compaction { .. }
             | AuditPayload::RecordQuarantined { .. }
             | AuditPayload::IngestDenied { .. }
+            | AuditPayload::ConversationErased { .. }
             | AuditPayload::Unknown { .. } => {}
         }
     }

crates/ourios-core/src/audit.rs

@@ -245,6 +245,21 @@ pub enum AuditPayload {
         /// The rejecting token's audit/metric label (RFC 0026 §3.4).
         token_name: String,
     },
+    /// A conversation was erased from the tenant (RFC 0047 §3.6): its rows
+    /// dropped by the compaction rewrite of every partition, then its
+    /// graph tuples deleted — in that order; the event is written after
+    /// both. System-scoped like [`Self::Compaction`]; the event's
+    /// `tenant_id` is the tenant the conversation lived in.
+    ConversationErased {
+        /// The raw conversation id (the promoted-column value).
+        conversation_id: String,
+        /// Partitions rewritten by the erasure pass.
+        partitions_rewritten: u64,
+        /// Rows dropped across those rewrites.
+        rows_dropped: u64,
+        /// Graph tuples deleted for the conversation object.
+        tuples_deleted: u64,
+    },
 }
 
 /// Stable on-disk `event_kind` ordinals (RFC 0005 §3.7 mapping).
@@ -272,6 +287,9 @@ pub const EVENT_KIND_RECORD_QUARANTINED: u8 = 7;
 /// `ingest_denied` — an authenticated cross-tenant write attempt was
 /// rejected pre-WAL (RFC 0026 §3.2).
 pub const EVENT_KIND_INGEST_DENIED: u8 = 8;
+/// `conversation_erased` — a conversation's rows were dropped by the
+/// compaction rewrite and its graph tuples deleted (RFC 0047 §3.6).
+pub const EVENT_KIND_CONVERSATION_ERASED: u8 = 9;
 
 /// Canonical `event_type` strings paired with the ordinals above
 /// (RFC 0005 §3.7 / RFC 0001 §6.4 / RFC 0009 §3.6).
@@ -293,6 +311,8 @@ pub const EVENT_TYPE_TEMPLATE_CREATED: &str = "template_created";
 pub const EVENT_TYPE_RECORD_QUARANTINED: &str = "record_quarantined";
 /// The string form of [`EVENT_KIND_INGEST_DENIED`].
 pub const EVENT_TYPE_INGEST_DENIED: &str = "ingest_denied";
+/// The string form of [`EVENT_KIND_CONVERSATION_ERASED`].
+pub const EVENT_TYPE_CONVERSATION_ERASED: &str = "conversation_erased";
 
 /// The `template_version` a leaf is born at (RFC 0017 §3.1). The
 /// [`TemplateChange::Created`] variant omits a version field — the invariant
@@ -320,6 +340,7 @@ impl AuditPayload {
             Self::Compaction { .. } => EVENT_KIND_COMPACTION,
             Self::RecordQuarantined { .. } => EVENT_KIND_RECORD_QUARANTINED,
             Self::IngestDenied { .. } => EVENT_KIND_INGEST_DENIED,
+            Self::ConversationErased { .. } => EVENT_KIND_CONVERSATION_ERASED,
             Self::Unknown { event_kind, .. } => *event_kind,
         }
     }
@@ -344,6 +365,7 @@ impl AuditPayload {
             Self::Compaction { .. } => EVENT_TYPE_COMPACTION,
             Self::RecordQuarantined { .. } => EVENT_TYPE_RECORD_QUARANTINED,
             Self::IngestDenied { .. } => EVENT_TYPE_INGEST_DENIED,
+            Self::ConversationErased { .. } => EVENT_TYPE_CONVERSATION_ERASED,
             Self::Unknown { event_type, .. } => event_type,
         }
     }
@@ -365,6 +387,7 @@ impl AuditPayload {
             | Self::Compaction { .. }
             | Self::RecordQuarantined { .. }
             | Self::IngestDenied { .. }
+            | Self::ConversationErased { .. }
             | Self::Unknown { .. } => false,
         }
     }

crates/ourios-core/src/auth/openfga/client.rs

@@ -39,7 +39,7 @@ const MAX_CACHE_ENTRIES: usize = 4096;
 const MAX_ERROR_BODY_BYTES: usize = 512;
 
 /// One relationship tuple / tuple key on the wire.
-#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
 pub struct TupleKey {
     /// `<type>:<id>` or a userset `<type>:<id>#<relation>`.
     pub user: String,
@@ -225,6 +225,42 @@ struct WriteBody<'a> {
     authorization_model_id: Option<&'a str>,
 }
 
+/// `Read` filter: every tuple on `object` (optionally one `relation`).
+#[derive(Serialize)]
+struct ReadTupleKey<'a> {
+    object: &'a str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    relation: Option<&'a str>,
+}
+
+#[derive(Serialize)]
+struct ReadBody<'a> {
+    tuple_key: ReadTupleKey<'a>,
+    page_size: u32,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    continuation_token: Option<&'a str>,
+}
+
+#[derive(Deserialize)]
+struct ReadResponse {
+    #[serde(default)]
+    tuples: Vec<ReadTuple>,
+    #[serde(default)]
+    continuation_token: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct ReadTuple {
+    key: TupleKey,
+}
+
+/// `Read` page size — `OpenFGA`'s maximum.
+const READ_PAGE_SIZE: u32 = 100;
+/// The bound on tuples one `read_by_object` returns: an object with more
+/// is not a conversation the emitter wrote (a few relations per
+/// participant), and an unbounded read is not fail-closed.
+const MAX_READ_TUPLES: usize = 10_000;
+
 /// The `OpenFGA` HTTP API over one store.
 #[derive(Clone)]
 pub struct OpenFgaClient {
@@ -393,6 +429,53 @@ impl OpenFgaClient {
         Ok(kept)
     }
 
+    /// `Read`: every tuple on `object` (RFC 0047 §3.6 — "no wildcard
+    /// delete exists": erasure reads the object's tuples, then deletes
+    /// them). Paginated to completion; bounded.
+    ///
+    /// # Errors
+    ///
+    /// [`OpenFgaError::Unavailable`] on transport/timeout/non-2xx;
+    /// [`OpenFgaError::BoundExceeded`] past the read bound.
+    pub async fn read_by_object(&self, object: &str) -> Result<Vec<TupleKey>, OpenFgaError> {
+        let mut tuples = Vec::new();
+        let mut continuation: Option<String> = None;
+        loop {
+            let body = ReadBody {
+                tuple_key: ReadTupleKey {
+                    object,
+                    relation: None,
+                },
+                page_size: READ_PAGE_SIZE,
+                continuation_token: continuation.as_deref(),
+            };
+            let response = self
+                .post("read", &body)?
+                .send()
+                .await
+                .map_err(|e| transport(&e))?;
+            let response = ok_status(response).await?;
+            let bytes = response
+                .bytes()
+                .await
+                .map_err(|e| OpenFgaError::Unavailable(format!("read read: {e}")))?;
+            let page: ReadResponse = serde_json::from_slice(&bytes)
+                .map_err(|e| OpenFgaError::Unavailable(format!("decode read: {e}")))?;
+            for tuple in page.tuples {
+                if tuples.len() >= MAX_READ_TUPLES {
+                    return Err(OpenFgaError::BoundExceeded {
+                        bound: MAX_READ_TUPLES,
+                    });
+                }
+                tuples.push(tuple.key);
+            }
+            match page.continuation_token {
+                Some(token) if !token.is_empty() => continuation = Some(token),
+                _ => return Ok(tuples),
+            }
+        }
+    }
+
     /// `Write`: add `writes` and remove `deletes` in one transactional,
     /// **idempotent** call (`on_duplicate` / `on_missing` = `ignore`, so a
     /// tuple already present or already gone is not an error). `OpenFGA`

crates/ourios-core/src/auth/openfga/mod.rs

@@ -215,6 +215,9 @@ pub const DEFAULT_LIST_TIMEOUT_MS: u64 = 2_000;
 pub const DEFAULT_SERVER_LIST_OBJECTS_DEADLINE_MS: u64 = 3_000;
 /// The `OpenFGA` object type of a conversation — the one bindable type in v1.
 pub const CONVERSATION_TYPE: &str = "conversation";
+/// The RFC 0027 MCP tools as graph objects (RFC 0047 §3.5):
+/// `tool:<T>/<name>`; the emitter writes their `parent` tuples per tenant.
+pub const MCP_TOOL_NAMES: [&str; 3] = ["query_logs", "list_templates", "template_drift"];
 
 impl fmt::Debug for OpenFgaConfig {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -602,13 +605,19 @@ pub struct TenantObjects {
 
 impl TenantObjects {
     /// The graph objects of `tenant`, or `None` when the tenant id cannot
-    /// form an object id.
+    /// form an object id — raw, or once its segment is encoded (the
+    /// encoding can only grow it; a tenant whose encoded segment plus the
+    /// `/` separator leaves no room for a conversation id has no
+    /// tenant-scoped objects).
     #[must_use]
     pub fn new(tenant: &str) -> Option<Self> {
         if !is_object_id(tenant) {
             return None;
         }
         let encoded = encode_tenant_segment(tenant);
+        if encoded.len() + 1 >= MAX_OBJECT_ID_BYTES {
+            return None;
+        }
         Some(Self {
             tenant_object: format!("{TENANT_TYPE}:{tenant}"),
             conversation_prefix: format!("{CONVERSATION_TYPE}:{encoded}/"),
@@ -635,6 +644,16 @@ impl TenantObjects {
         format!("{}{id}", self.conversation_prefix)
     }
 
+    /// Whether `conversation:<enc(T)>/<id>` is a valid object: `id` must be
+    /// an object id itself and the combined id half must fit `OpenFGA`'s
+    /// 256-byte limit. The emitter skips ids that do not.
+    #[must_use]
+    pub fn conversation_fits(&self, id: &str) -> bool {
+        is_object_id(id)
+            && self.conversation_prefix.len() - CONVERSATION_TYPE.len() - 1 + id.len()
+                <= MAX_OBJECT_ID_BYTES
+    }
+
     /// `tool:<enc(T)>/<name>`.
     #[must_use]
     pub fn tool(&self, name: &str) -> String {
@@ -908,6 +927,15 @@ mod tests {
         for bad in ["", "a b", "a:b", "a#b"] {
             assert!(TenantObjects::new(bad).is_none(), "{bad:?}");
         }
+        // The encoding may not push the segment past the object-id limit,
+        // and a conversation id must fit next to it.
+        let slashes = "/".repeat(90); // 90 raw bytes → 270 encoded
+        assert!(TenantObjects::new(&slashes).is_none());
+        let long = "x".repeat(200);
+        let t = TenantObjects::new(&long).expect("fits alone");
+        assert!(t.conversation_fits("c-1"));
+        assert!(!t.conversation_fits(&"y".repeat(60)), "201 + 60 > 256");
+        assert!(!t.conversation_fits("a b"));
     }
 
     /// The principal vocabulary renders exactly the model's type names.