//! Event queue state machine for buzz-acp.

//!

//! Manages per-channel event queues with per-channel in-flight tracking.

//! When the harness is ready to prompt the agent, it flushes the channel with

//! the oldest pending event, draining ALL events for that channel into a single

//! batch. Multiple channels can be in-flight simultaneously; each channel is

//! independent.

//!

//! ## Dedup modes

//!

//! - **Drop** (default) — while a prompt is in-flight for channel C, new events

//! for channel C are silently dropped (debug-logged). Events for other channels

//! still queue normally.

//! - **Queue** — all events accumulate; batched on the next flush cycle.

use nostr::{Event, ToBech32};

use std::collections::{HashMap, HashSet, VecDeque};

use std::sync::{Arc, OnceLock};

use std::time::{Duration, Instant};

use uuid::Uuid;

use crate::prompt_project::PromptProjectInfo;

use crate::config::DedupMode;

use crate::scope::SessionScope;

/// Maximum events queued per session scope before oldest events are dropped.

///

/// Under the `channel` policy there is exactly one scope per channel, so this

/// is the historical per-channel cap. Under the `thread` policy it caps each

/// thread partition; the channel as a whole is additionally bounded by

/// [`MAX_PENDING_PER_CHANNEL`] so per-thread partitioning cannot multiply the

/// total admitted backlog.

const MAX_PENDING_PER_SCOPE: usize = 500;

/// Aggregate cap on events queued across ALL scopes of a single channel.

///

/// Preserves the pre-thread-scoping backlog protection: moving the per-scope

/// limit to “per thread” must not let one channel with many threads hold an

/// unbounded multiple of the old cap. Equal to [`MAX_PENDING_PER_SCOPE`] so a

/// single-scope channel behaves exactly as before.

const MAX_PENDING_PER_CHANNEL: usize = 500;

/// A key that identifies a queue partition (session scope).

///

/// Lets the queue's public API accept either a bare channel [`Uuid`] (treated

/// as a conversation scope — the pre-thread-scoping default, and what the

/// queue's own unit tests use) or an explicit [`SessionScope`] (what the

/// harness passes once a thread scope has been resolved at admission). This

/// keeps the large existing channel-keyed test suite compiling unchanged while

/// the hot path routes by full scope.

pub trait IntoScope {

/// Convert into the owned [`SessionScope`] used as the partition key.

fn into_scope(self) -> SessionScope;

}

impl IntoScope for SessionScope {

fn into_scope(self) -> SessionScope {

self

}

}

impl IntoScope for &SessionScope {

fn into_scope(self) -> SessionScope {

self.clone()

}

}

impl IntoScope for Uuid {

fn into_scope(self) -> SessionScope {

SessionScope::Conversation { channel_id: self }

}

}

impl IntoScope for &Uuid {

fn into_scope(self) -> SessionScope {

SessionScope::Conversation { channel_id: *self }

}

}

/// Maximum events drained into a single batch.

const MAX_BATCH_EVENTS: usize = 50;

/// Maximum retry attempts before a batch is dead-lettered.

pub(crate) const MAX_RETRIES: u32 = 10;

/// Base retry delay in seconds (doubled each attempt).

const BASE_RETRY_DELAY_SECS: u64 = 5;

/// Cap on retry delay in seconds.

const MAX_RETRY_DELAY_SECS: u64 = 300;

/// Buffer added to `max_turn_duration` to derive the in-flight deadline.

const IN_FLIGHT_DEADLINE_BUFFER_SECS: u64 = 100;

/// Default in-flight deadline: default max_turn (7200s) + 100s buffer.

const DEFAULT_IN_FLIGHT_DEADLINE_SECS: u64 = 7300;

/// An event waiting in the queue.

#[derive(Debug, Clone)]

pub struct QueuedEvent {

pub channel_id: Uuid,

/// Session scope resolved once at admission. Under `channel` policy this is

/// always `Conversation { channel_id }`; under `thread` policy it is the

/// canonical thread scope. The queue partitions on this, never on the

/// channel alone. Invariant: `scope.channel_id() == channel_id`.

pub scope: SessionScope,

pub event: Event,

pub received_at: Instant,

/// Tag identifying which rule (or mode) matched this event.

pub prompt_tag: String,

/// Original-message routing for a kind:40003 edit, resolved once at

/// admission. `None` for ordinary events and for edits whose original

/// could not be fetched (routing then falls back to the edit target id).

pub edit: Option<ResolvedEdit>,

}

impl QueuedEvent {

fn into_batch_event(self) -> BatchEvent {

BatchEvent {

event: self.event,

prompt_tag: self.prompt_tag,

received_at: self.received_at,

edit: self.edit,

}

}

}

/// A single event inside a [`FlushBatch`].

#[derive(Debug, Clone)]

pub struct BatchEvent {

pub event: Event,

pub prompt_tag: String,

pub received_at: Instant,

/// See [`QueuedEvent::edit`].

pub edit: Option<ResolvedEdit>,

}

