jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 13:01:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add manual ACP session rotation command (#932)

Browse Source 
Signed-off-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: Max <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
tlongwell-block
2026-06-12 11:50:44 -04:00
committed by GitHub
parent df8896f13e
commit 00dc4915d4
3 changed files with 320 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
crates/buzz-acp/README.md
View File

@@ -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.

215
crates/buzz-acp/src/lib.rs
View File

@@ -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
&& buzz_event.event.content.trim() == "!shutdown"
&& buzz_event.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(pubkey_hex.as_str())
});
let is_shutdown = is_owner_control_command(
&buzz_event.event,
kind_u32,
"!shutdown",
&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
&& buzz_event.event.content.trim() == "!cancel"
&& buzz_event.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(pubkey_hex.as_str())
});
let is_cancel = is_owner_control_command(
&buzz_event.event,
kind_u32,
"!cancel",
&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::*;

148
crates/buzz-acp/src/pool.rs
View File

@@ -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.
/// `None` for heartbeat tasks (not cancellable) and after signal is consumed.
pub cancel_tx: Option<tokio::sync::oneshot::Sender<CancelMode>>,
/// Control signal for the in-flight prompt task.
/// `None` for heartbeat tasks (not controllable) and after signal is consumed.
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 ──────────────────────────────────────
// When cancel_rx is Some (channel tasks), wrap the prompt in select! so
// the main loop can interrupt it. Heartbeats (cancel_rx=None) take the
// simple await path — they are not cancellable.
let prompt_result = match cancel_rx {
// ── Control-aware prompt dispatch ─────────────────────────────────────
// When control_rx is Some (channel tasks), wrap the prompt in select! so
// the main loop can cancel, interrupt, or rotate it. Heartbeats
// (control_rx=None) take the simple await path — they are not controllable.
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);
// Cancel signal received. Guard against Race 1: the turn may
let control_signal = mode.unwrap_or(ControlSignal::Cancel);
// 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 {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
let retry_batch = match control_signal {
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
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 {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
let retry_batch = match control_signal {
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
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 {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
let retry_batch = match control_signal {
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
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 {
CancelMode::Interrupt => requeue_batch_if_queue(&ctx, batch),
CancelMode::Stop => None,
let retry_batch = match control_signal {
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
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!(
target: "pool::prompt",
"cancel signal arrived but turn already completed — treating as success"
if control_signal == ControlSignal::Rotate {
tracing::debug!(
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");
}
}