diff --git a/python/ktx-daemon/src/ktx_daemon/app.py b/python/ktx-daemon/src/ktx_daemon/app.py
index 63487439..6c790931 100644
--- a/python/ktx-daemon/src/ktx_daemon/app.py
+++ b/python/ktx-daemon/src/ktx_daemon/app.py
@@ -12,6 +12,7 @@ from typing import Any
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse, Response
+from pydantic import ValidationError
 
 from ktx_daemon import VERSION
 from ktx_daemon.code_execution import (
@@ -268,6 +269,14 @@ def create_app(
     ) -> SemanticLayerQueryResponse:
         try:
             return query_semantic_layer(request)
+        except ValidationError as error:
+            # A malformed source or response is a ktx contract fault, not a
+            # caller rejection; surface it as a server fault (500) so the Node
+            # client does not classify it as an expected query rejection, matching
+            # the subprocess transport (exit 1) and query_semantic_layer's own
+            # report. ValidationError subclasses ValueError, so catch it first.
+            logger.exception("Semantic query failed on a malformed source")
+            raise HTTPException(status_code=500, detail=str(error)) from error
         except ValueError as error:
             logger.warning("Semantic query rejected: %s", error)
             raise HTTPException(status_code=400, detail=str(error)) from error
diff --git a/python/ktx-daemon/tests/test_app.py b/python/ktx-daemon/tests/test_app.py
index fffc2899..e6ec9ed6 100644
--- a/python/ktx-daemon/tests/test_app.py
+++ b/python/ktx-daemon/tests/test_app.py
@@ -424,6 +424,24 @@ def test_semantic_query_endpoint_maps_value_error_to_400() -> None:
     assert "missing.order_count" in response.json()["detail"]
 
 
+def test_semantic_query_endpoint_maps_malformed_source_to_fault() -> None:
+    client = TestClient(create_app(), raise_server_exceptions=False)
+    invalid_source = {
+        key: value for key, value in ORDERS_SOURCE.items() if key != "table"
+    }
+
+    response = client.post(
+        "/semantic-layer/query",
+        json={
+            "sources": [invalid_source],
+            "dialect": "postgres",
+            "query": {"measures": ["orders.order_count"], "dimensions": []},
+        },
+    )
+
+    assert response.status_code == 500
+
+
 def test_semantic_validate_endpoint_returns_structured_validation() -> None:
     client = TestClient(create_app())
     invalid_source = {