couple fixes

Author: orual

bsky_agent/agents/pattern_config.toml

@@ -453,6 +453,7 @@ exclude_dids = [
     "did:plc:vszw3ess46odfhnzdsy4huae",
     "did:plc:lw7hallqsgrhwumgudkph44j",
     "did:plc:djpcowss7aokaydcdraoaovz",
+    "did:plc:zmdk74qov5y6ouh2vsooiqkl",
 ]
 require_agent_participation = true
 

crates/pattern_core/src/data_source/bluesky/inner.rs

@@ -844,7 +844,7 @@ impl BlueskyStreamInner {
         let (_sink, mut messages) = stream.into_stream();
 
         loop {
-            if *self.status.read() != StreamStatus::Stopped {
+            if *self.status.read() == StreamStatus::Stopped {
                 return Ok(());
             }
 
@@ -868,16 +868,15 @@ impl BlueskyStreamInner {
                 Some(result) = messages.next() => {
                     *self.last_message_time.write() = Some(Instant::now());
 
-                    if *self.status.read() == StreamStatus::Running {
-                        match result {
-                            Ok(msg) => {
-                                self.handle_message(msg);
-                            }
-                            Err(e) => {
-                                error!("BlueskyStream {} message error: {}", self.source_id, e);
-                            }
+                    match result {
+                        Ok(msg) => {
+                            self.handle_message(msg);
+                        }
+                        Err(e) => {
+                            error!("BlueskyStream {} message error: {}", self.source_id, e);
                         }
                     }
+
                 }
                 _ = tokio::time::sleep(Duration::from_secs(1)) => {
                     self.flush_expired_batches().await;
@@ -896,13 +895,15 @@ impl BlueskyStreamInner {
             } => {
                 *self.current_cursor.write() = Some(time_us);
 
-                if let Some(post) = self.parse_commit(&did, time_us, &commit) {
-                    if self.should_include_post(&post) {
-                        debug!(
-                            "BlueskyStream {} accepted post from {} ({})",
-                            self.source_id, post.did, post.uri
-                        );
-                        self.pending_batch.add_post(post);
+                if *self.status.read() == StreamStatus::Running {
+                    if let Some(post) = self.parse_commit(&did, time_us, &commit) {
+                        if self.should_include_post(&post) {
+                            debug!(
+                                "BlueskyStream {} accepted post from {} ({})",
+                                self.source_id, post.did, post.uri
+                            );
+                            self.pending_batch.add_post(post);
+                        }
                     }
                 }
             }