fix(history): reject falsy non-string message IDs; drop redundant keep_count guard

The or-chain used to resolve before_msg_id / message_id evaluated
falsy non-strings (False, 0) to '' before the isinstance(…, str) check
was reached, so they silently bypassed type validation.  Replace the
chain with explicit presence-aware logic that falls through on None or
empty string (preserving alias semantics) but rejects any other
non-string value with HTTP 400.

Also remove the now-tautological 'if keep_count >= 0:' guard, which
is always True after the negativity check introduced in the parent
commit.  The DB block is unconditional and indented accordingly.

Tests: add {'before_msg_id': False}, {'before_msg_id': 0},
{'message_id': 0} to the parametrised rejection suite.

36 passed, 0 failed (Python 3.14.4 / pytest 9.1.1).

routes/history/history_routes.py

@@ -391,26 +391,53 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
     async def truncate_session(request: Request, session_id: str):
         _verify_session_owner(request, session_id)
         try:
-            body = await request.json()
-            keep_count = int(body.get("keep_count", 0))
-            before_msg_id = str(body.get("before_msg_id") or body.get("message_id") or "").strip()
+            try:
+                body = await request.json()
+            except json.JSONDecodeError:
+                raise HTTPException(400, "Request body must be valid JSON")
+            if not isinstance(body, dict):
+                raise HTTPException(400, "Request body must be a JSON object")
+            # Resolve before_msg_id / message_id alias with explicit presence checks.
+            # None (JSON null) and empty string are both treated as "not provided here"
+            # and fall through to the alias, preserving the original OR-chain semantics.
+            # Any other non-string value (False, 0, True, 123, list, …) is rejected
+            # immediately so falsy non-strings cannot bypass type validation.
+            raw_msg_id = body.get("before_msg_id") if "before_msg_id" in body else None
+            if raw_msg_id is None or raw_msg_id == "":
+                raw_msg_id = body.get("message_id")
+            if raw_msg_id is not None and not isinstance(raw_msg_id, str):
+                raise HTTPException(400, "Message ID must be a string")
+            before_msg_id = (raw_msg_id or "").strip()
+            if "keep_count" not in body and not before_msg_id:
+                raise HTTPException(400, "keep_count or before_msg_id required")
+            raw_count = body.get("keep_count", 0)
+            # Keep integer-string clients working, but do not silently convert
+            # booleans or fractional numbers into destructive message counts.
+            if type(raw_count) not in (int, str):
+                raise HTTPException(400, "keep_count must be a non-negative integer")
+            try:
+                keep_count = int(raw_count)
+            except ValueError:
+                raise HTTPException(400, "keep_count must be a non-negative integer")
+            if keep_count < 0:
+                raise HTTPException(400, "keep_count must be a non-negative integer")
             deleted_sft_pairs: list[dict[str, str]] = []
