diff --git a/cli.py b/cli.py
index 3b1ecd8ae4..a5f4bbbd82 100644
--- a/cli.py
+++ b/cli.py
@@ -6664,6 +6664,18 @@ class HermesCLI:
                 focus_topic=focus_topic or None,
             )
             self.conversation_history = compressed
+            # _compress_context ends the old session and creates a new child
+            # session on the agent (run_agent.py::_compress_context). Sync the
+            # CLI's session_id so /status, /resume, exit summary, and title
+            # generation all point at the live continuation session, not the
+            # ended parent. Without this, subsequent end_session() calls target
+            # the already-closed parent and the child is orphaned.
+            if (
+                getattr(self.agent, "session_id", None)
+                and self.agent.session_id != self.session_id
+            ):
+                self.session_id = self.agent.session_id
+                self._pending_title = None
             new_tokens = estimate_messages_tokens_rough(self.conversation_history)
             summary = summarize_manual_compression(
                 original_history,
@@ -8182,6 +8194,20 @@ class HermesCLI:
             # Update history with full conversation
             self.conversation_history = result.get("messages", self.conversation_history) if result else self.conversation_history
 
+            # If auto-compression fired mid-turn, the agent created a new
+            # continuation session and mutated self.agent.session_id. Sync
+            # the CLI's session_id so /status, /resume, title generation,
+            # and the exit summary all target the live child session rather
+            # than the ended parent. Mirrors the gateway's post-run sync
+            # (gateway/run.py around line 9983).
+            if (
+                self.agent
+                and getattr(self.agent, "session_id", None)
+                and self.agent.session_id != self.session_id
+            ):
+                self.session_id = self.agent.session_id
+                self._pending_title = None
+
             # Get the final response
             response = result.get("final_response", "") if result else ""
 
@@ -10554,6 +10580,15 @@ def main(
                         user_message=effective_query,
                         conversation_history=cli.conversation_history,
                     )
+                    # Sync session_id if mid-run compression created a
+                    # continuation session. The exit line below reports
+                    # session_id to stderr for automation wrappers; without
+                    # this sync it would point at the ended parent.
+                    if (
+                        getattr(cli.agent, "session_id", None)
+                        and cli.agent.session_id != cli.session_id
+                    ):
+                        cli.session_id = cli.agent.session_id
                     response = result.get("final_response", "") if isinstance(result, dict) else str(result)
                     if response:
                         print(response)
diff --git a/hermes_state.py b/hermes_state.py
index af97f7fbd8..d692a51688 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -383,10 +383,19 @@ class SessionDB:
         return session_id
 
     def end_session(self, session_id: str, end_reason: str) -> None:
-        """Mark a session as ended."""
+        """Mark a session as ended.
+
+        No-ops when the session is already ended. The first end_reason wins:
+        compression-split sessions must keep their ``end_reason = 'compression'``
+        record even if a later stale ``end_session()`` call (e.g. from a
+        desynced CLI session_id after ``/resume`` or ``/branch``) targets them
+        with a different reason. Use ``reopen_session()`` first if you
+        intentionally need to re-end a closed session with a new reason.
+        """
         def _do(conn):
             conn.execute(
-                "UPDATE sessions SET ended_at = ?, end_reason = ? WHERE id = ?",
+                "UPDATE sessions SET ended_at = ?, end_reason = ? "
+                "WHERE id = ? AND ended_at IS NULL",
                 (time.time(), end_reason, session_id),
             )
         self._execute_write(_do)
diff --git a/tests/cli/test_manual_compress.py b/tests/cli/test_manual_compress.py
index d201f9cee5..9144c94b10 100644
--- a/tests/cli/test_manual_compress.py
+++ b/tests/cli/test_manual_compress.py
@@ -21,6 +21,7 @@ def test_manual_compress_reports_noop_without_success_banner(capsys):
     shell.agent = MagicMock()
     shell.agent.compression_enabled = True
     shell.agent._cached_system_prompt = ""
+    shell.agent.session_id = shell.session_id  # no-op compression: no split
     shell.agent._compress_context.return_value = (list(history), "")
 
     def _estimate(messages):
@@ -48,6 +49,7 @@ def test_manual_compress_explains_when_token_estimate_rises(capsys):
     shell.agent = MagicMock()
     shell.agent.compression_enabled = True
     shell.agent._cached_system_prompt = ""
+    shell.agent.session_id = shell.session_id  # no-op: no split
     shell.agent._compress_context.return_value = (compressed, "")
 
     def _estimate(messages):
@@ -64,3 +66,64 @@ def test_manual_compress_explains_when_token_estimate_rises(capsys):
     assert "✅ Compressed: 4 → 3 messages" in output
     assert "Rough transcript estimate: ~100 → ~120 tokens" in output
     assert "denser summaries" in output
+
+
+def test_manual_compress_syncs_session_id_after_split():
+    """Regression for cli.session_id desync after /compress.
+
+    _compress_context ends the parent session and creates a new child session,
+    mutating agent.session_id. Without syncing, cli.session_id still points
+    at the ended parent — causing /status, /resume, exit summary, and the
+    next end_session() call (e.g. from /resume <id>) to target the wrong row.
+    """
+    shell = _make_cli()
+    history = _make_history()
+    old_id = shell.session_id
+    new_child_id = "20260101_000000_child1"
+
+    compressed = [
+        {"role": "user", "content": "[summary]"},
+        history[-1],
+    ]
+    shell.conversation_history = history
+    shell.agent = MagicMock()
+    shell.agent.compression_enabled = True
+    shell.agent._cached_system_prompt = ""
+    # Simulate _compress_context mutating agent.session_id as a side effect.
+    def _fake_compress(*args, **kwargs):
+        shell.agent.session_id = new_child_id
+        return (compressed, "")
+    shell.agent._compress_context.side_effect = _fake_compress
+    shell.agent.session_id = old_id  # starts in sync
+    shell._pending_title = "stale title"
+
+    with patch("agent.model_metadata.estimate_messages_tokens_rough", return_value=100):
+        shell._manual_compress()
+
+    # CLI session_id must now point at the continuation child, not the parent.
+    assert shell.session_id == new_child_id
+    assert shell.session_id != old_id
+    # Pending title must be cleared — titles belong to the parent lineage and
+    # get regenerated for the continuation.
+    assert shell._pending_title is None
+
+
+def test_manual_compress_no_sync_when_session_id_unchanged():
+    """If compression is a no-op (agent.session_id didn't change), the CLI
+    must NOT clear _pending_title or otherwise disturb session state.
+    """
+    shell = _make_cli()
+    history = _make_history()
+    shell.conversation_history = history
+    shell.agent = MagicMock()
+    shell.agent.compression_enabled = True
+    shell.agent._cached_system_prompt = ""
+    shell.agent.session_id = shell.session_id
+    shell.agent._compress_context.return_value = (list(history), "")
+    shell._pending_title = "keep me"
+
+    with patch("agent.model_metadata.estimate_messages_tokens_rough", return_value=100):
+        shell._manual_compress()
+
+    # No split → pending title untouched.
+    assert shell._pending_title == "keep me"
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index d54d7b9fb0..bc1f7d7cbd 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -46,6 +46,37 @@ class TestSessionLifecycle:
         assert isinstance(session["ended_at"], float)
         assert session["end_reason"] == "user_exit"
 
+    def test_end_session_preserves_original_end_reason(self, db):
+        """The first end_reason wins — compression splits must not be
+        overwritten when a later stale ``end_session()`` call lands on the
+        same row (e.g. from a CLI session_id that desynced after compression
+        and then tried to /resume another session).
+        """
+        db.create_session(session_id="s1", source="cli")
+        db.end_session("s1", end_reason="compression")
+        first_ended_at = db.get_session("s1")["ended_at"]
+
+        # Simulate a stale CLI holding the old session_id and calling
+        # end_session() again with a different reason.
+        time.sleep(0.01)
+        db.end_session("s1", end_reason="resumed_other")
+
+        session = db.get_session("s1")
+        assert session["end_reason"] == "compression"
+        assert session["ended_at"] == first_ended_at
+
+    def test_end_session_after_reopen_allows_re_end(self, db):
+        """reopen_session() is the explicit escape hatch for re-ending a
+        closed session. After reopen, end_session() takes effect again.
+        """
+        db.create_session(session_id="s1", source="cli")
+        db.end_session("s1", end_reason="compression")
+        db.reopen_session("s1")
+        db.end_session("s1", end_reason="user_exit")
+
+        session = db.get_session("s1")
+        assert session["end_reason"] == "user_exit"
+
     def test_update_system_prompt(self, db):
         db.create_session(session_id="s1", source="cli")
         db.update_system_prompt("s1", "You are a helpful assistant.")