FILE: reportkit/__init__.py
```python
"""reportkit — multi-tenant report service."""
```

FILE: reportkit/schema.sql
```sql
-- reportkit schema
PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;

CREATE TABLE IF NOT EXISTS tenants (
    id          TEXT PRIMARY KEY,
    name        TEXT NOT NULL,
    suspended   INTEGER NOT NULL DEFAULT 0,
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE TABLE IF NOT EXISTS users (
    id          TEXT PRIMARY KEY,
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    email       TEXT NOT NULL,
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    UNIQUE (tenant_id, email)
);

CREATE TABLE IF NOT EXISTS roles (
    user_id     TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    role        TEXT NOT NULL CHECK (role IN ('viewer', 'editor', 'admin')),
    PRIMARY KEY (user_id, tenant_id)
);

-- Covering index for the hot permissions query (list_reports resolves perms
-- on every request: join users → roles → derive permissions).
CREATE INDEX IF NOT EXISTS idx_roles_user_tenant ON roles(user_id, tenant_id);

CREATE TABLE IF NOT EXISTS sessions (
    token       TEXT PRIMARY KEY,
    user_id     TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    expires_at  TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_sessions_token ON sessions(token);

CREATE TABLE IF NOT EXISTS reports (
    id          TEXT PRIMARY KEY,
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    owner_id    TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    title       TEXT NOT NULL,
    body        TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    updated_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_reports_owner    ON reports(owner_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_reports_tenant   ON reports(tenant_id);
CREATE INDEX IF NOT EXISTS idx_reports_title    ON reports(owner_id, title);

CREATE TABLE IF NOT EXISTS attachments (
    id          TEXT PRIMARY KEY,
    report_id   TEXT NOT NULL REFERENCES reports(id) ON DELETE CASCADE,
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    filename    TEXT NOT NULL,
    mime_type   TEXT NOT NULL DEFAULT 'application/octet-stream',
    size_bytes  INTEGER NOT NULL,
    data        BLOB NOT NULL,
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_attachments_report ON attachments(report_id);

CREATE TABLE IF NOT EXISTS ledger (
    id              TEXT PRIMARY KEY,
    tenant_id       TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    invoice_id      TEXT NOT NULL,
    amount_cents    INTEGER NOT NULL,
    currency        TEXT NOT NULL DEFAULT 'usd',
    event_type      TEXT NOT NULL,
    idempotency_key TEXT NOT NULL UNIQUE,
    created_at      TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
```

FILE: reportkit/db.py
```python
"""Database helpers: connection factory and schema bootstrap."""

import sqlite3
import pathlib

_SCHEMA = pathlib.Path(__file__).with_name("schema.sql").read_text()

_ALLOWED_SORT_COLUMNS = frozenset({"created_at", "title"})


def connect(path: str = ":memory:") -> sqlite3.Connection:
    """Return a connection with row_factory and foreign-key enforcement."""
    conn = sqlite3.connect(path, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode = WAL;")
    conn.execute("PRAGMA foreign_keys = ON;")
    return conn


def apply_schema(conn: sqlite3.Connection) -> None:
    """Idempotently apply the DDL schema."""
    conn.executescript(_SCHEMA)


def validate_sort_column(column: str) -> str:
    """
    Return *column* if it is an allowed sort key, otherwise raise ValueError.

    This is the single place where user-supplied sort values are sanitised
    before being interpolated into SQL.  Parameterised queries cannot be used
    for column names, so an allowlist is the correct defence.
    """
    if column not in _ALLOWED_SORT_COLUMNS:
        raise ValueError(
            f"Invalid sort column {column!r}. "
            f"Allowed: {sorted(_ALLOWED_SORT_COLUMNS)}"
        )
    return column
```

FILE: reportkit/auth.py
```python
"""Session resolution and permission helpers."""

import sqlite3
from typing import Optional

# ---------------------------------------------------------------------------
# Permission model
# ---------------------------------------------------------------------------

_ROLE_PERMISSIONS: dict[str, frozenset[str]] = {
    "viewer": frozenset({"report:read"}),
    "editor": frozenset({"report:read", "report:write"}),
    "admin":  frozenset({"report:read", "report:write", "report:delete", "user:manage"}),
}


def permissions_for_role(role: str) -> frozenset[str]:
    try:
        return _ROLE_PERMISSIONS[role]
    except KeyError:
        return frozenset()


# ---------------------------------------------------------------------------
# Session / caller resolution
# ---------------------------------------------------------------------------

class AuthenticationError(Exception):
    """Raised when a bearer token cannot be resolved to a live session."""


class AuthorizationError(Exception):
    """Raised when the resolved caller lacks a required permission."""


def _extract_token(request: dict) -> str:
    """Pull the bearer token out of the Authorization header."""
    auth_header: str = (request.get("headers") or {}).get("Authorization", "")
    if not auth_header.lower().startswith("bearer "):
        raise AuthenticationError("Missing or malformed Authorization header.")
    return auth_header[7:].strip()


def resolve_caller(conn: sqlite3.Connection, request: dict) -> dict:
    """
    Resolve the bearer token to a caller dict:

        {user_id, tenant_id, email, role, permissions}

    The query joins sessions → users → roles in one round-trip.  An index on
    (user_id, tenant_id) in the roles table keeps this fast even as the user
    table grows (see schema.sql idx_roles_user_tenant).
    """
    token = _extract_token(request)

    row = conn.execute(
        """
        SELECT
            u.id        AS user_id,
            u.tenant_id AS tenant_id,
            u.email     AS email,
            r.role      AS role,
            t.suspended AS suspended
        FROM sessions s
        JOIN users   u ON u.id        = s.user_id
        JOIN tenants t ON t.id        = u.tenant_id
        LEFT JOIN roles r
               ON r.user_id   = u.id
              AND r.tenant_id = u.tenant_id
        WHERE s.token      = ?
          AND s.expires_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
        """,
        (token,),
    ).fetchone()

    if row is None:
        raise AuthenticationError("Invalid or expired session token.")

    if row["suspended"]:
        raise AuthorizationError("Tenant account is suspended.")

    role = row["role"] or "viewer"
    return {
        "user_id":     row["user_id"],
        "tenant_id":   row["tenant_id"],
        "email":       row["email"],
        "role":        role,
        "permissions": permissions_for_role(role),
    }


def require_permission(caller: dict, permission: str) -> None:
    """Raise AuthorizationError if *caller* does not hold *permission*."""
    if permission not in caller["permissions"]:
        raise AuthorizationError(
            f"Permission denied: {permission!r} required, "
            f"caller holds {sorted(caller['permissions'])}."
        )
```

