From 099570cddc86fe0e70e4bb4510b4d8be25129b7e Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Tue, 26 May 2026 15:07:41 +0530 Subject: [PATCH] fix(channels): thread workspace identity through channel events to prevent cross-workspace persistence (#2633) --- .claude/memory.md | 11 + app/test/e2e/specs/mega-flow.spec.ts | 3 +- src/core/event_bus/events.rs | 29 ++ src/core/event_bus/events_tests.rs | 2 + .../channels/providers/telegram/bus.rs | 20 ++ .../channels/providers/telegram/bus_tests.rs | 141 ++++++++ src/openhuman/channels/routes_tests.rs | 1 + src/openhuman/channels/runtime/dispatch.rs | 3 + src/openhuman/memory_conversations/bus.rs | 330 +++++++++++++++++- 9 files changed, 533 insertions(+), 7 deletions(-) diff --git a/.claude/memory.md b/.claude/memory.md index 684b1b235..91818e6d0 100644 --- a/.claude/memory.md +++ b/.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`. 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. diff --git a/app/test/e2e/specs/mega-flow.spec.ts b/app/test/e2e/specs/mega-flow.spec.ts index 8b5942734..33fb23151 100644 --- a/app/test/e2e/specs/mega-flow.spec.ts +++ b/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({ diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index 3e3ce18b3..882424a6b 100644 --- a/src/core/event_bus/events.rs +++ b/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, + /// 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 { diff --git a/src/core/event_bus/events_tests.rs b/src/core/event_bus/events_tests.rs index 424db45fb..7825988ca 100644 --- a/src/core/event_bus/events_tests.rs +++ b/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", ), diff --git a/src/openhuman/channels/providers/telegram/bus.rs b/src/openhuman/channels/providers/telegram/bus.rs index 78b4ad82d..b58fb2028 100644 --- a/src/openhuman/channels/providers/telegram/bus.rs +++ b/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}" ); diff --git a/src/openhuman/channels/providers/telegram/bus_tests.rs b/src/openhuman/channels/providers/telegram/bus_tests.rs index 65ef6e6a9..3dfd72935 100644 --- a/src/openhuman/channels/providers/telegram/bus_tests.rs +++ b/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"); +} diff --git a/src/openhuman/channels/routes_tests.rs b/src/openhuman/channels/routes_tests.rs index 78630a764..8f1f28c56 100644 --- a/src/openhuman/channels/routes_tests.rs +++ b/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; diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index a94b6fa39..fd58d1cce 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/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(), }); } diff --git a/src/openhuman/memory_conversations/bus.rs b/src/openhuman/memory_conversations/bus.rs index 33b8e13f9..4e671acd3 100644 --- a/src/openhuman/memory_conversations/bus.rs +++ b/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"); + } }