T02: companies + contacts + import/export + N:M + soft-delete + GDPR + FTS

- Company CRUD with soft-delete, FTS search (tsvector + GIN), filter, pagination
- Contact CRUD with N:M company linking via company_contacts
- CSV/XLSX export, CSV import with dry-run preview
- GDPR hard-delete with deletion_log
- Audit log on all mutations
- 27 new tests (24 ACs), 56 total tests pass
- Migration 0002: contacts, company_contacts, FTS search_tsv
- Fixed T01 tests: POST→201, PATCH→PUT compatibility

.a0/briefings/T02_briefing.md

@@ -0,0 +1,106 @@
+# T02: Company + Contact + Import/Export System
+
+## Context
+- Project: LeoCRM (greenfield rewrite, Option C)
+- Repo: /a0/usr/workdir/dev-projects/leocrm
+- T01 COMPLETE: auth, multi-tenant, RBAC, sessions, audit, notifications all working
+- T01 commit: 7a7daf8 (pushed to Forgejo)
+- Tech: FastAPI + SQLAlchemy 2.0 async + PostgreSQL + Redis + Pydantic v2
+
+## Existing T01 Code to Build On
+- `app/models/company.py` (40 lines) — Company model skeleton, needs Contact + CompanyContact models
+- `app/routes/companies.py` (210 lines) — Company CRUD skeleton, needs expansion + Contact routes
+- `app/schemas/company.py` (23 lines) — Company schema, needs Contact schemas
+- `app/core/db/__init__.py` — Engine, Session, Base, TenantMixin, set_tenant_context
+- `app/deps.py` — get_current_user, require_admin, get_tenant_id
+- `app/core/audit.py` — log_audit function
+- `app/core/notifications.py` — create_notification
+- `tests/conftest.py` — Test fixtures with TRUNCATE CASCADE (DO NOT modify truncate list without checking table names)
+
+## Requirements (25)
+F-COMP-01..08, F-CONT-01..07, F-DATA-01..04, F-MIG-01, F-CORE-06, F-CORE-11, F-CORE-13, F-SEARCH-01, F-TEST-01
+
+## Acceptance Criteria (24)
+1. GET /api/v1/companies → 200 + paginated (total/page/page_size)
+2. GET /api/v1/companies?search=Tech → 200 + FTS results (tsvector)
+3. GET /api/v1/companies?industry=IT&sort_by=name&sort_order=asc → 200 + filtered+sorted
+4. POST /api/v1/companies valid → 201 + company object
+5. POST /api/v1/companies missing name → 422
+6. GET /api/v1/companies/{id} → 200 + detail inkl. contacts array
+7. PUT /api/v1/companies/{id} → 200 + updated
+8. DELETE /api/v1/companies/{id} → 204, deleted_at gesetzt (soft-delete)
+9. DELETE /api/v1/companies/{id}?cascade=true → 204, company + links geloescht
+10. POST /api/v1/companies/{id}/contacts/{cid} → 200, N:M link
+11. DELETE /api/v1/companies/{id}/contacts/{cid} → 204, N:M unlink
+12. GET /api/v1/companies/export?format=csv → 200 + text/csv
+13. GET /api/v1/companies/export?format=xlsx → 200 + openxmlformats
+14. GET /api/v1/contacts → 200 + paginated
+15. POST /api/v1/contacts mit company_ids array → 201 + N:M links
+16. GET /api/v1/contacts/{id} → 200 + detail inkl. companies array
+17. PUT /api/v1/contacts/{id} → 200
+18. DELETE /api/v1/contacts/{id} → 204, soft-delete
+19. DELETE /api/v1/contacts/{id}?gdpr=true → 204, hard-delete + deletion_log
+20. POST /api/v1/import CSV + entity_type=companies → 200 + result
+21. POST /api/v1/import/preview CSV → 200 + dry-run (no DB changes)
+22. GET /api/v1/companies/{id}/emails → 200 (empty array, mail plugin inactive)
+23. Audit log entry on every company/contact mutation
+24. Soft-deleted company not in GET list (deleted_at IS NULL filter)
+
+## Files to Create/Modify
+### New Files:
+- `app/models/contact.py` — Contact model + CompanyContact (N:M join table)
+- `app/schemas/contact.py` — Contact schemas (create/update/read/list)
+- `app/services/company_service.py` — Company CRUD + search + filter + pagination
+- `app/services/contact_service.py` — Contact CRUD + N:M linking
+- `app/services/import_export_service.py` — CSV import/export, XLSX export, dry-run preview
+- `app/routes/contacts.py` — Contact CRUD + N:M endpoints
+- `app/routes/import_export.py` — Import/export endpoints
+- `tests/test_companies.py` — Company CRUD + search + filter + export tests
+- `tests/test_contacts.py` — Contact CRUD + N:M + GDPR delete tests
+- `tests/test_import_export.py` — CSV import + preview + export tests
+
+### Modify:
+- `app/models/company.py` — Add soft-delete (deleted_at), FTS tsvector, ensure TenantMixin
+- `app/routes/companies.py` — Expand to full CRUD + search + filter + export + N:M endpoints
+- `app/schemas/company.py` — Add pagination, search, filter schemas
+- `app/models/__init__.py` — Register Contact, CompanyContact
+- `app/routes/__init__.py` — Register contacts + import_export routers
+- `app/schemas/__init__.py` — Register contact schemas
+- `app/services/__init__.py` — Register new services
+- `tests/conftest.py` — Add contacts, company_contacts to TRUNCATE list
+- `alembic/versions/` — New migration for contacts + company_contacts + FTS indexes
+
+## Dependencies to Install
+- `openpyxl>=3.1` — XLSX export (add to requirements.txt)
+
+## Forbidden Patterns
+- NO JWT tokens (session-based auth from T01)
+- NO SQLite (PostgreSQL only)
+- NO wildcard CORS
+- NO raw SQL without tenant context (use set_config or ORM filtering)
+- NO hardcoded secrets
+- NO legacy code reuse
+- NO .test TLD emails (Pydantic v2 rejects — use .com)
+- NO `SET LOCAL` with bound params (use `SELECT set_config()` instead)
+- NO raising HTTPException in middleware (return JSONResponse)
+- NO POST without status_code=201
+
+## Test Spec
+- Commands: `cd /a0/usr/workdir/dev-projects/leocrm && source venv/bin/activate && python -m pytest tests/test_companies.py tests/test_contacts.py tests/test_import_export.py -v --tb=short`
+- Coverage: `python -m pytest tests/test_companies.py tests/test_contacts.py tests/test_import_export.py --cov=app/routes/companies --cov=app/routes/contacts --cov=app/services --cov-report=term-missing`
+- Target: 85% for new modules
+- All 24 ACs must pass
+
+## Token Rule
+- Use `text_editor:read` for MODIFY, `text_editor:write` for NEW, `code_execution_tool:terminal` for test runs
+- Reference files by path, not inline
+- Multi-file output: separate files, not one big file
+
+## JSON Tool Examples
+Use this format for all tool calls:
+```json
+{"tool_name":"text_editor","tool_args":{"action":"write","path":"/a0/usr/workdir/dev-projects/leocrm/app/models/contact.py","content":"..."}}
+```
+```json
+{"tool_name":"code_execution_tool","tool_args":{"runtime":"terminal","session":0,"code":"cd /a0/usr/workdir/dev-projects/leocrm && source venv/bin/activate && python -m pytest tests/test_companies.py -v --tb=short"}}
+```

.a0/briefings/T03_briefing.md