FILE: reportkit/events.py
```python
"""
Lightweight synchronous event registry.

Handlers register against a string event name.  The dispatcher calls every
registered handler in registration order; exceptions in one handler do not
prevent others from running (they are collected and re-raised as a group).

The tenant.quota_exceeded handler is registered here but its implementation
is a no-op stub: the batch-import feed that emits the event was retired last
quarter.  The registration is kept so the platform team can wire in a real
implementation when the feed returns without touching the registry.
"""

from __future__ import annotations

import logging
from collections import defaultdict
from typing import Callable

logger = logging.getLogger(__name__)

# Type alias
Handler = Callable[[dict], None]

_registry: dict[str, list[Handler]] = defaultdict(list)


def register(event_name: str, handler: Handler) -> None:
    """Register *handler* to be called when *event_name* is dispatched."""
    _registry[event_name].append(handler)


def dispatch(event_name: str, payload: dict) -> None:
    """
    Call every handler registered for *event_name*.

    All handlers are invoked even if an earlier one raises; errors are logged
    and re-raised together after all handlers have run.
    """
    handlers = _registry.get(event_name, [])
    errors: list[Exception] = []
    for handler in handlers:
        try:
            handler(payload)
        except Exception as exc:  # noqa: BLE001
            logger.exception(
                "Event handler %r raised for event %r", handler, event_name
            )
            errors.append(exc)
    if errors:
        raise ExceptionGroup(f"Errors dispatching {event_name!r}", errors)


# ---------------------------------------------------------------------------
# Built-in handlers
# ---------------------------------------------------------------------------

def _on_report_created(payload: dict) -> None:
    logger.info(
        "report.created tenant=%s owner=%s report=%s",
        payload.get("tenant_id"),
        payload.get("owner_id"),
        payload.get("report_id"),
    )


def _on_report_deleted(payload: dict) -> None:
    logger.info(
        "report.deleted tenant=%s actor=%s report=%s",
        payload.get("tenant_id"),
        payload.get("actor_id"),
        payload.get("report_id"),
    )


def _on_tenant_quota_exceeded(payload: dict) -> None:
    """
    Stub handler for tenant.quota_exceeded.

    The batch-import feed that emits this event was retired; this handler is
    intentionally inert until the feed is reinstated.  When it returns, replace
    this body with a call to the tenant-suspension logic.
    """
    logger.debug(
        "tenant.quota_exceeded received (handler is a stub): tenant=%s",
        payload.get("tenant_id"),
    )


register("report.created",        _on_report_created)
register("report.deleted",        _on_report_deleted)
register("tenant.quota_exceeded", _on_tenant_quota_exceeded)
```

FILE: reportkit/content_scanner.py
```python
"""
Attachment content scanner.

Checks uploaded bytes for disallowed content markers before the attachment is
persisted.  The scanner is intentionally simple: it looks for a configurable
set of byte-string signatures.  In production this would delegate to a proper
AV/content-inspection service; the interface is kept stable so that swap-in is
straightforward.
"""

from __future__ import annotations

# Byte signatures that are unconditionally rejected.
# Extend this tuple to add new markers without changing call-sites.
_DISALLOWED_SIGNATURES: tuple[bytes, ...] = (
    b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE",   # EICAR test string
    b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR",  # EICAR header variant
)

MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024  # 25 MB


class ContentViolation(Exception):
    """Raised when a disallowed content marker is found in uploaded data."""


def scan(data: bytes, filename: str = "") -> None:
    """
    Scan *data* for disallowed content markers.

    Raises ContentViolation if any marker is found.
    Raises ValueError if the data exceeds the size limit.
    """
    if len(data) > MAX_ATTACHMENT_BYTES:
        raise ValueError(
            f"Attachment {filename!r} exceeds the 25 MB limit "
            f"({len(data):,} bytes)."
        )
    for sig in _DISALLOWED_SIGNATURES:
        if sig in data:
            raise ContentViolation(
                f"Disallowed content marker detected in {filename!r}."
            )
```

FILE: reportkit/billing.py
```python
"""
Billing webhook handler.

Verifies the HMAC-SHA256 signature supplied by the billing provider, then
applies the payment to the ledger.  The shared secret is read from the
environment variable BILLING_WEBHOOK_SECRET.

Idempotency: the provider retries deliveries it considers failed.  We use the
event's idempotency_key (derived from the provider's event ID) as a UNIQUE
constraint in the ledger table; a duplicate delivery returns 200 without
re-inserting.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import sqlite3
import uuid
import logging

logger = logging.getLogger(__name__)

_SECRET_ENV_VAR = "BILLING_WEBHOOK_SECRET"


class WebhookAuthError(Exception):
    """Raised when the HMAC signature on a webhook request is invalid."""


def _get_secret() -> bytes:
    secret = os.environ.get(_SECRET_ENV_VAR, "")
    if not secret:
        raise EnvironmentError(
            f"Environment variable {_SECRET_ENV_VAR!r} is not set."
        )
    return secret.encode()


def verify_signature(raw_body: bytes, signature_header: str) -> None:
    """
    Verify the HMAC-SHA256 signature sent by the billing provider.

    The provider sends:  X-Billing-Signature: sha256=<hex-digest>
    """
    if not signature_header.startswith("sha256="):
        raise WebhookAuthError("Signature header must start with 'sha256='.")
    expected_hex = signature_header[7:]
    secret = _get_secret()
    computed = hmac.new(secret, raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(computed, expected_hex):
        raise WebhookAuthError("Webhook signature verification failed.")


def handle_invoice_paid(
    conn: sqlite3.Connection,
    raw_body: bytes,
    signature_header: str,
) -> dict:
    """
    Process an invoice.paid webhook delivery.

    Returns a response dict suitable for the HTTP layer.
    """
    verify_signature(raw_body, signature_header)

    try:
        event = json.loads(raw_body)
    except json.JSONDecodeError as exc:
        return {"status": 400, "body": {"error": f"Invalid JSON: {exc}"}}

    event_type = event.get("type")
    if event_type != "invoice.paid":
        # Acknowledge unknown event types without processing them.
        return {"status": 200, "body": {"received": True, "processed": False}}

    data       = event.get("data", {})
    tenant_id  = data.get("tenant_id")
    invoice_id = data.get("invoice_id")
    amount     = data.get("amount_cents")
    currency   = data.get("currency", "usd")
    idem_key   = event.get("id")  # provider's unique event ID

    if not all([tenant_id, invoice_id, amount is not None, idem_key]):
        return {"status": 400, "body": {"error": "Missing required fields."}}

    try:
        conn.execute(
            """
            INSERT INTO ledger
                (id, tenant_id, invoice_id, amount_cents, currency,
                 event_type, idempotency_key)
            VALUES (?, ?, ?, ?, ?, 'invoice.paid', ?)
            """,
            (str(uuid.uuid4()), tenant_id, invoice_id, amount, currency, idem_key),
        )
        conn.commit()
        logger.info("Ledger entry created for invoice %s tenant %s", invoice_id, tenant_id)
    except sqlite3.IntegrityError:
        # Duplicate delivery — idempotency_key already present.
        logger.info(
            "Duplicate invoice.paid delivery ignored (idem_key=%s)", idem_key
        )

    return {"status": 200, "body": {"received": True, "processed": True}}
```

