"""INDL-34 Service Scheduling Phase 2 — API integration tests.

Runs against the FastAPI app + the indelis_test Postgres DB (see conftest.py).
S3 (boto3) and python-magic are monkeypatched so upload tests run without AWS
or libmagic. Redis rate-limiting fails open, so no Redis is required.

Spec: docs/34_INDL-34_Service_Scheduling_Phase2_Core.md
Functional: T-03..T-11.  Security: S-01..S-12 / SAC-01..SAC-12 (feasible subset).
"""
import uuid

import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy import func, select

from src.apps.auth.models.user import User
from src.apps.crew_members.models.crew_member import CrewMember
from src.apps.scheduling.models.scheduling_activity_log import SchedulingActivityLog
from src.apps.scheduling.models.scheduling_document import SchedulingDocument
from src.apps.scheduling.models.service_event import ServiceEvent
from src.apps.tenants.models.account import Account
from src.core.security import build_token_payload, create_access_token, hash_password

BASE = "/api/v1/scheduling"
LEGACY = "/api/v1/services"


# ── Fixtures ─────────────────────────────────────────────────────────────────

def _token(user, account) -> str:
    return create_access_token(build_token_payload(user, account))


async def _make_user(db_session, account, role, email):
    user = User(
        tenant_id=account.id,
        email=email,
        password_hash=hash_password("TestPassword123"),
        first_name=role.capitalize(),
        last_name="User",
        role=role,
        status="active",
    )
    db_session.add(user)
    await db_session.flush()
    return user


def _headers(user, account) -> dict:
    # The scheduling router derives tenant from request.state (set by
    # TenantMiddleware from the X-Tenant-ID header), not from the JWT — so the
    # tenant header is required in addition to the bearer token. This mirrors
    # production, where the frontend always sends the tenant header.
    return {
        "Authorization": f"Bearer {_token(user, account)}",
        "X-Tenant-ID": str(account.id),
    }


@pytest_asyncio.fixture
async def manager_headers(db_session, test_account):
    user = await _make_user(db_session, test_account, "manager", "mgr@testcemetery.com")
    return _headers(user, test_account)


@pytest_asyncio.fixture
async def staff_user(db_session, test_account):
    return await _make_user(db_session, test_account, "staff", "staff@testcemetery.com")


@pytest_asyncio.fixture
async def staff_headers(db_session, test_account, staff_user):
    return _headers(staff_user, test_account)


@pytest_asyncio.fixture
async def viewonly_headers(db_session, test_account):
    user = await _make_user(db_session, test_account, "view_only", "vo@testcemetery.com")
    return _headers(user, test_account)


@pytest_asyncio.fixture
async def other_tenant(db_session):
    account = Account(
        organization_name="Other Cemetery",
        subdomain="othercemetery",
        contact_email="admin@othercemetery.com",
        plan="starter",
        status="active",
    )
    db_session.add(account)
    await db_session.flush()
    return account


@pytest_asyncio.fixture
async def crew_member(db_session, test_account):
    cm = CrewMember(
        tenant_id=test_account.id,
        name="Alice Digger",
        email="alice@testcemetery.com",
        phone="613-555-0100",
        address="1 Grave Rd",
    )
    db_session.add(cm)
    await db_session.flush()
    return cm


def _payload(**overrides) -> dict:
    data = {
        "service_type": "Burial",
        "decedent_name": "John Doe",
        "date": "2026-08-01",
        "start_time": "10:00:00",
        "duration_minutes": 90,
        "plot_location": "Section A, Row 3, Plot 12",
        "section": "Garden of Peace",
        "officiant": "Rev. Smith",
        "family_contact_name": "Mary Doe",
        "family_contact_phone": "(613) 555-0199",
        "family_contact_email": "mary@example.com",
        "expected_attendees": 40,
        "event_note": "Graveside only.",
        "crew_member_ids": [],
        "run_sheet_items": [
            {"sort_order": 0, "time_label": "10:00", "details": "Arrival"},
            {"sort_order": 1, "time_label": "10:30", "details": "Committal"},
        ],
        "equipment_needed": "Lowering device",
        "notes_for_crew": "Bring tent.",
        "notify_family_confirmation": False,
        "notify_family_reminder_24h": False,
        "status": "draft",
    }
    data.update(overrides)
    return data