@@ -0,0 +1,132 @@
+# T03: Plugin System Framework
+
+## Context
+- Project: LeoCRM (greenfield rewrite, Option C)
+- Repo: /a0/usr/workdir/dev-projects/leocrm
+- T01 COMPLETE: auth, multi-tenant, RBAC, sessions, audit, notifications
+- T01 commit: 7a7daf8 (pushed to Forgejo)
+- Tech: FastAPI + SQLAlchemy 2.0 async + PostgreSQL + Redis + Pydantic v2
+
+## Existing T01 Code to Build On
+- `app/core/event_bus.py` — Event bus (0% coverage, needs integration)
+- `app/core/service_container.py` — DI container (0% coverage, needs integration)
+- `app/core/db/__init__.py` — Engine, Session, Base, TenantMixin, set_tenant_context
+- `app/deps.py` — get_current_user, require_admin, get_tenant_id
+- `app/core/audit.py` — log_audit function
+- `app/main.py` — create_app factory, mounts routers, middleware
+- `app/routes/__init__.py` — router aggregator
+- `tests/conftest.py` — Test fixtures with TRUNCATE CASCADE
+
+## Requirements (7)
+F-PLUGIN-01: Plugin-System für Module — Module als Plugins, Daten austauschbar
+F-PLUGIN-02: Plugin-Schnittstellen-Definition — API-Contract, Lifecycle-Hooks, Manifest, Abhängigkeiten
+F-CORE-01: Multi-Tenant-Architektur
+F-CORE-03: RBAC
+F-CORE-04: Audit-Log
+F-CORE-05: Event-System
+F-TEST-01: Test-Coverage
+
+## Acceptance Criteria (14)
+1. GET /api/v1/plugins → 200 + list of plugins with status
+2. POST /api/v1/plugins/{name}/install → 200, plugin status=installed, migrations run
+3. POST /api/v1/plugins/{name}/activate → 200, plugin status=active, routes registered
+4. POST /api/v1/plugins/{name}/deactivate → 200, plugin status=inactive, routes unregistered
+5. DELETE /api/v1/plugins/{name} → 200, plugin removed
+6. DELETE /api/v1/plugins/{name}?remove_data=true → 200, plugin tables dropped
+7. GET /api/v1/plugins/manifest → 200 + manifest schema documentation
+8. Plugin activation registers event listeners on event bus
+9. Plugin deactivation unregisters event listeners
+10. Plugin migration creates tables with tenant_id column
+11. Plugin migration validator rejects tables without tenant_id
+12. Plugin DB migrations tracked in plugin_migrations table
+13. Activating already-active plugin → idempotent (200, no error)
+14. Deactivating inactive plugin → idempotent (200)
+
+## Files to Create/Modify
+### New Files:
+- `app/models/plugin.py` — Plugin, PluginMigration (plugin_migrations tracking table)
+- `app/schemas/plugin.py` — Plugin schemas (manifest, status, install/activate response)
+- `app/services/plugin_service.py` — Plugin lifecycle: discover, install, activate, deactivate, uninstall
+- `app/routes/plugins.py` — Plugin endpoints (list/install/activate/deactivate/uninstall/manifest)
+- `app/plugins/__init__.py` — Plugin package init
+- `app/plugins/base.py` — BasePlugin abstract class with lifecycle hooks
+- `app/plugins/manifest.py` — PluginManifest Pydantic schema (name, version, dependencies, routes, events, migrations)
+- `app/plugins/registry.py` — Plugin registry (in-memory + DB-backed status)
+- `app/plugins/migration_runner.py` — Plugin DB migration runner + validator (tenant_id check)
+- `app/plugins/builtins/__init__.py` — Built-in plugins directory (empty for now, just structure)
+- `tests/test_plugins.py` — Plugin lifecycle tests (install/activate/deactivate/uninstall/idempotent)
+
+### Modify:
+- `app/models/__init__.py` — Register Plugin, PluginMigration
+- `app/routes/__init__.py` — Register plugins router
+- `app/schemas/__init__.py` — Register plugin schemas
+- `app/services/__init__.py` — Register plugin_service
+- `app/main.py` — Initialize plugin registry on startup (discover builtins)
+- `app/core/event_bus.py` — Ensure register/unregister listener API works for plugins
+- `app/core/service_container.py` — Ensure plugins can receive db, cache, event_bus, storage, notifications
+- `tests/conftest.py` — Add plugins, plugin_migrations to TRUNCATE list
+- `alembic/versions/` — New migration for plugins + plugin_migrations tables
+
+## Plugin Lifecycle Design
+```
+discovered → installed → active → inactive → uninstalled
+                ↑                      ↓
+                └──────────────────────┘ (can re-activate)
+```
+
+## Plugin Manifest Schema (Pydantic v2)
+```python
+class PluginManifest(BaseModel):
+    name: str  # unique identifier
+    version: str  # semver
+    display_name: str
+    description: str
+    dependencies: list[str] = []  # other plugin names required
+    routes: list[dict] = []  # route definitions
+    events: list[str] = []  # event names to listen
+    migrations: list[str] = []  # migration file names
+    permissions: list[str] = []  # required permissions
+```
+
+## BasePlugin Abstract Class
+```python
+class BasePlugin(ABC):
+    manifest: PluginManifest
+    
+    async def on_install(self, db, service_container): ...
+    async def on_activate(self, db, service_container, event_bus): ...
+    async def on_deactivate(self, db, service_container, event_bus): ...
+    async def on_uninstall(self, db, service_container): ...
+    def get_routes(self) -> list[APIRouter]: ...
+```
+
+## Forbidden Patterns
+- NO JWT tokens (session-based auth from T01)
+- NO SQLite (PostgreSQL only)
+- NO wildcard CORS
+- NO raw SQL without tenant context
+- NO hardcoded secrets
+- NO .test TLD emails (use .com)
+- NO `SET LOCAL` with bound params (use `SELECT set_config()`)
+- NO raising HTTPException in middleware (return JSONResponse)
+- NO POST without status_code=201 (where applicable)
+- NO plugin tables without tenant_id column (validator enforces)
+
+## Test Spec
+- Commands: `cd /a0/usr/workdir/dev-projects/leocrm && source venv/bin/activate && python -m pytest tests/test_plugins.py -v --tb=short`
+- Coverage: `python -m pytest tests/test_plugins.py --cov=app/plugins --cov-report=term-missing`
+- Target: 85% for plugin modules
+- All 14 ACs must pass
+
+## Token Rule
+- Use `text_editor:read` for MODIFY, `text_editor:write` for NEW, `code_execution_tool:terminal` for test runs
+- Reference files by path, not inline
+- Multi-file output: separate files, not one big file
+
+## JSON Tool Examples
+```json
+{"tool_name":"text_editor","tool_args":{"action":"write","path":"/a0/usr/workdir/dev-projects/leocrm/app/plugins/base.py","content":"..."}}
+```
+```json
+{"tool_name":"code_execution_tool","tool_args":{"runtime":"terminal","session":0,"code":"cd /a0/usr/workdir/dev-projects/leocrm && source venv/bin/activate && python -m pytest tests/test_plugins.py -v --tb=short"}}
+```

alembic/versions/0002_contacts_fts.py

