@@ -5,7 +5,10 @@
 
 import sqlalchemy as sa
 from pydantic import (
+    ConfigDict,
+    Discriminator,
     HttpUrl,
+    Tag,
     SerializerFunctionWrapHandler,
     model_serializer,
     model_validator,
@@ -464,7 +467,46 @@ class ProxyCompletionConfig(SQLModel):
 
 
 class Validator(SQLModel):
+    """Reference to a validator config saved in the guardrails service."""
+
     validator_config_id: UUID
+    is_enabled: bool = Field(
+        default=True,
+        description="When false, the reference is neither fetched nor run.",
+    )
+
+
+class InlineValidator(SQLModel):
+    """Validator spec stored directly in the config blob.
+
+    Per-type params (languages, ban_list_id, ...) are extras validated by the
+    guardrails service, so they are allowed through untouched here.
+    """
+
+    model_config = ConfigDict(extra="allow")  # pyright: ignore[reportAssignmentType]
+
+    type: str = Field(..., description="Guardrails validator type, e.g. ban_list")
+    on_fail: str | None = Field(default=None, description="Action on failure")
+    is_enabled: bool = Field(
+        default=True,
+        description="When false, the spec is kept in the blob but not run.",
+    )
+
+
+def _guardrail_entry_tag(value: Any) -> str:
+    if isinstance(value, dict):
+        has_ref = "validator_config_id" in value
+    else:
+        has_ref = hasattr(value, "validator_config_id")
+    return "ref" if has_ref else "inline"
+
+
+# Explicit tag: InlineValidator allows extras, so a smart union could also
+# swallow a reference dict that happens to carry a `type`.
+GuardrailEntry = Annotated[
+    Annotated[Validator, Tag("ref")] | Annotated[InlineValidator, Tag("inline")],
+    Discriminator(_guardrail_entry_tag),
+]
 
 
 class PromptTemplate(SQLModel):
@@ -500,12 +542,12 @@ def _default_proxy_provider(cls, data: Any) -> Any:
         description="Prompt template with {{input}} placeholder to wrap around the user input",
     )
 