FILE: reportkit/ops.py
```python
"""
Internal ops-console operations.

These functions are called by the ops console, which runs inside the private
network.  They are NOT exposed over the public HTTP surface.

purge_tenant  — delete every report belonging to a tenant.
set_role      — change a user's role within a tenant.

Both functions accept an actor_id for audit purposes; in a fuller
implementation this would write to an audit log table.
"""

from __future__ import annotations

import logging
import sqlite3

from reportkit.auth import permissions_for_role

logger = logging.getLogger(__name__)

_VALID_ROLES = frozenset({"viewer", "editor", "admin"})


def purge_tenant(
    conn: sqlite3.Connection,
    actor_id: str,
    tenant_id: str,
) -> int:
    """
    Delete every report belonging to *tenant_id*.

    Returns the number of reports deleted.
    Cascading deletes in the schema remove associated attachments automatically.
    """
    cursor = conn.execute(
        "DELETE FROM reports WHERE tenant_id = ?",
        (tenant_id,),
    )
    conn.commit()
    count = cursor.rowcount
    logger.info(
        "purge_tenant: actor=%s tenant=%s deleted_reports=%d",
        actor_id, tenant_id, count,
    )
    return count


def set_role(
    conn: sqlite3.Connection,
    actor_id: str,
    user_id: str,
    tenant_id: str,
    new_role: str,
) -> None:
    """
    Set *user_id*'s role within *tenant_id* to *new_role*.

    Raises ValueError for an unrecognised role name.
    """
    if new_role not in _VALID_ROLES:
        raise ValueError(
            f"Unknown role {new_role!r}. Valid roles: {sorted(_VALID_ROLES)}"
        )

    conn.execute(
        """
        INSERT INTO roles (user_id, tenant_id, role)
        VALUES (?, ?, ?)
        ON CONFLICT (user_id, tenant_id) DO UPDATE SET role = excluded.role
        """,
        (user_id, tenant_id, new_role),
    )
    conn.commit()
    logger.info(
        "set_role: actor=%s user=%s tenant=%s role=%s",
        actor_id, user_id, tenant_id, new_role,
    )
```

