diff --git a/hermes_cli/main.py b/hermes_cli/main.py
index 33a751731b..28d3009325 100644
--- a/hermes_cli/main.py
+++ b/hermes_cli/main.py
@@ -12179,6 +12179,12 @@ def main():
         default=False,
         help="Report the affected row count without writing",
     )
+    sessions_clean_markers.add_argument(
+        "--no-backup",
+        action="store_true",
+        default=False,
+        help="Skip the timestamped state.db backup taken before writing (not recommended)",
+    )
 
     sessions_optimize_storage = sessions_subparsers.add_parser(
         "optimize-storage",
diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py
index b8a06e3021..cf317b610c 100644
--- a/hermes_cli/sessions_cmd.py
+++ b/hermes_cli/sessions_cmd.py
@@ -1044,7 +1044,9 @@ def cmd_sessions(args, sessions_parser=None):
             print("Dry run — scanning for stale tool-call marker rows (#78148)…")
         else:
             print("Scanning for stale tool-call marker rows (#78148)…")
-        report = db.purge_stale_tool_call_markers(dry_run=args.dry_run)
+        report = db.purge_stale_tool_call_markers(
+            dry_run=args.dry_run, backup=not args.no_backup
+        )
         if report["rows_affected"] == 0:
             print("✓ No affected rows found — nothing to clean.")
         elif args.dry_run:
@@ -1053,6 +1055,8 @@ def cmd_sessions(args, sessions_parser=None):
                 f"ids {report['row_ids']}"
             )
         else:
+            if report["backup_path"]:
+                print(f"  backup: {report['backup_path']}")
             print(f"✓ Cleared {report['rows_affected']} row(s).")
 
     elif action == "optimize-storage":
diff --git a/hermes_state.py b/hermes_state.py
index c0cba359ec..3810dbcbd7 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -8443,7 +8443,9 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             self._remove_session_files(sessions_dir, sid)
         return count
 
-    def purge_stale_tool_call_markers(self, *, dry_run: bool = False) -> Dict[str, Any]:
+    def purge_stale_tool_call_markers(
+        self, *, dry_run: bool = False, backup: bool = True
+    ) -> Dict[str, Any]:
         """Permanently clear bare tool-call marker content (e.g. "[memory]")
         left in the ``messages`` table by sessions persisted before the
         #78148 fix in ``agent.conversation_loop``.
@@ -8460,10 +8462,20 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         and every other column on the row are left exactly as they are, so
         provider tool_call/tool_result pairing is unaffected.
 
-        With ``dry_run=True``, reports the affected row count/ids without
-        writing (read-only, no write lock taken).
+        Unlike the in-memory repair, this UPDATE is permanent and can't be
+        undone from within the DB. Since ``backup`` defaults to True, a
+        timestamped full snapshot is taken via ``VACUUM INTO`` (safe against
+        a live connection, unlike the raw-copy ``_backup_db_file`` used for
+        malformed-schema repair) before any row is touched — mirroring
+        ``repair_state_db_schema``'s backup-by-default convention for
+        destructive state.db operations. No snapshot is taken when there is
+        nothing to change.
 
-        Returns ``{"dry_run": bool, "rows_affected": int, "row_ids": [...]}``.
+        With ``dry_run=True``, reports the affected row count/ids without
+        writing or backing up (read-only, no write lock taken).
+
+        Returns ``{"dry_run": bool, "rows_affected": int, "row_ids": [...],
+        "backup_path": str|None}``.
         """
 
         def _find_affected(conn) -> List[int]:
@@ -8478,20 +8490,47 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                     affected.append(row["id"])
             return affected
 
+        with self._read_ctx() as conn:
+            affected_ids = _find_affected(conn)
+
         if dry_run:
