feat(#359): W4a Phase 1 — Import/Export Contribution-Architektur Kern

- Format-Registry (app/core/importexport_registry.py): FormatHandler-Protokoll,
  available_for() Schnittmenge, Singleton + Testing-Reset
- importexport_formats-Plugin (csv/json/xlsx), lifecycle-korrekt: on_activate
  registriert Handler in der Core-Registry, on_deactivate unregistriert
- ContactsContract: importexport-Beitrag (ie_*-Methoden) — Contacts besitzt
  seine Import/Export-Fachlogik jetzt selbst
- import_export_service.py: generische Engine, delegiert generisch ueber
  registry.list_discovered() an den besitzenden Contract (keine hartcodierten
  Plugin-Namen mehr); Signaturen identisch
- Funktionserhalt bewiesen: 45/45 import_export-Suite passed (inkl.
  Fehler-Multiplizitaet: 2 failed rows -> 3 total_errors, erreicht via
  ie_required-Weitergabe + ie_row_valid-Nur-contacts-Early-Return)

fixes #359 (Phase 1)

app/core/importexport_registry.py

@@ -0,0 +1,90 @@
+"""Import/Export format registry — formats are provided by plugins.
+
+A **format plugin** (e.g. ``importexport_formats``) registers handlers for
+its supported file formats (csv, json, xlsx, ...).  An **entity module**
+(e.g. contacts) contributes via its contract which formats it supports and
+provides the import/export logic for its data.
+
+The core orchestrator (routes + background jobs) resolves the intersection:
+
+    capabilities(entity) = entity.formats ∩ format_registry.registered
+
+The security policy layer (sensitive-data filter, tenant scoping, audit)
+runs in the orchestrator, independently of the module contribution.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Protocol, runtime_checkable
+
+logger = logging.getLogger(__name__)
+
+
+@runtime_checkable
+class FormatHandler(Protocol):
+    """Handler for one file format (e.g. csv, json, xlsx).
+
+    Provided by a format plugin via the registry.
+    """
+
+    format_id: str
+
+    @staticmethod
+    def parse(content: bytes) -> list[dict[str, Any]]:
+        """Parse file content into a list of row dicts."""
+        ...
+
+    @staticmethod
+    def serialize(rows: list[dict[str, Any]], headers: list[str]) -> bytes:
+        """Serialize rows into file bytes with the given column order."""
+        ...
+
+
+class ImportExportFormatRegistry:
+    """Registry for file-format handlers contributed by format plugins."""
+
+    def __init__(self) -> None:
+        self._formats: dict[str, FormatHandler] = {}
+
+    def register(self, handler: FormatHandler) -> None:
+        """Register (or replace) a format handler."""
+        self._formats[handler.format_id] = handler
+        logger.debug("Registered import/export format '%s'", handler.format_id)
+
+    def unregister(self, format_id: str) -> None:
+        self._formats.pop(format_id, None)
+        logger.debug("Unregistered import/export format '%s'", format_id)
+
+    def get(self, format_id: str) -> FormatHandler | None:
+        return self._formats.get(format_id)
+
+    def list_formats(self) -> list[str]:
+        return sorted(self._formats.keys())
+
+    def available_for(self, entity_formats: list[str]) -> list[str]:
+        """Return the intersection of an entity's declared formats and
+        the currently registered (plugin-provided) format handlers."""
+        return sorted(set(entity_formats) & set(self._formats.keys()))
+
+    def clear(self) -> None:
+        """Clear all registrations (testing only)."""
+        self._formats.clear()
+
+
+# ─── module-level singleton ─────────────────────────────────────────────────
+
+_registry: ImportExportFormatRegistry | None = None
+
+
+def get_format_registry() -> ImportExportFormatRegistry:
+    global _registry
+    if _registry is None:
+        _registry = ImportExportFormatRegistry()
+    return _registry
+
+
+def reset_format_registry_for_testing() -> ImportExportFormatRegistry:
+    global _registry
+    _registry = ImportExportFormatRegistry()
+    return _registry

app/plugins/builtins/contacts/contracts.py

@@ -106,6 +106,203 @@ class ContactsContract:
             c.deleted_at = datetime.now(UTC)
         return {"contacts_soft_deleted": len(contacts)}
 
+    # ─── Import/Export contribution (W4a, Spec #359) ───
+    # The contacts plugin owns its import/export domain logic; the core
+    # orchestrator resolves formats via the format registry and enforces
+    # the security policy (sensitive filter, tenant scoping, audit).
+
+    IE_COLUMNS = {
+        "contacts": ["firstname", "surname", "email", "phone", "mobile", "function", "department"],
+        "companies": ["name", "industry", "phone", "email", "website"],
+    }
+    IE_TARGET_FIELDS = {
+        "contacts": ["firstname", "surname", "email", "phone", "mobile", "function", "department"],
+        "companies": ["name", "industry", "phone", "email", "website"],
+    }
+    IE_VALIDATORS = {
+        "contacts": {"email": {"type": "email"}},
+        "companies": {"email": {"type": "email"}, "website": {"type": "url"}},
+    }
+
+    @staticmethod
+    def importexport_entities() -> list[str]:
+        """Entity types offered by this plugin's import/export."""
+        return ["contacts", "companies"]
+
+    @staticmethod
+    def importexport_formats() -> list[str]:
+        """File formats this plugin's import/export supports."""
+        return ["csv", "json", "xlsx"]
+
+    @staticmethod
+    def ie_columns(entity_type: str) -> list[str]:
+        return list(ContactsContract.IE_COLUMNS[entity_type])
+
+    @staticmethod
+    def ie_target_fields(entity_type: str) -> list[str]:
+        return list(ContactsContract.IE_TARGET_FIELDS[entity_type])
+
+    @staticmethod
+    def ie_validators(entity_type: str) -> dict[str, dict]:
+        return dict(ContactsContract.IE_VALIDATORS[entity_type])
+
+    @staticmethod
+    def ie_normalize_row(entity_type: str, row: dict[str, str]) -> dict[str, str]:
+        """Normalize an imported row to unified field names."""
+        if entity_type == "contacts":
+            firstname = (row.get("firstname") or row.get("first_name") or "").strip()
+            surname = (row.get("surname") or row.get("last_name") or "").strip()
+            row["firstname"] = firstname
+            row["surname"] = surname
+            if "email" not in row and "email_address" in row:
+                row["email"] = row["email_address"]
+            if "mobile" not in row and "phone_2" in row:
+                row["mobile"] = row["phone_2"]
+            if "function" not in row and "position" in row:
+                row["function"] = row["position"]
+            return row
+        name = (row.get("name") or row.get("company") or row.get("company_name") or "").strip()
+        row["name"] = name
+        if "email" not in row and "email_address" in row:
+            row["email"] = row["email_address"]
+        if "website" not in row and "url" in row:
+            row["website"] = row["url"]
+        if "website" not in row and "homepage" in row:
+            row["website"] = row["homepage"]
+        return row
+
+    @staticmethod
+    def ie_required(entity_type: str) -> list[str]:
+        """Required columns enforced by generic validate_row (old semantics)."""
+        return ["name"] if entity_type == "companies" else []
+
+    @staticmethod
+    def ie_row_valid(entity_type: str, row: dict[str, str]) -> tuple[bool, str]:
+        """Early either-or check for contacts only.
+
+        NOTE: companies' required-name check must NOT happen here —
+        generic validate_row(row, ['name'], validators) must see the row
+        so a missing name AND an invalid email yield two errors (as the
+        original semantics did).
+        """
+        if entity_type == "contacts":
+            if not row.get("firstname") and not row.get("surname"):
+                return False, "Missing required field: firstname or surname"
+        return True, ""
+
+    @staticmethod
+    async def ie_fetch_rows(
+        db: AsyncSession,
+        tenant_id: Any,
+        entity_type: str,
+        user_id: Any = None,
+        is_system_admin: bool = False,
+    ) -> tuple[list[str], list[dict[str, Any]]]:
+        """Fetch export rows (headers + row dicts), visibility-filtered."""
+        q = select(Contact).where(
+            Contact.tenant_id == tenant_id,
+            Contact.deleted_at.is_(None),
+        )
+        if entity_type == "companies":
+            q = q.where(Contact.type == "company").order_by(Contact.name)
+        else:
+            q = q.order_by(Contact.surname, Contact.firstname)
+        if user_id:
+            q = await apply_visibility_filter(
+                db, q, "contact", Contact, user_id, tenant_id, is_system_admin
+            )
+        records = (await db.execute(q)).scalars().all()
+        if entity_type == "companies":
+            headers = ["id", "type", "name", "email", "phone", "website", "city", "postalcode", "country"]
+            rows = [
+                {
+                    "id": str(c.id),
+                    "type": c.type or "company",
+                    "name": c.name or "",
+                    "email": c.email_1 or "",
+                    "phone": c.phone_1 or "",
+                    "website": c.website or "",
+                    "city": c.mailing_city or "",
+                    "postalcode": c.mailing_postalcode or "",
+                    "country": c.mailing_country or "",
+                }
+                for c in records
+            ]
+        else:
+            headers = ["id", "type", "firstname", "surname", "name", "email", "phone", "mobile", "city", "postalcode", "country"]
+            rows = [
+                {
+                    "id": str(c.id),
+                    "type": c.type or "person",
+                    "firstname": c.firstname or "",
+                    "surname": c.surname or "",
+                    "name": c.name or "",
+                    "email": c.email_1 or "",
+                    "phone": c.phone_1 or "",
+                    "mobile": c.phone_2 or "",
+                    "city": c.mailing_city or "",
+                    "postalcode": c.mailing_postalcode or "",
+                    "country": c.mailing_country or "",
+                }
+                for c in records
+            ]
+        return headers, rows
+
+    @staticmethod
+    async def ie_persist_row(
+        db: AsyncSession,
+        tenant_id: Any,
+        user_id: Any,
+        entity_type: str,
+        row: dict[str, str],
+    ) -> dict[str, Any]:
+        """Persist one imported row as Contact; returns the serialized record."""
+        from app.core.audit import log_audit
+        from app.services.contact_service import _serialize_contact
+
+        if entity_type == "companies":
+            contact = Contact(
+                tenant_id=tenant_id,
+                type="company",
+                name=row["name"].strip(),
+                displayname=row["name"].strip(),
+                email_1=row.get("email", "").strip() or None,
+                phone_1=row.get("phone", "").strip() or None,
+                website=row.get("website", "").strip() or None,
+                owner_id=user_id,
+                created_by=user_id,
+                updated_by=user_id,
+            )
+        else:
+            contact = Contact(
+                tenant_id=tenant_id,
+                type="person",
+                firstname=row["firstname"].strip() or None,
+                surname=row["surname"].strip() or None,
+                displayname=f"{row['firstname']} {row['surname']}".strip(),
+                email_1=row.get("email", "").strip() or None,
+                phone_1=row.get("phone", "").strip() or None,
+                phone_2=row.get("mobile", "").strip() or None,
+                owner_id=user_id,
+                created_by=user_id,
+                updated_by=user_id,
+            )
+        db.add(contact)
+        await db.flush()
+        await log_audit(
+            db,
+            tenant_id,
+            user_id,
+            "import",
+            "contact",
+            contact.id,
+            changes={
+                "type": entity_type.rstrip("s"),
+                "name": contact.name or f"{contact.firstname} {contact.surname}",
+            },
+        )
+        return _serialize_contact(contact)
+
     @classmethod
     def get_function(cls, name: str):
         """Return a callable exposed by this contract, or None if absent."""

app/plugins/builtins/importexport_formats/__init__.py

@@ -0,0 +1,5 @@
+"""Import/export formats plugin — provides the standard file formats."""
+
+from app.plugins.builtins.importexport_formats.plugin import ImportExportFormatsPlugin
+
+__all__ = ["ImportExportFormatsPlugin"]

app/plugins/builtins/importexport_formats/formats.py

@@ -0,0 +1,48 @@
+"""Standard format handlers — thin wrappers around the core helpers.
+
+No business logic here: parse/serialize only. Business logic (columns,
+normalization, validation, persistence) lives in the entity module's
+contract contribution (e.g. ContactsContract.importexport).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.services import import_export_helpers as helpers
+
+
+class CsvFormat:
+    format_id = "csv"
+
+    @staticmethod
+    def parse(filename: str, content: bytes) -> list[dict[str, Any]]:
+        return helpers.parse_csv(content)
+
+    @staticmethod
+    def serialize(rows: list[dict[str, Any]], headers: list[str]) -> bytes:
+        return helpers.write_csv(rows, headers)
+
+
+class JsonFormat:
+    format_id = "json"
+
+    @staticmethod
+    def parse(filename: str, content: bytes) -> list[dict[str, Any]]:
+        return helpers.parse_json(content)
+
+    @staticmethod
+    def serialize(rows: list[dict[str, Any]], headers: list[str]) -> bytes:
+        return helpers.write_json(rows)
+
+
+class XlsxFormat:
+    format_id = "xlsx"
+
+    @staticmethod
+    def parse(filename: str, content: bytes) -> list[dict[str, Any]]:
+        return helpers.parse_xlsx(content)
+
+    @staticmethod
+    def serialize(rows: list[dict[str, Any]], headers: list[str]) -> bytes:
+        return helpers.write_xlsx(rows, headers)

app/plugins/builtins/importexport_formats/plugin.py

@@ -0,0 +1,52 @@
+"""Import/export formats plugin — provides standard file formats as plugins.
+
+Part of the W4a contribution architecture (Spec #359): file formats live in
+this plugin instead of being hardcoded in the core orchestrator. New formats
+(e.g. PDF later) are added as additional format plugins without touching the
+core.
+
+Lifecycle: registers the bundled handlers (csv, json, xlsx) in the core
+format registry on activate, unregisters them on deactivate.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.core.importexport_registry import get_format_registry
+from app.plugins.base import BasePlugin
+from app.plugins.builtins.importexport_formats.formats import CsvFormat, JsonFormat, XlsxFormat
+from app.plugins.manifest import PluginManifest
+
+
+class ImportExportFormatsPlugin(BasePlugin):
+    """Bundles the standard import/export file formats."""
+
+    manifest = PluginManifest(
+        name="importexport_formats",
+        version="1.0.0",
+        display_name="Import/Export Formate",
+        description="Standard-Formate für Import/Export: CSV, JSON, XLSX.",
+        dependencies=[],
+        routes=[],
+        events=[],
+        migrations=[],
+        permissions=[],
+        is_core=True,
+    )
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._handlers: list[Any] = [CsvFormat(), JsonFormat(), XlsxFormat()]
+
+    async def on_activate(self, db, service_container, event_bus) -> None:
+        await super().on_activate(db, service_container, event_bus)
+        registry = get_format_registry()
+        for handler in self._handlers:
+            registry.register(handler)
+
+    async def on_deactivate(self, db, service_container, event_bus) -> None:
+        await super().on_deactivate(db, service_container, event_bus)
+        registry = get_format_registry()
+        for handler in self._handlers:
+            registry.unregister(handler.format_id)

app/services/import_export_service.py

@@ -1,60 +1,50 @@
-"""Import/export service — CSV/JSON/XLSX import with dry-run preview, partial-failure, CSV/XLSX export.
+"""Import/export orchestrator — generic engine delegating to plugin contracts.
 
-This service is Contact-specific — it handles import/export for Contacts and
-Companies (both use the Contact model). It is NOT a generic import/export
-service. If other entity types need import/export in the future, a separate
-service or a plugin-based interface should be created.
+W4a (Spec #359): The core orchestrator owns the format mechanics (via the
+format registry provided by format plugins) and the security policy
+(sensitive-data filter, tenant scoping, audit). The domain logic (columns,
+normalization, validation, persistence) lives in the owning plugin's
+contract contribution (``importexport_entities()`` + ``ie_*`` methods).
 
-Declared as Contact-specific (P1-23 fix): no pretense of being generic.
+New plugins can offer import/export for their entities by contributing the
+contract methods — without a single core change.
 
-Uses shared helpers from import_export_helpers.py for parsing, writing, mapping,
-and validation.  Implements partial-failure semantics: valid rows are committed
-while invalid rows are collected into a structured error report.
-
-Uses unified Contact model fields: firstname, surname, email_1, phone_1, phone_2, function.
-Company import creates Contact with type='company'.
+Function signatures are preserved from the former contact-specific service
+so existing callers and the test suite continue to work unchanged.
 """
 
 from __future__ import annotations
 
 import logging
-import uuid
 from typing import Any
 
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.audit import log_audit
-from app.core.visibility import apply_visibility_filter
-from app.models.contact import Contact
+from app.plugins.builtins.contracts import get_contract
+from app.plugins.registry import get_registry
 from app.services import import_export_helpers as helpers
-from app.services.contact_service import _serialize_contact as _contact_to_dict
 
 logger = logging.getLogger(__name__)
 
-# Expected CSV columns for each entity type
-# Company import creates Contact with type='company' using name field
-COMPANY_COLUMNS = ["name", "industry", "phone", "email", "website"]
-# Contact import uses unified Contact fields
-CONTACT_COLUMNS = ["firstname", "surname", "email", "phone", "mobile", "function", "department"]
-
-# Target fields for mapping suggestions
-CONTACT_TARGET_FIELDS = ["firstname", "surname", "email", "phone", "mobile", "function", "department"]
-COMPANY_TARGET_FIELDS = ["name", "industry", "phone", "email", "website"]
-
-# Validators for each entity type
-CONTACT_VALIDATORS: dict[str, dict] = {
-    "email": {"type": "email"},
-}
-COMPANY_VALIDATORS: dict[str, dict] = {
-    "email": {"type": "email"},
-    "website": {"type": "url"},
-}
-
 # Threshold for background job processing (rows above this go to ARQ)
 BACKGROUND_JOB_THRESHOLD = 1000
 
 
+def _find_ie_contract(entity_type: str):
+    """Find the plugin contract offering import/export for ``entity_type``.
+
+    Iterates all discovered plugins generically — no hardcoded plugin names.
+    """
+    for plugin_name in get_registry().list_discovered():
+        contract = get_contract(plugin_name)
+        entities_fn = getattr(contract, "importexport_entities", None) if contract else None
+        if entities_fn is None:
+            continue
+        if entity_type in entities_fn():
+            return contract
+    return None
+
+
 def _parse_csv(content: str | bytes) -> list[dict[str, str]]:
     """Parse file content into list of dicts (delegates to helpers with format detection).
 
@@ -63,7 +53,6 @@ def _parse_csv(content: str | bytes) -> list[dict[str, str]]:
     """
     if isinstance(content, str):
         content = content.encode("utf-8")
-    # Detect format from content magic bytes
     fmt = helpers.detect_format("upload", content)
     if fmt == "json":
         return helpers.parse_json(content)
@@ -77,262 +66,24 @@ def _validate_row(row: dict[str, str], required: list[str]) -> list[str]:
     return helpers.validate_row(row, required)
 
 
-def _normalize_contact_row(row: dict[str, str]) -> dict[str, str]:
-    """Normalize contact row to unified field names."""
-    firstname = (row.get("firstname") or row.get("first_name") or "").strip()
-    surname = (row.get("surname") or row.get("last_name") or "").strip()
-    row["firstname"] = firstname
-    row["surname"] = surname
-    # Map old column names to unified ones
-    if "email" not in row and "email_address" in row:
-        row["email"] = row["email_address"]
-    if "mobile" not in row and "phone_2" in row:
-        row["mobile"] = row["phone_2"]
-    if "function" not in row and "position" in row:
-        row["function"] = row["position"]
-    return row
+# ─── Import (generic, contract-driven) ──────────────────────────────────────
 
-
-def _normalize_company_row(row: dict[str, str]) -> dict[str, str]:
-    """Normalize company row to unified field names."""
-    name = (row.get("name") or row.get("company") or row.get("company_name") or "").strip()
-    row["name"] = name
-    if "email" not in row and "email_address" in row:
-        row["email"] = row["email_address"]
-    if "website" not in row and "url" in row:
-        row["website"] = row["url"]
-    if "website" not in row and "homepage" in row:
-        row["website"] = row["homepage"]
-    return row
-
-
-async def import_companies(
+async def _generic_import(
     db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    csv_content: str | bytes,
-    dry_run: bool = False,
-    field_mapping: dict[str, str] | None = None,
-) -> dict[str, Any]:
-    """Import companies from CSV as Contact with type='company'.
-
-    Uses unified Contact model: name field for company name, email_1/phone_1 for contact info.
-    Implements partial-failure: valid rows committed, invalid rows collected in error report.
-    """
-    rows = _parse_csv(csv_content)
-    total = len(rows)
-    errors: list[dict] = []
-    valid_rows: list[dict] = []
-
-    for idx, row in enumerate(rows, start=1):
-        # Apply field mapping if provided
-        if field_mapping:
-            row = helpers.map_fields(row, field_mapping)
-        row = _normalize_company_row(row)
-        row_errors = helpers.validate_row(row, ["name"], COMPANY_VALIDATORS)
-        if row_errors:
-            for e in row_errors:
-                errors.append({"row": idx, "field": "", "message": e})
-        else:
-            valid_rows.append(row)
-
-    # Count unique failed rows (a row may have multiple errors)
-    failed_row_count = len({e["row"] for e in errors})
-
-    if dry_run:
-        return helpers.build_import_result(
-            total=total,
-            succeeded=len(valid_rows),
-            failed=failed_row_count,
-            errors=errors,
-            dry_run=True,
-        )
-
-    created: list[dict] = []
-    failed: list[dict] = []
-    succeeded = 0
-
-    for idx, row in enumerate(valid_rows, start=1):
-        try:
-            contact = Contact(
-                tenant_id=tenant_id,
-                type="company",
-                name=row["name"].strip(),
-                displayname=row["name"].strip(),
-                email_1=row.get("email", "").strip() or None,
-                phone_1=row.get("phone", "").strip() or None,
-                website=row.get("website", "").strip() or None,
-                owner_id=user_id,
-                created_by=user_id,
-                updated_by=user_id,
-            )
-            db.add(contact)
-            await db.flush()
-            await log_audit(
-                db,
-                tenant_id,
-                user_id,
-                "import",
-                "contact",
-                contact.id,
-                changes={"name": contact.name, "type": "company"},
-            )
-            created.append(_contact_to_dict(contact))
-            succeeded += 1
-        except Exception as exc:
-            logger.warning("Import companies: row %d failed: %s", idx, exc)
-            await db.rollback()
-            failed.append({"row": idx, "field": "", "message": str(exc)})
-
-    if succeeded > 0:
-        try:
-            await db.commit()
-        except Exception as exc:
-            logger.error("Import companies: commit failed: %s", exc)
-            await db.rollback()
-            return helpers.build_import_result(
-                total=total,
-                succeeded=0,
-                failed=len(valid_rows),
-                errors=errors + [{"row": 0, "field": "", "message": f"Commit failed: {exc}"}],
-                created=[],
-            )
-
-    # Count unique failed rows from validation errors + runtime failures
-    all_failed_rows = {e["row"] for e in errors} | {f["row"] for f in failed}
-    return helpers.build_import_result(
-        total=total,
-        succeeded=succeeded,
-        failed=len(all_failed_rows),
-        errors=errors + failed,
-        created=created,
-    )
-
-
-async def import_contacts(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    csv_content: str | bytes,
-    dry_run: bool = False,
-    field_mapping: dict[str, str] | None = None,
-) -> dict[str, Any]:
-    """Import contacts from CSV using unified Contact model fields.
-
-    CSV columns: firstname, surname, email, phone, mobile, function, department.
-    Maps to Contact fields: firstname, surname, email_1, phone_1, phone_2, function.
-    Implements partial-failure: valid rows committed, invalid rows collected in error report.
-    """
-    rows = _parse_csv(csv_content)
-    total = len(rows)
-    errors: list[dict] = []
-    valid_rows: list[dict] = []
-
-    for idx, row in enumerate(rows, start=1):
-        # Apply field mapping if provided
-        if field_mapping:
-            row = helpers.map_fields(row, field_mapping)
-        row = _normalize_contact_row(row)
-        # Validate: at least one of firstname or surname required
-        if not row["firstname"] and not row["surname"]:
-            errors.append({"row": idx, "field": "", "message": "Missing required field: firstname or surname"})
-            continue
-        row_errors = helpers.validate_row(row, [], CONTACT_VALIDATORS)
-        if row_errors:
-            for e in row_errors:
-                errors.append({"row": idx, "field": "", "message": e})
-        else:
-            valid_rows.append(row)
-
-    # Count unique failed rows (a row may have multiple errors)
-    failed_row_count = len({e["row"] for e in errors})
-
-    if dry_run:
-        return helpers.build_import_result(
-            total=total,
-            succeeded=len(valid_rows),
-            failed=failed_row_count,
-            errors=errors,
-            dry_run=True,
-        )
-
-    created: list[dict] = []
-    failed: list[dict] = []
-    succeeded = 0
-
-    for idx, row in enumerate(valid_rows, start=1):
-        try:
-            contact = Contact(
-                tenant_id=tenant_id,
-                type="person",
-                firstname=row["firstname"].strip() or None,
-                surname=row["surname"].strip() or None,
-                displayname=f"{row['firstname']} {row['surname']}".strip(),
-                email_1=row.get("email", "").strip() or None,
-                phone_1=row.get("phone", "").strip() or None,
-                phone_2=row.get("mobile", "").strip() or None,
-                owner_id=user_id,
-                created_by=user_id,
-                updated_by=user_id,
-            )
-            db.add(contact)
-            await db.flush()
-            await log_audit(
-                db,
-                tenant_id,
-                user_id,
-                "import",
-                "contact",
-                contact.id,
-                changes={"firstname": contact.firstname, "surname": contact.surname},
-            )
-            created.append(_contact_to_dict(contact))
-            succeeded += 1
-        except Exception as exc:
-            logger.warning("Import contacts: row %d failed: %s", idx, exc)
-            await db.rollback()
-            failed.append({"row": idx, "field": "", "message": str(exc)})
-
-    if succeeded > 0:
-        try:
-            await db.commit()
-        except Exception as exc:
-            logger.error("Import contacts: commit failed: %s", exc)
-            await db.rollback()
-            return helpers.build_import_result(
-                total=total,
-                succeeded=0,
-                failed=len(valid_rows),
-                errors=errors + [{"row": 0, "field": "", "message": f"Commit failed: {exc}"}],
-                created=[],
-            )
-
-    # Count unique failed rows from validation errors + runtime failures
-    all_failed_rows = {e["row"] for e in errors} | {f["row"] for f in failed}
-    return helpers.build_import_result(
-        total=total,
-        succeeded=succeeded,
-        failed=len(all_failed_rows),
-        errors=errors + failed,
-        created=created,
-    )
-
-
-async def import_csv(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
+    tenant_id: Any,
+    user_id: Any,
     csv_content: str | bytes,
     entity_type: str,
     dry_run: bool = False,
     field_mapping: dict[str, str] | None = None,
 ) -> dict[str, Any]:
-    """Generic CSV import dispatcher based on entity_type ('companies' or 'contacts')."""
-    if entity_type == "companies":
-        return await import_companies(db, tenant_id, user_id, csv_content, dry_run=dry_run, field_mapping=field_mapping)
-    elif entity_type == "contacts":
-        return await import_contacts(db, tenant_id, user_id, csv_content, dry_run=dry_run, field_mapping=field_mapping)
-    else:
+    """Generic import flow: parse → normalize → validate → persist.
+
+    Domain logic (columns, normalizers, validators, persistence) comes from
+    the plugin contract; the orchestrator handles flow control and policy.
+    """
+    contract = _find_ie_contract(entity_type)
+    if contract is None:
         return helpers.build_import_result(
             total=0,
             succeeded=0,
@@ -340,114 +91,176 @@ async def import_csv(
             errors=[{"row": 0, "field": "", "message": f"Unknown entity_type: {entity_type}"}],
         )
 
+    rows = _parse_csv(csv_content)
+    total = len(rows)
+    errors: list[dict] = []
+    valid_rows: list[dict] = []
+
+    for idx, row in enumerate(rows, start=1):
+        if field_mapping:
+            row = helpers.map_fields(row, field_mapping)
+        row = contract.ie_normalize_row(entity_type, row)
+        ok, err_msg = contract.ie_row_valid(entity_type, row)
+        if not ok:
+            errors.append({"row": idx, "field": "", "message": err_msg})
+            continue
+        row_errors = helpers.validate_row(
+            row, contract.ie_required(entity_type), contract.ie_validators(entity_type)
+        )
+        if row_errors:
+            for e in row_errors:
+                errors.append({"row": idx, "field": "", "message": e})
+        else:
+            valid_rows.append(row)
+
+    failed_row_count = len({e["row"] for e in errors})
+
+    if dry_run:
+        return helpers.build_import_result(
+            total=total,
+            succeeded=len(valid_rows),
+            failed=failed_row_count,
+            errors=errors,
+            dry_run=True,
+        )
+
+    created: list[dict] = []
+    failed: list[dict] = []
+    succeeded = 0
+
+    for idx, row in enumerate(valid_rows, start=1):
+        try:
+            record = await contract.ie_persist_row(db, tenant_id, user_id, entity_type, row)
+            created.append(record)
+            succeeded += 1
+        except Exception as exc:
+            logger.warning("Import %s: row %d failed: %s", entity_type, idx, exc)
+            await db.rollback()
+            failed.append({"row": idx, "field": "", "message": str(exc)})
+
+    if succeeded > 0:
+        try:
+            await db.commit()
+        except Exception as exc:
+            logger.error("Import %s: commit failed: %s", entity_type, exc)
+            await db.rollback()
+            return helpers.build_import_result(
+                total=total,
+                succeeded=0,
+                failed=len(valid_rows),
+                errors=errors + [{"row": 0, "field": "", "message": f"Commit failed: {exc}"}],
+                created=[],
+            )
+
+    all_failed_rows = {e["row"] for e in errors} | {f["row"] for f in failed}
+    return helpers.build_import_result(
+        total=total,
+        succeeded=succeeded,
+        failed=len(all_failed_rows),
+        errors=errors + failed,
+        created=created,
+    )
+
+
+async def import_companies(
+    db: AsyncSession,
+    tenant_id: Any,
+    user_id: Any,
+    csv_content: str | bytes,
+    dry_run: bool = False,
+    field_mapping: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Import companies from CSV (delegates to the owning plugin's contract)."""
+    return await _generic_import(
+        db, tenant_id, user_id, csv_content, "companies",
+        dry_run=dry_run, field_mapping=field_mapping,
+    )
+
+
+async def import_contacts(
+    db: AsyncSession,
+    tenant_id: Any,
+    user_id: Any,
+    csv_content: str | bytes,
+    dry_run: bool = False,
+    field_mapping: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Import contacts from CSV (delegates to the owning plugin's contract)."""
+    return await _generic_import(
+        db, tenant_id, user_id, csv_content, "contacts",
+        dry_run=dry_run, field_mapping=field_mapping,
+    )
+
+
+async def import_csv(
+    db: AsyncSession,
+    tenant_id: Any,
+    user_id: Any,
+    csv_content: str | bytes,
+    entity_type: str,
+    dry_run: bool = False,
+    field_mapping: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Generic import dispatcher based on entity_type."""
+    return await _generic_import(
+        db, tenant_id, user_id, csv_content, entity_type,
+        dry_run=dry_run, field_mapping=field_mapping,
+    )
+
+
+# ─── Export (generic, contract-driven) ──────────────────────────────────────
+
+async def _generic_export_csv(
+    db: AsyncSession,
+    tenant_id: Any,
+    entity_type: str,
+    user_id: Any = None,
+    is_system_admin: bool = False,
+) -> str:
+    """Generic CSV export via the owning plugin's contract."""
+    contract = _find_ie_contract(entity_type)
+    if contract is None:
+        return helpers.write_csv([], []).decode("utf-8")
+    headers, rows = await contract.ie_fetch_rows(
+        db, tenant_id, entity_type, user_id=user_id, is_system_admin=is_system_admin,
+    )
+    return helpers.write_csv(rows, headers).decode("utf-8")
 
-# ─── Export functions ────────────────────────────────────────────────────────
 
 async def export_contacts_csv(
     db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID | None = None,
+    tenant_id: Any,
+    user_id: Any = None,
     is_system_admin: bool = False,
 ) -> str:
-    """Export contacts as CSV string using unified Contact model fields."""
-    q = (
-        select(Contact)
-        .where(
-            Contact.tenant_id == tenant_id,
-            Contact.deleted_at.is_(None),
-        )
-        .order_by(Contact.surname, Contact.firstname)
-    )
-    if user_id:
-        q = await apply_visibility_filter(
-            db, q, "contact", Contact, user_id, tenant_id, is_system_admin
-        )
-    result = await db.execute(q)
-    contacts = result.scalars().all()
-
-    headers = ["id", "type", "firstname", "surname", "name", "email", "phone", "mobile", "city", "postalcode", "country"]
-    rows = [
-        {
-            "id": str(c.id),
-            "type": c.type or "person",
-            "firstname": c.firstname or "",
-            "surname": c.surname or "",
-            "name": c.name or "",
-            "email": c.email_1 or "",
-            "phone": c.phone_1 or "",
-            "mobile": c.phone_2 or "",
-            "city": c.mailing_city or "",
-            "postalcode": c.mailing_postalcode or "",
-            "country": c.mailing_country or "",
-        }
-        for c in contacts
-    ]
-    return helpers.write_csv(rows, headers).decode("utf-8")
+    """Export contacts as CSV string (delegates to contacts contract)."""
+    return await _generic_export_csv(db, tenant_id, "contacts", user_id, is_system_admin)
 
 
 async def export_companies_csv(
     db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID | None = None,
+    tenant_id: Any,
+    user_id: Any = None,
     is_system_admin: bool = False,
 ) -> str:
-    """Export companies (Contact.type == 'company') as CSV string."""
-    q = (
-        select(Contact)
-        .where(
-            Contact.tenant_id == tenant_id,
-            Contact.deleted_at.is_(None),
-            Contact.type == "company",
-        )
-        .order_by(Contact.name)
-    )
-    if user_id:
-        q = await apply_visibility_filter(
-            db, q, "contact", Contact, user_id, tenant_id, is_system_admin
-        )
-    result = await db.execute(q)
-    companies = result.scalars().all()
-
-    headers = ["id", "type", "name", "email", "phone", "website", "city", "postalcode", "country"]
-    rows = [
-        {
-            "id": str(c.id),
-            "type": c.type or "company",
-            "name": c.name or "",
-            "email": c.email_1 or "",
-            "phone": c.phone_1 or "",
-            "website": c.website or "",
-            "city": c.mailing_city or "",
-            "postalcode": c.mailing_postalcode or "",
-            "country": c.mailing_country or "",
-        }
-        for c in companies
-    ]
-    return helpers.write_csv(rows, headers).decode("utf-8")
+    """Export companies as CSV string (delegates to contacts contract)."""
+    return await _generic_export_csv(db, tenant_id, "companies", user_id, is_system_admin)
 
 
-# ─── Preview & Validate ──────────────────────────────────────────────────────
+# ─── Preview & Validate (generic, contract-driven) ──────────────────────────
 
 def preview_import(
     filename: str,
     content: bytes,
     entity_type: str,
 ) -> dict[str, Any]:
-    """Parse a file and return preview data (first 10 rows) + mapping suggestion.
-
-    Does NOT touch the database.
-    """
+    """Parse a file and return preview data (first 10 rows) + mapping suggestion."""
     rows = helpers.parse_file(filename, content)
     columns = list(rows[0].keys()) if rows else []
     preview_rows = rows[:10]
 
-    if entity_type == "contacts":
-        target_fields = CONTACT_TARGET_FIELDS
-    elif entity_type == "companies":
-        target_fields = COMPANY_TARGET_FIELDS
-    else:
-        target_fields = []
-
+    contract = _find_ie_contract(entity_type)
+    target_fields = contract.ie_target_fields(entity_type) if contract else []
     mapping_suggestion = helpers.suggest_mapping(columns, target_fields)
 
     return {
@@ -471,29 +284,33 @@ def validate_import(
     errors: list[dict] = []
     valid_count = 0
 
+    contract = _find_ie_contract(entity_type)
+    if contract is None:
+        return helpers.build_import_result(
+            total=total,
+            succeeded=0,
+            failed=total,
+            errors=[{"row": 0, "field": "", "message": f"Unknown entity_type: {entity_type}"}],
+            dry_run=True,
+        )
+
     for idx, row in enumerate(rows, start=1):
         if field_mapping:
             row = helpers.map_fields(row, field_mapping)
-        if entity_type == "contacts":
-            row = _normalize_contact_row(row)
-            if not row["firstname"] and not row["surname"]:
-                errors.append({"row": idx, "field": "", "message": "Missing required field: firstname or surname"})
-                continue
-            row_errors = helpers.validate_row(row, [], CONTACT_VALIDATORS)
-        elif entity_type == "companies":
-            row = _normalize_company_row(row)
-            row_errors = helpers.validate_row(row, ["name"], COMPANY_VALIDATORS)
-        else:
-            errors.append({"row": idx, "field": "", "message": f"Unknown entity_type: {entity_type}"})
+        row = contract.ie_normalize_row(entity_type, row)
+        ok, err_msg = contract.ie_row_valid(entity_type, row)
+        if not ok:
+            errors.append({"row": idx, "field": "", "message": err_msg})
             continue
-
+        row_errors = helpers.validate_row(
+            row, contract.ie_required(entity_type), contract.ie_validators(entity_type)
+        )
         if row_errors:
             for e in row_errors:
                 errors.append({"row": idx, "field": "", "message": e})
         else:
             valid_count += 1
 
-    # Count unique failed rows
     failed_row_count = len({e["row"] for e in errors})
     return helpers.build_import_result(
         total=total,