impl BatchEvent {

/// Event id that replies and reactions for this event should target: the

/// visible original message for an edit, otherwise the event itself.

pub fn routing_event_id(&self) -> String {

reaction_target_id(&self.event)

}

/// Thread tags that route replies for this event. For an edit these are

/// the original message's tags; the edit's own bare `e` tag is never a

/// thread link. An unresolved edit routes as a top-level reply to its

/// target (see [`routing_event_id`](Self::routing_event_id)).

pub fn routing_thread_tags(&self) -> ThreadTags {

routing_thread_tags(&self.event, self.edit.as_ref())

}

/// See [`reply_thread`].

pub fn reply_thread(&self) -> String {

reply_thread(&self.event, self.edit.as_ref())

}

/// See [`ReplyRoute`].

pub fn reply_route(&self) -> ReplyRoute {

ReplyRoute {

root_event_id: self

.routing_thread_tags()

.root_event_id

.map(|root| root.to_ascii_lowercase()),

thread: self.reply_thread(),

}

}

}

/// Where a turn's `<context>` sends ordinary replies, as far as the

/// native-steer guard needs to know.

///

/// A native steer adds a message to a running turn without a new `<context>`,

/// so the turn keeps replying where its own `<context>` points. A message may

/// be steered natively only when that destination is also right for it.

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct ReplyRoute {

/// Routed thread root, lowercase; `None` for a top-level message.

root_event_id: Option<String>,

/// See [`reply_thread`].

thread: String,

}

impl ReplyRoute {

/// Whether a turn whose `<context>` was rendered for `self` also replies

/// correctly to `incoming`.

///

/// In a channel, a top-level message is answered in a new thread rooted at

/// that message, so two messages share a destination only when they share

/// a [`reply_thread`]. In a DM, a top-level message's `<context>` names no

/// reply target (see `format_prompt`): every top-level DM message replies

/// at the top of the same conversation. DM messages therefore share a

/// destination when they share a thread root, including none.

pub fn accepts_steer(&self, incoming: &ReplyRoute, is_dm: bool) -> bool {

if is_dm {

self.root_event_id == incoming.root_event_id

} else {

self.thread == incoming.thread

}

}

}

/// How the running turn's prompt routes replies, as the steer guard needs to

/// know.

///

/// The prompt task records each fact when it formats the turn's `<context>`.

/// The steer guard reads the recorded values instead of deriving them from

/// the incoming event again: channel metadata can become resolvable after the

/// turn starts, and the guard must apply the rule that matches the

/// `<context>` the agent actually received.

#[derive(Debug, Clone, Default)]

pub struct PromptRouting {

is_dm: Arc<OnceLock<bool>>,

trigger_anchor: Arc<OnceLock<bool>>,

}

impl PromptRouting {

/// Record whether the prompt rendered the channel as a DM

/// (`format_prompt`'s `is_dm`). Later calls are ignored: a turn has one

/// `<context>`.

pub fn record_dm(&self, is_dm: bool) {

let _ = self.is_dm.set(is_dm);

}

/// The recorded DM classification; `None` until it is recorded.

pub fn is_dm(&self) -> Option<bool> {

self.is_dm.get().copied()

}

/// Record whether the prompt anchored replies to its triggering message

/// rather than to the thread (see [`reply_anchor_is_trigger`]). Later

/// calls are ignored.

pub fn record_trigger_anchor(&self, trigger_anchor: bool) {

let _ = self.trigger_anchor.set(trigger_anchor);

}

/// The recorded anchor kind; `None` before the prompt is formatted.

pub fn trigger_anchor(&self) -> Option<bool> {

self.trigger_anchor.get().copied()

}

}

/// Reply destination of an in-flight turn, and how its prompt routed replies.

#[derive(Debug, Clone)]

struct InFlightReplyRoute {

route: ReplyRoute,

prompt: PromptRouting,

}

/// Why a batch's prior turn was cancelled — controls how `format_prompt`

/// frames the merged re-prompt.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

pub enum CancelReason {

/// A new request should **supersede** the interrupted work

/// (`MultipleEventHandling::Interrupt`).

Interrupt,

/// A message arrived while the agent was working; it should **continue**

/// and incorporate the message if relevant

/// (`MultipleEventHandling::Steer`, the default mid-turn path).

Steer,

}

/// A batch of events to prompt the agent with.

#[derive(Debug, Clone)]

pub struct FlushBatch {

pub channel_id: Uuid,

/// The single session scope every event in this batch belongs to. Events

/// from different scopes are never combined into one batch.

pub scope: SessionScope,

pub events: Vec<BatchEvent>,

/// Events from a cancelled batch that triggered this re-prompt.

/// Empty for normal (non-cancel) batches. When non-empty, `format_prompt()`

/// produces a merged prompt with annotated sections, framed per

/// [`cancel_reason`](Self::cancel_reason).

pub cancelled_events: Vec<BatchEvent>,

/// How the prior turn was cancelled, when [`cancelled_events`] is non-empty.

/// `None` for normal (non-merge) batches; falls back to the gentler

/// [`Steer`](CancelReason::Steer) framing if a merge somehow lacks a reason

/// (see [`MergeFraming::for_reason`]).

pub cancel_reason: Option<CancelReason>,

}