def _update_payload(**overrides) -> dict:
    """A valid ServiceUpdate body — same as create but WITHOUT `status`
    (ServiceUpdate has extra='forbid' and no status field)."""
    data = _payload(**overrides)
    data.pop("status", None)
    return data


async def _count_activity(db_session, service_id) -> int:
    return (
        await db_session.execute(
            select(func.count(SchedulingActivityLog.id)).where(
                SchedulingActivityLog.service_id == service_id
            )
        )
    ).scalar_one()


# ── T-03 / T-04 / SAC-03: create ─────────────────────────────────────────────

@pytest.mark.asyncio
async def test_create_awaiting_returns_201_with_crew_runsheet_and_scheduled_log(
    client: AsyncClient, staff_headers, db_session, crew_member
):
    body = _payload(status="awaiting_family_confirm", crew_member_ids=[str(crew_member.id)])
    res = await client.post(BASE, headers=staff_headers, json=body)
    assert res.status_code == 201, res.text
    data = res.json()["data"]
    assert data["status"] == "awaiting_family_confirm"
    assert len(data["crew"]) == 1 and data["crew"][0]["name"] == "Alice Digger"
    assert len(data["run_sheet_items"]) == 2
    # activity log has a `scheduled` entry (T-03)
    events = {e["event_type"] for e in data["activity_log"]}
    assert "scheduled" in events


@pytest.mark.asyncio
async def test_create_draft_logs_draft_saved(client: AsyncClient, staff_headers):
    res = await client.post(BASE, headers=staff_headers, json=_payload(status="draft"))
    assert res.status_code == 201, res.text
    data = res.json()["data"]
    assert data["status"] == "draft"
    events = {e["event_type"] for e in data["activity_log"]}
    assert "draft_saved" in events
    assert "confirmed" not in events


@pytest.mark.asyncio
async def test_create_with_confirmed_status_is_422(client: AsyncClient, staff_headers):
    # SAC-03 / S-03 / A04 — confirmed only reachable via PATCH /status.
    res = await client.post(BASE, headers=staff_headers, json=_payload(status="confirmed"))
    assert res.status_code == 422


@pytest.mark.asyncio
async def test_create_rejects_reminder_job_id_extra_field(client: AsyncClient, staff_headers):
    # S-12 / SAC-06
    res = await client.post(
        BASE, headers=staff_headers, json=_payload(reminder_job_id="fake")
    )
    assert res.status_code == 422


@pytest.mark.asyncio
async def test_create_response_hides_reminder_job_id(client: AsyncClient, staff_headers):
    # S-09 — reminder_job_id / metadata never surface in output.
    res = await client.post(BASE, headers=staff_headers, json=_payload())
    data = res.json()["data"]
    assert "reminder_job_id" not in data
    assert "tenant_id" not in data
    for entry in data["activity_log"]:
        assert "metadata" not in entry and "log_metadata" not in entry


# ── SAC-07: crew cross-tenant validation ─────────────────────────────────────

@pytest.mark.asyncio
async def test_create_with_foreign_crew_id_is_422_invalid_crew_member(
    client: AsyncClient, staff_headers, db_session, other_tenant
):
    foreign = CrewMember(
        tenant_id=other_tenant.id,
        name="Foreign Crew",
        email="foreign@other.com",
        phone="613-555-7777",
        address="Elsewhere",
    )
    db_session.add(foreign)
    await db_session.flush()
    res = await client.post(
        BASE, headers=staff_headers, json=_payload(crew_member_ids=[str(foreign.id)])
    )
    assert res.status_code == 422
    assert "invalid_crew_member" in res.text


# ── T-05 / T-06 / T-07: status FSM ───────────────────────────────────────────

async def _create_service(client, headers, **overrides) -> str:
    res = await client.post(BASE, headers=headers, json=_payload(**overrides))
    assert res.status_code == 201, res.text
    return res.json()["data"]["id"]