FILE: reportkit/handlers.py
```python
"""
HTTP surface for reportkit.

All public methods accept a *request* dict modelled after a minimal WSGI/ASGI
envelope:

    {
        "headers": {"Authorization": "Bearer <token>", ...},
        "body":    <bytes | dict>,          # dict for JSON, bytes for binary
        "query":   {"sort": "title", ...},  # URL query parameters
    }

Methods return a response dict:

    {"status": <int>, "body": <JSON-serialisable dict>}

No web framework is used; the class can be driven directly in tests or wrapped
by any WSGI/ASGI adapter.
"""

from __future__ import annotations

import sqlite3
import uuid
import logging

from reportkit.auth import (
    AuthenticationError,
    AuthorizationError,
    resolve_caller,
    require_permission,
)
from reportkit.content_scanner import scan, ContentViolation
from reportkit.db import validate_sort_column
from reportkit import events

logger = logging.getLogger(__name__)

_DEFAULT_SORT = "created_at"


def _ok(body: dict, status: int = 200) -> dict:
    return {"status": status, "body": body}


def _err(message: str, status: int) -> dict:
    return {"status": status, "body": {"error": message}}


class ReportHandlers:
    """
    HTTP surface for the report service.

    Instantiate once per process (or per test) with a live database connection.
    """

    def __init__(self, conn: sqlite3.Connection) -> None:
        self._conn = conn

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _authenticate(self, request: dict) -> dict:
        """Resolve caller or raise AuthenticationError / AuthorizationError."""
        return resolve_caller(self._conn, request)

    def _get_report_row(self, report_id: str, caller: dict):
        return self._conn.execute(
            "SELECT * FROM reports WHERE id = ? AND tenant_id = ?",
            (report_id, caller["tenant_id"]),
        ).fetchone()

    # ------------------------------------------------------------------
    # Public HTTP methods
    # ------------------------------------------------------------------

    def create_report(self, request: dict) -> dict:
        """POST /reports — create a report for the authenticated caller."""
        try:
            caller = self._authenticate(request)
        except (AuthenticationError, AuthorizationError) as exc:
            return _err(str(exc), 401)

        try:
            require_permission(caller, "report:write")
        except AuthorizationError as exc:
            return _err(str(exc), 403)

        body = request.get("body") or {}
        title = (body.get("title") or "").strip()
        content = (body.get("body") or "").strip()

        if not title:
            return _err("'title' is required.", 400)

        report_id = str(uuid.uuid4())
        try:
            self._conn.execute(
                """
                INSERT INTO reports (id, tenant_id, owner_id, title, body)
                VALUES (?, ?, ?, ?, ?)
                """,
                (report_id, caller["tenant_id"], caller["user_id"], title, content),
            )
            self._conn.commit()
        except sqlite3.Error as exc:
            logger.exception("create_report db error")
            return _err(f"Database error: {exc}", 500)

        events.dispatch(
            "report.created",
            {
                "report_id": report_id,
                "tenant_id": caller["tenant_id"],
                "owner_id":  caller["user_id"],
            },
        )

        row = self._get_report_row(report_id, caller)
        return _ok(dict(row), 201)

    def get_report(self, request: dict, report_id: str) -> dict:
        """GET /reports/<id> — return a single report."""
        try:
            caller = self._authenticate(request)
        except (AuthenticationError, AuthorizationError) as exc:
            return _err(str(exc), 401)

        try:
            require_permission(caller, "report:read")
        except AuthorizationError as exc:
            return _err(str(exc), 403)

        row = self._get_report_row(report_id, caller)
        if row is None:
            return _err("Report not found.", 404)

        return _ok(dict(row))

    def list_reports(self, request: dict) -> dict:
        """GET /reports — list the caller's own reports, newest first."""
        try:
            caller = self._authenticate(request)
        except (AuthenticationError, AuthorizationError) as exc:
            return _err(str(exc), 401)

        try:
            require_permission(caller, "report:read")
        except AuthorizationError as exc:
            return _err(str(exc), 403)

        rows = self._conn.execute(
            """
            SELECT * FROM reports
            WHERE owner_id  = ?
              AND tenant_id = ?
            ORDER BY created_at DESC
            """,
            (caller["user_id"], caller["tenant_id"]),
        ).fetchall()

        return _ok({"reports": [dict(r) for r in rows]})

    def search_reports(self, request: dict) -> dict:
        """
        GET /reports/search?q=<term>&sort=<created_at|title>

        Title search across the caller's own reports.  The sort column is
        validated against an allowlist before interpolation into SQL.
        """
        try:
            caller = self._authenticate(request)
        except (AuthenticationError, AuthorizationError) as exc:
            return _err(str(exc), 401)

        try:
            require_permission(caller, "report:read")
        except AuthorizationError as exc:
            return _err(str(exc), 403)

        query_params = request.get("query") or {}
        search_term  = (query_params.get("q") or "").strip()
        sort_raw     = (query_params.get("sort") or _DEFAULT_SORT).strip()

        try:
            sort_col = validate_sort_column(sort_raw)
        except ValueError as exc:
            return _err(str(exc), 400)

        like_pattern = f"%{search_term}%"

        # sort_col is safe: validated against a frozenset allowlist.
        sql = f"""
            SELECT * FROM reports
            WHERE owner_id  = ?
              AND tenant_id = ?
              AND title LIKE ?
            ORDER BY {sort_col} ASC
        """
        rows = self._conn.execute(
            sql,
            (caller["user_id"], caller["tenant_id"], like_pattern),
        ).fetchall()

        return _ok({"reports": [dict(r) for r in rows]})

    def delete_report(self, request: dict, report_id: str) -> dict:
        """DELETE /reports/<id> — delete a report."""
        try:
            caller = self._authenticate(request)
        except (AuthenticationError, AuthorizationError) as exc:
            return _err(str(exc), 401)

        try:
            require_permission(caller, "report:delete")
        except AuthorizationError as exc:
            return _err(str(exc), 403)

        row = self._get_report_row(report_id, caller)
        if row is None:
            return _err("Report not found.", 404)

        try:
            self._conn.execute(
                "DELETE FROM reports WHERE id = ? AND tenant_id = ?",
                (report_id, caller["tenant_id"]),
            )
            self._conn.commit()
        except sqlite3.Error as exc:
            logger.exception("delete_report db error")
            return _err(f"Database error: {exc}", 500)

        events.dispatch(
            "report.deleted",
            {
                "report_id": report_id,
                "tenant_id": caller["tenant_id"],
                "actor_id":  caller["user_id"],
            },
        )

        return _ok({"deleted": report_id})

    def upload_attachment(self, request: dict) -> dict:
        """
        POST /attachments

        Expects request["body"] to be a dict:
            {
                "report_id": str,
                "filename":  str,
                "mime_type": str,          # optional
                "data":      bytes,
            }

        The attachment is scanned for disallowed content before storage.
        """
        try:
            caller = self._authenticate(request)
        except (AuthenticationError, AuthorizationError) as exc:
            return _err(str(exc), 401)

        try:
            require_permission(caller, "report:write")
        except AuthorizationError as exc:
            return _err(str(exc), 403)

        body      = request.get("body") or {}
        report_id = (body.get("report_id") or "").strip()
        filename  = (body.get("filename")  or "").strip()
        mime_type = (body.get("mime_type") or "application/octet-stream").strip()
        data: bytes = body.get("data") or b""

        if not report_id:
            return _err("'report_id' is required.", 400)
        if not filename:
            return _err("'filename' is required.", 400)
        if not isinstance(data, (bytes, bytearray)):
            return _err("'data' must be bytes.", 400)

        # Verify the report exists and belongs to this tenant.
        report_row = self._get_report_row(report_id, caller)
        if report_row is None:
            return _err("Report not found.", 404)

        # Content scan before persistence.
        try:
            scan(bytes(data), filename)
        except ValueError as exc:
            return _err(str(exc), 413)
        except ContentViolation as exc:
            return _err(str(exc), 422)

        attachment_id = str(uuid.uuid4())
        try:
            self._conn.execute(
                """
                INSERT INTO attachments
                    (id, report_id, tenant_id, filename, mime_type, size_bytes, data)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    attachment_id,
                    report_id,
                    caller["tenant_id"],
                    filename,
                    mime_type,
                    len(data),
                    data,
                ),
            )
            self._conn.commit()
        except sqlite3.Error as exc:
            logger.exception("upload_attachment db error")
            return _err(f"Database error: {exc}", 500)

        return _ok(
            {
                "id":         attachment_id,
                "report_id":  report_id,
                "filename":   filename,
                "size_bytes": len(data),
            },
            201,
        )
```

FILE: tests/__init__.py
```python
```

FILE: tests/conftest.py
```python
"""Shared pytest fixtures."""

import uuid
import pytest

from reportkit.db import connect, apply_schema
from reportkit.handlers import ReportHandlers


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _seed(conn, *, role="editor"):
    """
    Insert a tenant, user, role, and session; return a dict with all IDs and
    the bearer token so tests can build request dicts easily.
    """
    tenant_id = str(uuid.uuid4())
    user_id   = str(uuid.uuid4())
    token     = str(uuid.uuid4())

    conn.execute(
        "INSERT INTO tenants (id, name) VALUES (?, ?)",
        (tenant_id, "Acme Corp"),
    )
    conn.execute(
        "INSERT INTO users (id, tenant_id, email) VALUES (?, ?, ?)",
        (user_id, tenant_id, "alice@example.com"),
    )
    conn.execute(
        "INSERT INTO roles (user_id, tenant_id, role) VALUES (?, ?, ?)",
        (user_id, tenant_id, role),
    )
    conn.execute(
        """
        INSERT INTO sessions (token, user_id, expires_at)
        VALUES (?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '+1 hour'))
        """,
        (token, user_id),
    )
    conn.commit()
    return {"tenant_id": tenant_id, "user_id": user_id, "token": token}


def _request(token: str, body=None, query=None) -> dict:
    return {
        "headers": {"Authorization": f"Bearer {token}"},
        "body":    body or {},
        "query":   query or {},
    }


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture()
def conn():
    c = connect(":memory:")
    apply_schema(c)
    return c


@pytest.fixture()
def editor_ctx(conn):
    return conn, _seed(conn, role="editor"), ReportHandlers(conn)


@pytest.fixture()
def viewer_ctx(conn):
    return conn, _seed(conn, role="viewer"), ReportHandlers(conn)


@pytest.fixture()
def admin_ctx(conn):
    return conn, _seed(conn, role="admin"), ReportHandlers(conn)
```