/// Per-channel event queue with per-channel in-flight enforcement.

///

/// # State Machine

///

/// ```text

/// State:

/// queues: Map<channel_id, VecDeque<QueuedEvent>> (capped at MAX_PENDING_PER_CHANNEL)

/// in_flight_channels: HashSet<Uuid>

/// in_flight_deadlines: Map<channel_id, Instant> (auto-expire after in_flight_deadline)

/// retry_after: Map<channel_id, Instant>

/// retry_counts: Map<channel_id, u32> (dead-letter after MAX_RETRIES)

/// dedup_mode: DedupMode

///

/// Transitions:

/// push(event):

/// if dedup_mode == Drop AND in_flight_channels.contains(event.channel_id):

/// debug log + discard

/// else if queues[channel].len() >= MAX_PENDING_PER_CHANNEL:

/// drop oldest (pop_front), warn, push_back new event

/// else:

/// queues[event.channel_id].push_back(event)

///

/// flush_next() → Option<FlushBatch>:

/// expire any stuck in-flight entries past their deadline

/// candidates = channels where queue non-empty

/// AND NOT in in_flight_channels

/// AND (no retry_after OR retry_after[c] <= now)

/// if candidates empty: return None

/// channel = pick candidate with oldest head event (min received_at)

/// events = drain up to MAX_BATCH_EVENTS from queues[channel]

/// in_flight_channels.insert(channel)

/// in_flight_deadlines.insert(channel, now + in_flight_deadline)

/// return Some(FlushBatch { channel, events })

///

/// mark_complete(channel_id):

/// in_flight_channels.remove(channel_id)

/// in_flight_deadlines.remove(channel_id)

/// retry_counts.remove(channel_id)

/// clean up expired retry_after entry if present

///

/// requeue(batch):

/// increment retry_counts[channel]

/// if retry_counts[channel] > MAX_RETRIES: dead-letter (log ERROR, return batch to caller)

/// else: push_front with original received_at, set exponential backoff retry_after with jitter

/// ```

pub struct EventQueue {

queues: HashMap<SessionScope, VecDeque<QueuedEvent>>,

in_flight_scopes: HashSet<SessionScope>,

/// Per-scope deadline for auto-expiring stuck in-flight entries.

in_flight_deadlines: HashMap<SessionScope, Instant>,

/// Number of events in each in-flight batch (for expiry logging).

in_flight_batch_sizes: HashMap<SessionScope, usize>,

/// Reply route of each in-flight turn: the [`ReplyRoute`] of the batch

/// event whose `<context>` routes the turn's replies (its last event).

in_flight_reply_routes: HashMap<SessionScope, InFlightReplyRoute>,

retry_after: HashMap<SessionScope, Instant>,

/// Per-scope retry attempt counter for exponential backoff / dead-lettering.

retry_counts: HashMap<SessionScope, u32>,

dedup_mode: DedupMode,

/// Events from cancelled batches, keyed by channel. Merged into the next

/// `FlushBatch` for that channel as `cancelled_events` so `format_prompt()`

/// can produce annotated "[Previous request — interrupted]" sections.

cancelled_batches: HashMap<SessionScope, Vec<BatchEvent>>,

/// Why each scope's cancelled batch was cancelled (steer vs interrupt).

/// Set by `requeue_as_cancelled`, consumed by `flush_next` to set

/// `FlushBatch::cancel_reason`. Keyed by scope, cleared on flush.

cancel_reasons: HashMap<SessionScope, CancelReason>,

/// Events withheld from `queues` while a goose-native steer is in flight

/// for that event. Invisible to `flush_next` / `has_flushable_work` /

/// `drain` (the events have been moved out of `queues`), so the queue's

/// no-double-deliver invariant holds without any change to the hot drain

/// path. Populated by [`mark_native_steer_pending`]; drained back to the

/// queue front by [`release_native_steer`] (preserving original

/// `received_at` fairness, same discipline as `requeue_preserve_timestamps`

/// at line 453). Bulk recovery on in-flight deadline expiry is performed

/// by `flush_next` / `has_flushable_work` (recover, not log-and-drop —

/// the events were never delivered to the agent).

withheld_native_steer: HashMap<SessionScope, Vec<QueuedEvent>>,

/// Duration after which an in-flight channel is auto-expired as orphaned.

/// Must be strictly greater than `max_turn_duration` so a turn running to

/// the hard cap returns via `mark_complete` before the backstop fires.

in_flight_deadline: Duration,

}