@pytest.mark.asyncio
async def test_patch_awaiting_to_confirmed_ok_logs_confirmed(
    client: AsyncClient, staff_headers
):
    sid = await _create_service(client, staff_headers, status="awaiting_family_confirm")
    res = await client.patch(
        f"{BASE}/{sid}/status", headers=staff_headers, json={"status": "confirmed"}
    )
    assert res.status_code == 200, res.text
    data = res.json()["data"]
    assert data["status"] == "confirmed"
    assert "confirmed" in {e["event_type"] for e in data["activity_log"]}


@pytest.mark.asyncio
async def test_patch_confirmed_to_awaiting_is_422(client: AsyncClient, staff_headers):
    sid = await _create_service(client, staff_headers, status="awaiting_family_confirm")
    await client.patch(f"{BASE}/{sid}/status", headers=staff_headers, json={"status": "confirmed"})
    # StatusUpdateRequest only permits 'confirmed'; awaiting is rejected at schema level.
    res = await client.patch(
        f"{BASE}/{sid}/status", headers=staff_headers,
        json={"status": "awaiting_family_confirm"},
    )
    assert res.status_code == 422


@pytest.mark.asyncio
async def test_patch_draft_to_confirmed_is_422_with_exact_message(
    client: AsyncClient, staff_headers
):
    # T-07 — a draft cannot jump to confirmed; exact FSM message.
    sid = await _create_service(client, staff_headers, status="draft")
    res = await client.patch(
        f"{BASE}/{sid}/status", headers=staff_headers, json={"status": "confirmed"}
    )
    assert res.status_code == 422
    assert (
        res.json()["message"]
        == "Status can only be updated from 'awaiting_family_confirm' to 'confirmed'."
    )


@pytest.mark.asyncio
async def test_patch_confirmed_is_idempotent_no_new_log(
    client: AsyncClient, staff_headers, db_session
):
    sid = await _create_service(client, staff_headers, status="awaiting_family_confirm")
    await client.patch(f"{BASE}/{sid}/status", headers=staff_headers, json={"status": "confirmed"})
    count_after_first = await _count_activity(db_session, uuid.UUID(sid))
    # Second confirm — must be 200, no new activity log row.
    res = await client.patch(
        f"{BASE}/{sid}/status", headers=staff_headers, json={"status": "confirmed"}
    )
    assert res.status_code == 200
    count_after_second = await _count_activity(db_session, uuid.UUID(sid))
    assert count_after_second == count_after_first


# ── S-03: PUT cannot change status ───────────────────────────────────────────

@pytest.mark.asyncio
async def test_put_with_status_field_is_422_and_status_unchanged(
    client: AsyncClient, staff_headers, db_session
):
    sid = await _create_service(client, staff_headers, status="awaiting_family_confirm")
    put_body = _payload()
    put_body["status"] = "confirmed"  # illegal extra field on ServiceUpdate
    res = await client.put(f"{BASE}/{sid}", headers=staff_headers, json=put_body)
    assert res.status_code == 422
    # persisted status must remain awaiting_family_confirm
    row = (
        await db_session.execute(select(ServiceEvent).where(ServiceEvent.id == uuid.UUID(sid)))
    ).scalar_one()
    await db_session.refresh(row)
    assert row.status == "awaiting_family_confirm"


# ── T-08: kanban / list view ─────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_kanban_returns_three_status_buckets(client: AsyncClient, staff_headers):
    await _create_service(client, staff_headers, status="draft")
    await _create_service(client, staff_headers, status="awaiting_family_confirm")
    res = await client.get(f"{BASE}?view=kanban", headers=staff_headers)
    assert res.status_code == 200
    data = res.json()["data"]
    assert set(data.keys()) == {"draft", "awaiting_family_confirm", "confirmed"}


@pytest.mark.asyncio
async def test_view_badvalue_is_422(client: AsyncClient, staff_headers):
    res = await client.get(f"{BASE}?view=badvalue", headers=staff_headers)
    assert res.status_code == 422