FILE: tests/test_auth.py
```python
"""Tests for authentication and authorisation helpers."""

import uuid
import pytest

from reportkit.auth import (
    resolve_caller,
    require_permission,
    permissions_for_role,
    AuthenticationError,
    AuthorizationError,
)
from reportkit.db import connect, apply_schema


@pytest.fixture()
def conn():
    c = connect(":memory:")
    apply_schema(c)
    return c


def _make_session(conn, role="editor"):
    tenant_id = str(uuid.uuid4())
    user_id   = str(uuid.uuid4())
    token     = str(uuid.uuid4())
    conn.execute("INSERT INTO tenants (id, name) VALUES (?, ?)", (tenant_id, "T"))
    conn.execute(
        "INSERT INTO users (id, tenant_id, email) VALUES (?, ?, ?)",
        (user_id, tenant_id, "u@example.com"),
    )
    conn.execute(
        "INSERT INTO roles (user_id, tenant_id, role) VALUES (?, ?, ?)",
        (user_id, tenant_id, role),
    )
    conn.execute(
        """
        INSERT INTO sessions (token, user_id, expires_at)
        VALUES (?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '+1 hour'))
        """,
        (token, user_id),
    )
    conn.commit()
    return token, user_id, tenant_id


def _req(token):
    return {"headers": {"Authorization": f"Bearer {token}"}}


class TestPermissionsForRole:
    def test_viewer(self):
        assert permissions_for_role("viewer") == {"report:read"}

    def test_editor(self):
        assert permissions_for_role("editor") == {"report:read", "report:write"}

    def test_admin(self):
        assert permissions_for_role("admin") >= {
            "report:read", "report:write", "report:delete", "user:manage"
        }

    def test_unknown_role_returns_empty(self):
        assert permissions_for_role("superuser") == frozenset()


class TestResolveCallerSuccess:
    def test_returns_caller_dict(self, conn):
        token, user_id, tenant_id = _make_session(conn, "editor")
        caller = resolve_caller(conn, _req(token))
        assert caller["user_id"]   == user_id
        assert caller["tenant_id"] == tenant_id
        assert caller["role"]      == "editor"
        assert "report:write" in caller["permissions"]

    def test_admin_has_all_permissions(self, conn):
        token, *_ = _make_session(conn, "admin")
        caller = resolve_caller(conn, _req(token))
        for perm in ("report:read", "report:write", "report:delete", "user:manage"):
            assert perm in caller["permissions"]


class TestResolveCallerFailures:
    def test_missing_header(self, conn):
        with pytest.raises(AuthenticationError):
            resolve_caller(conn, {"headers": {}})

    def test_bad_token(self, conn):
        with pytest.raises(AuthenticationError):
            resolve_caller(conn, _req("not-a-real-token"))

    def test_expired_session(self, conn):
        tenant_id = str(uuid.uuid4())
        user_id   = str(uuid.uuid4())
        token     = str(uuid.uuid4())
        conn.execute("INSERT INTO tenants (id, name) VALUES (?, ?)", (tenant_id, "T"))
        conn.execute(
            "INSERT INTO users (id, tenant_id, email) VALUES (?, ?, ?)",
            (user_id, tenant_id, "x@example.com"),
        )
        conn.execute(
            """
            INSERT INTO sessions (token, user_id, expires_at)
            VALUES (?, ?, '2000-01-01T00:00:00.000Z')
            """,
            (token, user_id),
        )
        conn.commit()
        with pytest.raises(AuthenticationError):
            resolve_caller(conn, _req(token))

    def test_suspended_tenant(self, conn):
        token, _, tenant_id = _make_session(conn, "editor")
        conn.execute(
            "UPDATE tenants SET suspended = 1 WHERE id = ?", (tenant_id,)
        )
        conn.commit()
        with pytest.raises(AuthorizationError, match="suspended"):
            resolve_caller(conn, _req(token))


class TestRequirePermission:
    def test_passes_when_held(self):
        caller = {"permissions": frozenset({"report:read"})}
        require_permission(caller, "report:read")  # no exception

    def test_raises_when_missing(self):
        caller = {"permissions": frozenset({"report:read"})}
        with pytest.raises(AuthorizationError):
            require_permission(caller, "report:delete")
```

