diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py
index ba3a151eba..2c96dd9f33 100644
--- a/gateway/platforms/api_server.py
+++ b/gateway/platforms/api_server.py
@@ -3216,7 +3216,12 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter):
             sessions_dir = Path(get_hermes_home()) / "sessions"
         except Exception:
             logger.debug("sessions dir unavailable for delete of %s", session_id, exc_info=True)
-        deleted = await asyncio.to_thread(db.delete_session, session_id, sessions_dir=sessions_dir)
+        from hermes_state_errors import SessionActiveWriteGuardError
+        try:
+            deleted = await asyncio.to_thread(
+                db.delete_session, session_id, sessions_dir=sessions_dir, exclude_active_write_guards=True)
+        except SessionActiveWriteGuardError as exc:
+            return _error_response(str(exc), 409, code="session_active_turn")
         if deleted:
             # A hard delete must also drop the gateway's durable channel→session routing entries
             # for the id, or the next inbound message resolves the SAME id and run_agent's
diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py
index 9c4627a3cf..6348b69ee3 100644
--- a/hermes_cli/sessions_cmd.py
+++ b/hermes_cli/sessions_cmd.py
@@ -562,12 +562,17 @@ def _export_markdown_single(db, args, export_one, output_dir, lineage_is_logical
             print(f"Export verification failed; not deleting session '{data.get('id')}': {reason}")
             return
         expected_messages.update(snapshots)
-    if not db.delete_session(
-        resolved_session_id, sessions_dir=_sessions_dir(), expected_delete_ids=delete_target_ids,
-        expected_display_messages=expected_messages,
-    ):
-        print(f"Exported, but session '{resolved_session_id}' was not deleted because its history or delegate set "
-              "changed after export.")
+    from hermes_state_errors import SessionActiveWriteGuardError
+    try:
+        if not db.delete_session(
+            resolved_session_id, sessions_dir=_sessions_dir(), expected_delete_ids=delete_target_ids,
+            expected_display_messages=expected_messages, exclude_active_write_guards=True,
+        ):
+            print(f"Exported, but session '{resolved_session_id}' was not deleted because its history or delegate set "
+                  "changed after export.")
+            return
+    except SessionActiveWriteGuardError as exc:
+        print(f"Exported, but session '{resolved_session_id}' was not deleted because it is active: {exc}")
         return
     delegates = len(delete_target_ids) - 1
     delegate_suffix = f" and {delegates} delegate session{'' if delegates == 1 else 's'}" if delegates else ""
@@ -588,8 +593,13 @@ def _cmd_delete(db, args):
             return
     elif _pinned_note:
         print(f"Warning: deleting a pinned session '{resolved_session_id}'.")
-    if not db.delete_session(resolved_session_id, sessions_dir=_sessions_dir()):
-        return _not_found(args.session_id)
+    from hermes_state_errors import SessionActiveWriteGuardError
+    try:
+        if not db.delete_session(resolved_session_id, sessions_dir=_sessions_dir(), exclude_active_write_guards=True):
+            return _not_found(args.session_id)
+    except SessionActiveWriteGuardError as exc:
+        print(f"Cannot delete active session: {exc}")
+        return 1
     print(f"Deleted session '{resolved_session_id}'.")
 
 
@@ -722,7 +732,7 @@ def _cmd_prune_or_archive(db, args, action):
         print("Cancelled.")
         return
     if prune:
-        print(f"Pruned {db.prune_sessions(sessions_dir=_sessions_dir(), **filters)} session(s).")
+        print(f"Pruned {db.prune_sessions(sessions_dir=_sessions_dir(), exclude_active_write_guards=True, **filters)} session(s).")
     else:
         print(f"Archived {db.archive_sessions(**filters)} session(s). They're hidden from listings "
               "but fully recoverable (nothing was deleted).")
diff --git a/hermes_cli/sessions_cmd_browse.py b/hermes_cli/sessions_cmd_browse.py
index 2f1311a577..1508abdac4 100644
--- a/hermes_cli/sessions_cmd_browse.py
+++ b/hermes_cli/sessions_cmd_browse.py
@@ -254,7 +254,7 @@ def _session_browse_picker(sessions: list, session_db=None) -> Optional[str]:
         except Exception:
             sessions_dir = None
         try:
-            return bool(session_db.delete_session(session_id, sessions_dir=sessions_dir))
+            return bool(session_db.delete_session(session_id, sessions_dir=sessions_dir, exclude_active_write_guards=True))
         except Exception:
             return False
     try:  # curses first; any failure (no curses module, odd terminal) falls back
diff --git a/hermes_cli/web_routers/sessions.py b/hermes_cli/web_routers/sessions.py
index 8fa5919809..f5e69b4176 100644
--- a/hermes_cli/web_routers/sessions.py
+++ b/hermes_cli/web_routers/sessions.py
@@ -28,7 +28,7 @@ from hermes_cli.web_routers._common import (
     CORRUPT_STORE_DETAIL, corrupt_store_as_status, log as _log, destructive_profile, http_failure,
 )
 from hermes_state import is_malformed_db_error
-from hermes_state_errors import StateDbReplacedError, is_transient_sqlite_error
+from hermes_state_errors import SessionActiveWriteGuardError, StateDbReplacedError, is_transient_sqlite_error
 from hermes_state_health import STORAGE_CORRUPT, note_storage_error, storage_state
 
 list_router = APIRouter()
@@ -114,7 +114,8 @@ def _prune_sessions(body: SessionPrune):
                 "sessions": [{k: r.get(k) for k in _PRUNE_ROW_KEYS} for r in rows]}
         sessions_dir = profile_home / "sessions"
         removed = db.prune_sessions(
-            sessions_dir=sessions_dir if sessions_dir.exists() else None, **filters)
+            sessions_dir=sessions_dir if sessions_dir.exists() else None,
+            exclude_active_write_guards=True, **filters)
         return {"ok": True, "removed": removed, "skipped_open": skipped_open}
     finally:
         db.close()