@pytest.mark.asyncio
async def test_list_pagination_and_page_size_cap(client: AsyncClient, staff_headers):
    res = await client.get(f"{BASE}?view=list&page=1&page_size=5", headers=staff_headers)
    assert res.status_code == 200
    body = res.json()
    # paginated envelope: items in `data`, total/page/page_size/pages at top level
    assert isinstance(body["data"], list)
    assert "total" in body and "page" in body and "page_size" in body and "pages" in body
    # page_size cap 100
    assert (await client.get(f"{BASE}?page_size=1000", headers=staff_headers)).status_code == 422


@pytest.mark.asyncio
async def test_list_status_filter(client: AsyncClient, staff_headers):
    await _create_service(client, staff_headers, status="draft")
    res = await client.get(f"{BASE}?status=draft", headers=staff_headers)
    assert res.status_code == 200
    for item in res.json()["data"]:
        assert item["status"] == "draft"


# ── T-10: full detail round-trip (incl. plot_location/section regression B1) ──

@pytest.mark.asyncio
async def test_detail_round_trips_plot_location_and_section(
    client: AsyncClient, staff_headers, crew_member
):
    sid = await _create_service(
        client, staff_headers, crew_member_ids=[str(crew_member.id)]
    )
    res = await client.get(f"{BASE}/{sid}", headers=staff_headers)
    assert res.status_code == 200
    data = res.json()["data"]
    assert data["plot_location"] == "Section A, Row 3, Plot 12"  # B1 regression
    assert data["section"] == "Garden of Peace"
    assert data["notes_for_crew"] == "Bring tent."
    assert [c["name"] for c in data["crew"]] == ["Alice Digger"]
    assert len(data["run_sheet_items"]) == 2
    assert "reminder_job_id" not in data


# ── update() logs changed fields incl. contact email/phone (B2 regression) ───

@pytest.mark.asyncio
async def test_update_changes_persist_and_log_updated(
    client: AsyncClient, staff_headers, db_session
):
    sid = await _create_service(client, staff_headers)
    updated = _update_payload(
        family_contact_email="changed@example.com",
        family_contact_phone="(613) 555-0000",
        decedent_name="Jane Roe",
    )
    res = await client.put(f"{BASE}/{sid}", headers=staff_headers, json=updated)
    assert res.status_code == 200, res.text
    data = res.json()["data"]
    assert data["family_contact_email"] == "changed@example.com"
    assert data["decedent_name"] == "Jane Roe"
    assert "updated" in {e["event_type"] for e in data["activity_log"]}


# ── DELETE service RBAC (S-08 / API5) ────────────────────────────────────────

@pytest.mark.asyncio
async def test_delete_service_as_staff_is_403(client: AsyncClient, staff_headers):
    sid = await _create_service(client, staff_headers)
    res = await client.delete(f"{BASE}/{sid}", headers=staff_headers)
    assert res.status_code == 403


@pytest.mark.asyncio
async def test_delete_service_as_manager_soft_deletes_and_logs(
    client: AsyncClient, staff_headers, manager_headers, db_session
):
    sid = await _create_service(client, staff_headers)
    res = await client.delete(f"{BASE}/{sid}", headers=manager_headers)
    assert res.status_code == 204
    # soft-deleted → subsequent GET 404
    assert (await client.get(f"{BASE}/{sid}", headers=manager_headers)).status_code == 404
    row = (
        await db_session.execute(select(ServiceEvent).where(ServiceEvent.id == uuid.UUID(sid)))
    ).scalar_one()
    await db_session.refresh(row)
    assert row.deleted_at is not None


# ── S-07 / API2: auth required on every endpoint ─────────────────────────────