FILE: tests/test_handlers.py
```python
"""Tests for the HTTP surface (ReportHandlers)."""

import uuid
import pytest

from reportkit.db import connect, apply_schema
from reportkit.handlers import ReportHandlers
from tests.conftest import _seed, _request


# ---------------------------------------------------------------------------
# create_report
# ---------------------------------------------------------------------------

class TestCreateReport:
    def test_editor_can_create(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.create_report(_request(ctx["token"], body={"title": "Q1 Report"}))
        assert resp["status"] == 201
        assert resp["body"]["title"] == "Q1 Report"
        assert resp["body"]["owner_id"] == ctx["user_id"]

    def test_viewer_cannot_create(self, viewer_ctx):
        conn, ctx, h = viewer_ctx
        resp = h.create_report(_request(ctx["token"], body={"title": "X"}))
        assert resp["status"] == 403

    def test_missing_title_returns_400(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.create_report(_request(ctx["token"], body={}))
        assert resp["status"] == 400

    def test_unauthenticated_returns_401(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.create_report({"headers": {}, "body": {"title": "X"}, "query": {}})
        assert resp["status"] == 401


# ---------------------------------------------------------------------------
# get_report
# ---------------------------------------------------------------------------

class TestGetReport:
    def _create(self, h, token, title="Test"):
        resp = h.create_report(_request(token, body={"title": title}))
        assert resp["status"] == 201
        return resp["body"]["id"]

    def test_owner_can_read(self, editor_ctx):
        conn, ctx, h = editor_ctx
        rid = self._create(h, ctx["token"])
        resp = h.get_report(_request(ctx["token"]), rid)
        assert resp["status"] == 200
        assert resp["body"]["id"] == rid

    def test_not_found_returns_404(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.get_report(_request(ctx["token"]), str(uuid.uuid4()))
        assert resp["status"] == 404

    def test_cross_tenant_isolation(self, conn):
        """A user from tenant B cannot read a report belonging to tenant A."""
        apply_schema(conn)
        h = ReportHandlers(conn)

        ctx_a = _seed(conn, role="editor")
        ctx_b = _seed(conn, role="editor")

        rid = h.create_report(
            _request(ctx_a["token"], body={"title": "Secret"})
        )["body"]["id"]

        resp = h.get_report(_request(ctx_b["token"]), rid)
        assert resp["status"] == 404


# ---------------------------------------------------------------------------
# list_reports
# ---------------------------------------------------------------------------

class TestListReports:
    def test_returns_own_reports_newest_first(self, editor_ctx):
        conn, ctx, h = editor_ctx
        h.create_report(_request(ctx["token"], body={"title": "First"}))
        h.create_report(_request(ctx["token"], body={"title": "Second"}))
        resp = h.list_reports(_request(ctx["token"]))
        assert resp["status"] == 200
        titles = [r["title"] for r in resp["body"]["reports"]]
        assert titles[0] == "Second"
        assert titles[1] == "First"

    def test_viewer_can_list(self, viewer_ctx):
        conn, ctx, h = viewer_ctx
        resp = h.list_reports(_request(ctx["token"]))
        assert resp["status"] == 200

    def test_empty_list_for_new_user(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.list_reports(_request(ctx["token"]))
        assert resp["body"]["reports"] == []


# ---------------------------------------------------------------------------
# search_reports
# ---------------------------------------------------------------------------

class TestSearchReports:
    def _populate(self, h, token):
        for title in ("Alpha report", "Beta report", "Gamma note"):
            h.create_report(_request(token, body={"title": title}))

    def test_search_by_title_substring(self, editor_ctx):
        conn, ctx, h = editor_ctx
        self._populate(h, ctx["token"])
        resp = h.search_reports(_request(ctx["token"], query={"q": "report"}))
        assert resp["status"] == 200
        assert len(resp["body"]["reports"]) == 2

    def test_sort_by_title(self, editor_ctx):
        conn, ctx, h = editor_ctx
        self._populate(h, ctx["token"])
        resp = h.search_reports(
            _request(ctx["token"], query={"q": "", "sort": "title"})
        )
        titles = [r["title"] for r in resp["body"]["reports"]]
        assert titles == sorted(titles)

    def test_invalid_sort_returns_400(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.search_reports(
            _request(ctx["token"], query={"sort": "'; DROP TABLE reports; --"})
        )
        assert resp["status"] == 400

    def test_sort_injection_rejected(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.search_reports(
            _request(ctx["token"], query={"sort": "owner_id"})
        )
        assert resp["status"] == 400

    def test_empty_query_returns_all(self, editor_ctx):
        conn, ctx, h = editor_ctx
        self._populate(h, ctx["token"])
        resp = h.search_reports(_request(ctx["token"], query={"q": ""}))
        assert len(resp["body"]["reports"]) == 3


# ---------------------------------------------------------------------------
# delete_report
# ---------------------------------------------------------------------------

class TestDeleteReport:
    def _create(self, h, token, title="To delete"):
        return h.create_report(_request(token, body={"title": title}))["body"]["id"]

    def test_admin_can_delete(self, admin_ctx):
        conn, ctx, h = admin_ctx
        rid = self._create(h, ctx["token"])
        resp = h.delete_report(_request(ctx["token"]), rid)
        assert resp["status"] == 200
        assert resp["body"]["deleted"] == rid

    def test_editor_cannot_delete(self, editor_ctx):
        conn, ctx, h = editor_ctx
        rid = self._create(h, ctx["token"])
        resp = h.delete_report(_request(ctx["token"]), rid)
        assert resp["status"] == 403

    def test_delete_nonexistent_returns_404(self, admin_ctx):
        conn, ctx, h = admin_ctx
        resp = h.delete_report(_request(ctx["token"]), str(uuid.uuid4()))
        assert resp["status"] == 404

    def test_deleted_report_not_retrievable(self, admin_ctx):
        conn, ctx, h = admin_ctx
        rid = self._create(h, ctx["token"])
        h.delete_report(_request(ctx["token"]), rid)
        resp = h.get_report(_request(ctx["token"]), rid)
        assert resp["status"] == 404


# ---------------------------------------------------------------------------
# upload_attachment
# ---------------------------------------------------------------------------

class TestUploadAttachment:
    def _create_report(self, h, token):
        return h.create_report(_request(token, body={"title": "R"}))["body"]["id"]

    def test_upload_succeeds(self, editor_ctx):
        conn, ctx, h = editor_ctx
        rid = self._create_report(h, ctx["token"])
        resp = h.upload_attachment(
            _request(
                ctx["token"],
                body={
                    "report_id": rid,
                    "filename":  "data.csv",
                    "data":      b"col1,col2\n1,2\n",
                },
            )
        )
        assert resp["status"] == 201
        assert resp["body"]["filename"] == "data.csv"

    def test_eicar_rejected(self, editor_ctx):
        conn, ctx, h = editor_ctx
        rid = self._create_report(h, ctx["token"])
        eicar = b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE"
        resp = h.upload_attachment(
            _request(
                ctx["token"],
                body={"report_id": rid, "filename": "evil.txt", "data": eicar},
            )
        )
        assert resp["status"] == 422

    def test_oversized_rejected(self, editor_ctx):
        conn, ctx, h = editor_ctx
        rid = self._create_report(h, ctx["token"])
        big = b"x" * (25 * 1024 * 1024 + 1)
        resp = h.upload_attachment(
            _request(
                ctx["token"],
                body={"report_id": rid, "filename": "big.bin", "data": big},
            )
        )
        assert resp["status"] == 413

    def test_missing_report_id_returns_400(self, editor_ctx):
        conn, ctx, h = editor_ctx
        resp = h.upload_attachment(
            _request(ctx["token"], body={"filename": "f.txt", "data": b"hi"})
        )
        assert resp["status"] == 400

    def test_viewer_cannot_upload(self, viewer_ctx):
        conn, ctx, h = viewer_ctx
        resp = h.upload_attachment(
            _request(
                ctx["token"],
                body={"report_id": str(uuid.uuid4()), "filename": "f.txt", "data": b""},
            )
        )
        assert resp["status"] == 403
```