@@ -440,7 +441,7 @@ async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions):
         raise HTTPException(status_code=400, detail="ids must contain at most 500 entries")
     profile = destructive_profile(body.profile, "POST /api/sessions/bulk-delete")
     deleted = await asyncio.to_thread(
-        _with_db, profile, lambda db: db.delete_sessions(body.ids), read_only=False)
+        _with_db, profile, lambda db: db.delete_sessions(body.ids, exclude_active_write_guards=True), read_only=False)
     return {"ok": True, "deleted": deleted}
 
 
@@ -725,7 +726,10 @@ async def delete_session_endpoint(session_id: str, profile: Optional[str] = None
         sid = _resolve_session_id(db, session_id)
         if not sid:
             return {"ok": True, "already_absent": True}
-        db.delete_session(sid, sessions_dir=_session_files_dir(profile))
+        try:
+            db.delete_session(sid, sessions_dir=_session_files_dir(profile), exclude_active_write_guards=True)
+        except SessionActiveWriteGuardError as exc:
+            raise HTTPException(status_code=409, detail=str(exc))
         return {"ok": True}
 
     return await asyncio.to_thread(_with_db, profile, _delete, read_only=False)
diff --git a/hermes_state_errors.py b/hermes_state_errors.py
index 22d2c9a95c..ec098af359 100644
--- a/hermes_state_errors.py
+++ b/hermes_state_errors.py
@@ -172,6 +172,10 @@ class SessionTurnLeaseLostError(RuntimeError):
     be persisting a newer turn, and landing this one would interleave a stale reply."""
 
 
+class SessionActiveWriteGuardError(RuntimeError):
+    """Raised when an active turn lease or compression lock rejects session deletion."""
+
+
 class StateDbReplacedError(RuntimeError):
     """The state.db path no longer names the file this SessionDB opened
     (out-of-band cp/mv/restore). In-place FTS repair and fail-open trigger
diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py
index d6166a6810..3714579185 100644
--- a/hermes_state_sessions.py
+++ b/hermes_state_sessions.py
@@ -1563,15 +1563,23 @@ class SessionSessionsMixin:
         self, session_id: str, sessions_dir: Optional[Path] = None,
         expected_delete_ids: Optional[List[str]] = None,
         expected_display_messages: Optional[Dict[str, List[Dict[str, Any]]]] = None,
+        exclude_active_write_guards: bool = False,
     ) -> bool:
         """Delete a session and its messages; delegate children cascade, branch/compression children
         are orphaned. Optional expected ids fence delegate drift; expected display snapshots fence
-        transcript drift. Both checks run inside the same write transaction as deletion."""
+        transcript drift. Both checks run inside the same write transaction as deletion.
+        With ``exclude_active_write_guards``, raises :class:`SessionActiveWriteGuardError` if the row
+        is protected by an active turn lease or compression lock."""
+        from hermes_state_errors import SessionActiveWriteGuardError
         removed_ids: List[str] = []
         expected_ids = set(expected_delete_ids) if expected_delete_ids is not None else None
         def _do(conn):
             if conn.execute("SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,)).fetchone() is None:
                 return False
+            if exclude_active_write_guards and self._write_guards_reject(conn, session_id):
+                raise SessionActiveWriteGuardError(
+                    f"session '{session_id}' has an active turn lease or compression lock"
+                )
             if expected_ids is not None and expected_ids != {
                 session_id, *_collect_delegate_child_ids(conn, [session_id])
             }:
@@ -1622,9 +1630,13 @@ class SessionSessionsMixin:
             self._remove_session_files(sessions_dir, session_id)
         return deleted
 
-    def delete_sessions(self, session_ids: List[str], sessions_dir: Optional[Path] = None) -> int:
+    def delete_sessions(
+        self, session_ids: List[str], sessions_dir: Optional[Path] = None,
+        exclude_active_write_guards: bool = False,
+    ) -> int:
         """Bulk delete with :meth:`delete_session` semantics per row, in ONE transaction. Unknown ids