@pytest.mark.asyncio
async def test_all_endpoints_require_auth(client: AsyncClient, test_account):
    # Send a valid tenant header but NO bearer token, so the 401 is specifically
    # an authentication failure (S-07 / API2) rather than a missing-tenant 401.
    th = {"X-Tenant-ID": str(test_account.id)}
    rid = str(uuid.uuid4())
    checks = [
        ("get", BASE, None),
        ("post", BASE, _payload()),
        ("get", f"{BASE}/{rid}", None),
        ("put", f"{BASE}/{rid}", _payload()),
        ("patch", f"{BASE}/{rid}/status", {"status": "confirmed"}),
        ("delete", f"{BASE}/{rid}", None),
        ("post", f"{BASE}/{rid}/documents", None),
        ("delete", f"{BASE}/{rid}/documents/{rid}", None),
        ("get", f"{BASE}/{rid}/activity", None),
    ]
    for method, url, json_body in checks:
        fn = getattr(client, method)
        res = await (
            fn(url, json=json_body, headers=th) if json_body is not None
            else fn(url, headers=th)
        )
        assert res.status_code == 401, f"{method.upper()} {url} -> {res.status_code}"


# ── SAC-01 / S-01: cross-tenant IDOR → 404 (not 403/200) ─────────────────────

@pytest_asyncio.fixture
async def foreign_service(db_session, other_tenant):
    """A service owned by Tenant B, created directly via ORM."""
    from datetime import date, time
    ev = ServiceEvent(
        tenant_id=other_tenant.id,
        service_type="Burial",
        scheduled_date=date(2026, 9, 1),
        scheduled_time=time(11, 0),
        status="awaiting_family_confirm",
        decedent_name="Foreign Decedent",
    )
    db_session.add(ev)
    await db_session.flush()
    return ev


@pytest.mark.asyncio
async def test_cross_tenant_get_is_404(client: AsyncClient, staff_headers, foreign_service):
    res = await client.get(f"{BASE}/{foreign_service.id}", headers=staff_headers)
    assert res.status_code == 404


@pytest.mark.asyncio
async def test_cross_tenant_put_is_404(client: AsyncClient, staff_headers, foreign_service):
    res = await client.put(
        f"{BASE}/{foreign_service.id}", headers=staff_headers, json=_update_payload()
    )
    assert res.status_code == 404


@pytest.mark.asyncio
async def test_cross_tenant_patch_status_is_404(
    client: AsyncClient, staff_headers, foreign_service
):
    res = await client.patch(
        f"{BASE}/{foreign_service.id}/status", headers=staff_headers,
        json={"status": "confirmed"},
    )
    assert res.status_code == 404


@pytest.mark.asyncio
async def test_cross_tenant_delete_is_404(
    client: AsyncClient, manager_headers, foreign_service
):
    res = await client.delete(f"{BASE}/{foreign_service.id}", headers=manager_headers)
    assert res.status_code == 404


@pytest.mark.asyncio
async def test_cross_tenant_activity_is_404(
    client: AsyncClient, staff_headers, foreign_service
):
    res = await client.get(f"{BASE}/{foreign_service.id}/activity", headers=staff_headers)
    assert res.status_code == 404


# ── SAC-02 / S-02: document IDOR ─────────────────────────────────────────────

@pytest.mark.asyncio
async def test_cross_tenant_document_delete_is_404_and_not_deleted(
    client: AsyncClient, staff_headers, db_session, other_tenant, foreign_service
):
    # A document belonging to Tenant B's service.
    foreign_user = await _make_user(db_session, other_tenant, "staff", "fs@other.com")
    doc = SchedulingDocument(
        tenant_id=other_tenant.id,
        service_id=foreign_service.id,
        file_name="cert.pdf",
        s3_key=f"scheduling/{other_tenant.id}/{foreign_service.id}/{uuid.uuid4()}.pdf",
        file_size=1024,
        mime_type="application/pdf",
        uploaded_by=foreign_user.id,
    )
    db_session.add(doc)
    await db_session.flush()

    # Tenant A tries to delete via its own (nonexistent) service path -> 404.
    res = await client.delete(
        f"{BASE}/{foreign_service.id}/documents/{doc.id}", headers=staff_headers
    )
    assert res.status_code == 404
    await db_session.refresh(doc)
    assert doc.deleted_at is None  # not soft-deleted


# ── T-09 + upload security (S-04/S-05/S-06) with S3 + magic monkeypatched ────

