From 3bea7a95ea38e75d2c3b42cc4dc6e88b495d1f6c Mon Sep 17 00:00:00 2001 From: Clifford Ressel Date: Fri, 3 Jul 2026 20:56:45 -0400 Subject: [PATCH 1/3] feat(acp,tui): forward native /compact and branch sessions via ACP session/fork When the agent advertises a compact command, /compact now forwards it in-session (no session swap, no summary injection); the custom summarize-and-swap path remains as fallback. New /fork picker entry branches at the current point via the unstable ACP session/fork, capability-gated per agent; rewind-to-message stays local. Shared session-replacement machinery extracted into backend/session_replace.rs. Also fixes an ETXTBSY flake in script-agent connection tests. --- ROADMAP.md | 4 +- nori-rs/acp/Cargo.toml | 2 +- nori-rs/acp/docs.md | 38 +- nori-rs/acp/src/backend/helpers.rs | 1 + nori-rs/acp/src/backend/mod.rs | 1 + nori-rs/acp/src/backend/session.rs | 3 + nori-rs/acp/src/backend/session_reducer.rs | 16 +- nori-rs/acp/src/backend/session_replace.rs | 146 ++++++ .../acp/src/backend/session_runtime_driver.rs | 102 ++-- nori-rs/acp/src/backend/spawn_and_relay.rs | 3 + nori-rs/acp/src/backend/submit_and_ops.rs | 6 + nori-rs/acp/src/backend/tests/mod.rs | 1 + nori-rs/acp/src/backend/tests/part8.rs | 434 ++++++++++++++++++ nori-rs/acp/src/backend/thread_goal.rs | 1 + nori-rs/acp/src/backend/transcript.rs | 1 + nori-rs/acp/src/connection/acp_connection.rs | 36 ++ .../src/connection/acp_connection_tests.rs | 14 +- nori-rs/acp/src/connection/docs.md | 3 +- nori-rs/mock-acp-agent/docs.md | 6 +- nori-rs/mock-acp-agent/src/main.rs | 100 +++- nori-rs/nori-protocol/docs.md | 3 +- nori-rs/nori-protocol/src/lib.rs | 10 + nori-rs/nori-protocol/src/session_runtime.rs | 5 + nori-rs/protocol/docs.md | 4 +- nori-rs/protocol/src/protocol/mod.rs | 5 + nori-rs/tui/docs.md | 18 +- nori-rs/tui/src/chatwidget/event_handlers.rs | 8 + nori-rs/tui/src/chatwidget/tests/part5.rs | 24 + nori-rs/tui/src/nori/fork_picker.rs | 147 ++++-- ..._tests__fork_picker_with_branch_entry.snap | 17 + nori-rs/tui/src/slash_command.rs | 2 +- nori-rs/tui/src/viewonly_transcript.rs | 1 + 32 files changed, 1013 insertions(+), 149 deletions(-) create mode 100644 nori-rs/acp/src/backend/session_replace.rs create mode 100644 nori-rs/acp/src/backend/tests/part8.rs create mode 100644 nori-rs/tui/src/nori/snapshots/nori_tui__nori__fork_picker__tests__fork_picker_with_branch_entry.snap diff --git a/ROADMAP.md b/ROADMAP.md index 1e330d363..3b857e8eb 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -25,7 +25,7 @@ session lifecycle message queue | Local agent registry | `βœ… done` | custom agents can be registered | keep local config registration simple | ACP Registry should complement this, not replace it yet | | Session lifecycle | `βœ… done` | navigate previous sessions | keep behavior stable while forking work continues | `load`, `list`, `resume`, lazy indexing, and `undo` are already working | | Session continuity | `βœ… done` | users can continue prior work | keep resume listing, speed, and metadata stable | context/usage metadata and `session_info_update` are finished | -| Session forking | `πŸ”΅ now` | sessions can branch and form trees | finish `fork` and tree-oriented flows | keep native ACP behavior ahead of local stand-ins | +| Session forking | `πŸ”΅ now` | sessions can branch and form trees | fork-at-message and tree-oriented flows (needs upstream ACP PR #629) | branch-at-head shipped via native `session/fork`, capability-gated (claude-code adapter supports it; codex/gemini do not advertise fork yet) | | Image attachments | `πŸ”΅ now` | users can send visual context | capability-aware image routing and transcript fidelity | images reach ACP, but the path needs polish | | Queued messages | `πŸ”΅ now` | users can keep typing during work | finish backend handling for queued turns | TUI support exists, but ACP backend behavior needs tightening | | Session configuration | `βœ… done` | agents can expose useful controls | custom options like thinking/effort level and plan/build modes | keep this driven by agent-provided config, not hardcoding | @@ -45,7 +45,7 @@ This table keeps the roadmap grounded in the current ACP spec and draft surface. | Session lifecycle | `session/load` | stable baseline | done | | Session lifecycle | `session/list` | stable | done | | Session lifecycle | `session/resume` | unstable landed | done | -| Session lifecycle | `session/fork` | unstable landed | todo | +| Session lifecycle | `session/fork` | unstable landed | branch-at-head done; fork-at-message blocked on upstream PR #629 | | Session lifecycle | `session/close` | unstable landed | handled | | Session lifecycle | `session/delete` | draft only | not planned near term | | Session lifecycle | undo, rewind, checkpoints | not first-class ACP today | local only | diff --git a/nori-rs/acp/Cargo.toml b/nori-rs/acp/Cargo.toml index bee3e14f8..974cacd02 100644 --- a/nori-rs/acp/Cargo.toml +++ b/nori-rs/acp/Cargo.toml @@ -8,7 +8,7 @@ license = { workspace = true } workspace = true [dependencies] -agent-client-protocol = { workspace = true } +agent-client-protocol = { workspace = true, features = ["unstable_session_fork"] } agent-client-protocol-schema = { workspace = true, features = ["unstable"] } anyhow = { workspace = true } axum = { workspace = true, default-features = false, features = [ diff --git a/nori-rs/acp/docs.md b/nori-rs/acp/docs.md index c744f9f73..85079dc67 100644 --- a/nori-rs/acp/docs.md +++ b/nori-rs/acp/docs.md @@ -718,7 +718,7 @@ This means `to_acp_mcp_servers()` has side effects (reads from keyring/file syst The server uses a normal `http://127.0.0.1:/mcp` URL because current Codex and Claude ACP adapters forward ACP MCP server entries into their underlying clients as ordinary HTTP MCP config. Avoiding `acp:` keeps startup compatible with those adapters while still keeping the tool implementation in process. -ACP session setup paths build the MCP server list in two phases: first convert configured MCP servers with `to_acp_mcp_servers()`, then let backend-owned features append local MCP servers when their own eligibility checks pass. The `nori-client` server requires HTTP MCP support. This setup applies to resumed, fresh, fallback, and compaction-created sessions. New backend construction commits the server before `session/new` or `session/load` so agents that eagerly initialize MCP during session setup can reach it. Compaction-created replacement sessions re-advertise the backend's existing `nori-client` server because the tools are backend-owned rather than ACP-session-owned; if the replacement `session/new` fails, the backend restores the previous initialized flag and keeps the existing endpoint alive. After compact replacement succeeds, the backend re-emits `SessionCapabilitiesChanged`, deriving `nori_client.advertised` from the actual MCP server list passed to that replacement session. Hook-only ACP sessions pass an empty list because hooks do not need user-configured or backend-owned MCP servers. +ACP session setup paths build the MCP server list in two phases: first convert configured MCP servers with `to_acp_mcp_servers()`, then let backend-owned features append local MCP servers when their own eligibility checks pass. The `nori-client` server requires HTTP MCP support. This setup applies to resumed, fresh, fallback, and replacement sessions (client-side compaction via `session/new` and `/fork` branching via `session/fork`, both through `replace_session()` in `backend/session_replace.rs`). New backend construction commits the server before `session/new` or `session/load` so agents that eagerly initialize MCP during session setup can reach it. Replacement sessions re-advertise the backend's existing `nori-client` server because the tools are backend-owned rather than ACP-session-owned; if the replacement request fails, the backend restores the previous initialized flag and keeps the existing endpoint alive. After replacement succeeds, the backend re-emits `SessionCapabilitiesChanged`, deriving `nori_client.advertised` from the actual MCP server list passed to that replacement session. Hook-only ACP sessions pass an empty list because hooks do not need user-configured or backend-owned MCP servers. The local `nori-client` MCP server is intentionally additive. User-configured MCP servers are still forwarded normally, and ineligible agents simply do not receive the loopback endpoint. `SessionCapabilitiesChanged` exposes both endpoint advertisement and initialization so clients do not need to infer either state from raw HTTP MCP support. Continuation chaining depends on the local MCP server actually being initialized for the current advertised endpoint, not just on HTTP MCP support or endpoint advertisement. Durable follow-up work for this surface lives in `@/docs/followups/nori-client-mcp.md`. @@ -861,7 +861,7 @@ Configuration: - `AcpBackendConfig.cli_version`: CLI version included in session metadata - `AcpBackendConfig.default_model`: Default model to apply at session start (from config.toml [default_models]) -- `AcpBackendConfig.initial_context`: Optional string injected into `pending_compact_summary` at spawn time. Used by the TUI's `/fork` command to pass a plain-text conversation summary into a new ACP session, giving the agent prior context without a protocol-level session fork. When `None` (the default), `pending_compact_summary` starts empty as before. The same `pending_compact_summary` mechanism is shared by `/compact` and `/resume`. +- `AcpBackendConfig.initial_context`: Optional string injected into `pending_compact_summary` at spawn time. Used by the TUI's `/fork` rewind-to-a-previous-message path to pass a plain-text conversation summary into a new ACP session, giving the agent prior context without a protocol-level session fork (branching at the current point uses native `session/fork` instead -- see Session Branching below). When `None` (the default), `pending_compact_summary` starts empty as before. The same `pending_compact_summary` mechanism is shared by `/compact`'s client-side fallback and `/resume`. - `AcpBackendConfig.session_context`: Optional string injected into `pending_hook_context` at spawn time (`spawn_and_relay.rs` / `session.rs`) only for ACP connections that do not advertise HTTP MCP support. Unlike `initial_context`, session context is prepended to the first user prompt **without** `SUMMARY_PREFIX` framing -- it appears as raw text before the user's message. If session start hooks also produce `::context::` lines, those are appended to the session context (both accumulate in the same `pending_hook_context` mutex). The context is consumed on the first prompt and not repeated. The TUI populates this with an embedded `` blurb (`@/nori-rs/tui/session_context.md`) that tells non-MCP agents they are running inside Nori CLI over ACP, points to the source repo, and names MCP-backed affordances as unavailable. **Re-exported Types:** @@ -1018,15 +1018,31 @@ The simpler `translator.rs` helper functions are unrelated to ACP session transl **Conversation Compaction:** -Unlike core's direct history manipulation, ACP uses a **prompt-based approach**: +`/compact` has two paths. The decision point is `start_prompt()` in `backend/session_reducer.rs`, at the moment the queued compact prompt starts: -1. `/compact` sends summarization prompt to agent -2. Agent's summary response is streamed to the TUI as deltas and captured in `pending_compact_summary` -3. A new ACP session is created (the old session's context is discarded) -4. The `ContextCompactedEvent` is emitted with the summary text cloned from `pending_compact_summary`, enabling the TUI to render a visual session boundary -5. Summary is prepended to the next user message (via `SUMMARY_PREFIX` framing) +**Native path (preferred):** When the agent advertises a slash command named exactly `compact` (via `available_commands_update`, persisted on the runtime's session state -- both the claude-agent-acp and codex-acp adapters advertise one), the reducer rewrites the queued prompt to `QueuedPromptKind::NativeCompact` with text `/compact` and forwards it as an ordinary `session/prompt`. The agent compacts **inside the same session**: no replacement session, no summary capture, no summary injection into the next prompt. On completion, `backend/session_runtime_driver.rs` emits `ClientEvent::ContextCompacted { summary: None }` only when the turn stopped with `EndTurn` -- a cancelled native compaction did not compact, so nothing is recorded. This intent distinction did not exist on the old path, where any streamed summary was captured regardless. -The `ContextCompactedEvent.summary` field is the coupling point between the ACP backend and the TUI's session boundary rendering. The TUI uses it to flush the streamed summary, show a "Context compacted" info message, insert a new session header, and reprint the summary as the first assistant message of the new session (see `@/nori-rs/tui/docs.md`). +**Client-side fallback:** For agents without a native `compact` command, the prompt-based approach remains: + +1. `/compact` sends `codex_core::compact::SUMMARIZATION_PROMPT` to the agent +2. The agent's summary response is streamed to the TUI as deltas and captured in `pending_compact_summary` +3. A replacement ACP session is created via `replace_session(SessionReplacement::NewSession)` in `backend/session_replace.rs` (the old session's context is discarded) +4. `ContextCompactedEvent` is emitted with the summary text cloned from `pending_compact_summary`, enabling the TUI to render a visual session boundary +5. The summary is prepended to the next user message (via `SUMMARY_PREFIX` framing) + +The `ContextCompactedEvent.summary` field is the coupling point between the ACP backend and the TUI's session boundary rendering. When the summary is present (fallback path), the TUI flushes the streamed summary, shows a "Context compacted" info message, inserts a new session header, and reprints the summary as the first assistant message of the new session. When the summary is `None` (native path), only the info message is shown because the session did not change (see `@/nori-rs/tui/docs.md`). + +**Session Branching (`session/fork`):** + +`Op::BranchSession` (from `@/nori-rs/protocol/src/protocol/mod.rs`, fired by the first entry of the TUI's `/fork` picker) forks the ACP session at its current state and swaps the runtime to the fork; the original session is preserved on the agent side and remains resumable. The flow lives in `backend/session_replace.rs`: + +- The op is routed through `SessionRuntimeInput::BranchSession` on the session-runtime channel so branching serializes with turn starts and completions. +- `handle_branch_session()` rejects with a user-facing error when the agent does not advertise the `fork` session capability (`AcpConnection::supports_fork()`) or when a turn is in progress (`SessionDriver::is_idle()`). +- On success it emits `ClientEvent::SessionBranched { new_session_id }`. `AcpConnection::fork_session()` applies the fork response's `config_options` before returning, so the live config snapshot is already fresh when clients react to the event. + +`replace_session(SessionReplacement::NewSession | Fork { from })` in `backend/session_replace.rs` is the shared session-replacement machinery for client-side compaction and branching: it re-assembles the MCP server list, re-registers the backend-owned `nori-client` goal MCP server when needed, swaps the live session id, and re-broadcasts `SessionCapabilitiesChanged`. It was extracted from the compact arm of `backend/session_runtime_driver.rs` when branching landed. On failure, the previous session stays active and goal-MCP state is restored. + +ACP `session/fork` is at-head only; picking an earlier message in the `/fork` picker still uses the local summary-based backtrack fork (see `@/nori-rs/tui/docs.md`). Fork-at-message is tracked upstream as ACP PR #629. There is deliberately a single `/fork` command with no `/branch` or `/rewind` aliases. **Session Resume** (`backend/mod.rs`, `connection.rs`): @@ -1153,7 +1169,7 @@ Error categorization operates on the `Debug`-formatted (`{e:?}`) anyhow error to **Retryable classification and the failure disposition:** `AcpErrorCategory::is_retryable()` returns `true` only for `ApiServerError` and `QuotaExceeded` -- transient server errors and rate/quota limits that clear on their own -- and `false` for the persistent categories a retry cannot fix. The retryable bool no longer rides on `EventMsg::Error`; `ErrorEvent` in `@/nori-rs/protocol/src/protocol/mod.rs` now carries only `message` and is display-only. Instead the disposition is threaded onto the prompt **completion** as a `nori_protocol::TurnFailure` (`Retryable`/`Fatal`) so a single ordered event decides both how the failure is surfaced and whether an unattended loop should keep going (the TUI consumes this in `@/nori-rs/tui/docs.md`, Loop Mode section). -The path: `send_prompt_error` in `backend/session_runtime_driver.rs` emits the `EventMsg::Error` for display and *returns* the retryable bool (computed as `category.is_retryable()` for user prompts; `Compact` and `GoalContinuation` failures are always non-retryable). The prompt-task error handler maps that bool to `Some(TurnFailure::Retryable | Fatal)` and sends it as `InboundEvent::PromptFailed { failure }` to the reducer (`backend/session_reducer.rs`). `reduce_prompt_failed` threads `failure` straight onto the `PromptCompleted` it pushes, while clean completions (`reduce_prompt_response`) carry `failure: None`. The two non-prompt-task producers of `PromptFailed` use fixed dispositions: the cancel watchdog's force-cancel/timeout sends `failure: None` (a clean forced cancel, not a failure), and unexpected agent process exit (`ConnectionEvent::ChildExited`) in `backend/spawn_and_relay.rs` sends `Some(TurnFailure::Fatal)`. The generic `send_error` op path in `backend/submit_and_ops.rs` only emits a display `EventMsg::Error` and is not part of the prompt-completion path. The bare `529` and underscore `rate_limit` matches above still route real overloaded/rate-limit strings into the retryable buckets; `send_prompt_error` logs the category, the resolved retryable bool, and the raw error for observability. +The path: `send_prompt_error` in `backend/session_runtime_driver.rs` emits the `EventMsg::Error` for display and *returns* the retryable bool (computed as `category.is_retryable()` for user prompts; compaction failures of either kind and `GoalContinuation` failures are always non-retryable). The prompt-task error handler maps that bool to `Some(TurnFailure::Retryable | Fatal)` and sends it as `InboundEvent::PromptFailed { failure }` to the reducer (`backend/session_reducer.rs`). `reduce_prompt_failed` threads `failure` straight onto the `PromptCompleted` it pushes, while clean completions (`reduce_prompt_response`) carry `failure: None`. The two non-prompt-task producers of `PromptFailed` use fixed dispositions: the cancel watchdog's force-cancel/timeout sends `failure: None` (a clean forced cancel, not a failure), and unexpected agent process exit (`ConnectionEvent::ChildExited`) in `backend/spawn_and_relay.rs` sends `Some(TurnFailure::Fatal)`. The generic `send_error` op path in `backend/submit_and_ops.rs` only emits a display `EventMsg::Error` and is not part of the prompt-completion path. The bare `529` and underscore `rate_limit` matches above still route real overloaded/rate-limit strings into the retryable buckets; `send_prompt_error` logs the category, the resolved retryable bool, and the raw error for observability. ### Things to Know @@ -1163,7 +1179,7 @@ Large modules use a directory layout (`foo/mod.rs` + submodules) instead of a si - Agent subprocess communication uses stdin/stdout with JSON-RPC 2.0 framing - The minimum supported ACP protocol version is V1 (`MINIMUM_SUPPORTED_VERSION`); initialize is sent with `ProtocolVersion::LATEST` -- `nori-acp` no longer has an `unstable` feature. Model selection rides the `SessionConfigOptionCategory::Model` variant, which the schema exposes only behind its own `unstable` feature; `nori-acp` turns that on unconditionally (`agent-client-protocol-schema = { features = ["unstable"] }` in `Cargo.toml`). The removed `unstable` feature previously gated only the deleted `session/set_model`/`SessionModelState` model-selection API +- `nori-acp` no longer has an `unstable` feature. Model selection rides the `SessionConfigOptionCategory::Model` variant, which the schema exposes only behind its own `unstable` feature; `nori-acp` turns that on unconditionally (`agent-client-protocol-schema = { features = ["unstable"] }` in `Cargo.toml`). The removed `unstable` feature previously gated only the deleted `session/set_model`/`SessionModelState` model-selection API. Session branching additionally enables the SDK's `unstable_session_fork` feature on the `agent-client-protocol` dependency, since `session/fork` is an unstable ACP extension - Approval requests are translated to use appropriate UI (exec approval for shell commands, patch approval for file edits) - Config loading uses Nori-specific paths (`~/.nori/cli/config.toml`) when the `nori-config` feature is enabled in the TUI - Transcript discovery is synchronous and intended for use in background threads (e.g., the TUI's `SystemInfo` collection thread) diff --git a/nori-rs/acp/src/backend/helpers.rs b/nori-rs/acp/src/backend/helpers.rs index 1fd57c33c..37fd0ce04 100644 --- a/nori-rs/acp/src/backend/helpers.rs +++ b/nori-rs/acp/src/backend/helpers.rs @@ -18,6 +18,7 @@ pub(crate) fn get_op_name(op: &Op) -> &'static str { Op::SearchHistoryRequest { .. } => "SearchHistoryRequest", Op::ListCustomPrompts => "ListCustomPrompts", Op::Compact => "Compact", + Op::BranchSession => "BranchSession", Op::Undo => "Undo", Op::UndoList => "UndoList", Op::UndoTo { .. } => "UndoTo", diff --git a/nori-rs/acp/src/backend/mod.rs b/nori-rs/acp/src/backend/mod.rs index 610782830..230a6dfba 100644 --- a/nori-rs/acp/src/backend/mod.rs +++ b/nori-rs/acp/src/backend/mod.rs @@ -406,6 +406,7 @@ mod nori_client_mcp; mod session; mod session_defaults; pub(crate) mod session_reducer; +mod session_replace; mod session_runtime_driver; mod spawn_and_relay; mod submit_and_ops; diff --git a/nori-rs/acp/src/backend/session.rs b/nori-rs/acp/src/backend/session.rs index 75b095135..2c5370f06 100644 --- a/nori-rs/acp/src/backend/session.rs +++ b/nori-rs/acp/src/backend/session.rs @@ -392,6 +392,9 @@ impl AcpBackend { .handle_permission_request(pending_request, current_policy) .await; } + session_runtime_driver::SessionRuntimeInput::BranchSession => { + runtime_backend.handle_branch_session().await; + } } } }); diff --git a/nori-rs/acp/src/backend/session_reducer.rs b/nori-rs/acp/src/backend/session_reducer.rs index 4e5d066da..c5dc70a97 100644 --- a/nori-rs/acp/src/backend/session_reducer.rs +++ b/nori-rs/acp/src/backend/session_reducer.rs @@ -162,7 +162,21 @@ fn reduce_prompt_submit( start_prompt(runtime, prompt, out); } -fn start_prompt(runtime: &mut SessionRuntime, prompt: QueuedPrompt, out: &mut ReduceOutput) { +fn start_prompt(runtime: &mut SessionRuntime, mut prompt: QueuedPrompt, out: &mut ReduceOutput) { + // When the agent advertises a native `compact` slash command, forward that + // command instead of the client-side summarization prompt: the agent then + // compacts inside the current session and no replacement session is needed. + if prompt.kind == nori_protocol::session_runtime::QueuedPromptKind::Compact + && runtime + .persisted + .available_commands + .iter() + .any(|command| command.name == "compact") + { + prompt.kind = nori_protocol::session_runtime::QueuedPromptKind::NativeCompact; + prompt.text = "/compact".to_string(); + } + let phase_before = session_phase_label(&runtime.phase); let request_id = new_request_id(); diff --git a/nori-rs/acp/src/backend/session_replace.rs b/nori-rs/acp/src/backend/session_replace.rs new file mode 100644 index 000000000..1070f7d9e --- /dev/null +++ b/nori-rs/acp/src/backend/session_replace.rs @@ -0,0 +1,146 @@ +//! Replacing the live ACP session while keeping the same agent connection. +//! +//! Two flows swap the runtime's session id: client-side compaction (summarize, +//! then continue in a fresh `session/new`) and `/fork`'s branch-from-current- +//! point (`session/fork`). The MCP re-registration and capability bookkeeping +//! around that swap is identical for both, so it lives here. + +use super::*; + +use nori_protocol::ClientEvent; + +pub(super) enum SessionReplacement { + /// `session/new` β€” used after client-side compaction. + NewSession, + /// `session/fork` from the given session β€” used by branching. + Fork { from: acp::SessionId }, +} + +impl AcpBackend { + /// Create the replacement session (new or forked), re-register the + /// backend-owned `nori-client` MCP server, swap the live session id, and + /// broadcast the refreshed session capabilities. + /// + /// On failure the previous session stays active and goal-MCP state is + /// restored. + pub(super) async fn replace_session( + &self, + replacement: SessionReplacement, + ) -> anyhow::Result { + let cwd = self.cwd.clone(); + let mut mcp_servers = crate::connection::mcp::to_acp_mcp_servers( + &self.mcp_servers, + self.mcp_oauth_credentials_store_mode, + ); + let previous_goal_mcp_connected = { + let goal_mcp_http_server = self.goal_mcp_http_server.lock().await; + if let Some(server) = goal_mcp_http_server.as_ref() { + mcp_servers.push(server.as_mcp_server()); + Some( + self.goal_mcp_connected + .swap(false, std::sync::atomic::Ordering::Relaxed), + ) + } else { + None + } + }; + let pending_nori_client_server = if previous_goal_mcp_connected.is_some() { + None + } else { + match nori_client_mcp::register_for_session( + &self.connection, + &mut mcp_servers, + Arc::clone(&self.thread_goal_state), + self.backend_event_tx.clone(), + Arc::clone(&self.transcript_recorder_cell), + Arc::clone(&self.goal_mcp_connected), + ) + .await + { + Ok(server) => server, + Err(err) => { + warn!("Failed to register goal MCP server for replacement session: {err}"); + None + } + } + }; + let nori_client_advertised = mcp_servers.iter().any(|server| { + matches!( + server, + acp::McpServer::Http(http) if http.name == "nori-client" + ) + }); + + let result = match &replacement { + SessionReplacement::NewSession => { + self.connection.create_session(&cwd, mcp_servers).await + } + SessionReplacement::Fork { from } => { + self.connection.fork_session(from, &cwd, mcp_servers).await + } + }; + + match result { + Ok(new_session_id) => { + if let Some(server) = pending_nori_client_server { + server.commit(&self.goal_mcp_http_server).await; + } + *self.session_id.write().await = new_session_id.clone(); + self.forward_and_record_client_event(ClientEvent::SessionCapabilitiesChanged( + nori_client_mcp::capabilities_update_for_nori_client( + &self.connection, + nori_client_advertised, + self.goal_mcp_connected + .load(std::sync::atomic::Ordering::Relaxed), + ), + )) + .await; + Ok(new_session_id) + } + Err(err) => { + if let Some(previous) = previous_goal_mcp_connected { + self.goal_mcp_connected + .store(previous, std::sync::atomic::Ordering::Relaxed); + } + Err(err) + } + } + } + + /// Handle `Op::BranchSession`: fork the current session at its current + /// state via ACP `session/fork` and switch the runtime to the fork. The + /// original session is preserved on the agent side and stays resumable. + pub(super) async fn handle_branch_session(&self) { + if !self.connection.supports_fork() { + self.send_error("This agent does not support branching sessions.") + .await; + return; + } + if !self.session_driver.lock().await.is_idle() { + self.send_error("Cannot branch while a turn is in progress.") + .await; + return; + } + let from = self.session_id.read().await.clone(); + // `fork_session` applies the fork response's config options before it + // returns, so the config snapshot is current when SessionBranched is + // emitted (clients may re-read config on that event). + match self + .replace_session(SessionReplacement::Fork { from }) + .await + { + Ok(new_session_id) => { + debug!("Branched to forked session: {new_session_id:?}"); + self.forward_and_record_client_event(ClientEvent::SessionBranched( + nori_protocol::SessionBranched { + new_session_id: new_session_id.to_string(), + }, + )) + .await; + } + Err(err) => { + self.send_error(&format!("Branching failed: {err:#}")).await; + } + } + } +} diff --git a/nori-rs/acp/src/backend/session_runtime_driver.rs b/nori-rs/acp/src/backend/session_runtime_driver.rs index 84ae0fefc..db2cab683 100644 --- a/nori-rs/acp/src/backend/session_runtime_driver.rs +++ b/nori-rs/acp/src/backend/session_runtime_driver.rs @@ -30,6 +30,10 @@ pub(crate) enum SessionRuntimeInput { pending_request: Box, current_policy: AskForApproval, }, + /// Fork the session at its current state via `session/fork` and swap to + /// the fork. Routed through this channel so branching serializes with + /// turn starts and completions. + BranchSession, } pub(crate) struct ReducerActions { @@ -54,6 +58,7 @@ fn client_event_kind(event: &ClientEvent) -> &'static str { ClientEvent::PlanSnapshot(_) => "plan_snapshot", ClientEvent::LoadCompleted => "load_completed", ClientEvent::ContextCompacted(_) => "context_compacted", + ClientEvent::SessionBranched(_) => "session_branched", ClientEvent::Warning(_) => "warning", ClientEvent::ReplayEntry(_) => "replay_entry", ClientEvent::ThreadGoalUpdated(_) => "thread_goal_updated", @@ -69,6 +74,11 @@ impl SessionDriver { } } + /// Whether no session request (prompt/load) is currently active. + pub(crate) fn is_idle(&self) -> bool { + self.runtime.phase == SessionPhase::Idle + } + pub(crate) fn apply(&mut self, event: InboundEvent) -> ReducerActions { let completed_prompt = matches!( event, @@ -278,7 +288,7 @@ impl AcpBackend { async fn dispatch_reducer_actions(&self, actions: ReducerActions) { match actions.completed_turn.as_ref().map(|turn| turn.prompt.kind) { - Some(QueuedPromptKind::Compact) => { + Some(QueuedPromptKind::Compact | QueuedPromptKind::NativeCompact) => { let mut completion_event = None; let mut non_completion_events = Vec::new(); for event in actions.events { @@ -318,7 +328,7 @@ impl AcpBackend { } } - async fn forward_and_record_client_event(&self, client_event: ClientEvent) { + pub(super) async fn forward_and_record_client_event(&self, client_event: ClientEvent) { match &client_event { ClientEvent::SessionPhaseChanged(phase) => { debug!( @@ -486,73 +496,14 @@ impl AcpBackend { }; *self.pending_compact_summary.lock().await = Some(summary.clone()); - let cwd = self.cwd.clone(); - let mut mcp_servers = crate::connection::mcp::to_acp_mcp_servers( - &self.mcp_servers, - self.mcp_oauth_credentials_store_mode, - ); - let previous_goal_mcp_connected = { - let goal_mcp_http_server = self.goal_mcp_http_server.lock().await; - if let Some(server) = goal_mcp_http_server.as_ref() { - mcp_servers.push(server.as_mcp_server()); - Some( - self.goal_mcp_connected - .swap(false, std::sync::atomic::Ordering::Relaxed), - ) - } else { - None - } - }; - let pending_nori_client_server = if previous_goal_mcp_connected.is_some() { - None - } else { - match nori_client_mcp::register_for_session( - &self.connection, - &mut mcp_servers, - Arc::clone(&self.thread_goal_state), - self.backend_event_tx.clone(), - Arc::clone(&self.transcript_recorder_cell), - Arc::clone(&self.goal_mcp_connected), - ) + match self + .replace_session(super::session_replace::SessionReplacement::NewSession) .await - { - Ok(server) => server, - Err(err) => { - warn!("Failed to register goal MCP server after compact: {err}"); - None - } - } - }; - let nori_client_advertised = mcp_servers.iter().any(|server| { - matches!( - server, - acp::McpServer::Http(http) if http.name == "nori-client" - ) - }); - match self.connection.create_session(&cwd, mcp_servers).await { + { Ok(new_session_id) => { - if let Some(server) = pending_nori_client_server { - server.commit(&self.goal_mcp_http_server).await; - } - debug!("Created new session after compact: {:?}", new_session_id); - *self.session_id.write().await = new_session_id; - self.forward_and_record_client_event( - ClientEvent::SessionCapabilitiesChanged( - nori_client_mcp::capabilities_update_for_nori_client( - &self.connection, - nori_client_advertised, - self.goal_mcp_connected - .load(std::sync::atomic::Ordering::Relaxed), - ), - ), - ) - .await; + debug!("Created new session after compact: {new_session_id:?}"); } Err(err) => { - if let Some(previous) = previous_goal_mcp_connected { - self.goal_mcp_connected - .store(previous, std::sync::atomic::Ordering::Relaxed); - } warn!("Failed to create new session after compact: {err}"); } } @@ -574,6 +525,19 @@ impl AcpBackend { }) .await; } + QueuedPromptKind::NativeCompact => { + // The agent compacted inside the current session; nothing to + // swap or inject. Record the compaction only for turns that + // ran to completion β€” a cancelled compaction did not compact. + if completed_turn.stop_reason + == agent_client_protocol_schema::v1::StopReason::EndTurn + { + self.forward_and_record_client_event(ClientEvent::ContextCompacted( + nori_protocol::ContextCompacted { summary: None }, + )) + .await; + } + } } } @@ -595,7 +559,9 @@ impl AcpBackend { match completed_turn.prompt.kind { QueuedPromptKind::User => {} QueuedPromptKind::GoalContinuation if can_chain_continuation => {} - QueuedPromptKind::GoalContinuation | QueuedPromptKind::Compact => return, + QueuedPromptKind::GoalContinuation + | QueuedPromptKind::Compact + | QueuedPromptKind::NativeCompact => return, } self.submit_goal_continuation_if_idle().await; @@ -763,7 +729,9 @@ impl AcpBackend { err: &anyhow::Error, ) -> nori_protocol::TurnFailure { let (message, retryable) = match prompt_kind { - QueuedPromptKind::Compact => (format!("Compact failed: {err}"), false), + QueuedPromptKind::Compact | QueuedPromptKind::NativeCompact => { + (format!("Compact failed: {err}"), false) + } QueuedPromptKind::GoalContinuation => { (format!("Goal continuation failed: {err}"), false) } diff --git a/nori-rs/acp/src/backend/spawn_and_relay.rs b/nori-rs/acp/src/backend/spawn_and_relay.rs index 8d6d7b65b..5b0064bfb 100644 --- a/nori-rs/acp/src/backend/spawn_and_relay.rs +++ b/nori-rs/acp/src/backend/spawn_and_relay.rs @@ -176,6 +176,9 @@ impl AcpBackend { .handle_permission_request(pending_request, current_policy) .await; } + session_runtime_driver::SessionRuntimeInput::BranchSession => { + runtime_backend.handle_branch_session().await; + } } } }); diff --git a/nori-rs/acp/src/backend/submit_and_ops.rs b/nori-rs/acp/src/backend/submit_and_ops.rs index 01678db8b..a74308fd5 100644 --- a/nori-rs/acp/src/backend/submit_and_ops.rs +++ b/nori-rs/acp/src/backend/submit_and_ops.rs @@ -182,6 +182,12 @@ impl AcpBackend { Op::Compact => { self.handle_compact(&id).await?; } + Op::BranchSession => { + let _ = self + .session_event_tx + .send(session_runtime_driver::SessionRuntimeInput::BranchSession) + .await; + } Op::ListCustomPrompts => { let dir = commands_dir(&self.nori_home); let event_tx = self.event_tx.clone(); diff --git a/nori-rs/acp/src/backend/tests/mod.rs b/nori-rs/acp/src/backend/tests/mod.rs index c652418d2..f8bb8bcb0 100644 --- a/nori-rs/acp/src/backend/tests/mod.rs +++ b/nori-rs/acp/src/backend/tests/mod.rs @@ -381,3 +381,4 @@ mod part3; mod part4; mod part5; mod part7; +mod part8; diff --git a/nori-rs/acp/src/backend/tests/part8.rs b/nori-rs/acp/src/backend/tests/part8.rs new file mode 100644 index 000000000..b384eacd3 --- /dev/null +++ b/nori-rs/acp/src/backend/tests/part8.rs @@ -0,0 +1,434 @@ +//! Native compact and session branching (ACP `session/fork`) behavior. +//! +//! These tests drive the backend against the mock ACP agent and assert only +//! wire-observable behavior: the prompt text the agent receives (via +//! `MOCK_AGENT_ECHO_PROMPT`), the session a prompt targets (via +//! `MOCK_AGENT_ECHO_SESSION_ID`), and the client events the backend emits. + +use super::*; +use pretty_assertions::assert_eq; + +/// Everything observable from one prompt turn: the streamed answer text and +/// any `ContextCompacted` events, terminated by `PromptCompleted`. +struct TurnCapture { + answer_text: String, + context_compacted: Vec, + prompt_completed: nori_protocol::PromptCompleted, +} + +/// Collect client events until the turn's `PromptCompleted` arrives. +async fn capture_turn( + client_event_rx: &mut mpsc::Receiver, +) -> TurnCapture { + use std::time::Duration; + + let mut answer_text = String::new(); + let mut context_compacted = Vec::new(); + let deadline = std::time::Instant::now() + Duration::from_secs(10); + + while std::time::Instant::now() < deadline { + let Ok(event) = tokio::time::timeout(Duration::from_secs(1), client_event_rx.recv()).await + else { + continue; + }; + match event { + Some(nori_protocol::ClientEvent::MessageDelta(delta)) + if delta.stream == nori_protocol::MessageStream::Answer => + { + answer_text.push_str(&delta.delta); + } + Some(nori_protocol::ClientEvent::ContextCompacted(compacted)) => { + context_compacted.push(compacted); + } + Some(nori_protocol::ClientEvent::PromptCompleted(prompt_completed)) => { + return TurnCapture { + answer_text, + context_compacted, + prompt_completed, + }; + } + Some(_) => {} + None => break, + } + } + panic!("turn did not complete within the deadline; answer so far: {answer_text:?}"); +} + +/// Wait until the backend reports the agent's advertised slash commands. +async fn wait_for_agent_commands(client_event_rx: &mut mpsc::Receiver) { + use std::time::Duration; + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + if let Ok(Some(nori_protocol::ClientEvent::AgentCommandsUpdate(_))) = + tokio::time::timeout(Duration::from_secs(1), client_event_rx.recv()).await + { + return; + } + } + panic!("agent commands update was not received"); +} + +/// Wait for a `SessionBranched` client event, panicking on timeout. +async fn wait_for_session_branched( + client_event_rx: &mut mpsc::Receiver, +) -> nori_protocol::SessionBranched { + use std::time::Duration; + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + if let Ok(Some(nori_protocol::ClientEvent::SessionBranched(branched))) = + tokio::time::timeout(Duration::from_secs(1), client_event_rx.recv()).await + { + return branched; + } + } + panic!("SessionBranched event was not received"); +} + +/// Wait for a control-channel error event and return its message. +async fn wait_for_error_message(event_rx: &mut mpsc::Receiver) -> String { + use std::time::Duration; + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + if let Ok(Some(event)) = tokio::time::timeout(Duration::from_secs(1), event_rx.recv()).await + && let EventMsg::Error(error) = event.msg + { + return error.message; + } + } + panic!("no error event was received"); +} + +fn mock_agent_available() -> bool { + let mock_config = + crate::registry::get_agent_config("mock-model").expect("mock-model should be registered"); + if std::path::Path::new(&mock_config.command).exists() { + return true; + } + eprintln!( + "Skipping test: mock_acp_agent not found at {}", + mock_config.command + ); + false +} + +async fn spawn_backend_for_test( + temp_dir: &tempfile::TempDir, +) -> ( + AcpBackend, + mpsc::Receiver, + mpsc::Receiver, +) { + use std::time::Duration; + + let (event_tx, mut event_rx) = mpsc::channel(64); + let (client_event_tx, client_event_rx) = mpsc::channel(64); + let config = build_test_config(temp_dir.path()); + let backend = spawn_test_backend(&config, event_tx, Some(client_event_tx)) + .await + .expect("Failed to spawn ACP backend"); + + // Drain the SessionConfigured event. + let _ = tokio::time::timeout(Duration::from_secs(2), event_rx.recv()) + .await + .expect("Should receive SessionConfigured event"); + + (backend, event_rx, client_event_rx) +} + +/// When the agent advertises a `compact` command, `/compact` must forward the +/// agent's own command in the current session: the agent receives the literal +/// text `/compact` (not the custom summarization prompt), no replacement +/// session is created, and the transcript records a summary-less compaction. +#[tokio::test] +#[serial] +async fn test_compact_forwards_native_command_when_advertised() { + if !mock_agent_available() { + return; + } + let _commands = EnvGuard::set("MOCK_AGENT_AVAILABLE_COMMANDS", "compact"); + let _echo_prompt = EnvGuard::set("MOCK_AGENT_ECHO_PROMPT", "1"); + let _echo_session = EnvGuard::set("MOCK_AGENT_ECHO_SESSION_ID", "1"); + + let temp_dir = tempfile::tempdir().expect("Failed to create temp dir"); + let (backend, _event_rx, mut client_event_rx) = spawn_backend_for_test(&temp_dir).await; + wait_for_agent_commands(&mut client_event_rx).await; + + backend + .submit(Op::Compact) + .await + .expect("Failed to submit Op::Compact"); + let compact_turn = capture_turn(&mut client_event_rx).await; + + assert!( + compact_turn.answer_text.contains("/compact"), + "agent should receive the literal /compact command, got: {:?}", + compact_turn.answer_text + ); + assert!( + !compact_turn + .answer_text + .contains("CONTEXT CHECKPOINT COMPACTION"), + "the custom summarization prompt must not be sent when the agent \ + advertises native compaction" + ); + assert_eq!( + compact_turn.context_compacted, + vec![nori_protocol::ContextCompacted { summary: None }], + "native compaction should be recorded without a client-side summary" + ); + + // The follow-up prompt stays on the original session and carries no + // injected summary. + backend + .submit(Op::UserInput { + items: vec![codex_protocol::user_input::UserInput::Text { + text: "hello after compact".to_string(), + }], + }) + .await + .expect("Failed to submit follow-up prompt"); + let follow_up_turn = capture_turn(&mut client_event_rx).await; + + assert!( + follow_up_turn.answer_text.contains("SESSION:0"), + "native compact must not swap to a new session, got: {:?}", + follow_up_turn.answer_text + ); + assert!( + follow_up_turn.answer_text.contains("hello after compact"), + "the follow-up prompt should be echoed back, got: {:?}", + follow_up_turn.answer_text + ); + assert!( + !follow_up_turn + .answer_text + .contains("Another language model started to solve this problem"), + "no compact summary may be injected into prompts after native compaction" + ); +} + +/// A cancelled native compaction must not record a compaction, and the session +/// must keep accepting prompts afterwards. +#[tokio::test] +#[serial] +async fn test_native_compact_cancelled_records_no_compaction() { + use std::time::Duration; + + if !mock_agent_available() { + return; + } + let _commands = EnvGuard::set("MOCK_AGENT_AVAILABLE_COMMANDS", "compact"); + let _stream = EnvGuard::set("MOCK_AGENT_STREAM_UNTIL_CANCEL", "1"); + + let temp_dir = tempfile::tempdir().expect("Failed to create temp dir"); + let (backend, _event_rx, mut client_event_rx) = spawn_backend_for_test(&temp_dir).await; + wait_for_agent_commands(&mut client_event_rx).await; + + backend + .submit(Op::Compact) + .await + .expect("Failed to submit Op::Compact"); + + // Wait for streaming to start, then interrupt. + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + assert!( + std::time::Instant::now() < deadline, + "compact turn never started streaming" + ); + if let Ok(Some(nori_protocol::ClientEvent::MessageDelta(_))) = + tokio::time::timeout(Duration::from_secs(1), client_event_rx.recv()).await + { + break; + } + } + backend + .submit(Op::Interrupt) + .await + .expect("Failed to submit Op::Interrupt"); + let cancelled_turn = capture_turn(&mut client_event_rx).await; + + assert_eq!( + cancelled_turn.prompt_completed.stop_reason, + nori_protocol::StopReason::Cancelled + ); + assert_eq!( + cancelled_turn.context_compacted, + Vec::new(), + "a cancelled compaction must not be recorded as a compaction" + ); + + // The session still accepts prompts: the next turn streams and can be + // cancelled cleanly. + backend + .submit(Op::UserInput { + items: vec![codex_protocol::user_input::UserInput::Text { + text: "still alive?".to_string(), + }], + }) + .await + .expect("Failed to submit follow-up prompt"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + assert!( + std::time::Instant::now() < deadline, + "follow-up turn never started streaming" + ); + if let Ok(Some(nori_protocol::ClientEvent::MessageDelta(_))) = + tokio::time::timeout(Duration::from_secs(1), client_event_rx.recv()).await + { + break; + } + } + backend + .submit(Op::Interrupt) + .await + .expect("Failed to submit follow-up interrupt"); + let follow_up_turn = capture_turn(&mut client_event_rx).await; + assert_eq!( + follow_up_turn.prompt_completed.stop_reason, + nori_protocol::StopReason::Cancelled + ); +} + +/// Branching with an agent that advertises `session/fork` swaps the runtime to +/// the forked session: subsequent prompts arrive on the new agent session. +#[tokio::test] +#[serial] +async fn test_branch_session_swaps_to_forked_session() { + if !mock_agent_available() { + return; + } + let _fork = EnvGuard::set("MOCK_AGENT_SUPPORT_SESSION_FORK", "1"); + let _echo_session = EnvGuard::set("MOCK_AGENT_ECHO_SESSION_ID", "1"); + + let temp_dir = tempfile::tempdir().expect("Failed to create temp dir"); + let (backend, _event_rx, mut client_event_rx) = spawn_backend_for_test(&temp_dir).await; + + backend + .submit(Op::BranchSession) + .await + .expect("Failed to submit Op::BranchSession"); + let branched = wait_for_session_branched(&mut client_event_rx).await; + assert_ne!( + branched.new_session_id, "0", + "fork must yield a new session" + ); + + backend + .submit(Op::UserInput { + items: vec![codex_protocol::user_input::UserInput::Text { + text: "hello branched session".to_string(), + }], + }) + .await + .expect("Failed to submit prompt after branch"); + let turn = capture_turn(&mut client_event_rx).await; + + assert!( + turn.answer_text + .contains(&format!("SESSION:{}", branched.new_session_id)), + "prompts after branching must target the forked session, got: {:?}", + turn.answer_text + ); +} + +/// Branching with an agent that does NOT advertise `session/fork` surfaces a +/// clear error and leaves the current session untouched. +#[tokio::test] +#[serial] +async fn test_branch_session_unsupported_agent_reports_error() { + if !mock_agent_available() { + return; + } + let _no_fork = EnvGuard::remove("MOCK_AGENT_SUPPORT_SESSION_FORK"); + let _echo_session = EnvGuard::set("MOCK_AGENT_ECHO_SESSION_ID", "1"); + + let temp_dir = tempfile::tempdir().expect("Failed to create temp dir"); + let (backend, mut event_rx, mut client_event_rx) = spawn_backend_for_test(&temp_dir).await; + + backend + .submit(Op::BranchSession) + .await + .expect("Failed to submit Op::BranchSession"); + let message = wait_for_error_message(&mut event_rx).await; + assert!( + message.contains("does not support branching"), + "unsupported branching should be reported as such, got: {message:?}" + ); + + // No branch may be reported when the agent lacks the capability. + while let Ok(Some(event)) = tokio::time::timeout( + std::time::Duration::from_millis(300), + client_event_rx.recv(), + ) + .await + { + assert!( + !matches!(event, nori_protocol::ClientEvent::SessionBranched(_)), + "no SessionBranched event may be emitted for an unsupported agent" + ); + } + + backend + .submit(Op::UserInput { + items: vec![codex_protocol::user_input::UserInput::Text { + text: "hello original session".to_string(), + }], + }) + .await + .expect("Failed to submit prompt after failed branch"); + let turn = capture_turn(&mut client_event_rx).await; + assert!( + turn.answer_text.contains("SESSION:0"), + "a failed branch must leave the original session active, got: {:?}", + turn.answer_text + ); +} + +/// The fork response's `config_options` replace the live session config +/// snapshot (the mock advertises `thought_level: high` for forked sessions). +#[tokio::test] +#[serial] +async fn test_branch_session_applies_forked_config_options() { + if !mock_agent_available() { + return; + } + let _fork = EnvGuard::set("MOCK_AGENT_SUPPORT_SESSION_FORK", "1"); + + let temp_dir = tempfile::tempdir().expect("Failed to create temp dir"); + let (backend, _event_rx, mut client_event_rx) = spawn_backend_for_test(&temp_dir).await; + + let thought_level_value = |options: &[acp::SessionConfigOption]| -> Option { + options.iter().find_map(|option| { + if option.id.to_string() != "thought_level" { + return None; + } + match &option.kind { + acp::SessionConfigKind::Select(select) => Some(select.current_value.to_string()), + _ => None, + } + }) + }; + + assert_eq!( + thought_level_value(&backend.config_options()).as_deref(), + Some("medium"), + "precondition: fresh mock sessions start at thought_level medium" + ); + + backend + .submit(Op::BranchSession) + .await + .expect("Failed to submit Op::BranchSession"); + let _ = wait_for_session_branched(&mut client_event_rx).await; + + assert_eq!( + thought_level_value(&backend.config_options()).as_deref(), + Some("high"), + "the fork response's config_options must replace the live snapshot" + ); +} diff --git a/nori-rs/acp/src/backend/thread_goal.rs b/nori-rs/acp/src/backend/thread_goal.rs index ceeb5b772..74802531d 100644 --- a/nori-rs/acp/src/backend/thread_goal.rs +++ b/nori-rs/acp/src/backend/thread_goal.rs @@ -92,6 +92,7 @@ impl ThreadGoalState { | ClientEvent::LoadCompleted | ClientEvent::QueueChanged(_) | ClientEvent::ContextCompacted(_) + | ClientEvent::SessionBranched(_) | ClientEvent::ReplayEntry(_) | ClientEvent::AgentCommandsUpdate(_) | ClientEvent::SessionCapabilitiesChanged(_) diff --git a/nori-rs/acp/src/backend/transcript.rs b/nori-rs/acp/src/backend/transcript.rs index fdab876a8..cf6ca2215 100644 --- a/nori-rs/acp/src/backend/transcript.rs +++ b/nori-rs/acp/src/backend/transcript.rs @@ -207,6 +207,7 @@ fn replay_entry_from_client_event( | nori_protocol::ClientEvent::LoadCompleted | nori_protocol::ClientEvent::QueueChanged(_) | nori_protocol::ClientEvent::ContextCompacted(_) + | nori_protocol::ClientEvent::SessionBranched(_) | nori_protocol::ClientEvent::ReplayEntry(_) | nori_protocol::ClientEvent::AgentCommandsUpdate(_) | nori_protocol::ClientEvent::SessionCapabilitiesChanged(_) diff --git a/nori-rs/acp/src/connection/acp_connection.rs b/nori-rs/acp/src/connection/acp_connection.rs index f440d4edc..b9b176674 100644 --- a/nori-rs/acp/src/connection/acp_connection.rs +++ b/nori-rs/acp/src/connection/acp_connection.rs @@ -699,6 +699,42 @@ impl AcpConnection { Ok(acp::SessionId::from(session_id.to_string())) } + /// Fork (branch) an existing session at its current state via + /// `session/fork` (unstable ACP extension, gated on the agent advertising + /// the `fork` session capability). + /// + /// The returned ID identifies the new session; the source session stays + /// valid on the agent side. Like `create_session`, the response's live + /// config options replace the current snapshot before this returns. + pub async fn fork_session( + &self, + session_id: &acp::SessionId, + cwd: &Path, + mcp_servers: Vec, + ) -> Result { + let response = self + .cx + .send_request( + acp::ForkSessionRequest::new(session_id.clone(), cwd).mcp_servers(mcp_servers), + ) + .block_task() + .await + .context("Failed to fork ACP session")?; + + if let Some(config_options) = response.config_options + && let Ok(mut state) = self.session_config_state.write() + { + state.config_options = config_options; + } + + Ok(response.session_id) + } + + /// Whether the connected agent advertises the `session/fork` capability. + pub fn supports_fork(&self) -> bool { + self.agent_capabilities.session_capabilities.fork.is_some() + } + /// List the agent's known sessions via ACP `session/list`. /// /// `cwd` is forwarded as the spec's working-directory filter. Cursor-based diff --git a/nori-rs/acp/src/connection/acp_connection_tests.rs b/nori-rs/acp/src/connection/acp_connection_tests.rs index 9e99e6967..883fe1380 100644 --- a/nori-rs/acp/src/connection/acp_connection_tests.rs +++ b/nori-rs/acp/src/connection/acp_connection_tests.rs @@ -936,17 +936,15 @@ async fn test_prompt_after_cancel_absorbs_empty_end_turn_tail() { fn script_agent_config(dir: &std::path::Path, body: &str) -> crate::registry::AcpAgentConfig { let script = dir.join("script-agent.sh"); std::fs::write(&script, body).expect("write script agent"); - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)) - .expect("chmod script agent"); - } crate::registry::AcpAgentConfig { agent: crate::registry::AgentKind::ClaudeCode, // placeholder, like custom agents provider_slug: "script-agent".to_string(), - command: script.to_string_lossy().to_string(), - args: vec![], + // Run the script through `sh` so the file is opened as data rather + // than exec'd directly: exec'ing a just-written file intermittently + // fails with ETXTBSY when a concurrently forking test process still + // holds the write fd (seen under full-workspace test load). + command: "sh".to_string(), + args: vec![script.to_string_lossy().to_string()], env: std::collections::HashMap::new(), provider_info: crate::registry::AcpProviderInfo::default(), auth_hint: "run: script-agent login".to_string(), diff --git a/nori-rs/acp/src/connection/docs.md b/nori-rs/acp/src/connection/docs.md index 5c3413941..4a963573a 100644 --- a/nori-rs/acp/src/connection/docs.md +++ b/nori-rs/acp/src/connection/docs.md @@ -31,7 +31,7 @@ Local ACP Agent (subprocess, own process group) ### Core Implementation -- **`AcpConnection`** (`acp_connection.rs`): The central `Send + Sync` type that owns the SDK connection task, the child's teardown handles, and the event receiver. Key methods: `create_session()`, `load_session()`, `list_sessions()`, `prompt()`, `cancel()`, `set_config_option()` +- **`AcpConnection`** (`acp_connection.rs`): The central `Send + Sync` type that owns the SDK connection task, the child's teardown handles, and the event receiver. Key methods: `create_session()`, `fork_session()`, `load_session()`, `list_sessions()`, `prompt()`, `cancel()`, `set_config_option()` - **`establish_connection()`**: A free function that registers all SDK handlers (notification, permission request, file read/write) and performs the initialization handshake. The caller (`spawn()`) supplies the `ConnectionEvent` channel so extra producers -- notably the child exit watcher -- can report through the same ordered stream - **Child exit watcher**: A background task OWNS the `Child`: it `wait()`s and reaps the process, publishes the exit status on a `watch` channel, and emits `ConnectionEvent::ChildExited { status, stderr_tail }` (defined in `mod.rs`) when the child dies. The connection holds only a `ChildHandle` (pid, exit-status receiver, kill `Notify`) -- never the `Child` itself - **stderr capture**: A background task logs each stderr line via tracing and keeps a bounded tail of recent lines. The tail is attached to startup failures and `ChildExited` events so the real cause (e.g. an auth hint) is user-visible @@ -49,6 +49,7 @@ Local ACP Agent (subprocess, own process group) - Without the `ChildExited` event, a dead child is a silent EOF that the ACP SDK layer treats as non-terminal -- pending requests would hang forever. The backend relay turns it into a visible error and fails any in-flight prompt - File write handlers restrict writes to the workspace directory or `/tmp` (canonicalized path check). File read handlers are unrestricted. Both emit synthetic `ToolCall` updates for TUI rendering - `AcpConnection::prompt()` absorbs stale cancel-tail responses (empty `end_turn` responses left over from a previous cancellation) by retrying until streamed updates arrive, keeping the reducer contract unchanged +- `fork_session()` sends the ACP `session/fork` request to branch an existing session at its current state; the returned id identifies the new session and the source session stays valid on the agent side. `session/fork` is an unstable ACP extension, so the crate enables the SDK's `unstable_session_fork` cargo feature on `agent-client-protocol` (see `@/nori-rs/acp/Cargo.toml`), and callers must gate on `supports_fork()`, which reports whether the agent advertised the `fork` session capability during initialize. Like `create_session()`, `fork_session()` replaces the live config snapshot with the fork response's `config_options` **before** returning, so the backend's `SessionBranched` event in `@/nori-rs/acp/src/backend/session_replace.rs` is emitted against a fresh snapshot - `list_sessions()` sends the ACP `session/list` request and drains cursor pagination internally (following `next_cursor` until exhausted, bounded by a page cap so a misbehaving agent that keeps returning a cursor cannot loop unbounded), concatenating every page in agent order. Each ACP `SessionInfo` is mapped into `AcpSessionSummary` (defined in `mod.rs`, re-exported from `nori_acp`), an owned boundary type carrying `session_id`, `cwd`, optional `title`, and optional `updated_at`. The boundary type decouples consumers from the raw ACP schema, letting `@/nori-rs/tui` -- which has no ACP-schema dependency -- render the agent-sourced `/resume` picker. It mirrors `load_session()`'s structure; the agent only honors `session/list` when it advertises the capability (projected as `agent.session_list`, see `@/nori-rs/acp/docs.md`) Created and maintained by Nori. diff --git a/nori-rs/mock-acp-agent/docs.md b/nori-rs/mock-acp-agent/docs.md index f39913576..f8756c8ec 100644 --- a/nori-rs/mock-acp-agent/docs.md +++ b/nori-rs/mock-acp-agent/docs.md @@ -20,9 +20,11 @@ Used by `@/nori-rs/tui-pty-e2e/` for end-to-end integration testing. The mock ag **Mock Behaviors**: Controlled via environment variables that the E2E tests set on the mock agent process. Each env var activates a specific behavior scenario. Key scenarios include multi-turn conversations, tool call streaming, permission requests, file operations, race condition simulations, and session lifecycle behaviors. -**Session Lifecycle Testing**: Several env vars control `session/load` behavior for testing the resume path in `@/nori-rs/acp/src/backend/session.rs`: +**Session Lifecycle Testing**: Several env vars control session lifecycle behavior (capability advertisement, `session/new`, `session/load`, `session/fork`) for testing the resume, replacement, and branching paths in `@/nori-rs/acp/src/backend/`: - `MOCK_AGENT_SUPPORT_LOAD_SESSION` -- when set, the agent advertises `load_session: true` in its capabilities during `initialize()` - `MOCK_AGENT_SUPPORT_SESSION_LIST` -- when set, the agent advertises the ACP `session/list` capability during `initialize()` and its `session/list` handler returns two canned `SessionInfo` rows; exercises the agent-sourced `/resume` picker wire path (`AcpConnection::list_sessions()` in `@/nori-rs/acp/src/connection/`, surfaced as `agent.session_list`) +- `MOCK_AGENT_SUPPORT_SESSION_FORK` -- when set, the agent advertises the ACP `session/fork` capability during `initialize()` and handles `session/fork` by minting a new session id seeded from the source session's config. Forked sessions deliberately come back with thought_level `"high"` so tests can verify the client applied the fork response's `config_options` instead of keeping its old snapshot. Exercises `AcpConnection::fork_session()` / `supports_fork()` and the branch flow in `@/nori-rs/acp/src/backend/session_replace.rs` +- `MOCK_AGENT_AVAILABLE_COMMANDS` -- comma-separated command names; `new_session()` spawns a task that sends an `available_commands_update` shortly after the response, mirroring how real adapters (e.g. claude-agent-acp) push commands right after `session/new`. Setting it to `compact` exercises the native-compact rewrite in `@/nori-rs/acp/src/backend/session_reducer.rs` - `MOCK_AGENT_MCP_HTTP` -- when set, the agent advertises HTTP MCP capability so the backend-owned `nori-client` MCP server in `@/nori-rs/acp/src/backend/nori_client_mcp.rs` can be tested through the normal `session/new` MCP server advertisement path - `MOCK_AGENT_INITIALIZE_NORI_CLIENT_DURING_NEW_SESSION` -- when set, `new_session()` eagerly sends an MCP `initialize` request to the advertised `nori-client` server before returning, mirroring agents that initialize advertised MCP servers during session setup - `MOCK_AGENT_FAIL_NEW_SESSION_FROM` -- when set to an integer N, `new_session()` returns an error once the generated session id is at least N, allowing tests to exercise replacement-session failures without breaking the initial backend startup @@ -33,6 +35,8 @@ Used by `@/nori-rs/tui-pty-e2e/` for end-to-end integration testing. The mock ag **Prompt Echo**: The `MOCK_AGENT_ECHO_PROMPT` env var causes the mock agent's `prompt()` handler to echo back the full prompt text it received. Used by session context tests in `@/nori-rs/acp/src/backend/tests/part5.rs` to verify that `AcpBackendConfig.session_context` is correctly prepended to the first user prompt and consumed after that. +**Session ID Echo**: The `MOCK_AGENT_ECHO_SESSION_ID` env var causes the mock agent's `prompt()` handler to start its response with a `SESSION:` text chunk, so tests can observe which session a prompt targeted (e.g. after a `session/fork` swap). It composes with `MOCK_AGENT_ECHO_PROMPT`: the session chunk is sent first, then the prompt echo (or the default canned response) follows. Used by the native-compact and branching tests in `@/nori-rs/acp/src/backend/tests/part8.rs`. + **Cancel Ignore**: The `MOCK_AGENT_IGNORE_CANCEL` env var causes the mock agent's `cancel()` handler to silently discard the cancellation instead of setting `cancel_requested`. Used with `MOCK_AGENT_STREAM_UNTIL_CANCEL` by `@/nori-rs/acp/src/backend/tests/part4.rs` to test the cancel timeout watchdog in `session_runtime_driver.rs` -- verifying that the backend force-cancels the prompt after the timeout even when the agent is unresponsive to `CancelNotification`. **Session Config Options**: The mock agent advertises live ACP session config options on `session/new` and `session/load`, and supports `session/set_config_option` for connection/TUI tests. The default config exposes `Model` plus `Thought Level`. Switching the model to `mock-model-fast` replaces `Thought Level` with a `Speed` selector, which lets tests verify that Nori replaces the full live config snapshot after a config mutation. diff --git a/nori-rs/mock-acp-agent/src/main.rs b/nori-rs/mock-acp-agent/src/main.rs index 880765f55..e5601199b 100644 --- a/nori-rs/mock-acp-agent/src/main.rs +++ b/nori-rs/mock-acp-agent/src/main.rs @@ -709,6 +709,15 @@ impl MockAgent { return Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)); } + // Echo the session id a prompt arrived on, so tests can observe which + // session subsequent prompts target (e.g. after a session/fork swap). + // Composes with MOCK_AGENT_ECHO_PROMPT: this chunk is sent first, then + // the prompt echo (or the default canned response) follows. + if std::env::var("MOCK_AGENT_ECHO_SESSION_ID").is_ok() { + self.send_text_chunk(session_id.clone(), &format!("SESSION:{session_id}\n")) + .await?; + } + // Echo back the full prompt text for verifying context injection. if std::env::var("MOCK_AGENT_ECHO_PROMPT").is_ok() { let user_text = arguments @@ -1341,11 +1350,25 @@ async fn main() -> acp::Result<()> { has_capabilities = true; } + let mut session_capabilities = acp::SessionCapabilities::new(); + let mut has_session_capabilities = false; + if std::env::var("MOCK_AGENT_SUPPORT_SESSION_LIST").is_ok() { eprintln!("Mock agent: advertising session/list capability"); - capabilities = capabilities.session_capabilities( - acp::SessionCapabilities::new().list(acp::SessionListCapabilities::new()), - ); + session_capabilities = + session_capabilities.list(acp::SessionListCapabilities::new()); + has_session_capabilities = true; + } + + if std::env::var("MOCK_AGENT_SUPPORT_SESSION_FORK").is_ok() { + eprintln!("Mock agent: advertising session/fork capability"); + session_capabilities = + session_capabilities.fork(acp::SessionForkCapabilities::new()); + has_session_capabilities = true; + } + + if has_session_capabilities { + capabilities = capabilities.session_capabilities(session_capabilities); has_capabilities = true; } @@ -1370,7 +1393,7 @@ async fn main() -> acp::Result<()> { let state = state.clone(); async move |arguments: acp::NewSessionRequest, responder: Responder, - _cx: ConnectionTo| { + cx: ConnectionTo| { let session_id = state.next_session_id.fetch_add(1, Ordering::SeqCst); if let Ok(fail_from) = std::env::var("MOCK_AGENT_FAIL_NEW_SESSION_FROM") && let Ok(fail_from) = fail_from.parse::() @@ -1400,6 +1423,40 @@ async fn main() -> acp::Result<()> { ); } + // Advertise slash commands shortly after the response so the + // client has registered the session before the update arrives, + // mirroring how real adapters (e.g. claude-agent-acp) push + // available_commands_update right after session/new. + if let Ok(commands) = std::env::var("MOCK_AGENT_AVAILABLE_COMMANDS") { + let agent = MockAgent { + cx: cx.clone(), + state: state.clone(), + }; + let session_id = acp::SessionId::new(session_key.clone()); + let commands: Vec = commands + .split(',') + .filter(|name| !name.is_empty()) + .map(|name| { + acp::AvailableCommand::new(name, format!("mock {name} command")) + }) + .collect(); + eprintln!( + "Mock agent: advertising {} available command(s)", + commands.len() + ); + cx.spawn(async move { + sleep(Duration::from_millis(20)).await; + agent + .send_update( + session_id, + acp::SessionUpdate::AvailableCommandsUpdate( + acp::AvailableCommandsUpdate::new(commands), + ), + ) + .await + })?; + } + responder.respond( acp::NewSessionResponse::new(acp::SessionId::new(session_key)) .config_options(config_options_for_state(&session_config)), @@ -1475,6 +1532,41 @@ async fn main() -> acp::Result<()> { }, agent_client_protocol::on_receive_request!(), ) + .on_receive_request( + { + let state = state.clone(); + async move |arguments: acp::ForkSessionRequest, + responder: Responder, + _cx: ConnectionTo| { + let new_id = state.next_session_id.fetch_add(1, Ordering::SeqCst); + eprintln!( + "Mock agent: session/fork from {} to {new_id}", + arguments.session_id + ); + let session_key = new_id.to_string(); + let session_config = { + let mut configs = state.session_configs.lock().unwrap(); + let mut config = configs + .get(&arguments.session_id.to_string()) + .cloned() + .unwrap_or_else(default_session_config); + // Forked sessions come back with thought_level "high" so + // tests can verify the client applied the fork response's + // config_options rather than keeping its old snapshot. + if config.thought_level.is_some() { + config.thought_level = Some("high".to_string()); + } + configs.insert(session_key.clone(), config.clone()); + config + }; + responder.respond( + acp::ForkSessionResponse::new(acp::SessionId::new(session_key)) + .config_options(config_options_for_state(&session_config)), + ) + } + }, + agent_client_protocol::on_receive_request!(), + ) .on_receive_request( { let state = state.clone(); diff --git a/nori-rs/nori-protocol/docs.md b/nori-rs/nori-protocol/docs.md index 9abf95887..32487b26b 100644 --- a/nori-rs/nori-protocol/docs.md +++ b/nori-rs/nori-protocol/docs.md @@ -26,7 +26,7 @@ agent_client_protocol_schema::SessionUpdate ### Core Implementation - **`ClientEventNormalizer`** maintains a `HashMap` keyed by `call_id`. `ToolCallUpdate` messages always upsert into that map: if the ACP agent never sent an initial `ToolCall`, the normalizer synthesizes a placeholder `ToolCall`, applies the update fields, and still emits a visible `ToolSnapshot`. -- **`SessionRuntime` support types** in `session_runtime.rs` define the reducer-owned ACP runtime model used by `nori-acp`: `SessionPhase`, `PersistedSessionState`, `ActiveRequestState`, `OpenMessage`, and `QueuedPrompt`. These types let the backend treat prompt turns, `session/load`, queued prompts, and ownership of tool/approval updates as one ordered state machine instead of reconstructing turn state from racing tasks. `QueuedPromptKind` distinguishes visible user prompts, compaction prompts, and hidden goal continuations so `@/nori-rs/acp/src/backend/session_reducer.rs` can preserve the right queue, transcript, and completion behavior for each path. `ActiveRequestState` keeps the last flushed assistant text so `PromptCompleted { last_agent_message, .. }` remains correct even when a later reasoning chunk closes the assistant buffer before the turn ends. +- **`SessionRuntime` support types** in `session_runtime.rs` define the reducer-owned ACP runtime model used by `nori-acp`: `SessionPhase`, `PersistedSessionState`, `ActiveRequestState`, `OpenMessage`, and `QueuedPrompt`. These types let the backend treat prompt turns, `session/load`, queued prompts, and ownership of tool/approval updates as one ordered state machine instead of reconstructing turn state from racing tasks. `QueuedPromptKind` distinguishes visible user prompts, compaction prompts (`Compact` for the client-side summarize-and-replace path, `NativeCompact` when the agent's own advertised `/compact` command is forwarded and the session is kept), and hidden goal continuations so `@/nori-rs/acp/src/backend/session_reducer.rs` can preserve the right queue, transcript, and completion behavior for each path. `ActiveRequestState` keeps the last flushed assistant text so `PromptCompleted { last_agent_message, .. }` remains correct even when a later reasoning chunk closes the assistant buffer before the turn ends. - **Session update normalization** keeps the first pass intentionally small: - `UserMessageChunk` becomes `MessageDelta { stream: User, .. }`, which lets replay paths reconstruct visible user history during `session/load`. - `CurrentModeUpdate` becomes `ClientEvent::SessionModeChanged { current_mode_id }`; the TUI resolves the id to a human label using its cached mode list. @@ -35,6 +35,7 @@ agent_client_protocol_schema::SessionUpdate - `UsageUpdate` also becomes `SessionUpdateInfo`, but the usage variant additionally carries `SessionUsageState` so the TUI can update footer context without reparsing the display string. - **Persisted session metadata** now includes `session_info` and `session_usage` alongside available commands, current mode, and config options. `nori-acp` owns persistence, but these structs live here so the reducer and replay pipeline share one runtime model. - **Thread-goal client events** carry the current goal snapshot (`objective`, lifecycle status, token usage, active time, and timestamps) or a clear notification. They are not derived from ACP provider messages; they are backend session-state projections emitted through the same `ClientEvent` stream as normalized ACP data. +- **Session lifecycle projections** `ContextCompacted` (optional summary; `None` when the agent compacted natively inside the same session) and `SessionBranched` (the new session id after an ACP `session/fork` swap) are likewise emitted by the ACP backend (`@/nori-rs/acp/src/backend/session_runtime_driver.rs` and `@/nori-rs/acp/src/backend/session_replace.rs`), not derived from provider messages. - **`is_generic_tool_call()`** gates initial `ToolCall` emission: tool calls with no `raw_input`, no `locations`, empty `content`, and no `/` in the title are suppressed (return empty `Vec`). The normalizer still records them internally so that later attributed `ToolCallUpdate` messages can refine the existing call without forcing the TUI to render a placeholder cell first. - **Invocation priority cascade** in `invocation_from_tool_call()` resolves what the tool is doing, in priority order: diff --git a/nori-rs/nori-protocol/src/lib.rs b/nori-rs/nori-protocol/src/lib.rs index 48d7600b2..9867e6bda 100644 --- a/nori-rs/nori-protocol/src/lib.rs +++ b/nori-rs/nori-protocol/src/lib.rs @@ -22,6 +22,7 @@ pub enum ClientEvent { LoadCompleted, QueueChanged(QueueChanged), ContextCompacted(ContextCompacted), + SessionBranched(SessionBranched), ReplayEntry(ReplayEntry), AgentCommandsUpdate(AgentCommandsUpdate), SessionCapabilitiesChanged(SessionCapabilitiesView), @@ -76,6 +77,15 @@ pub struct ContextCompacted { pub summary: Option, } +/// The ACP session was branched (forked at its current state via +/// `session/fork`). The runtime now targets the new session; the original +/// session is preserved on the agent side and remains resumable. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub struct SessionBranched { + pub new_session_id: String, +} + /// A set of commands advertised by the ACP agent. /// Each update fully replaces the previous set. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/nori-rs/nori-protocol/src/session_runtime.rs b/nori-rs/nori-protocol/src/session_runtime.rs index 6388275f5..1f2b40e43 100644 --- a/nori-rs/nori-protocol/src/session_runtime.rs +++ b/nori-rs/nori-protocol/src/session_runtime.rs @@ -197,7 +197,12 @@ pub enum TranscriptRole { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum QueuedPromptKind { User, + /// Client-side compaction: send a summarization prompt, then replace the + /// session and inject the summary into the next prompt. Compact, + /// Agent-native compaction: forward the agent's advertised `/compact` + /// command and stay in the current session. + NativeCompact, GoalContinuation, } diff --git a/nori-rs/protocol/docs.md b/nori-rs/protocol/docs.md index b55ac65c2..2d62ab30d 100644 --- a/nori-rs/protocol/docs.md +++ b/nori-rs/protocol/docs.md @@ -19,7 +19,7 @@ Path: @/nori-rs/protocol **Core Types:** `@/nori-rs/protocol/src/protocol/mod.rs` defines `Submission`, `Op`, `Event`, and `EventMsg`, which form the shared SQ/EQ contract between the UI and whichever backend owns the active session. -**Operations** (`@/nori-rs/protocol/src/protocol/mod.rs`) group backend commands into user-input, lifecycle, approval, history, undo, custom-prompt, and session-control surfaces. The `/goal` feature belongs to that typed command surface through `ThreadGoalGet`, `ThreadGoalSet`, and `ThreadGoalClear`, rather than being smuggled through a normal user prompt. +**Operations** (`@/nori-rs/protocol/src/protocol/mod.rs`) group backend commands into user-input, lifecycle, approval, history, undo, custom-prompt, and session-control surfaces. The `/goal` feature belongs to that typed command surface through `ThreadGoalGet`, `ThreadGoalSet`, and `ThreadGoalClear`, rather than being smuggled through a normal user prompt. Session-control ops include `Compact` and `BranchSession` (branch the ACP session at its current state via `session/fork`, handled in `@/nori-rs/acp/src/backend/session_replace.rs`). **Events** (`@/nori-rs/protocol/src/protocol/mod.rs`) carry shared control-plane updates back to TUI-facing code. Examples include turn lifecycle events, approval prompts, compact-summary notifications, undo results, prompt summaries, hook output, and history lookup results. ACP session-domain rendering uses `@/nori-rs/nori-protocol` instead. @@ -80,7 +80,7 @@ Path: @/nori-rs/protocol | Type | Purpose | |------|---------| -| `ContextCompactedEvent` | Carries an optional `summary: Option` field. When emitted by the ACP backend (`@/nori-rs/acp/`), the summary contains the compact summary text so the TUI can render a session boundary and reprint it. When emitted by the core backend (`@/nori-rs/core/`), the summary is `None` and the TUI shows only an info message. | +| `ContextCompactedEvent` | Carries an optional `summary: Option` field. The ACP backend (`@/nori-rs/acp/`) fills the summary only on its client-side compaction fallback, where the TUI renders a session boundary and reprints the summary. Agent-native ACP compaction (the agent compacts inside the same session) and the core backend (`@/nori-rs/core/`) emit `None`, and the TUI shows only an info message. | **Thread Goal Invariants:** diff --git a/nori-rs/protocol/src/protocol/mod.rs b/nori-rs/protocol/src/protocol/mod.rs index f6ee11042..5311437b8 100644 --- a/nori-rs/protocol/src/protocol/mod.rs +++ b/nori-rs/protocol/src/protocol/mod.rs @@ -239,6 +239,11 @@ pub enum Op { /// to generate a summary which will be returned as an AgentMessage event. Compact, + /// Branch the ACP session from its current state via `session/fork`. + /// The runtime swaps to the forked session; the original session is + /// preserved on the agent side and remains resumable. + BranchSession, + /// Request Codex to undo a turn (turn are stacked so it is the same effect as CMD + Z). Undo, diff --git a/nori-rs/tui/docs.md b/nori-rs/tui/docs.md index d7b572a33..70caac202 100644 --- a/nori-rs/tui/docs.md +++ b/nori-rs/tui/docs.md @@ -331,7 +331,7 @@ During background system info collection on unix, `check_worktree_cleanup()` run | `/login` | Log in to the current agent | | `/logout` | Show logout instructions | | `/switch-skillset [name]` | Switch between available skillsets (with optional direct name) | -| `/fork` | Rewind conversation to a previous message | +| `/fork` | Branch from the current point (native ACP `session/fork`) or rewind to a previous message | | `/browser` | Launch a headed Chrome browser the agent can control via CDP (Chrome DevTools Protocol) | | `/quit` | Exit Nori | | `/exit` | Exit Nori (alias for /quit) | @@ -462,24 +462,24 @@ When the ACP backend sends a `ContextCompactedEvent` with a summary, `on_context 3. Insert a `NoriSessionHeaderCell` (the "Nori CLI" card, same as starting a fresh session) by constructing a `SessionConfiguredEvent` from the current widget config state 4. Reprint the summary text as the first assistant message of the new session -When the event has no summary (core backend path), only the "Context compacted" info message is shown. This asymmetry exists because the core backend compacts history in-place without producing a summary for the TUI. +When the event has no summary, only the "Context compacted" info message is shown. This happens on two paths that both compact without producing a summary for the TUI: the core backend compacts history in-place, and the ACP backend's native compaction path forwards the agent's own `/compact` command so the agent compacts inside the same session -- no session boundary is rendered because no new session was created (see `@/nori-rs/acp/docs.md`, Conversation Compaction). The summary-bearing path is the ACP backend's client-side compaction fallback. **Fork Conversation (`/fork`) (`nori/fork_picker.rs`, `app_backtrack.rs`):** -The `/fork` slash command lets users rewind to a previous user message and branch the conversation from that point. It is only available when no task is running (`available_during_task = false`). The flow: +The `/fork` slash command lets users either branch the conversation at its current point (native ACP `session/fork`) or rewind to a previous user message (local summary-based fork). It is deliberately the single command for both -- there are no `/branch` or `/rewind` aliases. It is only available when no task is running (`available_during_task = false`). The flow: 1. `SlashCommand::Fork` dispatches `AppEvent::OpenForkPicker` -2. The handler calls `collect_user_messages()` in `app_backtrack.rs` to gather all user messages from the current session segment (messages after the last `SessionInfoCell`). If none exist, an info message is shown instead of the picker. -3. `fork_picker_params()` in `nori/fork_picker.rs` builds a `SelectionViewParams` with items displayed newest-first (reversed from chronological order). Message previews are truncated to 80 characters; multiline messages show only the first line with an ellipsis. -4. Selecting a message fires `AppEvent::ForkToMessage { nth_user_message, prefill }` -5. The `ForkToMessage` handler: +2. The handler gathers all user messages from the transcript via `app_backtrack.rs`. If none exist, an info message is shown instead of the picker. +3. `fork_picker_params()` in `nori/fork_picker.rs` builds a `SelectionViewParams` whose **first entry is "βŽ‡ Branch from current point"**, followed by message entries displayed newest-first (reversed from chronological order). Message previews are truncated to 80 characters; multiline messages show only the first line with an ellipsis. +4. Selecting the branch entry fires `Op::BranchSession`. The ACP backend forks the live session at its head via ACP `session/fork` and swaps to the fork; the original session is preserved on the agent side and stays resumable. The backend rejects branching with a user-facing error when the agent does not advertise the fork capability or a turn is in progress (see `@/nori-rs/acp/docs.md`, Session Branching). On success the backend emits `ClientEvent::SessionBranched`, which the TUI renders as an info cell ("Branched conversation -- the original session is preserved and can be reopened with /resume") in `chatwidget/event_handlers.rs`. +5. Selecting a message instead fires `AppEvent::ForkToMessage { cell_index, prefill }`, the local summary-based backtrack fork. That handler: - Calls `build_fork_summary()` to create a plain-text summary of the conversation up to (but not including) the selected message, formatted as `User: ...\nAssistant: ...\n` pairs - Shuts down the current conversation - Creates a new `ChatWidget` with `fork_context` set to the summary string - - Trims `transcript_cells` to the fork point via `trim_transcript_cells_to_nth_user()` so the TUI preserves visual history before the fork + - Trims `transcript_cells` to the fork point so the TUI preserves visual history before the fork - Prefills the composer with the selected message text -The fork context flows through `ChatWidgetInit.fork_context` -> `spawn_agent()` -> `spawn_acp_agent()` -> `AcpBackendConfig.initial_context`, which initializes the ACP backend's `pending_compact_summary`. This reuses the same mechanism as `/compact` and `/resume` -- the summary is prepended to the first user prompt in the new session, giving the agent prior conversation context without a protocol-level session fork. +The local fork context flows through `ChatWidgetInit.fork_context` -> `spawn_agent()` -> `spawn_acp_agent()` -> `AcpBackendConfig.initial_context`, which initializes the ACP backend's `pending_compact_summary`. This reuses the same mechanism as `/compact`'s client-side fallback and `/resume` -- the summary is prepended to the first user prompt in the new session, giving the agent prior conversation context without a protocol-level session fork. The rewind path stays local because ACP `session/fork` only forks at the session head; fork-at-message is tracked upstream as ACP PR #629. **Caller-injected agents (`nori cloud`):** `Cli.extra_agents` (a clap-skipped field on `@/nori-rs/tui/src/cli.rs`, never a CLI flag) carries extra `AgentConfigToml` registry entries from the caller. `run_main()` in `@/nori-rs/tui/src/lib.rs` appends them after the config's `[[agents]]` when initializing the agent registry. The CLI's `nori cloud` subcommand uses this to pin a synthetic `nori-cloud` entry that runs `nori-handroll cloud-acp` (see `@/nori-rs/cli/src/cloud.rs` and `@/nori-rs/cli/docs.md`); from the TUI's perspective it is an ordinary local ACP agent and `spawn_agent()` treats it like any other registry entry. There is no cloud-specific plumbing in the TUI -- the old `cloud_connection` threading through `Cli`/`App`/`ChatWidgetInit` was removed. diff --git a/nori-rs/tui/src/chatwidget/event_handlers.rs b/nori-rs/tui/src/chatwidget/event_handlers.rs index e6e7de86f..58d51ca26 100644 --- a/nori-rs/tui/src/chatwidget/event_handlers.rs +++ b/nori-rs/tui/src/chatwidget/event_handlers.rs @@ -1147,6 +1147,14 @@ impl ChatWidget { summary: context_compacted.summary, }); } + nori_protocol::ClientEvent::SessionBranched(_) => { + self.add_info_message( + "Branched conversation β€” the original session is preserved and can be \ + reopened with /resume" + .to_owned(), + None, + ); + } nori_protocol::ClientEvent::ReplayEntry(replay_entry) => { self.handle_client_replay_entry(replay_entry); } diff --git a/nori-rs/tui/src/chatwidget/tests/part5.rs b/nori-rs/tui/src/chatwidget/tests/part5.rs index 9a2ef5700..7b1f6c0e0 100644 --- a/nori-rs/tui/src/chatwidget/tests/part5.rs +++ b/nori-rs/tui/src/chatwidget/tests/part5.rs @@ -935,3 +935,27 @@ fn repeated_agent_messages_after_tool_use_share_one_separator() { assert_eq!(rendered.matches("Worked for").count(), 1, "{rendered}"); } + +/// A SessionBranched client event renders the branched-conversation info cell +/// (the runtime swapped to a fork; the original session stays resumable). +#[test] +fn session_branched_shows_info_message() { + let (mut chat, mut rx, _ops) = make_chatwidget_manual(); + + chat.handle_client_event(nori_protocol::ClientEvent::SessionBranched( + nori_protocol::SessionBranched { + new_session_id: "42".to_string(), + }, + )); + + let cells = drain_insert_history(&mut rx); + let combined = cells + .iter() + .map(|c| lines_to_single_string(c)) + .collect::>() + .join(""); + assert!( + combined.contains("Branched conversation"), + "expected branched-conversation info message: {combined:?}" + ); +} diff --git a/nori-rs/tui/src/nori/fork_picker.rs b/nori-rs/tui/src/nori/fork_picker.rs index 6cc235d76..fc9e6f4fd 100644 --- a/nori-rs/tui/src/nori/fork_picker.rs +++ b/nori-rs/tui/src/nori/fork_picker.rs @@ -1,7 +1,8 @@ -//! Fork picker component for rewinding conversations. +//! Fork picker component for branching and rewinding conversations. //! -//! This module provides the UI for selecting a previous user message -//! to rewind the conversation to. +//! This module provides the UI for either branching the conversation at its +//! current point (native ACP `session/fork`) or selecting a previous user +//! message to rewind to (local summary-based fork). use crate::app_event::AppEvent; use crate::app_event_sender::AppEventSender; @@ -37,31 +38,43 @@ pub fn fork_picker_params( messages: Vec<(usize, String)>, _app_event_tx: AppEventSender, ) -> SelectionViewParams { - let items: Vec = messages - .into_iter() - .rev() - .map(|(cell_index, message)| { - let preview = truncate_preview(&message); - let actions: Vec = vec![Box::new(move |tx| { - tx.send(AppEvent::ForkToMessage { - cell_index, - prefill: message.clone(), - }); - })]; - SelectionItem { - name: preview, - description: None, - is_current: false, - actions, - dismiss_on_select: true, - ..Default::default() - } - }) - .collect(); + // First entry: branch at the current point via the agent's native + // `session/fork` (no rewind). Earlier messages rewind locally instead. + let branch_actions: Vec = vec![Box::new(move |tx| { + tx.send(AppEvent::CodexOp(codex_core::protocol::Op::BranchSession)); + })]; + let mut items = vec![SelectionItem { + name: "βŽ‡ Branch from current point".to_string(), + description: Some("duplicate this conversation and continue; the original session is preserved (requires agent support)".to_string()), + is_current: false, + actions: branch_actions, + dismiss_on_select: true, + ..Default::default() + }]; + + items.extend(messages.into_iter().rev().map(|(cell_index, message)| { + let preview = truncate_preview(&message); + let actions: Vec = vec![Box::new(move |tx| { + tx.send(AppEvent::ForkToMessage { + cell_index, + prefill: message.clone(), + }); + })]; + SelectionItem { + name: preview, + description: None, + is_current: false, + actions, + dismiss_on_select: true, + ..Default::default() + } + })); SelectionViewParams { title: Some("Fork Conversation".to_string()), - subtitle: Some("Select a message to rewind to".to_string()), + subtitle: Some( + "Branch from the current point, or select a message to rewind to".to_string(), + ), footer_hint: Some(standard_popup_hint_line()), items, is_searchable: false, @@ -84,10 +97,62 @@ mod tests { } #[test] - fn fork_picker_with_no_messages_returns_empty_items() { + fn fork_picker_renders_branch_entry_and_messages() { + use crate::render::renderable::Renderable; + use ratatui::buffer::Buffer; + use ratatui::layout::Rect; + + let (tx, _rx) = make_tx(); + let params = fork_picker_params( + vec![ + (0, "refactor the parser".to_string()), + (1, "now add tests".to_string()), + ], + tx.clone(), + ); + let view = crate::bottom_pane::ListSelectionView::new(params, tx); + + let width = 64; + let height = view.desired_height(width); + let area = Rect::new(0, 0, width, height); + let mut buf = Buffer::empty(area); + view.render(area, &mut buf); + let rendered = (0..area.height) + .map(|row| { + (0..area.width) + .map(|col| { + let symbol = buf[(col, row)].symbol(); + if symbol.is_empty() { " " } else { symbol } + }) + .collect::() + }) + .collect::>() + .join("\n"); + insta::assert_snapshot!("fork_picker_with_branch_entry", rendered); + } + + #[test] + fn fork_picker_with_no_messages_offers_only_branch_entry() { let (tx, _rx) = make_tx(); let params = fork_picker_params(vec![], tx); - assert!(params.items.is_empty()); + assert_eq!(params.items.len(), 1); + assert_eq!(params.items[0].name, "βŽ‡ Branch from current point"); + } + + #[test] + fn fork_picker_branch_entry_fires_branch_session_op() { + let (tx, _rx) = make_tx(); + let params = fork_picker_params(vec![(0, "hello".to_string())], tx); + + let (verify_tx, mut verify_rx) = unbounded_channel::(); + let verify_sender = AppEventSender::new(verify_tx); + (params.items[0].actions[0])(&verify_sender); + + let event = verify_rx.try_recv().expect("should have received event"); + match event { + AppEvent::CodexOp(codex_core::protocol::Op::BranchSession) => {} + other => panic!("expected CodexOp(BranchSession), got {other:?}"), + } } #[test] @@ -110,10 +175,11 @@ mod tests { ]; let params = fork_picker_params(messages, tx); - assert_eq!(params.items.len(), 3); - assert_eq!(params.items[0].name, "third message"); - assert_eq!(params.items[1].name, "second message"); - assert_eq!(params.items[2].name, "first message"); + assert_eq!(params.items.len(), 4); + assert_eq!(params.items[0].name, "βŽ‡ Branch from current point"); + assert_eq!(params.items[1].name, "third message"); + assert_eq!(params.items[2].name, "second message"); + assert_eq!(params.items[3].name, "first message"); } #[test] @@ -123,9 +189,9 @@ mod tests { let messages = vec![(0, long_msg.clone())]; let params = fork_picker_params(messages, tx); - assert_eq!(params.items.len(), 1); - assert!(params.items[0].name.len() < long_msg.len()); - assert!(params.items[0].name.ends_with('…')); + assert_eq!(params.items.len(), 2); + assert!(params.items[1].name.len() < long_msg.len()); + assert!(params.items[1].name.ends_with('…')); } #[test] @@ -134,8 +200,8 @@ mod tests { let messages = vec![(0, "first line\nsecond line\nthird line".to_string())]; let params = fork_picker_params(messages, tx); - assert_eq!(params.items.len(), 1); - assert_eq!(params.items[0].name, "first line…"); + assert_eq!(params.items.len(), 2); + assert_eq!(params.items[1].name, "first line…"); } #[test] @@ -144,11 +210,12 @@ mod tests { let messages = vec![(0, "first".to_string()), (1, "second".to_string())]; let params = fork_picker_params(messages, tx); - // Execute the action for the first item (newest-first, so index 0 = message 1) - assert!(!params.items[0].actions.is_empty()); + // Execute the action for the first message item (item 0 is the branch + // entry; messages are newest-first, so index 1 = message 1) + assert!(!params.items[1].actions.is_empty()); let (verify_tx, mut verify_rx) = unbounded_channel::(); let verify_sender = AppEventSender::new(verify_tx); - (params.items[0].actions[0])(&verify_sender); + (params.items[1].actions[0])(&verify_sender); let event = verify_rx.try_recv().expect("should have received event"); match event { @@ -172,7 +239,7 @@ mod tests { // Last item in picker = oldest message (index 0) let (verify_tx, mut verify_rx) = unbounded_channel::(); let verify_sender = AppEventSender::new(verify_tx); - (params.items[1].actions[0])(&verify_sender); + (params.items[2].actions[0])(&verify_sender); let event = verify_rx.try_recv().expect("should have received event"); match event { diff --git a/nori-rs/tui/src/nori/snapshots/nori_tui__nori__fork_picker__tests__fork_picker_with_branch_entry.snap b/nori-rs/tui/src/nori/snapshots/nori_tui__nori__fork_picker__tests__fork_picker_with_branch_entry.snap new file mode 100644 index 000000000..765f4dad7 --- /dev/null +++ b/nori-rs/tui/src/nori/snapshots/nori_tui__nori__fork_picker__tests__fork_picker_with_branch_entry.snap @@ -0,0 +1,17 @@ +--- +source: tui/src/nori/fork_picker.rs +assertion_line: 128 +expression: rendered +--- + + Fork Conversation + Branch from the current point, or select a message to rewind + +β€Ί 1. βŽ‡ Branch from current point duplicate this conversation + and continue; the original + session is preserved + (requires agent support) + 2. now add tests + 3. refactor the parser + + ↑/k ↓/j to navigate, enter to confirm, esc to go back diff --git a/nori-rs/tui/src/slash_command.rs b/nori-rs/tui/src/slash_command.rs index 03017dbbe..6c67acc1e 100644 --- a/nori-rs/tui/src/slash_command.rs +++ b/nori-rs/tui/src/slash_command.rs @@ -67,7 +67,7 @@ impl SlashCommand { SlashCommand::Login => "log in to the current agent", SlashCommand::Logout => "show logout instructions", SlashCommand::SwitchSkillset => "switch between available skillsets", - SlashCommand::Fork => "rewind conversation to a previous message", + SlashCommand::Fork => "branch from the current point or rewind to a previous message", SlashCommand::Browser => "open a Chrome browser the agent can control via CDP", } } diff --git a/nori-rs/tui/src/viewonly_transcript.rs b/nori-rs/tui/src/viewonly_transcript.rs index 0e58768e1..eb3640bda 100644 --- a/nori-rs/tui/src/viewonly_transcript.rs +++ b/nori-rs/tui/src/viewonly_transcript.rs @@ -175,6 +175,7 @@ fn format_client_event(event: &nori_protocol::ClientEvent) -> Option { | nori_protocol::ClientEvent::LoadCompleted | nori_protocol::ClientEvent::QueueChanged(_) | nori_protocol::ClientEvent::ContextCompacted(_) + | nori_protocol::ClientEvent::SessionBranched(_) | nori_protocol::ClientEvent::ReplayEntry(_) | nori_protocol::ClientEvent::AgentCommandsUpdate(_) | nori_protocol::ClientEvent::SessionCapabilitiesChanged(_) From d12ecd9538bbe3f5cee50632eff84656dfdaf5a3 Mon Sep 17 00:00:00 2001 From: Clifford Ressel Date: Fri, 3 Jul 2026 21:13:02 -0400 Subject: [PATCH 2/3] fix(acp,tui): address self-review findings for compact/branch - Transcript loader now prefers the last recorded SessionBranched id so resuming a branched transcript targets the forked session - Cancelled fallback compaction no longer captures a truncated summary or replaces the session (same EndTurn invariant as native compact) - Branch busy-error wording covers loads, not just turns - /fork picker opens with the branch entry even with no user messages --- nori-rs/acp/src/backend/session_replace.rs | 5 +- .../acp/src/backend/session_runtime_driver.rs | 8 ++ nori-rs/acp/src/backend/tests/part8.rs | 92 +++++++++++++++++++ nori-rs/acp/src/transcript/loader.rs | 51 +++++++++- nori-rs/tui/src/app/event_handling.rs | 17 ++-- 5 files changed, 161 insertions(+), 12 deletions(-) diff --git a/nori-rs/acp/src/backend/session_replace.rs b/nori-rs/acp/src/backend/session_replace.rs index 1070f7d9e..323124ba2 100644 --- a/nori-rs/acp/src/backend/session_replace.rs +++ b/nori-rs/acp/src/backend/session_replace.rs @@ -116,8 +116,11 @@ impl AcpBackend { .await; return; } + // Rejects prompts and loads alike; prompts still queued after a + // cancelled turn are intentionally allowed through β€” they follow the + // user to the forked session. if !self.session_driver.lock().await.is_idle() { - self.send_error("Cannot branch while a turn is in progress.") + self.send_error("Cannot branch while the session is busy.") .await; return; } diff --git a/nori-rs/acp/src/backend/session_runtime_driver.rs b/nori-rs/acp/src/backend/session_runtime_driver.rs index db2cab683..50fade125 100644 --- a/nori-rs/acp/src/backend/session_runtime_driver.rs +++ b/nori-rs/acp/src/backend/session_runtime_driver.rs @@ -491,6 +491,14 @@ impl AcpBackend { self.maybe_submit_goal_continuation(completed_turn).await; } QueuedPromptKind::Compact => { + // A cancelled compaction did not compact: don't capture the + // truncated summary or replace the session (same invariant as + // the NativeCompact arm below). + if completed_turn.stop_reason + != agent_client_protocol_schema::v1::StopReason::EndTurn + { + return; + } let Some(summary) = completed_turn.last_agent_message.clone() else { return; }; diff --git a/nori-rs/acp/src/backend/tests/part8.rs b/nori-rs/acp/src/backend/tests/part8.rs index b384eacd3..69661cfb0 100644 --- a/nori-rs/acp/src/backend/tests/part8.rs +++ b/nori-rs/acp/src/backend/tests/part8.rs @@ -432,3 +432,95 @@ async fn test_branch_session_applies_forked_config_options() { "the fork response's config_options must replace the live snapshot" ); } + +/// A cancelled client-side (fallback) compaction must not be recorded as a +/// compaction either: no ContextCompacted, no session swap. +#[tokio::test] +#[serial] +async fn test_fallback_compact_cancelled_records_no_compaction() { + use std::time::Duration; + + if !mock_agent_available() { + return; + } + // No MOCK_AGENT_AVAILABLE_COMMANDS: /compact takes the client-side path. + let _no_commands = EnvGuard::remove("MOCK_AGENT_AVAILABLE_COMMANDS"); + let _stream = EnvGuard::set("MOCK_AGENT_STREAM_UNTIL_CANCEL", "1"); + let _echo_session = EnvGuard::set("MOCK_AGENT_ECHO_SESSION_ID", "1"); + + let temp_dir = tempfile::tempdir().expect("Failed to create temp dir"); + let (backend, _event_rx, mut client_event_rx) = spawn_backend_for_test(&temp_dir).await; + + backend + .submit(Op::Compact) + .await + .expect("Failed to submit Op::Compact"); + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + assert!( + std::time::Instant::now() < deadline, + "compact turn never started streaming" + ); + if let Ok(Some(nori_protocol::ClientEvent::MessageDelta(_))) = + tokio::time::timeout(Duration::from_secs(1), client_event_rx.recv()).await + { + break; + } + } + backend + .submit(Op::Interrupt) + .await + .expect("Failed to submit Op::Interrupt"); + let cancelled_turn = capture_turn(&mut client_event_rx).await; + + assert_eq!( + cancelled_turn.prompt_completed.stop_reason, + nori_protocol::StopReason::Cancelled + ); + assert_eq!( + cancelled_turn.context_compacted, + Vec::new(), + "a cancelled fallback compaction must not be recorded as a compaction" + ); + + // The original session is still the live one (no replacement session). + backend + .submit(Op::UserInput { + items: vec![codex_protocol::user_input::UserInput::Text { + text: "still here?".to_string(), + }], + }) + .await + .expect("Failed to submit follow-up prompt"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + let mut follow_up_text = String::new(); + loop { + assert!( + std::time::Instant::now() < deadline, + "follow-up turn never started streaming; got: {follow_up_text:?}" + ); + if let Ok(Some(nori_protocol::ClientEvent::MessageDelta(delta))) = + tokio::time::timeout(Duration::from_secs(1), client_event_rx.recv()).await + && delta.stream == nori_protocol::MessageStream::Answer + { + follow_up_text.push_str(&delta.delta); + if follow_up_text.contains("SESSION:") { + break; + } + } + } + assert!( + follow_up_text.contains("SESSION:0"), + "a cancelled fallback compact must not swap sessions, got: {follow_up_text:?}" + ); + backend + .submit(Op::Interrupt) + .await + .expect("Failed to submit follow-up interrupt"); + let follow_up_turn = capture_turn(&mut client_event_rx).await; + assert_eq!( + follow_up_turn.prompt_completed.stop_reason, + nori_protocol::StopReason::Cancelled + ); +} diff --git a/nori-rs/acp/src/transcript/loader.rs b/nori-rs/acp/src/transcript/loader.rs index 5626bcc70..4839aee4b 100644 --- a/nori-rs/acp/src/transcript/loader.rs +++ b/nori-rs/acp/src/transcript/loader.rs @@ -734,9 +734,20 @@ async fn load_transcript_from_path(path: &Path) -> io::Result { entries.push(parsed); } - let meta = + let mut meta = meta.ok_or_else(|| io::Error::other("transcript does not contain session metadata"))?; + // A branch (`session/fork`) swaps the live agent session after the + // metadata line was written; the last recorded SessionBranched event + // supersedes the spawn-time id so resume targets the forked session. + for parsed in &entries { + if let TranscriptEntry::ClientEvent(entry) = &parsed.entry + && let nori_protocol::ClientEvent::SessionBranched(branched) = &entry.event + { + meta.acp_session_id = Some(branched.new_session_id.clone()); + } + } + tracing::info!( target: "nori_resume", phase = "transcript_loader.load_transcript.done", @@ -763,6 +774,44 @@ mod tests { use pretty_assertions::assert_eq; use tempfile::TempDir; + /// After a `/fork` branch, the transcript's effective ACP session id must + /// be the forked session (recorded as a SessionBranched client event), not + /// the id captured in the session metadata at spawn time β€” otherwise + /// resuming the transcript reopens the pre-branch session and silently + /// drops post-branch context. + #[tokio::test] + async fn load_transcript_prefers_last_session_branched_id() { + let temp_dir = TempDir::new().unwrap(); + let recorder = TranscriptRecorder::new( + temp_dir.path(), + temp_dir.path(), + None, + "0.1.0", + Some("0".to_string()), + ) + .await + .unwrap(); + recorder + .record_client_event(&nori_protocol::ClientEvent::SessionBranched( + nori_protocol::SessionBranched { + new_session_id: "1".to_string(), + }, + )) + .await + .unwrap(); + recorder.flush().await.unwrap(); + recorder.shutdown().await.unwrap(); + + let transcript = load_transcript_from_path(recorder.transcript_path()) + .await + .unwrap(); + assert_eq!( + transcript.meta.acp_session_id.as_deref(), + Some("1"), + "resuming a branched transcript must target the forked session" + ); + } + #[tokio::test] async fn test_list_projects_empty() { let temp_dir = TempDir::new().unwrap(); diff --git a/nori-rs/tui/src/app/event_handling.rs b/nori-rs/tui/src/app/event_handling.rs index c418a87cf..ec109f273 100644 --- a/nori-rs/tui/src/app/event_handling.rs +++ b/nori-rs/tui/src/app/event_handling.rs @@ -1149,18 +1149,15 @@ impl App { .add_error_message(format!("Failed to launch browser: {err}")); } AppEvent::OpenForkPicker => { + // With no user messages the picker still offers branching at + // the current point (e.g. an empty-but-resumed session). let messages = crate::app_backtrack::collect_all_user_messages(&self.transcript_cells); - if messages.is_empty() { - self.chat_widget - .add_info_message("No messages to fork from.".to_string(), None); - } else { - let params = crate::nori::fork_picker::fork_picker_params( - messages, - self.app_event_tx.clone(), - ); - self.chat_widget.show_selection_view(params); - } + let params = crate::nori::fork_picker::fork_picker_params( + messages, + self.app_event_tx.clone(), + ); + self.chat_widget.show_selection_view(params); tui.frame_requester().schedule_frame(); } AppEvent::ForkToMessage { From 82d7fb34630a19b5404db3b9d8aa6d3aa91c8ad6 Mon Sep 17 00:00:00 2001 From: Clifford Ressel Date: Fri, 3 Jul 2026 21:20:03 -0400 Subject: [PATCH 3/3] chore(deps): fix cargo-deny advisories (anyhow bump, quick-xml bump, scoped ignores) New RustSec advisories landed upstream and fail cargo-deny on any fresh run (main included). anyhow 1.0.100->1.0.103 fixes RUSTSEC-2026-0190; quick-xml 0.38.0->0.38.4 stays current though no in-range fix exists yet for RUSTSEC-2026-0194/0195 (trusted-input paths only: wayland-scanner build-time XML, syntect bundled themes) so those are ignored with reasons, as is RUSTSEC-2026-0189 (rmcp Streamable HTTP *server* transport; this codebase is client-only, loopback server is axum). --- nori-rs/Cargo.lock | 10 +++++----- nori-rs/deny.toml | 3 +++ 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/nori-rs/Cargo.lock b/nori-rs/Cargo.lock index 206ddbbb5..051f1dfe0 100644 --- a/nori-rs/Cargo.lock +++ b/nori-rs/Cargo.lock @@ -220,9 +220,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.100" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "arboard" @@ -4392,7 +4392,7 @@ checksum = "3af6b589e163c5a788fab00ce0c0366f6efbb9959c2f9874b224936af7fce7e1" dependencies = [ "base64", "indexmap 2.12.0", - "quick-xml 0.38.0", + "quick-xml 0.38.4", "serde", "time", ] @@ -4653,9 +4653,9 @@ dependencies = [ [[package]] name = "quick-xml" -version = "0.38.0" +version = "0.38.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8927b0664f5c5a98265138b7e3f90aa19a6b21353182469ace36d4ac527b7b1b" +checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c" dependencies = [ "memchr", ] diff --git a/nori-rs/deny.toml b/nori-rs/deny.toml index cc7920b41..2274596e2 100644 --- a/nori-rs/deny.toml +++ b/nori-rs/deny.toml @@ -76,6 +76,9 @@ unmaintained = "workspace" ignore = [ { id = "RUSTSEC-2026-0002", reason = "lru 0.12.5 is pinned by the ratatui patch branch in [patch.crates-io]; track removal once ratatui can move to a fixed lru release" }, { id = "RUSTSEC-2026-0097", reason = "rand 0.8.x pinned by oauth2/zbus/secret-service; no custom logger uses ThreadRng in this codebase" }, + { id = "RUSTSEC-2026-0194", reason = "quick-xml quadratic attribute check: no fixed release in range (0.37.x via wayland-scanner build-time protocol XML, 0.38.x via plist/syntect bundled themes); neither path parses attacker-controlled XML. Track removal once a patched quick-xml lands and dependents move" }, + { id = "RUSTSEC-2026-0195", reason = "quick-xml NsReader memory exhaustion: same pinned versions and trusted-input paths as RUSTSEC-2026-0194" }, + { id = "RUSTSEC-2026-0189", reason = "rmcp 0.12 pinned by codex-rmcp-client; advisory is in rmcp's Streamable HTTP *server* transport and this codebase uses rmcp as an MCP client only (the backend's loopback nori-client MCP server is axum-based). Track upgrade to rmcp >=1.4" }, ] # If this is true, then cargo deny will use the git executable to fetch advisory database. # If this is false, then it uses a built-in git library.