FILE: tests/test_billing.py
```python
"""Tests for the billing webhook handler."""

import hashlib
import hmac
import json
import os
import uuid
import pytest

from reportkit.db import connect, apply_schema
from reportkit.billing import handle_invoice_paid, verify_signature, WebhookAuthError


SECRET = "test-secret-key"


@pytest.fixture(autouse=True)
def set_secret(monkeypatch):
    monkeypatch.setenv("BILLING_WEBHOOK_SECRET", SECRET)


@pytest.fixture()
def conn():
    c = connect(":memory:")
    apply_schema(c)
    # Seed a tenant so FK constraints pass.
    c.execute("INSERT INTO tenants (id, name) VALUES ('t1', 'Test Tenant')")
    c.commit()
    return c


def _sign(body: bytes) -> str:
    digest = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()
    return f"sha256={digest}"


def _event(tenant_id="t1", invoice_id=None, amount=1000, event_id=None):
    return {
        "id":   event_id or str(uuid.uuid4()),
        "type": "invoice.paid",
        "data": {
            "tenant_id":    tenant_id,
            "invoice_id":   invoice_id or str(uuid.uuid4()),
            "amount_cents": amount,
            "currency":     "usd",
        },
    }


class TestVerifySignature:
    def test_valid_signature_passes(self):
        body = b'{"hello": "world"}'
        verify_signature(body, _sign(body))  # no exception

    def test_invalid_signature_raises(self):
        body = b'{"hello": "world"}'
        with pytest.raises(WebhookAuthError):
            verify_signature(body, "sha256=deadbeef")

    def test_missing_prefix_raises(self):
        with pytest.raises(WebhookAuthError):
            verify_signature(b"body", "deadbeef")


class TestHandleInvoicePaid:
    def test_valid_event_creates_ledger_entry(self, conn):
        event = _event()
        body  = json.dumps(event).encode()
        resp  = handle_invoice_paid(conn, body, _sign(body))
        assert resp["status"] == 200
        row = conn.execute("SELECT * FROM ledger").fetchone()
        assert row is not None
        assert row["invoice_id"] == event["data"]["invoice_id"]

    def test_duplicate_delivery_is_idempotent(self, conn):
        event = _event(event_id="evt-001")
        body  = json.dumps(event).encode()
        sig   = _sign(body)
        handle_invoice_paid(conn, body, sig)
        handle_invoice_paid(conn, body, sig)  # second delivery
        count = conn.execute("SELECT COUNT(*) FROM ledger").fetchone()[0]
        assert count == 1

    def test_bad_signature_returns_error(self, conn):
        body = json.dumps(_event()).encode()
        with pytest.raises(WebhookAuthError):
            handle_invoice_paid(conn, body, "sha256=bad")

    def test_unknown_event_type_acknowledged(self, conn):
        event = {"id": "x", "type": "invoice.voided", "data": {}}
        body  = json.dumps(event).encode()
        resp  = handle_invoice_paid(conn, body, _sign(body))
        assert resp["status"] == 200
        assert resp["body"]["processed"] is False

    def test_missing_fields_returns_400(self, conn):
        event = {"id": "x", "type": "invoice.paid", "data": {}}
        body  = json.dumps(event).encode()
        resp  = handle_invoice_paid(conn, body, _sign(body))
        assert resp["status"] == 400


class TestMissingSecret:
    def test_raises_when_env_var_absent(self, conn, monkeypatch):
        monkeypatch.delenv("BILLING_WEBHOOK_SECRET", raising=False)
        body = json.dumps(_event()).encode()
        with pytest.raises(EnvironmentError):
            handle_invoice_paid(conn, body, "sha256=anything")
```

FILE: tests/test_ops.py
```python
"""Tests for the internal ops-console operations."""

import uuid
import pytest

from reportkit.db import connect, apply_schema
from reportkit.handlers import ReportHandlers
from reportkit.ops import purge_tenant, set_role
from tests.conftest import _seed, _request


@pytest.fixture()
def conn():
    c = connect(":memory:")
    apply_schema(c)
    return c


class TestPurgeTenant:
    def test_deletes_all_tenant_reports(self, conn):
        ctx = _seed(conn, role="editor")
        h   = ReportHandlers(conn)
        for i in range(3):
            h.create_report(_request(ctx["token"], body={"title": f"R{i}"}))

        count = purge_tenant(conn, actor_id="ops-bot", tenant_id=ctx["tenant_id"])
        assert count == 3

        remaining = conn.execute(
            "SELECT COUNT(*) FROM reports WHERE tenant_id = ?",
            (ctx["tenant_id"],),
        ).fetchone()[0]
        assert remaining == 0

    def test_does_not_affect_other_tenants(self, conn):
        ctx_a = _seed(conn, role="editor")
        ctx_b = _seed(conn, role="editor")
        h     = ReportHandlers(conn)

        h.create_report(_request(ctx_a["token"], body={"title": "A"}))
        h.create_report(_request(ctx_b["token"], body={"title": "B"}))

        purge_tenant(conn, actor_id="ops", tenant_id=ctx_a["tenant_id"])

        remaining_b = conn.execute(
            "SELECT COUNT(*) FROM reports WHERE tenant_id = ?",
            (ctx_b["tenant_id"],),
        ).fetchone()[0]
        assert remaining_b == 1

    def test_returns_zero_for_empty_tenant(self, conn):
        ctx   = _seed(conn, role="editor")
        count = purge_tenant(conn, actor_id="ops", tenant_id=ctx["tenant_id"])
        assert count == 0


class TestSetRole:
    def test_changes_role(self, conn):
        ctx = _seed(conn, role="viewer")
        set_role(conn, "ops", ctx["user_id"], ctx["tenant_id"], "admin")
        row = conn.execute(
            "SELECT role FROM roles WHERE user_id = ? AND tenant_id = ?",
            (ctx["user_id"], ctx["tenant_id"]),
        ).fetchone()
        assert row["role"] == "admin"

    def test_invalid_role_raises(self, conn):
        ctx = _seed(conn, role="viewer")
        with pytest.raises(ValueError, match="Unknown role"):
            set_role(conn, "ops", ctx["user_id"], ctx["tenant_id"], "superuser")

    def test_upsert_does_not_duplicate(self, conn):
        ctx = _seed(conn, role="viewer")
        set_role(conn, "ops", ctx["user_id"], ctx["tenant_id"], "editor")
        set_role(conn, "ops", ctx["user_id"], ctx["tenant_id"], "admin")
        count = conn.execute(
            "SELECT COUNT(*) FROM roles WHERE user_id = ?",
            (ctx["user_id"],),
        ).fetchone()[0]
        assert count == 1
```

