feat(core): add /new command to reset conversation while preserving session state (#2464)

Implements /new slash command that resets conversation history, tool call
cache, and context budget without reinitializing memory (SQLite + Qdrant),
MCP connections, or provider configuration.

- Fail-fast: new conversation ID created in SQLite before any state mutation
- Aborts in-flight JoinHandles (compression, sidequest, subgoal tasks)
- Cancels running plan token and shuts down active subagents
- Optional --no-digest flag skips carry-over session digest generation
- Optional --keep-plan flag preserves pending orchestration graph
- Background digest generation (30s bounded, fire-and-forget)
- TUI spinner emitted via status_tx before and after reset
- Channel-generic: works identically across CLI, TUI, and Telegram
- 3 unit tests for reset_compaction, FocusState::reset, SidequestState::reset

Closes #2451

Author: bug-ops

CHANGELOG.md

@@ -8,6 +8,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- feat(core): `/new` slash command — resets conversation context (messages, compaction state, tool caches, focus/sidequest, pending plans) while preserving memory, MCP connections, providers, and skills; creates a new `ConversationId` in SQLite for audit trail; generates a session digest for the outgoing conversation fire-and-forget unless `--no-digest` is passed; active sub-agents and background compression tasks are cancelled; `--keep-plan` preserves a pending plan graph; available in all channels (CLI, TUI, Telegram) via the unified `handle_builtin_command` path (closes #2451)
+
 - feat(acp): expose current model in `session/list` and emit `SessionInfoUpdate` on model change — each in-memory `SessionInfo` now carries `meta.currentModel`; after `session/set_config_option` with `configId=model` a `SessionInfoUpdate` notification with `meta.currentModel` is sent in addition to the existing `ConfigOptionUpdate`; same notification is sent after `session/set_session_model` (closes #2435)
 - feat(tools): adversarial policy agent — LLM-based pre-execution tool call validation against plain-language policies; configurable fail-closed/fail-open behavior (`fail_open = false` default); prompt injection hardening via code-fence param quoting; strict allow/deny response parsing; full `ToolExecutor` trait delegation; audit log `adversarial_policy_decision` field; executor chain order `PolicyGateExecutor → AdversarialPolicyGateExecutor → TrustGateExecutor`; gated on `policy-enforcer` feature; config `[tools.adversarial_policy]` (closes #2447)
 - feat(memory): Memex tool output archive — before compaction, `ToolOutput` bodies in the compaction range are saved to `tool_overflow` with `archive_type = 'archive'`; archived UUIDs are appended as a postfix after LLM summarization so references survive compaction; controlled by `[memory.compression] archive_tool_outputs = false`; archives are excluded from the short-lived cleanup job via `archive_type` column (migration 054, closes #2432)

crates/zeph-core/src/agent/context/assembly.rs

@@ -366,6 +366,177 @@ impl<C: Channel> Agent<C> {
             .retain(|m| m.role != Role::User || !m.content.starts_with(SESSION_DIGEST_PREFIX));
     }
 
+    /// Spawn a fire-and-forget background task to generate and persist a session digest for
+    /// `conversation_id`. No-op when digest is disabled or the conversation has no messages.
+    fn spawn_outgoing_digest(&self, conversation_id: Option<zeph_memory::ConversationId>) {
+        if !self.memory_state.digest_config.enabled {
+            return;
+        }
+        let non_system: Vec<_> = self
+            .msg
+            .messages
+            .iter()
+            .skip(1)
+            .filter(|m| m.role != zeph_llm::provider::Role::System)
+            .cloned()
+            .collect();
+        if non_system.is_empty() {
+            return;
+        }
+        let digest_config = self.memory_state.digest_config.clone();
+        let memory = self.memory_state.memory.clone();
+        let provider = self.provider.clone();
+        let tc = self.metrics.token_counter.clone();
+        let status_tx = self.session.status_tx.clone();
+        if let Some(ref tx) = self.session.status_tx {
+            let _ = tx.send("Generating session digest...".to_string());
+        }
+        tokio::spawn(async move {
+            if let (Some(mem), Some(cid)) = (memory, conversation_id) {
+                super::super::session_digest::generate_and_store_digest(
+                    &provider,
+                    &mem,
+                    cid,
+                    &non_system,
+                    &digest_config,
+                    &tc,
+                )
+                .await;
+            }
+            if let Some(tx) = status_tx {
+                let _ = tx.send(String::new());
+            }
+        });
+    }
+
+    /// Reset the conversation window for `/new`.
+    ///
+    /// Creates a new `ConversationId` in `SQLite` first (fail-fast: no state is mutated
+    /// if the `DB` call fails). Then resets all session-scoped state while preserving
+    /// cross-session state (memory, MCP, providers, skills).
+    ///
+    /// `keep_plan` — when `true`, `orchestration.pending_graph` is preserved.
+    /// `no_digest` — when `true`, skip generating a session digest for the outgoing
+    ///               conversation. Default behaviour: generate digest fire-and-forget.
+    ///
+    /// Returns the old and new `ConversationId` for the confirmation message.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if [`create_conversation`](zeph_memory::store::SqliteStore::create_conversation)
+    /// fails. In that case no agent state is modified.
+    pub(in crate::agent) async fn reset_conversation(
+        &mut self,
+        keep_plan: bool,
+        no_digest: bool,
+    ) -> Result<
+        (
+            Option<zeph_memory::ConversationId>,
+            Option<zeph_memory::ConversationId>,
+        ),
+        super::super::error::AgentError,
+    > {
+        // --- Step 1: create new ConversationId FIRST (fail-fast) ---
+        let new_conversation_id = if let Some(ref memory) = self.memory_state.memory {
+            match memory.sqlite().create_conversation().await {
+                Ok(id) => Some(id),
+                Err(e) => return Err(super::super::error::AgentError::Memory(e)),
+            }
+        } else {
+            None
+        };
+
+        let old_conversation_id = self.memory_state.conversation_id;
+
+        // --- Step 2: fire-and-forget digest for outgoing conversation ---
+        if !no_digest {
+            self.spawn_outgoing_digest(old_conversation_id);
+        }
+
+        // --- Step 3: TUI status ---
+        if let Some(ref tx) = self.session.status_tx {
+            let _ = tx.send("Resetting conversation...".to_string());
+        }
+
+        // --- Step 4: abort background compression tasks (context-compression) ---
+        #[cfg(feature = "context-compression")]
+        {
+            if let Some(h) = self.compression.pending_task_goal.take() {
+                h.abort();
+            }
+            if let Some(h) = self.compression.pending_sidequest_result.take() {
+                h.abort();
+            }
+            if let Some(h) = self.compression.pending_subgoal.take() {
+                h.abort();
+            }
+            self.compression.current_task_goal = None;
+            self.compression.task_goal_user_msg_hash = None;
+            self.compression.subgoal_registry =
+                crate::agent::compaction_strategy::SubgoalRegistry::default();
+            self.compression.subgoal_user_msg_hash = None;
+        }
+
+        // --- Step 5: cancel running plan and clear orchestration ---
+        if !keep_plan {
+            if let Some(token) = self.orchestration.plan_cancel_token.take() {
+                token.cancel();
+            }
+            self.orchestration.pending_graph = None;
+            self.orchestration.pending_goal_embedding = None;
+        }
+        // Cancel running sub-agents regardless of keep_plan.
+        if let Some(ref mut mgr) = self.orchestration.subagent_manager {
+            mgr.shutdown_all();
+        }
+
+        // --- Step 6: reset message history and caches ---
+        self.clear_history();
+        self.tool_orchestrator.clear_cache();
+
+        // Drain message queue, logging discarded entries.
+        let discarded = self.clear_queue();
+        if discarded > 0 {
+            tracing::debug!(
+                discarded,
+                "/new: discarded queued messages that arrived during reset"
+            );
+        }
+        self.msg.pending_image_parts.clear();
+
+        // --- Step 7: reset security URL sets ---
+        if let Ok(mut urls) = self.security.user_provided_urls.write() {
+            urls.clear();
+        }
+        self.security.flagged_urls.clear();
+
+        // --- Step 8: reset compaction and compression states ---
+        self.context_manager.reset_compaction();
+        self.focus.reset();
+        self.sidequest.reset();
+
+        // --- Step 9: reset misc session-scoped fields ---
+        self.debug_state.iteration_counter = 0;
+        self.last_persisted_message_id = None;
+        self.deferred_db_hide_ids.clear();
+        self.deferred_db_summaries.clear();
+        self.cached_filtered_tool_ids = None;
+        self.providers.cached_prompt_tokens = 0;
+
+        // --- Step 10: update conversation ID and memory state ---
+        self.memory_state.conversation_id = new_conversation_id;
+        self.memory_state.unsummarized_count = 0;
+        // Clear cached digest — the new conversation has no prior digest yet.
+        self.memory_state.cached_session_digest = None;
+
+        // --- Step 11: clear TUI status ---
+        if let Some(ref tx) = self.session.status_tx {
+            let _ = tx.send(String::new());
+        }
+
+        Ok((old_conversation_id, new_conversation_id))
+    }
+
     async fn fetch_document_rag(
         memory_state: &MemoryState,
         query: &str,

crates/zeph-core/src/agent/context_manager.rs

@@ -151,6 +151,15 @@ impl ContextManager {
         }
     }
 
+    /// Reset compaction state for a new conversation.
+    ///
+    /// Clears cooldown, exhaustion, and turn counters so the new conversation starts
+    /// with a clean compaction slate.
+    pub(crate) fn reset_compaction(&mut self) {
+        self.compaction = CompactionState::Ready;
+        self.turns_since_last_hard_compaction = None;
+    }
+
     /// Determine which compaction tier applies for the given token count.
     ///
     /// - `Hard` when `cached_tokens > budget * hard_compaction_threshold`
@@ -450,4 +459,14 @@ mod tests {
         let after_eviction = CompactionState::CompactedThisTurn { cooldown: 0 };
         assert!(after_eviction.is_compacted_this_turn());
     }
+
+    #[test]
+    fn reset_compaction_clears_exhausted_state() {
+        let mut cm = ContextManager::new();
+        cm.compaction = CompactionState::Exhausted { warned: true };
+        cm.turns_since_last_hard_compaction = Some(5);
+        cm.reset_compaction();
+        assert_eq!(cm.compaction, CompactionState::Ready);
+        assert!(cm.turns_since_last_hard_compaction.is_none());
+    }
 }

crates/zeph-core/src/agent/focus.rs

@@ -166,6 +166,23 @@ impl FocusState {
         }
     }
 
+    /// Reset focus state for a new conversation.
+    ///
+    /// Clears the active session and accumulated knowledge so the new conversation
+    /// starts without stale focus context.
+    pub(crate) fn reset(&mut self) {
+        self.knowledge_blocks.clear();
+        #[cfg(feature = "context-compression")]
+        {
+            self.active_marker = None;
+            self.compressing
+                .store(false, std::sync::atomic::Ordering::Release);
+        }
+        self.active_scope = None;
+        self.turns_since_focus = 0;
+        self.turns_since_reminder = 0;
+    }
+
     /// Increment turn counters. Called at the start of each user-message turn.
     pub(crate) fn tick(&mut self) {
         self.turns_since_focus = self.turns_since_focus.saturating_add(1);
@@ -450,4 +467,18 @@ mod tests {
         assert!(msg.content.contains("Summary B"));
         assert!(msg.metadata.focus_pinned, "knowledge block must be pinned");
     }
+
+    #[test]
+    fn reset_clears_all_session_state() {
+        let mut state = FocusState::new(FocusConfig::default());
+        state.knowledge_blocks.push("some knowledge".to_string());
+        state.active_scope = Some("active scope".to_string());
+        state.turns_since_focus = 7;
+        state.turns_since_reminder = 3;
+        state.reset();
+        assert!(state.knowledge_blocks.is_empty());
+        assert!(state.active_scope.is_none());
+        assert_eq!(state.turns_since_focus, 0);
+        assert_eq!(state.turns_since_reminder, 0);
+    }
 }

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

@@ -2646,6 +2646,7 @@ impl<C: Channel> Agent<C> {
     /// Returns `Some(true)` to break the loop (exit), `Some(false)` to continue to the next
     /// iteration, or `None` if the command was not recognized (caller should call
     /// `process_user_message`).
+    #[allow(clippy::too_many_lines)]
     async fn handle_builtin_command(
         &mut self,
         trimmed: &str,
@@ -2689,6 +2690,31 @@ impl<C: Channel> Agent<C> {
             return Ok(Some(false));
         }
 
+        if trimmed == "/new" || trimmed.starts_with("/new ") {
+            let args = trimmed.strip_prefix("/new").unwrap_or("").trim();
+            let keep_plan = args.split_whitespace().any(|a| a == "--keep-plan");
+            let no_digest = args.split_whitespace().any(|a| a == "--no-digest");
+            match self.reset_conversation(keep_plan, no_digest).await {
+                Ok((old_id, new_id)) => {
+                    let old = old_id.map_or_else(|| "none".to_string(), |id| id.0.to_string());
+                    let new = new_id.map_or_else(|| "none".to_string(), |id| id.0.to_string());
+                    let keep_note = if keep_plan { " (plan preserved)" } else { "" };
+                    self.channel
+                        .send(&format!(
+                            "New conversation started. Previous: {old} → Current: {new}{keep_note}"
+                        ))
+                        .await?;
+                }
+                Err(e) => {
+                    self.channel
+                        .send(&format!("Failed to start new conversation: {e}"))
+                        .await?;
+                }
+            }
+            let _ = self.channel.flush_chunks().await;
+            return Ok(Some(false));
+        }
+
         if trimmed == "/clear" {
             self.clear_history();
             self.tool_orchestrator.clear_cache();

crates/zeph-core/src/agent/session_digest.rs

@@ -69,6 +69,94 @@ use crate::channel::Channel;
 
 use super::Agent;
 
+/// Generate and persist a digest for a completed conversation from a background task.
+///
+/// Called fire-and-forget from `reset_conversation`. All errors are logged as warnings
+/// and swallowed.
+pub(super) async fn generate_and_store_digest(
+    provider: &zeph_llm::any::AnyProvider,
+    memory: &zeph_memory::semantic::SemanticMemory,
+    conversation_id: zeph_memory::ConversationId,
+    messages: &[zeph_llm::provider::Message],
+    digest_config: &crate::config::DigestConfig,
+    tc: &zeph_memory::TokenCounter,
+) {
+    if messages.is_empty() {
+        return;
+    }
+
+    let max_input = digest_config.max_input_messages;
+    let max_tokens = digest_config.max_tokens;
+
+    let slice = if messages.len() > max_input {
+        &messages[messages.len() - max_input..]
+    } else {
+        messages
+    };
+
+    let mut conv_text = String::new();
+    for msg in slice {
+        let role = match msg.role {
+            zeph_llm::provider::Role::User => "User",
+            zeph_llm::provider::Role::Assistant => "Assistant",
+            zeph_llm::provider::Role::System => "System",
+        };
+        let _ =
+            std::fmt::Write::write_fmt(&mut conv_text, format_args!("{role}: {}\n\n", msg.content));
+    }
+
+    let prompt = format!(
+        "You are a session summarizer. Read the following conversation excerpt and produce \
+         a compact digest (under {max_tokens} tokens) of the key facts, decisions, outcomes, \
+         and open questions from this session. Be specific and concise. \
+         Output ONLY the digest text, no preamble.\n\n\
+         Conversation:\n{conv_text}\n\
+         Digest:"
+    );
+
+    let chat_messages = vec![zeph_llm::provider::Message {
+        role: zeph_llm::provider::Role::User,
+        content: prompt,
+        parts: vec![],
+        metadata: zeph_llm::provider::MessageMetadata::default(),
+    }];
+
+    let timeout = Duration::from_secs(30);
+    let digest_text = tokio::select! {
+        () = async { tokio::time::sleep(timeout).await } => {
+            tracing::warn!("session digest (/new): LLM call timed out");
+            return;
+        }
+        result = provider.chat_with_named_provider(&digest_config.provider, &chat_messages) => {
+            match result {
+                Ok(text) => text,
+                Err(e) => {
+                    tracing::warn!("session digest (/new): LLM call failed: {e:#}");
+                    return;
+                }
+            }
+        }
+    };
+
+    let sanitized = sanitize_digest(&digest_text);
+    let final_text = truncate_digest(&sanitized, max_tokens, tc);
+    let token_count = i64::try_from(tc.count_tokens(&final_text)).unwrap_or(i64::MAX);
+
+    if let Err(e) = memory
+        .sqlite()
+        .save_session_digest(conversation_id, &final_text, token_count)
+        .await
+    {
+        tracing::warn!("session digest (/new): storage failed: {e:#}");
+    } else {
+        tracing::info!(
+            conversation_id = conversation_id.0,
+            tokens = token_count,
+            "session digest stored (via /new)"
+        );
+    }
+}
+
 impl<C: Channel> Agent<C> {
     /// Generate and persist a session digest at shutdown when digest is enabled.
     ///