From 4f7047201c11ae03ccb2d426d3156d890e4b5d8f Mon Sep 17 00:00:00 2001
From: SaguaroDev <74339271+SaguaroDev@users.noreply.github.com>
Date: Sun, 10 May 2026 18:16:32 -0400
Subject: [PATCH 1/6] feat(state): add messages.active flag + rewind primitives
 (#21910)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Schema v12 adds:
- messages.active (default 1) — soft-delete flag for /rewind
- sessions.rewind_count (default 0) — audit counter
- idx_messages_session_active deferred index

New SessionDB methods:
- rewind_to_message(session_id, target_message_id) — soft-deletes rows
  >= target_id, refuses non-user targets, increments rewind_count
- restore_rewound(session_id, since_message_id) — undo for stretch goal
- list_recent_user_messages — picker source

Existing methods get include_inactive kwarg (default False):
- get_messages, get_messages_as_conversation, search_messages.
  Rewound rows excluded from session_search by default — opt-in for audit.

The deferred index pattern (DEFERRED_INDEX_SQL run after _reconcile_columns)
avoids 'no such column: active' on legacy pre-v12 databases, since
executescript(SCHEMA_SQL) runs before column reconciliation.
---
 hermes_state.py            | 247 +++++++++++++++++++++++++++++++++++--
 tests/test_hermes_state.py |   6 +-
 2 files changed, 243 insertions(+), 10 deletions(-)

diff --git a/hermes_state.py b/hermes_state.py
index 51d9f0b406f9..231033b7cfc5 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -33,7 +33,7 @@
 
 DEFAULT_DB_PATH = get_hermes_home() / "state.db"
 
-SCHEMA_VERSION = 11
+SCHEMA_VERSION = 12
 
 # ---------------------------------------------------------------------------
 # WAL-compatibility fallback
@@ -218,6 +218,7 @@ def _log_wal_fallback_once(db_label: str, exc: Exception) -> None:
     handoff_state TEXT,
     handoff_platform TEXT,
     handoff_error TEXT,
+    rewind_count INTEGER NOT NULL DEFAULT 0,
     FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
 );
 
@@ -236,7 +237,8 @@ def _log_wal_fallback_once(db_label: str, exc: Exception) -> None:
     reasoning_content TEXT,
     reasoning_details TEXT,
     codex_reasoning_items TEXT,
-    codex_message_items TEXT
+    codex_message_items TEXT,
+    active INTEGER NOT NULL DEFAULT 1
 );
 
 CREATE TABLE IF NOT EXISTS state_meta (
@@ -250,6 +252,15 @@ def _log_wal_fallback_once(db_label: str, exc: Exception) -> None:
 CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp);
 """
 
+# Indexes that reference columns added in later schema versions must be
+# created AFTER _reconcile_columns() has had a chance to ADD them on
+# existing databases. SCHEMA_SQL above is run by sqlite executescript
+# which would otherwise fail on legacy DBs ("no such column: active").
+DEFERRED_INDEX_SQL = """
+CREATE INDEX IF NOT EXISTS idx_messages_session_active
+    ON messages(session_id, active, timestamp);
+"""
+
 FTS_SQL = """
 CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
     content
@@ -571,6 +582,13 @@ def _init_schema(self):
         # column gets created here.
         self._reconcile_columns(cursor)
 
+        # ── Deferred indexes ───────────────────────────────────────────
+        # Indexes that reference columns added by reconcile_columns()
+        # above must run AFTER reconciliation, not as part of SCHEMA_SQL
+        # (which sqlite3.executescript would fail to apply against a
+        # legacy table missing those columns).
+        cursor.executescript(DEFERRED_INDEX_SQL)
+
         # ── Schema version bookkeeping ─────────────────────────────────
         # Bump to current so future data migrations (if any) can gate on
         # version.  No version-gated column additions remain.
@@ -648,6 +666,22 @@ def _init_schema(self):
                     "COALESCE(tool_calls, '') "
                     "FROM messages"
                 )
