fix(channels): thread workspace identity through channel events to prevent cross-workspace persistence (#2633)

.claude/memory.md

@@ -266,3 +266,14 @@ Quick reference for anyone starting with Claude on this project. Updated by the
 - **`BootCheckTransport` is the right hook for frontend recovery** — `app/src/lib/bootCheck/index.ts` is the injection point for new recovery capabilities; don't add them directly to the BootCheck component.
 - **i18n bootCheck keys live in `-3.ts` chunks** — New keys must be added to all 13 language files simultaneously (the `-3.ts` chunk for each language).
 - **Workflow folder** — `workflow/` at repo root has 5 markdown files (00–05) defining the full PR workflow: pick issue → architectobot plan → user approval → codecrusher → architectobot verify → checks → memory-keeper → commit → push/PR.
+
+## Channel Event Workspace Routing (Issue #2602)
+
+- **Workspace identity is `PathBuf`** — Represented as the workspace directory path on `ChannelRuntimeContext` as `ctx.workspace_dir: Arc<PathBuf>`. Use `ctx.workspace_dir.as_ref().clone()` at publish sites. There is no abstract `WorkspaceId` type.
+- **`DomainEvent` workspace routing contract** — Publisher populates workspace field from context; subscriber compares against `self.workspace_dir` and early-returns with `log::debug!` on mismatch. Follow this pattern for any workspace-scoped `DomainEvent` variant.
+- **`ChannelMessageReceived` and `ChannelMessageProcessed` carry `workspace_dir`** — Added in PR for issue #2602. Guards in `ConversationPersistenceSubscriber` (memory_conversations/bus.rs) and `TelegramRemoteSubscriber` (telegram/bus.rs) prevent cross-workspace persistence during login/workspace-change races.
+
+## Pre-existing Upstream Failures (from issue #2602 session)
+
+- **Upstream `main` has 5 Vitest failures and 4 TypeScript compile errors** — Caused by missing iOS experimental dependencies: `@noble/ciphers/chacha`, `@noble/ciphers/webcrypto`, `qrcode.react`, `@tauri-apps/plugin-barcode-scanner`. Breaks `pnpm compile`, `pnpm build`, `pnpm test:coverage` on a clean checkout. Always verify by stashing changes and running checks on the base branch before blaming your PR.
+- **`cargo fmt` must run after codecrusher** — codecrusher does not reliably produce `cargo fmt`-clean Rust. Always run `cargo fmt --manifest-path Cargo.toml` after codecrusher finishes and before committing.

app/test/e2e/specs/mega-flow.spec.ts

@@ -206,7 +206,7 @@ describe('Mega flow — login + Gmail OAuth + Composio in one session', () => {
     // Login first — `oauth:success` is only meaningful for an authenticated user.
     await triggerDeepLink('openhuman://auth?token=mega-gmail-token');
     await waitForMockRequest('POST', '/telegram/login-tokens/', 15_000);
-    await waitForMockRequest('GET', '/auth/me', 10_000);
+    expect(await waitForMockRequest('GET', '/auth/me', 10_000)).toBeDefined();
     clearRequestLog();
 
     await triggerDeepLink('openhuman://oauth/success?integrationId=mock-gmail-int&provider=google');
@@ -591,6 +591,7 @@ describe('Mega flow — login + Gmail OAuth + Composio in one session', () => {
     });
     expect(auth.ok).toBe(true);
     clearRequestLog();
+    await browser.pause(500);
 
     // Seed composio state.
     setMockBehaviors({

src/core/event_bus/events.rs

@@ -3,6 +3,27 @@
 //! Events carry full payloads so subscribers have everything they need without
 //! secondary lookups. The broadcast channel clones each event per subscriber,
 //! which is fine — richness beats round-trips.
+//!
+//! ## Workspace-scoped events
+//!
+//! Some events are scoped to a specific workspace directory and must be
+//! validated by subscribers before acting on them.
+//!
+//! **Publisher contract**: when constructing a workspace-scoped event, the
+//! publisher must populate `workspace_dir` with the active workspace path at
+//! event creation time. This is typically available as `ctx.workspace_dir`
+//! on the channel runtime context.
+//!
+//! **Subscriber contract**: subscribers that persist or mutate workspace-
+//! specific data must compare the event's `workspace_dir` against their own
+//! workspace binding and silently drop events that do not match. This prevents
+//! stale in-flight events from a previous workspace from corrupting the newly
+//! active workspace's state when the user switches workspaces (e.g. logs out
+//! and back in) while events are in flight.
+//!
+//! **Current workspace-scoped variants**:
+//! - [`DomainEvent::ChannelMessageReceived`]
+//! - [`DomainEvent::ChannelMessageProcessed`]
 
 /// Top-level domain event. Non-exhaustive so new variants can be added
 /// without breaking existing match arms.
@@ -129,6 +150,10 @@ pub enum DomainEvent {
         reply_target: String,
         content: String,
         thread_ts: Option<String>,
+        /// Workspace directory active when this event was published.
+        /// Subscribers that persist data must reject events whose
+        /// `workspace_dir` does not match their own workspace binding.
+        workspace_dir: std::path::PathBuf,
     },
     /// A channel message was fully processed (LLM response sent or error).
     ChannelMessageProcessed {
@@ -141,6 +166,10 @@ pub enum DomainEvent {
         response: String,
         elapsed_ms: u64,
         success: bool,
+        /// Workspace directory active when this event was published.
+        /// Subscribers that persist data must reject events whose
+        /// `workspace_dir` does not match their own workspace binding.
+        workspace_dir: std::path::PathBuf,
     },
     /// A reaction event was received from a channel transport.
     ChannelReactionReceived {

src/core/event_bus/events_tests.rs

@@ -94,6 +94,7 @@ fn all_variants_have_correct_domain() {
                 reply_target: "r".into(),
                 content: "hi".into(),
                 thread_ts: None,
+                workspace_dir: std::path::PathBuf::from("/test"),
             },
             "channel",
         ),
@@ -108,6 +109,7 @@ fn all_variants_have_correct_domain() {
                 response: "hello".into(),
                 elapsed_ms: 0,
                 success: true,
+                workspace_dir: std::path::PathBuf::from("/test"),
             },
             "channel",
         ),

src/openhuman/channels/providers/telegram/bus.rs

@@ -56,8 +56,18 @@ impl EventHandler for TelegramRemoteSubscriber {
             DomainEvent::ChannelMessageReceived {
                 channel,
                 reply_target,
+                workspace_dir,
                 ..
             } if channel == "telegram" => {
+                if *workspace_dir != self.workspace_dir {
+                    tracing::debug!(
+                        "{LOG_PREFIX} dropping stale-workspace ChannelMessageReceived \
+                         event_ws={} self_ws={}",
+                        workspace_dir.display(),
+                        self.workspace_dir.display()
+                    );
+                    return;
+                }
                 tracing::debug!("{LOG_PREFIX} turn started reply_target={reply_target}");
                 self.set_busy(reply_target, true).await;
             }
@@ -66,8 +76,18 @@ impl EventHandler for TelegramRemoteSubscriber {
                 reply_target,
                 success,
                 elapsed_ms,
+                workspace_dir,
                 ..
             } if channel == "telegram" => {
+                if *workspace_dir != self.workspace_dir {
+                    tracing::debug!(
+                        "{LOG_PREFIX} dropping stale-workspace ChannelMessageProcessed \
+                         event_ws={} self_ws={}",
+                        workspace_dir.display(),
+                        self.workspace_dir.display()
+                    );
+                    return;
+                }
                 tracing::debug!(
                     "{LOG_PREFIX} turn finished reply_target={reply_target} success={success} elapsed_ms={elapsed_ms}"
                 );

src/openhuman/channels/providers/telegram/bus_tests.rs

@@ -17,6 +17,7 @@ async fn subscriber_marks_busy_on_received_and_clears_on_processed() {
             reply_target: "chat-99".into(),
             content: "hi".into(),
             thread_ts: Some("1".into()),
+            workspace_dir: dir.path().to_path_buf(),
         })
         .await;
 
@@ -35,6 +36,7 @@ async fn subscriber_marks_busy_on_received_and_clears_on_processed() {
             response: "ok".into(),
             elapsed_ms: 10,
             success: true,
+            workspace_dir: dir.path().to_path_buf(),
         })
         .await;
 
@@ -56,6 +58,7 @@ async fn subscriber_ignores_non_telegram_channel_events() {
             reply_target: "chat-99".into(),
             content: "hi".into(),
             thread_ts: None,
+            workspace_dir: dir.path().to_path_buf(),
         })
         .await;
 
@@ -63,3 +66,141 @@ async fn subscriber_ignores_non_telegram_channel_events() {
         .expect("store");
     assert!(!busy);
 }
+
+// ── Workspace-identity guard tests ───────────────────────────────────────────
+
+/// Positive control: matching workspace sets busy state as expected.
+#[tokio::test]
+async fn telegram_received_matching_workspace_sets_busy() {
+    let dir = tempdir().expect("tempdir");
+    let subscriber = TelegramRemoteSubscriber::new(dir.path().to_path_buf());
+
+    subscriber
+        .handle(&DomainEvent::ChannelMessageReceived {
+            channel: "telegram".into(),
+            message_id: "m1".into(),
+            sender: "alice".into(),
+            reply_target: "chat-10".into(),
+            content: "hi".into(),
+            thread_ts: None,
+            workspace_dir: dir.path().to_path_buf(),
+        })
+        .await;
+
+    let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-10")))
+        .expect("store");
+    assert!(busy, "matching workspace should mark busy");
+}
+
+/// Stale workspace on `ChannelMessageReceived` — busy must NOT be set.
+#[tokio::test]
+async fn telegram_received_stale_workspace_does_not_set_busy() {
+    let dir = tempdir().expect("tempdir");
+    let stale = tempdir().expect("stale tempdir");
+    let subscriber = TelegramRemoteSubscriber::new(dir.path().to_path_buf());
+
+    subscriber
+        .handle(&DomainEvent::ChannelMessageReceived {
+            channel: "telegram".into(),
+            message_id: "m1".into(),
+            sender: "alice".into(),
+            reply_target: "chat-20".into(),
+            content: "hi".into(),
+            thread_ts: None,
+            workspace_dir: stale.path().to_path_buf(),
+        })
+        .await;
+
+    let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-20")))
+        .expect("store");
+    assert!(!busy, "stale workspace should not set busy");
+}
+
+/// Matching workspace on `ChannelMessageProcessed` clears busy state correctly.
+#[tokio::test]
+async fn telegram_processed_matching_workspace_clears_busy() {
+    let dir = tempdir().expect("tempdir");
+    let subscriber = TelegramRemoteSubscriber::new(dir.path().to_path_buf());
+
+    // First mark as busy via a matching received event.
+    subscriber
+        .handle(&DomainEvent::ChannelMessageReceived {
+            channel: "telegram".into(),
+            message_id: "m1".into(),
+            sender: "alice".into(),
+            reply_target: "chat-30".into(),
+            content: "hi".into(),
+            thread_ts: None,
+            workspace_dir: dir.path().to_path_buf(),
+        })
+        .await;
+
+    let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-30")))
+        .expect("store");
+    assert!(busy, "should be busy after received");
+
+    // Now clear with a matching processed event.
+    subscriber
+        .handle(&DomainEvent::ChannelMessageProcessed {
+            channel: "telegram".into(),
+            message_id: "m1".into(),
+            sender: "alice".into(),
+            reply_target: "chat-30".into(),
+            content: "hi".into(),
+            thread_ts: None,
+            response: "done".into(),
+            elapsed_ms: 50,
+            success: true,
+            workspace_dir: dir.path().to_path_buf(),
+        })
+        .await;
+
+    let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-30")))
+        .expect("store");
+    assert!(!busy, "matching processed should clear busy");
+}
+
+/// Stale workspace on `ChannelMessageProcessed` — busy must NOT be cleared.
+#[tokio::test]
+async fn telegram_processed_stale_workspace_does_not_clear_busy() {
+    let dir = tempdir().expect("tempdir");
+    let stale = tempdir().expect("stale tempdir");
+    let subscriber = TelegramRemoteSubscriber::new(dir.path().to_path_buf());
+
+    // Mark as busy via a matching received event.
+    subscriber
+        .handle(&DomainEvent::ChannelMessageReceived {
+            channel: "telegram".into(),
+            message_id: "m1".into(),
+            sender: "alice".into(),
+            reply_target: "chat-40".into(),
+            content: "hi".into(),
+            thread_ts: None,
+            workspace_dir: dir.path().to_path_buf(),
+        })
+        .await;
+
+    let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-40")))
+        .expect("store");
+    assert!(busy, "should be busy after matching received");
+
+    // Attempt to clear with a stale-workspace processed event — must be ignored.
+    subscriber
+        .handle(&DomainEvent::ChannelMessageProcessed {
+            channel: "telegram".into(),
+            message_id: "m1".into(),
+            sender: "alice".into(),
+            reply_target: "chat-40".into(),
+            content: "hi".into(),
+            thread_ts: None,
+            response: "done".into(),
+            elapsed_ms: 50,
+            success: true,
+            workspace_dir: stale.path().to_path_buf(),
+        })
+        .await;
+
+    let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-40")))
+        .expect("store");
+    assert!(busy, "stale processed must not clear busy state");
+}

src/openhuman/channels/routes_tests.rs

@@ -525,6 +525,7 @@ async fn handle_runtime_command_telegram_new_status_and_sessions_round_trip() {
             reply_target: "chat-remote".into(),
             content: "work".into(),
             thread_ts: Some("42".into()),
+            workspace_dir: tempdir.path().to_path_buf(),
         })
         .await;
 

src/openhuman/channels/runtime/dispatch.rs

@@ -612,6 +612,7 @@ pub(crate) async fn process_channel_message(
         reply_target: msg.reply_target.clone(),
         content: msg.content.clone(),
         thread_ts: msg.thread_ts.clone(),
+        workspace_dir: ctx.workspace_dir.as_ref().clone(),
     });
 
     let target_channel = ctx.channels_by_name.get(&msg.channel).cloned();
@@ -1024,6 +1025,7 @@ pub(crate) async fn process_channel_message(
                     response: error_text.to_string(),
                     elapsed_ms: started_at.elapsed().as_millis() as u64,
                     success: false,
+                    workspace_dir: ctx.workspace_dir.as_ref().clone(),
                 });
                 return;
             }