@@ -0,0 +1,105 @@
+"""T02: contacts, company_contacts, FTS search_tsv on companies.
+
+Revision ID: 0002_contacts_fts
+Revises: 0001_initial
+Create Date: 2026-06-29
+"""
+
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0002_contacts_fts"
+down_revision: Union[str, None] = "0001_initial"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # --- companies: add FTS search_tsv computed column ---
+    op.execute(
+        """
+        ALTER TABLE companies
+        ADD COLUMN search_tsv TSVECTOR
+        GENERATED ALWAYS AS (
+            to_tsvector('english',
+                coalesce(name, '') || ' ' ||
+                coalesce(description, '') || ' ' ||
+                coalesce(industry, '')
+            )
+        ) STORED
+        """
+    )
+    op.create_index(
+        "ix_companies_search_vec",
+        "companies",
+        ["search_tsv"],
+        postgresql_using="gin",
+    )
+    op.create_index(
+        "ix_companies_industry",
+        "companies",
+        ["tenant_id", "industry"],
+    )
+
+    # --- contacts ---
+    op.create_table(
+        "contacts",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column("tenant_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("first_name", sa.String(100), nullable=False),
+        sa.Column("last_name", sa.String(100), nullable=False),
+        sa.Column("email", sa.String(255), nullable=True),
+        sa.Column("phone", sa.String(30), nullable=True),
+        sa.Column("mobile", sa.String(30), nullable=True),
+        sa.Column("position", sa.String(100), nullable=True),
+        sa.Column("department", sa.String(100), nullable=True),
+        sa.Column("linkedin_url", sa.String(500), nullable=True),
+        sa.Column("notes", sa.Text, nullable=True),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("updated_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+    op.create_index("ix_contacts_tenant_id", "contacts", ["tenant_id"])
+    op.create_index("ix_contacts_tenant_deleted", "contacts", ["tenant_id", "deleted_at"])
+    op.create_index("ix_contacts_tenant_name", "contacts", ["tenant_id", "last_name", "first_name"])
+    op.create_index("ix_contacts_email", "contacts", ["email"])
+
+    # --- company_contacts (N:M join) ---
+    op.create_table(
+        "company_contacts",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column("tenant_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("company_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("companies.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("contact_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("contacts.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("role_at_company", sa.String(100), nullable=True),
+        sa.Column("is_primary", sa.Boolean, nullable=False, server_default=sa.text("false")),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.UniqueConstraint("company_id", "contact_id", "tenant_id", name="uq_company_contact_tenant"),
+    )
+    op.create_index("ix_cc_company", "company_contacts", ["company_id"])
+    op.create_index("ix_cc_contact", "company_contacts", ["contact_id"])
+    op.create_index("ix_company_contacts_tenant_id", "company_contacts", ["tenant_id"])
+
+    # --- RLS on new tenant-scoped tables ---
+    for table in ["contacts", "company_contacts"]:
+        op.execute(f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY")
+        op.execute(
+            f"""CREATE POLICY tenant_isolation ON {table}
+            USING (tenant_id = current_setting('app.current_tenant_id')::uuid)"""
+        )
+
+
+def downgrade() -> None:
+    op.drop_table("company_contacts")
+    op.drop_table("contacts")
+    op.drop_index("ix_companies_search_vec", table_name="companies")
+    op.drop_index("ix_companies_industry", table_name="companies")
+    op.execute("ALTER TABLE companies DROP COLUMN search_tsv")

app/main.py

@@ -10,7 +10,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from app.config import get_settings
 from app.core.middleware import CSRFMiddleware
 from app.core.db import close_engine
-from app.routes import auth, users, roles, tenants, health, notifications, companies
+from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export
 
 
 @asynccontextmanager
@@ -42,6 +42,8 @@ def create_app() -> FastAPI:
     app.include_router(tenants.router)
     app.include_router(notifications.router)
     app.include_router(companies.router)
+    app.include_router(contacts.router)
+    app.include_router(import_export.router)
 
     return app
 

app/models/__init__.py

@@ -8,9 +8,11 @@ from app.models.audit import AuditLog, DeletionLog
 from app.models.notification import Notification
 from app.models.auth import PasswordResetToken, ApiToken
 from app.models.company import Company
+from app.models.contact import Contact, CompanyContact
 
 __all__ = [
     "Tenant", "User", "UserTenant", "Role", "Session",
     "AuditLog", "DeletionLog", "Notification",
     "PasswordResetToken", "ApiToken", "Company",
+    "Contact", "CompanyContact",
 ]

app/models/company.py

@@ -1,24 +1,27 @@
-"""Company model — minimal for cross-tenant test (AC #17)."""
+"""Company model — with soft-delete, FTS tsvector, and tenant scoping."""
 
 from __future__ import annotations
 
 import uuid
 from datetime import datetime
+from typing import Any
 
-from sqlalchemy import String, Integer, Numeric, Text, DateTime, ForeignKey, func, Index
+from sqlalchemy import String, Text, DateTime, ForeignKey, Index, Computed
-from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.dialects.postgresql import UUID as PGUUID, TSVECTOR
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
 
 
 class Company(Base, TenantMixin):
-    """Company entity (minimal fields for T01 cross-tenant test)."""
+    """Company entity with full-text search support."""
 
     __tablename__ = "companies"
     __table_args__ = (
         Index("ix_companies_tenant_deleted", "tenant_id", "deleted_at"),
         Index("ix_companies_tenant_name", "tenant_id", "name"),
+        Index("ix_companies_industry", "tenant_id", "industry"),
+        Index("ix_companies_search_vec", "search_tsv", postgresql_using="gin"),
     )
 
     id: Mapped[uuid.UUID] = mapped_column(
@@ -31,7 +34,18 @@ class Company(Base, TenantMixin):
     email: Mapped[str | None] = mapped_column(String(255), nullable=True)
     website: Mapped[str | None] = mapped_column(String(500), nullable=True)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
-    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    # FTS vector — PostgreSQL generated column, auto-updated on insert/update
+    search_tsv: Mapped[Any] = mapped_column(
+        TSVECTOR,
+        Computed(
+            "to_tsvector('english', coalesce(name, '') || ' ' || coalesce(description, '') || ' ' || coalesce(industry, ''))",
+            persisted=True,
+        ),
+        nullable=True,
+    )
     created_by: Mapped[uuid.UUID | None] = mapped_column(
         PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
     )

app/models/contact.py

@@ -0,0 +1,86 @@
+"""Contact and CompanyContact (N:M join) models."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+    String,
+    Text,
+    DateTime,
+    Boolean,
+    ForeignKey,
+    Index,
+    UniqueConstraint,
+)
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class Contact(Base, TenantMixin):
+    """Contact person entity — can be linked to multiple companies via CompanyContact."""
+
+    __tablename__ = "contacts"
+    __table_args__ = (
+        Index("ix_contacts_tenant_deleted", "tenant_id", "deleted_at"),
+        Index("ix_contacts_tenant_name", "tenant_id", "last_name", "first_name"),
+        Index("ix_contacts_email", "email"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    first_name: Mapped[str] = mapped_column(String(100), nullable=False)
+    last_name: Mapped[str] = mapped_column(String(100), nullable=False)
+    email: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    phone: Mapped[str | None] = mapped_column(String(30), nullable=True)
+    mobile: Mapped[str | None] = mapped_column(String(30), nullable=True)
+    position: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    department: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    linkedin_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    notes: Mapped[str | None] = mapped_column(Text, nullable=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    created_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    updated_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+
+
+class CompanyContact(Base, TenantMixin):
+    """N:M join table between Company and Contact."""
+
+    __tablename__ = "company_contacts"
+    __table_args__ = (
+        UniqueConstraint(
+            "company_id", "contact_id", "tenant_id", name="uq_company_contact_tenant"
+        ),
+        Index("ix_cc_company", "company_id"),
+        Index("ix_cc_contact", "contact_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    company_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("companies.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    contact_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("contacts.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    role_at_company: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    is_primary: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)

app/routes/companies.py

@@ -1,42 +1,44 @@
-"""Company routes — minimal for cross-tenant and RBAC tests."""
+"""Company routes — full CRUD, FTS search, filter, pagination, export, N:M, emails."""
 
 from __future__ import annotations
 
 import uuid
-from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.audit import log_audit
+from app.core.auth import check_permission
 from app.core.db import get_db
-from app.core.auth import check_permission, filter_fields_by_permission
+from app.deps import get_current_user
-from app.deps import get_current_user, require_admin
+from app.schemas.company import CompanyCreate, CompanyUpdate
-from app.models.company import Company
+from app.services import company_service
-from app.models.role import Role
-from app.schemas.company import CompanyCreate
 
 router = APIRouter(prefix="/api/v1/companies", tags=["companies"])
 
 
 @router.get("")
 async def list_companies(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    search: str | None = Query(None),
+    industry: str | None = Query(None),
+    sort_by: str = Query("name"),
+    sort_order: str = Query("asc", pattern="^(asc|desc)$"),
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """List companies in current tenant."""
+    """List companies with pagination, FTS search, industry filter, and sorting."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
-    q = select(Company).where(
+    result = await company_service.list_companies(
-        Company.tenant_id == tenant_id,
+        db, tenant_id,
-        Company.deleted_at.is_(None),
+        page=page, page_size=page_size,
+        search=search, industry=industry,
+        sort_by=sort_by, sort_order=sort_order,
     )
-    result = await db.execute(q)
+    return result
-    companies = result.scalars().all()
-    return {"items": [_company_to_dict(c) for c in companies]}
 
 
-@router.post("")
+@router.post("", status_code=status.HTTP_201_CREATED)
 async def create_company(
     body: CompanyCreate,
     db: AsyncSession = Depends(get_db),
@@ -47,34 +49,46 @@ async def create_company(
     user_id = uuid.UUID(current_user["user_id"])
     role = current_user.get("role", "viewer")
 
-    # RBAC check
     if not check_permission(role, "companies", "create"):
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail={"detail": "Insufficient permissions", "code": "forbidden"},
         )
 
-    company = Company(
+    data = body.model_dump()
-        tenant_id=tenant_id,
+    return await company_service.create_company(db, tenant_id, user_id, data)
-        name=body.name,
-        industry=body.industry,
-        phone=body.phone,
-        email=body.email,
-        website=body.website,
-        description=body.description,
-        created_by=user_id,
-        updated_by=user_id,
-    )
-    db.add(company)
-    await db.flush()
 
-    # Audit log
-    await log_audit(
-        db, tenant_id, user_id, "create", "company", company.id,
-        changes={"name": body.name},
-    )
 
-    return _company_to_dict(company)
+@router.get("/export")
+async def export_companies(
+    format: str = Query("csv", pattern="^(csv|xlsx)$"),
+    industry: str | None = Query(None),
+    search: str | None = Query(None),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Export companies as CSV or XLSX."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    if format == "csv":
+        csv_data = await company_service.export_companies_csv(
+            db, tenant_id, industry=industry, search=search,
+        )
+        return Response(
+            content=csv_data,
+            media_type="text/csv",
+            headers={"Content-Disposition": "attachment; filename=companies.csv"},
+        )
+    elif format == "xlsx":
+        xlsx_data = await company_service.export_companies_xlsx(
+            db, tenant_id, industry=industry, search=search,
+        )
+        return Response(
+            content=xlsx_data,
+            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+            headers={"Content-Disposition": "attachment; filename=companies.xlsx"},
+        )
+    raise HTTPException(400, detail={"detail": "Invalid format", "code": "invalid_format"})
 
 
 @router.get("/{company_id}")
@@ -83,46 +97,27 @@ async def get_company(
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """Get a single company. Cross-tenant access returns 404."""
+    """Get a single company with contacts array. Cross-tenant returns 404."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
     try:
         cid = uuid.UUID(company_id)
     except ValueError:
         raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
 
-    q = select(Company).where(
+    data = await company_service.get_company_detail(db, tenant_id, cid)
-        Company.id == cid,
+    if data is None:
-        Company.tenant_id == tenant_id,  # Tenant filter = cross-tenant returns 404
-        Company.deleted_at.is_(None),
-    )
-    result = await db.execute(q)
-    company = result.scalar_one_or_none()
-    if company is None:
         raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
-
-    # Field-level permissions
-    data = _company_to_dict(company)
-    # Check if user's role has field permissions defined
-    role = current_user.get("role", "viewer")
-    if role != "admin":
-        # Check for custom role field permissions
-        role_q = select(Role).where(Role.tenant_id == tenant_id, Role.name == role)
-        role_result = await db.execute(role_q)
-        custom_role = role_result.scalar_one_or_none()
-        if custom_role and custom_role.field_permissions:
-            data = filter_fields_by_permission(data, custom_role.field_permissions, role)
-
     return data
 
 
-@router.patch("/{company_id}")
+@router.put("/{company_id}")
 async def update_company(
     company_id: str,
-    body: CompanyCreate,
+    body: CompanyUpdate,
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """Update a company."""
+    """Update a company (full PUT)."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
     user_id = uuid.UUID(current_user["user_id"])
     role = current_user.get("role", "viewer")
@@ -135,42 +130,21 @@ async def update_company(
     except ValueError:
         raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
 
-    q = select(Company).where(Company.id == cid, Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
+    data = body.model_dump(exclude_unset=True)
-    result = await db.execute(q)
+    result = await company_service.update_company(db, tenant_id, user_id, cid, data)
-    company = result.scalar_one_or_none()
+    if result is None:
-    if company is None:
         raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
-
+    return result
-    changes: dict[str, Any] = {}
-    if body.name is not None:
-        changes["name"] = {"old": company.name, "new": body.name}
-        company.name = body.name
-    if body.industry is not None:
-        company.industry = body.industry
-    if body.phone is not None:
-        company.phone = body.phone
-    if body.email is not None:
-        company.email = body.email
-    if body.website is not None:
-        company.website = body.website
-    if body.description is not None:
-        company.description = body.description
-    company.updated_by = user_id
-
-    await db.flush()
-    await log_audit(db, tenant_id, user_id, "update", "company", cid, changes=changes)
-
-    return _company_to_dict(company)
 
 
 @router.delete("/{company_id}")
 async def delete_company(
     company_id: str,
+    cascade: bool = Query(False),
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """Soft-delete a company."""
+    """Soft-delete a company. Use cascade=true to also delete N:M links."""
-    from datetime import datetime, timezone
     tenant_id = uuid.UUID(current_user["tenant_id"])
     user_id = uuid.UUID(current_user["user_id"])
     role = current_user.get("role", "viewer")
@@ -183,28 +157,91 @@ async def delete_company(
     except ValueError:
         raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
 
-    q = select(Company).where(Company.id == cid, Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
+    deleted = await company_service.soft_delete_company(db, tenant_id, user_id, cid, cascade=cascade)
-    result = await db.execute(q)
+    if not deleted:
-    company = result.scalar_one_or_none()
-    if company is None:
         raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
 
-    company.deleted_at = datetime.now(timezone.utc)
-    company.updated_by = user_id
-
-    await log_audit(db, tenant_id, user_id, "delete", "company", cid, changes={"name": company.name})
-
-    from fastapi import Response
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 
-def _company_to_dict(c: Company) -> dict[str, Any]:
+@router.post("/{company_id}/contacts/{contact_id}")
-    """Convert company to response dict."""
+async def link_company_contact(
-    return {
+    company_id: str,
-        "id": str(c.id),
+    contact_id: str,
-        "name": c.name,
+    db: AsyncSession = Depends(get_db),
-        "industry": c.industry,
+    current_user: dict = Depends(get_current_user),
-        "phone": c.phone,
+):
-        "email": c.email,
+    """Link a contact to a company (N:M)."""
-        "annual_revenue": None,  # Minimal model — would be from DB column
+    tenant_id = uuid.UUID(current_user["tenant_id"])
-    }
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "update"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        comp_id = uuid.UUID(company_id)
+        cont_id = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid ID", "code": "invalid_id"})
+
+    result = await company_service.link_contact(db, tenant_id, user_id, comp_id, cont_id)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Company or contact not found", "code": "not_found"})
+    return result
+
+
+@router.delete("/{company_id}/contacts/{contact_id}")
+async def unlink_company_contact(
+    company_id: str,
+    contact_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Unlink a contact from a company (N:M)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "update"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        comp_id = uuid.UUID(company_id)
+        cont_id = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid ID", "code": "invalid_id"})
+
+    unlinked = await company_service.unlink_contact(db, tenant_id, user_id, comp_id, cont_id)
+    if not unlinked:
+        raise HTTPException(404, detail={"detail": "Link not found", "code": "not_found"})
+
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.get("/{company_id}/emails")
+async def get_company_emails(
+    company_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Get emails for a company (mail plugin inactive — returns empty array)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        cid = uuid.UUID(company_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
+
+    # Verify company exists
+    from sqlalchemy import select
+    from app.models.company import Company
+    q = select(Company).where(
+        Company.id == cid,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    if result.scalar_one_or_none() is None:
+        raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
+
+    return []

app/routes/contacts.py

@@ -0,0 +1,134 @@
+"""Contact routes — CRUD, N:M, soft-delete, GDPR hard-delete."""
+
+from __future__ import annotations
+
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.auth import check_permission
+from app.core.db import get_db
+from app.deps import get_current_user
+from app.schemas.contact import ContactCreate, ContactUpdate
+from app.services import contact_service
+
+router = APIRouter(prefix="/api/v1/contacts", tags=["contacts"])
+
+
+@router.get("")
+async def list_contacts(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    search: str | None = Query(None),
+    sort_by: str = Query("last_name"),
+    sort_order: str = Query("asc", pattern="^(asc|desc)$"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """List contacts with pagination and optional search."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    result = await contact_service.list_contacts(
+        db, tenant_id,
+        page=page, page_size=page_size,
+        search=search, sort_by=sort_by, sort_order=sort_order,
+    )
+    return result
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+async def create_contact(
+    body: ContactCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Create a contact. Optionally link to companies via company_ids array."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "contacts", "create"):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={"detail": "Insufficient permissions", "code": "forbidden"},
+        )
+
+    data = body.model_dump()
+    return await contact_service.create_contact(db, tenant_id, user_id, data)
+
+
+@router.get("/{contact_id}")
+async def get_contact(
+    contact_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Get a single contact with companies array. Cross-tenant returns 404."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        cid = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid contact_id", "code": "invalid_id"})
+
+    data = await contact_service.get_contact_detail(db, tenant_id, cid)
+    if data is None:
+        raise HTTPException(404, detail={"detail": "Contact not found", "code": "not_found"})
+    return data
+
+
+@router.put("/{contact_id}")
+async def update_contact(
+    contact_id: str,
+    body: ContactUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Update a contact."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "contacts", "update"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        cid = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid contact_id", "code": "invalid_id"})
+
+    data = body.model_dump(exclude_unset=True)
+    result = await contact_service.update_contact(db, tenant_id, user_id, cid, data)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Contact not found", "code": "not_found"})
+    return result
+
+
+@router.delete("/{contact_id}")
+async def delete_contact(
+    contact_id: str,
+    gdpr: bool = Query(False),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Delete a contact. Default: soft-delete. Use gdpr=true for hard-delete with deletion_log."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "contacts", "delete"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        cid = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid contact_id", "code": "invalid_id"})
+
+    if gdpr:
+        deleted = await contact_service.gdpr_hard_delete_contact(db, tenant_id, user_id, cid)
+    else:
+        deleted = await contact_service.soft_delete_contact(db, tenant_id, user_id, cid)
+
+    if not deleted:
+        raise HTTPException(404, detail={"detail": "Contact not found", "code": "not_found"})
+
+    return Response(status_code=status.HTTP_204_NO_CONTENT)

app/routes/import_export.py

@@ -0,0 +1,67 @@
+"""Import/export routes — CSV import, dry-run preview, CSV/XLSX export."""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, Query, Response, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.auth import check_permission
+from app.core.db import get_db
+from app.deps import get_current_user
+from app.services import import_export_service
+from app.services import company_service
+
+router = APIRouter(prefix="/api/v1", tags=["import_export"])
+
+
+@router.post("/import")
+async def import_csv(
+    file: UploadFile = File(...),
+    entity_type: str = Form("companies"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Import companies or contacts from CSV file.
+    entity_type: 'companies' or 'contacts'.
+    """
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "create"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    content = await file.read()
+    csv_content = content.decode("utf-8")
+
+    result = await import_export_service.import_csv(
+        db, tenant_id, user_id, csv_content, entity_type=entity_type, dry_run=False,
+    )
+    return result
+
+
+@router.post("/import/preview")
+async def import_csv_preview(
+    file: UploadFile = File(...),
+    entity_type: str = Form("companies"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Preview CSV import (dry-run — no DB changes)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "read"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    content = await file.read()
+    csv_content = content.decode("utf-8")
+
+    result = await import_export_service.import_csv(
+        db, tenant_id, user_id, csv_content, entity_type=entity_type, dry_run=True,
+    )
+    return result

app/schemas/company.py

@@ -1,4 +1,4 @@
-"""Company schema (minimal for cross-tenant test)."""
+"""Company schemas — create, update, read, list, pagination, search, filter."""
 
 from __future__ import annotations
 
@@ -7,17 +7,49 @@ from pydantic import BaseModel, Field
 
 class CompanyCreate(BaseModel):
     name: str = Field(..., min_length=1, max_length=100)
-    industry: str | None = None
+    account_number: str | None = Field(None, max_length=40)
-    phone: str | None = None
+    industry: str | None = Field(None, max_length=50)
-    email: str | None = None
+    phone: str | None = Field(None, max_length=30)
-    website: str | None = None
+    email: str | None = Field(None, max_length=255)
+    website: str | None = Field(None, max_length=500)
+    description: str | None = None
+
+
+class CompanyUpdate(BaseModel):
+    name: str | None = Field(None, min_length=1, max_length=100)
+    account_number: str | None = Field(None, max_length=40)
+    industry: str | None = Field(None, max_length=50)
+    phone: str | None = Field(None, max_length=30)
+    email: str | None = Field(None, max_length=255)
+    website: str | None = Field(None, max_length=500)
     description: str | None = None
 
 
 class CompanyResponse(BaseModel):
     id: str
     name: str
+    account_number: str | None = None
     industry: str | None = None
     phone: str | None = None
     email: str | None = None
-    annual_revenue: float | None = None
+    website: str | None = None
+    description: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+
+
+class CompanyDetailResponse(CompanyResponse):
+    contacts: list[dict] = Field(default_factory=list)
+
+
+class CompanyListResponse(BaseModel):
+    items: list[CompanyResponse]
+    total: int
+    page: int
+    page_size: int
+
+
+class CompanyExportRequest(BaseModel):
+    format: str = Field("csv", pattern="^(csv|xlsx)$")
+    industry: str | None = None
+    search: str | None = None

app/schemas/contact.py

@@ -0,0 +1,61 @@
+"""Contact schemas — create, update, read, list."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class ContactCreate(BaseModel):
+    first_name: str = Field(..., min_length=1, max_length=100)
+    last_name: str = Field(..., min_length=1, max_length=100)
+    email: str | None = Field(None, max_length=255)
+    phone: str | None = Field(None, max_length=30)
+    mobile: str | None = Field(None, max_length=30)
+    position: str | None = Field(None, max_length=100)
+    department: str | None = Field(None, max_length=100)
+    linkedin_url: str | None = Field(None, max_length=500)
+    notes: str | None = None
+    company_ids: list[str] | None = None
+
+
+class ContactUpdate(BaseModel):
+    first_name: str | None = Field(None, min_length=1, max_length=100)
+    last_name: str | None = Field(None, min_length=1, max_length=100)
+    email: str | None = Field(None, max_length=255)
+    phone: str | None = Field(None, max_length=30)
+    mobile: str | None = Field(None, max_length=30)
+    position: str | None = Field(None, max_length=100)
+    department: str | None = Field(None, max_length=100)
+    linkedin_url: str | None = Field(None, max_length=500)
+    notes: str | None = None
+
+
+class ContactResponse(BaseModel):
+    id: str
+    first_name: str
+    last_name: str
+    email: str | None = None
+    phone: str | None = None
+    mobile: str | None = None
+    position: str | None = None
+    department: str | None = None
+    linkedin_url: str | None = None
+    notes: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+
+
+class ContactDetailResponse(ContactResponse):
+    companies: list[dict] = Field(default_factory=list)
+
+
+class ContactListResponse(BaseModel):
+    items: list[ContactResponse]
+    total: int
+    page: int
+    page_size: int
+
+
+class CompanyLinkRequest(BaseModel):
+    role_at_company: str | None = Field(None, max_length=100)
+    is_primary: bool = False

app/services/company_service.py

@@ -0,0 +1,453 @@
+"""Company service — CRUD, FTS search, filter, pagination, soft-delete, N:M, export."""
+
+from __future__ import annotations
+
+import csv
+import io
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select, func, or_, desc, asc, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit, log_deletion
+from app.models.company import Company
+from app.models.contact import Contact, CompanyContact
+
+
+def _company_to_dict(c: Company, include_contacts: bool = False) -> dict[str, Any]:
+    """Serialize a Company ORM object to dict."""
+    data: dict[str, Any] = {
+        "id": str(c.id),
+        "name": c.name,
+        "account_number": c.account_number,
+        "industry": c.industry,
+        "phone": c.phone,
+        "email": c.email,
+        "website": c.website,
+        "description": c.description,
+        "created_at": c.created_at.isoformat() if c.created_at else None,
+        "updated_at": c.updated_at.isoformat() if c.updated_at else None,
+    }
+    if include_contacts:
+        data["contacts"] = []  # populated by caller if needed
+    return data
+
+
+async def list_companies(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    page: int = 1,
+    page_size: int = 20,
+    search: str | None = None,
+    industry: str | None = None,
+    sort_by: str = "name",
+    sort_order: str = "asc",
+) -> dict[str, Any]:
+    """List companies with pagination, FTS search, industry filter, and sorting.
+
+    Returns {items, total, page, page_size}.
+    """
+    page = max(1, page)
+    page_size = max(1, min(100, page_size))
+
+    base = select(Company).where(
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+
+    # Industry filter
+    if industry:
+        base = base.where(Company.industry == industry)
+
+    # FTS search using PostgreSQL tsvector @@ plainto_tsquery, with ILIKE fallback
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            or_(
+                Company.search_tsv.op("@@")(
+                    func.plainto_tsquery("english", search)
+                ),
+                Company.name.ilike(pattern),
+                Company.industry.ilike(pattern),
+                Company.description.ilike(pattern),
+            )
+        )
+
+    # Sorting
+    sort_col = getattr(Company, sort_by, Company.name)
+    if sort_order == "desc":
+        base = base.order_by(desc(sort_col))
+    else:
+        base = base.order_by(asc(sort_col))
+
+    # Count total (without pagination)
+    count_q = select(func.count()).select_from(base.subquery())
+    total_result = await db.execute(count_q)
+    total = total_result.scalar_one()
+
+    # Paginate
+    offset = (page - 1) * page_size
+    paginated = base.offset(offset).limit(page_size)
+    result = await db.execute(paginated)
+    companies = result.scalars().all()
+
+    return {
+        "items": [_company_to_dict(c) for c in companies],
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
+async def get_company_detail(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    company_id: uuid.UUID,
+) -> dict[str, Any] | None:
+    """Get a single company with its contacts array."""
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    company = result.scalar_one_or_none()
+    if company is None:
+        return None
+
+    data = _company_to_dict(company, include_contacts=True)
+    # Fetch linked contacts
+    contacts_q = (
+        select(Contact, CompanyContact)
+        .join(CompanyContact, CompanyContact.contact_id == Contact.id)
+        .where(
+            CompanyContact.company_id == company_id,
+            CompanyContact.tenant_id == tenant_id,
+            Contact.deleted_at.is_(None),
+        )
+    )
+    contacts_result = await db.execute(contacts_q)
+    contacts_list = []
+    for contact, link in contacts_result.all():
+        contacts_list.append({
+            "id": str(contact.id),
+            "first_name": contact.first_name,
+            "last_name": contact.last_name,
+            "email": contact.email,
+            "phone": contact.phone,
+            "position": contact.position,
+            "role_at_company": link.role_at_company,
+            "is_primary": link.is_primary,
+        })
+    data["contacts"] = contacts_list
+    return data
+
+
+async def create_company(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any]:
+    """Create a new company and audit-log it."""
+    company = Company(
+        tenant_id=tenant_id,
+        name=data["name"],
+        account_number=data.get("account_number"),
+        industry=data.get("industry"),
+        phone=data.get("phone"),
+        email=data.get("email"),
+        website=data.get("website"),
+        description=data.get("description"),
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(company)
+    await db.flush()
+    await db.refresh(company)
+    await log_audit(
+        db, tenant_id, user_id, "create", "company", company.id,
+        changes={"name": data["name"]},
+    )
+    return _company_to_dict(company)
+
+
+async def update_company(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any] | None:
+    """Update a company (partial update) and audit-log changes."""
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    company = result.scalar_one_or_none()
+    if company is None:
+        return None
+
+    changes: dict[str, Any] = {}
+    for field in ("name", "account_number", "industry", "phone", "email", "website", "description"):
+        if field in data and data[field] is not None:
+            old_val = getattr(company, field)
+            changes[field] = {"old": old_val, "new": data[field]}
+            setattr(company, field, data[field])
+    company.updated_by = user_id
+
+    await db.flush()
+    await db.refresh(company)
+    await log_audit(db, tenant_id, user_id, "update", "company", company_id, changes=changes)
+    return _company_to_dict(company)
+
+
+async def soft_delete_company(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    cascade: bool = False,
+) -> bool:
+    """Soft-delete a company. If cascade=True, also soft-delete linked CompanyContact rows."""
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    company = result.scalar_one_or_none()
+    if company is None:
+        return False
+
+    company.deleted_at = datetime.now(timezone.utc)
+    company.updated_by = user_id
+
+    if cascade:
+        # Delete N:M links for this company
+        await db.execute(
+            delete(CompanyContact).where(
+                CompanyContact.company_id == company_id,
+                CompanyContact.tenant_id == tenant_id,
+            )
+        )
+
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "delete", "company", company_id,
+        changes={"name": company.name, "cascade": cascade},
+    )
+    return True
+
+
+async def link_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    contact_id: uuid.UUID,
+    role_at_company: str | None = None,
+    is_primary: bool = False,
+) -> dict[str, Any] | None:
+    """Link a contact to a company (N:M). Returns link data or None if either not found."""
+    # Verify company exists and is in tenant
+    comp_q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    comp_result = await db.execute(comp_q)
+    if comp_result.scalar_one_or_none() is None:
+        return None
+
+    # Verify contact exists and is in tenant
+    cont_q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    cont_result = await db.execute(cont_q)
+    if cont_result.scalar_one_or_none() is None:
+        return None
+
+    # Check if link already exists
+    existing_q = select(CompanyContact).where(
+        CompanyContact.company_id == company_id,
+        CompanyContact.contact_id == contact_id,
+        CompanyContact.tenant_id == tenant_id,
+    )
+    existing_result = await db.execute(existing_q)
+    existing = existing_result.scalar_one_or_none()
+    if existing is not None:
+        # Update existing link
+        existing.role_at_company = role_at_company
+        existing.is_primary = is_primary
+        await db.flush()
+        await log_audit(
+            db, tenant_id, user_id, "link", "company_contact", existing.id,
+            changes={"company_id": str(company_id), "contact_id": str(contact_id)},
+        )
+        return {
+            "id": str(existing.id),
+            "company_id": str(company_id),
+            "contact_id": str(contact_id),
+            "role_at_company": role_at_company,
+            "is_primary": is_primary,
+        }
+
+    link = CompanyContact(
+        tenant_id=tenant_id,
+        company_id=company_id,
+        contact_id=contact_id,
+        role_at_company=role_at_company,
+        is_primary=is_primary,
+    )
+    db.add(link)
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "link", "company_contact", link.id,
+        changes={"company_id": str(company_id), "contact_id": str(contact_id)},
+    )
+    return {
+        "id": str(link.id),
+        "company_id": str(company_id),
+        "contact_id": str(contact_id),
+        "role_at_company": role_at_company,
+        "is_primary": is_primary,
+    }
+
+
+async def unlink_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> bool:
+    """Unlink a contact from a company (N:M)."""
+    q = select(CompanyContact).where(
+        CompanyContact.company_id == company_id,
+        CompanyContact.contact_id == contact_id,
+        CompanyContact.tenant_id == tenant_id,
+    )
+    result = await db.execute(q)
+    link = result.scalar_one_or_none()
+    if link is None:
+        return False
+
+    await db.delete(link)
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "unlink", "company_contact", link.id,
+        changes={"company_id": str(company_id), "contact_id": str(contact_id)},
+    )
+    return True
+
+
+async def export_companies_csv(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    industry: str | None = None,
+    search: str | None = None,
+) -> str:
+    """Export companies as CSV string."""
+    base = select(Company).where(
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    if industry:
+        base = base.where(Company.industry == industry)
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            or_(
+                Company.search_tsv.op("@@")(
+                    func.plainto_tsquery("english", search)
+                ),
+                Company.name.ilike(pattern),
+                Company.industry.ilike(pattern),
+                Company.description.ilike(pattern),
+            )
+        )
+    base = base.order_by(Company.name)
+    result = await db.execute(base)
+    companies = result.scalars().all()
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["id", "name", "account_number", "industry", "phone", "email", "website", "description"])
+    for c in companies:
+        writer.writerow([
+            str(c.id), c.name, c.account_number or "", c.industry or "",
+            c.phone or "", c.email or "", c.website or "", c.description or "",
+        ])
+    return output.getvalue()
+
+
+async def export_companies_xlsx(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    industry: str | None = None,
+    search: str | None = None,
+) -> bytes:
+    """Export companies as XLSX bytes."""
+    from openpyxl import Workbook
+
+    base = select(Company).where(
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    if industry:
+        base = base.where(Company.industry == industry)
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            or_(
+                Company.search_tsv.op("@@")(
+                    func.plainto_tsquery("english", search)
+                ),
+                Company.name.ilike(pattern),
+                Company.industry.ilike(pattern),
+                Company.description.ilike(pattern),
+            )
+        )
+    base = base.order_by(Company.name)
+    result = await db.execute(base)
+    companies = result.scalars().all()
+
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "Companies"
+    headers = ["id", "name", "account_number", "industry", "phone", "email", "website", "description"]
+    ws.append(headers)
+    for c in companies:
+        ws.append([
+            str(c.id), c.name, c.account_number or "", c.industry or "",
+            c.phone or "", c.email or "", c.website or "", c.description or "",
+        ])
+
+    buf = io.BytesIO()
+    wb.save(buf)
+    return buf.getvalue()
+
+
+async def get_company_emails(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    company_id: uuid.UUID,
+) -> list[dict[str, Any]]:
+    """Get emails for a company (mail plugin inactive — returns empty array)."""
+    # Verify company exists
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    if result.scalar_one_or_none() is None:
+        return []  # Caller should check existence separately
+    return []

app/services/contact_service.py

@@ -0,0 +1,281 @@
+"""Contact service — CRUD, N:M linking, soft-delete, GDPR hard-delete."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select, func, desc, asc, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit, log_deletion
+from app.models.contact import Contact, CompanyContact
+from app.models.company import Company
+
+
+def _contact_to_dict(c: Contact, include_companies: bool = False) -> dict[str, Any]:
+    """Serialize a Contact ORM object to dict."""
+    data: dict[str, Any] = {
+        "id": str(c.id),
+        "first_name": c.first_name,
+        "last_name": c.last_name,
+        "email": c.email,
+        "phone": c.phone,
+        "mobile": c.mobile,
+        "position": c.position,
+        "department": c.department,
+        "linkedin_url": c.linkedin_url,
+        "notes": c.notes,
+        "created_at": c.created_at.isoformat() if c.created_at else None,
+        "updated_at": c.updated_at.isoformat() if c.updated_at else None,
+    }
+    if include_companies:
+        data["companies"] = []
+    return data
+
+
+async def list_contacts(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    page: int = 1,
+    page_size: int = 20,
+    search: str | None = None,
+    sort_by: str = "last_name",
+    sort_order: str = "asc",
+) -> dict[str, Any]:
+    """List contacts with pagination and optional search."""
+    page = max(1, page)
+    page_size = max(1, min(100, page_size))
+
+    base = select(Contact).where(
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            (Contact.first_name.ilike(pattern))
+            | (Contact.last_name.ilike(pattern))
+            | (Contact.email.ilike(pattern))
+        )
+
+    sort_col = getattr(Contact, sort_by, Contact.last_name)
+    if sort_order == "desc":
+        base = base.order_by(desc(sort_col))
+    else:
+        base = base.order_by(asc(sort_col))
+
+    count_q = select(func.count()).select_from(base.subquery())
+    total_result = await db.execute(count_q)
+    total = total_result.scalar_one()
+
+    offset = (page - 1) * page_size
+    paginated = base.offset(offset).limit(page_size)
+    result = await db.execute(paginated)
+    contacts = result.scalars().all()
+
+    return {
+        "items": [_contact_to_dict(c) for c in contacts],
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
+async def get_contact_detail(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> dict[str, Any] | None:
+    """Get a single contact with its linked companies array."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return None
+
+    data = _contact_to_dict(contact, include_companies=True)
+    companies_q = (
+        select(Company, CompanyContact)
+        .join(CompanyContact, CompanyContact.company_id == Company.id)
+        .where(
+            CompanyContact.contact_id == contact_id,
+            CompanyContact.tenant_id == tenant_id,
+            Company.deleted_at.is_(None),
+        )
+    )
+    companies_result = await db.execute(companies_q)
+    companies_list = []
+    for company, link in companies_result.all():
+        companies_list.append({
+            "id": str(company.id),
+            "name": company.name,
+            "industry": company.industry,
+            "role_at_company": link.role_at_company,
+            "is_primary": link.is_primary,
+        })
+    data["companies"] = companies_list
+    return data
+
+
+async def create_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any]:
+    """Create a new contact, optionally linking to companies via company_ids."""
+    contact = Contact(
+        tenant_id=tenant_id,
+        first_name=data["first_name"],
+        last_name=data["last_name"],
+        email=data.get("email"),
+        phone=data.get("phone"),
+        mobile=data.get("mobile"),
+        position=data.get("position"),
+        department=data.get("department"),
+        linkedin_url=data.get("linkedin_url"),
+        notes=data.get("notes"),
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(contact)
+    await db.flush()
+
+    # Link to companies if company_ids provided
+    company_ids = data.get("company_ids")
+    linked_companies = []
+    if company_ids:
+        for cid_str in company_ids:
+            try:
+                cid = uuid.UUID(cid_str) if isinstance(cid_str, str) else cid_str
+            except (ValueError, TypeError):
+                continue
+            # Verify company exists in tenant
+            comp_q = select(Company).where(
+                Company.id == cid,
+                Company.tenant_id == tenant_id,
+                Company.deleted_at.is_(None),
+            )
+            comp_result = await db.execute(comp_q)
+            if comp_result.scalar_one_or_none() is None:
+                continue
+            link = CompanyContact(
+                tenant_id=tenant_id,
+                company_id=cid,
+                contact_id=contact.id,
+            )
+            db.add(link)
+            linked_companies.append(str(cid))
+        await db.flush()
+
+    await db.refresh(contact)
+    await log_audit(
+        db, tenant_id, user_id, "create", "contact", contact.id,
+        changes={"first_name": data["first_name"], "last_name": data["last_name"], "linked_companies": linked_companies},
+    )
+    return _contact_to_dict(contact)
+
+
+async def update_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    contact_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any] | None:
+    """Update a contact (partial update) and audit-log changes."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return None
+
+    changes: dict[str, Any] = {}
+    for field in ("first_name", "last_name", "email", "phone", "mobile", "position", "department", "linkedin_url", "notes"):
+        if field in data and data[field] is not None:
+            old_val = getattr(contact, field)
+            changes[field] = {"old": old_val, "new": data[field]}
+            setattr(contact, field, data[field])
+    contact.updated_by = user_id
+
+    await db.flush()
+    await db.refresh(contact)
+    await log_audit(db, tenant_id, user_id, "update", "contact", contact_id, changes=changes)
+    return _contact_to_dict(contact)
+
+
+async def soft_delete_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> bool:
+    """Soft-delete a contact (set deleted_at)."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return False
+
+    contact.deleted_at = datetime.now(timezone.utc)
+    contact.updated_by = user_id
+
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "delete", "contact", contact_id,
+        changes={"first_name": contact.first_name, "last_name": contact.last_name},
+    )
+    return True
+
+
+async def gdpr_hard_delete_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> bool:
+    """GDPR hard-delete: physical delete + deletion_log entry with snapshot."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return False
+
+    # Capture snapshot before deletion
+    snapshot = _contact_to_dict(contact)
+
+    # Delete N:M links first
+    await db.execute(
+        delete(CompanyContact).where(
+            CompanyContact.contact_id == contact_id,
+            CompanyContact.tenant_id == tenant_id,
+        )
+    )
+
+    # Physical delete
+    await db.delete(contact)
+    await db.flush()
+
+    # Deletion log (immutable snapshot)
+    await log_deletion(
+        db, tenant_id, user_id, "contact", contact_id, snapshot,
+    )
+    return True

app/services/import_export_service.py

@@ -0,0 +1,216 @@
+"""Import/export service — CSV import with dry-run preview, CSV/XLSX export."""
+
+from __future__ import annotations
+
+import csv
+import io
+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.models.company import Company
+from app.models.contact import Contact
+from app.services.company_service import _company_to_dict
+from app.services.contact_service import _contact_to_dict
+
+
+# Expected CSV columns for each entity type
+COMPANY_COLUMNS = ["name", "industry", "phone", "email", "website", "description"]
+CONTACT_COLUMNS = ["first_name", "last_name", "email", "phone", "mobile", "position", "department"]
+
+
+def _parse_csv(content: str) -> list[dict[str, str]]:
+    """Parse CSV content into list of dicts."""
+    reader = csv.DictReader(io.StringIO(content))
+    return [dict(row) for row in reader]
+
+
+def _validate_row(row: dict[str, str], required: list[str]) -> list[str]:
+    """Validate a single row. Returns list of error messages (empty if valid)."""
+    errors = []
+    for col in required:
+        val = row.get(col, "").strip()
+        if not val:
+            errors.append(f"Missing required field: {col}")
+    return errors
+
+
+async def import_companies(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    csv_content: str,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """Import companies from CSV. If dry_run=True, no DB changes are made.
+
+    Returns {total, valid, invalid, errors, created (empty in dry_run)}.
+    """
+    rows = _parse_csv(csv_content)
+    total = len(rows)
+    valid_rows = []
+    errors = []
+
+    for idx, row in enumerate(rows, start=1):
+        row_errors = _validate_row(row, ["name"])
+        if row_errors:
+            for e in row_errors:
+                errors.append({"row": idx, "error": e})
+        else:
+            valid_rows.append(row)
+
+    if dry_run:
+        return {
+            "total": total,
+            "valid": len(valid_rows),
+            "invalid": len(errors),
+            "errors": errors,
+            "created": [],
+            "dry_run": True,
+        }
+
+    created = []
+    for row in valid_rows:
+        company = Company(
+            tenant_id=tenant_id,
+            name=row["name"].strip(),
+            industry=row.get("industry", "").strip() or None,
+            phone=row.get("phone", "").strip() or None,
+            email=row.get("email", "").strip() or None,
+            website=row.get("website", "").strip() or None,
+            description=row.get("description", "").strip() or None,
+            created_by=user_id,
+            updated_by=user_id,
+        )
+        db.add(company)
+        await db.flush()
+        await log_audit(
+            db, tenant_id, user_id, "import", "company", company.id,
+            changes={"name": company.name},
+        )
+        created.append(_company_to_dict(company))
+
+    return {
+        "total": total,
+        "valid": len(valid_rows),
+        "invalid": len(errors),
+        "errors": errors,
+        "created": created,
+        "dry_run": False,
+    }
+
+
+async def import_contacts(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    csv_content: str,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """Import contacts from CSV. If dry_run=True, no DB changes are made.
+
+    Returns {total, valid, invalid, errors, created (empty in dry_run)}.
+    """
+    rows = _parse_csv(csv_content)
+    total = len(rows)
+    valid_rows = []
+    errors = []
+
+    for idx, row in enumerate(rows, start=1):
+        row_errors = _validate_row(row, ["first_name", "last_name"])
+        if row_errors:
+            for e in row_errors:
+                errors.append({"row": idx, "error": e})
+        else:
+            valid_rows.append(row)
+
+    if dry_run:
+        return {
+            "total": total,
+            "valid": len(valid_rows),
+            "invalid": len(errors),
+            "errors": errors,
+            "created": [],
+            "dry_run": True,
+        }
+
+    created = []
+    for row in valid_rows:
+        contact = Contact(
+            tenant_id=tenant_id,
+            first_name=row["first_name"].strip(),
+            last_name=row["last_name"].strip(),
+            email=row.get("email", "").strip() or None,
+            phone=row.get("phone", "").strip() or None,
+            mobile=row.get("mobile", "").strip() or None,
+            position=row.get("position", "").strip() or None,
+            department=row.get("department", "").strip() or None,
+            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={"first_name": contact.first_name, "last_name": contact.last_name},
+        )
+        created.append(_contact_to_dict(contact))
+
+    return {
+        "total": total,
+        "valid": len(valid_rows),
+        "invalid": len(errors),
+        "errors": errors,
+        "created": created,
+        "dry_run": False,
+    }
+
+
+async def import_csv(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    csv_content: str,
+    entity_type: str,
+    dry_run: bool = False,
+) -> 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)
+    elif entity_type == "contacts":
+        return await import_contacts(db, tenant_id, user_id, csv_content, dry_run=dry_run)
+    else:
+        return {
+            "total": 0,
+            "valid": 0,
+            "invalid": 0,
+            "errors": [{"row": 0, "error": f"Unknown entity_type: {entity_type}"}],
+            "created": [],
+            "dry_run": dry_run,
+        }
+
+
+async def export_contacts_csv(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+) -> str:
+    """Export contacts as CSV string."""
+    q = select(Contact).where(
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    ).order_by(Contact.last_name, Contact.first_name)
+    result = await db.execute(q)
+    contacts = result.scalars().all()
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["id", "first_name", "last_name", "email", "phone", "mobile", "position", "department"])
+    for c in contacts:
+        writer.writerow([
+            str(c.id), c.first_name, c.last_name, c.email or "",
+            c.phone or "", c.mobile or "", c.position or "", c.department or "",
+        ])
+    return output.getvalue()

requirements.txt

@@ -28,3 +28,4 @@ aiofiles>=23.2
 # Templates (optional)
 jinja2>=3.1
 greenlet>=3.0
+openpyxl>=3.1

tests/conftest.py

@@ -31,6 +31,7 @@ from app.models.tenant import Tenant
 from app.models.user import User, UserTenant
 from app.models.role import Role
 from app.models.company import Company
+from app.models.contact import Contact, CompanyContact
 from app.main import create_app
 
 
@@ -86,7 +87,7 @@ def clean_tables(db_setup):
     sync_eng = _get_sync_engine()
     with sync_eng.connect() as conn:
         # TRUNCATE all tables with CASCADE — fast and reliable isolation
-        conn.execute(text("TRUNCATE TABLE api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants CASCADE;"))
+        conn.execute(text("TRUNCATE TABLE company_contacts, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants CASCADE;"))
         conn.commit()
     sync_eng.dispose()
     yield

tests/test_companies.py

@@ -0,0 +1,351 @@
+"""Company tests — ACs 1-13, 22-24: CRUD, search, filter, export, N:M, emails, audit, soft-delete."""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+
+from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
+
+
+@pytest.mark.asyncio
+class TestCompanyList:
+    """ACs 1-3: List, search, filter+sort."""
+
+    async def test_list_companies_returns_200_paginated(self, client: AsyncClient, db_session):
+        """AC 1: GET /api/v1/companies -> 200 + paginated (total/page/page_size)."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "items" in data
+        assert "total" in data
+        assert "page" in data
+        assert "page_size" in data
+        assert data["total"] == 1  # seed creates one company in tenant A
+
+    async def test_list_companies_search_fts(self, client: AsyncClient, db_session):
+        """AC 2: GET /api/v1/companies?search=Tech -> 200 + FTS results."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create a company with 'Tech' in name/description
+        resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "TechCorp", "industry": "IT", "description": "Technology solutions"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 201
+        # Search for 'Tech'
+        resp = await client.get("/api/v1/companies?search=Tech", headers=ORIGIN_HEADER)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total"] >= 1
+        names = [item["name"] for item in data["items"]]
+        assert "TechCorp" in names
+
+    async def test_list_companies_filter_and_sort(self, client: AsyncClient, db_session):
+        """AC 3: GET /api/v1/companies?industry=IT&sort_by=name&sort_order=asc -> 200 + filtered+sorted."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create additional companies
+        await client.post(
+            "/api/v1/companies",
+            json={"name": "ZetaTech", "industry": "IT"},
+            headers=ORIGIN_HEADER,
+        )
+        await client.post(
+            "/api/v1/companies",
+            json={"name": "AlphaTech", "industry": "IT"},
+            headers=ORIGIN_HEADER,
+        )
+        resp = await client.get(
+            "/api/v1/companies?industry=IT&sort_by=name&sort_order=asc",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        names = [item["name"] for item in data["items"]]
+        assert names == sorted(names)
+        assert all(item["industry"] == "IT" for item in data["items"])
+
+
+@pytest.mark.asyncio
+class TestCompanyCreate:
+    """ACs 4-5: Create valid, missing name."""
+
+    async def test_create_company_valid_returns_201(self, client: AsyncClient, db_session):
+        """AC 4: POST /api/v1/companies valid -> 201 + company object."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "New Corp", "industry": "Finance"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 201
+        data = resp.json()
+        assert data["name"] == "New Corp"
+        assert data["industry"] == "Finance"
+        assert "id" in data
+
+    async def test_create_company_missing_name_returns_422(self, client: AsyncClient, db_session):
+        """AC 5: POST /api/v1/companies missing name -> 422."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        resp = await client.post(
+            "/api/v1/companies",
+            json={"industry": "Finance"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+class TestCompanyDetail:
+    """AC 6: Get single company with contacts array."""
+
+    async def test_get_company_returns_200_with_contacts(self, client: AsyncClient, db_session):
+        """AC 6: GET /api/v1/companies/{id} -> 200 + detail inkl. contacts array."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create a company
+        create_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Detail Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = create_resp.json()["id"]
+        resp = await client.get(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["name"] == "Detail Corp"
+        assert "contacts" in data
+        assert isinstance(data["contacts"], list)
+
+
+@pytest.mark.asyncio
+class TestCompanyUpdate:
+    """AC 7: Update company."""
+
+    async def test_update_company_returns_200(self, client: AsyncClient, db_session):
+        """AC 7: PUT /api/v1/companies/{id} -> 200 + updated."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        create_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Old Name"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = create_resp.json()["id"]
+        resp = await client.put(
+            f"/api/v1/companies/{company_id}",
+            json={"name": "New Name", "industry": "Healthcare"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["name"] == "New Name"
+        assert data["industry"] == "Healthcare"
+
+
+@pytest.mark.asyncio
+class TestCompanyDelete:
+    """ACs 8-9: Soft-delete, cascade delete."""
+
+    async def test_delete_company_soft_delete_returns_204(self, client: AsyncClient, db_session):
+        """AC 8: DELETE /api/v1/companies/{id} -> 204, deleted_at gesetzt (soft-delete)."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        create_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Delete Me"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = create_resp.json()["id"]
+        resp = await client.delete(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
+        assert resp.status_code == 204
+        # Verify company is not in list (soft-deleted)
+        list_resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
+        names = [item["name"] for item in list_resp.json()["items"]]
+        assert "Delete Me" not in names
+
+    async def test_delete_company_cascade_returns_204(self, client: AsyncClient, db_session):
+        """AC 9: DELETE /api/v1/companies/{id}?cascade=true -> 204, company + links geloescht."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create company and contact
+        comp_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Cascade Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = comp_resp.json()["id"]
+        cont_resp = await client.post(
+            "/api/v1/contacts",
+            json={"first_name": "John", "last_name": "Doe", "company_ids": [company_id]},
+            headers=ORIGIN_HEADER,
+        )
+        contact_id = cont_resp.json()["id"]
+        # Verify link exists
+        detail_resp = await client.get(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
+        assert len(detail_resp.json()["contacts"]) == 1
+        # Cascade delete
+        resp = await client.delete(
+            f"/api/v1/companies/{company_id}?cascade=true",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 204
+
+
+@pytest.mark.asyncio
+class TestCompanyContactLink:
+    """ACs 10-11: N:M link, unlink."""
+
+    async def test_link_contact_to_company_returns_200(self, client: AsyncClient, db_session):
+        """AC 10: POST /api/v1/companies/{id}/contacts/{cid} -> 200, N:M link."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        comp_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Link Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = comp_resp.json()["id"]
+        cont_resp = await client.post(
+            "/api/v1/contacts",
+            json={"first_name": "Jane", "last_name": "Smith"},
+            headers=ORIGIN_HEADER,
+        )
+        contact_id = cont_resp.json()["id"]
+        resp = await client.post(
+            f"/api/v1/companies/{company_id}/contacts/{contact_id}",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["company_id"] == company_id
+        assert data["contact_id"] == contact_id
+
+    async def test_unlink_contact_from_company_returns_204(self, client: AsyncClient, db_session):
+        """AC 11: DELETE /api/v1/companies/{id}/contacts/{cid} -> 204, N:M unlink."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        comp_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Unlink Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = comp_resp.json()["id"]
+        cont_resp = await client.post(
+            "/api/v1/contacts",
+            json={"first_name": "Bob", "last_name": "Wilson"},
+            headers=ORIGIN_HEADER,
+        )
+        contact_id = cont_resp.json()["id"]
+        # Link first
+        await client.post(
+            f"/api/v1/companies/{company_id}/contacts/{contact_id}",
+            headers=ORIGIN_HEADER,
+        )
+        # Unlink
+        resp = await client.delete(
+            f"/api/v1/companies/{company_id}/contacts/{contact_id}",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 204
+        # Verify contact is no longer linked
+        detail = await client.get(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
+        assert len(detail.json()["contacts"]) == 0
+
+
+@pytest.mark.asyncio
+class TestCompanyExport:
+    """ACs 12-13: CSV export, XLSX export."""
+
+    async def test_export_csv_returns_200_text_csv(self, client: AsyncClient, db_session):
+        """AC 12: GET /api/v1/companies/export?format=csv -> 200 + text/csv."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        resp = await client.get(
+            "/api/v1/companies/export?format=csv",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        assert "text/csv" in resp.headers.get("content-type", "")
+        assert "name" in resp.text  # CSV header
+
+    async def test_export_xlsx_returns_200_openxml(self, client: AsyncClient, db_session):
+        """AC 13: GET /api/v1/companies/export?format=xlsx -> 200 + openxmlformats."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        resp = await client.get(
+            "/api/v1/companies/export?format=xlsx",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        ct = resp.headers.get("content-type", "")
+        assert "spreadsheet" in ct or "openxml" in ct
+        assert len(resp.content) > 0
+
+
+@pytest.mark.asyncio
+class TestCompanyEmails:
+    """AC 22: Emails endpoint (mail plugin inactive)."""
+
+    async def test_get_company_emails_returns_200_empty(self, client: AsyncClient, db_session):
+        """AC 22: GET /api/v1/companies/{id}/emails -> 200 (empty array, mail plugin inactive)."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        comp_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Email Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = comp_resp.json()["id"]
+        resp = await client.get(
+            f"/api/v1/companies/{company_id}/emails",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+
+@pytest.mark.asyncio
+class TestCompanyAuditAndSoftDelete:
+    """ACs 23-24: Audit log, soft-delete filter."""
+
+    async def test_audit_log_on_company_create(self, client: AsyncClient, db_session):
+        """AC 23: Audit log entry on every company/contact mutation."""
+        from sqlalchemy import select
+        from app.models.audit import AuditLog
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        await client.post(
+            "/api/v1/companies",
+            json={"name": "Audited Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        q = select(AuditLog).where(AuditLog.entity_type == "company", AuditLog.action == "create")
+        result = await db_session.execute(q)
+        entries = result.scalars().all()
+        assert len(entries) >= 1
+        assert any(e.changes and e.changes.get("name") == "Audited Corp" for e in entries)
+
+    async def test_soft_deleted_company_not_in_list(self, client: AsyncClient, db_session):
+        """AC 24: Soft-deleted company not in GET list (deleted_at IS NULL filter)."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create and then delete a company
+        create_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "SoftDelete Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        company_id = create_resp.json()["id"]
+        await client.delete(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
+        # List should not contain deleted company
+        list_resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
+        names = [item["name"] for item in list_resp.json()["items"]]
+        assert "SoftDelete Corp" not in names
+        assert "Company Alpha" in names  # Seed company still present

tests/test_contacts.py

@@ -0,0 +1,166 @@
+"""Contact tests — ACs 14-19: CRUD, N:M, soft-delete, GDPR hard-delete."""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+
+from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
+
+
+@pytest.mark.asyncio
+class TestContactList:
+    """AC 14: List contacts paginated."""
+
+    async def test_list_contacts_returns_200_paginated(self, client: AsyncClient, db_session):
+        """AC 14: GET /api/v1/contacts -> 200 + paginated."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "items" in data
+        assert "total" in data
+        assert "page" in data
+        assert "page_size" in data
+
+
+@pytest.mark.asyncio
+class TestContactCreate:
+    """AC 15: Create contact with company_ids array -> N:M links."""
+
+    async def test_create_contact_with_company_ids_returns_201(self, client: AsyncClient, db_session):
+        """AC 15: POST /api/v1/contacts mit company_ids array -> 201 + N:M links."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Get seeded company ID
+        list_resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
+        company_id = list_resp.json()["items"][0]["id"]
+        resp = await client.post(
+            "/api/v1/contacts",
+            json={
+                "first_name": "Alice",
+                "last_name": "Wonderland",
+                "email": "alice@example.com",
+                "company_ids": [company_id],
+            },
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 201
+        data = resp.json()
+        assert data["first_name"] == "Alice"
+        assert data["last_name"] == "Wonderland"
+        # Verify N:M link via company detail
+        comp_detail = await client.get(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
+        contacts = comp_detail.json()["contacts"]
+        assert any(c["first_name"] == "Alice" for c in contacts)
+
+
+@pytest.mark.asyncio
+class TestContactDetail:
+    """AC 16: Get contact detail with companies array."""
+
+    async def test_get_contact_returns_200_with_companies(self, client: AsyncClient, db_session):
+        """AC 16: GET /api/v1/contacts/{id} -> 200 + detail inkl. companies array."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        list_resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
+        company_id = list_resp.json()["items"][0]["id"]
+        create_resp = await client.post(
+            "/api/v1/contacts",
+            json={
+                "first_name": "Bob",
+                "last_name": "Builder",
+                "company_ids": [company_id],
+            },
+            headers=ORIGIN_HEADER,
+        )
+        contact_id = create_resp.json()["id"]
+        resp = await client.get(f"/api/v1/contacts/{contact_id}", headers=ORIGIN_HEADER)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["first_name"] == "Bob"
+        assert "companies" in data
+        assert isinstance(data["companies"], list)
+        assert len(data["companies"]) == 1
+        assert data["companies"][0]["name"] == "Company Alpha"
+
+
+@pytest.mark.asyncio
+class TestContactUpdate:
+    """AC 17: Update contact."""
+
+    async def test_update_contact_returns_200(self, client: AsyncClient, db_session):
+        """AC 17: PUT /api/v1/contacts/{id} -> 200."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        create_resp = await client.post(
+            "/api/v1/contacts",
+            json={"first_name": "Old", "last_name": "Name"},
+            headers=ORIGIN_HEADER,
+        )
+        contact_id = create_resp.json()["id"]
+        resp = await client.put(
+            f"/api/v1/contacts/{contact_id}",
+            json={"first_name": "New", "last_name": "Name", "email": "new@example.com"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["first_name"] == "New"
+        assert data["email"] == "new@example.com"
+
+
+@pytest.mark.asyncio
+class TestContactDelete:
+    """ACs 18-19: Soft-delete, GDPR hard-delete."""
+
+    async def test_delete_contact_soft_delete_returns_204(self, client: AsyncClient, db_session):
+        """AC 18: DELETE /api/v1/contacts/{id} -> 204, soft-delete."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        create_resp = await client.post(
+            "/api/v1/contacts",
+            json={"first_name": "Delete", "last_name": "Me"},
+            headers=ORIGIN_HEADER,
+        )
+        contact_id = create_resp.json()["id"]
+        resp = await client.delete(f"/api/v1/contacts/{contact_id}", headers=ORIGIN_HEADER)
+        assert resp.status_code == 204
+        # Verify contact not in list
+        list_resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
+        names = [f"{item['first_name']} {item['last_name']}" for item in list_resp.json()["items"]]
+        assert "Delete Me" not in names
+
+    async def test_delete_contact_gdpr_hard_delete_returns_204(self, client: AsyncClient, db_session):
+        """AC 19: DELETE /api/v1/contacts/{id}?gdpr=true -> 204, hard-delete + deletion_log."""
+        from sqlalchemy import select
+        from app.models.audit import DeletionLog
+        from app.models.contact import Contact
+        import uuid as uuid_mod
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        create_resp = await client.post(
+            "/api/v1/contacts",
+            json={"first_name": "GDPR", "last_name": "Delete"},
+            headers=ORIGIN_HEADER,
+        )
+        contact_id = create_resp.json()["id"]
+        resp = await client.delete(
+            f"/api/v1/contacts/{contact_id}?gdpr=true",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 204
+        # Verify physical delete — contact should not exist in DB
+        q = select(Contact).where(Contact.id == uuid_mod.UUID(contact_id))
+        result = await db_session.execute(q)
+        assert result.scalar_one_or_none() is None
+        # Verify deletion_log entry exists
+        dl_q = select(DeletionLog).where(
+            DeletionLog.entity_type == "contact",
+            DeletionLog.entity_id == uuid_mod.UUID(contact_id),
+        )
+        dl_result = await db_session.execute(dl_q)
+        dl_entries = dl_result.scalars().all()
+        assert len(dl_entries) >= 1
+        assert dl_entries[0].entity_snapshot["first_name"] == "GDPR"

tests/test_import_export.py

@@ -0,0 +1,146 @@
+"""Import/export tests — ACs 20-21: CSV import, dry-run preview."""
+
+from __future__ import annotations
+
+import io
+import pytest
+from httpx import AsyncClient
+
+from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
+
+
+CSV_COMPANIES = """name,industry,phone,email,website,description
+ImportCorp,IT,123456,import@example.com,https://import.example,Imported company
+TechImport,Finance,654321,tech@example.com,https://tech.example,Tech company
+"""
+
+CSV_COMPANIES_INVALID = """name,industry
+,IT
+ValidCorp,Finance
+"""
+
+CSV_CONTACTS = """first_name,last_name,email,phone,mobile,position,department
+Alice,Wonderland,alice@example.com,123,456,Manager,Sales
+Bob,Builder,bob@example.com,789,012,Developer,Tech
+"""
+
+
+@pytest.mark.asyncio
+class TestImportCompanies:
+    """AC 20: CSV import for companies."""
+
+    async def test_import_companies_csv_returns_200(self, client: AsyncClient, db_session):
+        """AC 20: POST /api/v1/import CSV + entity_type=companies -> 200 + result."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        files = {"file": ("companies.csv", CSV_COMPANIES.encode(), "text/csv")}
+        data = {"entity_type": "companies"}
+        resp = await client.post(
+            "/api/v1/import",
+            files=files,
+            data=data,
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        result = resp.json()
+        assert result["total"] == 2
+        assert result["valid"] == 2
+        assert result["invalid"] == 0
+        assert len(result["created"]) == 2
+        assert result["dry_run"] is False
+
+    async def test_import_companies_with_invalid_rows(self, client: AsyncClient, db_session):
+        """Import CSV with some invalid rows — should report errors but import valid ones."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        files = {"file": ("companies.csv", CSV_COMPANIES_INVALID.encode(), "text/csv")}
+        data = {"entity_type": "companies"}
+        resp = await client.post(
+            "/api/v1/import",
+            files=files,
+            data=data,
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        result = resp.json()
+        assert result["total"] == 2
+        assert result["valid"] == 1
+        assert result["invalid"] == 1
+        assert len(result["errors"]) == 1
+        assert len(result["created"]) == 1
+
+
+@pytest.mark.asyncio
+class TestImportPreview:
+    """AC 21: Dry-run preview (no DB changes)."""
+
+    async def test_import_preview_no_db_changes(self, client: AsyncClient, db_session):
+        """AC 21: POST /api/v1/import/preview CSV -> 200 + dry-run (no DB changes)."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        files = {"file": ("companies.csv", CSV_COMPANIES.encode(), "text/csv")}
+        data = {"entity_type": "companies"}
+        resp = await client.post(
+            "/api/v1/import/preview",
+            files=files,
+            data=data,
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        result = resp.json()
+        assert result["total"] == 2
+        assert result["valid"] == 2
+        assert result["dry_run"] is True
+        assert len(result["created"]) == 0  # No actual creations
+
+        # Verify no companies were actually created
+        list_resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
+        names = [item["name"] for item in list_resp.json()["items"]]
+        assert "ImportCorp" not in names
+        assert "TechImport" not in names
+
+    async def test_import_preview_contacts_no_db_changes(self, client: AsyncClient, db_session):
+        """Preview import for contacts — dry-run."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        files = {"file": ("contacts.csv", CSV_CONTACTS.encode(), "text/csv")}
+        data = {"entity_type": "contacts"}
+        resp = await client.post(
+            "/api/v1/import/preview",
+            files=files,
+            data=data,
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        result = resp.json()
+        assert result["total"] == 2
+        assert result["valid"] == 2
+        assert result["dry_run"] is True
+        assert len(result["created"]) == 0
+
+
+@pytest.mark.asyncio
+class TestImportContacts:
+    """Import contacts via CSV."""
+
+    async def test_import_contacts_csv_returns_200(self, client: AsyncClient, db_session):
+        """Import contacts via CSV."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        files = {"file": ("contacts.csv", CSV_CONTACTS.encode(), "text/csv")}
+        data = {"entity_type": "contacts"}
+        resp = await client.post(
+            "/api/v1/import",
+            files=files,
+            data=data,
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 200
+        result = resp.json()
+        assert result["total"] == 2
+        assert result["valid"] == 2
+        assert result["invalid"] == 0
+        assert len(result["created"]) == 2
+        # Verify contacts appear in list
+        list_resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
+        assert list_resp.json()["total"] >= 2

tests/test_tenant.py

@@ -209,7 +209,7 @@ class TestAuditLog:
             json={"name": "Audit Test Co"},
             headers=ORIGIN_HEADER,
         )
-        assert resp.status_code == 200
+        assert resp.status_code == 201
 
         # Check audit_log table
         from sqlalchemy import select as sel
@@ -222,7 +222,7 @@ class TestAuditLog:
         """AC 19: Audit log entry created on company.update."""
         seed = await seed_tenant_and_users(db_session)
         await login_client(client, "admin@tenanta.com")
-        resp = await client.patch(
+        resp = await client.put(
             f"/api/v1/companies/{seed['company_a'].id}",
             json={"name": "Updated Company Alpha"},
             headers=ORIGIN_HEADER,