feat(rfc-026): stream_ingest/stream_manage Cedar actions and read-only stream status

crates/omnigraph-cli/src/planes.rs

@@ -404,6 +404,7 @@ mod tests {
     use clap::Parser;
 
     use super::*;
+    use crate::cli::PolicyCommand;
 
     #[test]
     fn scope_flag_matrix_matches_capabilities() {
@@ -474,6 +475,45 @@ mod tests {
         );
     }
 
+    #[test]
+    fn policy_explain_accepts_stream_action_wire_names_and_kebab_aliases() {
+        let parse_action = |spelling: &str| {
+            let command = Cli::try_parse_from([
+                "omnigraph",
+                "policy",
+                "explain",
+                "--actor",
+                "act-operator",
+                "--action",
+                spelling,
+            ])
+            .unwrap()
+            .command;
+            match command {
+                Command::Policy {
+                    command: PolicyCommand::Explain { action, .. },
+                } => action,
+                other => panic!("expected policy explain command, got {other:?}"),
+            }
+        };
+
+        for (canonical, alias, expected) in [
+            (
+                "stream_ingest",
+                "stream-ingest",
+                omnigraph_policy::PolicyAction::StreamIngest,
+            ),
+            (
+                "stream_manage",
+                "stream-manage",
+                omnigraph_policy::PolicyAction::StreamManage,
+            ),
+        ] {
+            assert_eq!(parse_action(canonical), expected);
+            assert_eq!(parse_action(alias), expected);
+        }
+    }
+
     #[test]
     fn every_capability_describes_distinctly() {
         let phrases = [

crates/omnigraph-policy/src/lib.rs

@@ -71,6 +71,29 @@ pub enum PolicyAction {
     /// is double-gated: `invoke_query` to reach the tool, plus `change` for
     /// the write itself.
     InvokeQuery,
+    /// RFC-026: gates admitting rows into a table's streaming lane. Per-graph
+    /// and **graph-scoped**: the experimental streaming profile is main-only,
+    /// so a branch dimension would describe a topology the lane refuses. A
+    /// rule that sets `branch_scope` or `target_branch_scope` on it is
+    /// rejected by `validate()`.
+    ///
+    /// Deliberately separate from `Change`: a stream append acknowledges
+    /// durability without graph visibility, so an operator can grant
+    /// high-rate ingestion without granting direct-lane writes (or the
+    /// reverse). See RFC-026 §4.6.
+    #[value(name = "stream_ingest", alias = "stream-ingest")]
+    StreamIngest,
+    /// RFC-026: gates streaming *lifecycle management* — enable/disable,
+    /// fold, quiesce, resume, and abort-drain. Per-graph and graph-scoped for
+    /// the same main-only reason as `StreamIngest`.
+    ///
+    /// Split from `StreamIngest` because the blast radii differ in kind:
+    /// ingestion adds rows, while management can seal a lane, drain
+    /// acknowledged data, or reopen it at a new epoch. Read-only stream
+    /// status is authorized like other graph operational metadata, not by
+    /// this action.
+    #[value(name = "stream_manage", alias = "stream-manage")]
+    StreamManage,
 }
 
 impl PolicyAction {
@@ -86,6 +109,8 @@ impl PolicyAction {
             Self::Admin => "admin",
             Self::GraphList => "graph_list",
             Self::InvokeQuery => "invoke_query",
+            Self::StreamIngest => "stream_ingest",
+            Self::StreamManage => "stream_manage",
         }
     }
 
@@ -116,7 +141,9 @@ impl PolicyAction {
             | Self::BranchDelete
             | Self::BranchMerge
             | Self::Admin
-            | Self::InvokeQuery => PolicyResourceKind::Graph,
+            | Self::InvokeQuery
+            | Self::StreamIngest
+            | Self::StreamManage => PolicyResourceKind::Graph,
         }
     }
 }
@@ -173,6 +200,8 @@ impl FromStr for PolicyAction {
             "admin" => Ok(Self::Admin),
             "graph_list" => Ok(Self::GraphList),
             "invoke_query" => Ok(Self::InvokeQuery),
+            "stream_ingest" => Ok(Self::StreamIngest),
+            "stream_manage" => Ok(Self::StreamManage),
             other => bail!("unknown policy action '{other}'"),
         }
     }
@@ -845,6 +874,8 @@ namespace Omnigraph {
     action "branch_merge" appliesTo { principal: Actor, resource: Graph, context: RequestContext };
     action "admin" appliesTo { principal: Actor, resource: Graph, context: RequestContext };
     action "invoke_query" appliesTo { principal: Actor, resource: Graph, context: RequestContext };
+    action "stream_ingest" appliesTo { principal: Actor, resource: Graph, context: RequestContext };
+    action "stream_manage" appliesTo { principal: Actor, resource: Graph, context: RequestContext };
 
     action "graph_list" appliesTo { principal: Actor, resource: Server, context: RequestContext };
 }
@@ -1060,7 +1091,7 @@ rules:
     }
     use super::{
         PolicyAction, PolicyCompiler, PolicyConfig, PolicyEngine, PolicyExpectation, PolicyRequest,
-        PolicyTestCase, PolicyTestConfig,
+        PolicyResourceKind, PolicyTestCase, PolicyTestConfig,
     };
 
     #[test]
@@ -1413,6 +1444,53 @@ rules:
         );
     }
 