-            with self._read_ctx() as conn:
-                affected_ids = _find_affected(conn)
-            return {"dry_run": True, "rows_affected": len(affected_ids), "row_ids": affected_ids}
+            return {
+                "dry_run": True,
+                "rows_affected": len(affected_ids),
+                "row_ids": affected_ids,
+                "backup_path": None,
+            }
+
+        if not affected_ids:
+            return {
+                "dry_run": False,
+                "rows_affected": 0,
+                "row_ids": [],
+                "backup_path": None,
+            }
+
+        backup_path: Optional[str] = None
+        if backup:
+            import datetime
+
+            stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+            dest = self.db_path.with_name(
+                f"{self.db_path.name}.pre-clean-markers-backup-{stamp}"
+            )
+            with self._lock:
+                self._conn.execute("VACUUM INTO ?", (str(dest),))
+            backup_path = str(dest)
+            logger.info("Backed up state.db to %s before clean-markers write", backup_path)
 
         def _do(conn):
-            affected_ids = _find_affected(conn)
-            if affected_ids:
-                placeholders = ",".join("?" * len(affected_ids))
+            ids = _find_affected(conn)
+            if ids:
+                placeholders = ",".join("?" * len(ids))
                 conn.execute(
                     f"UPDATE messages SET content = '' WHERE id IN ({placeholders})",
-                    affected_ids,
+                    ids,
                 )
-            return affected_ids
+            return ids
 
         affected_ids = self._execute_write(_do)
         if affected_ids:
@@ -8499,7 +8538,12 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 "Permanently cleared %d stale tool-call marker row(s) in state.db (#78148)",
                 len(affected_ids),
             )
-        return {"dry_run": False, "rows_affected": len(affected_ids), "row_ids": affected_ids}
+        return {
+            "dry_run": False,
+            "rows_affected": len(affected_ids),
+            "row_ids": affected_ids,
+            "backup_path": backup_path,
+        }
 
     # ── Meta key/value (for scheduler bookkeeping) ──
 
diff --git a/tests/test_stale_tool_call_marker_session_repair.py b/tests/test_stale_tool_call_marker_session_repair.py
index 41b5d8f12e..ed27fe4351 100644
--- a/tests/test_stale_tool_call_marker_session_repair.py
+++ b/tests/test_stale_tool_call_marker_session_repair.py
@@ -188,6 +188,9 @@ class TestPurgeStaleToolCallMarkers:
                 report = db.purge_stale_tool_call_markers(dry_run=False)
                 assert report["dry_run"] is False
                 assert report["rows_affected"] == 1
+                # Backup defaults to on for a destructive, irreversible write.
+                assert report["backup_path"] is not None
+                assert Path(report["backup_path"]).exists()
 
                 row = db._conn.execute(
                     "SELECT content, tool_calls FROM messages WHERE role = 'assistant' "
@@ -200,6 +203,43 @@ class TestPurgeStaleToolCallMarkers:
                 # Running again finds nothing left to clean — idempotent.
                 second = db.purge_stale_tool_call_markers(dry_run=False)
                 assert second["rows_affected"] == 0
+                assert second["backup_path"] is None  # nothing to change, nothing to back up
+            finally:
+                db.close()
+
+    def test_no_backup_when_flag_false(self):
+        import tempfile
+        from pathlib import Path
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmp:
+            db = SessionDB(db_path=Path(tmp) / "t.db")
+            try:
+                self._seed_polluted_db(db)
+
+                report = db.purge_stale_tool_call_markers(dry_run=False, backup=False)
+                assert report["rows_affected"] == 1
+                assert report["backup_path"] is None
+                # No extra file created beside the DB.
+                siblings = list(Path(tmp).glob("t.db.*backup*"))
+                assert siblings == []
+            finally:
+                db.close()
+
+    def test_dry_run_never_backs_up(self):
+        import tempfile
+        from pathlib import Path
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmp:
+            db = SessionDB(db_path=Path(tmp) / "t.db")
+            try:
+                self._seed_polluted_db(db)
+
+                report = db.purge_stale_tool_call_markers(dry_run=True)
+                assert report["backup_path"] is None
+                siblings = list(Path(tmp).glob("t.db.*backup*"))
+                assert siblings == []
             finally:
                 db.close()