diff --git a/alembic/versions/0008_metadata_fields.py b/alembic/versions/0008_metadata_fields.py
new file mode 100644
index 000000000..922127613
--- /dev/null
+++ b/alembic/versions/0008_metadata_fields.py
@@ -0,0 +1,39 @@
+"""custom metadata field catalog
+
+Revision ID: 0008_metadata_fields
+Revises: 0007_add_knowledge_delete_anonymous
+Create Date: 2026-07-16 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0008_metadata_fields"
+down_revision: str | Sequence[str] | None = "0007_add_knowledge_delete_anonymous"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "metadata_fields",
+        sa.Column("key", sa.String(length=64), nullable=False),
+        sa.Column("metadata_type", sa.String(length=16), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint("key", name="pk_metadata_fields"),
+    )
+    op.create_index(
+        "ix_metadata_fields_metadata_type",
+        "metadata_fields",
+        ["metadata_type"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_metadata_fields_metadata_type", table_name="metadata_fields")
+    op.drop_table("metadata_fields")
diff --git a/frontend/app/api/queries/useGetSearchQuery.ts b/frontend/app/api/queries/useGetSearchQuery.ts
index 33b98d3af..788386d09 100644
--- a/frontend/app/api/queries/useGetSearchQuery.ts
+++ b/frontend/app/api/queries/useGetSearchQuery.ts
@@ -3,6 +3,7 @@ import {
   useQuery,
   useQueryClient,
 } from "@tanstack/react-query";
+import type { MetadataGroup } from "@/components/metadata-filter-builder";
 import type { ParsedQueryData } from "@/contexts/knowledge-filter-context";
 import { SEARCH_CONSTANTS } from "@/lib/constants";
 import { buildSearchPayloadFilters } from "@/lib/filter-normalization";
@@ -16,6 +17,7 @@ export interface SearchPayload {
     document_types?: string[];
     owners?: string[];
     connector_types?: string[];
+    metadata?: MetadataGroup;
   };
 }
 
@@ -39,6 +41,7 @@ export interface ChunkResult {
   index?: number;
   allowed_users?: string[];
   allowed_groups?: string[];
+  metadata?: Record<string, { type: string; value: unknown }>;
 }
 
 export interface File {
diff --git a/frontend/app/chat/_types/types.ts b/frontend/app/chat/_types/types.ts
index fc89edda3..9a85fe37d 100644
--- a/frontend/app/chat/_types/types.ts
+++ b/frontend/app/chat/_types/types.ts
@@ -88,6 +88,7 @@ export interface SelectedFilters {
   document_types: string[];
   owners: string[];
   connector_types: string[];
+  metadata?: import("@/components/metadata-filter-builder").MetadataGroup;
 }
 
 export interface KnowledgeFilterData {
@@ -109,6 +110,7 @@ export interface RequestBody {
     document_types?: string[];
     owners?: string[];
     connector_types?: string[];
+    metadata?: import("@/components/metadata-filter-builder").MetadataGroup;
   };
   filter_id?: string;
   limit?: number;
diff --git a/frontend/components/cloud-picker/custom-metadata-editor.tsx b/frontend/components/cloud-picker/custom-metadata-editor.tsx
new file mode 100644
index 000000000..47ee13bcd
--- /dev/null
+++ b/frontend/components/cloud-picker/custom-metadata-editor.tsx
@@ -0,0 +1,177 @@
+"use client";
+
+import { Plus, Trash2 } from "lucide-react";
+import { useEffect, useMemo, useState } from "react";
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+  SelectValue,
+} from "@/components/ui/select";
+import type { CustomMetadataEntry, CustomMetadataType } from "./types";
+
+interface MetadataFieldSuggestion {
+  key: string;
+  type: CustomMetadataType;
+}
+
+interface Props {
+  value: CustomMetadataEntry[];
+  onChange: (value: CustomMetadataEntry[]) => void;
+}
+
+const defaultValue = (type: CustomMetadataType): string | number | boolean => {
+  if (type === "number") return 0;
+  if (type === "boolean") return false;
+  return "";
+};
+
+export function CustomMetadataEditor({ value, onChange }: Props) {
+  const [suggestions, setSuggestions] = useState<MetadataFieldSuggestion[]>([]);
+
+  useEffect(() => {
+    let active = true;
+    fetch("/api/metadata/fields")
+      .then((response) => (response.ok ? response.json() : { fields: [] }))
+      .then((result) => {
+        if (active) setSuggestions(result.fields ?? []);
+      })
+      .catch(() => undefined);
+    return () => {
+      active = false;
+    };
+  }, []);
+
+  const fieldTypes = useMemo(
+    () => new Map(suggestions.map((field) => [field.key, field.type])),
+    [suggestions],
+  );
+
+  const update = (index: number, patch: Partial<CustomMetadataEntry>) => {
+    const next = [...value];
+    next[index] = { ...next[index], ...patch };
+    onChange(next);
+  };
+
+  return (
+    <div className="mt-6 space-y-3 border-t pt-4">
+      <div>
+        <div className="text-sm font-semibold">Custom metadata</div>
+        <div className="text-sm text-muted-foreground">
+          Applied to every document selected for this ingest.
+        </div>
+      </div>
+      <datalist id="custom-metadata-keys">
+        {suggestions.map((field) => (
+          <option key={field.key} value={field.key} />
+        ))}
+      </datalist>
+      {value.map((entry, index) => {
+        const knownType = fieldTypes.get(entry.key.trim().toLowerCase());
+        return (
+          <div
+            className="grid grid-cols-[minmax(0,1fr)_130px_minmax(0,1fr)_36px] gap-2"
+            key={`${index}-${entry.key}`}
+          >
+            <Input
+              aria-label="Metadata key"
+              list="custom-metadata-keys"
+              placeholder="supplier"
+              value={entry.key}
+              onChange={(event) => {
+                const key = event.target.value.toLowerCase();
+                const suggestedType = fieldTypes.get(key);
+                update(index, {
+                  key,
+                  ...(suggestedType && suggestedType !== entry.type
+                    ? {
+                        type: suggestedType,
+                        value: defaultValue(suggestedType),
+                      }
+                    : {}),
+                });
+              }}
+            />
+            <Select
+              disabled={!!knownType}
+              value={knownType ?? entry.type}
+              onValueChange={(type: CustomMetadataType) =>
+                update(index, { type, value: defaultValue(type) })
+              }
+            >
+              <SelectTrigger aria-label="Metadata type">
+                <SelectValue />
+              </SelectTrigger>
+              <SelectContent>
+                <SelectItem value="string">Text</SelectItem>
+                <SelectItem value="number">Number</SelectItem>
+                <SelectItem value="date">Date</SelectItem>
+                <SelectItem value="boolean">Boolean</SelectItem>
+              </SelectContent>
+            </Select>
+            {entry.type === "boolean" ? (
+              <Select
+                value={String(entry.value)}
+                onValueChange={(next) =>
+                  update(index, { value: next === "true" })
+                }
+              >
+                <SelectTrigger aria-label="Metadata value">
+                  <SelectValue />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem value="true">True</SelectItem>
+                  <SelectItem value="false">False</SelectItem>
+                </SelectContent>
+              </Select>
+            ) : (
+              <Input
+                aria-label="Metadata value"
+                type={
+                  entry.type === "date"
+                    ? "date"
+                    : entry.type === "number"
+                      ? "number"
+                      : "text"
+                }
+                value={String(entry.value)}
+                onChange={(event) =>
+                  update(index, {
+                    value:
+                      entry.type === "number"
+                        ? Number(event.target.value)
+                        : event.target.value,
+                  })
+                }
+              />
+            )}
+            <Button
+              aria-label="Remove metadata"
+              onClick={() =>
+                onChange(value.filter((_, itemIndex) => itemIndex !== index))
+              }
+              size="icon"
+              type="button"
+              variant="ghost"
+            >
+              <Trash2 className="h-4 w-4" />
+            </Button>
+          </div>
+        );
+      })}
+      <Button
+        onClick={() =>
+          onChange([...value, { key: "", type: "string", value: "" }])
+        }
+        size="sm"
+        type="button"
+        variant="outline"
+      >
+        <Plus className="h-4 w-4" /> Add metadata
+      </Button>
+    </div>
+  );
+}
diff --git a/frontend/components/cloud-picker/ingest-settings.tsx b/frontend/components/cloud-picker/ingest-settings.tsx
index 684dc84f5..1bdc18fd5 100644
--- a/frontend/components/cloud-picker/ingest-settings.tsx
+++ b/frontend/components/cloud-picker/ingest-settings.tsx
@@ -35,6 +35,7 @@ import {
 } from "@/components/ui/tooltip";
 import { useAuth } from "@/contexts/auth-context";
 import { knowledgeToIngestSettings } from "@/lib/ingest-settings-knowledge";
+import { CustomMetadataEditor } from "./custom-metadata-editor";
 import type { IngestSettings as IngestSettingsType } from "./types";
 
 interface IngestSettingsProps {
@@ -285,6 +286,12 @@ export const IngestSettings = ({
             </div>
           </div>
         )}
+        {showAdvancedSettings && (
+          <CustomMetadataEditor
+            value={currentSettings.metadata ?? []}
+            onChange={(metadata) => handleSettingsChange({ metadata })}
+          />
+        )}
 
         <div className={showAdvancedSettings ? "" : "mt-6"}>
           {showShared && (
diff --git a/frontend/components/cloud-picker/types.ts b/frontend/components/cloud-picker/types.ts
index fc125cbef..d70007f12 100644
--- a/frontend/components/cloud-picker/types.ts
+++ b/frontend/components/cloud-picker/types.ts
@@ -115,6 +115,17 @@ export interface IngestSettings {
   embeddingModel: string;
   /** When true, index without an owner so all users in the instance can retrieve the document. COS only. */
   shared?: boolean;
+  /** Typed custom metadata applied to every document in the ingest batch. */
+  metadata?: CustomMetadataEntry[];
+}
+
+export type CustomMetadataType = "string" | "number" | "date" | "boolean";
+export type CustomMetadataValue = string | number | boolean;
+
+export interface CustomMetadataEntry {
+  key: string;
+  type: CustomMetadataType;
+  value: CustomMetadataValue;
 }
 
 /** Inline error message if chunk settings are invalid; otherwise null. */
diff --git a/frontend/components/knowledge-dropdown.tsx b/frontend/components/knowledge-dropdown.tsx
index 1120083b9..98bfe3385 100644
--- a/frontend/components/knowledge-dropdown.tsx
+++ b/frontend/components/knowledge-dropdown.tsx
@@ -14,6 +14,7 @@ import { useEffect, useRef, useState } from "react";
 import { toast } from "sonner";
 import type { File as SearchFile } from "@/app/api/queries/useGetSearchQuery";
 import { useGetTasksQuery } from "@/app/api/queries/useGetTasksQuery";
+import { IngestSettings } from "@/components/cloud-picker/ingest-settings";
 import { DuplicateHandlingDialog } from "@/components/duplicate-handling-dialog";
 import { Button } from "@/components/ui/button";
 import {
@@ -34,6 +35,7 @@ import { Label } from "@/components/ui/label";
 import { useIsCloudBrand } from "@/contexts/brand-context";
 import { useTask } from "@/contexts/task-context";
 import { usePermissions } from "@/hooks/use-permissions";
+import { useSessionIngestSettings } from "@/hooks/useSessionIngestSettings";
 import {
   trackButton,
   trackProcessFailure,
@@ -136,6 +138,13 @@ export function KnowledgeDropdown() {
   const [folderPath, setFolderPath] = useState("");
   const [folderLoading, setFolderLoading] = useState(false);
   const [fileUploading, setFileUploading] = useState(false);
+  const [ingestSettings, setIngestSettings] = useSessionIngestSettings();
+  const [ingestSettingsOpen, setIngestSettingsOpen] = useState(true);
+  const [pendingLocalUpload, setPendingLocalUpload] = useState<{
+    files: File[];
+    replace: boolean;
+    source: "file" | "folder";
+  } | null>(null);
   const [isNavigatingToCloud, setIsNavigatingToCloud] = useState(false);
   const [bucketConnectorConfigured, setBucketConnectorConfigured] = useState<
     Record<string, boolean>
@@ -336,7 +345,11 @@ export function KnowledgeDropdown() {
           resetFileInput();
           return;
         }
-        await uploadFile(file, false);
+        setPendingLocalUpload({
+          files: [file],
+          replace: false,
+          source: "file",
+        });
       } catch (error) {
         console.error("[Duplicate Check] Exception:", error);
         toast.error("Failed to check for duplicates", {
@@ -359,7 +372,13 @@ export function KnowledgeDropdown() {
     });
 
     try {
-      await uploadFileUtil(file, replace);
+      await uploadFileUtil(
+        file,
+        replace,
+        false,
+        undefined,
+        ingestSettings.metadata,
+      );
       refetchTasks();
     } catch (error) {
       trackProcessFailure({
@@ -405,7 +424,11 @@ export function KnowledgeDropdown() {
 
     for (const batch of batches) {
       try {
-        const result = await uploadFiles(batch, replace);
+        const result = await uploadFiles(
+          batch,
+          replace,
+          ingestSettings.metadata,
+        );
         addTask(result.taskId, { source: "folder" });
       } catch (error) {
         trackProcessFailure({
@@ -428,14 +451,12 @@ export function KnowledgeDropdown() {
   const handleOverwriteFile = async () => {
     if (pendingFolderUpload) {
       isFolderOverwriteConfirmedRef.current = true;
-      const { allFiles, duplicateNames, unsupportedCount } =
-        pendingFolderUpload;
-      await uploadFolderBatches(allFiles, true);
-      const unsupportedMessage =
-        unsupportedCount > 0 ? `, skipped ${unsupportedCount} unsupported` : "";
-      toast.success(
-        `Processed ${allFiles.length} file(s), including ${duplicateNames.length} overwrite(s)${unsupportedMessage}`,
-      );
+      const { allFiles } = pendingFolderUpload;
+      setPendingLocalUpload({
+        files: allFiles,
+        replace: true,
+        source: "folder",
+      });
       resetDuplicateDialogState();
       return;
     }
@@ -462,7 +483,11 @@ export function KnowledgeDropdown() {
         return oldData;
       });
 
-      await uploadFile(pendingFile, true);
+      setPendingLocalUpload({
+        files: [pendingFile],
+        replace: true,
+        source: "file",
+      });
 
       resetDuplicateDialogState();
     }
@@ -473,22 +498,13 @@ export function KnowledgeDropdown() {
       if (isFolderOverwriteConfirmedRef.current) {
         isFolderOverwriteConfirmedRef.current = false;
       } else {
-        const { nonDuplicateFiles, duplicateNames, unsupportedCount } =
-          pendingFolderUpload;
+        const { nonDuplicateFiles } = pendingFolderUpload;
         if (nonDuplicateFiles.length > 0) {
-          await uploadFolderBatches(nonDuplicateFiles, false);
-          const extraParts: string[] = [];
-          if (duplicateNames.length > 0) {
-            extraParts.push(`skipped ${duplicateNames.length} duplicate(s)`);
-          }
-          if (unsupportedCount > 0) {
-            extraParts.push(`skipped ${unsupportedCount} unsupported`);
-          }
-          const suffix =
-            extraParts.length > 0 ? `, ${extraParts.join(", ")}` : "";
-          toast.success(
-            `Processed ${nonDuplicateFiles.length} file(s)${suffix}`,
-          );
+          setPendingLocalUpload({
+            files: nonDuplicateFiles,
+            replace: false,
+            source: "folder",
+          });
         } else {
           toast.info(
             "Skipped duplicate files. All selected files were duplicates, so nothing was uploaded.",
@@ -592,12 +608,11 @@ export function KnowledgeDropdown() {
         return;
       }
 
-      await uploadFolderBatches(nonDuplicateFiles, false);
-      const unsupportedMessage =
-        unsupportedCount > 0 ? `, skipped ${unsupportedCount} unsupported` : "";
-      toast.success(
-        `Successfully processed ${nonDuplicateFiles.length} file(s)${unsupportedMessage}`,
-      );
+      setPendingLocalUpload({
+        files: nonDuplicateFiles,
+        replace: false,
+        source: "folder",
+      });
     } catch (error) {
       console.error("Folder upload error:", error);
       toast.error("Folder upload failed", {
@@ -741,6 +756,17 @@ export function KnowledgeDropdown() {
   // Comprehensive loading state
   const isLoading = fileUploading || folderLoading || isNavigatingToCloud;
 
+  const confirmLocalUpload = async () => {
+    if (!pendingLocalUpload) return;
+    const pending = pendingLocalUpload;
+    setPendingLocalUpload(null);
+    if (pending.source === "file" && pending.files.length === 1) {
+      await uploadFile(pending.files[0], pending.replace);
+    } else {
+      await uploadFolderBatches(pending.files, pending.replace);
+    }
+  };
+
   if (!mounted) {
     return (
       <div className="flex h-12 pointer-events-none">
@@ -915,6 +941,36 @@ export function KnowledgeDropdown() {
         duplicateLabel={duplicateFilename}
         duplicateNames={pendingFolderUpload?.duplicateNames}
       />
+
+      <Dialog
+        open={!!pendingLocalUpload}
+        onOpenChange={(open) => !open && setPendingLocalUpload(null)}
+      >
+        <DialogContent className="max-w-2xl">
+          <DialogHeader>
+            <DialogTitle>Configure ingest</DialogTitle>
+            <DialogDescription>
+              These settings and metadata apply to all{" "}
+              {pendingLocalUpload?.files.length ?? 0} selected document(s).
+            </DialogDescription>
+          </DialogHeader>
+          <IngestSettings
+            isOpen={ingestSettingsOpen}
+            onOpenChange={setIngestSettingsOpen}
+            settings={ingestSettings}
+            onSettingsChange={setIngestSettings}
+          />
+          <div className="flex justify-end gap-2">
+            <Button
+              variant="outline"
+              onClick={() => setPendingLocalUpload(null)}
+            >
+              Cancel
+            </Button>
+            <Button onClick={confirmLocalUpload}>Upload</Button>
+          </div>
+        </DialogContent>
+      </Dialog>
     </>
   );
 }
diff --git a/frontend/components/knowledge-filter-panel.tsx b/frontend/components/knowledge-filter-panel.tsx
index 660882958..c92c876f9 100644
--- a/frontend/components/knowledge-filter-panel.tsx
+++ b/frontend/components/knowledge-filter-panel.tsx
@@ -15,6 +15,10 @@ import {
   FilterIconPopover,
   type IconKey,
 } from "@/components/filter-icon-popover";
+import {
+  MetadataFilterBuilder,
+  type MetadataGroup,
+} from "@/components/metadata-filter-builder";
 import { Button } from "@/components/ui/button";
 import {
   Card,
@@ -103,6 +107,7 @@ export function KnowledgeFilterPanel() {
     document_types: ["*"] as string[], // Default to wildcard
     owners: ["*"] as string[], // Default to wildcard
     connector_types: ["*"] as string[], // Default to wildcard
+    metadata: undefined as MetadataGroup | undefined,
   });
   const [resultLimit, setResultLimit] = useState(10);
   const [scoreThreshold, setScoreThreshold] = useState(0);
@@ -131,6 +136,7 @@ export function KnowledgeFilterPanel() {
         document_types: filters.document_types ?? ["*"],
         owners: filters.owners ?? ["*"],
         connector_types: filters.connector_types ?? ["*"],
+        metadata: filters.metadata,
       };
 
       setSelectedFilters(processedFilters);
@@ -154,6 +160,7 @@ export function KnowledgeFilterPanel() {
         document_types: filters.document_types ?? ["*"],
         owners: filters.owners ?? ["*"],
         connector_types: filters.connector_types ?? ["*"],
+        metadata: filters.metadata,
       });
       setResultLimit(parsedFilterData.limit || 10);
       setScoreThreshold(parsedFilterData.scoreThreshold || 0);
@@ -247,7 +254,7 @@ export function KnowledgeFilterPanel() {
   };
 
   const handleFilterChange = (
-    facetType: keyof typeof selectedFilters,
+    facetType: "data_sources" | "document_types" | "owners" | "connector_types",
     newValues: string[],
   ) => {
     setSelectedFilters((prev) => ({
@@ -326,6 +333,7 @@ export function KnowledgeFilterPanel() {
                 />
               </div>
             </div>
+
             {!createMode && selectedFilter?.created_at && (
               <div className="space-y-2 text-xs text-right text-muted-foreground">
                 <span className="text-placeholder-foreground">Created</span>{" "}
@@ -436,6 +444,16 @@ export function KnowledgeFilterPanel() {
               />
             </div>
 
+            <div className="space-y-2">
+              <Label>Custom metadata</Label>
+              <MetadataFilterBuilder
+                value={selectedFilters.metadata}
+                onChange={(metadata) =>
+                  setSelectedFilters((previous) => ({ ...previous, metadata }))
+                }
+              />
+            </div>
+
             {/* Result Limit Control - exactly like search page */}
             <div className="space-y-4">
               <div className="space-y-2">
diff --git a/frontend/components/metadata-filter-builder.tsx b/frontend/components/metadata-filter-builder.tsx
new file mode 100644
index 000000000..b436466c4
--- /dev/null
+++ b/frontend/components/metadata-filter-builder.tsx
@@ -0,0 +1,404 @@
+"use client";
+
+import { Plus, Trash2 } from "lucide-react";
+import { useEffect, useId, useState } from "react";
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+  SelectValue,
+} from "@/components/ui/select";
+
+export type MetadataOperator =
+  | "equals"
+  | "not_equals"
+  | "in"
+  | "not_in"
+  | "contains"
+  | "not_contains"
+  | "exists"
+  | "not_exists"
+  | "gt"
+  | "gte"
+  | "lt"
+  | "lte"
+  | "between";
+
+export interface MetadataCondition {
+  key: string;
+  operator: MetadataOperator;
+  value?: unknown;
+}
+
+export interface MetadataGroup {
+  op: "and" | "or";
+  conditions: Array<MetadataCondition | MetadataGroup>;
+}
+
+interface MetadataField {
+  key: string;
+  type: "string" | "number" | "date" | "boolean";
+}
+
+const isGroup = (
+  item: MetadataCondition | MetadataGroup,
+): item is MetadataGroup => "conditions" in item;
+
+const operatorsFor = (type: MetadataField["type"] | undefined) => {
+  const base: MetadataOperator[] = [
+    "equals",
+    "not_equals",
+    "in",
+    "not_in",
+    "exists",
+    "not_exists",
+  ];
+  if (type === "string") base.push("contains", "not_contains");
+  if (type === "number" || type === "date") {
+    base.push("gt", "gte", "lt", "lte", "between");
+  }
+  return base;
+};
+
+const scalarValue = (raw: string, type: MetadataField["type"] | undefined) =>
+  type === "number" ? Number(raw) : type === "boolean" ? raw === "true" : raw;
+
+function ConditionEditor({
+  condition,
+  fields,
+  onChange,
+  onRemove,
+}: {
+  condition: MetadataCondition;
+  fields: MetadataField[];
+  onChange: (condition: MetadataCondition) => void;
+  onRemove: () => void;
+}) {
+  const field = fields.find((candidate) => candidate.key === condition.key);
+  const valueListId = useId();
+  const [suggestedValues, setSuggestedValues] = useState<unknown[]>([]);
+  useEffect(() => {
+    if (!condition.key) return;
+    fetch(
+      `/api/metadata/fields/${encodeURIComponent(condition.key)}/values?limit=50`,
+    )
+      .then((response) => (response.ok ? response.json() : { values: [] }))
+      .then((result) =>
+        setSuggestedValues(
+          (result.values ?? []).map((item: { value: unknown }) => item.value),
+        ),
+      )
+      .catch(() => undefined);
+  }, [condition.key]);
+  const noValue =
+    condition.operator === "exists" || condition.operator === "not_exists";
+  const isList = condition.operator === "in" || condition.operator === "not_in";
+  const between = condition.operator === "between";
+  const inputType =
+    field?.type === "date"
+      ? "date"
+      : field?.type === "number"
+        ? "number"
+        : "text";
+
+  return (
+    <div className="grid grid-cols-[minmax(0,1fr)_150px_minmax(0,1fr)_36px] gap-2">
+      <Select
+        value={condition.key}
+        onValueChange={(key) =>
+          onChange({ key, operator: "equals", value: "" })
+        }
+      >
+        <SelectTrigger>
+          <SelectValue placeholder="Metadata field" />
+        </SelectTrigger>
+        <SelectContent>
+          {fields.map((item) => (
+            <SelectItem key={item.key} value={item.key}>
+              {item.key}
+            </SelectItem>
+          ))}
+        </SelectContent>
+      </Select>
+      <Select
+        value={condition.operator}
+        onValueChange={(operator: MetadataOperator) =>
+          onChange({
+            ...condition,
+            operator,
+            value:
+              operator === "between" ? { gte: "", lte: "" } : condition.value,
+          })
+        }
+      >
+        <SelectTrigger>
+          <SelectValue />
+        </SelectTrigger>
+        <SelectContent>
+          {operatorsFor(field?.type).map((operator) => (
+            <SelectItem key={operator} value={operator}>
+              {operator.replaceAll("_", " ")}
+            </SelectItem>
+          ))}
+        </SelectContent>
+      </Select>
+      {noValue ? (
+        <div />
+      ) : between ? (
+        <div className="flex gap-2">
+          <Input
+            aria-label="Lower bound"
+            type={inputType}
+            value={String((condition.value as { gte?: unknown })?.gte ?? "")}
+            onChange={(event) =>
+              onChange({
+                ...condition,
+                value: {
+                  ...(condition.value as object),
+                  gte: scalarValue(event.target.value, field?.type),
+                },
+              })
+            }
+          />
+          <Input
+            aria-label="Upper bound"
+            type={inputType}
+            value={String((condition.value as { lte?: unknown })?.lte ?? "")}
+            onChange={(event) =>
+              onChange({
+                ...condition,
+                value: {
+                  ...(condition.value as object),
+                  lte: scalarValue(event.target.value, field?.type),
+                },
+              })
+            }
+          />
+        </div>
+      ) : field?.type === "boolean" ? (
+        <Select
+          value={String(condition.value ?? false)}
+          onValueChange={(value) =>
+            onChange({ ...condition, value: value === "true" })
+          }
+        >
+          <SelectTrigger>
+            <SelectValue />
+          </SelectTrigger>
+          <SelectContent>
+            <SelectItem value="true">True</SelectItem>
+            <SelectItem value="false">False</SelectItem>
+          </SelectContent>
+        </Select>
+      ) : (
+        <>
+          <datalist id={valueListId}>
+            {suggestedValues.map((value) => (
+              <option key={String(value)} value={String(value)} />
+            ))}
+          </datalist>
+          <Input
+            aria-label="Filter value"
+            list={field?.type === "string" ? valueListId : undefined}
+            placeholder={isList ? "value one, value two" : "Value"}
+            type={isList ? "text" : inputType}
+            value={
+              Array.isArray(condition.value)
+                ? condition.value.join(", ")
+                : String(condition.value ?? "")
+            }
+            onChange={(event) =>
+              onChange({
+                ...condition,
+                value: isList
+                  ? event.target.value
+                      .split(",")
+                      .map((item) => scalarValue(item.trim(), field?.type))
+                      .filter((item) => item !== "")
+                  : scalarValue(event.target.value, field?.type),
+              })
+            }
+          />
+        </>
+      )}
+      <Button
+        aria-label="Remove condition"
+        onClick={onRemove}
+        size="icon"
+        type="button"
+        variant="ghost"
+      >
+        <Trash2 className="h-4 w-4" />
+      </Button>
+    </div>
+  );
+}
+
+function GroupEditor({
+  group,
+  fields,
+  onChange,
+  removable,
+  onRemove,
+}: {
+  group: MetadataGroup;
+  fields: MetadataField[];
+  onChange: (group: MetadataGroup) => void;
+  removable?: boolean;
+  onRemove?: () => void;
+}) {
+  const replace = (index: number, item: MetadataCondition | MetadataGroup) => {
+    const conditions = [...group.conditions];
+    conditions[index] = item;
+    onChange({ ...group, conditions });
+  };
+  return (
+    <div className="space-y-3 rounded-lg border p-3">
+      <div className="flex items-center gap-2">
+        <span className="text-sm text-muted-foreground">Match</span>
+        <Select
+          value={group.op}
+          onValueChange={(op: "and" | "or") => onChange({ ...group, op })}
+        >
+          <SelectTrigger className="w-24">
+            <SelectValue />
+          </SelectTrigger>
+          <SelectContent>
+            <SelectItem value="and">All</SelectItem>
+            <SelectItem value="or">Any</SelectItem>
+          </SelectContent>
+        </Select>
+        <span className="text-sm text-muted-foreground">conditions</span>
+        {removable && (
+          <Button
+            className="ml-auto"
+            onClick={onRemove}
+            size="sm"
+            type="button"
+            variant="ghost"
+          >
+            Remove group
+          </Button>
+        )}
+      </div>
+      {group.conditions.map((item, index) =>
+        isGroup(item) ? (
+          <GroupEditor
+            fields={fields}
+            group={item}
+            // biome-ignore lint/suspicious/noArrayIndexKey: expression nodes have no persisted UI identifier
+            key={index}
+            onChange={(next) => replace(index, next)}
+            onRemove={() =>
+              onChange({
+                ...group,
+                conditions: group.conditions.filter(
+                  (_, current) => current !== index,
+                ),
+              })
+            }
+            removable
+          />
+        ) : (
+          <ConditionEditor
+            condition={item}
+            fields={fields}
+            // biome-ignore lint/suspicious/noArrayIndexKey: expression nodes have no persisted UI identifier
+            key={index}
+            onChange={(next) => replace(index, next)}
+            onRemove={() =>
+              onChange({
+                ...group,
+                conditions: group.conditions.filter(
+                  (_, current) => current !== index,
+                ),
+              })
+            }
+          />
+        ),
+      )}
+      <div className="flex gap-2">
+        <Button
+          disabled={!fields.length}
+          onClick={() =>
+            onChange({
+              ...group,
+              conditions: [
+                ...group.conditions,
+                { key: fields[0]?.key ?? "", operator: "equals", value: "" },
+              ],
+            })
+          }
+          size="sm"
+          type="button"
+          variant="outline"
+        >
+          <Plus className="h-4 w-4" /> Condition
+        </Button>
+        <Button
+          onClick={() =>
+            onChange({
+              ...group,
+              conditions: [...group.conditions, { op: "and", conditions: [] }],
+            })
+          }
+          size="sm"
+          type="button"
+          variant="outline"
+        >
+          <Plus className="h-4 w-4" /> Group
+        </Button>
+      </div>
+    </div>
+  );
+}
+
+export function MetadataFilterBuilder({
+  value,
+  onChange,
+}: {
+  value?: MetadataGroup;
+  onChange: (value: MetadataGroup | undefined) => void;
+}) {
+  const [fields, setFields] = useState<MetadataField[]>([]);
+  useEffect(() => {
+    fetch("/api/metadata/fields")
+      .then((response) => (response.ok ? response.json() : { fields: [] }))
+      .then((result) =>
+        setFields(
+          (result.fields ?? []).filter((field: MetadataField) => !!field.type),
+        ),
+      )
+      .catch(() => undefined);
+  }, []);
+
+  if (!value) {
+    return (
+      <Button
+        disabled={!fields.length}
+        onClick={() => onChange({ op: "and", conditions: [] })}
+        size="sm"
+        type="button"
+        variant="outline"
+      >
+        <Plus className="h-4 w-4" /> Add metadata filters
+      </Button>
+    );
+  }
+  return (
+    <div className="space-y-2">
+      <GroupEditor fields={fields} group={value} onChange={onChange} />
+      <Button
+        onClick={() => onChange(undefined)}
+        size="sm"
+        type="button"
+        variant="ghost"
+      >
+        Clear metadata filters
+      </Button>
+    </div>
+  );
+}
diff --git a/frontend/contexts/knowledge-filter-context.tsx b/frontend/contexts/knowledge-filter-context.tsx
index a538484af..3aeaa7e8e 100644
--- a/frontend/contexts/knowledge-filter-context.tsx
+++ b/frontend/contexts/knowledge-filter-context.tsx
@@ -8,6 +8,7 @@ import React, {
   useState,
 } from "react";
 import { FilterColor, IconKey } from "@/components/filter-icon-popover";
+import type { MetadataGroup } from "@/components/metadata-filter-builder";
 
 interface KnowledgeFilter {
   id: string;
@@ -26,6 +27,7 @@ export interface ParsedQueryData {
     document_types: string[];
     owners: string[];
     connector_types: string[];
+    metadata?: MetadataGroup;
   };
   limit: number;
   scoreThreshold: number;
@@ -103,6 +105,7 @@ export function KnowledgeFilterProvider({
             document_types: raw.filters?.document_types ?? ["*"],
             owners: raw.filters?.owners ?? ["*"],
             connector_types: raw.filters?.connector_types ?? ["*"],
+            metadata: raw.filters?.metadata,
           },
           limit: raw.limit ?? 10,
           scoreThreshold: raw.scoreThreshold ?? 0,
diff --git a/frontend/lib/filter-normalization.ts b/frontend/lib/filter-normalization.ts
index 59427f288..818add639 100644
--- a/frontend/lib/filter-normalization.ts
+++ b/frontend/lib/filter-normalization.ts
@@ -1,8 +1,11 @@
+import type { MetadataGroup } from "@/components/metadata-filter-builder";
+
 export interface FilterInput {
   data_sources?: string[];
   document_types?: string[];
   owners?: string[];
   connector_types?: string[];
+  metadata?: MetadataGroup;
 }
 
 export interface NormalizedSelectedFilters {
@@ -10,6 +13,7 @@ export interface NormalizedSelectedFilters {
   document_types: string[];
   owners: string[];
   connector_types: string[];
+  metadata?: MetadataGroup;
 }
 
 function normalizeFilterDimension(values?: string[]): string[] {
@@ -27,6 +31,7 @@ function normalizeSelectedFilters(
     document_types: normalizeFilterDimension(filters?.document_types),
     owners: normalizeFilterDimension(filters?.owners),
     connector_types: normalizeFilterDimension(filters?.connector_types),
+    metadata: filters?.metadata,
   };
 }
 
@@ -48,6 +53,9 @@ export function buildSearchPayloadFilters(
   if (normalized.connector_types.length > 0) {
     payloadFilters.connector_types = normalized.connector_types;
   }
+  if (normalized.metadata?.conditions.length) {
+    payloadFilters.metadata = normalized.metadata;
+  }
 
   return Object.keys(payloadFilters).length > 0 ? payloadFilters : undefined;
 }
diff --git a/frontend/lib/upload-utils.ts b/frontend/lib/upload-utils.ts
index 2793a43cc..4cbaaad48 100644
--- a/frontend/lib/upload-utils.ts
+++ b/frontend/lib/upload-utils.ts
@@ -82,12 +82,14 @@ export async function duplicateCheck(
 export async function uploadFiles(
   files: File[],
   replace = false,
+  metadata?: unknown[],
 ): Promise<{ taskId: string; fileCount: number }> {
   const formData = new FormData();
   for (const file of files) {
     formData.append("file", file);
   }
   formData.append("replace_duplicates", replace.toString());
+  if (metadata?.length) formData.append("metadata", JSON.stringify(metadata));
 
   const uploadResponse = await fetch("/api/router/upload_ingest", {
     method: "POST",
@@ -129,6 +131,7 @@ export async function uploadFile(
   replace = false,
   createFilter = false,
   callbacks?: UploadFileCallbacks,
+  metadata?: unknown[],
 ): Promise<UploadFileResult> {
   try {
     const formData = new FormData();
@@ -137,6 +140,7 @@ export async function uploadFile(
     if (createFilter) {
       formData.append("create_filter", "true");
     }
+    if (metadata?.length) formData.append("metadata", JSON.stringify(metadata));
 
     const uploadResponse = await fetch("/api/router/upload_ingest", {
       method: "POST",
diff --git a/sdks/mcp/README.md b/sdks/mcp/README.md
index 85746ea60..f215414ef 100644
--- a/sdks/mcp/README.md
+++ b/sdks/mcp/README.md
@@ -108,6 +108,8 @@ All tools are auto-exposed from the `/v1/` API, and they are available immediate
 | `openrag_get_chat` | Get a specific chat conversation by ID. |
 | `openrag_delete_chat` | Delete a chat conversation by ID. |
 | `openrag_search` | Semantic search over the knowledge base. Supports filters, score threshold, data sources. |
+| `openrag_list_metadata_fields` | List typed custom metadata fields visible to the caller. |
+| `openrag_list_metadata_values` | List suggested values for a visible custom metadata field. |
 | `openrag_ingest` | Ingest documents (files, URLs, text) into the knowledge base. Returns a `task_id`. |
 | `openrag_get_task_status` | Check the status of an ingestion task by `task_id`. |
 | `openrag_delete_document` | Delete a document from the knowledge base by filename. |
@@ -122,4 +124,4 @@ All tools are auto-exposed from the `/v1/` API, and they are available immediate
 
 ## License
 
-Apache 2.0 — see [LICENSE](../../LICENSE) for details.
\ No newline at end of file
+Apache 2.0 — see [LICENSE](../../LICENSE) for details.
diff --git a/sdks/python/README.md b/sdks/python/README.md
index d3c0b11ad..ce2159196 100644
--- a/sdks/python/README.md
+++ b/sdks/python/README.md
@@ -132,13 +132,22 @@ for result in results.results:
     print(f"  {result.text[:100]}...")
 
 # Search with filters
-from openrag_sdk import SearchFilters
+from openrag_sdk import MetadataFilterCondition, MetadataFilterGroup, SearchFilters
 
 results = await client.search.query(
     "API documentation",
     filters=SearchFilters(
         data_sources=["api-docs.pdf"],
-        document_types=["application/pdf"]
+        document_types=["application/pdf"],
+        metadata=MetadataFilterGroup(
+            op="and",
+            conditions=[
+                MetadataFilterCondition(key="supplier", operator="equals", value="Dell"),
+                MetadataFilterCondition(
+                    key="contract_end", operator="gte", value="2026-01-01"
+                ),
+            ],
+        ),
     ),
     limit=5,
     score_threshold=0.5
@@ -148,6 +157,17 @@ results = await client.search.query(
 ## Documents
 
 ```python
+from openrag_sdk import MetadataEntry
+
+# Typed metadata is available to search, chat, saved filters, and MCP.
+result = await client.documents.ingest(
+    file_path="./dell-contract.pdf",
+    metadata=[
+        MetadataEntry(key="supplier", type="string", value="Dell"),
+        MetadataEntry(key="contract_end", type="date", value="2026-12-31"),
+    ],
+)
+
 # Ingest a file (waits for completion by default)
 result = await client.documents.ingest(file_path="./report.pdf")
 print(f"Status: {result.status}")
diff --git a/sdks/python/openrag_sdk/__init__.py b/sdks/python/openrag_sdk/__init__.py
index 86bddffc2..f4b36bf8b 100644
--- a/sdks/python/openrag_sdk/__init__.py
+++ b/sdks/python/openrag_sdk/__init__.py
@@ -56,6 +56,9 @@
     KnowledgeFilterSearchResponse,
     KnowledgeSettings,
     Message,
+    MetadataEntry,
+    MetadataFilterCondition,
+    MetadataFilterGroup,
     SearchFilters,
     SearchResponse,
     SearchResult,
@@ -98,6 +101,9 @@
     "ConversationDetail",
     "ConversationListResponse",
     "Message",
+    "MetadataEntry",
+    "MetadataFilterCondition",
+    "MetadataFilterGroup",
     "SettingsResponse",
     "SettingsUpdateOptions",
     "SettingsUpdateResponse",
diff --git a/sdks/python/openrag_sdk/documents.py b/sdks/python/openrag_sdk/documents.py
index 414144c3d..398da880f 100644
--- a/sdks/python/openrag_sdk/documents.py
+++ b/sdks/python/openrag_sdk/documents.py
@@ -1,11 +1,17 @@
 """OpenRAG SDK documents client."""
 
 import asyncio
+import json
 from pathlib import Path
 from typing import TYPE_CHECKING, BinaryIO
 
 from .exceptions import NotFoundError
-from .models import DeleteDocumentResponse, IngestResponse, IngestTaskStatus
+from .models import (
+    DeleteDocumentResponse,
+    IngestResponse,
+    IngestTaskStatus,
+    MetadataEntry,
+)
 
 if TYPE_CHECKING:
     from .client import OpenRAGClient
@@ -26,6 +32,7 @@ async def ingest(
         wait: bool = True,
         poll_interval: float = 1.0,
         timeout: float = 300.0,
+        metadata: list[MetadataEntry | dict] | None = None,
     ) -> IngestResponse | IngestTaskStatus:
         """
         Ingest a document into the knowledge base.