+    #[test]
+    fn stream_actions_are_graph_scoped_and_reject_branch_qualifiers() {
+        // RFC-026's experimental streaming profile is main-only, so a branch
+        // dimension on either stream action would describe a topology the
+        // lane refuses. Both qualifiers are rejected at validate().
+        for (action, qualifier) in [
+            ("stream_ingest", "branch_scope: any"),
+            ("stream_ingest", "target_branch_scope: any"),
+            ("stream_manage", "branch_scope: any"),
+            ("stream_manage", "target_branch_scope: any"),
+        ] {
+            let policy: PolicyConfig = serde_yaml::from_str(&format!(
+                r#"
+version: 1
+groups:
+  team: [act-alice]
+rules:
+  - id: team-stream
+    allow:
+      actors: {{ group: team }}
+      actions: [{action}]
+      {qualifier}
+"#
+            ))
+            .unwrap();
+            let err = policy.validate().unwrap_err().to_string();
+            let scope = qualifier.split(':').next().unwrap();
+            assert!(
+                err.contains(scope) && err.contains(action),
+                "{scope} on {action} must be rejected: {err}"
+            );
+        }
+
+        // Both are per-graph resources, not server-scoped.
+        for action in [PolicyAction::StreamIngest, PolicyAction::StreamManage] {
+            assert_eq!(action.resource_kind(), PolicyResourceKind::Graph);
+            // The stable policy/YAML wire name round-trips through FromStr.
+            // The CLI owns a separate parser test because it derives its
+            // accepted values through clap's ValueEnum implementation.
+            assert_eq!(
+                action.as_str().parse::<PolicyAction>().unwrap(),
+                action,
+                "wire name must round-trip through FromStr"
+            );
+        }
+    }
+
     #[test]
     fn server_scoped_rule_cannot_use_branch_scope() {
         let policy: PolicyConfig = serde_yaml::from_str(

crates/omnigraph/src/db/manifest.rs

@@ -643,6 +643,21 @@ impl Snapshot {
         self.stream_lifecycles.iter()
     }
 
+    /// Test-only seam for proving that public status resolves the live table
+    /// registration by immutable identity instead of trusting this explicitly
+    /// diagnostic alias.
+    #[cfg(all(test, feature = "failpoints"))]
+    pub(crate) fn set_stream_diagnostic_table_key_for_test(
+        &mut self,
+        identity: TableIdentity,
+        table_key: &str,
+    ) {
+        self.stream_lifecycles
+            .get_mut(&identity)
+            .expect("test snapshot must contain the requested stream lifecycle")
+            .diagnostic_table_key = table_key.to_string();
+    }
+
     /// Exact durable pointer to the graph-global RFC-026 token participant.
     pub(crate) fn stream_token_authority(&self) -> &StreamTokenAuthorityEntry {
         &self.stream_token_authority

crates/omnigraph/src/db/manifest/stream.rs

@@ -418,6 +418,17 @@ pub(crate) enum LastFoldOutcome {
     StrictBlocked,
 }
 
+impl LastFoldOutcome {
+    /// The exact durable wire name, so a status projection reports the stored
+    /// value instead of minting a second spelling of it.
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Published => "PUBLISHED",
+            Self::StrictBlocked => "STRICT_BLOCKED",
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct LastFoldSummary {

crates/omnigraph/src/db/mod.rs

@@ -14,8 +14,8 @@ pub(crate) use omnigraph::{DeferredTableFork, WriteAuthorityToken, WriteTxn};
 pub use omnigraph::{
     CleanupPolicyOptions, InitOptions, MergeOutcome, Omnigraph, OpenMode, PendingIndex,
     RepairAction, RepairClassification, RepairOptions, RepairStats, SchemaApplyOptions,
-    SchemaApplyResult, SkipReason, StreamingProfileResult, TableCleanupStats, TableOptimizeStats,
-    TableRepairStats,
+    SchemaApplyResult, SkipReason, StreamStatus, StreamTableStatus, StreamingProfileResult,
+    TableCleanupStats, TableOptimizeStats, TableRepairStats,
 };
 
 use crate::error::{OmniError, Result};

crates/omnigraph/src/db/omnigraph.rs

@@ -41,10 +41,12 @@ mod schema_apply;
 mod stream_enrollment;
 mod stream_ingest;
 mod stream_profile;
+mod stream_status;
 mod table_ops;
 
 pub use optimize::{CleanupPolicyOptions, SkipReason, TableCleanupStats, TableOptimizeStats};
 pub use stream_profile::StreamingProfileResult;
+pub use stream_status::{StreamStatus, StreamTableStatus};
 pub use repair::{
     RepairAction, RepairClassification, RepairOptions, RepairStats, TableRepairStats,
 };

crates/omnigraph/src/db/omnigraph/stream_profile.rs

@@ -54,8 +54,14 @@ impl Omnigraph {
         enabled: bool,
         actor: Option<&str>,
     ) -> Result<StreamingProfileResult> {
+        // Streaming lifecycle management, not general graph administration:
+        // the flag opens or closes a lane, so it belongs with fold, quiesce,
+        // and resume under `stream_manage` (RFC-026 §4.6). P1 shipped on the
+        // reserved `Admin` action because `stream_manage` did not exist yet;
+        // migrating is in-window precisely because §4.7 declares the
+        // experimental profile's authorization surface unstable.
         self.enforce(
-            omnigraph_policy::PolicyAction::Admin,
+            omnigraph_policy::PolicyAction::StreamManage,
             &omnigraph_policy::ResourceScope::Graph,
             actor,
         )?;