impl EventQueue {

/// Create a new empty event queue with the given dedup mode.

///

/// Uses [`DEFAULT_IN_FLIGHT_DEADLINE_SECS`] for the in-flight backstop.

/// Call [`with_in_flight_deadline`](Self::with_in_flight_deadline) to

/// derive the deadline from the configured `max_turn_duration`.

pub fn new(dedup_mode: DedupMode) -> Self {

Self {

queues: HashMap::new(),

in_flight_scopes: HashSet::new(),

in_flight_deadlines: HashMap::new(),

in_flight_batch_sizes: HashMap::new(),

in_flight_reply_routes: HashMap::new(),

retry_after: HashMap::new(),

retry_counts: HashMap::new(),

dedup_mode,

cancelled_batches: HashMap::new(),

cancel_reasons: HashMap::new(),

withheld_native_steer: HashMap::new(),

in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS),

}

}

/// Set the in-flight backstop deadline from the configured max turn

/// duration, preserving the 100s buffer for cancel-drain grace + respawn.

pub fn with_in_flight_deadline(mut self, max_turn_duration_secs: u64) -> Self {

self.in_flight_deadline =

Duration::from_secs(max_turn_duration_secs + IN_FLIGHT_DEADLINE_BUFFER_SECS);

self

}

/// Monotonically extend an existing in-flight deadline for `channel_id`.

///

/// Called when a successful steer grants a fresh turn budget. The new

/// deadline is `max(current, now + max_turn_secs + buffer)` — it never

/// moves backward. If the channel is not in-flight (already completed

/// via `mark_complete`), this is a no-op: a late ack never resurrects

/// a deadline.

pub fn extend_in_flight_deadline<K: IntoScope>(&mut self, scope: K, max_turn_secs: u64) {

let scope = scope.into_scope();

if let Some(current) = self.in_flight_deadlines.get_mut(&scope) {

let extended = Instant::now()

+ Duration::from_secs(max_turn_secs + IN_FLIGHT_DEADLINE_BUFFER_SECS);

if extended > *current {

tracing::info!(

channel_id = %scope.channel_id(),

scope = %scope.telemetry_label(),

"extending in-flight deadline by {max_turn_secs}s + {IN_FLIGHT_DEADLINE_BUFFER_SECS}s buffer"

);

*current = extended;

}

}

}

/// Push an event into the queue for its channel.

///

/// In [`DedupMode::Drop`], events for any currently in-flight channel are

/// silently discarded (debug-logged).

///

/// Returns `true` if the event was accepted, `false` if dropped.

pub fn push(&mut self, event: QueuedEvent) -> bool {

debug_assert_eq!(

event.scope.channel_id(),

event.channel_id,

"QueuedEvent.scope must belong to its channel_id"

);

if matches!(self.dedup_mode, DedupMode::Drop)

&& self.in_flight_scopes.contains(&event.scope)

{

tracing::debug!(

channel_id = %event.channel_id,

scope = %event.scope.telemetry_label(),

"dropping event for in-flight scope (drop mode)"

);

return false;

}

let channel_id = event.channel_id;

let scope = event.scope.clone();

let queue = self.queues.entry(scope.clone()).or_default();

// Enforce per-scope depth cap: drop oldest in this partition.

if queue.len() >= MAX_PENDING_PER_SCOPE {

queue.pop_front();

tracing::warn!(

channel_id = %channel_id,

scope = %scope.telemetry_label(),

limit = MAX_PENDING_PER_SCOPE,

"per-scope queue depth cap reached — dropped oldest event"

);

}

queue.push_back(event);

// Enforce the aggregate per-channel cap across all scopes so thread

// partitioning cannot multiply the admitted backlog.

self.enforce_channel_cap(channel_id);

true

}

/// Total queued events across every scope belonging to `channel_id`.

fn channel_event_total(&self, channel_id: Uuid) -> usize {

self.queues

.iter()

.filter(|(s, _)| s.channel_id() == channel_id)

.map(|(_, q)| q.len())

.sum()

}

/// Drop the globally-oldest queued event(s) across a channel's scopes until

/// its aggregate depth is within [`MAX_PENDING_PER_CHANNEL`]. Preserves

/// cross-scope FIFO fairness by always evicting the oldest head event.

fn enforce_channel_cap(&mut self, channel_id: Uuid) {

while self.channel_event_total(channel_id) > MAX_PENDING_PER_CHANNEL {

// Find the channel's scope whose head event is oldest.

let victim = self

.queues

.iter()

.filter(|(s, q)| s.channel_id() == channel_id && !q.is_empty())

.min_by_key(|(_, q)| q.front().unwrap().received_at)

.map(|(s, _)| s.clone());

let Some(scope) = victim else { break };

if let Some(q) = self.queues.get_mut(&scope) {

q.pop_front();

if q.is_empty() {

self.queues.remove(&scope);

}

}

tracing::warn!(

channel_id = %channel_id,

limit = MAX_PENDING_PER_CHANNEL,

"aggregate per-channel queue cap reached — dropped oldest event"

);

}

}