@@ -37,6 +44,7 @@ async def ingest(
             wait: If True, poll until ingestion completes. If False, return immediately.
             poll_interval: Seconds between status checks when waiting.
             timeout: Maximum seconds to wait for completion.
+            metadata: Typed custom metadata applied to the document.
 
         Returns:
             IngestTaskStatus with final status if wait=True.
@@ -46,6 +54,9 @@ async def ingest(
             ValueError: If neither file_path nor file is provided.
             TimeoutError: If ingestion doesn't complete within timeout.
         """
+        form_data = (
+            {"metadata": self._serialize_metadata(metadata)} if metadata else None
+        )
         if file_path is not None:
             path = Path(file_path)
             with open(path, "rb") as f:
@@ -54,6 +65,7 @@ async def ingest(
                     "POST",
                     "/api/v1/documents/ingest",
                     files=files,
+                    data=form_data,
                 )
         elif file is not None:
             if filename is None:
@@ -63,6 +75,7 @@ async def ingest(
                 "POST",
                 "/api/v1/documents/ingest",
                 files=files,
+                data=form_data,
             )
         else:
             raise ValueError("Either file_path or file must be provided")
@@ -80,6 +93,17 @@ async def ingest(
             timeout=timeout,
         )
 
+    @staticmethod
+    def _serialize_metadata(metadata: list[MetadataEntry | dict]) -> str:
+        return json.dumps(
+            [
+                item.model_dump(mode="json")
+                if isinstance(item, MetadataEntry)
+                else item
+                for item in metadata
+            ]
+        )
+
     async def get_task_status(self, task_id: str) -> IngestTaskStatus:
         """
         Get the status of an ingestion task.
diff --git a/sdks/python/openrag_sdk/models.py b/sdks/python/openrag_sdk/models.py
index b0eee0864..4217f596e 100644
--- a/sdks/python/openrag_sdk/models.py
+++ b/sdks/python/openrag_sdk/models.py
@@ -1,6 +1,8 @@
 """OpenRAG SDK data models."""
 
-from typing import Literal
+from __future__ import annotations
+
+from typing import Any, Literal
 
 from pydantic import BaseModel, Field
 
@@ -14,6 +16,7 @@ class Source(BaseModel):
     score: float
     page: int | None = None
     mimetype: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 class ChatResponse(BaseModel):
@@ -60,6 +63,7 @@ class SearchResult(BaseModel):
     score: float
     page: int | None = None
     mimetype: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 class SearchResponse(BaseModel):
@@ -168,6 +172,46 @@ class SearchFilters(BaseModel):
 
     data_sources: list[str] | None = None
     document_types: list[str] | None = None
+    owners: list[str] | None = None
+    connector_types: list[str] | None = None
+    metadata: MetadataFilterGroup | None = None
+
+
+class MetadataFilterCondition(BaseModel):
+    """One typed custom metadata comparison."""
+
+    key: str
+    operator: Literal[
+        "equals",
+        "not_equals",
+        "in",
+        "not_in",
+        "contains",
+        "not_contains",
+        "exists",
+        "not_exists",
+        "gt",
+        "gte",
+        "lt",
+        "lte",
+        "between",
+    ]
+    value: Any | None = None
+
+
+class MetadataFilterGroup(BaseModel):
+    """Nested AND/OR expression over custom metadata conditions."""
+
+    op: Literal["and", "or"] = "and"
+    conditions: list[MetadataFilterCondition | MetadataFilterGroup]
+
+
+class MetadataEntry(BaseModel):
+    """Typed custom metadata attached to an ingested document."""
+
+    key: str
+    type: Literal["string", "number", "date", "boolean"]
+    value: str | int | float | bool | list[str | int | float | bool]
 
 
 # Settings update models
@@ -213,7 +257,7 @@ class KnowledgeFilterQueryData(BaseModel):
     """Query configuration stored in a knowledge filter."""
 
     query: str | None = None
-    filters: dict[str, list[str]] | None = None
+    filters: dict[str, Any] | None = None
     limit: int | None = None
     score_threshold: float | None = Field(default=None, alias="scoreThreshold")
     color: str | None = None
diff --git a/sdks/typescript/README.md b/sdks/typescript/README.md
index 6488d1f72..0b9251ed5 100644
--- a/sdks/typescript/README.md
+++ b/sdks/typescript/README.md
@@ -161,6 +161,13 @@ const results = await client.search.query("API documentation", {
   filters: {
     data_sources: ["api-docs.pdf"],
     document_types: ["application/pdf"],
+    metadata: {
+      op: "and",
+      conditions: [
+        { key: "supplier", operator: "equals", value: "Dell" },
+        { key: "contract_end", operator: "gte", value: "2026-01-01" },
+      ],
+    },
   },
   limit: 5,
   scoreThreshold: 0.5,
@@ -170,6 +177,15 @@ const results = await client.search.query("API documentation", {
 ## Documents
 
 ```typescript
+// Typed metadata is available to search, chat, saved filters, and MCP.
+await client.documents.ingest({
+  filePath: "./dell-contract.pdf",
+  metadata: [
+    { key: "supplier", type: "string", value: "Dell" },
+    { key: "contract_end", type: "date", value: "2026-12-31" },
+  ],
+});
+
 // Ingest a file (waits for completion by default)
 const result = await client.documents.ingest({
   filePath: "./report.pdf",
diff --git a/sdks/typescript/src/documents.ts b/sdks/typescript/src/documents.ts
index 274fce9a2..047cf038d 100644
--- a/sdks/typescript/src/documents.ts
+++ b/sdks/typescript/src/documents.ts
@@ -9,6 +9,7 @@ import type {
   IngestResponse,
   IngestTaskStatus,
   NotFoundError,
+  MetadataEntry,
 } from "./types";
 
 export interface IngestOptions {
@@ -24,6 +25,8 @@ export interface IngestOptions {
   pollInterval?: number;
   /** Maximum seconds to wait for completion. Default: 300. */
   timeout?: number;
+  /** Typed custom metadata applied to the document. */
+  metadata?: MetadataEntry[];
 }
 
 export class DocumentsClient {
@@ -63,6 +66,9 @@ export class DocumentsClient {
     } else {
       throw new Error("Either filePath or file must be provided");
     }
+    if (options.metadata?.length) {
+      formData.append("metadata", JSON.stringify(options.metadata));
+    }
 
     const response = await this.client._request(
       "POST",
diff --git a/sdks/typescript/src/index.ts b/sdks/typescript/src/index.ts
index 3283f118c..c93daefc6 100644
--- a/sdks/typescript/src/index.ts
+++ b/sdks/typescript/src/index.ts
@@ -52,6 +52,11 @@ export {
   ChatCreateOptions,
   SearchQueryOptions,
   SearchFilters,
+  MetadataType,
+  MetadataValue,
+  MetadataEntry,
+  MetadataCondition,
+  MetadataGroup,
   // Chat types
   ChatResponse,
   StreamEvent,
diff --git a/sdks/typescript/src/types.ts b/sdks/typescript/src/types.ts
index 2d2336ee8..86c31241f 100644
--- a/sdks/typescript/src/types.ts
+++ b/sdks/typescript/src/types.ts
@@ -9,6 +9,7 @@ export interface Source {
   score: number;
   page?: number | null;
   mimetype?: string | null;
+  metadata?: Record<string, { type: MetadataType; value: MetadataValue }>;
 }
 
 export interface ChatResponse {
@@ -43,6 +44,7 @@ export interface SearchResult {
   score: number;
   page?: number | null;
   mimetype?: string | null;
+  metadata?: Record<string, { type: MetadataType; value: MetadataValue }>;
 }
 
 export interface SearchResponse {
@@ -52,6 +54,42 @@ export interface SearchResponse {
 export interface SearchFilters {
   data_sources?: string[];
   document_types?: string[];
+  owners?: string[];
+  connector_types?: string[];
+  metadata?: MetadataGroup;
+}
+
+export type MetadataType = "string" | "number" | "date" | "boolean";
+export type MetadataValue = string | number | boolean | Array<string | number | boolean>;
+
+export interface MetadataEntry {
+  key: string;
+  type: MetadataType;
+  value: MetadataValue;
+}
+
+export interface MetadataCondition {
+  key: string;
+  operator:
+    | "equals"
+    | "not_equals"
+    | "in"
+    | "not_in"
+    | "contains"
+    | "not_contains"
+    | "exists"
+    | "not_exists"
+    | "gt"
+    | "gte"
+    | "lt"
+    | "lte"
+    | "between";
+  value?: unknown;
+}
+
+export interface MetadataGroup {
+  op: "and" | "or";
+  conditions: Array<MetadataCondition | MetadataGroup>;
 }
 
 // Document types
@@ -169,6 +207,7 @@ export interface KnowledgeFilterQueryData {
     document_types?: string[];
     owners?: string[];
     connector_types?: string[];
+    metadata?: MetadataGroup;
   };
   /** Maximum number of results. */
   limit?: number;
diff --git a/src/api/connectors.py b/src/api/connectors.py
index 7251537a3..f6887846f 100644
--- a/src/api/connectors.py
+++ b/src/api/connectors.py
@@ -915,6 +915,15 @@ async def connector_sync(
     if denied := await _connector_access_denied(request, session, connector_type):
         return denied
 
+    metadata_entries = body.settings.get("metadata") if isinstance(body.settings, dict) else None
+    if metadata_entries is not None:
+        try:
+            from services.custom_metadata_service import CustomMetadataService
+
+            CustomMetadataService().normalize_entries(metadata_entries)
+        except ValueError as exc:
+            return JSONResponse({"error": f"Invalid metadata: {exc}"}, status_code=400)
+
     max_files = body.max_files
     selected_files_raw = body.selected_files
     selected_files = None
diff --git a/src/api/custom_metadata.py b/src/api/custom_metadata.py
new file mode 100644
index 000000000..26aeadcc5
--- /dev/null
+++ b/src/api/custom_metadata.py
@@ -0,0 +1,128 @@
+"""Custom metadata discovery endpoints scoped by document visibility."""
+
+import re
+
+from fastapi import Depends, Query
+from fastapi.responses import JSONResponse
+
+from config.settings import get_index_name
+from dependencies import (
+    get_session_manager,
+    require_api_key_permission,
+    require_permission,
+)
+from services.custom_metadata_service import CustomMetadataService
+from session_manager import User
+
+
+async def _fields(user: User, session_manager) -> JSONResponse:
+    client = session_manager.get_user_opensearch_client(user.user_id, user.jwt_token)
+    result = await client.search(
+        index=get_index_name(),
+        body={
+            "size": 0,
+            "aggs": {
+                "metadata": {
+                    "nested": {"path": "metadata_entries"},
+                    "aggs": {"keys": {"terms": {"field": "metadata_entries.key", "size": 200}}},
+                }
+            },
+        },
+    )
+    field_types = await CustomMetadataService().get_field_types()
+    buckets = result.get("aggregations", {}).get("metadata", {}).get("keys", {}).get("buckets", [])
+    return JSONResponse(
+        {
+            "fields": [
+                {
+                    "key": bucket["key"],
+                    "type": field_types.get(bucket["key"]),
+                    "document_count": bucket.get("doc_count", 0),
+                }
+                for bucket in buckets
+                if bucket.get("key") in field_types
+            ]
+        }
+    )
+
+
+async def _values(key: str, query: str, limit: int, user: User, session_manager) -> JSONResponse:
+    key = key.strip().lower()
+    field_types = await CustomMetadataService().get_field_types()
+    metadata_type = field_types.get(key)
+    if metadata_type is None:
+        return JSONResponse({"error": f"Unknown custom metadata key '{key}'"}, status_code=404)
+    value_field = CustomMetadataService.VALUE_FIELDS[metadata_type]
+    include = f".*{re.escape(query)}.*" if query else None
+    terms: dict = {"field": f"metadata_entries.{value_field}", "size": min(limit, 100)}
+    if include and metadata_type == "string":
+        terms["include"] = include
+    client = session_manager.get_user_opensearch_client(user.user_id, user.jwt_token)
+    result = await client.search(
+        index=get_index_name(),
+        body={
+            "size": 0,
+            "aggs": {
+                "metadata": {
+                    "nested": {"path": "metadata_entries"},
+                    "aggs": {
+                        "selected_key": {
+                            "filter": {"term": {"metadata_entries.key": key}},
+                            "aggs": {"values": {"terms": terms}},
+                        }
+                    },
+                }
+            },
+        },
+    )
+    buckets = (
+        result.get("aggregations", {})
+        .get("metadata", {})
+        .get("selected_key", {})
+        .get("values", {})
+        .get("buckets", [])
+    )
+    return JSONResponse(
+        {
+            "key": key,
+            "type": metadata_type,
+            "values": [
+                {"value": bucket.get("key"), "document_count": bucket.get("doc_count", 0)}
+                for bucket in buckets
+            ],
+        }
+    )
+
+
+async def list_fields(
+    session_manager=Depends(get_session_manager),
+    user: User = Depends(require_permission("search:use")),
+):
+    return await _fields(user, session_manager)
+
+
+async def list_values(
+    key: str,
+    query: str = Query(""),
+    limit: int = Query(20, ge=1, le=100),
+    session_manager=Depends(get_session_manager),
+    user: User = Depends(require_permission("search:use")),
+):
+    return await _values(key, query, limit, user, session_manager)
+
+
+async def list_fields_v1(
+    session_manager=Depends(get_session_manager),
+    user: User = Depends(require_api_key_permission("search:use")),
+):
+    return await _fields(user, session_manager)
+
+
+async def list_values_v1(
+    key: str,
+    query: str = Query(""),
+    limit: int = Query(20, ge=1, le=100),
+    session_manager=Depends(get_session_manager),
+    user: User = Depends(require_api_key_permission("search:use")),
+):
+    return await _values(key, query, limit, user, session_manager)
diff --git a/src/api/knowledge_filter.py b/src/api/knowledge_filter.py
index 1f9e1588d..9f95d2b70 100644
--- a/src/api/knowledge_filter.py
+++ b/src/api/knowledge_filter.py
@@ -60,6 +60,8 @@ def normalize_query_data(query_data: str | dict) -> str:
         "owners": filters.get("owners", ["*"]),
         "connector_types": filters.get("connector_types", ["*"]),
     }
+    if isinstance(filters.get("metadata"), dict):
+        normalized_filters["metadata"] = filters["metadata"]
 
     normalized = {
         "query": data.get("query", ""),
@@ -73,6 +75,17 @@ def normalize_query_data(query_data: str | dict) -> str:
     return json.dumps(normalized)
 
 
+async def validate_metadata_query_data(normalized_query_data: str) -> None:
+    data = json.loads(normalized_query_data)
+    expression = (data.get("filters") or {}).get("metadata")
+    if not expression:
+        return
+    from services.custom_metadata_service import CustomMetadataService
+
+    service = CustomMetadataService()
+    service.compile_expression(expression, await service.get_field_types())
+
+
 class CreateFilterBody(BaseModel):
     name: str
     description: str = ""
@@ -113,6 +126,7 @@ async def create_knowledge_filter(
 
     try:
         normalized_query_data = normalize_query_data(body.queryData)
+        await validate_metadata_query_data(normalized_query_data)
     except Exception as e:
         logger.error(f"Failed to normalize query_data: {e}")
         return JSONResponse({"error": f"Invalid queryData format: {str(e)}"}, status_code=400)
@@ -195,6 +209,14 @@ async def update_knowledge_filter(
 
     existing_filter = existing_result["filter"]
 
+    query_data = body.queryData if body.queryData is not None else existing_filter["query_data"]
+    try:
+        normalized_query_data = normalize_query_data(query_data)
+        await validate_metadata_query_data(normalized_query_data)
+    except Exception as e:
+        logger.error(f"Failed to normalize query_data: {e}")
+        return JSONResponse({"error": f"Invalid queryData format: {str(e)}"}, status_code=400)
+
     delete_result = await knowledge_filter_service.delete_knowledge_filter(
         filter_id, user_id=user.user_id, jwt_token=jwt_token
     )
@@ -203,13 +225,6 @@ async def update_knowledge_filter(
             {"error": "Failed to delete existing knowledge filter"}, status_code=500
         )
 
-    query_data = body.queryData if body.queryData is not None else existing_filter["query_data"]
-    try:
-        normalized_query_data = normalize_query_data(query_data)
-    except Exception as e:
-        logger.error(f"Failed to normalize query_data: {e}")
-        return JSONResponse({"error": f"Invalid queryData format: {str(e)}"}, status_code=400)
-
     updated_filter = {
         "id": filter_id,
         "name": body.name if body.name is not None else existing_filter["name"],
diff --git a/src/api/router.py b/src/api/router.py
index 8b76ba27a..3bd49fef9 100644
--- a/src/api/router.py
+++ b/src/api/router.py
@@ -31,6 +31,7 @@ async def upload_ingest_router(
     replace_duplicates: str = Form("true"),
     create_filter: str = Form("false"),
     preview: str = Form("false"),
+    metadata_json: str | None = Form(None, alias="metadata"),
     document_service=Depends(get_document_service),
     langflow_file_service=Depends(get_langflow_file_service),
     session_manager=Depends(get_session_manager),
@@ -54,6 +55,19 @@ async def upload_ingest_router(
         preview_mode=preview_mode,
     )
 
+    metadata_entries = None
+    if isinstance(metadata_json, str) and metadata_json:
+        try:
+            parsed_metadata = json.loads(metadata_json)
+            if not isinstance(parsed_metadata, list):
+                raise ValueError("metadata must be a JSON array of typed key/value entries")
+            from services.custom_metadata_service import CustomMetadataService
+
+            CustomMetadataService().normalize_entries(parsed_metadata)
+            metadata_entries = parsed_metadata
+        except (json.JSONDecodeError, ValueError) as e:
+            return JSONResponse({"error": f"Invalid metadata: {e}"}, status_code=400)
+
     if disable_ingest_with_langflow:
         logger.debug("Routing to traditional OpenRAG upload via task service")
         return await _traditional_upload_ingest_task(
@@ -65,6 +79,7 @@ async def upload_ingest_router(
             task_service=task_service,
             user=user,
             settings_json=settings_json,
+            metadata_entries=metadata_entries,
         )
 
     logger.debug("Routing to Langflow upload-ingest pipeline via task service")
@@ -80,6 +95,7 @@ async def upload_ingest_router(
         session_manager=session_manager,
         task_service=task_service,
         user=user,
+        metadata_entries=metadata_entries,
     )
 
 
@@ -92,6 +108,7 @@ async def _traditional_upload_ingest_task(
     task_service,
     user: User,
     settings_json: str | None = None,
+    metadata_entries: list[dict] | None = None,
 ):
     """Task-based traditional upload and ingest for single/multiple files"""
     try:
@@ -104,6 +121,9 @@ async def _traditional_upload_ingest_task(
                 settings = json.loads(settings_json)
             except json.JSONDecodeError as e:
                 return JSONResponse({"error": f"Invalid settings JSON: {e}"}, status_code=400)
+        if metadata_entries is not None:
+            settings = dict(settings or {})
+            settings["metadata"] = metadata_entries
 
         user_id = user.user_id
         user_name = user.name
@@ -197,6 +217,7 @@ async def _langflow_upload_ingest_task(
     session_manager,
     task_service,
     user: User,
+    metadata_entries: list[dict] | None = None,
 ):
     """Task-based langflow upload and ingest for single/multiple files"""
     try:
@@ -217,6 +238,9 @@ async def _langflow_upload_ingest_task(
                 tweaks = json.loads(tweaks_json)
             except json.JSONDecodeError as e:
                 return JSONResponse({"error": f"Invalid tweaks JSON: {e}"}, status_code=400)
+        if metadata_entries is not None:
+            settings = dict(settings or {})
+            settings["metadata"] = metadata_entries
 
         user_id = user.user_id
         user_name = user.name
diff --git a/src/api/schemas/custom_metadata.py b/src/api/schemas/custom_metadata.py
new file mode 100644
index 000000000..a1becfb53
--- /dev/null
+++ b/src/api/schemas/custom_metadata.py
@@ -0,0 +1,51 @@
+"""Public request models for typed custom metadata filters."""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+
+class MetadataFilterCondition(BaseModel):
+    key: str = Field(pattern=r"^[a-z][a-z0-9_]{0,63}$")
+    operator: Literal[
+        "equals",
+        "not_equals",
+        "in",
+        "not_in",
+        "contains",
+        "not_contains",
+        "exists",
+        "not_exists",
+        "gt",
+        "gte",
+        "lt",
+        "lte",
+        "between",
+    ]
+    value: Any | None = None
+
+
+class MetadataFilterGroup(BaseModel):
+    op: Literal["and", "or"] = "and"
+    conditions: list[MetadataFilterCondition | MetadataFilterGroup] = Field(
+        min_length=1,
+        max_length=50,
+    )
+
+
+class SearchFiltersRequest(BaseModel):
+    data_sources: list[str] | None = None
+    document_types: list[str] | None = None
+    owners: list[str] | None = None
+    connector_types: list[str] | None = None
+    metadata: MetadataFilterGroup | None = None
+
+    model_config = {"extra": "forbid"}
+
+
+def dump_search_filters(filters: SearchFiltersRequest | dict[str, Any] | None):
+    if isinstance(filters, SearchFiltersRequest):
+        return filters.model_dump(exclude_none=True)
+    return filters
diff --git a/src/api/v1/_filter_resolution.py b/src/api/v1/_filter_resolution.py
index 417e306ca..33f11db00 100644
--- a/src/api/v1/_filter_resolution.py
+++ b/src/api/v1/_filter_resolution.py
@@ -17,11 +17,11 @@
 _FILTER_DIMENSIONS = ("data_sources", "document_types", "owners", "connector_types")
 
 
-def _strip_wildcards(filters: dict[str, Any] | None) -> dict[str, list[str]]:
+def _strip_wildcards(filters: dict[str, Any] | None) -> dict[str, Any]:
     """Keep only filter dimensions that contain concrete values."""
     if not filters:
         return {}
-    cleaned: dict[str, list[str]] = {}
+    cleaned: dict[str, Any] = {}
     for key in _FILTER_DIMENSIONS:
         values = filters.get(key)
         if not values or not isinstance(values, list):
@@ -29,6 +29,9 @@ def _strip_wildcards(filters: dict[str, Any] | None) -> dict[str, list[str]]:
         if "*" in values:
             continue
         cleaned[key] = values
+    metadata_expression = filters.get("metadata")
+    if isinstance(metadata_expression, dict) and metadata_expression:
+        cleaned["metadata"] = metadata_expression
     return cleaned
 
 
@@ -83,7 +86,9 @@ def merge_filter_overrides(
 
     filters: dict[str, Any] | None = resolved["filters"]
     if "filters" in provided_fields:
-        inline_filters = request_body.filters
+        from api.schemas.custom_metadata import dump_search_filters
+
+        inline_filters = dump_search_filters(request_body.filters)
         if inline_filters:
             filters = {**resolved["filters"], **inline_filters}
         else:
diff --git a/src/api/v1/chat.py b/src/api/v1/chat.py
index b3d6cb97f..699c663d1 100644
--- a/src/api/v1/chat.py
+++ b/src/api/v1/chat.py
@@ -7,12 +7,12 @@
 
 import json
 import time
-from typing import Any
 
 from fastapi import Depends, HTTPException, Request
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel
 
+from api.schemas.custom_metadata import SearchFiltersRequest, dump_search_filters
 from api.v1._filter_resolution import merge_filter_overrides, resolve_filter_id
 from auth_context import set_auth_context, set_score_threshold, set_search_filters, set_search_limit
 from dependencies import (
@@ -35,7 +35,7 @@ class ChatV1Body(BaseModel):
     message: str
     stream: bool = False
     chat_id: str | None = None
-    filters: dict[str, Any] | None = None
+    filters: SearchFiltersRequest | None = None
     limit: int = 10
     score_threshold: float = 0
     filter_id: str | None = None
@@ -53,6 +53,7 @@ def _extract_sources(item: dict) -> list[dict]:
                     "score": result.get("score", 0),
                     "page": result.get("page"),
                     "mimetype": result.get("mimetype"),
+                    "metadata": result.get("metadata", {}),
                 }
             )
     return sources
@@ -137,7 +138,7 @@ async def chat_create_endpoint(
 
         await _assert_owns(body.chat_id, storage_user_id)
 
-    resolved_filters = body.filters
+    resolved_filters = dump_search_filters(body.filters)
     resolved_limit = body.limit
     resolved_score_threshold = body.score_threshold
     if body.filter_id:
@@ -151,6 +152,14 @@ async def chat_create_endpoint(
             resolved, body
         )
 
+    if resolved_filters:
+        from services.custom_metadata_service import CustomMetadataService
+
+        try:
+            await CustomMetadataService().build_filter_clauses(resolved_filters)
+        except ValueError as exc:
+            return JSONResponse({"error": str(exc)}, status_code=400)
+
     if resolved_filters:
         set_search_filters(resolved_filters)
     set_search_limit(resolved_limit)
diff --git a/src/api/v1/documents.py b/src/api/v1/documents.py
index e58590e2f..69dcd25db 100644
--- a/src/api/v1/documents.py
+++ b/src/api/v1/documents.py
@@ -45,6 +45,7 @@ async def ingest_endpoint(
     tweaks: str | None = Form(None),
     replace_duplicates: str = Form("true"),
     create_filter: str = Form("false"),
+    metadata: str | None = Form(None),
     document_service=Depends(get_document_service),
     langflow_file_service=Depends(get_langflow_file_service),
     session_manager=Depends(get_session_manager),
@@ -73,6 +74,7 @@ async def ingest_endpoint(
         # it. Pass an explicit value so the Form("false") default sentinel is not
         # forwarded when this function is called directly (not via form parsing).
         preview="false",
+        metadata_json=metadata if isinstance(metadata, str) else None,
         document_service=document_service,
         langflow_file_service=langflow_file_service,
         session_manager=session_manager,
diff --git a/src/api/v1/search.py b/src/api/v1/search.py
index a8ac58624..8ece4db59 100644
--- a/src/api/v1/search.py
+++ b/src/api/v1/search.py
@@ -5,12 +5,11 @@
 Uses API key authentication.
 """
 
-from typing import Any
-
 from fastapi import Depends
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
+from api.schemas.custom_metadata import SearchFiltersRequest, dump_search_filters
 from api.v1._filter_resolution import merge_filter_overrides, resolve_filter_id
 from auth_context import set_auth_context
 from dependencies import (
@@ -27,7 +26,7 @@
 
 class SearchV1Body(BaseModel):
     query: str
-    filters: dict[str, Any] | None = None
+    filters: SearchFiltersRequest | None = None
     limit: int = 10
     score_threshold: float = 0
     filter_id: str | None = None
@@ -48,7 +47,7 @@ async def search_endpoint(
     # resolving filters so search_tool() can still identify the caller.
     set_auth_context(user.user_id, user.jwt_token)
 
-    resolved_filters = body.filters
+    resolved_filters = dump_search_filters(body.filters)
     resolved_limit = body.limit
     resolved_score_threshold = body.score_threshold
     if body.filter_id:
@@ -89,12 +88,15 @@ async def search_endpoint(
                 "score": item.get("score"),
                 "page": item.get("page"),
                 "mimetype": item.get("mimetype"),
+                "metadata": item.get("metadata", {}),
             }
             for item in result.get("results", [])
         ]
 
         return JSONResponse({"results": results})
 
+    except ValueError as e:
+        return JSONResponse({"error": str(e)}, status_code=400)
     except OpenSearchDiskSpaceError as e:
         logger.error("Search blocked by disk space constraint", error=str(e), user_id=user.user_id)
         return JSONResponse({"error": DISK_SPACE_ERROR_MESSAGE}, status_code=507)
diff --git a/src/app/routes/internal.py b/src/app/routes/internal.py
index 097a488bc..fb3a7b50c 100644
--- a/src/app/routes/internal.py
+++ b/src/app/routes/internal.py
@@ -12,6 +12,7 @@
     auth,
     chat,
     connectors,
+    custom_metadata,
     docling,
     documents,
     files,
@@ -120,6 +121,15 @@ def register_internal_routes(app: FastAPI):
 
     # Search endpoint
     app.add_api_route("/search", search.search, methods=["POST"], tags=["internal"])
+    app.add_api_route(
+        "/metadata/fields", custom_metadata.list_fields, methods=["GET"], tags=["internal"]
+    )
+    app.add_api_route(
+        "/metadata/fields/{key}/values",
+        custom_metadata.list_values,
+        methods=["GET"],
+        tags=["internal"],
+    )
 
     # File listing/search endpoints
     app.add_api_route("/files", files.list_files, methods=["GET"], tags=["internal"])
diff --git a/src/app/routes/public_v1.py b/src/app/routes/public_v1.py
index 557745e2c..0b745a719 100644
--- a/src/app/routes/public_v1.py
+++ b/src/app/routes/public_v1.py
@@ -2,6 +2,7 @@
 
 from fastapi import FastAPI
 
+from api import custom_metadata
 from api.v1 import (
     chat as v1_chat,
 )
@@ -42,6 +43,18 @@ def register_public_v1_routes(app: FastAPI):
 
     # Search endpoint
     app.add_api_route("/v1/search", v1_search.search_endpoint, methods=["POST"], tags=["public"])
+    app.add_api_route(
+        "/v1/metadata/fields",
+        custom_metadata.list_fields_v1,
+        methods=["GET"],
+        tags=["public"],
+    )
+    app.add_api_route(
+        "/v1/metadata/fields/{key}/values",
+        custom_metadata.list_values_v1,
+        methods=["GET"],
+        tags=["public"],
+    )
 
     # Documents endpoints
     app.add_api_route(
diff --git a/src/config/settings.py b/src/config/settings.py
index 2363c31b2..826f6110b 100644
--- a/src/config/settings.py
+++ b/src/config/settings.py
@@ -710,6 +710,19 @@ def _get_min_env_int(key: str, default: int, minimum: int) -> int:
             "modified_time": {"type": "date"},
             "indexed_time": {"type": "date"},
             "metadata": {"type": "object"},
+            "custom_metadata": {"type": "object", "enabled": False},
+            "metadata_entries": {
+                "type": "nested",
+                "properties": {
+                    "key": {"type": "keyword"},
+                    "type": {"type": "keyword"},
+                    "string_value": {"type": "keyword"},
+                    "string_value_text": {"type": "text"},
+                    "number_value": {"type": "double"},
+                    "date_value": {"type": "date"},
+                    "boolean_value": {"type": "boolean"},
+                },
+            },
         }
     },
 }
diff --git a/src/db/models/__init__.py b/src/db/models/__init__.py
index 5a33d69c6..8f8c764c8 100644
--- a/src/db/models/__init__.py
+++ b/src/db/models/__init__.py
@@ -7,6 +7,7 @@
 from db.models.api_key import ApiKey
 from db.models.audit_log import AuditLog
 from db.models.conversation import Conversation
+from db.models.metadata_field import MetadataField
 from db.models.migration_status import MigrationStatus
 from db.models.permission import Permission
 from db.models.role import Role
@@ -22,6 +23,7 @@
     "AuditLog",
     "Conversation",
     "MigrationStatus",
+    "MetadataField",
     "Permission",
     "Role",
     "RolePermission",
diff --git a/src/db/models/metadata_field.py b/src/db/models/metadata_field.py
new file mode 100644
index 000000000..1059b996a
--- /dev/null
+++ b/src/db/models/metadata_field.py
@@ -0,0 +1,14 @@
+"""Workspace-wide custom metadata field catalog."""
+
+from datetime import UTC, datetime
+
+from sqlmodel import Field, SQLModel
+
+
+class MetadataField(SQLModel, table=True):
+    __tablename__ = "metadata_fields"
+
+    key: str = Field(primary_key=True, max_length=64)
+    metadata_type: str = Field(max_length=16, index=True)
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
diff --git a/src/db/repositories/__init__.py b/src/db/repositories/__init__.py
index 8988b30e1..f3c14f0c1 100644
--- a/src/db/repositories/__init__.py
+++ b/src/db/repositories/__init__.py
@@ -3,6 +3,7 @@
 from db.repositories.api_key_repo import ApiKeyRepo
 from db.repositories.audit_repo import AuditRepo
 from db.repositories.conversation_repo import ConversationRepo
+from db.repositories.metadata_field_repo import MetadataFieldRepo
 from db.repositories.permission_repo import PermissionRepo
 from db.repositories.preferences_repo import PreferencesRepo
 from db.repositories.role_repo import RoleRepo
@@ -10,6 +11,8 @@
 from db.repositories.user_repo import UserRepo
 from db.repositories.workspace_config_repo import (
     SECTIONS as WORKSPACE_CONFIG_SECTIONS,
+)
+from db.repositories.workspace_config_repo import (
     WorkspaceConfigRepo,
 )
 
@@ -17,6 +20,7 @@
     "ApiKeyRepo",
     "AuditRepo",
     "ConversationRepo",
+    "MetadataFieldRepo",
     "PermissionRepo",
     "PreferencesRepo",
     "RoleRepo",
diff --git a/src/db/repositories/metadata_field_repo.py b/src/db/repositories/metadata_field_repo.py
new file mode 100644
index 000000000..ffb76bdc2
--- /dev/null
+++ b/src/db/repositories/metadata_field_repo.py
@@ -0,0 +1,31 @@
+"""Persistence interface for the custom metadata field catalog."""
+
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models.metadata_field import MetadataField
+
+
+class MetadataFieldRepo:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def get(self, key: str) -> MetadataField | None:
+        return await self.session.get(MetadataField, key)
+
+    async def list_all(self) -> list[MetadataField]:
+        result = await self.session.execute(select(MetadataField).order_by(MetadataField.key))
+        return list(result.scalars().all())
+
+    async def add(self, key: str, metadata_type: str) -> MetadataField:
+        row = MetadataField(
+            key=key,
+            metadata_type=metadata_type,
+            created_at=datetime.now(UTC),
+            updated_at=datetime.now(UTC),
+        )
+        self.session.add(row)
+        await self.session.flush()
+        return row
diff --git a/src/mcp_http/server.py b/src/mcp_http/server.py
index ec35ade22..968b962d8 100644
--- a/src/mcp_http/server.py
+++ b/src/mcp_http/server.py
@@ -96,10 +96,19 @@
             "Returns matching document chunks with relevance scores. "
             "Optionally pass `filter_id` to scope results to a knowledge "
             "filter's data_sources, or inline `filters` (data_sources, "
-            "document_types, owners, connector_types) for a per-call scope. "
+            "document_types, owners, connector_types, and a nested metadata "
+            "expression) for a per-call scope. "
             "If both are provided, inline filters override per-field."
         ),
     },
+    ("/v1/metadata/fields", "GET"): {
+        "name": "openrag_list_metadata_fields",
+        "description": "List typed custom metadata fields visible in documents the caller can access.",
+    },
+    ("/v1/metadata/fields/{key}/values", "GET"): {
+        "name": "openrag_list_metadata_values",
+        "description": "List suggested values for a visible custom metadata field.",
+    },
     # Documents endpoints
     # NOTE: /v1/documents/ingest is intentionally NOT customized here because
     # it is excluded from MCP exposure entirely (see route_maps below).
diff --git a/src/models/processors.py b/src/models/processors.py
index c8b7bee4a..9d13cdf1f 100644
--- a/src/models/processors.py
+++ b/src/models/processors.py
@@ -417,6 +417,7 @@ async def process_document_standard(
         ocr: bool | None = None,
         picture_descriptions: bool | None = None,
         shared: bool = False,
+        metadata: list[dict[str, Any]] | None = None,
     ):
         """
         Standard processing pipeline for non-Langflow processors:
@@ -618,6 +619,7 @@ async def process_document_standard(
             allowed_principals=allowed_principals,
             allowed_principal_labels=allowed_principal_labels,
             is_sample_data=is_sample_data,
+            metadata=list(metadata or []),
         )
         parser_name = slim_doc.get("parser")
         if not parser_name:
@@ -790,6 +792,8 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                             standard_kwargs[param] = int(raw)
                         except (TypeError, ValueError):
                             pass
+                if isinstance(s.get("metadata"), list):
+                    standard_kwargs["metadata"] = s["metadata"]
 
             config = get_openrag_config()
             standard_kwargs["ocr"] = config.knowledge.ocr
@@ -1271,6 +1275,8 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                                     standard_kwargs[param] = int(raw)
                                 except (TypeError, ValueError):
                                     pass
+                        if isinstance(s.get("metadata"), list):
+                            standard_kwargs["metadata"] = s["metadata"]
                     config = get_openrag_config()
                     standard_kwargs["ocr"] = config.knowledge.ocr
                     standard_kwargs["picture_descriptions"] = config.knowledge.picture_descriptions
@@ -1545,6 +1551,7 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                 document_id=file_hash,
                 original_filename=original_filename,
                 original_mimetype=original_mimetype,
+                metadata=self.settings.get("metadata") if isinstance(self.settings, dict) else None,
             )
 
             # Langflow returns "success" even when no text was extracted
diff --git a/src/services/chat_service.py b/src/services/chat_service.py
index 9fdd03fc1..4b5f9aca7 100644
--- a/src/services/chat_service.py
+++ b/src/services/chat_service.py
@@ -155,26 +155,9 @@ async def langflow_chat(
         # Build the complete filter expression like the search service does
         filter_expression: dict[str, Any] = {}
         if filters:
-            filter_clauses = []
-            # Map frontend filter names to backend field names
-            field_mapping = {
-                "data_sources": "filename",
-                "document_types": "mimetype",
-                "owners": "owner",
-                "connector_types": "connector_type",
-            }
-
-            for filter_key, values in filters.items():
-                if values is not None and isinstance(values, list) and len(values) > 0:
-                    # Map frontend key to backend field name
-                    field_name = field_mapping.get(filter_key, filter_key)
+            from services.custom_metadata_service import CustomMetadataService
 
-                    if len(values) == 1:
-                        # Single value filter
-                        filter_clauses.append({"term": {field_name: values[0]}})
-                    else:
-                        # Multiple values filter
-                        filter_clauses.append({"terms": {field_name: values}})
+            filter_clauses = await CustomMetadataService().build_filter_clauses(filters)
 
             if filter_clauses:
                 filter_expression["filter"] = filter_clauses
@@ -273,25 +256,9 @@ async def langflow_nudges_chat(
         filter_clauses = []
 
         if filters:
-            # Map frontend filter names to backend field names
-            field_mapping = {
-                "data_sources": "filename",
-                "document_types": "mimetype",
-                "owners": "owner",
-                "connector_types": "connector_type",
-            }
+            from services.custom_metadata_service import CustomMetadataService
 
-            for filter_key, values in filters.items():
-                if values is not None and isinstance(values, list) and len(values) > 0:
-                    # Map frontend key to backend field name
-                    field_name = field_mapping.get(filter_key, filter_key)
-
-                    if len(values) == 1:
-                        # Single value filter
-                        filter_clauses.append({"term": {field_name: values[0]}})
-                    else:
-                        # Multiple values filter
-                        filter_clauses.append({"terms": {field_name: values}})
+            filter_clauses = await CustomMetadataService().build_filter_clauses(filters)
 
             if filter_clauses:
                 has_user_filters = True
diff --git a/src/services/custom_metadata_service.py b/src/services/custom_metadata_service.py
new file mode 100644
index 000000000..27c2da15d
--- /dev/null
+++ b/src/services/custom_metadata_service.py
@@ -0,0 +1,377 @@
+"""Validation, indexing, discovery, and query compilation for custom metadata."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import date
+from typing import Any, Literal
+
+MetadataType = Literal["string", "number", "date", "boolean"]
+
+
+@dataclass(frozen=True)
+class NormalizedMetadata:
+    source: dict[str, dict[str, Any]]
+    index_entries: list[dict[str, Any]]
+
+
+class CustomMetadataService:
+    """Present one small metadata interface to ingest and retrieval callers."""
+
+    VALUE_FIELDS: dict[str, str] = {
+        "string": "string_value",
+        "number": "number_value",
+        "date": "date_value",
+        "boolean": "boolean_value",
+    }
+    PROTECTED_KEYS = frozenset(
+        {
+            "allowed_groups",
+            "allowed_principal_labels",
+            "allowed_principals",
+            "allowed_users",
+            "chunk_embedding",
+            "chunk_overlap",
+            "chunk_size",
+            "connector_file_id",
+            "connector_type",
+            "connector_types",
+            "custom_metadata",
+            "created_time",
+            "data_sources",
+            "document_id",
+            "document_types",
+            "embedding_dimensions",
+            "embedding_model",
+            "file_size",
+            "filename",
+            "indexed_time",
+            "ingest_run_id",
+            "is_sample_data",
+            "metadata",
+            "metadata_entries",
+            "mimetype",
+            "modified_time",
+            "owner",
+            "owner_email",
+            "owner_name",
+            "owners",
+            "page",
+            "parser",
+            "source_url",
+            "text",
+            "user_permissions",
+            "group_permissions",
+        }
+    )
+    KEY_PATTERN = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
+
+    def __init__(self, session_factory=None):
+        self.session_factory = session_factory
+
+    def _session_factory(self):
+        if self.session_factory is not None:
+            return self.session_factory
+        from db.engine import SessionLocal, init_engine
+
+        if SessionLocal is None:
+            init_engine()
+            from db.engine import SessionLocal as initialized_session_factory
+
+            return initialized_session_factory
+        return SessionLocal
+
+    async def register_entries(self, entries: list[dict[str, Any]] | None) -> None:
+        normalized = self.normalize_entries(entries)
+        if not normalized.source:
+            return
+        session_factory = self._session_factory()
+        if session_factory is None:
+            raise RuntimeError("Metadata field registry database is unavailable")
+        from db.repositories.metadata_field_repo import MetadataFieldRepo
+
+        async with session_factory() as session:
+            repo = MetadataFieldRepo(session)
+            for key, item in normalized.source.items():
+                existing = await repo.get(key)
+                metadata_type = item["type"]
+                if existing is None:
+                    await repo.add(key, metadata_type)
+                elif existing.metadata_type != metadata_type:
+                    raise ValueError(
+                        f"Custom metadata '{key}' expected {existing.metadata_type}, "
+                        f"received {metadata_type}"
+                    )
+            await session.commit()
+
+    async def get_field_types(self) -> dict[str, str]:
+        session_factory = self._session_factory()
+        if session_factory is None:
+            return {}
+        from db.repositories.metadata_field_repo import MetadataFieldRepo
+
+        async with session_factory() as session:
+            rows = await MetadataFieldRepo(session).list_all()
+            return {row.key: row.metadata_type for row in rows}
+
+    async def build_filter_clauses(self, filters: dict[str, Any] | None) -> list[dict[str, Any]]:
+        """Compile built-in and custom public filters through one interface."""
+        field_mapping = {
+            "data_sources": "filename",
+            "document_types": "mimetype",
+            "owners": "owner",
+            "connector_types": "connector_type",
+        }
+        clauses: list[dict[str, Any]] = []
+        for key, value in (filters or {}).items():
+            if key == "metadata":
+                clauses.append(self.compile_expression(value, await self.get_field_types()))
+                continue
+            field = field_mapping.get(key)
+            if field is None:
+                raise ValueError(f"Unknown filter field '{key}'")
+            if not isinstance(value, list):
+                raise ValueError(f"Filter '{key}' must be a list")
+            if "*" in value:
+                continue
+            if not value:
+                clauses.append({"term": {field: "__IMPOSSIBLE_VALUE__"}})
+            elif len(value) == 1:
+                clauses.append({"term": {field: value[0]}})
+            else:
+                clauses.append({"terms": {field: value}})
+        return clauses
+
+    def normalize_entries(self, entries: list[dict[str, Any]] | None) -> NormalizedMetadata:
+        source: dict[str, dict[str, Any]] = {}
+        index_entries: list[dict[str, Any]] = []
+
+        if len(entries or []) > 50:
+            raise ValueError("A document cannot have more than 50 custom metadata fields")
+
+        for raw in entries or []:
+            key = str(raw.get("key") or "").strip().lower()
+            metadata_type = str(raw.get("type") or "")
+            value = raw.get("value")
+            if not key:
+                raise ValueError("Custom metadata key is required")
+            if key in self.PROTECTED_KEYS or key.startswith("chunk_embedding_"):
+                raise ValueError(f"Custom metadata key '{key}' is protected")
+            if not self.KEY_PATTERN.fullmatch(key):
+                raise ValueError(
+                    f"Custom metadata key '{key}' must use lowercase letters, numbers, and underscores"
+                )
+            if key in source:
+                raise ValueError(f"Custom metadata key '{key}' is duplicate")
+            if metadata_type not in self.VALUE_FIELDS:
+                raise ValueError(f"Unsupported custom metadata type for '{key}': {metadata_type}")
+
+            normalized_value = self._normalize_value(key, metadata_type, value)
+            source[key] = {"type": metadata_type, "value": normalized_value}
+            index_entry = {
+                "key": key,
+                "type": metadata_type,
+                self.VALUE_FIELDS[metadata_type]: normalized_value,
+            }
+            if metadata_type == "string":
+                index_entry["string_value_text"] = normalized_value
+            index_entries.append(index_entry)
+
+        return NormalizedMetadata(source=source, index_entries=index_entries)
+
+    def entries_from_mapping(self, values: dict[str, Any] | None) -> list[dict[str, Any]]:
+        """Convert trusted Langflow/custom maps to the typed public representation."""
+        entries: list[dict[str, Any]] = []
+        for key, raw_value in (values or {}).items():
+            canonical_key = str(key).strip().lower()
+            if canonical_key in self.PROTECTED_KEYS or raw_value is None:
+                continue
+            if isinstance(raw_value, dict) and {"type", "value"} <= raw_value.keys():
+                entries.append(
+                    {
+                        "key": canonical_key,
+                        "type": raw_value["type"],
+                        "value": raw_value["value"],
+                    }
+                )
+                continue
+            sample = raw_value[0] if isinstance(raw_value, list) and raw_value else raw_value
+            if isinstance(sample, bool):
+                metadata_type = "boolean"
+            elif isinstance(sample, (int, float)):
+                metadata_type = "number"
+            elif isinstance(sample, str):
+                metadata_type = "string"
+            else:
+                continue
+            entries.append({"key": canonical_key, "type": metadata_type, "value": raw_value})
+        return entries
+
+    def compile_expression(
+        self,
+        expression: dict[str, Any] | None,
+        field_types: dict[str, str],
+    ) -> dict[str, Any]:
+        """Validate a public metadata expression and compile it to OpenSearch DSL."""
+        if not expression:
+            return {"match_all": {}}
+        if self._condition_count(expression) > 50:
+            raise ValueError("Custom metadata filters cannot exceed 50 conditions")
+        return self._compile_node(expression, field_types, depth=0)
+
+    def _condition_count(self, node: dict[str, Any]) -> int:
+        conditions = node.get("conditions")
+        if not isinstance(conditions, list):
+            return 1
+        return sum(
+            self._condition_count(condition)
+            for condition in conditions
+            if isinstance(condition, dict)
+        )
+
+    def _compile_node(
+        self,
+        node: dict[str, Any],
+        field_types: dict[str, str],
+        *,
+        depth: int,
+    ) -> dict[str, Any]:
+        if depth > 5:
+            raise ValueError("Custom metadata filter nesting cannot exceed 5 levels")
+        if "conditions" in node:
+            op = str(node.get("op") or "and").lower()
+            if op not in {"and", "or"}:
+                raise ValueError("Custom metadata filter group op must be 'and' or 'or'")
+            conditions = node.get("conditions")
+            if not isinstance(conditions, list) or not conditions:
+                raise ValueError("Custom metadata filter group requires conditions")
+            compiled = [
+                self._compile_node(item, field_types, depth=depth + 1) for item in conditions
+            ]
+            if op == "and":
+                return {"bool": {"must": compiled}}
+            return {"bool": {"should": compiled, "minimum_should_match": 1}}
+
+        key = str(node.get("key") or "").strip().lower()
+        if key not in field_types:
+            raise ValueError(f"Unknown custom metadata key '{key}'")
+        metadata_type = field_types[key]
+        operator = str(node.get("operator") or "").lower()
+        return self._compile_condition(key, metadata_type, operator, node.get("value"))
+
+    def _compile_condition(
+        self,
+        key: str,
+        metadata_type: str,
+        operator: str,
+        value: Any,
+    ) -> dict[str, Any]:
+        value_field = self.VALUE_FIELDS.get(metadata_type)
+        if value_field is None:
+            raise ValueError(f"Unsupported registered type for '{key}': {metadata_type}")
+        key_clause = {"term": {"metadata_entries.key": key}}
+        field = f"metadata_entries.{value_field}"
+
+        positive: dict[str, Any]
+        if operator in {"exists", "not_exists"}:
+            positive = {"nested": {"path": "metadata_entries", "query": key_clause}}
+            return self._negate(positive) if operator == "not_exists" else positive
+        if operator in {"equals", "not_equals"}:
+            self._require_scalar(key, operator, value)
+            normalized = self._normalize_value(key, metadata_type, value)
+            value_clause = {"term": {field: normalized}}
+        elif operator in {"in", "not_in"}:
+            if not isinstance(value, list) or not value:
+                raise ValueError(f"Operator '{operator}' for '{key}' requires a non-empty list")
+            normalized = self._normalize_value(key, metadata_type, value)
+            value_clause = {"terms": {field: normalized}}
+        elif operator in {"contains", "not_contains"}:
+            if metadata_type != "string":
+                raise ValueError(f"Operator '{operator}' is only valid for string metadata")
+            self._require_scalar(key, operator, value)
+            normalized = self._normalize_value(key, metadata_type, value)
+            value_clause = {"match_phrase": {"metadata_entries.string_value_text": normalized}}
+        elif operator in {"gt", "gte", "lt", "lte"}:
+            if metadata_type not in {"number", "date"}:
+                raise ValueError(f"Operator '{operator}' requires number or date metadata")
+            self._require_scalar(key, operator, value)
+            normalized = self._normalize_value(key, metadata_type, value)
+            value_clause = {"range": {field: {operator: normalized}}}
+        elif operator == "between":
+            if metadata_type not in {"number", "date"} or not isinstance(value, dict):
+                raise ValueError("Operator 'between' requires number or date bounds")
+            bounds = {
+                bound: self._normalize_value(key, metadata_type, bound_value)
+                for bound, bound_value in value.items()
+                if bound in {"gt", "gte", "lt", "lte"}
+            }
+            if not bounds:
+                raise ValueError("Operator 'between' requires at least one bound")
+            value_clause = {"range": {field: bounds}}
+        else:
+            raise ValueError(f"Unsupported custom metadata operator '{operator}'")
+
+        positive = {
+            "nested": {
+                "path": "metadata_entries",
+                "query": {"bool": {"must": [key_clause, value_clause]}},
+            }
+        }
+        if operator in {"not_equals", "not_in", "not_contains"}:
+            return self._negate(positive)
+        return positive
+
+    @staticmethod
+    def _negate(clause: dict[str, Any]) -> dict[str, Any]:
+        return {"bool": {"must_not": [clause]}}
+
+    @staticmethod
+    def _require_scalar(key: str, operator: str, value: Any) -> None:
+        if isinstance(value, (list, dict)):
+            raise ValueError(f"Operator '{operator}' for '{key}' requires a scalar value")
+
+    @staticmethod
+    def _normalize_value(key: str, metadata_type: str, value: Any) -> Any:
+        values = value if isinstance(value, list) else [value]
+        if len(values) > 100:
+            raise ValueError(f"Custom metadata '{key}' cannot contain more than 100 values")
+        normalized: list[Any] = []
+        for item in values:
+            if metadata_type == "string":
+                if not isinstance(item, str):
+                    raise ValueError(f"Custom metadata '{key}' must be a string")
+                if len(item) > 2048:
+                    raise ValueError(f"Custom metadata '{key}' exceeds 2048 characters")
+                normalized.append(item)
+            elif metadata_type == "number":
+                if isinstance(item, bool) or not isinstance(item, (int, float)):
+                    raise ValueError(f"Custom metadata '{key}' must be a number")
+                normalized.append(item)
+            elif metadata_type == "boolean":
+                if not isinstance(item, bool):
+                    raise ValueError(f"Custom metadata '{key}' must be a boolean")
+                normalized.append(item)
+            else:
+                if not isinstance(item, str):
+                    raise ValueError(f"Custom metadata '{key}' must be an ISO date")
+                try:
+                    date.fromisoformat(item)
+                except ValueError as exc:
+                    raise ValueError(f"Custom metadata '{key}' must be an ISO date") from exc
+                normalized.append(item)
+        return normalized if isinstance(value, list) else normalized[0]
+
+
+CUSTOM_METADATA_MAPPING: dict[str, Any] = {
+    "type": "nested",
+    "properties": {
+        "key": {"type": "keyword"},
+        "type": {"type": "keyword"},
+        "string_value": {"type": "keyword"},
+        "string_value_text": {"type": "text"},
+        "number_value": {"type": "double"},
+        "date_value": {"type": "date"},
+        "boolean_value": {"type": "boolean"},
+    },
+}
diff --git a/src/services/document_index_writer.py b/src/services/document_index_writer.py
index a1eff9767..7f3da027a 100644
--- a/src/services/document_index_writer.py
+++ b/src/services/document_index_writer.py
@@ -12,6 +12,7 @@
 from dataclasses import dataclass, field
 from typing import Any
 
+from services.custom_metadata_service import CustomMetadataService
 from utils.embedding_fields import ensure_embedding_field_exists
 from utils.embeddings import create_index_body
 from utils.group_acl import unique_acl_principal_labels, unique_acl_principals
@@ -43,6 +44,7 @@ class DocumentIndexContext:
     parser: str | None = None
     chunk_size: int | None = None
     chunk_overlap: int | None = None
+    metadata: list[dict[str, Any]] = field(default_factory=list)
 
 
 @dataclass
@@ -59,6 +61,7 @@ class DocumentIndexWriter:
 
     def __init__(self, opensearch_client: Any | None = None):
         self.opensearch_client = opensearch_client
+        self.custom_metadata = CustomMetadataService()
 
     def _get_write_client(self) -> Any:
         from config.settings import clients
@@ -106,6 +109,7 @@ async def index_chunks(
 
         now = datetime.datetime.now(datetime.UTC).isoformat()
         bulk_body: list[dict[str, Any]] = []
+        custom_entries_by_key: dict[str, dict[str, Any]] = {}
         for chunk in chunks:
             if len(chunk.vector) != dimensions:
                 raise ValueError(
@@ -113,14 +117,23 @@ async def index_chunks(
                     f"expected {dimensions}, got {len(chunk.vector)} for {chunk.chunk_id}"
                 )
             bulk_body.append({"index": {"_index": index_name, "_id": chunk.chunk_id}})
-            bulk_body.append(
-                self._build_chunk_document(
-                    context=context,
-                    chunk=chunk,
-                    embedding_field=embedding_field,
-                    indexed_time=now,
-                )
+            document = self._build_chunk_document(
+                context=context,
+                chunk=chunk,
+                embedding_field=embedding_field,
+                indexed_time=now,
             )
+            for key, item in document.get("custom_metadata", {}).items():
+                candidate = {"key": key, "type": item["type"], "value": item["value"]}
+                existing = custom_entries_by_key.get(key)
+                if existing is not None and existing["type"] != candidate["type"]:
+                    raise ValueError(
+                        f"Custom metadata '{key}' has conflicting types in one ingest batch"
+                    )
+                custom_entries_by_key[key] = candidate
+            bulk_body.append(document)
+
+        await self.custom_metadata.register_entries(list(custom_entries_by_key.values()))
 
         result = await client.bulk(body=bulk_body, refresh=refresh)
         self._raise_for_bulk_errors(result)
@@ -194,6 +207,7 @@ def _build_chunk_document(
         indexed_time: str,
     ) -> dict[str, Any]:
         metadata = self._normalized_metadata(chunk.metadata)
+        custom_metadata = self._custom_metadata(context, metadata)
         document_id = context.document_id or str(metadata.get("document_id") or chunk.chunk_id)
         filename = context.filename or str(metadata.get("filename") or "")
         mimetype = context.mimetype or str(metadata.get("mimetype") or "")
@@ -219,7 +233,8 @@ def _build_chunk_document(
                 context.allowed_principal_labels
             ),
             "indexed_time": indexed_time,
-            "metadata": metadata.get("metadata", {}),
+            "custom_metadata": custom_metadata.source,
+            "metadata_entries": custom_metadata.index_entries,
         }
 
         parser = context.parser or metadata.get("parser")
@@ -257,6 +272,58 @@ def _build_chunk_document(
 
         return doc
 
+    def _custom_metadata(
+        self,
+        context: DocumentIndexContext,
+        chunk_metadata: dict[str, Any],
+    ):
+        explicit = chunk_metadata.get("metadata")
+        if isinstance(explicit, list):
+            chunk_entries = explicit
+        elif isinstance(explicit, dict):
+            chunk_entries = self.custom_metadata.entries_from_mapping(explicit)
+        else:
+            chunk_entries = []
+
+        known_chunk_fields = {
+            "allowed_groups",
+            "allowed_principal_labels",
+            "allowed_principals",
+            "allowed_users",
+            "chunk_overlap",
+            "chunk_size",
+            "connector_file_id",
+            "connector_type",
+            "created_time",
+            "document_id",
+            "file_size",
+            "filename",
+            "filesize",
+            "langflow_chunk_id",
+            "metadata",
+            "mimetype",
+            "modified_time",
+            "page",
+            "parser",
+            "source_url",
+        }
+        arbitrary = {
+            key: value for key, value in chunk_metadata.items() if key not in known_chunk_fields
+        }
+        chunk_entries.extend(self.custom_metadata.entries_from_mapping(arbitrary))
+
+        chunk_normalized = self.custom_metadata.normalize_entries(chunk_entries)
+        context_normalized = self.custom_metadata.normalize_entries(context.metadata)
+        merged_source = {**chunk_normalized.source, **context_normalized.source}
+        merged_entries = {entry["key"]: entry for entry in chunk_normalized.index_entries}
+        merged_entries.update({entry["key"]: entry for entry in context_normalized.index_entries})
+        from services.custom_metadata_service import NormalizedMetadata
+
+        return NormalizedMetadata(
+            source=merged_source,
+            index_entries=list(merged_entries.values()),
+        )
+
     @staticmethod
     def _normalized_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
         normalized = dict(metadata or {})
diff --git a/src/services/langflow_file_service.py b/src/services/langflow_file_service.py
index 3be3c48dd..531ea7bb6 100644
--- a/src/services/langflow_file_service.py
+++ b/src/services/langflow_file_service.py
@@ -209,6 +209,7 @@ def _configure_ingest_callback(
         chunk_size: int | None = None,
         chunk_overlap: int | None = None,
         connector_file_id: str | None = None,
+        metadata: list[dict[str, Any]] | None = None,
     ) -> tuple[str | None, str | None]:
         if self.ingest_token_service is None:
             logger.warning(
@@ -245,6 +246,7 @@ def _configure_ingest_callback(
             chunk_size=chunk_size,
             chunk_overlap=chunk_overlap,
             connector_file_id=connector_file_id,
+            metadata=list(metadata or []),
         )
         token = self.ingest_token_service.create_token(context)
         logger.info(
@@ -374,6 +376,7 @@ async def run_ingestion_flow(
         docling_task_id: str | None = None,
         original_filename: str | None = None,
         original_mimetype: str | None = None,
+        metadata: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         """
         Trigger the ingestion flow with provided file paths.
@@ -495,6 +498,7 @@ async def run_ingestion_flow(
             chunk_size=chunk_size,
             chunk_overlap=chunk_overlap,
             connector_file_id=connector_file_id,
+            metadata=metadata,
         )
         headers.update(
             self._ingest_callback_global_var_headers(
@@ -915,6 +919,7 @@ async def upload_and_ingest_file(
         allowed_principal_labels: list[dict[str, Any]] | None = None,
         original_filename: str | None = None,
         original_mimetype: str | None = None,
+        metadata: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         """
         Two-phase Docling upload + Langflow ingest operation.
@@ -1061,6 +1066,9 @@ async def upload_and_ingest_file(
                 allowed_principal_labels=allowed_principal_labels,
                 original_filename=original_filename,
                 original_mimetype=original_mimetype,
+                metadata=metadata
+                if metadata is not None
+                else (settings.get("metadata") if isinstance(settings, dict) else None),
             )
             total_duration = round(time.time() - total_start_time, 2)
             logger.info(f"[LF] Ingestion completed successfully in {total_duration}s")
diff --git a/src/services/langflow_ingest_token_service.py b/src/services/langflow_ingest_token_service.py
index 0f4eccf96..2c6c21398 100644
--- a/src/services/langflow_ingest_token_service.py
+++ b/src/services/langflow_ingest_token_service.py
@@ -168,6 +168,7 @@ def _context_to_payload(context: DocumentIndexContext) -> dict[str, Any]:
             "chunk_size": context.chunk_size,
             "chunk_overlap": context.chunk_overlap,
             "connector_file_id": context.connector_file_id,
+            "metadata": list(context.metadata),
         }
 
     @staticmethod
@@ -200,4 +201,5 @@ def _payload_to_context(payload: dict[str, Any]) -> DocumentIndexContext:
             chunk_size=payload.get("chunk_size"),
             chunk_overlap=payload.get("chunk_overlap"),
             connector_file_id=payload.get("connector_file_id"),
+            metadata=list(payload.get("metadata") or []),
         )
diff --git a/src/services/search_service.py b/src/services/search_service.py
index 0a49c5e23..c78c06506 100644
--- a/src/services/search_service.py
+++ b/src/services/search_service.py
@@ -10,6 +10,7 @@
 from auth_context import get_auth_context
 from config.embedding_constants import OPENAI_DEFAULT_EMBEDDING_MODEL
 from config.settings import clients, get_embedding_model, get_index_name, get_openrag_config
+from services.custom_metadata_service import CustomMetadataService
 from utils.container_utils import transform_localhost_url
 from utils.logging_config import get_logger
 
@@ -70,6 +71,7 @@ class SearchService:
     def __init__(self, session_manager=None, models_service=None):
         self.session_manager = session_manager
         self.models_service = models_service
+        self.custom_metadata = CustomMetadataService()
         self._configure_provider_env()
 
     def _configure_provider_env(self):
@@ -139,28 +141,7 @@ async def search_tool(self, query: str, embedding_model: str = None) -> dict[str
             # Build filter clauses first so we can use them in model detection
             filter_clauses: list[dict[str, Any]] = []
             if filters:
-                # Map frontend filter names to backend field names
-                field_mapping = {
-                    "data_sources": "filename",
-                    "document_types": "mimetype",
-                    "owners": "owner",
-                    "connector_types": "connector_type",
-                }
-
-                for filter_key, values in filters.items():
-                    if values is not None and isinstance(values, list):
-                        # Map frontend key to backend field name
-                        field_name = field_mapping.get(filter_key, filter_key)
-
-                        if len(values) == 0:
-                            # Empty array means "match nothing" - use impossible filter
-                            filter_clauses.append({"term": {field_name: "__IMPOSSIBLE_VALUE__"}})
-                        elif len(values) == 1:
-                            # Single value filter
-                            filter_clauses.append({"term": {field_name: values[0]}})
-                        else:
-                            # Multiple values filter
-                            filter_clauses.append({"terms": {field_name: values}})
+                filter_clauses = await self.custom_metadata.build_filter_clauses(filters)
 
             try:
                 # Build aggregation query with filters applied
@@ -287,28 +268,7 @@ async def embed_with_model(model_name):
             # Wildcard query - no embedding needed
             filter_clauses = []
             if filters:
-                # Map frontend filter names to backend field names
-                field_mapping = {
-                    "data_sources": "filename",
-                    "document_types": "mimetype",
-                    "owners": "owner",
-                    "connector_types": "connector_type",
-                }
-
-                for filter_key, values in filters.items():
-                    if values is not None and isinstance(values, list):
-                        # Map frontend key to backend field name
-                        field_name = field_mapping.get(filter_key, filter_key)
-
-                        if len(values) == 0:
-                            # Empty array means "match nothing" - use impossible filter
-                            filter_clauses.append({"term": {field_name: "__IMPOSSIBLE_VALUE__"}})
-                        elif len(values) == 1:
-                            # Single value filter
-                            filter_clauses.append({"term": {field_name: values[0]}})
-                        else:
-                            # Multiple values filter
-                            filter_clauses.append({"terms": {field_name: values}})
+                filter_clauses = await self.custom_metadata.build_filter_clauses(filters)
 
         # Build query body
         if is_wildcard_match_all:
@@ -447,6 +407,7 @@ async def embed_with_model(model_name):
                 "allowed_users",
                 "allowed_groups",
                 "allowed_principal_labels",
+                "custom_metadata",
             ],
             "size": limit,
         }
@@ -582,6 +543,7 @@ async def embed_with_model(model_name):
                     "allowed_users": source.get("allowed_users", []),
                     "allowed_groups": source.get("allowed_groups", []),
                     "allowed_principal_labels": source.get("allowed_principal_labels", []),
+                    "metadata": source.get("custom_metadata", source.get("metadata", {})),
                 }
             )
 
diff --git a/src/utils/embeddings.py b/src/utils/embeddings.py
index 01b6c5086..5f4a979c9 100644
--- a/src/utils/embeddings.py
+++ b/src/utils/embeddings.py
@@ -1,3 +1,4 @@
+from services.custom_metadata_service import CUSTOM_METADATA_MAPPING
 from utils.embedding_fields import build_knn_vector_field, get_embedding_field_name
 from utils.logging_config import get_logger
 
@@ -53,6 +54,8 @@ async def create_index_body(
         "modified_time": {"type": "date"},
         "indexed_time": {"type": "date"},
         "metadata": {"type": "object"},
+        "custom_metadata": {"type": "object", "enabled": False},
+        "metadata_entries": CUSTOM_METADATA_MAPPING,
     }
 
     if embedding_dimensions:
diff --git a/src/utils/opensearch_init.py b/src/utils/opensearch_init.py
index 4dca27f83..abcdfc248 100644
--- a/src/utils/opensearch_init.py
+++ b/src/utils/opensearch_init.py
@@ -21,6 +21,7 @@
     get_index_name,
     get_openrag_config,
 )
+from services.custom_metadata_service import CUSTOM_METADATA_MAPPING
 from utils.embeddings import create_index_body
 from utils.logging_config import get_logger
 from utils.telemetry import Category, MessageId, TelemetryClient
@@ -227,7 +228,11 @@ async def _ensure_opensearch_index():
             await _ensure_field_mappings(
                 clients.opensearch,
                 index_name,
-                {"allowed_principal_labels": ACL_PRINCIPAL_LABELS_MAPPING},
+                {
+                    "allowed_principal_labels": ACL_PRINCIPAL_LABELS_MAPPING,
+                    "custom_metadata": {"type": "object", "enabled": False},
+                    "metadata_entries": CUSTOM_METADATA_MAPPING,
+                },
             )
             await _ensure_index_replicas(clients.opensearch, index_name)
             return
diff --git a/tests/unit/api/test_v1_filter_resolution.py b/tests/unit/api/test_v1_filter_resolution.py
index 649c0d7ef..348835370 100644
--- a/tests/unit/api/test_v1_filter_resolution.py
+++ b/tests/unit/api/test_v1_filter_resolution.py
@@ -2,7 +2,7 @@
 
 from pydantic import BaseModel
 
-from api.v1._filter_resolution import merge_filter_overrides
+from api.v1._filter_resolution import _strip_wildcards, merge_filter_overrides
 
 
 class _RequestBody(BaseModel):
@@ -53,3 +53,14 @@ def test_merge_filter_overrides_merges_partial_inline_filters_per_field():
     assert filters == {"data_sources": ["beta.md"], "owners": ["alice"]}
     assert limit == 5
     assert score_threshold == 0.4
+
+
+def test_strip_wildcards_preserves_custom_metadata_expression():
+    expression = {
+        "op": "and",
+        "conditions": [{"key": "supplier", "operator": "equals", "value": "Dell"}],
+    }
+
+    assert _strip_wildcards(
+        {"data_sources": ["*"], "owners": ["alice"], "metadata": expression}
+    ) == {"owners": ["alice"], "metadata": expression}
diff --git a/tests/unit/services/test_custom_metadata_service.py b/tests/unit/services/test_custom_metadata_service.py
new file mode 100644
index 000000000..659d9dddd
--- /dev/null
+++ b/tests/unit/services/test_custom_metadata_service.py
@@ -0,0 +1,172 @@
+import pytest
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlmodel import SQLModel
+
+from services.custom_metadata_service import CustomMetadataService
+from services.document_index_writer import (
+    DocumentIndexChunk,
+    DocumentIndexContext,
+    DocumentIndexWriter,
+)
+
+
+@pytest_asyncio.fixture
+async def metadata_session_factory():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as connection:
+        await connection.run_sync(SQLModel.metadata.create_all)
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    yield factory
+    await engine.dispose()
+
+
+def test_normalize_entries_builds_typed_source_and_index_values():
+    service = CustomMetadataService()
+
+    normalized = service.normalize_entries(
+        [
+            {"key": "supplier", "type": "string", "value": "Dell"},
+            {"key": "contract_end", "type": "date", "value": "2026-12-31"},
+        ]
+    )
+
+    assert normalized.source == {
+        "supplier": {"type": "string", "value": "Dell"},
+        "contract_end": {"type": "date", "value": "2026-12-31"},
+    }
+    assert normalized.index_entries == [
+        {
+            "key": "supplier",
+            "type": "string",
+            "string_value": "Dell",
+            "string_value_text": "Dell",
+        },
+        {"key": "contract_end", "type": "date", "date_value": "2026-12-31"},
+    ]
+
+
+def test_normalize_entries_rejects_protected_and_duplicate_keys():
+    service = CustomMetadataService()
+
+    with pytest.raises(ValueError, match="protected"):
+        service.normalize_entries([{"key": "Owner", "type": "string", "value": "other"}])
+
+    with pytest.raises(ValueError, match="duplicate"):
+        service.normalize_entries(
+            [
+                {"key": "supplier", "type": "string", "value": "Dell"},
+                {"key": "supplier", "type": "string", "value": "Lenovo"},
+            ]
+        )
+
+
+def test_compile_expression_supports_nested_logic_and_type_aware_operators():
+    service = CustomMetadataService()
+
+    query = service.compile_expression(
+        {
+            "op": "and",
+            "conditions": [
+                {"key": "supplier", "operator": "equals", "value": "Dell"},
+                {
+                    "op": "or",
+                    "conditions": [
+                        {
+                            "key": "contract_end",
+                            "operator": "between",
+                            "value": {"gte": "2026-01-01", "lte": "2026-12-31"},
+                        },
+                        {
+                            "key": "contract_type",
+                            "operator": "contains",
+                            "value": "support",
+                        },
+                    ],
+                },
+            ],
+        },
+        {
+            "supplier": "string",
+            "contract_end": "date",
+            "contract_type": "string",
+        },
+    )
+
+    assert "must" in query["bool"]
+    assert query["bool"]["must"][0]["nested"]["path"] == "metadata_entries"
+    nested_or = query["bool"]["must"][1]["bool"]
+    assert nested_or["minimum_should_match"] == 1
+    assert nested_or["should"][0]["nested"]["query"]["bool"]["must"][1] == {
+        "range": {
+            "metadata_entries.date_value": {
+                "gte": "2026-01-01",
+                "lte": "2026-12-31",
+            }
+        }
+    }
+
+
+def test_compile_expression_requires_scalar_values_except_for_set_operators():
+    service = CustomMetadataService()
+
+    with pytest.raises(ValueError, match="requires a scalar value"):
+        service.compile_expression(
+            {"key": "supplier", "operator": "equals", "value": ["Dell"]},
+            {"supplier": "string"},
+        )
+
+    compiled = service.compile_expression(
+        {"key": "supplier", "operator": "in", "value": ["Dell", "Lenovo"]},
+        {"supplier": "string"},
+    )
+    assert compiled["nested"]["query"]["bool"]["must"][1] == {
+        "terms": {"metadata_entries.string_value": ["Dell", "Lenovo"]}
+    }
+
+
+def test_document_writer_persists_document_and_langflow_custom_metadata():
+    writer = DocumentIndexWriter()
+    context = DocumentIndexContext(
+        document_id="doc-1",
+        filename="contract.pdf",
+        mimetype="application/pdf",
+        embedding_model="model",
+        metadata=[{"key": "supplier", "type": "string", "value": "Dell"}],
+    )
+    chunk = DocumentIndexChunk(
+        chunk_id="chunk-1",
+        text="terms",
+        vector=[0.1],
+        metadata={"contract_number": 1234},
+    )
+
+    document = writer._build_chunk_document(
+        context=context,
+        chunk=chunk,
+        embedding_field="embedding",
+        indexed_time="now",
+    )
+
+    assert document["custom_metadata"]["supplier"]["value"] == "Dell"
+    assert document["custom_metadata"]["contract_number"] == {
+        "type": "number",
+        "value": 1234,
+    }
+    assert {entry["key"] for entry in document["metadata_entries"]} == {
+        "supplier",
+        "contract_number",
+    }
+
+
+@pytest.mark.asyncio
+async def test_registry_first_type_wins_and_mismatches_are_rejected(
+    metadata_session_factory,
+):
+    service = CustomMetadataService(session_factory=metadata_session_factory)
+
+    await service.register_entries([{"key": "contract_number", "type": "string", "value": "A-123"}])
+
+    assert await service.get_field_types() == {"contract_number": "string"}
+    with pytest.raises(ValueError, match="expected string"):
+        await service.register_entries([{"key": "contract_number", "type": "number", "value": 123}])
diff --git a/tests/unit/test_langflow_ingest_callback.py b/tests/unit/test_langflow_ingest_callback.py
index 1405fe282..79f7c8ba4 100644
--- a/tests/unit/test_langflow_ingest_callback.py
+++ b/tests/unit/test_langflow_ingest_callback.py
@@ -117,12 +117,14 @@ def test_ingest_token_round_trips_connector_file_id():
         owner="user-1",
         ingest_run_id="run-1",
         connector_file_id="my-bucket::報告書.pdf",
+        metadata=[{"key": "supplier", "type": "string", "value": "Dell"}],
     )
     token = token_service.create_token(context)
 
     restored_context, _jti = token_service.validate_token(token)
 
     assert restored_context.connector_file_id == "my-bucket::報告書.pdf"
+    assert restored_context.metadata == [{"key": "supplier", "type": "string", "value": "Dell"}]
 
 
 @pytest.mark.asyncio