diff --git a/Cargo.lock b/Cargo.lock
index 28fd83211..733175a5e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -852,6 +852,7 @@ dependencies = [
  "http",
  "jacquard-api",
  "jacquard-common",
+ "knot-capability",
  "parking_lot",
  "reqwest 0.13.1",
  "serde",
@@ -4941,6 +4942,10 @@ dependencies = [
  "moka",
 ]
 
+[[package]]
+name = "knot-capability"
+version = "0.0.1"
+
 [[package]]
 name = "knot-cob"
 version = "2.0.0"
@@ -5553,6 +5558,7 @@ dependencies = [
  "knot-acl",
  "knot-atproto",
  "knot-cache",
+ "knot-capability",
  "knot-cob",
  "knot-cobs",
  "knot-config",
diff --git a/Cargo.toml b/Cargo.toml
index e3aa2c88c..8558b79ce 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -29,6 +29,7 @@ print_stdout = "deny"
 print_stderr = "warn"
 
 [workspace.dependencies]
+knot-capability = { path = "crates/knot-capability" }
 lexicons = { path = "crates/lexicons" }
 trusted-proxies = { path = "crates/trusted-proxies" }
 
diff --git a/bobbin/crates/bobbin-sim/tests/determinism_leak.rs b/bobbin/crates/bobbin-sim/tests/determinism_leak.rs
index 738feef7f..be1b66039 100644
--- a/bobbin/crates/bobbin-sim/tests/determinism_leak.rs
+++ b/bobbin/crates/bobbin-sim/tests/determinism_leak.rs
@@ -142,8 +142,8 @@ fn cold_start_under_live_load_is_byte_deterministic() {
                 "expected 250 (replay+live) frames",
             );
             assert_eq!(
-                result.first_report.edge_count, 250,
-                "diversified owners should yield 250 distinct edge keys",
+                result.first_report.edge_count, 251,
+                "250 diversified repo subjects, plus the one (sh.tangled.repo, Global) feed key they all share",
             );
         }
     }
diff --git a/bobbin/crates/bobbin-sim/tests/warming_buffer.rs b/bobbin/crates/bobbin-sim/tests/warming_buffer.rs
index 4ecde479d..fd6dd2d5c 100644
--- a/bobbin/crates/bobbin-sim/tests/warming_buffer.rs
+++ b/bobbin/crates/bobbin-sim/tests/warming_buffer.rs
@@ -180,8 +180,10 @@ fn warming_to_ready_promote_drains_residual_via_parallel_slingshot_wave() {
     );
     assert_eq!(
         report.edge_count,
-        stars * 2 + 1,
-        "each drained star contributes a primary edge plus a sh.tangled.feed.star.by mirror edge; promoter repo adds one. got {} for {stars} stars",
+        stars * 2 + 3,
+        "each drained star contributes a primary edge plus a sh.tangled.feed.star.by mirror edge; \
+         the promoter repo adds one, and the (sh.tangled.feed.star, Global) and \
+         (sh.tangled.repo, Global) feed keys add one each. got {} for {stars} stars",
         report.edge_count,
     );
 }
diff --git a/bobbin/crates/ingest/src/lib.rs b/bobbin/crates/ingest/src/lib.rs
index 86616a12c..b6e381705 100644
--- a/bobbin/crates/ingest/src/lib.rs
+++ b/bobbin/crates/ingest/src/lib.rs
@@ -1859,7 +1859,7 @@ mod tests {
             true,
             true,
         );
-        assert!(gate.has_knot_acl(&KnotHostKey::new(&native_host)).await);
+        assert!(gate.admit(&KnotHostKey::new(&native_host)).await.is_some());
 
         let registry = KnotRegistry::new();
         let (store, issue_states, pull_statuses, cov, resolver) = fresh();
diff --git a/bobbin/crates/knot-ingest/Cargo.toml b/bobbin/crates/knot-ingest/Cargo.toml
index fbcdcc04d..0dfe4c4f7 100644
--- a/bobbin/crates/knot-ingest/Cargo.toml
+++ b/bobbin/crates/knot-ingest/Cargo.toml
@@ -10,6 +10,7 @@ bobbin-edge-index = { workspace = true }
 bobbin-knot-proxy = { workspace = true }
 bobbin-runtime = { workspace = true }
 bobbin-types = { workspace = true }
+knot-capability = { workspace = true }
 jacquard-api = { workspace = true, features = ["com_atproto", "streaming"] }
 jacquard-common = { workspace = true }
 chrono = { workspace = true }
diff --git a/bobbin/crates/knot-ingest/src/client.rs b/bobbin/crates/knot-ingest/src/client.rs
index 773366b17..2ca8b7fbe 100644
--- a/bobbin/crates/knot-ingest/src/client.rs
+++ b/bobbin/crates/knot-ingest/src/client.rs
@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -12,6 +13,7 @@ use http::{HeaderMap, StatusCode};
 use jacquard_common::DefaultStr;
 use jacquard_common::types::did::Did;
 use jacquard_common::types::nsid::Nsid;
+use knot_capability::Capability;
 use serde::Deserialize;
 use thiserror::Error;
 use url::Url;
@@ -128,18 +130,36 @@ impl KnotClient {
         Ok(Self::new(ReqwestHttp::shared(client)))
     }
 
-    pub async fn capabilities(&self, host: &KnotHost) -> Result<Vec<String>, KnotClientError> {
+    pub async fn capabilities(
+        &self,
+        host: &KnotHost,
+    ) -> Result<HashSet<Capability>, KnotClientError> {
         let mut url = host.xrpc_url(&nsid(VERSION_NSID));
         url.set_query(None);
         let bytes = self.get_json(url).await?;
         let resp: VersionWire = serde_json::from_slice(&bytes)?;
-        Ok(resp.capabilities.unwrap_or_default())
+        Ok(resp
+            .capabilities
+            .unwrap_or_default()
+            .into_iter()
+            .filter_map(|token| Capability::parse(&token))
+            .collect())
     }
 
-    pub async fn list_members(&self, host: &KnotHost) -> Result<AclListing, KnotClientError> {
-        let subject = authority(host);
-        self.drain(host, LIST_MEMBERS_NSID, subject, None, 0, Vec::new())
-            .await
+    pub async fn list_members(
+        &self,
+        host: &KnotHost,
+        knot: &Did<DefaultStr>,
+    ) -> Result<AclListing, KnotClientError> {
+        self.drain(
+            host,
+            LIST_MEMBERS_NSID,
+            knot.as_ref().to_owned(),
+            None,
+            0,
+            Vec::new(),
+        )
+        .await
     }
 
     pub async fn list_collaborators(
@@ -296,13 +316,17 @@ mod tests {
             .and(path("/xrpc/sh.tangled.knot.version"))
             .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                 "version": "1.0.0 (deadbeef)",
-                "capabilities": ["knot-acl"]
+                "capabilities": ["knot-acl", "shuck-oysters", "repo-did-input"]
             })))
             .mount(&server)
             .await;
 
         let caps = client().capabilities(&endpoint(&server)).await.unwrap();
-        assert_eq!(caps, vec!["knot-acl".to_owned()]);
+        assert_eq!(
+            caps,
+            HashSet::from([Capability::KnotAcl, Capability::RepoDidInput]),
+            "a token we never heard of is dropped instead of poisoning the set"
+        );
     }
 
     #[tokio::test]
@@ -317,7 +341,11 @@ mod tests {
             .await;
 
         let caps = client().capabilities(&endpoint(&server)).await.unwrap();
-        assert!(caps.is_empty());
+        assert_eq!(
+            caps,
+            HashSet::new(),
+            "a knot that declares no capabilities serves none"
+        );
     }
 
     #[tokio::test]