-        are skipped (UI selection can race another tab's delete). Returns the number deleted."""
+        are skipped (UI selection can race another tab's delete). With ``exclude_active_write_guards``,
+        rows protected by an active turn lease or compression lock are skipped. Returns the number deleted."""
         unique_ids = list({sid for sid in session_ids or () if isinstance(sid, str) and sid})
         if not unique_ids:
             return 0
@@ -1635,6 +1647,11 @@ class SessionSessionsMixin:
             ).fetchall()]
             if not existing:
                 return 0
+            if exclude_active_write_guards:
+                active_ids = {sid for sid in existing if self._write_guards_reject(conn, sid)}
+                existing = [sid for sid in existing if sid not in active_ids]
+                if not existing:
+                    return 0
             removed_ids.extend(_delete_delegate_children(conn, existing))
             for chunk in _id_chunks(existing):
                 ph = _session_ids_placeholders(chunk)
diff --git a/tests/hermes_cli/test_web_routers_destructive_profile_scope.py b/tests/hermes_cli/test_web_routers_destructive_profile_scope.py
index f862eca90e..2220456ab1 100644
--- a/tests/hermes_cli/test_web_routers_destructive_profile_scope.py
+++ b/tests/hermes_cli/test_web_routers_destructive_profile_scope.py
@@ -93,7 +93,7 @@ def client(monkeypatch, homes):
 class _StubDB:
     """Just enough session store for the delete/prune bodies; records nothing itself."""
 
-    def delete_sessions(self, ids):
+    def delete_sessions(self, ids, **_kwargs):
         return len(ids)
 
     def delete_empty_sessions(self):
diff --git a/tests/hermes_cli/test_web_server_sessiondb_eventloop.py b/tests/hermes_cli/test_web_server_sessiondb_eventloop.py
index ee0060a279..c45f18554f 100644
--- a/tests/hermes_cli/test_web_server_sessiondb_eventloop.py
+++ b/tests/hermes_cli/test_web_server_sessiondb_eventloop.py
@@ -22,7 +22,7 @@ def test_bulk_delete_sessiondb_work_runs_off_event_loop(monkeypatch):
     db_modes: list[bool] = []
 
     class _DB:
-        def delete_sessions(self, ids):
+        def delete_sessions(self, ids, **kwargs):
             db_threads.append(threading.get_ident())
             assert ids == ["one", "two"]
             return 2
diff --git a/tests/hermes_state/test_delete_session_write_guards.py b/tests/hermes_state/test_delete_session_write_guards.py
new file mode 100644
index 0000000000..5ab9b65972
--- /dev/null
+++ b/tests/hermes_state/test_delete_session_write_guards.py
@@ -0,0 +1,65 @@
+"""Invariant tests for entry-side deletion refusal on active write guards (#123583)."""
+
+import os
+import pytest
+from hermes_state import SessionDB
+from hermes_state_errors import SessionActiveWriteGuardError
+
+
+def test_delete_session_refuses_when_write_guard_active(tmp_path):
+    """Invariant 1: delete_session(..., exclude_active_write_guards=True) refuses inside
+    the transaction while an active turn lease or compression lock protects the row."""
+    path = tmp_path / "state.db"
+    db = SessionDB(path)
+    db.create_session("sess-lease", source="test")
+    db.create_session("sess-cmp", source="test")
+
+    turn_holder = f"pid={os.getpid()}:turn=1"
+    cmp_holder = f"pid={os.getpid()}:cmp=1"
+
+    assert db.try_acquire_session_turn_lease("sess-lease", turn_holder, ttl_seconds=300.0) is True
+    assert db.try_acquire_compression_lock("sess-cmp", cmp_holder, ttl_seconds=300.0) is True
+
+    # 1. Active turn lease -> raises SessionActiveWriteGuardError and row survives
+    with pytest.raises(SessionActiveWriteGuardError):
+        db.delete_session("sess-lease", exclude_active_write_guards=True)
+    assert db.get_session("sess-lease") is not None
+
+    # 2. Active compression lock -> raises SessionActiveWriteGuardError and row survives
+    with pytest.raises(SessionActiveWriteGuardError):
+        db.delete_session("sess-cmp", exclude_active_write_guards=True)
+    assert db.get_session("sess-cmp") is not None
+
+    # 3. Released guards -> deletion succeeds
+    db.release_session_turn_lease("sess-lease", turn_holder)
+    db.release_compression_lock("sess-cmp", cmp_holder)
+
+    assert db.delete_session("sess-lease", exclude_active_write_guards=True) is True
+    assert db.get_session("sess-lease") is None
+
+    assert db.delete_session("sess-cmp", exclude_active_write_guards=True) is True
+    assert db.get_session("sess-cmp") is None
+    db.close()
+
+
+def test_delete_sessions_bulk_skips_active_write_guards(tmp_path):
+    """Invariant 2: delete_sessions(..., exclude_active_write_guards=True) atomically
+    skips rows with active guards and removes only the idle ones."""
+    path = tmp_path / "state.db"
+    db = SessionDB(path)
+    db.create_session("bulk-active", source="test")
+    db.create_session("bulk-idle", source="test")
+
+    turn_holder = f"pid={os.getpid()}:turn=bulk"
+    assert db.try_acquire_session_turn_lease("bulk-active", turn_holder, ttl_seconds=300.0) is True
+
+    deleted_count = db.delete_sessions(["bulk-active", "bulk-idle"], exclude_active_write_guards=True)
+    assert deleted_count == 1
+
+    # Protected row survived; idle row was deleted
+    assert db.get_session("bulk-active") is not None
+    assert db.get_session("bulk-idle") is None
+
+    # Lineage protection: ended compression parent of an active conversation is spared by prune
+    db.release_session_turn_lease("bulk-active", turn_holder)
+    db.close()
diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py
index 7b6e85b8b4..9e449a57c7 100644
--- a/tui_gateway/methods_session.py
+++ b/tui_gateway/methods_session.py
@@ -1086,9 +1086,12 @@ def _(rid, params: dict) -> dict:
     with _profile_db(params, writer=True) as db:
         if db is None:
             return _db_unavailable_error(rid, code=5036)
+        from hermes_state_errors import SessionActiveWriteGuardError
         try:
             home = Path(profile_home) if profile_home is not None else get_hermes_home()
-            deleted = db.delete_session(target, sessions_dir=home / "sessions")
+            deleted = db.delete_session(target, sessions_dir=home / "sessions", exclude_active_write_guards=True)
+        except SessionActiveWriteGuardError:
+            return _err(rid, 4023, "cannot delete an active session")
         except Exception as e:
             return _err(rid, 5036, f"delete failed: {e}")
     return _ok(rid, {"deleted": target}) if deleted else _err(rid, 4007, "session not found")
diff --git a/website/docs/user-guide/features/web-dashboard.md b/website/docs/user-guide/features/web-dashboard.md
index 7b164fad86..4bee7a415d 100644
--- a/website/docs/user-guide/features/web-dashboard.md
+++ b/website/docs/user-guide/features/web-dashboard.md
@@ -528,7 +528,7 @@ Full-text search across message content. Query parameter: `q`. Returns matching
 
 ### DELETE /api/sessions/\{session_id\}
 
-Deletes a session and its message history.
+Deletes a session and its message history. Returns `409 Conflict` if the session has an active turn lease or compression lock.
 
 ### GET /api/logs
 
diff --git a/website/docs/user-guide/sessions.md b/website/docs/user-guide/sessions.md
index 582c2b184d..4403ac6bf2 100644
--- a/website/docs/user-guide/sessions.md
+++ b/website/docs/user-guide/sessions.md
@@ -470,6 +470,8 @@ hermes sessions delete 20250305_091523_a1b2c3d4 --yes
 
 Deleting a session that is still open in a running chat does not stop that chat: its next save recreates the session under the same id with the full in-memory transcript. Close the chat first if you want the session gone.
 
+Deleting a session while a turn is actively executing or compressing is refused (exits with code 1) to prevent transcript loss under the live agent. Wait for the active turn or compression to complete before deleting.
+
 ### Rename a Session
 
 ```bash