-    input_guardrails: list[Validator] | None = Field(
+    input_guardrails: list[GuardrailEntry] | None = Field(
         default=None,
         description="Guardrails applied to validate/sanitize the input before the LLM call",
     )
 
-    output_guardrails: list[Validator] | None = Field(
+    output_guardrails: list[GuardrailEntry] | None = Field(
         default=None,
         description="Guardrails applied to validate/sanitize the output after the LLM call",
     )
 
@@ -11,14 +11,18 @@
 from opentelemetry import trace
 
 from app.core.config import settings
-from app.models.llm.request import Validator
+from app.models.llm.request import InlineValidator, Validator
 
 logger = logging.getLogger(__name__)
 tracer = trace.get_tracer(__name__)
 
 
 GUARDRAILS_PROXY_TIMEOUT_SECONDS = 30.0
 
+# The service logs validator_log.stage from the spec and defaults to input.
+GUARDRAIL_STAGE_INPUT = "input"
+GUARDRAIL_STAGE_OUTPUT = "output"
+
 
 def _guardrails_headers(
     organization_id: int | None, project_id: int | None
@@ -148,7 +152,7 @@ def applied(self) -> bool:
 def apply_guardrails(
     *,
     text: str,
-    validators: list[Validator] | None,
+    validators: Sequence[Validator | InlineValidator] | None,
     job_id: UUID,
     project_id: int | None,
     organization_id: int | None,
@@ -164,30 +168,45 @@ def apply_guardrails(
 
     Args:
         text: Primary text to validate. Sent as `input` to the service.
-        validators: Validator references (config IDs). When None/empty the
-            function short-circuits with a no-op outcome.
+        validators: Config-ID references and/or inline specs, in run order.
+            Entries with is_enabled=False are skipped; when nothing enabled
+            remains the function short-circuits with a no-op outcome.
         output_text: When provided, sent as `output` to the service for
             output-guardrail validators that compare input/output pairs.
             Also routes the IDs through the output-config fetch path.
     """
-    if not validators:
+    enabled: list[Validator | InlineValidator] = []
+    for entry in validators or []:
+        if entry.is_enabled:
+            enabled.append(entry)
+    if not enabled:
         return GuardrailsOutcome(
             safe_text=text, error=None, bypassed=False, rephrase_needed=False, raw={}
         )
 
     is_output = output_text is not None
-    input_cfgs, output_cfgs = list_validators_config(
-        organization_id=organization_id,
-        project_id=project_id,
-        input_validator_configs=None if is_output else validators,
-        output_validator_configs=validators if is_output else None,
-    )
-    resolved = output_cfgs if is_output else input_cfgs
+    references: list[Validator] = []
+    for entry in enabled:
+        if isinstance(entry, Validator):
+            references.append(entry)
+
+    fetched: list[dict[str, Any]] = []
+    if references:
+        input_cfgs, output_cfgs = list_validators_config(
+            organization_id=organization_id,
+            project_id=project_id,
+            input_validator_configs=None if is_output else references,
+            output_validator_configs=references if is_output else None,
+        )
+        fetched = output_cfgs if is_output else input_cfgs
+
+    stage = GUARDRAIL_STAGE_OUTPUT if is_output else GUARDRAIL_STAGE_INPUT
+    resolved = _prepare_validators(enabled, fetched, stage, job_id)
     if not resolved:
         logger.info(
             f"[apply_guardrails] No validator configs resolved upstream; skipping "
             f"POST /guardrails. job_id={job_id}, requested_ids="
-            f"{[str(v.validator_config_id) for v in validators]}, "
+            f"{[str(v.validator_config_id) for v in references]}, "
             f"is_output={is_output}"
         )
         return GuardrailsOutcome(
@@ -237,6 +256,40 @@ def apply_guardrails(
     )
 
 
+def _prepare_validators(
+    entries: Sequence[Validator | InlineValidator],
+    fetched_rows: list[dict[str, Any]],
+    stage: str,
+    job_id: UUID,
+) -> list[dict[str, Any]]:
+    """Build the POST `validators` list in blob order, since the service runs
+    them in the order received while the fetch returns rows by created_at."""
+    rows_by_id = {str(row.get("id")): row for row in fetched_rows}
+
+    prepared: list[dict[str, Any]] = []
+    for entry in entries:
+        if isinstance(entry, InlineValidator):
+            # The service's validator configs are extra="forbid", so is_enabled must go.
+            spec = entry.model_dump(mode="json", exclude={"is_enabled"})
+            # Service's on_fail is non-nullable; omitting it applies its default.
+            if spec.get("on_fail") is None:
+                spec.pop("on_fail", None)
+            spec["stage"] = stage
+            prepared.append(spec)
+            continue
+
+        config_id = str(entry.validator_config_id)
+        row = rows_by_id.get(config_id)
+        if row is None:
+            logger.warning(
+                f"[_prepare_validators] Validator config not returned upstream; "
+                f"dropping it | job_id={job_id}, validator_config_id={config_id}"
+            )
+            continue
+        prepared.append(row)
+    return prepared
+
+
 def summarize_validator_results(outcome: GuardrailsOutcome) -> list[dict[str, Any]]:
     """Per-validator name/outcome/text summary, extracted from the raw
     guardrails service response, for use in llm_call metadata."""
@@ -259,7 +312,7 @@ def summarize_validator_results(outcome: GuardrailsOutcome) -> list[dict[str, An
 
 def run_guardrails_validation(
     input_text: str,
-    guardrail_config: Sequence[Validator | dict[str, Any]],
+    guardrail_config: Sequence[Validator | InlineValidator | dict[str, Any]],
     job_id: UUID,
     project_id: int | None,
     organization_id: int | None,
@@ -282,12 +335,13 @@ def run_guardrails_validation(
     Returns:
         JSON response from the guardrails service with validation results.
     """
-    validators = [
-        validator.model_dump(mode="json")
-        if isinstance(validator, Validator)
-        else validator
-        for validator in guardrail_config
-    ]
+    validators: list[dict[str, Any]] = []
+    for validator in guardrail_config:
+        if isinstance(validator, dict):
+            validators.append(validator)
+        else:
+            # Unknown to the service's extra="forbid" validator schemas.
+            validators.append(validator.model_dump(mode="json", exclude={"is_enabled"}))
 
     payload = {
         "request_id": str(job_id),
 
@@ -1,4 +1,12 @@
-from app.models.llm.request import build_kaapi_completion_config
+from uuid import uuid4
+
+from app.models.llm.request import (
+    ConfigBlob,
+    InlineValidator,
+    NativeCompletionConfig,
+    Validator,
+    build_kaapi_completion_config,
+)
 from app.services.llm.mappers import kaapi_params_as_dict
 
 
@@ -67,3 +75,76 @@ def test_unset_temperature_survives_json_round_trip(self) -> None:
         assert "temperature" not in kaapi_params_as_dict(
             round_tripped.completion.params
         )
+
+
+def _blob(input_guardrails: list) -> ConfigBlob:
+    return ConfigBlob(
+        completion=NativeCompletionConfig(
+            provider="openai-native", type="text", params={"model": "gpt-4o"}
+        ),
+        input_guardrails=input_guardrails,
+    )
+
+
+class TestGuardrailEntryParsing:
+    def test_dict_with_config_id_parses_to_validator(self) -> None:
+        config_id = uuid4()
+        blob = _blob([{"validator_config_id": str(config_id)}])
+
+        (entry,) = blob.input_guardrails or []
+        assert isinstance(entry, Validator)
+        assert entry.validator_config_id == config_id
+        assert entry.is_enabled is True
+
+    def test_dict_with_type_parses_to_inline_and_keeps_extras(self) -> None:
+        blob = _blob(
+            [{"type": "ban_list", "on_fail": "fix", "languages": ["en", "hi"]}]
+        )
+
+        (entry,) = blob.input_guardrails or []
+        assert isinstance(entry, InlineValidator)
+        assert entry.model_dump() == {
+            "type": "ban_list",
+            "on_fail": "fix",
+            "is_enabled": True,
+            "languages": ["en", "hi"],
+        }
+
+    def test_ref_dict_carrying_type_still_parses_to_validator(self) -> None:
+        # InlineValidator allows extras, so without the discriminator a smart
+        # union could claim this ref.
+        config_id = uuid4()
+        blob = _blob([{"validator_config_id": str(config_id), "type": "ban_list"}])
+
+        (entry,) = blob.input_guardrails or []
+        assert isinstance(entry, Validator)
+        assert entry.model_dump() == {
+            "validator_config_id": config_id,
+            "is_enabled": True,
+        }
+
+    def test_revalidating_parsed_entries_keeps_types(self) -> None:
+        parsed = _blob(
+            [{"validator_config_id": str(uuid4())}, {"type": "pii_remover"}]
+        ).input_guardrails
+
+        rebuilt = _blob(list(parsed or [])).input_guardrails or []
+
+        assert [type(e) for e in rebuilt] == [Validator, InlineValidator]
+
+    def test_is_enabled_false_survives_json_round_trip(self) -> None:
+        config_id = uuid4()
+        dumped = _blob(
+            [
+                {"validator_config_id": str(config_id), "is_enabled": False},
+                {"type": "pii_remover", "is_enabled": False},
+            ]
+        ).model_dump(mode="json")
+
+        ref, inline = ConfigBlob.model_validate(dumped).input_guardrails or []
+
+        assert isinstance(ref, Validator)
+        assert ref.validator_config_id == config_id
+        assert ref.is_enabled is False
+        assert isinstance(inline, InlineValidator)
+        assert inline.is_enabled is False