/// Try to flush the next batch.

///

/// Returns `None` if all non-in-flight, non-throttled queues are empty.

/// Otherwise picks the channel with the oldest pending event (FIFO fairness

/// across channels), drains ALL events for that channel into a single batch,

/// inserts into `in_flight_channels`, and returns the batch.

pub fn flush_next(&mut self) -> Option<FlushBatch> {

let now = Instant::now();

// Auto-expire any stuck in-flight entries that missed mark_complete.

let expired: Vec<SessionScope> = self

.in_flight_deadlines

.iter()

.filter(|(_, deadline)| now >= **deadline)

.map(|(scope, _)| scope.clone())

.collect();

for scope in expired {

let lost_events = self.in_flight_batch_sizes.remove(&scope).unwrap_or(0);

tracing::error!(

channel_id = %scope.channel_id(),

scope = %scope.telemetry_label(),

lost_events,

deadline_secs = self.in_flight_deadline.as_secs(),

"BUG: in-flight scope expired without mark_complete — \

auto-releasing; {lost_events} dispatched event(s) orphaned"

);

self.in_flight_scopes.remove(&scope);

self.in_flight_deadlines.remove(&scope);

self.in_flight_reply_routes.remove(&scope);

// Recover any withheld goose-native steer events for the expired

// scope back to the queue front so normal dispatch delivers

// them. Unlike the in-flight batch above (already delivered to a

// now-hung prompt — nothing to recover), these events were never

// delivered to the agent.

self.recover_withheld_for_expired_scope(&scope);

}

// Find the scope whose head event has the oldest received_at,

// excluding in-flight scopes and throttled scopes.

let scope = self

.queues

.iter()

.filter(|(scope, q)| {

!q.is_empty()

&& !self.in_flight_scopes.contains(scope)

&& self.retry_after.get(scope).is_none_or(|&t| t <= now)

})

.min_by_key(|(_, q)| q.front().unwrap().received_at)

.map(|(scope, _)| scope.clone());

// Fallback: if no queued events are ready but a scope has cancelled

// events waiting (e.g., explicit !cancel with no new @mention), flush

// those as a regular batch (re-dispatch unchanged).

let scope = match scope {

Some(scope) => scope,

None => {

let cancelled_scope = self

.cancelled_batches

.keys()

.find(|scope| !self.in_flight_scopes.contains(scope))

.cloned();

let scope = cancelled_scope?;

// Move cancelled events into the regular events slot.

// No new events to merge — re-dispatch the original batch.

let cancelled = self.cancelled_batches.remove(&scope).unwrap_or_default();

let cancel_reason = self.cancel_reasons.remove(&scope);

self.in_flight_scopes.insert(scope.clone());

self.in_flight_deadlines

.insert(scope.clone(), now + self.in_flight_deadline);

self.in_flight_batch_sizes

.insert(scope.clone(), cancelled.len());

self.record_in_flight_reply_route(&scope, &cancelled);

return Some(FlushBatch {

channel_id: scope.channel_id(),

scope,

events: cancelled,

cancelled_events: vec![],

cancel_reason,

});

}

};

let channel_id = scope.channel_id();

// Drain up to MAX_BATCH_EVENTS; leave any remainder in the queue.

let queue = self.queues.entry(scope.clone()).or_default();

let drain_count = MAX_BATCH_EVENTS.min(queue.len());

let mut events: Vec<BatchEvent> = queue

.drain(..drain_count)

.map(QueuedEvent::into_batch_event)

.collect();

// Relay replay delivers stored events newest-first (`ORDER BY

// created_at DESC`), but batch consumers — `format_prompt` scope and

// reply-anchor selection — require the LAST event to be the newest.

// Stable sort: same-second events keep delivery order.

events.sort_by_key(|be| be.event.created_at);

// Remove the queue entry if now empty.

if self.queues.get(&scope).is_some_and(|q| q.is_empty()) {

self.queues.remove(&scope);

}

self.in_flight_scopes.insert(scope.clone());

self.in_flight_deadlines

.insert(scope.clone(), now + self.in_flight_deadline);

self.in_flight_batch_sizes

.insert(scope.clone(), events.len());

self.record_in_flight_reply_route(&scope, &events);

// Merge any cancelled events stored by requeue_as_cancelled().

let cancelled_events = self.cancelled_batches.remove(&scope).unwrap_or_default();

let cancel_reason = if cancelled_events.is_empty() {

self.cancel_reasons.remove(&scope);

None

} else {

self.cancel_reasons.remove(&scope)

};

Some(FlushBatch {

channel_id,

scope,

events,

cancelled_events,

cancel_reason,

})

}

/// Earliest retry throttle for queued work not already in flight.

///

/// Includes expired deadlines: an event-loop iteration can cross eligibility

