feat(properties): add custom fields without a migration

Modelled on HubSpot's custom properties: the definition lives in a table,
the values live in each record's existing JSON attributes. Adding a field
is an INSERT, so it can happen mid-conversation through MCP and the next
record can use it immediately.

The cost of that is that the database enforces nothing about the values,
so data_type is enforced in application code and every write path has to
come through it — otherwise the type on a definition is decoration.
Values are coerced rather than merely checked, because callers are agents
and HTTP clients and "42" is a number expressed loosely; what is rejected
is genuinely ambiguous, like "quite large" for a number.

A value whose property has no definition is rejected rather than stored.
A typo sitting in the database looks exactly like data, and a registry
whose set of fields is not actually the set of fields is worse than none.
Workspaces with nothing defined keep the old free-form behaviour, so this
does not break attributes already in use.

name and data_type are not updatable: renaming orphans every value stored
under the old key, and retyping leaves values that no longer satisfy the
type. Deleting a definition leaves existing values alone rather than
rewriting every record, so undoing a mistaken delete is just recreating
the property.

Two ways in, sharing one validator: a logged-in person, and an integration
key for MaskanX. An agent gets no shortcut around the rules a person is
held to.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

alembic/versions/c2e5f8a41b76_add_custom_properties.py

@@ -0,0 +1,62 @@
+"""add crm_custom_properties
+
+Fields an operator (or an agent, through MCP) adds without a migration.
+
+The definition lives here; the values live in each record's existing
+`attributes` JSON. That split is the whole point: adding a field is an
+INSERT, so it can happen mid-conversation and be usable immediately.
+
+Revision ID: c2e5f8a41b76
+Revises: b1c4d7e29a03
+Create Date: 2026-08-04
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = 'c2e5f8a41b76'
+down_revision: Union[str, None] = 'b1c4d7e29a03'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'crm_custom_properties',
+        sa.Column('id', sa.String(length=36), nullable=False),
+        sa.Column('tenant_id', sa.String(length=36), nullable=False),
+        sa.Column('object_type', sa.String(length=40), nullable=False),
+        sa.Column('name', sa.String(length=80), nullable=False),
+        sa.Column('label', sa.String(length=160), nullable=False),
+        sa.Column('description', sa.Text(), nullable=True),
+        sa.Column('data_type', sa.String(length=24), nullable=False),
+        sa.Column('options', sa.JSON(), nullable=False),
+        sa.Column('is_required', sa.Boolean(), nullable=False),
+        sa.Column('created_by', sa.String(length=120), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(['tenant_id'], ['crm_tenants.id'], ondelete='CASCADE'),
+        sa.PrimaryKeyConstraint('id'),
+        # `name` is the key written into each record's attributes, so it has
+        # to be unique per object type — two definitions sharing a key would
+        # fight over the same stored value.
+        sa.UniqueConstraint(
+            'tenant_id', 'object_type', 'name',
+            name='uq_crm_custom_properties_object_name',
+        ),
+    )
+    op.create_index(
+        'ix_crm_custom_properties_tenant_object',
+        'crm_custom_properties',
+        ['tenant_id', 'object_type'],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        'ix_crm_custom_properties_tenant_object',
+        table_name='crm_custom_properties',
+    )
+    op.drop_table('crm_custom_properties')

app/api/contacts.py

@@ -18,6 +18,7 @@ from app.services import (
     add_audit,
     add_event,
     apply_updates,
+    validated_attributes,
     contact_to_out,
     model_or_404,
     verify_optional_reference,
@@ -83,7 +84,11 @@ def create_contact(payload: ContactCreate, user: Writer, db: Database) -> Contac
         user.tenant_id,
     )
     verify_optional_reference(db, User, payload.owner_id, user.tenant_id)
-    contact = Contact(tenant_id=user.tenant_id, **payload.model_dump())
+    values = payload.model_dump()
+    values["attributes"] = validated_attributes(
+        db, user.tenant_id, "contact", {}, values.get("attributes"),
+    )
+    contact = Contact(tenant_id=user.tenant_id, **values)
     db.add(contact)
     try:
         db.flush()
@@ -145,6 +150,11 @@ def update_contact(
         user.tenant_id,
     )
     verify_optional_reference(db, User, values.get("owner_id"), user.tenant_id)
+    if "attributes" in values:
+        values["attributes"] = validated_attributes(
+            db, user.tenant_id, "contact", contact.attributes,
+            values["attributes"],
+        )
     apply_updates(contact, values)
     add_audit(
         db,

app/api/properties.py

@@ -0,0 +1,205 @@
+"""Custom property definitions — fields added without a migration.
+
+Two ways in, on purpose:
+
+- `/properties` for a logged-in person, from the CRM's own settings screen.
+- `/integrations/properties` for MaskanX, authenticated by an integration
+  key. That is what an agent reaches through MCP when a conversation asks
+  for a new field.
+
+Both go through the same validation. An agent gets no shortcut around the
+rules a person is held to — if anything it needs them more, since it will
+happily invent a property name from a half-sentence.
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Header, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from app.core.security import CurrentUser, Database
+from app.models import CustomProperty
+from app.properties import PropertyError, validate_definition
+from app.schemas import (
+    CustomPropertyCreate,
+    CustomPropertyOut,
+    CustomPropertyUpdate,
+)
+
+router = APIRouter(prefix="/properties", tags=["Properties"])
+integration_router = APIRouter(prefix="/integrations", tags=["Integrations"])
+
+
+def _list_for(db, tenant_id: str, object_type: str | None) -> list[CustomProperty]:
+    query = select(CustomProperty).where(CustomProperty.tenant_id == tenant_id)
+    if object_type:
+        query = query.where(CustomProperty.object_type == object_type)
+    return list(db.scalars(query.order_by(CustomProperty.label)))
+
+
+def _create(
+    db,
+    tenant_id: str,
+    payload: CustomPropertyCreate,
+    created_by: str,
+) -> CustomProperty:
+    try:
+        validate_definition(
+            name=payload.name,
+            object_type=payload.object_type,
+            data_type=payload.data_type,
+            options=payload.options,
+        )
+    except PropertyError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+
+    prop = CustomProperty(
+        tenant_id=tenant_id,
+        object_type=payload.object_type,
+        name=payload.name,
+        label=payload.label,
+        description=payload.description,
+        data_type=payload.data_type,
+        options=payload.options,
+        is_required=payload.is_required,
+        created_by=created_by,
+    )
+    db.add(prop)
+    try:
+        db.commit()
+    except IntegrityError as exc:
+        db.rollback()
+        # Told plainly rather than as a constraint name: an agent that
+        # re-runs a request needs to know the field already exists, not
+        # that a unique index fired.
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=(
+                f"A property named {payload.name!r} already exists on "
+                f"{payload.object_type}."
+            ),
+        ) from exc
+    return prop
+
+
+@router.get("", response_model=list[CustomPropertyOut])
+def list_properties(
+    user: CurrentUser,
+    db: Database,
+    object_type: str | None = None,
+) -> list[CustomProperty]:
+    return _list_for(db, user.tenant_id, object_type)
+
+
+@router.post("", response_model=CustomPropertyOut, status_code=status.HTTP_201_CREATED)
+def create_property(
+    payload: CustomPropertyCreate,
+    user: CurrentUser,
+    db: Database,
+) -> CustomProperty:
+    return _create(db, user.tenant_id, payload, created_by=user.email)
+
+
+@router.patch("/{property_id}", response_model=CustomPropertyOut)
+def update_property(
+    property_id: str,
+    payload: CustomPropertyUpdate,
+    user: CurrentUser,
+    db: Database,
+) -> CustomProperty:
+    """Update a property's presentation, never its identity or type.
+
+    `name` and `data_type` are not updatable — see `CustomPropertyUpdate`.
+    Narrowing an enumeration is allowed even though records may already
+    hold a value that is no longer offered: those values stay readable and
+    simply cannot be chosen again, which is the behaviour an operator
+    retiring an option expects.
+    """
+    prop = db.scalar(
+        select(CustomProperty).where(
+            CustomProperty.id == property_id,
+            CustomProperty.tenant_id == user.tenant_id,
+        ),
+    )
+    if prop is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+
+    if payload.options is not None and prop.data_type != "enumeration":
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Options only apply to an enumeration, not to a {prop.data_type}.",
+        )
+    if payload.options is not None and not payload.options:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="An enumeration needs at least one option.",
+        )
+
+    for field, value in payload.model_dump(exclude_unset=True).items():
+        if value is not None:
+            setattr(prop, field, value)
+    db.commit()
+    return prop
+
+
+@router.delete("/{property_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_property(property_id: str, user: CurrentUser, db: Database) -> None:
+    """Remove a definition.
+
+    Values already written into records are left where they are. Rewriting
+    every record to strip a key would be a large, silent, irreversible
+    write in response to a small request; leaving them means undoing a
+    mistaken delete is just recreating the property.
+    """
+    prop = db.scalar(
+        select(CustomProperty).where(
+            CustomProperty.id == property_id,
+            CustomProperty.tenant_id == user.tenant_id,
+        ),
+    )
+    if prop is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+    db.delete(prop)
+    db.commit()
+
+
+# --- integration key access, for MaskanX and its MCP tools ---
+
+
+@integration_router.get("/properties", response_model=list[CustomPropertyOut])
+def list_properties_via_integration(
+    db: Database,
+    object_type: str | None = None,
+    integration_key: Annotated[str | None, Header(alias="X-Integration-Key")] = None,
+) -> list[CustomProperty]:
+    from app.api.integrations import _authenticate_integration
+
+    credential = _authenticate_integration(db, integration_key)
+    properties = _list_for(db, credential.tenant_id, object_type)
+    db.commit()
+    return properties
+
+
+@integration_router.post(
+    "/properties",
+    response_model=CustomPropertyOut,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_property_via_integration(
+    payload: CustomPropertyCreate,
+    db: Database,
+    integration_key: Annotated[str | None, Header(alias="X-Integration-Key")] = None,
+) -> CustomProperty:
+    from app.api.integrations import _authenticate_integration
+
+    credential = _authenticate_integration(db, integration_key)
+    return _create(
+        db,
+        credential.tenant_id,
+        payload,
+        created_by=f"integration:{credential.name}",
+    )

app/api/router.py

@@ -8,6 +8,7 @@ from app.api import (
     contacts,
     dashboard,
     integrations,
+    properties,
     leads,
 )
 
@@ -19,5 +20,7 @@ api_router.include_router(leads.router)
 api_router.include_router(activities.router)
 api_router.include_router(catalog.router)
 api_router.include_router(campaigns.router)
+api_router.include_router(properties.router)
+api_router.include_router(properties.integration_router)
 api_router.include_router(integrations.router)
 

app/models.py

@@ -628,3 +628,54 @@ class Campaign(Base, TimestampMixin):
     metadata_json: Mapped[dict[str, Any]] = mapped_column(
         JSON, default=dict, nullable=False,
     )
+
+
+class CustomProperty(Base, TimestampMixin):
+    """A field an operator added, without a schema migration.
+
+    Modelled on how HubSpot treats properties: the definition lives in a
+    table, and the values live in the record's existing `attributes` JSON.
+    That is the whole point — adding a field is an INSERT, not a migration,
+    so an agent can create one mid-conversation and the next record can use
+    it immediately.
+
+    The cost of that choice is that values are not typed by the database,
+    so `data_type` is enforced in application code on write. Anything that
+    writes a custom property value has to go through that validation; a
+    direct UPDATE to `attributes` bypasses it.
+
+    `name` is the stable key stored inside `attributes`; `label` is what a
+    person reads. They are kept separate for the same reason HubSpot keeps
+    them separate: renaming a label must not orphan every value already
+    written under the old key.
+    """
+
+    __tablename__ = "crm_custom_properties"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id",
+            "object_type",
+            "name",
+            name="uq_crm_custom_properties_object_name",
+        ),
+        Index("ix_crm_custom_properties_tenant_object", "tenant_id", "object_type"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_id)
+    tenant_id: Mapped[str] = mapped_column(
+        ForeignKey("crm_tenants.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    # Which kind of record carries it: contact, lead, organization, campaign.
+    object_type: Mapped[str] = mapped_column(String(40), nullable=False)
+    name: Mapped[str] = mapped_column(String(80), nullable=False)
+    label: Mapped[str] = mapped_column(String(160), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text)
+    data_type: Mapped[str] = mapped_column(String(24), nullable=False, default="string")
+    # Allowed values for an enumeration. Empty for every other type.
+    options: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    is_required: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
+    # Where it came from. An agent-created field is marked so a human can
+    # tell at a glance which fields they defined and which one a
+    # conversation produced.
+    created_by: Mapped[str | None] = mapped_column(String(120))

app/properties.py

@@ -0,0 +1,173 @@
+"""Validation for custom property definitions and their values.
+
+Custom property values live in a JSON column, so the database enforces
+nothing about them. Everything that writes one has to come through here, or
+the `data_type` on the definition is decoration.
+
+Pure functions: no session, no I/O, so the rules can be tested directly.
+"""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Any
+
+# A property name becomes a key inside a record's `attributes` JSON and is
+# referenced by agents and API callers, so it is restricted to something
+# that survives being a JSON key, a query parameter and a column header.
+NAME_PATTERN = r"^[a-z][a-z0-9_]{1,63}$"
+
+DATA_TYPES = ("string", "number", "boolean", "date", "enumeration")
+
+OBJECT_TYPES = ("contact", "lead", "organization", "campaign")
+
+
+class PropertyError(ValueError):
+    """Raised when a definition or a value is not usable."""
+
+
+def validate_definition(
+    *,
+    name: str,
+    object_type: str,
+    data_type: str,
+    options: list[str] | None,
+) -> None:
+    """Check a property definition before it is stored.
+
+    Rejects an enumeration with no options: a field whose only valid values
+    are none of them cannot ever be filled in, and an agent creating one by
+    mistake would produce a form nobody can submit.
+    """
+    import re
+
+    if object_type not in OBJECT_TYPES:
+        raise PropertyError(
+            f"Unknown object type {object_type!r}. "
+            f"Available: {', '.join(OBJECT_TYPES)}.",
+        )
+    if data_type not in DATA_TYPES:
+        raise PropertyError(
+            f"Unknown data type {data_type!r}. Available: {', '.join(DATA_TYPES)}.",
+        )
+    if not re.match(NAME_PATTERN, name or ""):
+        raise PropertyError(
+            f"{name!r} is not a usable property name. Use lowercase letters, "
+            f"digits and underscores, starting with a letter, e.g. "
+            f"'budget_range'.",
+        )
+    if data_type == "enumeration" and not options:
+        raise PropertyError(
+            "An enumeration needs at least one option; otherwise no value "
+            "could ever be valid for it.",
+        )
+    if data_type != "enumeration" and options:
+        raise PropertyError(
+            f"Options only apply to an enumeration, not to a {data_type}.",
+        )
+
+
+def coerce_value(definition: Any, value: Any) -> Any:
+    """Return `value` in the shape its definition calls for.
+
+    Coerces rather than merely checking, because callers are agents and
+    HTTP clients: a number arriving as the string "42" is a well-formed
+    intention expressed loosely, and rejecting it would be pedantry. What
+    is rejected is anything genuinely ambiguous — "quite large" is not a
+    number by any reading.
+
+    `None` clears the value and is always allowed here; whether a property
+    may be empty is `is_required`'s business, checked separately, because
+    the two questions have different answers on a partial update.
+    """
+    if value is None:
+        return None
+
+    data_type = definition.data_type
+
+    if data_type == "string":
+        return str(value)
+
+    if data_type == "number":
+        try:
+            number = float(value)
+        except (TypeError, ValueError) as exc:
+            raise PropertyError(
+                f"{definition.label!r} expects a number, not {value!r}.",
+            ) from exc
+        # Keep whole numbers whole: 3.0 stored for a count reads oddly and
+        # round-trips into JSON as 3.0 forever.
+        return int(number) if number.is_integer() else number
+
+    if data_type == "boolean":
+        if isinstance(value, bool):
+            return value
+        text = str(value).strip().lower()
+        if text in ("true", "yes", "1"):
+            return True
+        if text in ("false", "no", "0"):
+            return False
+        raise PropertyError(
+            f"{definition.label!r} expects true or false, not {value!r}.",
+        )
+
+    if data_type == "date":
+        if isinstance(value, (date, datetime)):
+            return value.isoformat()
+        try:
+            return date.fromisoformat(str(value)[:10]).isoformat()
+        except ValueError as exc:
+            raise PropertyError(
+                f"{definition.label!r} expects a date as YYYY-MM-DD, "
+                f"not {value!r}.",
+            ) from exc
+
+    if data_type == "enumeration":
+        text = str(value)
+        if text not in (definition.options or []):
+            raise PropertyError(
+                f"{text!r} is not one of the allowed values for "
+                f"{definition.label!r}: "
+                f"{', '.join(definition.options or []) or 'none'}.",
+            )
+        return text
+
+    raise PropertyError(f"Unknown data type {data_type!r}.")
+
+
+def apply_values(
+    definitions: list[Any],
+    current: dict[str, Any],
+    incoming: dict[str, Any],
+) -> dict[str, Any]:
+    """Merge validated custom property values into a record's attributes.
+
+    Only keys that have a definition are touched. An unknown key is
+    rejected rather than stored: silently accepting one would let a typo
+    ("budjet_range") sit in the database looking like data, and the whole
+    reason for a registry is that the set of fields is knowable.
+
+    Required properties are enforced only against what is being written,
+    not against the merged result — a partial update that does not mention
+    a required field is not an attempt to clear it.
+    """
+    by_name = {definition.name: definition for definition in definitions}
+
+    unknown = sorted(set(incoming) - set(by_name))
+    if unknown:
+        known = ", ".join(sorted(by_name)) or "none"
+        raise PropertyError(
+            f"No such property: {', '.join(unknown)}. Defined here: {known}.",
+        )
+
+    merged = dict(current)
+    for name, raw in incoming.items():
+        definition = by_name[name]
+        coerced = coerce_value(definition, raw)
+        if coerced is None and definition.is_required:
+            raise PropertyError(f"{definition.label!r} is required and cannot be empty.")
+        if coerced is None:
+            merged.pop(name, None)
+        else:
+            merged[name] = coerced
+    return merged

app/schemas.py

@@ -470,3 +470,42 @@ class CampaignOut(ApiModel):
     # Meta has not attributed yet, and the gap between the two is worth
     # seeing rather than hiding.
     crm_leads: int = 0
+
+
+class CustomPropertyCreate(ApiModel):
+    object_type: str = Field(min_length=2, max_length=40)
+    name: str = Field(min_length=2, max_length=80)
+    label: str = Field(min_length=1, max_length=160)
+    description: str | None = None
+    data_type: str = Field(default="string", max_length=24)
+    options: list[str] = Field(default_factory=list)
+    is_required: bool = False
+
+
+class CustomPropertyUpdate(ApiModel):
+    """A partial update.
+
+    `name`, `object_type` and `data_type` are absent on purpose. They are
+    the identity and meaning of the field: changing a name orphans every
+    value already stored under it, and changing a type leaves stored values
+    that no longer satisfy it. Delete and recreate instead — deliberately
+    more effort, because it loses data.
+    """
+
+    label: str | None = Field(default=None, min_length=1, max_length=160)
+    description: str | None = None
+    options: list[str] | None = None
+    is_required: bool | None = None
+
+
+class CustomPropertyOut(ApiModel):
+    id: str
+    object_type: str
+    name: str
+    label: str
+    description: str | None
+    data_type: str
+    options: list[str]
+    is_required: bool
+    created_by: str | None
+    created_at: datetime

app/services.py

@@ -247,3 +247,52 @@ def next_lead_position(db: Session, tenant_id: str, stage_id: str) -> int:
 
 def decimal_or_zero(value: Decimal | None) -> Decimal:
     return value or Decimal("0")
+
+
+def validated_attributes(
+    db: Session,
+    tenant_id: str,
+    object_type: str,
+    current: dict[str, Any],
+    incoming: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Merge incoming custom property values after validating them.
+
+    Custom property values live in a JSON column, so the database enforces
+    nothing about them — every write path has to come through here or the
+    `data_type` on a definition is decoration.
+
+    Values whose property has no definition are rejected rather than
+    stored. Silently accepting one would let a typo sit in the database
+    looking like data, and a registry whose set of fields is not actually
+    the set of fields is worse than none.
+    """
+    from sqlalchemy import select
+
+    from app.models import CustomProperty
+    from app.properties import PropertyError, apply_values
+
+    if incoming is None:
+        return current
+
+    definitions = list(
+        db.scalars(
+            select(CustomProperty).where(
+                CustomProperty.tenant_id == tenant_id,
+                CustomProperty.object_type == object_type,
+            ),
+        ),
+    )
+    # Nothing defined yet: keep the previous behaviour of storing whatever
+    # arrives, so adding this validation does not break workspaces that
+    # already use `attributes` as a free-form bag.
+    if not definitions:
+        return incoming
+
+    try:
+        return apply_values(definitions, current, incoming)
+    except PropertyError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc

tests/test_properties.py

@@ -0,0 +1,372 @@
+"""Custom properties: fields added without a migration.
+
+The definitions live in a table and the values in each record's JSON
+`attributes`, so nothing in the database enforces a type. These tests are
+mostly about the validation that stands in for that, and about the two ways
+in — a logged-in person, and an integration key, which is what an agent
+reaches through MCP.
+"""
+import pytest
+from fastapi.testclient import TestClient
+
+from app.properties import PropertyError, coerce_value, validate_definition
+
+
+class _Definition:
+    """Stands in for a stored definition in the pure-function tests."""
+
+    def __init__(self, data_type, options=None, label="Field", is_required=False):
+        self.data_type = data_type
+        self.options = options or []
+        self.label = label
+        self.is_required = is_required
+        self.name = "field"
+
+
+# --- defining ---
+
+
+def test_a_name_must_be_a_usable_key():
+    """It becomes a JSON key and a query parameter, not just a label."""
+    for bad in ("Budget Range", "1st_field", "budget-range", "", "A"):
+        with pytest.raises(PropertyError):
+            validate_definition(
+                name=bad, object_type="contact", data_type="string", options=None,
+            )
+
+
+def test_a_good_name_is_accepted():
+    validate_definition(
+        name="budget_range", object_type="contact", data_type="string", options=None,
+    )
+
+
+def test_an_enumeration_with_no_options_is_refused():
+    """No value could ever be valid for it."""
+    with pytest.raises(PropertyError, match="at least one option"):
+        validate_definition(
+            name="stage", object_type="lead", data_type="enumeration", options=[],
+        )
+
+
+def test_options_on_a_non_enumeration_are_refused():
+    with pytest.raises(PropertyError, match="only apply to an enumeration"):
+        validate_definition(
+            name="notes", object_type="lead", data_type="string", options=["a"],
+        )
+
+
+def test_an_unknown_object_type_lists_the_real_ones():
+    with pytest.raises(PropertyError, match="contact"):
+        validate_definition(
+            name="x", object_type="spaceship", data_type="string", options=None,
+        )
+
+
+# --- values ---
+
+
+def test_a_number_arriving_as_a_string_is_accepted():
+    """Callers are agents and HTTP clients; "42" is a number loosely typed."""
+    assert coerce_value(_Definition("number"), "42") == 42
+
+
+def test_a_whole_number_stays_whole():
+    assert coerce_value(_Definition("number"), "3.0") == 3
+
+
+def test_a_decimal_survives():
+    assert coerce_value(_Definition("number"), "3.5") == 3.5
+
+
+def test_something_genuinely_unnumeric_is_refused():
+    with pytest.raises(PropertyError, match="expects a number"):
+        coerce_value(_Definition("number"), "quite large")
+
+
+@pytest.mark.parametrize("value", ["true", "Yes", "1", True])
+def test_truthy_spellings_of_boolean(value):
+    assert coerce_value(_Definition("boolean"), value) is True
+
+
+@pytest.mark.parametrize("value", ["false", "No", "0", False])
+def test_falsy_spellings_of_boolean(value):
+    assert coerce_value(_Definition("boolean"), value) is False
+
+
+def test_an_ambiguous_boolean_is_refused():
+    with pytest.raises(PropertyError, match="true or false"):
+        coerce_value(_Definition("boolean"), "maybe")
+
+
+def test_a_date_is_normalised_to_iso():
+    assert coerce_value(_Definition("date"), "2026-08-04") == "2026-08-04"
+
+
+def test_a_datetime_string_keeps_only_the_date():
+    assert coerce_value(_Definition("date"), "2026-08-04T10:30:00Z") == "2026-08-04"
+
+
+def test_an_unparseable_date_says_what_it_wanted():
+    with pytest.raises(PropertyError, match="YYYY-MM-DD"):
+        coerce_value(_Definition("date"), "next tuesday")
+
+
+def test_an_enumeration_value_must_be_offered():
+    definition = _Definition("enumeration", options=["hot", "warm"], label="Temp")
+
+    assert coerce_value(definition, "hot") == "hot"
+    with pytest.raises(PropertyError, match="hot, warm"):
+        coerce_value(definition, "lukewarm")
+
+
+def test_none_always_clears_a_value():
+    """Whether it may be empty is is_required's business, checked elsewhere."""
+    assert coerce_value(_Definition("number", is_required=True), None) is None
+
+
+# --- the API ---
+
+
+@pytest.fixture()
+def service_key(client: TestClient, auth_headers: dict[str, str]) -> str:
+    created = client.post(
+        "/api/v1/integrations/credentials",
+        headers=auth_headers,
+        json={"name": "MaskanX"},
+    )
+    return created.json()["key"]
+
+
+def _definition(**overrides) -> dict:
+    payload = {
+        "object_type": "contact",
+        "name": "budget_range",
+        "label": "Budget range",
+        "data_type": "enumeration",
+        "options": ["50-70 lakh", "70-90 lakh", "1 crore+"],
+    }
+    payload.update(overrides)
+    return payload
+
+
+def test_a_property_can_be_created_and_listed(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    created = client.post(
+        "/api/v1/properties", headers=auth_headers, json=_definition(),
+    )
+    assert created.status_code == 201, created.text
+
+    listed = client.get("/api/v1/properties", headers=auth_headers).json()
+    assert [p["name"] for p in listed] == ["budget_range"]
+    assert listed[0]["options"] == ["50-70 lakh", "70-90 lakh", "1 crore+"]
+
+
+def test_properties_can_be_filtered_by_object_type(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    client.post("/api/v1/properties", headers=auth_headers, json=_definition())
+    client.post(
+        "/api/v1/properties",
+        headers=auth_headers,
+        json=_definition(
+            object_type="lead", name="site_visit_done",
+            label="Site visit done", data_type="boolean", options=[],
+        ),
+    )
+
+    contacts = client.get(
+        "/api/v1/properties?object_type=contact", headers=auth_headers,
+    ).json()
+
+    assert [p["name"] for p in contacts] == ["budget_range"]
+
+
+def test_the_same_property_twice_is_a_readable_conflict(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    """An agent re-running a request needs to know it already exists."""
+    client.post("/api/v1/properties", headers=auth_headers, json=_definition())
+    again = client.post(
+        "/api/v1/properties", headers=auth_headers, json=_definition(),
+    )
+
+    assert again.status_code == 409
+    assert "already exists" in again.json()["error"]["message"]
+
+
+def test_an_invalid_definition_is_refused_with_a_reason(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    response = client.post(
+        "/api/v1/properties",
+        headers=auth_headers,
+        json=_definition(name="Budget Range"),
+    )
+
+    assert response.status_code == 422
+    assert "budget_range" in response.json()["error"]["message"]
+
+
+def test_a_label_can_be_changed_without_orphaning_values(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    created = client.post(
+        "/api/v1/properties", headers=auth_headers, json=_definition(),
+    ).json()
+
+    updated = client.patch(
+        f"/api/v1/properties/{created['id']}",
+        headers=auth_headers,
+        json={"label": "Buying budget"},
+    )
+
+    assert updated.status_code == 200
+    assert updated.json()["label"] == "Buying budget"
+    # The key values are stored under is untouched.
+    assert updated.json()["name"] == "budget_range"
+
+
+def test_a_property_can_be_deleted(client: TestClient, auth_headers: dict) -> None:
+    created = client.post(
+        "/api/v1/properties", headers=auth_headers, json=_definition(),
+    ).json()
+
+    assert (
+        client.delete(
+            f"/api/v1/properties/{created['id']}", headers=auth_headers,
+        ).status_code
+        == 204
+    )
+    assert client.get("/api/v1/properties", headers=auth_headers).json() == []
+
+
+def test_reading_properties_needs_a_login(client: TestClient) -> None:
+    assert client.get("/api/v1/properties").status_code == 401
+
+
+# --- through an integration key, which is what MCP uses ---
+
+
+def test_an_integration_can_create_a_property(
+    client: TestClient, auth_headers: dict, service_key: str,
+) -> None:
+    created = client.post(
+        "/api/v1/integrations/properties",
+        headers={"X-Integration-Key": service_key},
+        json=_definition(),
+    )
+
+    assert created.status_code == 201, created.text
+    # Marked as machine-made, so a human can tell which fields a
+    # conversation produced.
+    assert created.json()["created_by"].startswith("integration:")
+
+
+def test_an_integration_gets_no_shortcut_around_validation(
+    client: TestClient, service_key: str,
+) -> None:
+    response = client.post(
+        "/api/v1/integrations/properties",
+        headers={"X-Integration-Key": service_key},
+        json=_definition(data_type="enumeration", options=[]),
+    )
+
+    assert response.status_code == 422
+
+
+def test_an_integration_property_is_visible_to_people(
+    client: TestClient, auth_headers: dict, service_key: str,
+) -> None:
+    """A field created in chat has to show up in the CRM's own screens."""
+    client.post(
+        "/api/v1/integrations/properties",
+        headers={"X-Integration-Key": service_key},
+        json=_definition(),
+    )
+
+    listed = client.get("/api/v1/properties", headers=auth_headers).json()
+
+    assert [p["name"] for p in listed] == ["budget_range"]
+
+
+def test_creating_a_property_needs_a_valid_key(client: TestClient) -> None:
+    response = client.post("/api/v1/integrations/properties", json=_definition())
+
+    assert response.status_code == 401
+
+
+# --- values on records ---
+
+
+def _contact(client, auth_headers, attributes):
+    return client.post(
+        "/api/v1/contacts",
+        headers=auth_headers,
+        json={
+            "first_name": "Asha",
+            "last_name": "Menon",
+            "primary_email": "asha@example.com",
+            "attributes": attributes,
+        },
+    )
+
+
+def test_a_value_is_validated_against_its_definition(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    client.post("/api/v1/properties", headers=auth_headers, json=_definition())
+
+    good = _contact(client, auth_headers, {"budget_range": "70-90 lakh"})
+    assert good.status_code == 201, good.text
+    assert good.json()["attributes"]["budget_range"] == "70-90 lakh"
+
+
+def test_a_value_outside_the_enumeration_is_refused(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    client.post("/api/v1/properties", headers=auth_headers, json=_definition())
+
+    response = _contact(client, auth_headers, {"budget_range": "a few rupees"})
+
+    assert response.status_code == 422
+    assert "70-90 lakh" in response.json()["error"]["message"]
+
+
+def test_an_undefined_property_is_refused_rather_than_stored(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    """A typo left in the database looks exactly like data."""
+    client.post("/api/v1/properties", headers=auth_headers, json=_definition())
+
+    response = _contact(client, auth_headers, {"budjet_range": "70-90 lakh"})
+
+    assert response.status_code == 422
+    assert "budjet_range" in response.json()["error"]["message"]
+
+
+def test_attributes_stay_free_form_when_nothing_is_defined(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    """Adding this validation must not break workspaces already using them."""
+    response = _contact(client, auth_headers, {"anything": "at all"})
+
+    assert response.status_code == 201
+    assert response.json()["attributes"] == {"anything": "at all"}
+
+
+def test_a_number_property_is_coerced_on_the_way_in(
+    client: TestClient, auth_headers: dict,
+) -> None:
+    client.post(
+        "/api/v1/properties",
+        headers=auth_headers,
+        json=_definition(
+            name="bedrooms", label="Bedrooms", data_type="number", options=[],
+        ),
+    )
+
+    response = _contact(client, auth_headers, {"bedrooms": "3"})
+
+    assert response.json()["attributes"]["bedrooms"] == 3