@@ -1152,6 +1154,7 @@ pub(crate) async fn process_channel_message(
         response: response_text,
         elapsed_ms: started_at.elapsed().as_millis() as u64,
         success,
+        workspace_dir: ctx.workspace_dir.as_ref().clone(),
     });
 }
 

src/openhuman/memory_conversations/bus.rs

@@ -98,16 +98,26 @@ impl EventHandler for ConversationPersistenceSubscriber {
                 reply_target,
                 content,
                 thread_ts,
+                workspace_dir,
             } => {
-                let workspace_dir = match self.workspace_dir_snapshot() {
-                    Ok(workspace_dir) => workspace_dir,
+                let my_workspace = match self.workspace_dir_snapshot() {
+                    Ok(d) => d,
                     Err(error) => {
                         log::warn!("{LOG_PREFIX} failed to resolve workspace: {error}");
                         return;
                     }
                 };
+                if *workspace_dir != my_workspace {
+                    log::debug!(
+                        "{LOG_PREFIX} dropping stale-workspace event \
+                         event_ws={} self_ws={}",
+                        workspace_dir.display(),
+                        my_workspace.display()
+                    );
+                    return;
+                }
                 if let Err(error) = persist_channel_turn(
-                    &workspace_dir,
+                    &my_workspace,
                     ChannelTurnDescriptor {
                         channel,
                         message_id,
@@ -138,17 +148,27 @@ impl EventHandler for ConversationPersistenceSubscriber {
                 response,
                 elapsed_ms,
                 success,
+                workspace_dir,
                 ..
             } => {
-                let workspace_dir = match self.workspace_dir_snapshot() {
-                    Ok(workspace_dir) => workspace_dir,
+                let my_workspace = match self.workspace_dir_snapshot() {
+                    Ok(d) => d,
                     Err(error) => {
                         log::warn!("{LOG_PREFIX} failed to resolve workspace: {error}");
                         return;
                     }
                 };
+                if *workspace_dir != my_workspace {
+                    log::debug!(
+                        "{LOG_PREFIX} dropping stale-workspace event \
+                         event_ws={} self_ws={}",
+                        workspace_dir.display(),
+                        my_workspace.display()
+                    );
+                    return;
+                }
                 if let Err(error) = persist_channel_turn(
-                    &workspace_dir,
+                    &my_workspace,
                     ChannelTurnDescriptor {
                         channel,
                         message_id,
@@ -335,6 +355,7 @@ mod tests {
                 reply_target: "general".into(),
                 content: "hello".into(),
                 thread_ts: Some("thread-1".into()),
+                workspace_dir: temp.path().to_path_buf(),
             })
             .await;
         subscriber
@@ -348,6 +369,7 @@ mod tests {
                 response: "hi there".into(),
                 elapsed_ms: 42,
                 success: true,
+                workspace_dir: temp.path().to_path_buf(),
             })
             .await;
 
@@ -379,6 +401,7 @@ mod tests {
                 reply_target: "chat-1".into(),
                 content: "hello".into(),
                 thread_ts: Some("100".into()),
+                workspace_dir: temp.path().to_path_buf(),
             })
             .await;
         subscriber
@@ -389,6 +412,7 @@ mod tests {
                 reply_target: "chat-1".into(),
                 content: "follow-up".into(),
                 thread_ts: Some("200".into()),
+                workspace_dir: temp.path().to_path_buf(),
             })
             .await;
 
@@ -409,6 +433,7 @@ mod tests {
             reply_target: "room-1".into(),
             content: "hello".into(),
             thread_ts: None,
+            workspace_dir: temp.path().to_path_buf(),
         };
 
         subscriber.handle(&event).await;
@@ -446,4 +471,297 @@ mod tests {
         assert_eq!(non_empty_trimmed("   "), None);
         assert_eq!(non_empty_trimmed(""), None);
     }
+
+    // ── Workspace-identity guard tests ───────────────────────────────────────
+
+    /// Positive control: a `ChannelMessageReceived` event whose workspace matches
+    /// the subscriber's workspace IS persisted.
+    #[tokio::test]
+    async fn received_matching_workspace_is_persisted() {
+        let temp = TempDir::new().expect("tempdir");
+        let subscriber = ConversationPersistenceSubscriber::new(temp.path().to_path_buf());
+
+        subscriber
+            .handle(&DomainEvent::ChannelMessageReceived {
+                channel: "slack".into(),
+                message_id: "m1".into(),
+                sender: "bob".into(),
+                reply_target: "dev".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                workspace_dir: temp.path().to_path_buf(),
+            })
+            .await;
+
+        let messages =
+            super::super::get_messages(temp.path().to_path_buf(), "channel:slack_bob_dev")
+                .expect("messages");
+        assert_eq!(messages.len(), 1);
+        assert_eq!(messages[0].id, "user:m1");
+    }
+
+    /// `ChannelMessageReceived` with a mismatched workspace must be silently dropped —
+    /// nothing persisted in the subscriber's workspace.
+    #[tokio::test]
+    async fn received_stale_workspace_is_dropped() {
+        let temp = TempDir::new().expect("tempdir");
+        let stale = TempDir::new().expect("stale tempdir");
+        let subscriber = ConversationPersistenceSubscriber::new(temp.path().to_path_buf());
+
+        subscriber
+            .handle(&DomainEvent::ChannelMessageReceived {
+                channel: "slack".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "general".into(),
+                content: "should not persist".into(),
+                thread_ts: None,
+                workspace_dir: stale.path().to_path_buf(),
+            })
+            .await;
+
+        // No thread should have been created in temp (the subscriber's workspace).
+        let threads = super::super::list_threads(temp.path().to_path_buf()).expect("threads");
+        assert!(
+            threads.is_empty(),
+            "stale-workspace event must not create a thread"
+        );
+    }
+
+    /// `ChannelMessageProcessed` with matching workspace is appended correctly
+    /// (positive control for the processed-event guard).
+    #[tokio::test]
+    async fn processed_matching_workspace_is_appended() {
+        let temp = TempDir::new().expect("tempdir");
+        let subscriber = ConversationPersistenceSubscriber::new(temp.path().to_path_buf());
+
+        // Seed the received event first so a thread exists.
+        subscriber
+            .handle(&DomainEvent::ChannelMessageReceived {
+                channel: "slack".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "general".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                workspace_dir: temp.path().to_path_buf(),
+            })
+            .await;
+
+        subscriber
+            .handle(&DomainEvent::ChannelMessageProcessed {
+                channel: "slack".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "general".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                response: "hi there".into(),
+                elapsed_ms: 10,
+                success: true,
+                workspace_dir: temp.path().to_path_buf(),
+            })
+            .await;
+
+        let messages =
+            super::super::get_messages(temp.path().to_path_buf(), "channel:slack_alice_general")
+                .expect("messages");
+        assert_eq!(messages.len(), 2);
+        assert_eq!(messages[1].id, "assistant:m1");
+    }
+
+    /// `ChannelMessageProcessed` with a mismatched workspace must not be appended,
+    /// even if a prior `ChannelMessageReceived` for the correct workspace was already
+    /// persisted.
+    #[tokio::test]
+    async fn processed_stale_workspace_is_dropped() {
+        let temp = TempDir::new().expect("tempdir");
+        let stale = TempDir::new().expect("stale tempdir");
+        let subscriber = ConversationPersistenceSubscriber::new(temp.path().to_path_buf());
+
+        // Persist the inbound message from the correct workspace.
+        subscriber
+            .handle(&DomainEvent::ChannelMessageReceived {
+                channel: "slack".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "general".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                workspace_dir: temp.path().to_path_buf(),
+            })
+            .await;
+
+        // Then try to process with a stale workspace — must be dropped.
+        subscriber
+            .handle(&DomainEvent::ChannelMessageProcessed {
+                channel: "slack".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "general".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                response: "should not persist".into(),
+                elapsed_ms: 10,
+                success: true,
+                workspace_dir: stale.path().to_path_buf(),
+            })
+            .await;
+
+        let messages =
+            super::super::get_messages(temp.path().to_path_buf(), "channel:slack_alice_general")
+                .expect("messages");
+        // Only the user turn should be present; the stale processed event must be dropped.
+        assert_eq!(messages.len(), 1);
+        assert_eq!(messages[0].id, "user:m1");
+    }
+
+    /// Simulate the exact workspace-switch race:
+    /// 1. `ChannelMessageReceived` from workspace A — persisted.
+    /// 2. `ChannelMessageProcessed` from workspace B — dropped.
+    /// 3. `ChannelMessageProcessed` from workspace A — persisted.
+    /// Verify only workspace A's events appear.
+    #[tokio::test]
+    async fn workspace_switch_mid_conversation() {
+        let workspace_a = TempDir::new().expect("workspace_a");
+        let workspace_b = TempDir::new().expect("workspace_b");
+
+        // Subscriber is bound to workspace A.
+        let subscriber = ConversationPersistenceSubscriber::new(workspace_a.path().to_path_buf());
+
+        subscriber
+            .handle(&DomainEvent::ChannelMessageReceived {
+                channel: "telegram".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "chat-1".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                workspace_dir: workspace_a.path().to_path_buf(),
+            })
+            .await;
+
+        // Stale processed event from workspace B — must be dropped.
+        subscriber
+            .handle(&DomainEvent::ChannelMessageProcessed {
+                channel: "telegram".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "chat-1".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                response: "from workspace B — must be dropped".into(),
+                elapsed_ms: 5,
+                success: true,
+                workspace_dir: workspace_b.path().to_path_buf(),
+            })
+            .await;
+
+        // Correct processed event from workspace A — must be persisted.
+        subscriber
+            .handle(&DomainEvent::ChannelMessageProcessed {
+                channel: "telegram".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "chat-1".into(),
+                content: "hello".into(),
+                thread_ts: None,
+                response: "from workspace A — should persist".into(),
+                elapsed_ms: 10,
+                success: true,
+                workspace_dir: workspace_a.path().to_path_buf(),
+            })
+            .await;
+
+        let messages = super::super::get_messages(
+            workspace_a.path().to_path_buf(),
+            "channel:telegram_alice_chat-1",
+        )
+        .expect("messages");
+
+        assert_eq!(messages.len(), 2, "only user + correct assistant turn");
+        assert_eq!(messages[0].id, "user:m1");
+        assert_eq!(messages[1].id, "assistant:m1");
+        assert_eq!(
+            messages[1].content, "from workspace A — should persist",
+            "workspace B response must not have been written"
+        );
+    }
+
+    /// Events from 3 different wrong workspaces all get dropped; nothing persists.
+    #[tokio::test]
+    async fn multiple_stale_workspaces_all_dropped() {
+        let temp = TempDir::new().expect("tempdir");
+        let stale_a = TempDir::new().expect("stale_a");
+        let stale_b = TempDir::new().expect("stale_b");
+        let stale_c = TempDir::new().expect("stale_c");
+
+        let subscriber = ConversationPersistenceSubscriber::new(temp.path().to_path_buf());
+
+        for (i, stale) in [&stale_a, &stale_b, &stale_c].iter().enumerate() {
+            subscriber
+                .handle(&DomainEvent::ChannelMessageReceived {
+                    channel: "discord".into(),
+                    message_id: format!("m{i}"),
+                    sender: "alice".into(),
+                    reply_target: "room-1".into(),
+                    content: format!("msg {i}"),
+                    thread_ts: None,
+                    workspace_dir: stale.path().to_path_buf(),
+                })
+                .await;
+        }
+
+        let threads = super::super::list_threads(temp.path().to_path_buf()).expect("threads");
+        assert!(
+            threads.is_empty(),
+            "no events from wrong workspaces should create a thread"
+        );
+    }
+
+    /// After a stale event is dropped, a subsequent matching-workspace event is
+    /// still persisted correctly.
+    #[tokio::test]
+    async fn correct_workspace_after_stale_events() {
+        let temp = TempDir::new().expect("tempdir");
+        let stale = TempDir::new().expect("stale tempdir");
+        let subscriber = ConversationPersistenceSubscriber::new(temp.path().to_path_buf());
+
+        // Stale event first.
+        subscriber
+            .handle(&DomainEvent::ChannelMessageReceived {
+                channel: "slack".into(),
+                message_id: "m0".into(),
+                sender: "alice".into(),
+                reply_target: "general".into(),
+                content: "stale".into(),
+                thread_ts: None,
+                workspace_dir: stale.path().to_path_buf(),
+            })
+            .await;
+
+        // Now a matching-workspace event.
+        subscriber
+            .handle(&DomainEvent::ChannelMessageReceived {
+                channel: "slack".into(),
+                message_id: "m1".into(),
+                sender: "alice".into(),
+                reply_target: "general".into(),
+                content: "valid".into(),
+                thread_ts: None,
+                workspace_dir: temp.path().to_path_buf(),
+            })
+            .await;
+
+        let messages =
+            super::super::get_messages(temp.path().to_path_buf(), "channel:slack_alice_general")
+                .expect("messages");
+        assert_eq!(
+            messages.len(),
+            1,
+            "only the valid event should be persisted"
+        );
+        assert_eq!(messages[0].id, "user:m1");
+        assert_eq!(messages[0].content, "valid");
+    }
 }