/// before it arms its timer. The caller must gate on idle pool capacity and

/// dispatch when woken. Dispatch either makes the scope in-flight or releases

/// it with `mark_complete`, which clears the expired throttle even when a

/// busy session owner holds the batch. Empty/removed scopes never arm a timer.

pub fn next_retry_deadline(&self) -> Option<Instant> {

self.retry_after

.iter()

.filter(|(scope, _)| {

!self.in_flight_scopes.contains(*scope)

&& self.queues.get(*scope).is_some_and(|q| !q.is_empty())

})

.map(|(_, &deadline)| deadline)

.min()

}

/// Mark the prompt for `channel_id` as complete.

///

/// Removes the channel from `in_flight_channels` and `in_flight_deadlines`.

///

/// If the channel was NOT requeued (no active `retry_after` throttle), the

/// retry counter is reset — the channel is healthy and the next failure

/// starts fresh. If the channel WAS requeued, `retry_counts` is left intact

/// so the backoff sequence continues on the next attempt.

///

/// Also cleans up any already-expired `retry_after` entry.

pub fn mark_complete<K: IntoScope>(&mut self, scope: K) {

let scope = scope.into_scope();

self.in_flight_scopes.remove(&scope);

self.in_flight_deadlines.remove(&scope);

self.in_flight_batch_sizes.remove(&scope);

self.in_flight_reply_routes.remove(&scope);

let now = Instant::now();

match self.retry_after.get(&scope) {

// Active throttle → scope was requeued; keep retry_counts intact.

Some(&deadline) if deadline > now => {}

// Expired or absent throttle → successful completion; reset counter

// and clean up the stale retry_after entry.

Some(_) => {

self.retry_after.remove(&scope);

self.retry_counts.remove(&scope);

}

None => {

self.retry_counts.remove(&scope);

}

}

}

/// Re-queue a batch of events that failed to process.

///

/// Events are pushed back to the **front** of the channel's queue so they

/// are processed first on the next flush cycle. This prevents event loss

/// when session creation or `session/prompt` fails transiently.

///

/// Original `received_at` timestamps are preserved so the channel retains

/// its fairness position. The retry delay comes from exponential backoff,

/// not from resetting received_at.

///

/// After [`MAX_RETRIES`] attempts the batch is dead-lettered: logged at

/// ERROR and returned to the caller (rather than requeued) so a visible

/// failure notice can be posted to the channel. Returns `None` when the

/// batch was requeued for another attempt.

///

/// Note: does NOT remove from `in_flight_channels` — caller must call

/// `mark_complete` separately.

pub fn requeue(&mut self, batch: FlushBatch) -> Option<FlushBatch> {

let channel_id = batch.channel_id;

let scope = batch.scope.clone();

let attempt = {

let count = self.retry_counts.entry(scope.clone()).or_insert(0);

*count += 1;

*count

};

if attempt > MAX_RETRIES {

tracing::error!(

channel_id = %channel_id,

attempt,

events = batch.events.len(),

"dead-lettering batch after {} retries — discarding {} events",

MAX_RETRIES,

batch.events.len(),

);

self.retry_counts.remove(&scope);

// Also clear retry_after so fresh traffic on this scope isn't

// throttled by stale backoff from the discarded poison batch.

self.retry_after.remove(&scope);

return Some(batch);

}

// Exponential backoff: BASE * 2^(attempt-1), capped at MAX, with ±20% jitter.

let base_secs = BASE_RETRY_DELAY_SECS.saturating_mul(1u64 << (attempt - 1).min(6));

let capped_secs = base_secs.min(MAX_RETRY_DELAY_SECS);

// Jitter: multiply by 0.8..1.2 using subsecond nanos as entropy source.

let jitter = {

let nanos = std::time::SystemTime::now()

.duration_since(std::time::UNIX_EPOCH)

.unwrap_or_default()

.subsec_nanos();

0.8 + (nanos as f64 / u32::MAX as f64) * 0.4

};

let delay = Duration::from_secs_f64(capped_secs as f64 * jitter);

tracing::warn!(

channel_id = %channel_id,

attempt,

max = MAX_RETRIES,

delay_secs = delay.as_secs_f64(),

events = batch.events.len(),

"requeueing failed batch with backoff"

);

let queue = self.queues.entry(scope.clone()).or_default();

// Push to front in reverse order so original order is preserved.

for be in batch.events.into_iter().rev() {

queue.push_front(QueuedEvent {

channel_id,

scope: scope.clone(),

event: be.event,

prompt_tag: be.prompt_tag,

received_at: be.received_at, // preserve original timestamp (#46)

edit: be.edit,

});

}

// Enforce per-scope cap: trim oldest (back) events if requeue pushed

// the partition over the limit. Without this, repeated requeue+push

// cycles can grow the queue unboundedly.

while queue.len() > MAX_PENDING_PER_SCOPE {

queue.pop_back();

tracing::warn!(

channel_id = %channel_id,

scope = %scope.telemetry_label(),

limit = MAX_PENDING_PER_SCOPE,

"requeue overflow — dropped oldest event to enforce cap"

);

}

self.retry_after.insert(scope, Instant::now() + delay);

self.enforce_channel_cap(channel_id);

None

}

