fix(acp): close PromptChannelGuard reload race and stale-event leak

CHANGELOG.md

@@ -134,6 +134,25 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   strings that enable both `session` and `acp-http` together, so a narrower single bundle like
   `ide` was never checked with `--all-targets` anywhere, letting test-only `#[cfg(...)]` gates
   narrower than the production code they exercise ship and pass CI indefinitely.
+- `zeph-acp`: fixed two follow-on hazards in `PromptChannelGuard` (issues #6666, #6667,
+  discovered while reviewing #6661's fix). First, `do_close_session`/`do_delete_session` remove
+  a session's entry without waiting for or aborting any turn still in flight on it; if that
+  session is then reloaded/resumed under the same `SessionId` before the stale guard is dropped,
+  the guard's `Drop` would find the fresh entry and overwrite its new, live `output_rx` with the
+  old, now-dead receiver from the turn that predated the close. `SessionEntry` now carries a
+  `generation` stamp assigned at construction; the guard captures it at acquisition time and
+  skips the restore if the entry's current generation no longer matches. Second, a receiver
+  restored after the enclosing task was aborted mid-`drain_agent_events` could carry over
+  `LoopbackEvent`s the still-running agent loop had already queued (including a legitimate
+  `Flush`), so the next `session/prompt` call on that session could receive one of these stale
+  events as its first `rx.recv()` instead of a genuinely new one — and since `drain_agent_events`
+  returns on the turn's first `Flush` while the agent loop can keep running and queue further
+  events afterward (e.g. a second `Flush` from a post-response self-check), draining once at
+  `Drop` time only caught the events queued by that instant. `acquire_prompt_channels` now also
+  drains under the sessions lock, right after confirming no turn is currently in flight on that
+  session — every event still queued at that point is provably an orphan from a prior turn —
+  closing the whole inter-turn window instead of one snapshot; `Drop`'s drain remains as a cheap
+  early filter.
 
 ## [0.22.3] - 2026-07-22
 ### Fixed

crates/zeph-acp/src/agent/mod.rs

@@ -265,11 +265,28 @@ pub(crate) struct SessionConfigSeed {
     temperature_preset: zeph_config::AcpTemperaturePreset,
 }
 
+/// Monotonic counter assigned to every `SessionEntry` at construction (`make_session_entry`).
+///
+/// Lets `turn::PromptChannelGuard`, captured at the start of a turn, detect at restore time
+/// whether the session map still holds the *same* entry it started with. `do_load_session` /
+/// `do_resume_session` both early-return without inserting anything if the `SessionId` is
+/// already present in the map — so a fresh `SessionEntry` only ever lands under an id that a
+/// prior `do_close_session`/`do_delete_session` has already `remove()`-d. Neither of those
+/// removals waits for or aborts any turn still in flight on that session, so a
+/// `PromptChannelGuard` acquired before the close can outlive it and still be holding the
+/// (now orphaned) receiver when the id is reloaded/resumed (#6666). The fresh entry gets a new
+/// generation, so the stale guard's later `Drop` can tell its receiver is no longer the live
+/// one and skip clobbering the reloaded session's `output_rx`.
+static SESSION_ENTRY_GENERATION: AtomicU64 = AtomicU64::new(0);
+
 pub(crate) struct SessionEntry {
     pub(crate) input_tx: mpsc::Sender<ChannelMessage>,
     /// Receiver is owned solely by the `prompt()` handler.
     /// `Mutex` instead of `RefCell` so `SessionEntry` is `Send`.
     pub(crate) output_rx: Mutex<Option<mpsc::Receiver<LoopbackEvent>>>,
+    /// Identity stamp from [`SESSION_ENTRY_GENERATION`], assigned once at construction.
+    /// See that constant's doc for why this exists.
+    pub(crate) generation: u64,
     pub(crate) cancel_signal: Arc<tokio::sync::Notify>,
     /// Epoch milliseconds; updated on every prompt.
     pub(crate) last_active_ms: AtomicU64,

crates/zeph-acp/src/agent/session.rs

@@ -33,9 +33,9 @@ use super::elicitation;
 #[cfg(test)]
 use super::{AgentSpawner, ZephAcpAgent};
 use super::{
-    DEFAULT_MODE_ID, NotifyReceiver, NotifySender, SessionConfigSeed, SessionContext, SessionEntry,
-    ZephAcpAgentState, build_config_options, build_mode_state, model_meta,
-    session_event_to_updates,
+    DEFAULT_MODE_ID, NotifyReceiver, NotifySender, SESSION_ENTRY_GENERATION, SessionConfigSeed,
+    SessionContext, SessionEntry, ZephAcpAgentState, build_config_options, build_mode_state,
+    model_meta, session_event_to_updates,
 };
 
 const LOOPBACK_CHANNEL_CAPACITY: usize = 64;
@@ -1255,6 +1255,7 @@ impl ZephAcpAgentState {
         SessionEntry {
             input_tx: handle.input_tx,
             output_rx: Mutex::new(Some(handle.output_rx)),
+            generation: SESSION_ENTRY_GENERATION.fetch_add(1, Ordering::Relaxed),
             cancel_signal: handle.cancel_signal,
             last_active_ms: AtomicU64::new(now_ms),
             created_at: chrono::Utc::now(),