Add manual ACP session rotation command (#932)

Signed-off-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: Max <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>

crates/buzz-acp/README.md

@@ -139,7 +139,17 @@ Controls which authors' events the harness forwards to the agent. Events from di
 | `anyone` | Forward all events (no author filtering). |
 | `nobody` | Drop all inbound events. Agent only acts on heartbeat prompts. |
 
-The gate applies to **all** inbound events — @mentions, DMs, thread replies, and any event delivered by the relay. The `!shutdown` command is checked **before** the gate, so the owner can always shut down the agent regardless of mode.
+The gate applies to **all** inbound events — @mentions, DMs, thread replies, and any event delivered by the relay. Owner control commands are checked **before** the gate, so the owner can still manage the harness regardless of mode:
+
+| Command | Effect |
+|---------|--------|
+| `!shutdown` | Gracefully exits the harness. |
+| `!cancel` | Cancels the current in-flight turn for that channel, if any. |
+| `!rotate` | Rotates the ACP session for that channel. If a turn is in-flight, it is cancelled and the channel session is invalidated when the task returns; otherwise the cached idle session is invalidated immediately. The next queued/received event starts a fresh session. |
+
+Use `!cancel` to stop only the current turn; it is a no-op when the channel is idle. Use `!rotate` when you want the next turn in the channel to start from a fresh ACP session, even if the channel is currently idle.
+
+Owner control commands must be kind:9 stream messages from the owner, must mention this agent with a `p` tag, and are consumed by the harness instead of being forwarded to the agent.
 
 > **Note:** The default mode is `owner-only`. Agents without a registered `agent_owner_pubkey` will not respond to any events until the owner is resolved. Set `--respond-to anyone` to disable the gate entirely.
 

crates/buzz-acp/src/lib.rs

@@ -28,7 +28,7 @@ use filter::SubscriptionRule;
 use futures_util::FutureExt;
 use nostr::{PublicKey, ToBech32};
 use pool::{
-    AgentPool, CancelMode, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource,
+    AgentPool, ControlSignal, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource,
     SessionState,
 };
 use queue::{prepend_base_prompt, EventQueue, QueuedEvent, ThreadTags};
@@ -540,7 +540,7 @@ fn handle_relay_observer_control_event(
         return;
     };
 
-    let fired = cancel_in_flight_task(pool, channel_id, CancelMode::Stop);
+    let fired = signal_in_flight_task(pool, channel_id, ControlSignal::Cancel);
     let status = if fired { "sent" } else { "no_active_turn" };
     if let Some(observer) = observer {
         observer.emit(
@@ -1471,12 +1471,12 @@ async fn tokio_main() -> Result<()> {
 
                             // ── Shutdown command handling ─────────────────────
                             // Check: kind:9, content "!shutdown", from owner, mentions THIS agent.
-                            let is_shutdown = kind_u32 == KIND_STREAM_MESSAGE
+                            let is_shutdown = is_owner_control_command(
-                                && buzz_event.event.content.trim() == "!shutdown"
+                                &buzz_event.event,
-                                && buzz_event.event.tags.iter().any(|t| {
+                                kind_u32,
-                                    t.as_slice().first().map(|s| s.as_str()) == Some("p")
+                                "!shutdown",
-                                        && t.as_slice().get(1).map(|s| s.as_str()) == Some(pubkey_hex.as_str())
+                                &pubkey_hex,
-                                });
+                            );
                             if is_shutdown {
                                 let owner = owner_cache.get();
                                 if let Some(owner) = owner {
@@ -1504,16 +1504,20 @@ async fn tokio_main() -> Result<()> {
                             // Mode-independent: !cancel fires regardless of
                             // --multiple-event-handling. It is explicit user
                             // intent, not an automatic policy decision.
-                            let is_cancel = kind_u32 == KIND_STREAM_MESSAGE
+                            let is_cancel = is_owner_control_command(
-                                && buzz_event.event.content.trim() == "!cancel"
+                                &buzz_event.event,
-                                && buzz_event.event.tags.iter().any(|t| {
+                                kind_u32,
-                                    t.as_slice().first().map(|s| s.as_str()) == Some("p")
+                                "!cancel",
-                                        && t.as_slice().get(1).map(|s| s.as_str()) == Some(pubkey_hex.as_str())
+                                &pubkey_hex,
-                                });
+                            );
                             if is_cancel {
                                 if let Some(owner) = owner_cache.get() {
                                     if buzz_event.event.pubkey.to_hex() == *owner {
-                                        let fired = cancel_in_flight_task(&mut pool, buzz_event.channel_id, CancelMode::Stop);
+                                        let fired = signal_in_flight_task(
+                                            &mut pool,
+                                            buzz_event.channel_id,
+                                            ControlSignal::Cancel,
+                                        );
                                         if !fired {
                                             tracing::warn!(
                                                 channel_id = %buzz_event.channel_id,
@@ -1527,6 +1531,53 @@ async fn tokio_main() -> Result<()> {
                             }
                             // ── End cancel command handling ───────────────────
 
+                            // ── Rotate command handling ─────────────────────
+                            // Mirrors !shutdown / !cancel: kind:9, content
+                            // "!rotate", from owner, mentions THIS agent.
+                            //
+                            // Rotation is explicit owner intent to start the
+                            // next turn in this channel with a fresh ACP
+                            // session. It is consumed by the harness and never
+                            // forwarded to the agent. If a turn is in-flight,
+                            // cancel it, drop its triggering batch, and
+                            // invalidate the channel session when the task
+                            // returns. If idle, invalidate the cached channel
+                            // session immediately. Queued future events remain
+                            // queued and will create a fresh session on dispatch.
+                            let is_rotate = is_owner_control_command(
+                                &buzz_event.event,
+                                kind_u32,
+                                "!rotate",
+                                &pubkey_hex,
+                            );
+                            if is_rotate {
+                                if let Some(owner) = owner_cache.get() {
+                                    if buzz_event.event.pubkey.to_hex() == *owner {
+                                        let fired = signal_in_flight_task(
+                                            &mut pool,
+                                            buzz_event.channel_id,
+                                            ControlSignal::Rotate,
+                                        );
+                                        if fired {
+                                            tracing::info!(
+                                                channel_id = %buzz_event.channel_id,
+                                                "!rotate received — cancelling in-flight turn and rotating session"
+                                            );
+                                        } else {
+                                            let invalidated = pool.invalidate_channel_sessions(buzz_event.channel_id);
+                                            tracing::info!(
+                                                channel_id = %buzz_event.channel_id,
+                                                invalidated,
+                                                "!rotate received — invalidated idle channel session(s)"
+                                            );
+                                        }
+                                        continue; // consume event — do NOT push to queue
+                                    }
+                                }
+                                // Not from owner — fall through to normal prompt handling.
+                            }
+                            // ── End rotate command handling ──────────────────
+
                             // ── Inbound author gate ──────────────────────────
                             // Coarse security policy: drop events from disallowed
                             // authors before they reach subscription rules or the
@@ -1608,7 +1659,11 @@ async fn tokio_main() -> Result<()> {
                                     }
                                 };
                                 if should_cancel {
-                                    cancel_in_flight_task(&mut pool, buzz_event.channel_id, CancelMode::Interrupt);
+                                    signal_in_flight_task(
+                                        &mut pool,
+                                        buzz_event.channel_id,
+                                        ControlSignal::Interrupt,
+                                    );
                                 }
                             }
                             // ── End mode gate ────────────────────────────────
@@ -1877,20 +1932,44 @@ enum LoopAction {
     Exit,
 }
 
-// ── cancel_in_flight_task ─────────────────────────────────────────────────────
+// ── Owner control commands ───────────────────────────────────────────────────
 
-/// Send a cancel signal to the in-flight task for `channel_id`.
+fn event_mentions_agent(event: &nostr::Event, agent_pubkey_hex: &str) -> bool {
+    event.tags.iter().any(|t| {
+        t.as_slice().first().map(|s| s.as_str()) == Some("p")
+            && t.as_slice().get(1).map(|s| s.as_str()) == Some(agent_pubkey_hex)
+    })
+}
+
+fn is_owner_control_command(
+    event: &nostr::Event,
+    kind_u32: u32,
+    command: &str,
+    agent_pubkey_hex: &str,
+) -> bool {
+    kind_u32 == KIND_STREAM_MESSAGE
+        && event.content.trim() == command
+        && event_mentions_agent(event, agent_pubkey_hex)
+}
+
+// ── signal_in_flight_task ─────────────────────────────────────────────────────
+
+/// Send a control signal to the in-flight task for `channel_id`.
 /// Returns `true` if a signal was sent, `false` if no in-flight task was found.
-fn cancel_in_flight_task(pool: &mut AgentPool, channel_id: uuid::Uuid, mode: CancelMode) -> bool {
+fn signal_in_flight_task(
+    pool: &mut AgentPool,
+    channel_id: uuid::Uuid,
+    mode: ControlSignal,
+) -> bool {
     let entry = pool
         .task_map_mut()
         .values_mut()
         .find(|m| m.channel_id == Some(channel_id));
 
     if let Some(meta) = entry {
-        if let Some(tx) = meta.cancel_tx.take() {
+        if let Some(tx) = meta.control_tx.take() {
             let _ = tx.send(mode);
-            tracing::info!(channel = %channel_id, "cancel signal sent to in-flight task");
+            tracing::info!(channel = %channel_id, ?mode, "control signal sent to in-flight task");
             return true;
         }
     }
@@ -1941,7 +2020,7 @@ fn dispatch_pending(
 
         // Prompt text is now built inside run_prompt_task (needs async for
         // context fetching). Pass None for prompt_text; batch carries the data.
-        let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<CancelMode>();
+        let (control_tx, control_rx) = tokio::sync::oneshot::channel::<ControlSignal>();
 
         let abort_handle = pool.join_set.spawn(async move {
             pool::run_prompt_task(
@@ -1950,7 +2029,7 @@ fn dispatch_pending(
                 None,
                 ctx_clone,
                 result_tx,
-                Some(cancel_rx),
+                Some(control_rx),
             )
             .await;
         });
@@ -1961,7 +2040,7 @@ fn dispatch_pending(
                 agent_index,
                 channel_id: Some(channel_id),
                 recoverable_batch,
-                cancel_tx: Some(cancel_tx),
+                control_tx: Some(control_tx),
             },
         );
         dispatched_channels.push((channel_id, typing_scope));
@@ -2368,7 +2447,7 @@ fn dispatch_heartbeat(
             agent_index,
             channel_id: None,
             recoverable_batch: None,
-            cancel_tx: None,
+            control_tx: None,
         },
     );
     *heartbeat_in_flight = true;
@@ -2685,6 +2764,92 @@ fn build_mcp_servers(config: &Config) -> Vec<McpServer> {
 
 // ── Tests ─────────────────────────────────────────────────────────────────────
 
+#[cfg(test)]
+mod owner_control_command_tests {
+    use super::*;
+    use nostr::{EventBuilder, Keys, Kind, Tag};
+
+    fn make_event(kind: u32, content: &str, p_hex: Option<&str>) -> nostr::Event {
+        let keys = Keys::generate();
+        let tags = match p_hex {
+            Some(hex) => vec![Tag::parse(["p", hex]).expect("p tag")],
+            None => vec![],
+        };
+        EventBuilder::new(Kind::Custom(kind as u16), content)
+            .tags(tags)
+            .sign_with_keys(&keys)
+            .unwrap()
+    }
+
+    #[test]
+    fn owner_control_command_requires_kind_content_and_agent_mention() {
+        let agent = "ab".repeat(32);
+
+        let event = make_event(KIND_STREAM_MESSAGE, " !rotate ", Some(&agent));
+        assert!(is_owner_control_command(
+            &event,
+            KIND_STREAM_MESSAGE,
+            "!rotate",
+            &agent
+        ));
+
+        let wrong_kind = make_event(1, "!rotate", Some(&agent));
+        assert!(!is_owner_control_command(&wrong_kind, 1, "!rotate", &agent));
+
+        let wrong_content = make_event(KIND_STREAM_MESSAGE, "!cancel", Some(&agent));
+        assert!(!is_owner_control_command(
+            &wrong_content,
+            KIND_STREAM_MESSAGE,
+            "!rotate",
+            &agent
+        ));
+
+        let no_mention = make_event(KIND_STREAM_MESSAGE, "!rotate", None);
+        assert!(!is_owner_control_command(
+            &no_mention,
+            KIND_STREAM_MESSAGE,
+            "!rotate",
+            &agent
+        ));
+    }
+
+    #[tokio::test]
+    async fn signal_in_flight_task_sends_rotate_once() {
+        let mut pool = AgentPool::from_slots(vec![]);
+        let channel_id = Uuid::new_v4();
+        let other_channel_id = Uuid::new_v4();
+        let (control_tx, control_rx) = tokio::sync::oneshot::channel();
+
+        let abort_handle = pool.join_set.spawn(async {});
+        pool.task_map_mut().insert(
+            abort_handle.id(),
+            pool::TaskMeta {
+                agent_index: 0,
+                channel_id: Some(channel_id),
+                recoverable_batch: None,
+                control_tx: Some(control_tx),
+            },
+        );
+
+        assert!(!signal_in_flight_task(
+            &mut pool,
+            other_channel_id,
+            ControlSignal::Rotate
+        ));
+        assert!(signal_in_flight_task(
+            &mut pool,
+            channel_id,
+            ControlSignal::Rotate
+        ));
+        assert_eq!(control_rx.await.unwrap(), ControlSignal::Rotate);
+        assert!(!signal_in_flight_task(
+            &mut pool,
+            channel_id,
+            ControlSignal::Rotate
+        ));
+    }
+}
+
 #[cfg(test)]
 mod owner_cache_tests {
     use super::*;

crates/buzz-acp/src/pool.rs

@@ -52,9 +52,9 @@ pub struct TaskMeta {
     pub channel_id: Option<Uuid>,
     /// Clone of batch for Queue mode panic recovery.
     pub recoverable_batch: Option<FlushBatch>,
-    /// Cancel signal for the in-flight prompt task.
+    /// Control signal for the in-flight prompt task.
-    /// `None` for heartbeat tasks (not cancellable) and after signal is consumed.
+    /// `None` for heartbeat tasks (not controllable) and after signal is consumed.
-    pub cancel_tx: Option<tokio::sync::oneshot::Sender<CancelMode>>,
+    pub control_tx: Option<tokio::sync::oneshot::Sender<ControlSignal>>,
 }
 
 /// Agent-level model capabilities. Populated on first session creation.
@@ -117,6 +117,13 @@ impl SessionState {
         self.heartbeat_turn_count = 0;
         self.core_sections.clear();
     }
+
+    #[cfg(test)]
+    fn has_channel_state(&self, channel_id: &Uuid) -> bool {
+        self.sessions.contains_key(channel_id)
+            || self.turn_counts.contains_key(channel_id)
+            || self.core_sections.contains_key(channel_id)
+    }
 }
 
 /// An agent with its session state, owned by the pool or a running task.
@@ -159,13 +166,30 @@ pub enum PromptSource {
     Heartbeat,
 }
 
-/// How an in-flight channel turn should be cancelled.
+/// Apply state effects for Race 1, where a control signal arrives just after the
+/// prompt completed naturally. The prompt result has already been consumed by
+/// `select!`, so the harness must synthesize a successful result while still
+/// honoring any load-bearing control signal semantics.
+fn apply_completed_before_control_signal(
+    state: &mut SessionState,
+    source: &PromptSource,
+    control_signal: ControlSignal,
+) {
+    if control_signal == ControlSignal::Rotate {
+        state.invalidate(source);
+    }
+}
+
+/// Control signal for an in-flight channel turn.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
-pub enum CancelMode {
+pub enum ControlSignal {
     /// Stop the current turn and drop its triggering batch.
-    Stop,
+    Cancel,
     /// Stop the current turn and requeue its triggering batch for a merged re-prompt.
     Interrupt,
+    /// Stop the current turn and drop its triggering batch. The session is
+    /// invalidated just like cancel; the next turn creates a fresh session.
+    Rotate,
 }
 
 /// Outcome of a prompt task.
@@ -609,7 +633,7 @@ pub async fn run_prompt_task(
     prompt_text: Option<String>,
     ctx: Arc<PromptContext>,
     result_tx: mpsc::UnboundedSender<PromptResult>,
-    cancel_rx: Option<tokio::sync::oneshot::Receiver<CancelMode>>,
+    control_rx: Option<tokio::sync::oneshot::Receiver<ControlSignal>>,
 ) {
     // ── Determine source and resolve/create session ───────────────────────
 
@@ -1008,11 +1032,11 @@ pub async fn run_prompt_task(
         None => vec![prompt_text.as_str()],
     };
 
-    // ── Cancel-aware prompt dispatch ──────────────────────────────────────
+    // ── Control-aware prompt dispatch ─────────────────────────────────────
-    // When cancel_rx is Some (channel tasks), wrap the prompt in select! so
+    // When control_rx is Some (channel tasks), wrap the prompt in select! so
-    // the main loop can interrupt it. Heartbeats (cancel_rx=None) take the
+    // the main loop can cancel, interrupt, or rotate it. Heartbeats
-    // simple await path — they are not cancellable.
+    // (control_rx=None) take the simple await path — they are not controllable.
-    let prompt_result = match cancel_rx {
+    let prompt_result = match control_rx {
         None => {
             // Heartbeat / non-cancellable path.
             agent
@@ -1035,8 +1059,8 @@ pub async fn run_prompt_task(
                     ctx.max_turn_duration,
                 ) => result,
                 mode = rx => {
-                    let cancel_mode = mode.unwrap_or(CancelMode::Stop);
+                    let control_signal = mode.unwrap_or(ControlSignal::Cancel);
-                    // Cancel signal received. Guard against Race 1: the turn may
+                    // Control signal received. Guard against Race 1: the turn may
                     // have completed naturally just as cancel fired.
                     if agent.acp.has_in_flight_prompt() {
                         // Prompt is genuinely in-flight — cancel it.
@@ -1051,9 +1075,9 @@ pub async fn run_prompt_task(
                             Ok(stop_reason) => {
                                 log_stop_reason(&source, &stop_reason);
                                 agent.state.invalidate(&source);
-                                let retry_batch = match cancel_mode {
+                                let retry_batch = match control_signal {
-                                    CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
+                                    ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
-                                    CancelMode::Stop => None,
+                                    ControlSignal::Cancel | ControlSignal::Rotate => None,
                                 };
                                 let _ = result_tx.send(PromptResult {
                                     agent,
@@ -1065,9 +1089,9 @@ pub async fn run_prompt_task(
                             }
                             Err(AcpError::AgentExited) => {
                                 agent.state.invalidate_all();
-                                let retry_batch = match cancel_mode {
+                                let retry_batch = match control_signal {
-                                    CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
+                                    ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
-                                    CancelMode::Stop => None,
+                                    ControlSignal::Cancel | ControlSignal::Rotate => None,
                                 };
                                 let _ = result_tx.send(PromptResult {
                                     agent,
@@ -1080,9 +1104,9 @@ pub async fn run_prompt_task(
                             Err(AcpError::IdleTimeout(_) | AcpError::HardTimeout) => {
                                 // Cancel drain timed out — agent state uncertain.
                                 agent.state.invalidate(&source);
-                                let retry_batch = match cancel_mode {
+                                let retry_batch = match control_signal {
-                                    CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
+                                    ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
-                                    CancelMode::Stop => None,
+                                    ControlSignal::Cancel | ControlSignal::Rotate => None,
                                 };
                                 let _ = result_tx.send(PromptResult {
                                     agent,
@@ -1094,9 +1118,9 @@ pub async fn run_prompt_task(
                             }
                             Err(e) => {
                                 agent.state.invalidate(&source);
-                                let retry_batch = match cancel_mode {
+                                let retry_batch = match control_signal {
-                                    CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
+                                    ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
-                                    CancelMode::Stop => None,
+                                    ControlSignal::Cancel | ControlSignal::Rotate => None,
                                 };
                                 let _ = result_tx.send(PromptResult {
                                     agent,
@@ -1122,9 +1146,21 @@ pub async fn run_prompt_task(
                         // and last_prompt_id was cleared by the success path.
                         //
                         // MUST send a PromptResult or the main loop deadlocks.
-                        tracing::debug!(
+                        if control_signal == ControlSignal::Rotate {
-                            target: "pool::prompt",
+                            tracing::debug!(
-                            "cancel signal arrived but turn already completed — treating as success"
+                                target: "pool::prompt",
+                                "rotate signal arrived but turn already completed — invalidating session"
+                            );
+                        } else {
+                            tracing::debug!(
+                                target: "pool::prompt",
+                                "control signal arrived but turn already completed — treating as success"
+                            );
+                        }
+                        apply_completed_before_control_signal(
+                            &mut agent.state,
+                            &source,
+                            control_signal,
                         );
                         let _ = result_tx.send(PromptResult {
                             agent,
@@ -2473,11 +2509,50 @@ mod tests {
         s.sessions.insert(ch_b, "sess-b".into());
         s.turn_counts.insert(ch_a, 5);
         s.turn_counts.insert(ch_b, 3);
+        s.core_sections.insert(ch_a, "core-a".into());
+        s.core_sections.insert(ch_b, "core-b".into());
         s.heartbeat_session = Some("sess-hb".into());
         s.heartbeat_turn_count = 7;
         (s, ch_a, ch_b)
     }
 
+    #[test]
+    fn test_rotate_after_natural_completion_invalidates_channel_state() {
+        let (mut s, ch_a, ch_b) = make_state();
+
+        apply_completed_before_control_signal(
+            &mut s,
+            &PromptSource::Channel(ch_a),
+            ControlSignal::Rotate,
+        );
+
+        assert!(!s.sessions.contains_key(&ch_a));
+        assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.core_sections.contains_key(&ch_a));
+        assert!(!s.has_channel_state(&ch_a));
+        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
+        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
+        assert_eq!(s.heartbeat_session.as_deref(), Some("sess-hb"));
+        assert_eq!(s.heartbeat_turn_count, 7);
+    }
+
+    #[test]
+    fn test_cancel_after_natural_completion_preserves_channel_state() {
+        let (mut s, ch_a, ch_b) = make_state();
+
+        apply_completed_before_control_signal(
+            &mut s,
+            &PromptSource::Channel(ch_a),
+            ControlSignal::Cancel,
+        );
+
+        assert_eq!(s.sessions.get(&ch_a).unwrap(), "sess-a");
+        assert_eq!(*s.turn_counts.get(&ch_a).unwrap(), 5);
+        assert_eq!(s.core_sections.get(&ch_a).unwrap(), "core-a");
+        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
+    }
+
     #[test]
     fn test_invalidate_channel_clears_session_and_turn_count() {
         let (mut s, ch_a, ch_b) = make_state();
@@ -2485,9 +2560,12 @@ mod tests {
 
         assert!(!s.sessions.contains_key(&ch_a));
         assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.core_sections.contains_key(&ch_a));
+        assert!(!s.has_channel_state(&ch_a));
         // ch_b untouched
         assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
         assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
         // heartbeat untouched
         assert_eq!(s.heartbeat_session.as_deref(), Some("sess-hb"));
         assert_eq!(s.heartbeat_turn_count, 7);
@@ -2504,6 +2582,8 @@ mod tests {
         assert_eq!(s.sessions.len(), 2);
         assert_eq!(*s.turn_counts.get(&ch_a).unwrap(), 5);
         assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.core_sections.get(&ch_a).unwrap(), "core-a");
+        assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
     }
 
     #[test]
@@ -2513,6 +2593,7 @@ mod tests {
 
         assert!(s.sessions.is_empty());
         assert!(s.turn_counts.is_empty());
+        assert!(s.core_sections.is_empty());
         assert!(s.heartbeat_session.is_none());
         assert_eq!(s.heartbeat_turn_count, 0);
     }
@@ -2528,6 +2609,8 @@ mod tests {
         assert_eq!(s.turn_counts.len(), 2);
         assert_eq!(*s.turn_counts.get(&ch_a).unwrap(), 5);
         assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.core_sections.get(&ch_a).unwrap(), "core-a");
+        assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
     }
 
     #[test]
@@ -2536,6 +2619,7 @@ mod tests {
         s.invalidate_all(); // should not panic
         assert!(s.sessions.is_empty());
         assert!(s.turn_counts.is_empty());
+        assert!(s.core_sections.is_empty());
     }
 
     #[test]
@@ -2544,9 +2628,12 @@ mod tests {
         assert!(s.invalidate_channel(&ch_a));
         assert!(!s.sessions.contains_key(&ch_a));
         assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.core_sections.contains_key(&ch_a));
+        assert!(!s.has_channel_state(&ch_a));
         // ch_b untouched
         assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
         assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
         // heartbeat untouched
         assert_eq!(s.heartbeat_session.as_deref(), Some("sess-hb"));
         assert_eq!(s.heartbeat_turn_count, 7);
@@ -2573,7 +2660,10 @@ mod tests {
         }
         assert!(!s.sessions.contains_key(&ch_a));
         assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.core_sections.contains_key(&ch_a));
+        assert!(!s.has_channel_state(&ch_a));
         assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
         assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
     }
 }