/// Re-queue a **complete** flushed batch preserving original `received_at`

/// timestamps.

///

/// Used when a batch was flushed but could not run — no agent was available,

/// or the batch's session-owning worker was busy (thread-scope affinity

/// hold) — so we retry without penalizing the scope's fairness position and

/// without imposing a retry throttle.

///

/// Restores the **entire** batch, not just `events`: any

/// [`cancelled_events`](FlushBatch::cancelled_events) and their

/// [`cancel_reason`](FlushBatch::cancel_reason) are returned to the pending

/// cancelled-carryover so the next flush reconstructs the same merged

/// (interrupt/steer) prompt. Dropping them here would silently lose the

/// original request of an interrupted turn.

///

/// Does NOT set `retry_after`. Does NOT remove from `in_flight_scopes` —

/// caller must call `mark_complete` separately.

pub fn requeue_preserve_timestamps(&mut self, batch: FlushBatch) {

let channel_id = batch.channel_id;

let scope = batch.scope.clone();

// Restore cancelled carryover FIRST so it precedes any carryover a

// concurrent cancel may have already staged for this scope, preserving

// original-before-newer ordering. `flush_next` re-merges it as the next

// batch's `cancelled_events`.

if !batch.cancelled_events.is_empty() {

let existing = self.cancelled_batches.remove(&scope).unwrap_or_default();

let mut restored = batch.cancelled_events;

restored.extend(existing);

self.cancelled_batches.insert(scope.clone(), restored);

if let Some(reason) = batch.cancel_reason {

// Keep the most recent reason if one was already staged.

self.cancel_reasons.entry(scope.clone()).or_insert(reason);

}

}

let queue = self.queues.entry(scope.clone()).or_default();

// Push to front in reverse order so original order is preserved.

for be in batch.events.into_iter().rev() {

queue.push_front(QueuedEvent {

channel_id,

scope: scope.clone(),

event: be.event,

prompt_tag: be.prompt_tag,

received_at: be.received_at,

edit: be.edit,

});

}

// Enforce per-scope cap: trim newest (back) events if over limit.

while queue.len() > MAX_PENDING_PER_SCOPE {

queue.pop_back();

tracing::warn!(

channel_id = %channel_id,

scope = %scope.telemetry_label(),

limit = MAX_PENDING_PER_SCOPE,

"requeue_preserve overflow — dropped newest event to enforce cap"

);

}

self.enforce_channel_cap(channel_id);

}

/// Requeue a cancelled batch so its events appear as `cancelled_events`

/// in the next `FlushBatch` for this channel (enabling the annotated

/// merged-prompt format in `format_prompt()`).

///

/// `reason` records why the turn was cancelled (steer vs interrupt) so the

/// merged prompt is framed correctly. On a double-cancel, the most recent

/// reason wins.

///

/// Unlike `requeue_preserve_timestamps`, events are NOT pushed back into

/// the generic queue — they are stored separately and merged by

/// `flush_next()`. No retry throttle, no backoff.

pub fn requeue_as_cancelled(&mut self, batch: FlushBatch, reason: CancelReason) {

let scope = batch.scope.clone();

let entry = self.cancelled_batches.entry(scope.clone()).or_default();

// Preserve any already-cancelled events from a prior cancel (double-cancel).

entry.extend(batch.cancelled_events);

entry.extend(batch.events);

self.cancel_reasons.insert(scope, reason);

}

/// Returns `true` if any channel has pending events that are not in-flight

/// and not throttled by `retry_after`.

///

/// Also auto-expires any stuck in-flight entries whose deadline has passed.

/// This is a `&mut self` method so expiry can happen without requiring a

/// full `flush_next` call.

pub fn has_flushable_work(&mut self) -> bool {

let now = Instant::now();

// Auto-expire stuck in-flight entries (same logic as flush_next).

let expired: Vec<SessionScope> = self

.in_flight_deadlines

.iter()

.filter(|(_, deadline)| now >= **deadline)

.map(|(scope, _)| scope.clone())

.collect();

for scope in expired {

let lost_events = self.in_flight_batch_sizes.remove(&scope).unwrap_or(0);

tracing::error!(

channel_id = %scope.channel_id(),

scope = %scope.telemetry_label(),

lost_events,

deadline_secs = self.in_flight_deadline.as_secs(),

"BUG: in-flight scope expired without mark_complete — \

auto-releasing; {lost_events} dispatched event(s) orphaned"

);

self.in_flight_scopes.remove(&scope);

self.in_flight_deadlines.remove(&scope);

self.in_flight_reply_routes.remove(&scope);

// Symmetric with the flush_next expiry block: recover withheld

// goose-native steer events for the expired scope so they are

// not permanently orphaned in the side table.

self.recover_withheld_for_expired_scope(&scope);

}

self.queues.iter().any(|(scope, q)| {

!q.is_empty()

&& !self.in_flight_scopes.contains(scope)

&& self.retry_after.get(scope).is_none_or(|&t| t <= now)

}) || self

.cancelled_batches

.keys()

.any(|scope| !self.in_flight_scopes.contains(scope))

}

