Skip to content

fix(acp): acknowledge consumed owner commands in channel #3877

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tcballard wants to merge 1 commit into
base: main
Choose a base branch
from
+101 −17
Open

fix(acp): acknowledge consumed owner commands in channel #3877

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 93 additions & 8 deletions crates/buzz-acp/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2044,6 +2044,15 @@ async fn tokio_main() -> Result<()> {
sender = %buzz_event.event.pubkey.to_hex(),
"shutdown command from owner — exiting gracefully"
);
let thread_tags =
queue::parse_thread_tags(&buzz_event.event);
pool::post_notice(
&ctx.rest_client,
buzz_event.channel_id,
&thread_tags,
SHUTDOWN_NOTICE,
)
.await;
let _ = shutdown_tx.send(());
continue;
}
Expand Down Expand Up @@ -2074,12 +2083,19 @@ async fn tokio_main() -> Result<()> {
buzz_event.channel_id,
ControlSignal::Cancel,
);
let notice = cancel_notice(fired);
if !fired {
tracing::warn!(
channel_id = %buzz_event.channel_id,
"!cancel received but no in-flight task — no-op"
);
}
spawn_notice(
Some(&ctx.rest_client),
buzz_event.channel_id,
queue::parse_thread_tags(&buzz_event.event),
notice.to_string(),
);
continue; // consume event — do NOT push to queue
}
}
Expand Down Expand Up @@ -2112,19 +2128,27 @@ async fn tokio_main() -> Result<()> {
buzz_event.channel_id,
ControlSignal::Rotate,
);
if fired {
let invalidated = if fired {
tracing::info!(
channel_id = %buzz_event.channel_id,
"!rotate received — cancelling in-flight turn and rotating session"
);
0
} 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)"
);
}
invalidated
};
spawn_notice(
Some(&ctx.rest_client),
buzz_event.channel_id,
queue::parse_thread_tags(&buzz_event.event),
rotate_notice(fired, invalidated).to_string(),
);
continue; // consume event — do NOT push to queue
}
}
Expand Down Expand Up @@ -3028,10 +3052,45 @@ fn is_auth_error(error: &acp::AcpError) -> bool {
message.contains("Re-authenticate") || message.contains("API Error: 401")
}

const SHUTDOWN_NOTICE: &str = "Shutting down.";

fn cancel_notice(turn_active: bool) -> &'static str {
if turn_active {
"Cancelled the current turn."
} else {
"Nothing to cancel — no turn in flight."
}
}

fn rotate_notice(turn_active: bool, invalidated_sessions: usize) -> &'static str {
if turn_active {
"Cancelled the current turn — the next one starts from a fresh session."
} else if invalidated_sessions > 0 {
"Session rotated — the next turn starts fresh."
} else {
"No session to rotate — the next turn already starts fresh."
}
}

/// Spawn a task that posts a user-visible notice to the relay.
fn spawn_notice(
rest_client: Option<&relay::RestClient>,
channel_id: Uuid,
thread_tags: ThreadTags,
content: String,
) {
if let Some(rest) = rest_client {
let rest = rest.clone();
tokio::spawn(async move {
pool::post_notice(&rest, channel_id, &thread_tags, &content).await;
});
}
}

/// Spawn a task that posts a user-visible failure notice to the relay.
///
/// Shared by the hard-cap immediate dead-letter path and the retries-exhausted
/// dead-letter path so neither duplicates the tokio::spawn block.
/// dead-letter path so neither duplicates the thread-tag extraction.
fn spawn_failure_notice(
rest_client: Option<&relay::RestClient>,
batch: &FlushBatch,
Expand All @@ -3043,11 +3102,7 @@ fn spawn_failure_notice(
.last()
.map(|be| queue::parse_thread_tags(&be.event))
.unwrap_or_default();
let rest = rest.clone();
let channel_id = batch.channel_id;
tokio::spawn(async move {
pool::post_failure_notice(&rest, channel_id, &thread_tags, &content).await;
});
spawn_notice(Some(rest), batch.channel_id, thread_tags, content);
}
}

Expand Down Expand Up @@ -4314,6 +4369,36 @@ mod owner_control_command_tests {
));
}

#[test]
fn cancel_notices_distinguish_active_and_idle_turns() {
assert_eq!(cancel_notice(true), "Cancelled the current turn.");
assert_eq!(
cancel_notice(false),
"Nothing to cancel — no turn in flight."
);
}

#[test]
fn rotate_notices_distinguish_active_cached_and_empty_sessions() {
assert_eq!(
rotate_notice(true, 0),
"Cancelled the current turn — the next one starts from a fresh session."
);
assert_eq!(
rotate_notice(false, 1),
"Session rotated — the next turn starts fresh."
);
assert_eq!(
rotate_notice(false, 0),
"No session to rotate — the next turn already starts fresh."
);
}

#[test]
fn shutdown_notice_is_concise() {
assert_eq!(SHUTDOWN_NOTICE, "Shutting down.");
}

#[test]
fn mode_gate_signal_maps_handling_to_control_signal() {
let owner = "a".repeat(64);
Expand Down
17 changes: 8 additions & 9 deletions crates/buzz-acp/src/pool.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3800,11 +3800,10 @@ pub(crate) async fn reaction_add(rest: &crate::relay::RestClient, event_id: &str
}
}

/// Best-effort: post a visible failure notice (kind:9) to a channel after a
/// batch is dead-lettered. Replies into the thread of `thread_tags` when the
/// triggering event was threaded. Errors are logged and swallowed — the
/// notice must never take down the main loop.
pub(crate) async fn post_failure_notice(
/// Best-effort: post a visible notice (kind:9) to a channel. Replies into the
/// thread of `thread_tags` when the triggering event was threaded. Errors are
/// logged and swallowed — the notice must never take down the main loop.
pub(crate) async fn post_notice(
rest: &crate::relay::RestClient,
channel_id: Uuid,
thread_tags: &ThreadTags,
Expand All @@ -3826,21 +3825,21 @@ pub(crate) async fn post_failure_notice(
match buzz_sdk::build_message(channel_id, content, thread_ref.as_ref(), &[], false, &[]) {
Ok(b) => b,
Err(e) => {
tracing::warn!(channel = %channel_id, "failure notice: build failed: {e}");
tracing::warn!(channel = %channel_id, "notice: build failed: {e}");
return;
}
};
let event = match builder.sign_with_keys(&rest.keys) {
Ok(e) => e,
Err(e) => {
tracing::warn!(channel = %channel_id, "failure notice: sign failed: {e}");
tracing::warn!(channel = %channel_id, "notice: sign failed: {e}");
return;
}
};
match tokio::time::timeout(Duration::from_secs(5), rest.submit_event(&event)).await {
Ok(Ok(_)) => {}
Ok(Err(e)) => tracing::warn!(channel = %channel_id, "failure notice failed: {e}"),
Err(_) => tracing::warn!(channel = %channel_id, "failure notice timed out"),
Ok(Err(e)) => tracing::warn!(channel = %channel_id, "notice failed: {e}"),
Err(_) => tracing::warn!(channel = %channel_id, "notice timed out"),
}
}

Expand Down