+            if current_version < 12:
+                # v12: messages.active flag for /rewind soft-deletion.
+                # The declarative reconcile_columns() above adds the
+                # column itself; this UPDATE is belt-and-suspenders to
+                # ensure any rows that pre-existed the ADD COLUMN have
+                # active=1 rather than NULL.  Required because SQLite's
+                # ALTER TABLE ADD COLUMN populates existing rows with
+                # the DEFAULT, but only when the DEFAULT is a literal —
+                # NULL would otherwise be possible on some older builds
+                # or if reconciliation ever ran without a DEFAULT.
+                try:
+                    cursor.execute(
+                        "UPDATE messages SET active = 1 WHERE active IS NULL"
+                    )
+                except sqlite3.OperationalError:
+                    pass
             if current_version < SCHEMA_VERSION:
                 cursor.execute(
                     "UPDATE schema_version SET version = ?",
@@ -1596,11 +1630,24 @@ def _do(conn):
 
         self._execute_write(_do)
 
-    def get_messages(self, session_id: str) -> List[Dict[str, Any]]:
-        """Load all messages for a session, ordered by insertion order."""
+    def get_messages(
+        self, session_id: str, include_inactive: bool = False
+    ) -> List[Dict[str, Any]]:
+        """Load messages for a session in insertion order.
+
+        By default only active messages are returned. Pass
+        ``include_inactive=True`` to load soft-deleted rows (e.g. for
+        audit / debug views of rewound history). See
+        :meth:`rewind_to_message` for the soft-delete mechanic.
+
+        Ordered by AUTOINCREMENT id (true insertion order) rather than
+        timestamp — see c03acca50 for the WSL2 clock-regression rationale.
+        """
+        active_clause = "" if include_inactive else " AND active = 1"
         with self._lock:
             cursor = self._conn.execute(
-                "SELECT * FROM messages WHERE session_id = ? ORDER BY id",
+                "SELECT * FROM messages WHERE session_id = ?"
+                f"{active_clause} ORDER BY id",
                 (session_id,),
             )
             rows = cursor.fetchall()
@@ -1882,23 +1929,32 @@ def resolve_resume_session_id(self, session_id: str) -> str:
         return session_id
 
     def get_messages_as_conversation(
-        self, session_id: str, include_ancestors: bool = False
+        self,
+        session_id: str,
+        include_ancestors: bool = False,
+        include_inactive: bool = False,
     ) -> List[Dict[str, Any]]:
         """
         Load messages in the OpenAI conversation format (role + content dicts).
         Used by the gateway to restore conversation history.
+
+        By default only active messages are returned. Pass
+        ``include_inactive=True`` to load soft-deleted (rewound) rows
+        as well. See :meth:`rewind_to_message`.
         """
         session_ids = [session_id]
         if include_ancestors:
             session_ids = self._session_lineage_root_to_tip(session_id)
 
+        active_clause = "" if include_inactive else " AND active = 1"
         with self._lock:
             placeholders = ",".join("?" for _ in session_ids)
             rows = self._conn.execute(
                 "SELECT role, content, tool_call_id, tool_calls, tool_name, "
                 "finish_reason, reasoning, reasoning_content, reasoning_details, "
                 "codex_reasoning_items, codex_message_items "
-                f"FROM messages WHERE session_id IN ({placeholders}) ORDER BY id",
+                f"FROM messages WHERE session_id IN ({placeholders})"
+                f"{active_clause} ORDER BY id",
                 tuple(session_ids),
             ).fetchall()
 
@@ -1987,6 +2043,175 @@ def _is_duplicate_replayed_user_message(messages: List[Dict[str, Any]], msg: Dic
                 return False
         return False
 
+    # =========================================================================
+    # Rewind (soft-delete) — see /rewind slash command + issue #21910
+    # =========================================================================
+
+    def rewind_to_message(
+        self, session_id: str, target_message_id: int
+    ) -> Dict[str, Any]:
+        """Soft-delete all messages with id >= ``target_message_id`` in *session_id*.
+
+        The target message itself becomes inactive as well so the caller
+        can pre-fill it as the next user prompt without it appearing
+        twice in the replayed transcript.  Rewound rows are kept on
+        disk with ``active=0`` for audit / forensic inspection — use
+        :meth:`get_messages` with ``include_inactive=True`` to see them.
+
+        Returns a dict::
+
+            {
+                "rewound_count": int,    # number of rows newly flipped to active=0
+                "target_message": dict,  # full row dict of the target
+                "new_head_id":   int|None  # id of the last still-active row, or None
+            }
+
+        Raises ``ValueError`` if the target message does not exist in
+        *session_id* or if its role is not ``"user"``.
+
+        Always increments ``sessions.rewind_count`` — even when the
+        target is already inactive — so the counter accurately reflects
+        the number of rewind operations performed against the session.
+        Idempotent on the ``active`` flag: re-rewinding past the same
+        target is a no-op on row state but still bumps the counter.
+        """
+
+        # 1) Validate target up-front (read-only, outside the write txn).
+        with self._lock:
+            row = self._conn.execute(
+                "SELECT * FROM messages WHERE id = ? AND session_id = ?",
+                (target_message_id, session_id),
+            ).fetchone()
+        if row is None:
+            raise ValueError(
+                f"message {target_message_id} not found in session {session_id}"
+            )
+        target_row = dict(row)
+        if target_row.get("role") != "user":
+            raise ValueError(
+                f"rewind target must be a 'user' message (got role="
+                f"{target_row.get('role')!r}, id={target_message_id})"
+            )
+
+        # Decode content for callers (prefill the prompt buffer).
+        target_row["content"] = self._decode_content(target_row.get("content"))
+
+        rewound: List[int] = []
+
+        def _do(conn):
+            cursor = conn.execute(
+                "SELECT id FROM messages "
+                "WHERE session_id = ? AND id >= ? AND active = 1",
+                (session_id, target_message_id),
+            )
+            ids = [r[0] for r in cursor.fetchall()]
+            if ids:
+                placeholders = ",".join("?" for _ in ids)
+                conn.execute(
+                    f"UPDATE messages SET active = 0 WHERE id IN ({placeholders})",
+                    ids,
+                )
+            conn.execute(
+                "UPDATE sessions SET rewind_count = COALESCE(rewind_count, 0) + 1 "
+                "WHERE id = ?",
+                (session_id,),
+            )
+            return ids
+
+        rewound = self._execute_write(_do)
+
+        # 2) Compute new head id (largest still-active row id in session).
+        with self._lock:
+            head_row = self._conn.execute(
+                "SELECT MAX(id) FROM messages WHERE session_id = ? AND active = 1",
+                (session_id,),
+            ).fetchone()
+        new_head_id = head_row[0] if head_row and head_row[0] is not None else None
+
+        return {
+            "rewound_count": len(rewound),
+            "target_message": target_row,
+            "new_head_id": new_head_id,
+        }
+
+    def restore_rewound(self, session_id: str, since_message_id: int) -> int:
+        """Mark inactive messages with id >= *since_message_id* active again.
+
+        Returns the number of rows flipped back to ``active=1``.
+        Intended for undo-of-rewind and test cleanup; not wired to a
+        slash command in v1.
+        """
+        def _do(conn):
+            cursor = conn.execute(
+                "SELECT id FROM messages "
+                "WHERE session_id = ? AND id >= ? AND active = 0",
+                (session_id, since_message_id),
+            )
+            ids = [r[0] for r in cursor.fetchall()]
+            if ids:
+                placeholders = ",".join("?" for _ in ids)
+                conn.execute(
+                    f"UPDATE messages SET active = 1 WHERE id IN ({placeholders})",
+                    ids,
+                )
+            return len(ids)
+
+        return self._execute_write(_do)
+
+    def list_recent_user_messages(
+        self,
+        session_id: str,
+        limit: int = 20,
+        include_inactive: bool = False,
+    ) -> List[Dict[str, Any]]:
+        """Return the *limit* most-recent user messages, newest first.
+
+        Each entry is a dict with keys ``id``, ``timestamp``, ``preview``.
+        ``preview`` is the first 80 characters of the message content
+        (with line breaks collapsed to spaces). Used by the /rewind
+        slash command picker.
+
+        By default only active messages are returned.
+        """
+        active_clause = "" if include_inactive else " AND active = 1"
+        with self._lock:
+            cursor = self._conn.execute(
+                "SELECT id, timestamp, content FROM messages "
+                "WHERE session_id = ? AND role = 'user'"
+                f"{active_clause} "
+                "ORDER BY id DESC LIMIT ?",
+                (session_id, int(limit)),
+            )
+            rows = cursor.fetchall()
+
+        result: List[Dict[str, Any]] = []
+        for row in rows:
+            decoded = self._decode_content(row["content"])
+            if isinstance(decoded, list):
+                # Multimodal — flatten text parts.
+                text_parts = [
+                    p.get("text", "") for p in decoded
+                    if isinstance(p, dict) and p.get("type") == "text"
+                ]
+                preview = " ".join(t for t in text_parts if t).strip()
+                if not preview:
+                    preview = "[multimodal content]"
+            elif isinstance(decoded, str):
+                preview = decoded
+            else:
+                preview = ""
+            preview = " ".join(preview.split())  # collapse whitespace
+            if len(preview) > 80:
+                preview = preview[:77] + "..."
+            result.append(
+                {
+                    "id": row["id"],
+                    "timestamp": row["timestamp"],
+                    "preview": preview,
+                }
+            )
+        return result
+
     # =========================================================================
     # Search
     # =========================================================================
@@ -2084,6 +2309,7 @@ def search_messages(
         limit: int = 20,
         offset: int = 0,
         sort: str = None,
+        include_inactive: bool = False,
     ) -> List[Dict[str, Any]]:
         """
         Full-text search across session messages using FTS5.
@@ -2105,6 +2331,9 @@ def search_messages(
         The short-CJK LIKE fallback already orders by timestamp DESC and
         ignores ``sort``. The trigram CJK path honours ``sort`` like the main
         FTS5 path.
+
+        Rewound (``active=0``) rows are excluded by default. Pass
+        ``include_inactive=True`` to search every row.
         """
         if not query or not query.strip():
             return []
@@ -2135,6 +2364,8 @@ def search_messages(
         # Build WHERE clauses dynamically
         where_clauses = ["messages_fts MATCH ?"]
         params: list = [query]
+        if not include_inactive:
+            where_clauses.append("m.active = 1")
 
         if source_filter is not None:
             source_placeholders = ",".join("?" for _ in source_filter)
@@ -2214,6 +2445,8 @@ def search_messages(
                 trigram_query = " ".join(parts)
                 tri_where = ["messages_fts_trigram MATCH ?"]
                 tri_params: list = [trigram_query]
+                if not include_inactive:
+                    tri_where.append("m.active = 1")
                 if source_filter is not None:
                     tri_where.append(f"s.source IN ({','.join('?' for _ in source_filter)})")
                     tri_params.extend(source_filter)
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index 3bae763b9412..4ffb282f88cf 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -1447,7 +1447,7 @@ def test_tables_exist(self, db):
     def test_schema_version(self, db):
         cursor = db._conn.execute("SELECT version FROM schema_version")
         version = cursor.fetchone()[0]
-        assert version == 11
+        assert version == 12
 
     def test_title_column_exists(self, db):
         """Verify the title column was created in the sessions table."""
@@ -1744,7 +1744,7 @@ def test_migration_from_v2(self, tmp_path):
 
         # Verify migration
         cursor = migrated_db._conn.execute("SELECT version FROM schema_version")
-        assert cursor.fetchone()[0] == 11
+        assert cursor.fetchone()[0] == 12
 
         # Verify title column exists and is NULL for existing sessions
         session = migrated_db.get_session("existing")
@@ -2939,7 +2939,7 @@ def test_v10_to_v11_upgrade_backfills_tool_fields(self, tmp_path):
                 "SELECT version FROM schema_version LIMIT 1"
             ).fetchone()
             version = row["version"] if hasattr(row, "keys") else row[0]
-            assert version == 11
+            assert version == 12
         finally:
             session_db.close()
 

From 48c465e0a80ef2d6ad70f770002ea2dfae06d92b Mon Sep 17 00:00:00 2001
From: SaguaroDev <74339271+SaguaroDev@users.noreply.github.com>
Date: Sun, 10 May 2026 18:17:57 -0400
Subject: [PATCH 2/6] feat(memory): add rewound kwarg to on_session_switch hook

---
 agent/memory_manager.py  | 6 ++++++
 agent/memory_provider.py | 5 +++++
 2 files changed, 11 insertions(+)

diff --git a/agent/memory_manager.py b/agent/memory_manager.py
index 79547139086f..b95568b9f8b1 100644
--- a/agent/memory_manager.py
+++ b/agent/memory_manager.py
@@ -460,6 +460,7 @@ def on_session_switch(
         *,
         parent_session_id: str = "",
         reset: bool = False,
+        rewound: bool = False,
         **kwargs,
     ) -> None:
         """Notify all providers that the agent's session_id has rotated.
@@ -472,6 +473,10 @@ def on_session_switch(
         per-session state so subsequent writes land in the correct
         session's record. See ``MemoryProvider.on_session_switch`` for
         the full contract.
+
+        ``rewound=True`` signals that session_id is unchanged but the
+        transcript was truncated; providers caching per-turn document
+        state should invalidate.
         """
         if not new_session_id:
             return
@@ -481,6 +486,7 @@ def on_session_switch(
                     new_session_id,
                     parent_session_id=parent_session_id,
                     reset=reset,
+                    rewound=rewound,
                     **kwargs,
                 )
             except Exception as e:
diff --git a/agent/memory_provider.py b/agent/memory_provider.py
index c9abc48c7a92..5f068a0e5bd2 100644
--- a/agent/memory_provider.py
+++ b/agent/memory_provider.py
@@ -166,6 +166,7 @@ def on_session_switch(
         *,
         parent_session_id: str = "",
         reset: bool = False,
+        rewound: bool = False,
         **kwargs,
     ) -> None:
         """Called when the agent switches session_id mid-process.
@@ -195,6 +196,10 @@ def on_session_switch(
             (``_session_turns``, ``_turn_counter``, etc.) when this is
             set. ``False`` for ``/resume`` / ``/branch`` / compression
             where the logical conversation continues under the new id.
+        rewound:
+            ``True`` if session_id is unchanged but the transcript was
+            truncated; providers caching per-turn document state should
+            invalidate.
 
         Default is no-op for backward compatibility.
         """

From 7845d4d774d4e2eea53d6207d5cd8c7f33bee79d Mon Sep 17 00:00:00 2001
From: SaguaroDev <74339271+SaguaroDev@users.noreply.github.com>
Date: Sun, 10 May 2026 18:19:19 -0400
Subject: [PATCH 3/6] feat(cli): add /rewind slash command and handler

---
 cli.py                 | 127 +++++++++++++++++++++++++++++++++++++++++
 hermes_cli/commands.py |   1 +
 2 files changed, 128 insertions(+)

diff --git a/cli.py b/cli.py
index 423b96a73d65..407bcfa5eaa8 100644
--- a/cli.py
+++ b/cli.py
@@ -6428,6 +6428,131 @@ def _handle_branch_command(self, cmd_original: str) -> None:
         _cprint(f"  Original session: {parent_session_id}")
         _cprint(f"  Branch session:   {new_session_id}")
 
+    def _handle_rewind_command(self, cmd_original: str) -> None:
+        """Handle /rewind — pick a previous user message and rewind to it.
+
+        Soft-deletes (active=0) every message at-or-after the chosen user
+        turn, reloads the active transcript, and pre-fills the prompt
+        buffer with the chosen message so the user can edit and resubmit.
+        Memory providers are notified with ``rewound=True`` so per-turn
+        caches invalidate.
+        """
+        if not self.conversation_history:
+            _cprint("  No conversation to rewind — send a message first.")
+            return
+        if not self._session_db:
+            from hermes_state import format_session_db_unavailable
+            _cprint(f"  {format_session_db_unavailable()}")
+            return
+
+        try:
+            items = self._session_db.list_recent_user_messages(
+                self.session_id, limit=10
+            )
+        except Exception as e:
+            _cprint(f"  Failed to load message history: {e}")
+            return
+
+        if not items:
+            _cprint("  No user messages to rewind to.")
+            return
+
+        labels = [f"{i+1}. {item['preview']}" for i, item in enumerate(items)]
+        idx = self._run_curses_picker(
+            title="Rewind to message", items=labels, default_index=0
+        )
+        if idx is None:
+            return  # cancelled
+
+        try:
+            selected_id = items[idx]["id"]
+        except (IndexError, KeyError, TypeError):
+            _cprint("  Invalid selection.")
+            return
+
+        try:
+            result = self._session_db.rewind_to_message(
+                self.session_id, selected_id
+            )
+        except ValueError as e:
+            _cprint(f"  Rewind failed: {e}")
+            return
+        except Exception as e:
+            _cprint(f"  Rewind failed: {e}")
+            return
+
+        # Reload active-only transcript
+        try:
+            self.conversation_history = (
+                self._session_db.get_messages_as_conversation(self.session_id)
+            )
+        except Exception:
+            # Defensive — leave history alone if reload fails
+            pass
+
+        # Agent state surgery — mirrors /branch path
+        if self.agent:
+            try:
+                self.agent.reset_session_state()
+            except Exception:
+                pass
+            if hasattr(self.agent, "_invalidate_system_prompt"):
+                try:
+                    self.agent._invalidate_system_prompt()
+                except Exception:
+                    pass
+            if hasattr(self.agent, "_last_flushed_db_idx"):
+                try:
+                    self.agent._last_flushed_db_idx = len(self.conversation_history)
+                except Exception:
+                    pass
+
+            _mm = getattr(self.agent, "_memory_manager", None)
+            if _mm is not None:
+                try:
+                    _mm.on_session_switch(
+                        self.session_id,
+                        parent_session_id="",
+                        reset=False,
+                        rewound=True,
+                    )
+                except Exception:
+                    pass
+
+        # Pre-fill the prompt buffer with the chosen message text so the
+        # user can edit-and-resubmit.  Gateway / non-prompt-toolkit
+        # invocations have no live buffer, so fall back to printing.
+        target_msg = result.get("target_message") or {}
+        target_text = target_msg.get("content") or ""
+        if isinstance(target_text, list):
+            # Multimodal — flatten text parts for the buffer prefill.
+            parts = [
+                p.get("text", "") for p in target_text
+                if isinstance(p, dict) and p.get("type") == "text"
+            ]
+            target_text = "\n".join(t for t in parts if t)
+
+        prefilled = False
+        app = getattr(self, "_app", None)
+        if app is not None and isinstance(target_text, str) and target_text:
+            try:
+                buf = app.current_buffer
+                if buf is not None and hasattr(buf, "text"):
+                    buf.text = target_text
+                    if hasattr(buf, "cursor_position"):
+                        buf.cursor_position = len(target_text)
+                    prefilled = True
+            except Exception:
+                prefilled = False
+
+        rewound_count = result.get("rewound_count", 0)
+        _cprint(
+            f"  ↶ Rewound {rewound_count} message(s)."
+            " Edit and resubmit, or send a new message."
+        )
+        if not prefilled and isinstance(target_text, str) and target_text:
+            _cprint(f"  Previous message: {target_text}")
+
     def save_conversation(self):
         """Save the current conversation to a JSON snapshot under ~/.hermes/sessions/saved/.
 
@@ -7909,6 +8034,8 @@ def process_command(self, command: str) -> bool:
             self.undo_last()
         elif canonical == "branch":
             self._handle_branch_command(cmd_original)
+        elif canonical == "rewind":
+            self._handle_rewind_command(cmd_original)
         elif canonical == "save":
             self.save_conversation()
         elif canonical == "cron":
diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py
index 1e42fb9421eb..d98c1d4d324a 100644
--- a/hermes_cli/commands.py
+++ b/hermes_cli/commands.py
@@ -77,6 +77,7 @@ class CommandDef:
                cli_only=True),
     CommandDef("retry", "Retry the last message (resend to agent)", "Session"),
     CommandDef("undo", "Remove the last user/assistant exchange", "Session"),
+    CommandDef("rewind", "Rewind to a previous user message and re-prompt", "Session"),
     CommandDef("title", "Set a title for the current session", "Session",
                args_hint="[name]"),
     CommandDef("handoff", "Hand off this session to a messaging platform (Telegram, Discord, etc.)", "Session",

From 572e042ca2fa069f9739d850e4c1cbccd51f4efc Mon Sep 17 00:00:00 2001
From: SaguaroDev <74339271+SaguaroDev@users.noreply.github.com>
Date: Sun, 10 May 2026 18:21:56 -0400
Subject: [PATCH 4/6] test(rewind): cover DB primitives and CLI handler
 (#21910)

---
 tests/cli/test_rewind_command.py | 194 +++++++++++++++++++++++++++++++
 tests/test_hermes_state.py       | 178 ++++++++++++++++++++++++++++
 2 files changed, 372 insertions(+)
 create mode 100644 tests/cli/test_rewind_command.py

diff --git a/tests/cli/test_rewind_command.py b/tests/cli/test_rewind_command.py
new file mode 100644
index 000000000000..24d768edd2b7
--- /dev/null
+++ b/tests/cli/test_rewind_command.py
@@ -0,0 +1,194 @@
+"""Tests for the /rewind slash command — pick a previous user message and re-prompt.
+
+These tests exercise ``HermesCLI._handle_rewind_command`` against a real
+:class:`SessionDB`, with the curses picker patched to return a known index.
+"""
+
+import os
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def session_db(tmp_path):
+    """Create a real SessionDB for testing."""
+    os.environ["HERMES_HOME"] = str(tmp_path / ".hermes")
+    os.makedirs(tmp_path / ".hermes", exist_ok=True)
+    from hermes_state import SessionDB
+    db = SessionDB(db_path=tmp_path / ".hermes" / "test_rewind.db")
+    yield db
+    db.close()
+
+
+@pytest.fixture
+def cli_instance(tmp_path, session_db):
+    """Minimal HermesCLI-like MagicMock for testing _handle_rewind_command."""
+    cli = MagicMock()
+    cli._session_db = session_db
+    cli.session_id = "20260510_120000_rewind"
+    cli.model = "anthropic/claude-sonnet-4.6"
+    cli.session_start = datetime.now()
+    cli._app = None  # No prompt_toolkit app — exercise the print-fallback path.
+    cli.agent = None
+
+    # Seed: 3 user turns w/ assistant responses.
+    session_db.create_session(
+        session_id=cli.session_id,
+        source="cli",
+        model=cli.model,
+    )
+    session_db.append_message(cli.session_id, role="user", content="first thing")
+    session_db.append_message(cli.session_id, role="assistant", content="reply 1")
+    session_db.append_message(cli.session_id, role="user", content="second thing")
+    session_db.append_message(cli.session_id, role="assistant", content="reply 2")
+    session_db.append_message(cli.session_id, role="user", content="third thing")
+    session_db.append_message(cli.session_id, role="assistant", content="reply 3")
+
+    cli.conversation_history = session_db.get_messages_as_conversation(cli.session_id)
+    return cli
+
+
+class TestRewindHandler:
+    """Test _handle_rewind_command CLI handler behaviour."""
+
+    def test_rewind_truncates_history(self, cli_instance, session_db):
+        """Picking a user message rewinds the DB and reloads conversation_history."""
+        from cli import HermesCLI
+
+        # list_recent_user_messages returns newest-first → index 0 is "third",
+        # index 1 is "second", index 2 is "first". Pick "second".
+        cli_instance._run_curses_picker = MagicMock(return_value=1)
+        HermesCLI._handle_rewind_command(cli_instance, "/rewind")
+
+        contents = [m["content"] for m in cli_instance.conversation_history]
+        assert contents == ["first thing", "reply 1"]
+
+    def test_rewind_cancel_returns_silently(self, cli_instance, session_db):
+        """Picker returning None (cancel) leaves history untouched."""
+        from cli import HermesCLI
+        before = list(cli_instance.conversation_history)
+
+        cli_instance._run_curses_picker = MagicMock(return_value=None)
+        HermesCLI._handle_rewind_command(cli_instance, "/rewind")
+
+        assert cli_instance.conversation_history == before
+        # No rewind row flipped in DB either
+        all_rows = session_db.get_messages(cli_instance.session_id, include_inactive=True)
+        active_rows = session_db.get_messages(cli_instance.session_id)
+        assert len(all_rows) == len(active_rows)
+
+    def test_rewind_handler_no_session_db(self, cli_instance):
+        """Without a SessionDB, the handler bails gracefully and leaves history intact."""
+        from cli import HermesCLI
+        cli_instance._session_db = None
+        before = list(cli_instance.conversation_history)
+        cli_instance._run_curses_picker = MagicMock()
+
+        # Should not raise, should not call the picker either.
+        HermesCLI._handle_rewind_command(cli_instance, "/rewind")
+        assert cli_instance._run_curses_picker.call_count == 0
+        assert cli_instance.conversation_history == before
+
+    def test_rewind_handler_empty_history(self, cli_instance):
+        """Empty conversation_history bails before touching the DB."""
+        from cli import HermesCLI
+        cli_instance.conversation_history = []
+        cli_instance._run_curses_picker = MagicMock()
+
+        HermesCLI._handle_rewind_command(cli_instance, "/rewind")
+        assert cli_instance._run_curses_picker.call_count == 0
+
+    def test_memory_provider_notified_on_rewind(self, cli_instance, session_db):
+        """Memory manager hook fires with rewound=True after a successful rewind."""
+        from cli import HermesCLI
+
+        mm = MagicMock()
+        agent = MagicMock()
+        agent._memory_manager = mm
+        # Mirror the hasattr checks in the handler — make these accessor-friendly.
+        agent._last_flushed_db_idx = 0
+        cli_instance.agent = agent
+        cli_instance._run_curses_picker = MagicMock(return_value=0)
+
+        HermesCLI._handle_rewind_command(cli_instance, "/rewind")
+
+        assert mm.on_session_switch.call_count == 1
+        args, kwargs = mm.on_session_switch.call_args
+        assert args[0] == cli_instance.session_id
+        assert kwargs["rewound"] is True
+        assert kwargs["reset"] is False
+        # parent_session_id is empty for a same-session rewind
+        assert kwargs["parent_session_id"] == ""
+
+    def test_rewind_no_user_messages(self, tmp_path, session_db):
+        """Session with only system/assistant messages → 'no user messages' bail."""
+        from cli import HermesCLI
+
+        cli = MagicMock()
+        cli._session_db = session_db
+        cli.session_id = "20260510_120001_empty"
+        cli._app = None
+        cli.agent = None
+        session_db.create_session(session_id=cli.session_id, source="cli")
+        session_db.append_message(cli.session_id, role="assistant", content="greetings")
+        cli.conversation_history = session_db.get_messages_as_conversation(cli.session_id)
+        cli._run_curses_picker = MagicMock()
+
+        HermesCLI._handle_rewind_command(cli, "/rewind")
+        assert cli._run_curses_picker.call_count == 0  # bailed before opening picker
+
+    def test_rewind_prefills_prompt_buffer_when_app_present(
+        self, cli_instance, session_db
+    ):
+        """When ``self._app`` is wired, the chosen message lands in current_buffer."""
+        from cli import HermesCLI
+
+        buf = SimpleNamespace(text="", cursor_position=0)
+        cli_instance._app = SimpleNamespace(current_buffer=buf)
+        cli_instance._run_curses_picker = MagicMock(return_value=1)
+
+        HermesCLI._handle_rewind_command(cli_instance, "/rewind")
+
+        assert buf.text == "second thing"
+        assert buf.cursor_position == len("second thing")
+
+    def test_rewind_invokes_agent_state_reset(self, cli_instance, session_db):
+        """Successful rewind calls reset_session_state and updates _last_flushed_db_idx."""
+        from cli import HermesCLI
+
+        agent = MagicMock()
+        agent._last_flushed_db_idx = 999
+        cli_instance.agent = agent
+        cli_instance._run_curses_picker = MagicMock(return_value=0)
+
+        HermesCLI._handle_rewind_command(cli_instance, "/rewind")
+
+        assert agent.reset_session_state.called
+        assert agent._invalidate_system_prompt.called
+        # After rewinding to the newest user message (index 0), only that single
+        # user message is left active, so _last_flushed_db_idx should reflect 0
+        # (rewind soft-deletes the target message itself per spec).
+        assert agent._last_flushed_db_idx == len(cli_instance.conversation_history)
+
+
+class TestRewindCommandDef:
+    """The CommandDef registration for /rewind."""
+
+    def test_rewind_in_registry(self):
+        from hermes_cli.commands import COMMAND_REGISTRY
+        names = [c.name for c in COMMAND_REGISTRY]
+        assert "rewind" in names
+
+    def test_rewind_in_session_category(self):
+        from hermes_cli.commands import COMMAND_REGISTRY
+        rewind = next(c for c in COMMAND_REGISTRY if c.name == "rewind")
+        assert rewind.category == "Session"
+
+    def test_rewind_resolves(self):
+        from hermes_cli.commands import resolve_command
+        result = resolve_command("rewind")
+        assert result is not None
+        assert result.name == "rewind"
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index 4ffb282f88cf..397b5f7b22f1 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -2943,3 +2943,181 @@ def test_v10_to_v11_upgrade_backfills_tool_fields(self, tmp_path):
         finally:
             session_db.close()
 
+
+
+# =========================================================================
+# /rewind primitives (#21910)
+# =========================================================================
+
+class TestRewindPrimitives:
+    """SessionDB.rewind_to_message + restore_rewound + list_recent_user_messages."""
+
+    def _seed(self, db, session_id="rw1"):
+        db.create_session(session_id=session_id, source="cli")
+        return session_id
+
+    def test_rewind_to_first_user_message(self, db):
+        """One user msg, rewind to it — that row becomes inactive."""
+        sid = self._seed(db)
+        db.append_message(sid, role="user", content="hello")
+        uid = db.get_messages(sid)[0]["id"]
+
+        result = db.rewind_to_message(sid, uid)
+
+        assert result["rewound_count"] == 1
+        assert result["target_message"]["content"] == "hello"
+        assert result["new_head_id"] is None
+        assert db.get_messages(sid) == []
+        assert len(db.get_messages(sid, include_inactive=True)) == 1
+
+    def test_rewind_mid_session(self, db):
+        """4 user turns w/ assistant responses, rewind to turn 2."""
+        sid = self._seed(db, "rw_mid")
+        ids = []
+        for i in range(1, 5):
+            db.append_message(sid, role="user", content=f"u{i}")
+            db.append_message(sid, role="assistant", content=f"a{i}")
+        rows = db.get_messages(sid)
+        # turn 2 user id
+        user_msgs = [r for r in rows if r["role"] == "user"]
+        target_id = user_msgs[1]["id"]  # u2
+
+        result = db.rewind_to_message(sid, target_id)
+
+        # u2 + a2 + u3 + a3 + u4 + a4 = 6 messages flipped
+        assert result["rewound_count"] == 6
+        active = db.get_messages(sid)
+        contents = [m["content"] for m in active]
+        assert contents == ["u1", "a1"]
+
+    def test_rewind_then_continue(self, db):
+        """Rewind, append a new message, conversation reflects active head only."""
+        sid = self._seed(db, "rw_cont")
+        db.append_message(sid, role="user", content="u1")
+        db.append_message(sid, role="assistant", content="a1")
+        db.append_message(sid, role="user", content="u2")
+        db.append_message(sid, role="assistant", content="a2")
+        rows = db.get_messages(sid)
+        target_id = [r for r in rows if r["role"] == "user"][1]["id"]
+
+        db.rewind_to_message(sid, target_id)
+        db.append_message(sid, role="user", content="u2-edited")
+
+        conv = db.get_messages_as_conversation(sid)
+        contents = [m["content"] for m in conv]
+        assert contents == ["u1", "a1", "u2-edited"]
+
+    def test_rewind_across_tool_call_boundary(self, db):
+        """Assistant + tool result rows are all flipped together when target precedes them."""
+        sid = self._seed(db, "rw_tool")
+        db.append_message(sid, role="user", content="first")
+        db.append_message(sid, role="assistant", content="ok")
+        db.append_message(sid, role="user", content="use a tool")
+        tool_calls = [
+            {"id": "call_1", "function": {"name": "web_search", "arguments": "{}"}},
+        ]
+        db.append_message(sid, role="assistant", content="", tool_calls=tool_calls)
+        db.append_message(
+            sid, role="tool", content="result", tool_name="web_search",
+            tool_call_id="call_1",
+        )
+        rows = db.get_messages(sid)
+        target_id = [r for r in rows if r["role"] == "user"][1]["id"]  # "use a tool"
+
+        result = db.rewind_to_message(sid, target_id)
+
+        # target user + assistant w/ tool_calls + tool result = 3 rows
+        assert result["rewound_count"] == 3
+        active = db.get_messages(sid)
+        roles = [m["role"] for m in active]
+        assert roles == ["user", "assistant"]
+
+    def test_rewound_rows_preserved_in_db(self, db):
+        """include_inactive=True surfaces rewound rows; rewind_count is bumped."""
+        sid = self._seed(db, "rw_audit")
+        db.append_message(sid, role="user", content="keep me forever")
+        uid = db.get_messages(sid)[0]["id"]
+
+        db.rewind_to_message(sid, uid)
+
+        all_rows = db.get_messages(sid, include_inactive=True)
+        assert len(all_rows) == 1
+        assert all_rows[0]["content"] == "keep me forever"
+        sess = db.get_session(sid)
+        assert sess["rewind_count"] == 1
+
+        # Re-rewinding past the same target is a no-op on rows but still bumps the counter.
+        db.rewind_to_message(sid, uid)
+        sess = db.get_session(sid)
+        assert sess["rewind_count"] == 2
+
+    def test_session_search_excludes_inactive_by_default(self, db):
+        """search_messages skips rewound content unless include_inactive=True."""
+        sid = self._seed(db, "rw_search1")
+        db.append_message(sid, role="user", content="UNIQ_REWIND_TOKEN once")
+        uid = db.get_messages(sid)[0]["id"]
+        db.rewind_to_message(sid, uid)
+
+        hits = db.search_messages("UNIQ_REWIND_TOKEN")
+        assert hits == []
+
+    def test_session_search_includes_inactive_when_opted_in(self, db):
+        """Opting in returns the rewound rows."""
+        sid = self._seed(db, "rw_search2")
+        db.append_message(sid, role="user", content="UNIQTOKEN42 hello")
+        uid = db.get_messages(sid)[0]["id"]
+        db.rewind_to_message(sid, uid)
+
+        hits = db.search_messages("UNIQTOKEN42", include_inactive=True)
+        assert len(hits) >= 1
+
+    def test_rewind_target_must_be_user_role(self, db):
+        """ValueError when target is assistant or tool."""
+        sid = self._seed(db, "rw_role")
+        db.append_message(sid, role="user", content="u1")
+        db.append_message(sid, role="assistant", content="a1")
+        rows = db.get_messages(sid)
+        assistant_id = [r for r in rows if r["role"] == "assistant"][0]["id"]
+
+        with pytest.raises(ValueError):
+            db.rewind_to_message(sid, assistant_id)
+
+    def test_rewind_target_must_belong_to_session(self, db):
+        """ValueError when target id belongs to a different session."""
+        sid_a = self._seed(db, "rw_sa")
+        sid_b = self._seed(db, "rw_sb")
+        db.append_message(sid_a, role="user", content="hi from A")
+        a_id = db.get_messages(sid_a)[0]["id"]
+
+        # Try to rewind session B to a message that belongs to session A.
+        with pytest.raises(ValueError):
+            db.rewind_to_message(sid_b, a_id)
+
+    def test_list_recent_user_messages_excludes_inactive_by_default(self, db):
+        sid = self._seed(db, "rw_list")
+        db.append_message(sid, role="user", content="u1")
+        db.append_message(sid, role="assistant", content="a1")
+        db.append_message(sid, role="user", content="u2")
+        rows = db.get_messages(sid)
+        u2_id = [r for r in rows if r["role"] == "user"][1]["id"]
+
+        db.rewind_to_message(sid, u2_id)
+
+        items = db.list_recent_user_messages(sid)
+        assert [i["preview"] for i in items] == ["u1"]
+        items_all = db.list_recent_user_messages(sid, include_inactive=True)
+        assert sorted(i["preview"] for i in items_all) == ["u1", "u2"]
+
+    def test_restore_rewound_round_trips(self, db):
+        sid = self._seed(db, "rw_restore")
+        db.append_message(sid, role="user", content="u1")
+        db.append_message(sid, role="assistant", content="a1")
+        rows = db.get_messages(sid)
+        u_id = [r for r in rows if r["role"] == "user"][0]["id"]
+
+        db.rewind_to_message(sid, u_id)
+        assert db.get_messages(sid) == []
+
+        restored = db.restore_rewound(sid, u_id)
+        assert restored == 2
+        assert len(db.get_messages(sid)) == 2

From ad395b396693503a22a33e54c3b4cf04a2d91167 Mon Sep 17 00:00:00 2001
From: SaguaroDev <74339271+SaguaroDev@users.noreply.github.com>
Date: Sun, 10 May 2026 18:42:03 -0400
Subject: [PATCH 5/6] fix(memory): keep rewound kwarg in **kwargs to preserve
 hook signature

Adding rewound as an explicit named kwarg on MemoryManager.on_session_switch
broke existing tests that assert exact equality on the extra kwargs dict
captured by providers (test_manager_fans_out_to_all_providers,
test_manager_reset_flag_preserved).

The MemoryProvider base class still documents rewound as a known kwarg,
but the manager now passes it through **kwargs like any other forwarded
flag. The CLI rewind handler already passes rewound=True keyword-style,
so behavior is identical.
---
 agent/memory_manager.py | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/agent/memory_manager.py b/agent/memory_manager.py
index b95568b9f8b1..38f76347e1ee 100644
--- a/agent/memory_manager.py
+++ b/agent/memory_manager.py
@@ -460,7 +460,6 @@ def on_session_switch(
         *,
         parent_session_id: str = "",
         reset: bool = False,
-        rewound: bool = False,
         **kwargs,
     ) -> None:
         """Notify all providers that the agent's session_id has rotated.
@@ -474,9 +473,10 @@ def on_session_switch(
         session's record. See ``MemoryProvider.on_session_switch`` for
         the full contract.
 
-        ``rewound=True`` signals that session_id is unchanged but the
-        transcript was truncated; providers caching per-turn document
-        state should invalidate.
+        Callers may pass ``rewound=True`` (forwarded via ``**kwargs``)
+        to signal that session_id is unchanged but the transcript was
+        truncated; providers caching per-turn document state should
+        invalidate.
         """
         if not new_session_id:
             return
@@ -486,7 +486,6 @@ def on_session_switch(
                     new_session_id,
                     parent_session_id=parent_session_id,
                     reset=reset,
-                    rewound=rewound,
                     **kwargs,
                 )
             except Exception as e:

From 7bb57680fd756d9e32acfb290b68afc2c0578fe5 Mon Sep 17 00:00:00 2001
From: SaguaroDev <74339271+SaguaroDev@users.noreply.github.com>
Date: Sun, 10 May 2026 18:45:32 -0400
Subject: [PATCH 6/6] feat(tui): wire /rewind through command.dispatch +
 prefill payload (#21910)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds the TUI half of the /rewind feature so the Ink terminal UI gets
the same affordance as the prompt_toolkit CLI.

Python side (tui_gateway/server.py):
- /rewind added to _PENDING_INPUT_COMMANDS so slash.exec rejects it
  and the TUI falls through to command.dispatch (the only path with
  access to live session state + memory hooks).
- New command.dispatch branch for name == "rewind":
  v1 auto-picks the most recent user turn (Claude-Code-style single-
  step undo), calls SessionDB.rewind_to_message, refreshes the
  in-memory history, fires _memory_manager.on_session_switch with
  rewound=True, and returns the new "prefill" payload.
- A dedicated picker overlay (multi-step rewind) is tracked as a
  follow-up to #21910.

TS side (ui-tui/src/):
- New "prefill" variant on CommandDispatchResponse + asCommandDispatch
  validator. Mirrors "send" but does NOT auto-submit; the client drops
  the message into the composer for editing.
- createSlashHandler renders the optional notice via sys() and calls
  ctx.composer.setInput(d.message), letting the user edit-and-resubmit
  the rewound turn — the core UX promised by the issue.

Tests:
- 7 new tui_gateway tests covering prefill payload shape, in-memory
  history truncation, DB soft-delete, memory-provider notification
  (rewound=True), busy-session refusal, missing-session error, and
  registry placement in _PENDING_INPUT_COMMANDS.
- Extended asCommandDispatch vitest covering the new prefill variant
  (with + without notice, and rejection of malformed payloads).

Out of scope for v1 (tracked as #21910 follow-up):
- Dedicated picker overlay in Ink (the multi-step rewind UI). v1 auto-
  picks the most recent user turn, matching the most common case.
- Gateway platforms (Telegram, Discord, etc.) — issue scopes v1 to
  CLI + TUI only.
---
 tests/tui_gateway/test_rewind_command.py      | 154 ++++++++++++++++++
 tui_gateway/server.py                         |  89 ++++++++++
 .../src/__tests__/asCommandDispatch.test.ts   |  11 ++
 ui-tui/src/app/createSlashHandler.ts          |  13 ++
 ui-tui/src/gatewayTypes.ts                    |   1 +
 ui-tui/src/lib/rpc.ts                         |   8 +
 6 files changed, 276 insertions(+)
 create mode 100644 tests/tui_gateway/test_rewind_command.py

diff --git a/tests/tui_gateway/test_rewind_command.py b/tests/tui_gateway/test_rewind_command.py
new file mode 100644
index 000000000000..ae2de14e2346
--- /dev/null
+++ b/tests/tui_gateway/test_rewind_command.py
@@ -0,0 +1,154 @@
+"""Tests for /rewind handling in tui_gateway.
+
+The TUI routes ``/rewind`` through ``command.dispatch`` (it's in
+``_PENDING_INPUT_COMMANDS`` because the CLI handler queues input the
+slash-worker subprocess can't read). The server handles it directly,
+mutates SessionDB to soft-delete rows, refreshes the in-memory session
+history, fires the memory-provider hook with ``rewound=True``, and
+returns ``{"type": "prefill", "message": <text>, "notice": ...}`` so
+the Ink client drops the message into the composer for editing.
+See issue #21910.
+"""
+
+from __future__ import annotations
+
+import importlib
+import threading
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from hermes_state import SessionDB
+
+
+@pytest.fixture()
+def hermes_home(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    yield home
+
+
+@pytest.fixture()
+def server(hermes_home):
+    with patch.dict(
+        "sys.modules",
+        {
+            "hermes_cli.env_loader": MagicMock(),
+            "hermes_cli.banner": MagicMock(),
+        },
+    ):
+        mod = importlib.import_module("tui_gateway.server")
+        yield mod
+        mod._sessions.clear()
+        mod._pending.clear()
+        mod._answers.clear()
+        mod._methods.clear()
+        importlib.reload(mod)
+
+
+@pytest.fixture()
+def db(hermes_home):
+    return SessionDB(db_path=hermes_home / "state.db")
+
+
+@pytest.fixture()
+def session_with_history(server, db):
+    """Build a session with 3 user turns + assistant replies persisted in DB."""
+    sid = "sid-rewind"
+    session_key = "tui-rewind-1"
+    db.create_session(session_key, source="tui")
+    for i in range(1, 4):
+        db.append_message(session_key, "user", f"question {i}")
+        db.append_message(session_key, "assistant", f"answer {i}")
+    history = db.get_messages_as_conversation(session_key)
+    agent = MagicMock()
+    agent._memory_manager = MagicMock()
+    agent._last_flushed_db_idx = len(history)
+    s = {
+        "session_key": session_key,
+        "history": list(history),
+        "history_lock": threading.Lock(),
+        "history_version": 0,
+        "running": False,
+        "agent": agent,
+        "attached_images": [],
+        "cols": 120,
+    }
+    server._sessions[sid] = s
+    # Wire the DB cache so _get_db() returns our fixture.
+    server._db = db
+    return sid, session_key, s, agent
+
+
+def _call(server, method, **params):
+    return server._methods[method](1, params)
+
+
+def test_rewind_returns_prefill_with_target_text(server, session_with_history):
+    sid, session_key, s, agent = session_with_history
+    resp = _call(server, "command.dispatch", session_id=sid, name="rewind", arg="")
+    result = resp["result"]
+    assert result["type"] == "prefill"
+    # v1 auto-picks the most recent user turn — "question 3"
+    assert result["message"] == "question 3"
+    assert "Rewound" in result["notice"]
+
+
+def test_rewind_truncates_in_memory_history(server, session_with_history, db):
+    sid, session_key, s, agent = session_with_history
+    _call(server, "command.dispatch", session_id=sid, name="rewind", arg="")
+    # After rewinding to "question 3", active history should be 4 rows:
+    # user q1, asst a1, user q2, asst a2
+    assert len(s["history"]) == 4
+    roles = [m["role"] for m in s["history"]]
+    assert roles == ["user", "assistant", "user", "assistant"]
+    # version bumped
+    assert s["history_version"] == 1
+
+
+def test_rewind_soft_deletes_rows_in_db(server, session_with_history, db):
+    sid, session_key, _, _ = session_with_history
+    _call(server, "command.dispatch", session_id=sid, name="rewind", arg="")
+    # All rows still present
+    all_rows = db.get_messages(session_key, include_inactive=True)
+    assert len(all_rows) == 6
+    # 2 inactive (the "question 3" row + its trailing siblings — here just
+    # "question 3" + "answer 3", since target was the q3 user row).
+    active = [r for r in all_rows if r["active"] == 1]
+    assert len(active) == 4
+    # rewind_count bumped
+    sess = db.get_session(session_key)
+    assert sess["rewind_count"] == 1
+
+
+def test_rewind_notifies_memory_provider(server, session_with_history):
+    sid, session_key, _, agent = session_with_history
+    _call(server, "command.dispatch", session_id=sid, name="rewind", arg="")
+    agent._memory_manager.on_session_switch.assert_called_once()
+    args, kwargs = agent._memory_manager.on_session_switch.call_args
+    assert args[0] == session_key
+    assert kwargs["rewound"] is True
+    assert kwargs["reset"] is False
+
+
+def test_rewind_refuses_when_session_busy(server, session_with_history):
+    sid, _, s, _ = session_with_history
+    s["running"] = True
+    resp = _call(server, "command.dispatch", session_id=sid, name="rewind", arg="")
+    assert "error" in resp
+    assert "busy" in resp["error"]["message"].lower()
+
+
+def test_rewind_errors_when_no_active_session(server):
+    resp = _call(server, "command.dispatch", session_id="no-such-sid", name="rewind", arg="")
+    assert "error" in resp
+    assert "no active session" in resp["error"]["message"].lower()
+
+
+def test_rewind_in_pending_input_commands(server):
+    """Registry sanity: /rewind must be in _PENDING_INPUT_COMMANDS so
+    slash.exec rejects it and the TUI falls through to command.dispatch."""
+    assert "rewind" in server._PENDING_INPUT_COMMANDS
diff --git a/tui_gateway/server.py b/tui_gateway/server.py
index de2888a6de74..619a8b43c529 100644
--- a/tui_gateway/server.py
+++ b/tui_gateway/server.py
@@ -4406,6 +4406,7 @@ def _(rid, params: dict) -> dict:
         "steer",
         "plan",
         "goal",
+        "rewind",
     }
 )
 
@@ -4791,6 +4792,94 @@ def _(rid, params: dict) -> dict:
             {"type": "send", "notice": notice, "message": state.goal},
         )
 
+    if name == "rewind":
+        # /rewind: pick the most-recent user message and prefill the
+        # composer with its text after soft-deleting everything that
+        # came after it in the transcript. v1 auto-picks the latest
+        # user turn (Claude-Code-style single-step undo); a multi-step
+        # picker UI is tracked as a follow-up to issue #21910.
+        if not session:
+            return _err(rid, 4001, "no active session to rewind")
+        if session.get("running"):
+            return _err(
+                rid, 4009, "session busy — /interrupt the current turn before /rewind"
+            )
+        db = _get_db()
+        if db is None:
+            return _db_unavailable_error(rid, code=5008)
+        session_key = session.get("session_key", "")
+        if not session_key:
+            return _err(rid, 4001, "no session key for rewind")
+        try:
+            recents = db.list_recent_user_messages(session_key, limit=10)
+        except Exception as e:
+            return _err(rid, 5008, f"rewind: failed to load history: {e}")
+        if not recents:
+            return _err(rid, 4018, "no user messages to rewind to")
+        # v1: auto-pick the most recent user turn. The Ink UI does not
+        # yet host a dedicated picker overlay (#21910 follow-up).
+        target_id = recents[0]["id"]
+        try:
+            result = db.rewind_to_message(session_key, target_id)
+        except ValueError as e:
+            return _err(rid, 4004, f"rewind: {e}")
+        except Exception as e:
+            return _err(rid, 5008, f"rewind: {e}")
+        # Reload the active-only transcript into the in-memory session
+        # history so subsequent turns see the truncated view.
+        try:
+            active = db.get_messages_as_conversation(session_key)
+        except Exception:
+            active = []
+        with session["history_lock"]:
+            session["history"] = list(active)
+            session["history_version"] = int(session.get("history_version", 0)) + 1
+        # Notify memory providers — same hook /branch fires, plus the
+        # rewound flag so providers caching per-turn document state
+        # know to invalidate. See #6672 + #21910.
+        agent = session.get("agent")
+        if agent is not None:
+            mm = getattr(agent, "_memory_manager", None)
+            if mm is not None:
+                try:
+                    mm.on_session_switch(
+                        session_key,
+                        parent_session_id="",
+                        reset=False,
+                        rewound=True,
+                    )
+                except Exception:
+                    pass
+            if hasattr(agent, "_invalidate_system_prompt"):
+                try:
+                    agent._invalidate_system_prompt()
+                except Exception:
+                    pass
+            if hasattr(agent, "_last_flushed_db_idx"):
+                try:
+                    agent._last_flushed_db_idx = len(active)
+                except Exception:
+                    pass
+        target_msg = result.get("target_message") or {}
+        target_text = target_msg.get("content") or ""
+        if isinstance(target_text, list):
+            parts = [
+                p.get("text", "") for p in target_text
+                if isinstance(p, dict) and p.get("type") == "text"
+            ]
+            target_text = "\n".join(t for t in parts if t)
+        if not isinstance(target_text, str):
+            target_text = ""
+        rewound_count = result.get("rewound_count", 0)
+        notice = (
+            f"↶ Rewound {rewound_count} message(s). "
+            "Edit and resubmit, or send a new message."
+        )
+        return _ok(
+            rid,
+            {"type": "prefill", "message": target_text, "notice": notice},
+        )
+
     if name in {"snapshot", "snap"}:
         subcommand = arg.split(maxsplit=1)[0].lower() if arg else ""
         if subcommand in {"restore", "rewind"}:
diff --git a/ui-tui/src/__tests__/asCommandDispatch.test.ts b/ui-tui/src/__tests__/asCommandDispatch.test.ts
index dfa7595174e1..5dac25fab7bd 100644
--- a/ui-tui/src/__tests__/asCommandDispatch.test.ts
+++ b/ui-tui/src/__tests__/asCommandDispatch.test.ts
@@ -15,6 +15,15 @@ describe('asCommandDispatch', () => {
       type: 'send',
       message: 'hello world'
     })
+    expect(asCommandDispatch({ type: 'prefill', message: 'edit me' })).toEqual({
+      type: 'prefill',
+      message: 'edit me'
+    })
+    expect(asCommandDispatch({ type: 'prefill', message: 'edit me', notice: '↶ rewound' })).toEqual({
+      type: 'prefill',
+      message: 'edit me',
+      notice: '↶ rewound'
+    })
   })
 
   it('rejects malformed payloads', () => {
@@ -23,5 +32,7 @@ describe('asCommandDispatch', () => {
     expect(asCommandDispatch({ type: 'skill', name: 1 })).toBeNull()
     expect(asCommandDispatch({ type: 'send' })).toBeNull()
     expect(asCommandDispatch({ type: 'send', message: 42 })).toBeNull()
+    expect(asCommandDispatch({ type: 'prefill' })).toBeNull()
+    expect(asCommandDispatch({ type: 'prefill', message: 42 })).toBeNull()
   })
 })
diff --git a/ui-tui/src/app/createSlashHandler.ts b/ui-tui/src/app/createSlashHandler.ts
index 0164ef0d568d..71e2536d89f2 100644
--- a/ui-tui/src/app/createSlashHandler.ts
+++ b/ui-tui/src/app/createSlashHandler.ts
@@ -119,6 +119,19 @@ export function createSlashHandler(ctx: SlashHandlerContext): (cmd: string) => b
               }
               return d.message?.trim() ? send(d.message) : sys(`/${parsed.name}: empty message`)
             }
+
+            if (d.type === 'prefill') {
+              // /rewind returns prefill: drop the chosen text into the
+              // composer so the user can edit and resubmit, instead of
+              // submitting it immediately like 'send'.
+              if (d.notice?.trim()) {
+                sys(d.notice)
+              }
+              if (d.message) {
+                ctx.composer.setInput(d.message)
+              }
+              return
+            }
           })
           .catch(guardedErr)
       })
diff --git a/ui-tui/src/gatewayTypes.ts b/ui-tui/src/gatewayTypes.ts
index ab85c39fbddf..85fdf6178432 100644
--- a/ui-tui/src/gatewayTypes.ts
+++ b/ui-tui/src/gatewayTypes.ts
@@ -48,6 +48,7 @@ export type CommandDispatchResponse =
   | { target: string; type: 'alias' }
   | { message?: string; name: string; type: 'skill' }
   | { message: string; notice?: string; type: 'send' }
+  | { message: string; notice?: string; type: 'prefill' }
 
 // ── Config ───────────────────────────────────────────────────────────
 
diff --git a/ui-tui/src/lib/rpc.ts b/ui-tui/src/lib/rpc.ts
index 81dc70318646..76862f073666 100644
--- a/ui-tui/src/lib/rpc.ts
+++ b/ui-tui/src/lib/rpc.ts
@@ -34,6 +34,14 @@ export const asCommandDispatch = (value: unknown): CommandDispatchResponse | nul
     }
   }
 
+  if (t === 'prefill' && typeof o.message === 'string') {
+    return {
+      type: 'prefill',
+      message: o.message,
+      notice: typeof o.notice === 'string' ? o.notice : undefined,
+    }
+  }
+
   return null
 }