/// Returns `true` if any undispatched work remains for a channel that is

/// NOT currently in-flight — *including* work held back only by a

/// `retry_after` backoff throttle.

///

/// This is deliberately broader than [`has_flushable_work`](Self::has_flushable_work):

/// that method excludes `retry_after`-throttled channels because they are

/// not flushable *right now*, but the events are still queued and MUST be

/// delivered once the backoff deadline passes. Idle-pool-sleep teardown

/// must gate on this, not on flushability — a failed turn requeued with a

/// future backoff deadline is real queued work, and sleeping on it (while

/// the maintenance timer is disabled and lazy re-wake is itself gated by

/// flushability) would strand the batch until unrelated traffic arrives.

///

/// Covers the three tables where undispatched, non-in-flight work can

/// live: non-empty `queues` (throttled or not), pending `cancelled_batches`,

/// and `withheld_native_steer` events. Read-only (no in-flight expiry) —

/// in-flight liveness is gated separately by [`has_in_flight`](Self::has_in_flight).

pub fn has_undispatched_work(&self) -> bool {

let has_queued = self

.queues

.iter()

.any(|(scope, q)| !q.is_empty() && !self.in_flight_scopes.contains(scope));

let has_cancelled = self

.cancelled_batches

.keys()

.any(|scope| !self.in_flight_scopes.contains(scope));

let has_withheld = self

.withheld_native_steer

.iter()

.any(|(scope, v)| !v.is_empty() && !self.in_flight_scopes.contains(scope));

has_queued || has_cancelled || has_withheld

}

/// Number of pending partitions (session scopes) with queued events.

///

/// Under `channel` policy this equals the number of channels with pending

/// events; under `thread` policy it counts distinct thread partitions.

pub fn pending_channels(&self) -> usize {

self.queues.len()

}

/// Whether `scope` still has work that can be reconstructed into a batch.

///

/// Busy-owner hold timestamps are derived from pending queue state. Queue

/// cap eviction can retire a scope without going through a pool cleanup

/// path, so the dispatch loop uses this seam to prune orphaned holds before

/// scheduling their deadline wakeups.

pub(crate) fn has_pending_scope(&self, scope: &SessionScope) -> bool {

self.queues

.get(scope)

.is_some_and(|queue| !queue.is_empty())

|| self

.cancelled_batches

.get(scope)

.is_some_and(|events| !events.is_empty())

}

/// Number of queued events for a specific scope (or channel, treated as its

/// conversation scope). Test-only.

#[cfg(test)]

pub fn queued_event_count<K: IntoScope>(&self, scope: K) -> usize {

self.queues.get(&scope.into_scope()).map_or(0, |q| q.len())

}

/// Force a channel's retry-attempt counter to `count`, simulating `count`

/// prior failed attempts without needing to drive fake flush/requeue

/// cycles through the queue (which would leave artifact events behind).

/// Test-only — lets integration tests outside this module exercise

/// `requeue()`'s dead-letter threshold directly.

#[cfg(test)]

pub fn set_retry_count_for_test<K: IntoScope>(&mut self, scope: K, count: u32) {

self.retry_counts.insert(scope.into_scope(), count);

}

/// Drop all queued (non-in-flight) events for a channel.

///

/// Used when the agent is removed from a channel — any pending events

/// for that channel are stale and should not be prompted. Does NOT

/// affect in-flight prompts (those will complete normally; the agent

/// may fail to act if it lost access, but that's handled by the relay).

///

/// Also clears any `retry_after` throttle for the channel.

///

/// Returns the visible event IDs that own lifecycle reactions for every

/// dropped event — queued, cancelled carryover, and withheld native

/// steers — so the caller can clean up any 👀 added at queue-push time.

/// Edits report their original message (see [`reaction_target_id`]).

pub fn drain_channel(&mut self, channel_id: Uuid) -> Vec<String> {

// Channel-wide cleanup must find and clear EVERY child thread scope for

// this channel, not just the conversation scope.

let mut ids: Vec<String> = Vec::new();

let mut seen: HashSet<String> = HashSet::new();

let mut collect = |event: &Event| {

let id = reaction_target_id(event);

if seen.insert(id.clone()) {

ids.push(id);

}

};

self.queues.retain(|s, q| {

if s.channel_id() != channel_id {

return true;

}

q.iter().for_each(|e| collect(&e.event));

false

});

self.cancelled_batches.retain(|s, events| {