@pytest_asyncio.fixture
def mock_s3_and_magic(monkeypatch):
    """Patch boto3 client + python-magic in the scheduling service module.

    Returns a dict recording put_object calls so tests can assert S3 side effects.
    """
    import src.apps.scheduling.services.scheduling_service as svc

    calls = {"put_object": [], "presign": []}

    class _FakeS3:
        def put_object(self, **kwargs):
            calls["put_object"].append(kwargs)
            return {}

        def generate_presigned_url(self, op, Params=None, ExpiresIn=None):
            calls["presign"].append({"Params": Params, "ExpiresIn": ExpiresIn})
            key = (Params or {}).get("Key", "")
            return f"https://s3.example.test/{key}?X-Amz-Expires={ExpiresIn}"

    monkeypatch.setattr(svc, "_make_s3_client", lambda: _FakeS3())

    # Magic-byte sniffer: PDF header -> pdf, HTML markers -> text/html.
    def _fake_sniff(content: bytes):
        if content.startswith(b"%PDF"):
            return "application/pdf"
        head = content[:64].lstrip().lower()
        if head.startswith(b"<!doctype html") or head.startswith(b"<html"):
            return "text/html"
        return "application/octet-stream"

    monkeypatch.setattr(svc, "_sniff_mime", _fake_sniff)
    return calls


@pytest.mark.asyncio
async def test_upload_valid_pdf_stores_and_returns_presigned_url(
    client: AsyncClient, staff_headers, mock_s3_and_magic
):
    # T-09 / SAC-05 / S-10
    sid = await _create_service(client, staff_headers)
    files = {"file": ("death_certificate.pdf", b"%PDF-1.4 fake body", "application/pdf")}
    res = await client.post(f"{BASE}/{sid}/documents", headers=staff_headers, files=files)
    assert res.status_code == 201, res.text
    data = res.json()["data"]
    assert data["file_name"] == "death_certificate.pdf"
    assert data["mime_type"] == "application/pdf"
    assert data["download_url"].startswith("https://s3.example.test/")
    # S3 key server-built: scheduling/<uuid>/<uuid>/<uuid>.pdf
    put = mock_s3_and_magic["put_object"][0]
    import re
    assert re.match(
        r"^scheduling/[0-9a-f-]{36}/[0-9a-f-]{36}/[0-9a-f-]{36}\.pdf$", put["Key"]
    )
    # SAC-08 / S-10: presign TTL <= 900s
    assert mock_s3_and_magic["presign"][-1]["ExpiresIn"] <= 900


@pytest.mark.asyncio
async def test_upload_traversal_filename_is_422_no_s3_put(
    client: AsyncClient, staff_headers, mock_s3_and_magic
):
    # S-04 / SAC-05
    sid = await _create_service(client, staff_headers)
    files = {"file": ("../../etc/passwd", b"%PDF-1.4 body", "application/pdf")}
    res = await client.post(f"{BASE}/{sid}/documents", headers=staff_headers, files=files)
    assert res.status_code == 422
    assert mock_s3_and_magic["put_object"] == []


@pytest.mark.asyncio
async def test_upload_mime_spoof_html_as_pdf_is_415_no_s3_put(
    client: AsyncClient, staff_headers, mock_s3_and_magic
):
    # S-05 / SAC-04 — declared pdf, magic bytes say HTML.
    sid = await _create_service(client, staff_headers)
    files = {"file": ("evil.pdf", b"<!DOCTYPE html><script>alert(1)</script>", "application/pdf")}
    res = await client.post(f"{BASE}/{sid}/documents", headers=staff_headers, files=files)
    assert res.status_code == 415
    assert mock_s3_and_magic["put_object"] == []


@pytest.mark.asyncio
async def test_upload_oversized_file_is_413_no_s3_put(
    client: AsyncClient, staff_headers, mock_s3_and_magic
):
    # S-06 — 25 MB + 1 byte, valid PDF header; must reject before any S3 put.
    sid = await _create_service(client, staff_headers)
    big = b"%PDF-1.4" + b"0" * (25 * 1024 * 1024 + 1)
    files = {"file": ("big.pdf", big, "application/pdf")}
    res = await client.post(f"{BASE}/{sid}/documents", headers=staff_headers, files=files)
    assert res.status_code == 413
    assert mock_s3_and_magic["put_object"] == []