-            if keep_count >= 0:
-                db = SessionLocal()
-                try:
-                    all_db_messages = db.query(DbChatMessage).filter(
-                        DbChatMessage.session_id == session_id
-                    ).order_by(DbChatMessage.timestamp).all()
-                    if before_msg_id:
-                        resolved_keep_count = _keep_count_before_message(all_db_messages, before_msg_id)
-                        if resolved_keep_count is None:
-                            raise HTTPException(404, "Message not found")
-                        keep_count = resolved_keep_count
-                    for pos, row in enumerate(all_db_messages):
-                        row._will_delete_for_sft = pos >= keep_count
-                    deleted_sft_pairs = _deleted_sft_pairs_from_db_rows(all_db_messages)
-                finally:
-                    db.close()
+            # keep_count is always >= 0 here; the block is unconditional.
+            db = SessionLocal()
+            try:
+                all_db_messages = db.query(DbChatMessage).filter(
+                    DbChatMessage.session_id == session_id
+                ).order_by(DbChatMessage.timestamp).all()
+                if before_msg_id:
+                    resolved_keep_count = _keep_count_before_message(all_db_messages, before_msg_id)
+                    if resolved_keep_count is None:
+                        raise HTTPException(404, "Message not found")
+                    keep_count = resolved_keep_count
+                for pos, row in enumerate(all_db_messages):
+                    row._will_delete_for_sft = pos >= keep_count
+                deleted_sft_pairs = _deleted_sft_pairs_from_db_rows(all_db_messages)
+            finally:
+                db.close()
             result = session_manager.truncate_messages(session_id, keep_count)
             _remove_deleted_sft_trace_rows(
                 owner=effective_user(request),

tests/test_history_truncate_validation.py

@@ -0,0 +1,98 @@
+"""Invalid truncation requests must leave saved chat history intact."""
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from tests.helpers.database import isolated_session_database
+
+
+@pytest.fixture
+def history_client(tmp_path, monkeypatch):
+    from core.models import ChatMessage
+    from routes.history import history_routes
+    from routes import session_routes
+
+    with isolated_session_database(tmp_path) as (manager, database):
+        monkeypatch.setattr(history_routes, "SessionLocal", database.SessionLocal)
+        monkeypatch.setattr(session_routes, "SessionLocal", database.SessionLocal)
+        monkeypatch.setenv("AUTH_ENABLED", "true")
+        manager.create_session("truncate-test", "Test", "http://example.test", "test", owner="alice")
+        for text in ("first", "second", "third"):
+            manager.add_message("truncate-test", ChatMessage("user", text))
+        app = FastAPI()
+        app.state.auth_manager = SimpleNamespace(is_configured=True)
+
+        @app.middleware("http")
+        async def owner(request, call_next):
+            request.state.current_user = request.headers.get("X-Test-Owner", "alice")
+            return await call_next(request)
+
+        app.include_router(history_routes.setup_history_routes(manager))
+        with TestClient(app, raise_server_exceptions=False) as client:
+            yield client, manager, database
+
+
+def saved_messages(database):
+    with database.SessionLocal() as db:
+        return [row.content for row in db.query(database.ChatMessage).filter(
+            database.ChatMessage.session_id == "truncate-test"
+        ).order_by(database.ChatMessage.timestamp).all()]
+
+
+@pytest.mark.parametrize("payload", [
+    {}, {"unrelated": 1}, {"before_msg_id": ""},
+    {"keep_count": None}, {"keep_count": False}, {"keep_count": True},
+    {"keep_count": 1.9}, {"keep_count": -1}, {"keep_count": "bad"},
+    {"keep_count": []}, {"keep_count": {}},
+    {"before_msg_id": True}, {"before_msg_id": ["message"]},
+    {"before_msg_id": False}, {"before_msg_id": 0}, {"message_id": 0},
+    [], None, "two", 2, False,
+])
+def test_invalid_target_preserves_history(history_client, payload):
+    client, manager, database = history_client
+    import json
+    response = client.post("/api/session/truncate-test/truncate", content=json.dumps(payload), headers={"Content-Type": "application/json"})
+    assert response.status_code == 400, response.text
+    assert saved_messages(database) == ["first", "second", "third"]
+    assert [m.content for m in manager.get_session("truncate-test").history] == ["first", "second", "third"]
+
+
+def test_malformed_json_preserves_history(history_client):
+    client, _, database = history_client
+    response = client.post("/api/session/truncate-test/truncate", content="{", headers={"Content-Type": "application/json"})
+    assert response.status_code == 400, response.text
+    assert saved_messages(database) == ["first", "second", "third"]
+
+
+@pytest.mark.parametrize("count,expected", [(0, []), (2, ["first", "second"]), (9, ["first", "second", "third"]), ("2", ["first", "second"])])
+def test_explicit_count_remains_supported(history_client, count, expected):
+    client, _, database = history_client
+    response = client.post("/api/session/truncate-test/truncate", json={"keep_count": count})
+    assert response.status_code == 200, response.text
+    assert saved_messages(database) == expected
+
+
+@pytest.mark.parametrize("field", ["before_msg_id", "message_id"])
+def test_message_id_boundary_remains_supported(history_client, field):
+    client, _, database = history_client
+    with database.SessionLocal() as db:
+        message_id = db.query(database.ChatMessage).filter(database.ChatMessage.content == "second").one().id
+    response = client.post("/api/session/truncate-test/truncate", json={field: message_id})
+    assert response.status_code == 200, response.text
+    assert saved_messages(database) == ["first"]
+
+
+def test_missing_message_id_does_not_clear_history(history_client):
+    client, _, database = history_client
+    response = client.post("/api/session/truncate-test/truncate", json={"before_msg_id": "missing"})
+    assert response.status_code == 404
+    assert saved_messages(database) == ["first", "second", "third"]
+
+
+def test_owner_gate_still_precedes_validation(history_client):
+    client, _, database = history_client
+    response = client.post("/api/session/truncate-test/truncate", json={}, headers={"X-Test-Owner": "bob"})
+    assert response.status_code == 404
+    assert saved_messages(database) == ["first", "second", "third"]