fix(nip-ar): bound live sync heads by the token's channels

A channel-scoped token could sync channel A and then receive the full
head of an artifact that moved to channel B, when its pubkey could read
B. The live head read now drops a head outside the token's channels, so
the sync sends the source removal instead. When no readable frame can
remove an artifact from the client's copy, the sync ends with resync.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>

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

@@ -345,7 +345,7 @@ pub async fn handle_sync(
             return;
         }
     };
-    let reader = match conn.auth_state_snapshot() {
+    let (reader, bound) = match conn.auth_state_snapshot() {
         AuthState::Authenticated(ctx) => {
             if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) {
                 refuse(&sub_id, "restricted: insufficient scope", &conn, &state).await;
@@ -359,7 +359,7 @@ pub async fn handle_sync(
                     return;
                 }
             }
-            ctx.pubkey.to_bytes().to_vec()
+            (ctx.pubkey.to_bytes().to_vec(), ctx.channel_ids)
         }
         _ => {
             refuse(&sub_id, "auth-required: not authenticated", &conn, &state).await;
@@ -428,7 +428,15 @@ pub async fn handle_sync(
     drop(permit);
 
     run_live(
-        &sub_id, owner, &filter, &reader, in_set, &handle, &conn, &state,
+        &sub_id,
+        owner,
+        &filter,
+        &reader,
+        bound.as_deref(),
+        in_set,
+        &handle,
+        &conn,
+        &state,
     )
     .await;
 }
@@ -645,12 +653,15 @@ async fn try_send_owned(
 /// Rules 3–5 after `EOSE`: read each changed head, one read at a time, and
 /// send it. A dropped frame or a resync request ends the sync. `in_set` is
 /// the client's copy as this sync sent it: the `d` of each artifact in scope.
+/// `bound` is a channel-scoped token's channels: the reader reads no head
+/// outside them, even where its pubkey could.
 #[allow(clippy::too_many_arguments)]
 async fn run_live(
     sub_id: &str,
     owner: u64,
     filter: &ArtifactSyncFilter,
     reader: &[u8],
+    bound: Option<&[Uuid]>,
     mut in_set: HashSet<Uuid>,
     handle: &SyncHandle,
     conn: &ConnectionState,
@@ -672,7 +683,11 @@ async fn run_live(
                 .artifact_head(conn.tenant.community(), reader, d, &filter.channels)
                 .await
             {
-                Ok(head) => head,
+                Ok(head) => head.filter(|(head, _)| {
+                    bound.is_none_or(|allowed| {
+                        head.channel_id.is_some_and(|id| allowed.contains(&id))
+                    })
+                }),
                 Err(e) => {
                     warn!(conn_id = %conn.conn_id, "artifact sync head read failed: {e}");
                     close(sub_id, owner, "error: sync read failed", conn, state).await;
@@ -690,8 +705,11 @@ async fn run_live(
                 continue;
             }
             // A head the reader cannot read leaves only the source removal.
+            // Without one, nothing the reader may read can remove the
+            // artifact from its copy, so the copy is no longer exact.
             let Some(event) = head.map(|(head, _)| head.event).or(removal) else {
-                continue;
+                close(sub_id, owner, "resync: artifact left scope", conn, state).await;
+                return;
             };
             #[cfg(test)]
             super::close::test_seam::pause_at(&test_seam::LIVE_SEND_PAUSE).await;

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

@@ -384,6 +384,49 @@ async fn membership_removal_resyncs_the_whole_sync() {
         .is_empty());
 }
 
+/// Rule 3 under a channel-scoped token: the token bounds what the reader
+/// can read, so a move to a channel outside it sends the source removal,
+/// never the destination head, even though the pubkey can read that
+/// channel. Mutation: drop the token bound on the live head read → the
+/// moved head is sent → RED.
+#[tokio::test]
+#[ignore = "requires Postgres and Redis"]
+async fn scoped_token_gets_only_the_removal_for_a_move_outside_it() {
+    let f = fixture().await;
+    let owner = Keys::generate();
+    let source = create_channel(&f.state, &f.tenant, &owner, &[]).await;
+    let destination = create_channel(&f.state, &f.tenant, &owner, &[]).await;
+    let d = Uuid::new_v4();
+    let created = revision(&f, &owner, source, d, "create", None).await;
+
+    let (conn, mut rx) = connect(&f, &owner);
+    match &mut *conn.auth_state.lock().unwrap() {
+        crate::connection::AuthState::Authenticated(ctx) => {
+            ctx.channel_ids = Some(vec![source]);
+        }
+        _ => unreachable!("test connections are authenticated"),
+    }
+    let task = start_sync(&f, &conn, sync_filter(&[source]));
+    assert_eq!(initial_set(&mut rx).await, vec![created.id.to_hex()]);
+
+    let moved = revision(&f, &owner, destination, d, "move", Some(&created)).await;
+    let message = frame(&mut rx).await;
+    assert_eq!(
+        (message[0].as_str(), message[2]["kind"].as_u64()),
+        (Some("EVENT"), Some(45011)),
+        "{message}"
+    );
+    assert_ne!(message[2]["id"].as_str(), Some(moved.id.to_hex().as_str()));
+    no_frame(&mut rx).await;
+
+    super::close::handle_close("s".into(), Arc::clone(&conn), Arc::clone(&f.state)).await;
+    assert_eq!(frame(&mut rx).await, json!(["CLOSED", "s", ""]));
+    tokio::time::timeout(Duration::from_secs(5), task)
+        .await
+        .expect("sync task ends on CLOSE")
+        .unwrap();
+}
+
 /// Rule 2: an unreadable or unknown channel gets `CLOSED restricted` and
 /// no `EOSE`, with the same reason for both.
 #[tokio::test]