@pytest.mark.asyncio
async def test_staff_cannot_delete_another_users_document_403(
    client: AsyncClient, staff_headers, manager_headers, db_session, mock_s3_and_magic
):
    # API5 / S-08 — staff may only delete own uploads; a manager-owned doc -> 403.
    sid = await _create_service(client, staff_headers)
    # Manager uploads a document.
    files = {"file": ("permit.pdf", b"%PDF-1.4 body", "application/pdf")}
    up = await client.post(f"{BASE}/{sid}/documents", headers=manager_headers, files=files)
    assert up.status_code == 201, up.text
    doc_id = up.json()["data"]["id"]
    # Staff (not the uploader, below manager) attempts delete -> 403.
    res = await client.delete(f"{BASE}/{sid}/documents/{doc_id}", headers=staff_headers)
    assert res.status_code == 403


@pytest.mark.asyncio
async def test_uploader_can_delete_own_document(
    client: AsyncClient, staff_headers, db_session, mock_s3_and_magic
):
    sid = await _create_service(client, staff_headers)
    files = {"file": ("mine.pdf", b"%PDF-1.4 body", "application/pdf")}
    up = await client.post(f"{BASE}/{sid}/documents", headers=staff_headers, files=files)
    doc_id = up.json()["data"]["id"]
    res = await client.delete(f"{BASE}/{sid}/documents/{doc_id}", headers=staff_headers)
    assert res.status_code == 204


# ── S-09: activity endpoint allow-list ───────────────────────────────────────

@pytest.mark.asyncio
async def test_activity_endpoint_omits_metadata(client: AsyncClient, staff_headers):
    sid = await _create_service(client, staff_headers)
    res = await client.get(f"{BASE}/{sid}/activity", headers=staff_headers)
    assert res.status_code == 200
    entries = res.json()["data"]
    assert entries, "expected at least the create activity entry"
    allowed = {
        "id", "event_type", "from_status", "to_status",
        "description", "actor_name", "created_at",
    }
    for e in entries:
        assert set(e.keys()) == allowed
        assert "metadata" not in e


# ── HIGH-1 regression: legacy /services PUT status guard ─────────────────────

@pytest_asyncio.fixture
async def legacy_service(db_session, test_account, staff_user):
    from datetime import date, time
    ev = ServiceEvent(
        tenant_id=test_account.id,
        created_by=staff_user.id,
        service_type="Burial",
        scheduled_date=date(2026, 8, 15),
        scheduled_time=time(9, 0),
        status="awaiting_family_confirm",
    )
    db_session.add(ev)
    await db_session.flush()
    return ev


def _legacy_put_body(status: str) -> dict:
    return {
        "service_type": "Burial",
        "date": "2026-08-15",
        "start_time": "09:00:00",
        "status": status,
    }


@pytest.mark.asyncio
async def test_legacy_put_confirmed_is_422(
    client: AsyncClient, staff_headers, legacy_service
):
    res = await client.put(
        f"{LEGACY}/{legacy_service.id}", headers=staff_headers,
        json=_legacy_put_body("confirmed"),
    )
    assert res.status_code == 422


@pytest.mark.asyncio
async def test_legacy_put_draft_ok(client: AsyncClient, staff_headers, legacy_service):
    res = await client.put(
        f"{LEGACY}/{legacy_service.id}", headers=staff_headers,
        json=_legacy_put_body("draft"),
    )
    assert res.status_code == 200, res.text


@pytest.mark.asyncio
async def test_legacy_put_awaiting_ok(client: AsyncClient, staff_headers, legacy_service):
    res = await client.put(
        f"{LEGACY}/{legacy_service.id}", headers=staff_headers,
        json=_legacy_put_body("awaiting_family_confirm"),
    )
    assert res.status_code == 200, res.text