@@ -326,7 +354,7 @@ mod tests {
         let host = endpoint(&server);
         Mock::given(method("GET"))
             .and(path("/xrpc/sh.tangled.knot.listMembers"))
-            .and(query_param("subject", authority(&host).as_str()))
+            .and(query_param("subject", "did:web:knot.oyster.cafe"))
             .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                 "items": [
                     {"subject": "did:plc:boltless", "addedBy": "did:plc:akshay", "createdAt": "2026-06-01T00:00:00Z"},
@@ -336,7 +364,8 @@ mod tests {
             .mount(&server)
             .await;
 
-        let listing = client().list_members(&host).await.unwrap();
+        let knot = did("did:web:knot.oyster.cafe");
+        let listing = client().list_members(&host, &knot).await.unwrap();
         assert_eq!(listing.completeness, Completeness::Complete);
         assert_eq!(
             listing.entries,
@@ -374,7 +403,8 @@ mod tests {
             .mount(&server)
             .await;
 
-        let listing = client().list_members(&host).await.unwrap();
+        let knot = did("did:web:knot.oyster.cafe");
+        let listing = client().list_members(&host, &knot).await.unwrap();
         assert_eq!(listing.completeness, Completeness::Complete);
         let subjects: Vec<_> = listing.entries.into_iter().map(|m| m.subject).collect();
         assert_eq!(
@@ -418,7 +448,8 @@ mod tests {
             .mount(&server)
             .await;
 
-        let listing = client().list_members(&host).await.unwrap();
+        let knot = did("did:web:knot.oyster.cafe");
+        let listing = client().list_members(&host, &knot).await.unwrap();
         assert_eq!(
             listing.completeness,
             Completeness::Truncated,
diff --git a/bobbin/crates/knot-ingest/src/firehose.rs b/bobbin/crates/knot-ingest/src/firehose.rs
index c8f5532ce..ba64ed78e 100644
--- a/bobbin/crates/knot-ingest/src/firehose.rs
+++ b/bobbin/crates/knot-ingest/src/firehose.rs
@@ -7,12 +7,11 @@ use jacquard_api::com_atproto::sync::subscribe_repos;
 use jacquard_common::DefaultStr;
 use jacquard_common::types::cid::Cid;
 use jacquard_common::types::did::Did;
+use jacquard_common::types::nsid::Nsid;
+use jacquard_common::types::recordkey::Rkey;
 use serde::Deserialize;
 use thiserror::Error;
 
-pub const MEMBER_INVITE_COLLECTION: &str = "sh.tangled.knot.memberInvite";
-pub const COLLABORATOR_INVITE_COLLECTION: &str = "sh.tangled.repo.collaboratorInvite";
-
 const TYPE_COMMIT: &str = "#commit";
 const TYPE_INFO: &str = "#info";
 
@@ -55,8 +54,8 @@ pub struct Commit {
 #[derive(Clone, Debug, PartialEq)]
 pub struct RecordOp {
     pub action: OpAction,
-    pub collection: String,
-    pub rkey: String,
+    pub collection: Nsid<DefaultStr>,
+    pub rkey: Rkey<DefaultStr>,
     pub record: Option<Bytes>,
 }
 
@@ -73,6 +72,29 @@ impl OpAction {
     }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum InviteEvent {
+    KnotMember,
+    RepoCollaborator,
+}
+
+impl InviteEvent {
+    const KINDS: [Self; 2] = [Self::KnotMember, Self::RepoCollaborator];
+
+    pub const fn collection(self) -> &'static str {
+        match self {
+            Self::KnotMember => "sh.tangled.knot.memberInvite",
+            Self::RepoCollaborator => "sh.tangled.repo.collaboratorInvite",
+        }
+    }
+
+    pub fn of(collection: &Nsid<DefaultStr>) -> Option<Self> {
+        Self::KINDS
+            .into_iter()
+            .find(|kind| kind.collection() == collection.as_str())
+    }
+}
+
 #[derive(Deserialize)]
 struct Header {
     op: Option<i64>,
@@ -162,6 +184,8 @@ fn record_of(
         subscribe_repos::RepoOpAction::Other(_) => return None,
     };
     let (collection, rkey) = op.path.split_once('/')?;
+    let collection = Nsid::<DefaultStr>::new_owned(collection).ok()?;
+    let rkey = Rkey::<DefaultStr>::new_owned(rkey).ok()?;
     if !action.is_delete() && op.cid.is_none() {
         return None;
     }
@@ -178,8 +202,8 @@ fn record_of(
     };
     Some(RecordOp {
         action,
-        collection: collection.to_owned(),
-        rkey: rkey.to_owned(),
+        collection,
+        rkey,
         record,
     })
 }
@@ -303,8 +327,8 @@ mod tests {
         assert_eq!(commit.records.len(), 1);
         let op = &commit.records[0];
         assert_eq!(op.action, OpAction::Create);
-        assert_eq!(op.collection, GIT_REF_COLLECTION);
-        assert_eq!(op.rkey, "refs~2fheads~2fmain");
+        assert_eq!(op.collection.as_str(), GIT_REF_COLLECTION);
+        assert_eq!(op.rkey.as_str(), "refs~2fheads~2fmain");
         assert_eq!(op.record.as_deref(), Some(record.as_slice()));
     }
 
@@ -339,6 +363,32 @@ mod tests {
         assert!(commit.records.is_empty());
     }
 
+    #[test]
+    fn ops_whose_path_is_not_a_collection_and_record_key_are_dropped() {
+        let record = unhex(FROZEN_BARE_REF);
+        let cases = [
+            ("collection that is not an nsid", "not an nsid", "3mug"),
+            (
+                "record key with a slash in it",
+                InviteEvent::KnotMember.collection(),
+                "did:plc:limpet/extra",
+            ),
+        ];
+
+        cases.into_iter().for_each(|(case, collection, rkey)| {
+            let bytes = commit_frame(
+                "did:plc:scallop",
+                11,
+                vec![invite_op(collection, rkey, Some(TEST_CID.to_vec()))],
+                car(&[&record]),
+            );
+            let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else {
+                panic!("expected a commit frame");
+            };
+            assert!(commit.records.is_empty(), "an op with a {case} survived");
+        });
+    }
+
     #[test]
     fn error_frame_body_is_decoded_after_the_header() {
         let mut bytes = encode_one(&Val::Map(vec![(t("op"), Val::Nint(0))]));
@@ -398,7 +448,7 @@ pub(crate) mod testcbor {
     pub(crate) const GIT_REF_COLLECTION: &str = "sh.tangled.git.ref";
 
     pub(crate) const FROZEN_PUSHED_REF: &str = "a4637368617828616261626162616261626162616261626162616261626162616261626162616261626162616261626524747970657273682e74616e676c65642e6769742e7265666d782d746e676c2d656469746f726e6469643a706c633a6c696d70657473782d746e676c2d707573682d6f7074696f6e738167736b69702d6369";
-    pub(crate) const FROZEN_BARE_REF: &str = "a2637368617840636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646524747970657273682e74616e676c65642e6769742e726566";
+    pub(crate) const FROZEN_BARE_REF: &str = "a2637368617840636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646524747970657273682e74616e676c65642e6769742e726566";
 
     #[derive(Clone)]
     pub(crate) enum Val {
diff --git a/bobbin/crates/knot-ingest/src/gate.rs b/bobbin/crates/knot-ingest/src/gate.rs
index 9bf6447e3..bd967e3cc 100644
--- a/bobbin/crates/knot-ingest/src/gate.rs
+++ b/bobbin/crates/knot-ingest/src/gate.rs
@@ -1,16 +1,18 @@
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::sync::Arc;
-use std::sync::Mutex;
 use std::time::Duration;
 
+use parking_lot::Mutex;
+
 use bobbin_runtime::Clock;
 use bobbin_types::knot_acl::KnotHostKey;
+use knot_capability::Capability;
 use tokio::time::Instant;
 
-use crate::client::{KnotClient, knot_endpoint};
+use crate::client::{KnotClient, KnotClientError, knot_endpoint};
+use crate::stream::Feed;
 
-const KNOT_ACL_CAPABILITY: &str = "knot-acl";
-const LEGACY_REPROBE_INTERVAL: Duration = Duration::from_secs(300);
+const NO_ACL_REPROBE_INTERVAL: Duration = Duration::from_secs(300);
 const ERROR_REPROBE_INTERVAL: Duration = Duration::from_secs(60);
 
 struct ProbeRecord {
@@ -23,7 +25,7 @@ pub struct CapabilityGate {
     clock: Arc<dyn Clock>,
     dev: bool,
     allow_private: bool,
-    native: Mutex<HashSet<KnotHostKey>>,
+    admitted: Mutex<HashMap<KnotHostKey, Feed>>,
     last_probe: Mutex<HashMap<KnotHostKey, ProbeRecord>>,
 }
 
@@ -34,56 +36,65 @@ impl CapabilityGate {
             clock,
             dev,
             allow_private,
-            native: Mutex::new(HashSet::new()),
+            admitted: Mutex::new(HashMap::new()),
             last_probe: Mutex::new(HashMap::new()),
         }
     }
 
     pub fn is_native(&self, host: &KnotHostKey) -> bool {
-        self.native.lock().unwrap().contains(host)
+        self.feed(host).is_some()
+    }
+
+    fn feed(&self, host: &KnotHostKey) -> Option<Feed> {
+        self.admitted.lock().get(host).copied()
     }
 
-    pub async fn has_knot_acl(&self, host: &KnotHostKey) -> bool {
-        if self.is_native(host) {
-            return true;
+    pub async fn admit(&self, host: &KnotHostKey) -> Option<Feed> {
+        if let Some(feed) = self.feed(host) {
+            return Some(feed);
         }
         let now = self.clock.now_instant();
         if self.throttled(host, now) {
-            return false;
+            return None;
         }
         match self.probe(host).await {
-            Ok(true) => {
-                self.native.lock().unwrap().insert(host.clone());
-                true
+            Ok(Some(feed)) => {
+                self.admitted.lock().insert(host.clone(), feed);
+                Some(feed)
             }
-            Ok(false) => {
-                self.mark(host, now, LEGACY_REPROBE_INTERVAL);
-                false
+            Ok(None) => {
+                self.mark(host, now, NO_ACL_REPROBE_INTERVAL);
+                None
             }
             Err(err) => {
                 tracing::warn!(host = %host, error = %err, "knot capability probe failed");
                 self.mark(host, now, ERROR_REPROBE_INTERVAL);
-                false
+                None
             }
         }
     }
 
-    async fn probe(&self, host: &KnotHostKey) -> Result<bool, crate::client::KnotClientError> {
+    async fn probe(&self, host: &KnotHostKey) -> Result<Option<Feed>, KnotClientError> {
         let endpoint = knot_endpoint(host.as_str(), self.dev, self.allow_private)?;
         let caps = self.client.capabilities(&endpoint).await?;
-        Ok(caps.iter().any(|cap| cap == KNOT_ACL_CAPABILITY))
+        Ok(caps.contains(&Capability::KnotAcl).then(|| {
+            if caps.contains(&Capability::AtprotoFirehose) {
+                Feed::Atproto
+            } else {
+                Feed::Legacy
+            }
+        }))
     }
 
     fn throttled(&self, host: &KnotHostKey, now: Instant) -> bool {
         self.last_probe
             .lock()
-            .unwrap()
             .get(host)
             .is_some_and(|rec| now.saturating_duration_since(rec.at) < rec.retry_after)
     }
 
     fn mark(&self, host: &KnotHostKey, now: Instant, retry_after: Duration) {
-        self.last_probe.lock().unwrap().insert(
+        self.last_probe.lock().insert(
             host.clone(),
             ProbeRecord {
                 at: now,
@@ -160,55 +171,64 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn declares_knot_acl() {
-        let server = MockServer::start().await;
-        mount_version(&server, json!(["knot-acl"]), 1).await;
-        let (gate, host) = gate(&server, Arc::new(ManualClock::new()));
-        assert!(gate.has_knot_acl(&host).await);
-        assert!(gate.is_native(&host));
-    }
+    async fn mounted_capabilities_pick_which_feed() {
+        for (caps, wanted) in [
+            (json!(["knot-acl"]), Some(Feed::Legacy)),
+            (json!(["knot-acl", "atproto-firehose"]), Some(Feed::Atproto)),
+            (json!([]), None),
+        ] {
+            let mounted = caps.to_string();
+            let server = MockServer::start().await;
+            mount_version(&server, caps, 1).await;
+            let (gate, host) = gate(&server, Arc::new(ManualClock::new()));
 
-    #[tokio::test]
-    async fn legacy_knot_without_capability() {
-        let server = MockServer::start().await;
-        mount_version(&server, json!([]), 1).await;
-        let (gate, host) = gate(&server, Arc::new(ManualClock::new()));
-        assert!(!gate.has_knot_acl(&host).await);
-        assert!(!gate.is_native(&host));
+            assert_eq!(gate.admit(&host).await, wanted, "knot serving {mounted}");
+            assert_eq!(
+                gate.feed(&host),
+                wanted,
+                "{mounted} latched the admitted feed"
+            );
+            assert_eq!(
+                gate.is_native(&host),
+                wanted.is_some(),
+                "{mounted} counts as native"
+            );
+        }
     }
 
     #[tokio::test]
-    async fn native_is_latched_and_survives_probe_error() {
+    async fn admission_is_latched_and_survives_probe_error() {
         let server = MockServer::start().await;
         mount_version(&server, json!(["knot-acl"]), 1).await;
         let clock = Arc::new(ManualClock::new());
         let (gate, host) = gate(&server, clock.clone());
-        assert!(gate.has_knot_acl(&host).await);
+        assert_eq!(gate.admit(&host).await, Some(Feed::Legacy));
 
         server.reset().await;
-        clock.advance(LEGACY_REPROBE_INTERVAL + Duration::from_secs(1));
-        assert!(
-            gate.has_knot_acl(&host).await,
-            "latched native never re-probes"
+        clock.advance(NO_ACL_REPROBE_INTERVAL + Duration::from_secs(1));
+        assert_eq!(
+            gate.admit(&host).await,
+            Some(Feed::Legacy),
+            "latch never re-probes"
         );
         assert!(gate.is_native(&host));
     }
 
     #[tokio::test]
-    async fn legacy_throttled_then_reprobed_after_interval() {
+    async fn unadmitted_throttled_then_reprobed_after_interval() {
         let server = MockServer::start().await;
         mount_version(&server, json!([]), 2).await;
         let clock = Arc::new(ManualClock::new());
         let (gate, host) = gate(&server, clock.clone());
-        assert!(!gate.has_knot_acl(&host).await);
+        assert_eq!(gate.admit(&host).await, None);
         clock.advance(Duration::from_secs(60));
-        assert!(!gate.has_knot_acl(&host).await);
-        clock.advance(LEGACY_REPROBE_INTERVAL);
-        assert!(!gate.has_knot_acl(&host).await);
+        assert_eq!(gate.admit(&host).await, None);
+        clock.advance(NO_ACL_REPROBE_INTERVAL);
+        assert_eq!(gate.admit(&host).await, None);
     }
 
     #[tokio::test]
-    async fn legacy_upgrade_is_detected_on_reprobe() {
+    async fn acl_upgrade_is_detected_on_reprobe() {
         let server = MockServer::start().await;
         Mock::given(method("GET"))
             .and(path("/xrpc/sh.tangled.knot.version"))
@@ -226,9 +246,9 @@ mod tests {
             .await;
         let clock = Arc::new(ManualClock::new());
         let (gate, host) = gate(&server, clock.clone());
-        assert!(!gate.has_knot_acl(&host).await);
-        clock.advance(LEGACY_REPROBE_INTERVAL + Duration::from_secs(1));
-        assert!(gate.has_knot_acl(&host).await);
+        assert_eq!(gate.admit(&host).await, None);
+        clock.advance(NO_ACL_REPROBE_INTERVAL + Duration::from_secs(1));
+        assert_eq!(gate.admit(&host).await, Some(Feed::Legacy));
         assert!(gate.is_native(&host));
     }
 
@@ -243,13 +263,10 @@ mod tests {
             .await;
         let clock = Arc::new(ManualClock::new());
         let (gate, host) = gate(&server, clock.clone());
-        assert!(!gate.has_knot_acl(&host).await);
+        assert_eq!(gate.admit(&host).await, None);
         clock.advance(Duration::from_secs(1));
-        assert!(
-            !gate.has_knot_acl(&host).await,
-            "error reprobe is throttled"
-        );
+        assert_eq!(gate.admit(&host).await, None, "error reprobe is throttled");
         clock.advance(ERROR_REPROBE_INTERVAL);
-        assert!(!gate.has_knot_acl(&host).await);
+        assert_eq!(gate.admit(&host).await, None);
     }
 }
diff --git a/bobbin/crates/knot-ingest/src/legacy.rs b/bobbin/crates/knot-ingest/src/legacy.rs
new file mode 100644
index 000000000..836b41d6a
--- /dev/null
+++ b/bobbin/crates/knot-ingest/src/legacy.rs
@@ -0,0 +1,137 @@
+use jacquard_common::DefaultStr;
+use jacquard_common::types::did::Did;
+use jacquard_common::types::nsid::Nsid;
+use serde::Deserialize;
+
+use crate::stream::{Cursor, FeedHandler, Outcome};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RosterUpdate {
+    KnotMember,
+    RepoCollaborator,
+}
+
+impl RosterUpdate {
+    const KINDS: [Self; 2] = [Self::KnotMember, Self::RepoCollaborator];
+
+    const fn nsid(self) -> &'static str {
+        match self {
+            Self::KnotMember => "sh.tangled.knot.memberUpdate",
+            Self::RepoCollaborator => "sh.tangled.repo.collaboratorUpdate",
+        }
+    }
+
+    fn of(nsid: &Nsid<DefaultStr>) -> Option<Self> {
+        Self::KINDS
+            .into_iter()
+            .find(|kind| kind.nsid() == nsid.as_str())
+    }
+}
+
+#[derive(Deserialize)]
+struct FrameWire {
+    nsid: Nsid<DefaultStr>,
+    event: EventWire,
+    created: i64,
+}
+
+#[derive(Deserialize)]
+struct EventWire {
+    repo: Option<Did<DefaultStr>>,
+}
+
+pub(crate) fn dispatch(text: &str, handler: &dyn FeedHandler, cursor: &mut Cursor) -> Outcome {
+    let Ok(frame) = serde_json::from_str::<FrameWire>(text) else {
+        return Outcome::Idle;
+    };
+    if frame.created <= 0 {
+        return Outcome::Idle;
+    }
+    cursor.advance(frame.created);
+    match (RosterUpdate::of(&frame.nsid), frame.event.repo) {
+        (Some(RosterUpdate::KnotMember), _) => handler.roster_changed(),
+        (Some(RosterUpdate::RepoCollaborator), Some(repo)) => handler.repo_roster_changed(&repo),
+        _ => {}
+    }
+    Outcome::Progressed
+}
+
+#[cfg(test)]
+mod tests {
+    use parking_lot::Mutex;
+
+    use super::*;
+    use crate::stream::Feed;
+
+    #[derive(Default)]
+    struct Recorder(Mutex<Vec<String>>);
+
+    impl FeedHandler for Recorder {
+        fn roster_changed(&self) {
+            self.0.lock().push("roster".to_owned());
+        }
+
+        fn repo_roster_changed(&self, repo: &Did<DefaultStr>) {
+            self.0.lock().push(repo.as_ref().to_owned());
+        }
+    }
+
+    const MEMBER: &str = r#"{"rkey":"3mug","nsid":"sh.tangled.knot.memberUpdate",
+        "event":{"op":"add","subject":"did:plc:limpet"},"created":6521339806120}"#;
+    const MEMBER_OLD: &str = r#"{"rkey":"3mug","nsid":"sh.tangled.knot.memberUpdate",
+        "event":{"op":"add","subject":"did:plc:limpet"},"created":7}"#;
+    const MEMBER_UNDATED: &str = r#"{"rkey":"3mug","nsid":"sh.tangled.knot.memberUpdate",
+        "event":{"op":"add","subject":"did:plc:limpet"},"created":0}"#;
+    const COLLABORATOR: &str = r#"{"rkey":"3mug","nsid":"sh.tangled.repo.collaboratorUpdate",
+        "event":{"op":"remove","subject":"did:plc:limpet","repo":"did:plc:scallop"},"created":7}"#;
+    const REF: &str = r#"{"rkey":"3mug","nsid":"sh.tangled.git.refUpdate",
+        "event":{"ref":"refs/heads/main"},"created":9}"#;
+    const GARBLED: &str = r#"{"nsid":"sh.tangled.knot.memberUpdate"}"#;
+    const NOT_AN_NSID: &str = r#"{"rkey":"3mug","nsid":"not an nsid at all",
+        "event":{"op":"add","subject":"did:plc:limpet"},"created":7}"#;
+
+    #[test]
+    fn each_legacy_frame_moves_cursor_and_calls_handler() {
+        let cases: [(&str, &str, i64, bool, i64, &[&str]); 7] = [
+            ("member update", MEMBER, 0, true, 6521339806120, &["roster"]),
+            (
+                "collaborator update",
+                COLLABORATOR,
+                0,
+                true,
+                7,
+                &["did:plc:scallop"],
+            ),
+            ("unrelated ref update", REF, 4, true, 9, &[]),
+            (
+                "older member update",
+                MEMBER_OLD,
+                100,
+                true,
+                100,
+                &["roster"],
+            ),
+            ("garbled frame", GARBLED, 3, false, 3, &[]),
+            ("undated frame", MEMBER_UNDATED, 5, false, 5, &[]),
+            ("frame with a junk nsid", NOT_AN_NSID, 3, false, 3, &[]),
+        ];
+
+        cases
+            .into_iter()
+            .for_each(|(case, frame, start, news, wanted, calls)| {
+                let handler = Recorder::default();
+                let mut cursor = Cursor::at(Feed::Legacy, start);
+
+                let outcome = dispatch(frame, &handler, &mut cursor);
+
+                assert_eq!(
+                    matches!(outcome, Outcome::Progressed),
+                    news,
+                    "whether {case} progressed"
+                );
+                assert_eq!(cursor.seq(), wanted, "cursor after {case}");
+                assert_eq!(cursor.feed(), Feed::Legacy, "feed after {case}");
+                assert_eq!(*handler.0.lock(), calls, "handler calls from {case}");
+            });
+    }
+}
diff --git a/bobbin/crates/knot-ingest/src/lib.rs b/bobbin/crates/knot-ingest/src/lib.rs
index fa7dadbe4..fda2e157f 100644
--- a/bobbin/crates/knot-ingest/src/lib.rs
+++ b/bobbin/crates/knot-ingest/src/lib.rs
@@ -1,6 +1,7 @@
 pub mod client;
 pub mod firehose;
 pub mod gate;
+pub mod legacy;
 pub mod orchestrator;
 pub mod registry;
 pub mod roster;
@@ -11,4 +12,4 @@ pub use gate::CapabilityGate;
 pub use orchestrator::Orchestrator;
 pub use registry::KnotRegistry;
 pub use roster::{Cursor, Roster};
-pub use stream::{StreamConfig, run_stream};
+pub use stream::{Feed, StreamConfig, run_stream};
diff --git a/bobbin/crates/knot-ingest/src/orchestrator.rs b/bobbin/crates/knot-ingest/src/orchestrator.rs
index 7e6a4b96e..bfe8ed7fb 100644
--- a/bobbin/crates/knot-ingest/src/orchestrator.rs
+++ b/bobbin/crates/knot-ingest/src/orchestrator.rs
@@ -1,5 +1,5 @@
-use std::collections::{HashMap, HashSet};
 use parking_lot::Mutex;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -19,7 +19,7 @@ use crate::client::{AclListing, Completeness, KnotClient, KnotClientError, knot_
 use crate::gate::CapabilityGate;
 use crate::registry::KnotRegistry;
 use crate::roster::{AclKind, Cursor, Roster};
-use crate::stream::{FeedHandler, StreamConfig, run_stream};
+use crate::stream::{Feed, FeedHandler, StreamConfig, run_stream};
 
 const POLL_INTERVAL: Duration = Duration::from_secs(30);
 const RECONCILE_INTERVAL: Duration = Duration::from_secs(300);
@@ -66,15 +66,15 @@ impl Orchestrator {
             .into_iter()
             .filter(|host| !subscribed.contains_key(host) && !unspawnable.contains(host))
             .collect();
-        let approved: Vec<KnotHostKey> = futures::stream::iter(candidates)
-            .map(|host| async move { self.gate.has_knot_acl(&host).await.then_some(host) })
+        let approved: Vec<(KnotHostKey, Feed)> = futures::stream::iter(candidates)
+            .map(|host| async move { self.gate.admit(&host).await.map(|feed| (host, feed)) })
             .buffer_unordered(PROBE_CONCURRENCY)
             .filter_map(|approved| async move { approved })
             .collect()
             .await;
         approved
             .into_iter()
-            .for_each(|host| match self.spawn(&host) {
+            .for_each(|(host, feed)| match self.spawn(&host, feed) {
                 Some(token) => {
                     subscribed.insert(host, token);
                 }
@@ -85,12 +85,12 @@ impl Orchestrator {
             });
     }
 
-    fn spawn(&self, host: &KnotHostKey) -> Option<CancellationToken> {
+    fn spawn(&self, host: &KnotHostKey, feed: Feed) -> Option<CancellationToken> {
         let endpoint = knot_endpoint(host.as_str(), self.dev, self.allow_private).ok()?;
         let knot_did = host_to_knot_did(host.as_str())?;
         let roster = Arc::new(Mutex::new(Roster::new(
             self.store.clone(),
-            knot_did,
+            knot_did.clone(),
             self.registry.clone(),
             host.clone(),
         )));
@@ -107,8 +107,9 @@ impl Orchestrator {
             acl: nudge_tx,
         };
         let stream_endpoint = endpoint.clone();
+        let stream_cursor = crate::stream::Cursor::start(feed);
         tokio::spawn(async move {
-            run_stream(&stream_cfg, &stream_endpoint, &handlers, 0).await;
+            run_stream(&stream_cfg, &stream_endpoint, stream_cursor, &handlers).await;
         });
 
         let client = self.client.clone();
@@ -120,6 +121,7 @@ impl Orchestrator {
             let ctx = ReconcileCtx {
                 client: &client,
                 endpoint: &endpoint,
+                knot: &knot_did,
                 host: &host_owned,
                 registry: &registry,
                 roster: &roster,
@@ -158,22 +160,19 @@ impl KnotHandlers {
 }
 
 impl FeedHandler for KnotHandlers {
-    fn member_invite(&self, _subject: &Did<DefaultStr>) {
+    fn roster_changed(&self) {
         self.nudge(Nudge::Full);
     }
 
-    fn collaborator_invite(&self, repo: &Did<DefaultStr>, _subject: &Did<DefaultStr>) {
+    fn repo_roster_changed(&self, repo: &Did<DefaultStr>) {
         self.nudge(Nudge::Repo(repo.clone()));
     }
-
-    fn outdated_cursor(&self) {
-        self.nudge(Nudge::Full);
-    }
 }
 
 struct ReconcileCtx<'a> {
     client: &'a KnotClient,
     endpoint: &'a KnotHost,
+    knot: &'a Did<DefaultStr>,
     host: &'a KnotHostKey,
     registry: &'a KnotRegistry,
     roster: &'a Mutex<Roster>,
@@ -259,23 +258,17 @@ async fn apply_batch(ctx: &ReconcileCtx<'_>, batch: NudgeBatch) {
 async fn reconcile_once(ctx: &ReconcileCtx<'_>) {
     let horizon = ctx.roster.lock().max_cursor();
 
-    let members = ctx.client.list_members(ctx.endpoint).await;
+    let members = ctx.client.list_members(ctx.endpoint, ctx.knot).await;
     reconcile_listing(ctx, AclKind::Member, members, horizon);
 
     futures::stream::iter(ctx.registry.repos(ctx.host))
-        .for_each(|repo| async move {
-            reconcile_repo_once(ctx, &repo, horizon).await
-        })
+        .for_each(|repo| async move { reconcile_repo_once(ctx, &repo, horizon).await })
         .await;
 
     ctx.roster.lock().purge_legacy();
 }
 
-async fn reconcile_repo_once(
-    ctx: &ReconcileCtx<'_>,
-    repo: &Did<DefaultStr>,
-    horizon: Cursor,
-) {
+async fn reconcile_repo_once(ctx: &ReconcileCtx<'_>, repo: &Did<DefaultStr>, horizon: Cursor) {
     let collaborators = ctx.client.list_collaborators(ctx.endpoint, repo).await;
     reconcile_listing(ctx, AclKind::Collaborator(repo), collaborators, horizon);
 }
@@ -356,6 +349,7 @@ mod tests {
     struct Harness {
         server: MockServer,
         endpoint: KnotHost,
+        knot: Did<DefaultStr>,
         host: KnotHostKey,
         registry: Arc<KnotRegistry>,
         store: Arc<EdgeStore>,
@@ -368,6 +362,7 @@ mod tests {
             ReconcileCtx {
                 client: &self.client,
                 endpoint: &self.endpoint,
+                knot: &self.knot,
                 host: &self.host,
                 registry: &self.registry,
                 roster: &self.roster,
@@ -384,7 +379,7 @@ mod tests {
         let knot_did = host_to_knot_did(host.as_str()).unwrap();
         let roster = Arc::new(Mutex::new(Roster::new(
             store.clone(),
-            knot_did,
+            knot_did.clone(),
             registry.clone(),
             host.clone(),
         )));
@@ -392,6 +387,7 @@ mod tests {
         Harness {
             server,
             endpoint,
+            knot: knot_did,
             host,
             registry,
             store,
@@ -570,27 +566,27 @@ mod tests {
     }
 
     #[test]
-    fn handlers_nudge_full_for_member_and_outdated_events() {
+    fn handlers_nudge_full_for_roster_changes() {
         let (tx, mut rx) = mpsc::channel(4);
         let handlers = KnotHandlers {
             host: KnotHostKey::new("oyster.cafe"),
             acl: tx,
         };
-        handlers.member_invite(&did("did:plc:boltless"));
-        handlers.outdated_cursor();
+        handlers.roster_changed();
+        handlers.roster_changed();
         assert!(matches!(rx.try_recv(), Ok(Nudge::Full)));
         assert!(matches!(rx.try_recv(), Ok(Nudge::Full)));
         assert!(rx.try_recv().is_err());
     }
 
     #[test]
-    fn handlers_nudge_repo_for_collaborator_invites() {
+    fn handlers_nudge_repo_for_repo_roster_changes() {
         let (tx, mut rx) = mpsc::channel(4);
         let handlers = KnotHandlers {
             host: KnotHostKey::new("oyster.cafe"),
             acl: tx,
         };
-        handlers.collaborator_invite(&did("did:plc:scallop"), &did("did:plc:olaren"));
+        handlers.repo_roster_changed(&did("did:plc:scallop"));
         assert_eq!(rx.try_recv(), Ok(Nudge::Repo(did("did:plc:scallop"))),);
     }
 
@@ -599,6 +595,7 @@ mod tests {
         let Harness {
             server,
             endpoint,
+            knot,
             host,
             registry,
             store,
@@ -631,6 +628,7 @@ mod tests {
             let ctx = ReconcileCtx {
                 client: &client,
                 endpoint: &endpoint,
+                knot: &knot,
                 host: &loop_host,
                 registry: &loop_registry,
                 roster: &loop_roster,
diff --git a/bobbin/crates/knot-ingest/src/roster.rs b/bobbin/crates/knot-ingest/src/roster.rs
index 92f9a787c..a1d23f273 100644
--- a/bobbin/crates/knot-ingest/src/roster.rs
+++ b/bobbin/crates/knot-ingest/src/roster.rs
@@ -104,23 +104,22 @@ impl Roster {
     }
 
     pub fn reap(&mut self, kind: AclKind<'_>, present: &HashSet<Did<DefaultStr>>, horizon: Cursor) {
-        let stale: Vec<Did<DefaultStr>> = self
-            .seen
-            .iter()
-            .filter_map(|(key, state)| {
-                let subject = match (key, kind) {
-                    (DedupKey::Member(subject), AclKind::Member) => subject,
-                    (DedupKey::Collaborator(edge_repo, subject), AclKind::Collaborator(repo))
-                        if edge_repo == repo =>
-                    {
-                        subject
-                    }
-                    _ => return None,
-                };
-                (state.present && state.cursor <= horizon && !present.contains(subject))
-                    .then(|| subject.clone())
-            })
-            .collect();
+        let stale: Vec<Did<DefaultStr>> =
+            self.seen
+                .iter()
+                .filter_map(|(key, state)| {
+                    let subject = match (key, kind) {
+                        (DedupKey::Member(subject), AclKind::Member) => subject,
+                        (
+                            DedupKey::Collaborator(edge_repo, subject),
+                            AclKind::Collaborator(repo),
+                        ) if edge_repo == repo => subject,
+                        _ => return None,
+                    };
+                    (state.present && state.cursor <= horizon && !present.contains(subject))
+                        .then(|| subject.clone())
+                })
+                .collect();
         stale
             .into_iter()
             .for_each(|subject| self.retire(kind, subject));
diff --git a/bobbin/crates/knot-ingest/src/stream.rs b/bobbin/crates/knot-ingest/src/stream.rs
index 1d92406f3..d20be6712 100644
--- a/bobbin/crates/knot-ingest/src/stream.rs
+++ b/bobbin/crates/knot-ingest/src/stream.rs
@@ -10,14 +10,91 @@ use tokio_util::sync::CancellationToken;
 use url::Url;
 
 use crate::client::authority;
-use crate::firehose::{self, COLLABORATOR_INVITE_COLLECTION, Frame, MEMBER_INVITE_COLLECTION};
+use crate::firehose::{self, Frame, InviteEvent};
+use crate::legacy;
 
 const SUBSCRIBE_REPOS_PATH: &str = "xrpc/com.atproto.sync.subscribeRepos";
+const EVENTS_PATH: &str = "events";
 const RECONNECT_INITIAL: Duration = Duration::from_secs(1);
 const RECONNECT_MAX: Duration = Duration::from_secs(60);
 const HEALTHY_SESSION_MIN: Duration = Duration::from_secs(15);
 const LIVENESS_TIMEOUT: Duration = Duration::from_secs(90);
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Feed {
+    Atproto,
+    Legacy,
+}
+
+impl Feed {
+    fn spec(self) -> (Self, &'static str, &'static str) {
+        match self {
+            Self::Atproto => (Self::Legacy, SUBSCRIBE_REPOS_PATH, "atproto firehose"),
+            Self::Legacy => (Self::Atproto, EVENTS_PATH, "knot event stream"),
+        }
+    }
+
+    fn other(self) -> Self {
+        self.spec().0
+    }
+
+    fn label(self) -> &'static str {
+        self.spec().2
+    }
+
+    pub fn live(self, clock: &dyn Clock) -> Cursor {
+        match self {
+            Self::Atproto => Cursor::start(self),
+            Self::Legacy => Cursor::at(
+                self,
+                i64::try_from(clock.now_unix_micros().raw().saturating_mul(1_000))
+                    .unwrap_or(i64::MAX),
+            ),
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct Cursor {
+    feed: Feed,
+    seq: i64,
+}
+
+impl Cursor {
+    pub const fn start(feed: Feed) -> Self {
+        Self { feed, seq: 0 }
+    }
+
+    pub(crate) const fn at(feed: Feed, seq: i64) -> Self {
+        Self { feed, seq }
+    }
+
+    pub const fn feed(self) -> Feed {
+        self.feed
+    }
+
+    pub const fn seq(self) -> i64 {
+        self.seq
+    }
+
+    pub fn advance(&mut self, seq: i64) {
+        self.seq = self.seq.max(seq);
+    }
+
+    fn switched(self) -> Self {
+        Self::start(self.feed.other())
+    }
+
+    fn read(&mut self, message: &WsMessage, handler: &dyn FeedHandler) -> Outcome {
+        match (self.feed, message) {
+            (Feed::Atproto, WsMessage::Binary(bytes)) => dispatch(bytes, handler, self),
+            (Feed::Legacy, WsMessage::Text(text)) => legacy::dispatch(text, handler, self),
+            (_, WsMessage::Text(_) | WsMessage::Binary(_)) => Outcome::Mismatch,
+            _ => Outcome::Idle,
+        }
+    }
+}
+
 #[derive(Clone)]
 pub struct StreamConfig {
     pub ws: Arc<dyn WsTransport>,
@@ -26,45 +103,66 @@ pub struct StreamConfig {
 }
 
 pub trait FeedHandler: Send + Sync + 'static {
-    fn member_invite(&self, _subject: &Did<DefaultStr>) {}
-    fn collaborator_invite(&self, _repo: &Did<DefaultStr>, _subject: &Did<DefaultStr>) {}
-    fn outdated_cursor(&self) {}
+    fn roster_changed(&self);
+    fn repo_roster_changed(&self, repo: &Did<DefaultStr>);
 }
 
+#[derive(Debug)]
 enum SessionEnd {
     Cancelled,
     Closed { progressed: bool },
     ConnectFailed,
+    Mismatch,
     Resync,
+    Unroutable,
 }
 
-enum Outcome {
+pub(crate) enum Outcome {
     Progressed,
     Idle,
+    Mismatch,
     Resync,
 }
 
 pub async fn run_stream(
     cfg: &StreamConfig,
     host: &KnotHost,
+    mut cursor: Cursor,
     handler: &dyn FeedHandler,
-    initial_cursor: i64,
 ) {
-    let mut cursor = initial_cursor;
+    let mut proven = false;
     let mut backoff = RECONNECT_INITIAL;
     loop {
         if cfg.cancel.is_cancelled() {
             return;
         }
         let started = cfg.clock.now_instant();
-        let end = run_session(cfg, host, handler, &mut cursor).await;
+        let mut position = session_cursor(cursor, cfg.clock.as_ref());
+        let end = run_session(cfg, host, handler, &mut position).await;
+        cursor = position;
+        proven |= feed_confirmed(&end);
+        if wrong_feed(&end, proven) {
+            let feed = cursor.feed();
+            let (knot, tried, next) = (authority(host), feed.label(), feed.other().label());
+            match &end {
+                SessionEnd::Mismatch => {
+                    tracing::warn!(host = %knot, tried, next, "switching feeds, since this knot sent the other one's frames")
+                }
+                _ => {
+                    tracing::info!(host = %knot, tried, next, "this knot won't serve the feed we asked for, so we try the other")
+                }
+            }
+            (cursor, proven) = (cursor.switched(), false);
+        }
         match end {
-            SessionEnd::Cancelled => return,
+            SessionEnd::Cancelled | SessionEnd::Unroutable => return,
             SessionEnd::Resync => {
-                cursor = 0;
+                cursor = Cursor::start(cursor.feed());
                 backoff = RECONNECT_INITIAL;
             }
-            SessionEnd::Closed { progressed: false } | SessionEnd::ConnectFailed => {}
+            SessionEnd::ConnectFailed
+            | SessionEnd::Mismatch
+            | SessionEnd::Closed { progressed: false } => {}
             SessionEnd::Closed { progressed: true } => {
                 let elapsed = cfg.clock.now_instant().saturating_duration_since(started);
                 if session_was_healthy(&end, elapsed) {
@@ -81,6 +179,21 @@ pub async fn run_stream(
     }
 }
 
+fn feed_confirmed(end: &SessionEnd) -> bool {
+    matches!(end, SessionEnd::Resync | SessionEnd::Closed { .. })
+}
+
+fn wrong_feed(end: &SessionEnd, proven: bool) -> bool {
+    matches!(end, SessionEnd::Mismatch) || matches!(end, SessionEnd::ConnectFailed if !proven)
+}
+
+fn session_cursor(cursor: Cursor, clock: &dyn Clock) -> Cursor {
+    match (cursor.feed(), cursor.seq()) {
+        (feed @ Feed::Legacy, 0) => feed.live(clock),
+        _ => cursor,
+    }
+}
+
 fn session_was_healthy(end: &SessionEnd, elapsed: Duration) -> bool {
     matches!(end, SessionEnd::Closed { progressed: true }) && elapsed >= HEALTHY_SESSION_MIN
 }
@@ -94,17 +207,19 @@ async fn run_session(
     cfg: &StreamConfig,
     host: &KnotHost,
     handler: &dyn FeedHandler,
-    cursor: &mut i64,
+    cursor: &mut Cursor,
 ) -> SessionEnd {
-    let Some(url) = subscribe_url(host, *cursor) else {
-        return SessionEnd::ConnectFailed;
+    let feed = cursor.feed();
+    let Some(url) = feed_url(host, *cursor) else {
+        tracing::warn!(host = %authority(host), feed = feed.label(), cursor = cursor.seq(), "knot's own hostname won't form a websocket url");
+        return SessionEnd::Unroutable;
     };
     let conn = tokio::select! {
         _ = cfg.cancel.cancelled() => return SessionEnd::Cancelled,
         res = cfg.ws.connect(url) => match res {
             Ok(conn) => conn,
             Err(err) => {
-                tracing::warn!(host = %authority(host), error = %err, "firehose connect failed");
+                tracing::warn!(host = %authority(host), feed = feed.label(), error = %err, "couldn't open knot stream");
                 return SessionEnd::ConnectFailed;
             }
         },
@@ -118,32 +233,32 @@ async fn run_session(
         let message = tokio::select! {
             _ = cfg.cancel.cancelled() => return SessionEnd::Cancelled,
             _ = cfg.clock.sleep(LIVENESS_TIMEOUT) => {
-                tracing::warn!(host = %authority(host), "firehose went quiet, reconnecting");
+                tracing::warn!(host = %authority(host), feed = feed.label(), "nothing came down the stream, reconnecting");
                 return SessionEnd::Closed { progressed };
             }
             message = stream.next() => message,
         };
         match message {
             None => return SessionEnd::Closed { progressed },
-            Some(Ok(WsMessage::Binary(bytes))) => match dispatch(&bytes, handler, cursor) {
-                Outcome::Resync => return SessionEnd::Resync,
-                Outcome::Progressed => progressed = true,
-                Outcome::Idle => {}
-            },
             Some(Ok(WsMessage::Ping(payload))) => {
                 let _ = sink.send(WsMessage::Pong(payload)).await;
             }
             Some(Ok(WsMessage::Close { .. })) => return SessionEnd::Closed { progressed },
-            Some(Ok(_)) => {}
+            Some(Ok(message)) => match cursor.read(&message, handler) {
+                Outcome::Resync => return SessionEnd::Resync,
+                Outcome::Mismatch => return SessionEnd::Mismatch,
+                Outcome::Progressed => progressed = true,
+                Outcome::Idle => {}
+            },
             Some(Err(err)) => {
-                tracing::warn!(host = %authority(host), error = %err, "firehose read error");
+                tracing::warn!(host = %authority(host), feed = feed.label(), error = %err, "read failed mid-stream");
                 return SessionEnd::Closed { progressed };
             }
         }
     }
 }
 
-fn dispatch(data: &[u8], handler: &dyn FeedHandler, cursor: &mut i64) -> Outcome {
+fn dispatch(data: &[u8], handler: &dyn FeedHandler, cursor: &mut Cursor) -> Outcome {
     let frame = match firehose::decode_frame(data) {
         Ok(frame) => frame,
         Err(err) => {
@@ -153,12 +268,12 @@ fn dispatch(data: &[u8], handler: &dyn FeedHandler, cursor: &mut i64) -> Outcome
     };
     match frame {
         Frame::Commit(commit) => {
-            *cursor = (*cursor).max(commit.seq);
+            cursor.advance(commit.seq);
             dispatch_commit(&commit, handler);
             Outcome::Progressed
         }
         _ if firehose::is_outdated_cursor(&frame) => {
-            handler.outdated_cursor();
+            handler.roster_changed();
             Outcome::Resync
         }
         _ if firehose::is_future_cursor(&frame) => {
@@ -183,34 +298,27 @@ fn log_undecodable(err: &firehose::FirehoseError) {
 }
 
 fn dispatch_commit(commit: &firehose::Commit, handler: &dyn FeedHandler) {
-    for op in &commit.records {
-        match op.collection.as_str() {
-            MEMBER_INVITE_COLLECTION => {
-                if let Ok(subject) = Did::new_owned(op.rkey.as_str()) {
-                    handler.member_invite(&subject);
-                }
-            }
-            COLLABORATOR_INVITE_COLLECTION => {
-                if let Ok(subject) = Did::new_owned(op.rkey.as_str()) {
-                    handler.collaborator_invite(&commit.repo, &subject);
-                }
-            }
-            _ => {}
-        }
-    }
+    commit
+        .records
+        .iter()
+        .filter_map(|op| InviteEvent::of(&op.collection))
+        .for_each(|invite| match invite {
+            InviteEvent::KnotMember => handler.roster_changed(),
+            InviteEvent::RepoCollaborator => handler.repo_roster_changed(&commit.repo),
+        });
 }
 
-fn subscribe_url(host: &KnotHost, cursor: i64) -> Option<Url> {
+fn feed_url(host: &KnotHost, cursor: Cursor) -> Option<Url> {
     let scheme = if host.url().scheme() == "https" {
         "wss"
     } else {
         "ws"
     };
-    let base = format!("{scheme}://{}/{SUBSCRIBE_REPOS_PATH}", authority(host));
-    let full = if cursor != 0 {
-        format!("{base}?cursor={cursor}")
-    } else {
-        base
+    let (_, path, _) = cursor.feed().spec();
+    let base = format!("{scheme}://{}/{path}", authority(host));
+    let full = match cursor.seq() {
+        0 => base,
+        seq => format!("{base}?cursor={seq}"),
     };
     Url::parse(&full).ok()
 }
@@ -221,7 +329,8 @@ mod tests {
     use std::sync::atomic::{AtomicUsize, Ordering};
 
     use bobbin_runtime::{
-        NetworkError, SystemClock, WsConnectFuture, WsMessageFuture, WsSendFuture, WsSink, WsStream,
+        NetworkError, SimClock, SystemClock, UnixMicros, WsConnectFuture, WsMessageFuture,
+        WsSendFuture, WsSink, WsStream,
     };
     use bytes::Bytes;
     use parking_lot::Mutex;
@@ -231,9 +340,8 @@ mod tests {
 
     #[derive(Clone, Debug, PartialEq)]
     enum Entry {
-        Member(Did<DefaultStr>),
-        Collaborator(Did<DefaultStr>, Did<DefaultStr>),
-        Outdated,
+        Full,
+        Repo(Did<DefaultStr>),
     }
 
     #[derive(Default)]
@@ -252,16 +360,12 @@ mod tests {
     }
 
     impl FeedHandler for RecordingHandler {
-        fn member_invite(&self, subject: &Did<DefaultStr>) {
-            self.push(Entry::Member(subject.clone()));
+        fn roster_changed(&self) {
+            self.push(Entry::Full);
         }
 
-        fn collaborator_invite(&self, repo: &Did<DefaultStr>, subject: &Did<DefaultStr>) {
-            self.push(Entry::Collaborator(repo.clone(), subject.clone()));
-        }
-
-        fn outdated_cursor(&self) {
-            self.push(Entry::Outdated);
+        fn repo_roster_changed(&self, repo: &Did<DefaultStr>) {
+            self.push(Entry::Repo(repo.clone()));
         }
     }
 
@@ -290,8 +394,10 @@ mod tests {
     struct ScriptWs {
         sessions: Mutex<Vec<VecDeque<WsMessage>>>,
         sent: Arc<Mutex<Vec<WsMessage>>>,
+        urls: Mutex<Vec<String>>,
         connects: AtomicUsize,
         fail: bool,
+        fail_first: usize,
     }
 
     impl ScriptWs {
@@ -299,8 +405,10 @@ mod tests {
             Self {
                 sessions: Mutex::new(sessions),
                 sent: Arc::new(Mutex::new(Vec::new())),
+                urls: Mutex::new(Vec::new()),
                 connects: AtomicUsize::new(0),
                 fail: false,
+                fail_first: 0,
             }
         }
 
@@ -310,9 +418,10 @@ mod tests {
     }
 
     impl WsTransport for ScriptWs {
-        fn connect(&self, _url: Url) -> WsConnectFuture {
-            self.connects.fetch_add(1, Ordering::SeqCst);
-            if self.fail {
+        fn connect(&self, url: Url) -> WsConnectFuture {
+            let attempt = self.connects.fetch_add(1, Ordering::SeqCst);
+            self.urls.lock().push(url.to_string());
+            if self.fail || attempt < self.fail_first {
                 return Box::pin(async {
                     Err(NetworkError::Connect("scripted failure".to_owned()))
                 });
@@ -363,6 +472,31 @@ mod tests {
         Did::new_owned(s).unwrap()
     }
 
+    type Script = Vec<VecDeque<WsMessage>>;
+
+    const FIREHOSE_PATH: &str = "xrpc/com.atproto.sync.subscribeRepos";
+    const EVENTS_URL: &str = "ws://oyster.cafe/events";
+    const MEMBER_UPDATE: &str = r#"{"rkey":"3mug","nsid":"sh.tangled.knot.memberUpdate",
+        "event":{"op":"add","subject":"did:plc:limpet"},"created":11}"#;
+    const STRAY_REF: &str = r#"{"nsid":"sh.tangled.git.refUpdate","created":9}"#;
+
+    fn text(json: &str) -> WsMessage {
+        WsMessage::Text(json.to_owned())
+    }
+
+    async fn drive(refusals: usize, sessions: Script) -> (Vec<Entry>, Vec<String>) {
+        let ws = Arc::new(ScriptWs {
+            fail_first: refusals,
+            ..ScriptWs::new(sessions)
+        });
+        let handler = RecordingHandler::default();
+        let config = cfg(ws.clone(), CancellationToken::new());
+        let host = KnotHost::parse("http://oyster.cafe").unwrap();
+        let stream = run_stream(&config, &host, Cursor::start(Feed::Atproto), &handler);
+        let _ = tokio::time::timeout(Duration::from_secs(30), stream).await;
+        (handler.entries(), ws.urls.lock().clone())
+    }
+
     #[tokio::test]
     async fn session_dispatches_invites_refs_pongs_and_advances_cursor() {
         let handler = RecordingHandler::default();
@@ -372,13 +506,13 @@ mod tests {
             invite_commit(
                 "did:plc:limpet",
                 100,
-                crate::firehose::MEMBER_INVITE_COLLECTION,
+                InviteEvent::KnotMember.collection(),
                 "did:plc:boltless",
             ),
             invite_commit(
                 repo,
                 200,
-                crate::firehose::COLLABORATOR_INVITE_COLLECTION,
+                InviteEvent::RepoCollaborator.collection(),
                 "did:plc:olaren",
             ),
             WsMessage::Ping(Bytes::from_static(b"ka")),
@@ -414,19 +548,13 @@ mod tests {
         let ws: Arc<dyn WsTransport> = Arc::new(script);
         let config = cfg(ws, CancellationToken::new());
         let host = KnotHost::parse("http://oyster.cafe").unwrap();
-        let mut cursor = 0i64;
+        let mut cursor = Cursor::start(Feed::Atproto);
 
         let end = run_session(&config, &host, &handler, &mut cursor).await;
 
         assert!(matches!(end, SessionEnd::Closed { progressed: true }));
-        assert_eq!(cursor, 400);
-        assert_eq!(
-            handler.entries(),
-            vec![
-                Entry::Member(did("did:plc:boltless")),
-                Entry::Collaborator(did(repo), did("did:plc:olaren")),
-            ],
-        );
+        assert_eq!(cursor.seq(), 400);
+        assert_eq!(handler.entries(), vec![Entry::Full, Entry::Repo(did(repo))]);
         let sent = sent.lock();
         assert_eq!(sent.len(), 1);
         assert!(matches!(&sent[0], WsMessage::Pong(p) if p.as_ref() == b"ka"));
@@ -449,12 +577,12 @@ mod tests {
         ])]));
         let config = cfg(ws, CancellationToken::new());
         let host = KnotHost::parse("http://oyster.cafe").unwrap();
-        let mut cursor = 42i64;
+        let mut cursor = Cursor::at(Feed::Atproto, 42);
 
         let end = run_session(&config, &host, &handler, &mut cursor).await;
 
         assert!(matches!(end, SessionEnd::Resync));
-        assert_eq!(handler.entries(), vec![Entry::Outdated]);
+        assert_eq!(handler.entries(), vec![Entry::Full]);
     }
 
     #[tokio::test]
@@ -474,7 +602,7 @@ mod tests {
         ])]));
         let config = cfg(ws, CancellationToken::new());
         let host = KnotHost::parse("http://oyster.cafe").unwrap();
-        let mut cursor = 42i64;
+        let mut cursor = Cursor::at(Feed::Atproto, 42);
 
         let end = run_session(&config, &host, &handler, &mut cursor).await;
 
@@ -490,7 +618,7 @@ mod tests {
             invite_commit(
                 "did:plc:limpet",
                 500,
-                crate::firehose::MEMBER_INVITE_COLLECTION,
+                InviteEvent::KnotMember.collection(),
                 "did:plc:boltless",
             ),
             WsMessage::Close {
@@ -500,16 +628,13 @@ mod tests {
         ])]));
         let config = cfg(ws, CancellationToken::new());
         let host = KnotHost::parse("http://oyster.cafe").unwrap();
-        let mut cursor = 0i64;
+        let mut cursor = Cursor::start(Feed::Atproto);
 
         let end = run_session(&config, &host, &handler, &mut cursor).await;
 
         assert!(matches!(end, SessionEnd::Closed { progressed: true }));
-        assert_eq!(cursor, 500);
-        assert_eq!(
-            handler.entries(),
-            vec![Entry::Member(did("did:plc:boltless"))]
-        );
+        assert_eq!(cursor.seq(), 500);
+        assert_eq!(handler.entries(), vec![Entry::Full]);
     }
 
     #[tokio::test(start_paused = true)]
@@ -525,7 +650,7 @@ mod tests {
             session(vec![invite_commit(
                 "did:plc:limpet",
                 700,
-                crate::firehose::MEMBER_INVITE_COLLECTION,
+                InviteEvent::KnotMember.collection(),
                 "did:plc:boltless",
             )]),
             session(vec![]),
@@ -535,7 +660,13 @@ mod tests {
         let host = KnotHost::parse("http://oyster.cafe").unwrap();
         let task_handler = handler.clone();
         let task = tokio::spawn(async move {
-            run_stream(&config, &host, task_handler.as_ref(), 900).await;
+            run_stream(
+                &config,
+                &host,
+                Cursor::at(Feed::Atproto, 900),
+                task_handler.as_ref(),
+            )
+            .await;
         });
         for _ in 0..30 {
             tokio::time::advance(Duration::from_secs(1)).await;
@@ -548,7 +679,7 @@ mod tests {
 
         assert_eq!(
             handler.entries(),
-            vec![Entry::Outdated, Entry::Member(did("did:plc:boltless"))],
+            vec![Entry::Full, Entry::Full],
             "an outdated cursor resyncs to live and keeps delivering after the reconnect"
         );
         assert!(
@@ -557,39 +688,181 @@ mod tests {
         );
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn refused_firehose_falls_back_to_knot_event_stream() {
+        let sessions = vec![session(vec![text(MEMBER_UPDATE)]), session(vec![])];
+
+        let (entries, urls) = drive(1, sessions).await;
+
+        assert_eq!(
+            entries,
+            vec![Entry::Full],
+            "legacy frame arrives after the refusal"
+        );
+        assert!(urls[0].ends_with(FIREHOSE_PATH), "firehose is tried first");
+        let live = urls[1].strip_prefix(&format!("{EVENTS_URL}?cursor="));
+        let started_live = live.is_some_and(|at| at.parse::<i64>().is_ok_and(|nanos| nanos > 0));
+        assert!(
+            started_live,
+            "fresh legacy subscription starts at the present"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn text_frame_on_firehose_switches_to_legacy_feed() {
+        let stray = session(vec![text(STRAY_REF)]);
+        let member = session(vec![text(MEMBER_UPDATE)]);
+
+        let (entries, urls) = drive(0, vec![stray, member]).await;
+
+        assert_eq!(
+            entries,
+            vec![Entry::Full],
+            "legacy frame shows up once we switch"
+        );
+        assert!(
+            urls[1].starts_with(EVENTS_URL),
+            "legacy path follows, got {}",
+            urls[1]
+        );
+    }
+
+    #[test]
+    fn wrong_kind_frame_reads_as_mismatch() {
+        let handler = RecordingHandler::default();
+
+        [
+            (Cursor::start(Feed::Atproto), text("{}")),
+            (Cursor::start(Feed::Legacy), WsMessage::Binary(Bytes::new())),
+        ]
+        .into_iter()
+        .for_each(|(mut cursor, frame)| {
+            assert!(matches!(cursor.read(&frame, &handler), Outcome::Mismatch));
+            assert_eq!(cursor.seq(), 0, "mismatched frame leaves the cursor alone");
+        });
+    }
+
+    #[test]
+    fn only_fresh_legacy_subscription_starts_at_present() {
+        let clock = SystemClock::new();
+        let live = session_cursor(Cursor::start(Feed::Legacy), &clock);
+        assert_eq!(live.feed(), Feed::Legacy);
+        assert!(live.seq() > 0, "fresh legacy feed resumes from now");
+
+        [
+            Cursor::at(Feed::Legacy, 1788245839553422000),
+            Cursor::start(Feed::Atproto),
+            Cursor::at(Feed::Atproto, 42),
+        ]
+        .into_iter()
+        .for_each(|at| {
+            assert_eq!(session_cursor(at, &clock), at, "resumed as it stands");
+        });
+    }
+
+    #[test]
+    fn switching_feeds_never_reuses_the_other_clocks_number() {
+        let legacy_nanos = 1788245839553422000;
+        let clock = SimClock::at(UnixMicros::new(1_788_245_839_553_422));
+
+        let from_legacy = Cursor::at(Feed::Legacy, legacy_nanos).switched();
+        assert_eq!(from_legacy.feed(), Feed::Atproto);
+        assert_eq!(
+            from_legacy.seq(),
+            0,
+            "a wall clock reading is not a firehose sequence number"
+        );
+
+        let from_atproto = Cursor::at(Feed::Atproto, 42).switched();
+        assert_eq!(from_atproto.feed(), Feed::Legacy);
+        assert_eq!(
+            from_atproto.seq(),
+            0,
+            "a firehose sequence number is not a wall clock reading"
+        );
+        assert!(
+            session_cursor(from_atproto, &clock).seq() >= legacy_nanos,
+            "the legacy feed opens at the present instead"
+        );
+    }
+
+    #[test]
+    fn session_end_says_whether_feed_is_proven_or_wrong() {
+        [
+            (SessionEnd::Resync, true, false),
+            (SessionEnd::Closed { progressed: true }, true, false),
+            (SessionEnd::Closed { progressed: false }, true, false),
+            (SessionEnd::ConnectFailed, false, true),
+            (SessionEnd::Mismatch, false, true),
+            (SessionEnd::Unroutable, false, false),
+        ]
+        .into_iter()
+        .for_each(|(end, proves, wrong)| {
+            assert_eq!(
+                feed_confirmed(&end),
+                proves,
+                "{end:?} proves the feed: {proves}"
+            );
+            assert_eq!(
+                wrong_feed(&end, false),
+                wrong,
+                "{end:?} is the wrong feed: {wrong}"
+            );
+        });
+
+        assert!(
+            !wrong_feed(&SessionEnd::ConnectFailed, true),
+            "proven feed is kept"
+        );
+    }
+
     #[tokio::test]
     async fn pre_cancelled_stream_returns_without_connecting() {
         let handler = RecordingHandler::default();
         let ws: Arc<dyn WsTransport> = Arc::new(ScriptWs {
-            sessions: Mutex::new(Vec::new()),
-            sent: Arc::new(Mutex::new(Vec::new())),
-            connects: AtomicUsize::new(0),
             fail: true,
+            ..ScriptWs::new(Vec::new())
         });
         let cancel = CancellationToken::new();
         cancel.cancel();
         let config = cfg(ws, cancel);
         let host = KnotHost::parse("http://oyster.cafe").unwrap();
 
-        run_stream(&config, &host, &handler, 0).await;
+        run_stream(&config, &host, Cursor::start(Feed::Atproto), &handler).await;
     }
 
     #[test]
-    fn subscribe_url_carries_scheme_path_and_cursor() {
-        let host = KnotHost::parse("http://oyster.cafe").unwrap();
-        assert_eq!(
-            subscribe_url(&host, 0).unwrap().as_str(),
-            "ws://oyster.cafe/xrpc/com.atproto.sync.subscribeRepos"
-        );
-        assert_eq!(
-            subscribe_url(&host, 42).unwrap().as_str(),
-            "ws://oyster.cafe/xrpc/com.atproto.sync.subscribeRepos?cursor=42"
-        );
+    fn each_feed_url_has_scheme_path_and_cursor() {
+        let plain = KnotHost::parse("http://oyster.cafe").unwrap();
         let secure = KnotHost::parse("https://nel.pet").unwrap();
-        assert_eq!(
-            subscribe_url(&secure, 7).unwrap().as_str(),
-            "wss://nel.pet/xrpc/com.atproto.sync.subscribeRepos?cursor=7"
-        );
+
+        [
+            (
+                &plain,
+                Cursor::start(Feed::Atproto),
+                format!("ws://oyster.cafe/{FIREHOSE_PATH}"),
+            ),
+            (
+                &plain,
+                Cursor::at(Feed::Atproto, 42),
+                format!("ws://oyster.cafe/{FIREHOSE_PATH}?cursor=42"),
+            ),
+            (&plain, Cursor::start(Feed::Legacy), EVENTS_URL.to_owned()),
+            (
+                &secure,
+                Cursor::at(Feed::Atproto, 7),
+                format!("wss://nel.pet/{FIREHOSE_PATH}?cursor=7"),
+            ),
+            (
+                &secure,
+                Cursor::at(Feed::Legacy, 7),
+                "wss://nel.pet/events?cursor=7".to_owned(),
+            ),
+        ]
+        .into_iter()
+        .for_each(|(host, cursor, wanted)| {
+            assert_eq!(feed_url(host, cursor).unwrap().as_str(), wanted);
+        });
     }
 
     #[test]
diff --git a/bobbin/crates/resolver/src/legacy_upgrade.rs b/bobbin/crates/resolver/src/legacy_upgrade.rs
index 61cc0daf6..8016fe5c4 100644
--- a/bobbin/crates/resolver/src/legacy_upgrade.rs
+++ b/bobbin/crates/resolver/src/legacy_upgrade.rs
@@ -93,7 +93,6 @@ pub fn synthesize_created_at(bytes: &[u8], fallback_rfc3339: &str) -> Option<all
 #[derive(Clone, Copy, Debug)]
 enum FieldRule {
     DropIfEmptyString,
-    NullToEmptyArray,
     EnsureArray,
 }
 
@@ -101,11 +100,11 @@ fn scrub_rules(nsid: &str) -> &'static [(&'static str, FieldRule)] {
     match nsid {
         "sh.tangled.actor.profile" => &[("preferredHandle", FieldRule::DropIfEmptyString)],
         "sh.tangled.label.op" => &[
-            ("add", FieldRule::NullToEmptyArray),
-            ("delete", FieldRule::NullToEmptyArray),
+            ("add", FieldRule::EnsureArray),
+            ("delete", FieldRule::EnsureArray),
         ],
         "sh.tangled.repo.pull" => &[
-            ("rounds", FieldRule::NullToEmptyArray),
+            ("rounds", FieldRule::EnsureArray),
             ("versions", FieldRule::EnsureArray),
         ],
         _ => &[],
@@ -128,7 +127,6 @@ pub fn scrub_record_bytes<S: BosStr + AsRef<str>>(
             (FieldRule::DropIfEmptyString, Some(serde_json::Value::String(s))) if s.is_empty() => {
                 Some((*field, *rule))
             }
-            (FieldRule::NullToEmptyArray, Some(serde_json::Value::Null)) => Some((*field, *rule)),
             (FieldRule::EnsureArray, Some(serde_json::Value::Null) | None) => Some((*field, *rule)),
             _ => None,
         })
@@ -140,7 +138,7 @@ pub fn scrub_record_bytes<S: BosStr + AsRef<str>>(
         FieldRule::DropIfEmptyString => {
             obj.remove(*field);
         }
-        FieldRule::NullToEmptyArray | FieldRule::EnsureArray => {
+        FieldRule::EnsureArray => {
             obj.insert(
                 (*field).to_owned(),
                 serde_json::Value::Array(alloc::vec::Vec::new()),
@@ -606,6 +604,17 @@ mod tests {
         assert!(obj.get("add").is_some_and(|v| v.is_array()));
     }
 
+    #[test]
+    fn label_op_decodes_when_a_required_array_is_absent_rather_than_null() {
+        let json = br#"{"$type":"sh.tangled.label.op","add":[{"key":"at://did:plc:limpet/sh.tangled.label.definition/k","value":"v"}],"performedAt":"2026-05-01T00:00:00Z","subject":"at://did:plc:limpet/sh.tangled.repo.issue/3aaa"}"#;
+        let decoded = DecodedRecord::try_decode(&nsid("sh.tangled.label.op"), json)
+            .expect("label.op without a delete key must scrub-recover");
+        assert!(matches!(
+            decoded,
+            DecodedRecord::Canon(r) if matches!(*r, Record::LabelOp(_))
+        ));
+    }
+
     #[test]
     fn scrub_passes_through_when_label_op_arrays_are_non_null() {
         let json = br#"{"$type":"sh.tangled.label.op","add":[],"delete":[],"performedAt":"2026-05-01T00:00:00Z","subject":"at://did:plc:limpet/sh.tangled.repo.issue/3aaa"}"#;
diff --git a/crates/knot-capability/Cargo.toml b/crates/knot-capability/Cargo.toml
new file mode 100644
index 000000000..9fb78042e
--- /dev/null
+++ b/crates/knot-capability/Cargo.toml
@@ -0,0 +1,9 @@
+[package]
+name = "knot-capability"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+
+[lints]
+workspace = true
diff --git a/crates/knot-capability/src/lib.rs b/crates/knot-capability/src/lib.rs
new file mode 100644
index 000000000..3223d8570
--- /dev/null
+++ b/crates/knot-capability/src/lib.rs
@@ -0,0 +1,70 @@
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
+pub enum Capability {
+    KnotAcl,
+    RepoDidInput,
+    KeepCommit,
+    AtprotoFirehose,
+}
+
+impl Capability {
+    const ALL: &'static [Capability] = &[
+        Capability::KnotAcl,
+        Capability::RepoDidInput,
+        Capability::KeepCommit,
+        Capability::AtprotoFirehose,
+    ];
+
+    pub const SERVED: &'static [Capability] = &[
+        Capability::KnotAcl,
+        Capability::RepoDidInput,
+        Capability::AtprotoFirehose,
+    ];
+
+    pub const fn token(self) -> &'static str {
+        match self {
+            Capability::KnotAcl => "knot-acl",
+            Capability::RepoDidInput => "repo-did-input",
+            Capability::KeepCommit => "knot-keepcommit",
+            Capability::AtprotoFirehose => "atproto-firehose",
+        }
+    }
+
+    pub fn parse(token: &str) -> Option<Self> {
+        Self::ALL.iter().copied().find(|cap| cap.token() == token)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn every_capability_parses_back_from_its_token() {
+        Capability::ALL.iter().copied().for_each(|cap| {
+            assert_eq!(Capability::parse(cap.token()), Some(cap), "{cap:?}");
+        });
+    }
+
+    #[test]
+    fn tokens_are_distinct() {
+        let tokens: Vec<&str> = Capability::ALL.iter().map(|cap| cap.token()).collect();
+        let mut unique = tokens.clone();
+        unique.sort_unstable();
+        unique.dedup();
+        assert_eq!(unique.len(), tokens.len(), "two variants share a token");
+    }
+
+    #[test]
+    fn unknown_tokens_are_rejected() {
+        ["", "knot_acl", "KNOT-ACL", "atproto-firehose "]
+            .iter()
+            .for_each(|token| assert_eq!(Capability::parse(token), None, "{token:?}"));
+    }
+
+    #[test]
+    fn served_capabilities_are_real_capabilities() {
+        Capability::SERVED.iter().copied().for_each(|cap| {
+            assert!(Capability::ALL.contains(&cap), "{cap:?} is not in ALL");
+        });
+    }
+}
diff --git a/knot2/crates/knot-fixtures/src/lib.rs b/knot2/crates/knot-fixtures/src/lib.rs
index fbe4ebe53..0c3fb0e0a 100644
--- a/knot2/crates/knot-fixtures/src/lib.rs
+++ b/knot2/crates/knot-fixtures/src/lib.rs
@@ -6,6 +6,10 @@ pub const AUTHOR_NAME: &str = "nel";
 pub const AUTHOR_EMAIL: &str = "nel@oyster.cafe";
 pub const PINNED_DATE: &str = "2026-01-01T00:00:00 +0000";
 
+pub const UNSHARED_SSH: &str = "-o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
+     -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes \
+     -o ControlMaster=no -o ControlPath=none";
+
 pub fn command(cwd: &Path) -> Command {
     let mut command = Command::new("git");
     command
diff --git a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs
index 5f6c3b2b0..586cee277 100644
--- a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs
+++ b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs
@@ -537,10 +537,7 @@ async fn the_lfs_round_trip_gate_holds_over_both_transports_and_the_fork() {
 
     let media = media_bytes();
     let media_oid = LfsOid::from_digest(Sha256::digest(&media).into());
-    let ssh = format!(
-        "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
-    );
+    let ssh = format!("ssh -i {key_path} {}", knot_fixtures::UNSHARED_SSH);
     let path_env = std::env::var("PATH").unwrap_or_default();
     let home = scratch.path().to_str().unwrap().to_string();
     let env: Vec<(String, String)> = [
@@ -651,10 +648,7 @@ async fn many_objects_ride_default_git_lfs_concurrency_over_both_transports() {
     let (key_path, public_line) = keygen(scratch.path());
     let world = spawn_world(public_line).await;
 
-    let ssh = format!(
-        "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
-    );
+    let ssh = format!("ssh -i {key_path} {}", knot_fixtures::UNSHARED_SSH);
     let path_env = std::env::var("PATH").unwrap_or_default();
     let home = scratch.path().to_str().unwrap().to_string();
     let env: Vec<(String, String)> = [
@@ -797,10 +791,7 @@ async fn the_lfs_stack_is_conformant_with_the_reference_server_and_client() {
     ]
     .into();
 
-    let ssh = format!(
-        "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
-    );
+    let ssh = format!("ssh -i {key_path} {}", knot_fixtures::UNSHARED_SSH);
     let path_env = std::env::var("PATH").unwrap_or_default();
     let home = scratch.path().to_str().unwrap().to_string();
     let knot_env: Vec<(String, String)> = [
diff --git a/knot2/crates/knot-sim/tests/ssh.rs b/knot2/crates/knot-sim/tests/ssh.rs
index bd8ca9a0e..b1b487f24 100644
--- a/knot2/crates/knot-sim/tests/ssh.rs
+++ b/knot2/crates/knot-sim/tests/ssh.rs
@@ -201,10 +201,7 @@ async fn spawn(published_line: String) -> Server {
 }
 
 fn ssh_command(key_path: &str) -> String {
-    format!(
-        "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
-    )
+    format!("ssh -i {key_path} {}", knot_fixtures::UNSHARED_SSH)
 }
 
 fn seed_work(work: &Path) -> String {
diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs
index db38e2f59..b03f9e47c 100644
--- a/knot2/crates/knot-ssh/tests/ssh_push.rs
+++ b/knot2/crates/knot-ssh/tests/ssh_push.rs
@@ -9,6 +9,7 @@ use knot_atproto::Atproto;
 use knot_cob::{CobHome, CobStore};
 use knot_cobs::{CollaboratorsChange, Grant, MembersChange, Registration, RegistryChange};
 use knot_git::{ArchiveLimit, Layout, Repo};
+use knot_fixtures::UNSHARED_SSH;
 use knot_index::{Index, Resolved};
 use knot_pack::MaxWireBytes;
 use knot_runtime::{
@@ -489,10 +490,7 @@ async fn spawn_server_core(
 }
 
 fn ssh_command(key_path: &str) -> String {
-    format!(
-        "ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes"
-    )
+    format!("ssh -i {key_path} {UNSHARED_SSH}")
 }
 
 async fn git_ssh(cwd: &Path, key: &str, args: &[&str]) -> (bool, String) {
@@ -1137,8 +1135,7 @@ async fn a_filled_key_set_refuses_an_unregistered_key_and_an_acl_write_reopens_t
 
     let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
     let two_ids = format!(
-        "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
+        "ssh -i {unregistered_path} -i {} {UNSHARED_SSH}",
         fx.key_path
     );
     let (ok, out) = {
@@ -1177,8 +1174,7 @@ async fn key_recognition_edge_cases() {
 
     let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
     let two_ids = format!(
-        "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
+        "ssh -i {unregistered_path} -i {} {UNSHARED_SSH}",
         fx.key_path
     );
     let (ok, out) = {
@@ -1321,8 +1317,7 @@ async fn a_handle_in_the_url_identifies_a_visitor_and_lets_a_multi_key_client_fi
 
     let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered");
     let two_ids = format!(
-        "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
-         -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes",
+        "ssh -i {unregistered_path} -i {} {UNSHARED_SSH}",
         fx.key_path
     );
     let identified = format!("ssh://nel.pet@127.0.0.1:{}/{REPO_DID}", fx.server.port);
@@ -1812,19 +1807,9 @@ fn ssh_bare(key_path: &str, port: u16) -> (bool, String) {
 
 fn ssh_bare_as(key_path: &str, user: &str, port: u16) -> (bool, String) {
     let out = Command::new("ssh")
+        .args(["-i", key_path])
+        .args(UNSHARED_SSH.split_whitespace())
         .args([
-            "-i",
-            key_path,
-            "-o",
-            "IdentitiesOnly=yes",
-            "-o",
-            "StrictHostKeyChecking=no",
-            "-o",
-            "UserKnownHostsFile=/dev/null",
-            "-o",
-            "PreferredAuthentications=publickey",
-            "-o",
-            "BatchMode=yes",
             "-p",
             &port.to_string(),
             &format!("{user}@127.0.0.1"),
@@ -2016,19 +2001,9 @@ fn trickled_lfs_upload(
 ) -> (bool, String) {
     use std::io::Write;
     let mut child = Command::new("ssh")
+        .args(["-i", key_path])
+        .args(UNSHARED_SSH.split_whitespace())
         .args([
-            "-i",
-            key_path,
-            "-o",
-            "IdentitiesOnly=yes",
-            "-o",
-            "StrictHostKeyChecking=no",
-            "-o",
-            "UserKnownHostsFile=/dev/null",
-            "-o",
-            "PreferredAuthentications=publickey",
-            "-o",
-            "BatchMode=yes",
             "-p",
             &port.to_string(),
             "git@127.0.0.1",
diff --git a/knot2/crates/knot-xrpc/Cargo.toml b/knot2/crates/knot-xrpc/Cargo.toml
index 5a5972b9c..965fd9fe5 100644
--- a/knot2/crates/knot-xrpc/Cargo.toml
+++ b/knot2/crates/knot-xrpc/Cargo.toml
@@ -6,6 +6,7 @@ rust-version.workspace = true
 license.workspace = true
 
 [dependencies]
+knot-capability = { workspace = true }
 knot-types = { workspace = true }
 knot-git = { workspace = true }
 knot-lfs = { workspace = true }
diff --git a/knot2/crates/knot-xrpc/src/service.rs b/knot2/crates/knot-xrpc/src/service.rs
index 9caf10ffb..ee6ad15f9 100644
--- a/knot2/crates/knot-xrpc/src/service.rs
+++ b/knot2/crates/knot-xrpc/src/service.rs
@@ -5,6 +5,7 @@ use axum::extract::State;
 use axum::response::{IntoResponse, Response};
 use serde::Serialize;
 
+use knot_capability::Capability;
 use knot_runtime::{Clock, HttpTransport};
 use knot_types::AccountDid;
 
@@ -19,7 +20,7 @@ const WIRE_VERSION: &str = "v1.15.0";
 #[derive(Serialize)]
 struct VersionWire {
     version: &'static str,
-    capabilities: [&'static str; 2],
+    capabilities: Vec<&'static str>,
 }
 
 #[derive(Serialize)]
@@ -53,7 +54,10 @@ struct OwnerWire {
 pub(crate) async fn version() -> Response {
     Json(VersionWire {
         version: WIRE_VERSION,
-        capabilities: ["knot-acl", "repo-did-input"],
+        capabilities: Capability::SERVED
+            .iter()
+            .map(|capability| capability.token())
+            .collect(),
     })
     .into_response()
 }
diff --git a/knot2/crates/knot-xrpc/tests/reads.rs b/knot2/crates/knot-xrpc/tests/reads.rs
index 45e5595b2..03c7c78f7 100644
--- a/knot2/crates/knot-xrpc/tests/reads.rs
+++ b/knot2/crates/knot-xrpc/tests/reads.rs
@@ -7,6 +7,7 @@ use futures::stream;
 use http::{HeaderMap, StatusCode, header};
 use tokio_tungstenite::tungstenite;
 
+use knot_capability::Capability;
 use knot_index::{Resolved, Standing};
 use knot_types::{AccountDid, Oid, RepoDid};
 use knot_xrpc::{ArchiveLimit, ResponseLimit};
@@ -2013,10 +2014,13 @@ async fn service_metadata_endpoints_answer() {
     let world = World::new();
     let wire = get_json(&world, "/xrpc/sh.tangled.knot.version").await;
     assert_eq!(wire["version"], "v1.15.0");
-    assert_eq!(
-        wire["capabilities"],
-        serde_json::json!(["knot-acl", "repo-did-input"])
-    );
+    let served = [
+        Capability::KnotAcl,
+        Capability::RepoDidInput,
+        Capability::AtprotoFirehose,
+    ]
+    .map(Capability::token);
+    assert_eq!(wire["capabilities"], serde_json::json!(served));
 
     let owner = get_json(&world, "/xrpc/sh.tangled.owner").await;
     assert_eq!(owner["owner"], OWNER);
diff --git a/knot2/interop/atproto_reads_test.go b/knot2/interop/atproto_reads_test.go
index ad32f05c4..cba947355 100644
--- a/knot2/interop/atproto_reads_test.go
+++ b/knot2/interop/atproto_reads_test.go
@@ -15,26 +15,19 @@ import (
 	"github.com/bluesky-social/indigo/atproto/identity"
 	"github.com/bluesky-social/indigo/atproto/repo"
 	"github.com/bluesky-social/indigo/atproto/syntax"
-	lexutil "github.com/bluesky-social/indigo/lex/util"
 	indigoxrpc "github.com/bluesky-social/indigo/xrpc"
 	"github.com/samber/lo"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	cbg "github.com/whyrusleeping/cbor-gen"
+	"tangled.org/core/api/tangled"
 )
 
-const memberInviteNSID = "sh.tangled.knot.memberInvite"
-
 type MemberInvite struct {
 	CreatedAt string `json:"createdAt"`
 	Editor    string `json:"x-tngl-editor"`
 }
 
-func (invite *MemberInvite) MarshalCBOR(w io.Writer) error {
-	return fmt.Errorf("the interop suite only reads")
-}
-
-
 func (invite *MemberInvite) UnmarshalCBOR(r io.Reader) error {
 	maj, count, err := cbg.CborReadHeader(r)
 	if err != nil {
@@ -91,7 +84,6 @@ func TestKnotServesComAtprotoReads(t *testing.T) {
 	did := os.Getenv("KNOT_TEST_DID")
 	collection := os.Getenv("KNOT_TEST_COLLECTION")
 	rkey := os.Getenv("KNOT_TEST_RKEY")
-	lexutil.RegisterType(memberInviteNSID, &MemberInvite{})
 
 	latest, err := comatproto.SyncGetLatestCommit(ctx, xc, did)
 	require.NoError(t, err, "sync.getLatestCommit answers an unmodified client")
@@ -144,11 +136,17 @@ func TestKnotServesComAtprotoReads(t *testing.T) {
 	records, err := comatproto.RepoListRecords(ctx, xc, collection, "", 50, did, false)
 	require.NoError(t, err, "repo.listRecords answers an unmodified client")
 	assert.Equal(t, "at://"+did+"/"+collection+"/"+rkey, records.Records[0].Uri)
-	invite, ok := records.Records[0].Value.Val.(*MemberInvite)
-	require.True(t, ok, "the value decodes as the registered lexicon type")
-	assert.Equal(t, "did:web:olaren.dev", invite.Editor)
+	invite, ok := records.Records[0].Value.Val.(*tangled.KnotMemberInvite)
+	require.True(t, ok, "the value decodes as the generated lexicon type")
 	assert.Equal(t, "1970-01-01T00:16:40Z", invite.CreatedAt)
 
+	raw, _, err := full.GetRecordBytes(ctx, nsid, recordKey)
+	require.NoError(t, err, "the stored record bytes come back out of the signed tree")
+	var stored MemberInvite
+	require.NoError(t, stored.UnmarshalCBOR(bytes.NewReader(raw)))
+	assert.Equal(t, "did:web:olaren.dev", stored.Editor, "the knot serves the undeclared editor field")
+	assert.Equal(t, "1970-01-01T00:16:40Z", stored.CreatedAt)
+
 	describe, err := comatproto.RepoDescribeRepo(ctx, xc, did)
 	require.NoError(t, err, "repo.describeRepo answers an unmodified client")
 	assert.Equal(t, did, describe.Did)