FILE: tests/test_content_scanner.py
```python
"""Tests for the attachment content scanner."""

import pytest
from reportkit.content_scanner import scan, ContentViolation, MAX_ATTACHMENT_BYTES


class TestScan:
    def test_clean_data_passes(self):
        scan(b"hello world", "file.txt")  # no exception

    def test_eicar_string_rejected(self):
        with pytest.raises(ContentViolation):
            scan(b"prefix EICAR-STANDARD-ANTIVIRUS-TEST-FILE suffix", "bad.txt")

    def test_eicar_header_variant_rejected(self):
        with pytest.raises(ContentViolation):
            scan(b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR", "bad2.txt")

    def test_oversized_raises_value_error(self):
        with pytest.raises(ValueError, match="25 MB"):
            scan(b"x" * (MAX_ATTACHMENT_BYTES + 1), "big.bin")

    def test_exactly_at_limit_passes(self):
        scan(b"x" * MAX_ATTACHMENT_BYTES, "ok.bin")  # no exception

    def test_empty_bytes_passes(self):
        scan(b"", "empty.txt")
```

FILE: tests/test_events.py
```python
"""Tests for the event registry."""

import pytest
from reportkit import events as ev


@pytest.fixture(autouse=True)
def clean_registry():
    """Isolate each test by saving and restoring the registry."""
    original = dict(ev._registry)
    yield
    ev._registry.clear()
    ev._registry.update(original)


class TestRegisterAndDispatch:
    def test_handler_is_called(self):
        received = []
        ev.register("test.event", lambda p: received.append(p))
        ev.dispatch("test.event", {"x": 1})
        assert received == [{"x": 1}]

    def test_multiple_handlers_all_called(self):
        log = []
        ev.register("multi", lambda p: log.append("a"))
        ev.register("multi", lambda p: log.append("b"))
        ev.dispatch("multi", {})
        assert log == ["a", "b"]

    def test_no_handlers_is_silent(self):
        ev.dispatch("no.handlers", {})  # no exception

    def test_failing_handler_does_not_block_others(self):
        log = []

        def bad(p):
            raise RuntimeError("boom")

        ev.register("err.event", bad)
        ev.register("err.event", lambda p: log.append("ok"))

        with pytest.raises(ExceptionGroup):
            ev.dispatch("err.event", {})

        assert "ok" in log

    def test_exception_group_raised_on_error(self):
        ev.register("fail", lambda p: (_ for _ in ()).throw(ValueError("oops")))
        with pytest.raises(ExceptionGroup) as exc_info:
            ev.dispatch("fail", {})
        assert len(exc_info.value.exceptions) == 1
```

FILE: NOTES.md
```markdown
# reportkit — design notes

## Decisions

### No web framework
The spec requires a plain class whose methods take a `request` dict.
`ReportHandlers` is that class.  It can be wrapped by any WSGI/ASGI adapter
(e.g. a two-line `wsgiref` shim) without touching business logic.

### SQLite + WAL mode
WAL mode allows concurrent readers alongside a single writer, which is
appropriate for a service that is read-heavy (list/search) and write-light.
Foreign keys are enforced at the connection level (`PRAGMA foreign_keys = ON`).

### Hot permissions query
The spec calls out the permissions join as the hottest query.  The schema
carries a covering index `idx_roles_user_tenant ON roles(user_id, tenant_id)`.
`resolve_caller` performs a single three-table join (sessions → users → roles)
per request rather than separate round-trips.  If the service were to scale
beyond a single process, this is the first candidate for a short-lived
in-process cache keyed on `(token, expires_at)`.

### Sort-column injection defence
`search_reports` accepts a user-supplied sort column.  SQLite parameterised
queries cannot bind column names, so the value is validated against a
`frozenset` allowlist in `db.validate_sort_column` before string interpolation.
An invalid value returns HTTP 400.  This is the only place in the codebase
where a user value is interpolated into SQL.

### Billing idempotency
The billing provider retries deliveries.  The `ledger` table has a `UNIQUE`
constraint on `idempotency_key` (the provider's event ID).  A duplicate
`INSERT` raises `sqlite3.IntegrityError`, which is caught and silently ignored;
the endpoint returns 200 so the provider stops retrying.

### HMAC verification
`billing.verify_signature` uses `hmac.compare_digest` to avoid timing attacks.
The shared secret is read from the environment variable
`BILLING_WEBHOOK_SECRET`; the function raises `EnvironmentError` at call time
(not import time) if the variable is absent, so the rest of the service starts
normally even if billing is not configured.

### Content scanning
`content_scanner.scan` checks for a tuple of byte-string signatures before
persistence.  The interface is a single `scan(data, filename)` function that
raises typed exceptions (`ContentViolation`, `ValueError`).  Swapping in a
real AV service means replacing the body of `scan` without changing any
call-site.

### tenant.quota_exceeded handler
The event is registered and the handler is a documented no-op stub.  The
batch-import feed that emits it was retired; keeping the registration means
the platform team can activate it by replacing the stub body rather than
touching the registry wiring.

### Ops console functions
`purge_tenant` and `set_role` are plain functions in `ops.py`, not HTTP
endpoints.  They are called directly by the internal console over the private
network.  Cascading deletes in the schema (`ON DELETE CASCADE`) mean
`purge_tenant` only needs to delete from `reports`; attachments follow
automatically.

### Attachments stored as BLOBs
For simplicity, attachment bytes are stored directly in SQLite.  At 25 MB per
file this is workable for low-to-moderate volume.  The natural upgrade path is
to store a reference (e.g. an S3 key) instead of the bytes; the
`upload_attachment` handler is the only place that would change.

## Deliberate omissions

- **Pagination** — `list_reports` and `search_reports` return all matching
  rows.  A `LIMIT`/`OFFSET` or cursor-based scheme would be added before
  production.
- **Audit log** — `purge_tenant` and `set_role` log to the Python logger.  A
  proper audit table is the obvious next step.
- **Rate limiting** — not implemented; belongs in a gateway layer.
- **HTTPS / TLS** — the class is transport-agnostic; TLS is the adapter's
  responsibility.
- **Migrations** — `apply_schema` is idempotent (`CREATE TABLE IF NOT EXISTS`)
  but is not a migration system.  Alembic or a hand-rolled migration table
  would be needed for schema evolution.
- **`ExceptionGroup`** — used in `events.py`; requires Python 3.11+.  On
  3.10 this can be replaced with a plain `RuntimeError` wrapping the list of
  causes.
```