diff --git a/api/src/services/audit.py b/api/src/services/audit.py
new file mode 100644
index 0000000..474fec4
--- /dev/null
+++ b/api/src/services/audit.py
@@ -0,0 +1,119 @@
+"""Сервис аудита: единственная точка записи `audit_event` (spec audit-trail, ТЗ 8.2).
+
+Журнал append-only: у сервиса один публичный метод — `record`; методов изменения и удаления нет и не будет
+(проверяется тестом `tests/test_audit_service.py`, он же сканирует `src/` на UPDATE/DELETE по `AuditEvent`).
+
+Запись выполняется в ТЕКУЩЕЙ транзакции сессии: сервис делает `add` + `flush` и не вызывает `commit`/`rollback` —
+фиксация и откат остаются за вызывающим кодом, поэтому откат изменения данных откатывает и строку аудита.
+"""
+
+import datetime
+import decimal
+import enum
+import math
+import uuid
+from collections.abc import Mapping
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.models.audit_ex import AuditEvent
+from src.models.enums import AuditAction, AuditSection
+
+
+def _scalar_to_jsonable(value: Any) -> Any:
+    if value is None or isinstance(value, bool | int | str):
+        return value
+    if isinstance(value, enum.Enum):
+        return to_jsonable(value.value)
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            raise ValueError("Нечисловое значение float (NaN/Infinity) нельзя записать в jsonb")
+        return value
+    if isinstance(value, decimal.Decimal):
+        return format(value, "f")
+    if isinstance(value, uuid.UUID):
+        return str(value)
+    if isinstance(value, datetime.datetime | datetime.date | datetime.time):
+        return value.isoformat()
+    raise TypeError(f"Значение типа {type(value).__name__} не сериализуется в jsonb для audit_event")
+
+
+def to_jsonable(value: Any) -> Any:
+    """Приводит значение к JSON-совместимому виду для jsonb.
+
+    UUID -> str; Decimal -> строка без экспоненты (деньги не превращаются в float); datetime/date/time -> isoformat;
+    Enum -> `.value`; Mapping/list/tuple/set -> рекурсивно. Неизвестный тип — `TypeError`, а не молчаливое `str()`.
+    """
+    if isinstance(value, Mapping):
+        return {str(key): to_jsonable(item) for key, item in value.items()}
+    if isinstance(value, list | tuple | set | frozenset):
+        return [to_jsonable(item) for item in value]
+    return _scalar_to_jsonable(value)
+
+
+def _normalize_payload(name: str, payload: Mapping[str, Any] | None) -> Any:
+    if payload is None:
+        return None
+    if not isinstance(payload, Mapping):
+        raise TypeError(f"{name} должен быть словарём (Mapping) или None, получено {type(payload).__name__}")
+    return to_jsonable(payload)
+
+
+class AuditService:
+    """Запись событий журнала аудита в сессию вызывающего (ручной DI: сессия приходит в конструктор)."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    async def record(  # noqa: PLR0913 - сигнатура зафиксирована спекой audit-trail (все поля audit_event)
+        self,
+        *,
+        action: AuditAction,
+        section: AuditSection,
+        object_type: str,
+        object_id: uuid.UUID | None = None,
+        user_id: uuid.UUID | None = None,
+        consolidation_version_id: uuid.UUID | None = None,
+        description: str | None = None,
+        before_value: Mapping[str, Any] | None = None,
+        after_value: Mapping[str, Any] | None = None,
+        comment: str | None = None,
+        result: str | None = None,
+        changed_field: str | None = None,
+        ip_address: str | None = None,
+        user_agent: str | None = None,
+    ) -> AuditEvent:
+        """Добавляет событие в текущую транзакцию и делает `flush` (id, ранние нарушения ограничений); без `commit`.
+
+        `user_id=None` — системное действие. `occurred_at` ставит сервис (UTC), параметром он не принимается.
+        """
+        if not isinstance(action, AuditAction):
+            raise TypeError(f"action должен быть членом AuditAction, получено {type(action).__name__}: {action!r}")
+        if not isinstance(section, AuditSection):
+            raise TypeError(f"section должен быть членом AuditSection, получено {type(section).__name__}: {section!r}")
+        if not isinstance(object_type, str):
+            raise TypeError(f"object_type должен быть строкой, получено {type(object_type).__name__}")
+        if not object_type.strip():
+            raise ValueError("object_type не может быть пустым")
+
+        event = AuditEvent(
+            occurred_at=datetime.datetime.now(datetime.UTC),
+            user_id=user_id,
+            action=action,
+            section=section,
+            object_type=object_type,
+            object_id=object_id,
+            consolidation_version_id=consolidation_version_id,
+            description=description,
+            before_value=_normalize_payload("before_value", before_value),
+            after_value=_normalize_payload("after_value", after_value),
+            comment=comment,
+            result=result,
+            changed_field=changed_field,
+            ip_address=ip_address,
+            user_agent=user_agent,
+        )
+        self._session.add(event)
+        await self._session.flush()
+        return event
diff --git a/api/tests/test_audit_service.py b/api/tests/test_audit_service.py
new file mode 100644
index 0000000..2378472
--- /dev/null
+++ b/api/tests/test_audit_service.py
@@ -0,0 +1,381 @@
+"""Сервис аудита (spec audit-trail, задача 21): системное событие, откат, append-only, статический запрет UPDATE/DELETE.
+
+Быстрые тесты (валидация, сканер, статический обход `src/`) идут без БД; интеграционные — на `db_session`.
+"""
+
+import ast
+import datetime
+import inspect
+import re
+import uuid
+from decimal import Decimal
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.models.audit_ex import AuditEvent
+from src.models.directories import RegionalOffice
+from src.models.enums import AuditAction, AuditSection
+from src.services.audit import AuditService, to_jsonable
+
+SRC_DIR = Path(__file__).resolve().parent.parent / "src"
+
+# ---------------------------------------------------------------------------------------------------------------------
+# Статический сканер: обращения, изменяющие или удаляющие audit_event
+# ---------------------------------------------------------------------------------------------------------------------
+
+MUTATING_CALLS = {"update", "delete"}
+AUDIT_FIELDS = {
+    "occurred_at",
+    "user_id",
+    "action",
+    "section",
+    "object_type",
+    "object_id",
+    "consolidation_version_id",
+    "description",
+    "before_value",
+    "after_value",
+    "comment",
+    "result",
+    "changed_field",
+    "ip_address",
+    "user_agent",
+    "id",
+}
+RAW_SQL_MUTATION = re.compile(
+    r"\b(update|delete\s+from|truncate(\s+table)?)\s+(only\s+)?(\"?dkb\"?\.)?\"?audit_event\b", re.I
+)
+
+
+def _call_name(node: ast.Call) -> str | None:
+    func = node.func
+    if isinstance(func, ast.Name):
+        return func.id
+    if isinstance(func, ast.Attribute):
+        return func.attr
+    return None
+
+
+def _mentions_audit_event(node: ast.AST) -> bool:
+    return any(
+        (isinstance(sub, ast.Name) and sub.id == "AuditEvent")
+        or (isinstance(sub, ast.Attribute) and sub.attr == "AuditEvent")
+        for sub in ast.walk(node)
+    )
+
+
+def _is_audit_event_expr(node: ast.AST | None) -> bool:
+    """Выражение, дающее экземпляр AuditEvent: `AuditEvent(...)` / `mod.AuditEvent(...)`."""
+    return isinstance(node, ast.Call) and _mentions_audit_event(node.func)
+
+
+def _tainted_names(tree: ast.AST) -> set[str]:
+    """Имена, связанные с экземплярами AuditEvent: присваивание `AuditEvent(...)`, аннотация `: AuditEvent`."""
+    names: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Assign) and _is_audit_event_expr(node.value):
+            names.update(t.id for t in node.targets if isinstance(t, ast.Name))
+        elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
+            if _mentions_audit_event(node.annotation) or _is_audit_event_expr(node.value):
+                names.add(node.target.id)
+        elif isinstance(node, ast.arg) and node.annotation is not None and _mentions_audit_event(node.annotation):
+            names.add(node.arg)
+        elif isinstance(node, ast.For | ast.comprehension) and _is_audit_event_expr(getattr(node, "iter", None)):
+            pass
+    return names
+
+
+def find_audit_event_mutations(source: str, filename: str = "<src>") -> list[str]:
+    """Находки вида `file:line: описание` — любое изменение/удаление audit_event в исходнике.
+
+    Ловит: `update(AuditEvent...)` / `delete(AuditEvent...)` (в т.ч. `sa.update`, `AuditEvent.__table__`);
+    `.delete(x)`/`delete(x)`, где x — экземпляр AuditEvent (внутримодульное отслеживание имён); присваивание/удаление
+    атрибутов таких экземпляров и `AuditEvent.<поле> = ...`; сырой SQL UPDATE/DELETE/TRUNCATE по audit_event.
+    """
+    tree = ast.parse(source, filename=filename)
+    tainted = _tainted_names(tree)
+    findings: list[str] = []
+
+    def add(node: ast.AST, message: str) -> None:
+        findings.append(f"{filename}:{getattr(node, 'lineno', 0)}: {message}")
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Call):
+            name = _call_name(node)
+            args: list[ast.AST] = [*node.args, *(kw.value for kw in node.keywords)]
+            if name in MUTATING_CALLS:
+                if any(_mentions_audit_event(arg) for arg in args):
+                    add(node, f"{name}() по AuditEvent")
+                elif any(isinstance(arg, ast.Name) and arg.id in tainted for arg in args):
+                    add(node, f"{name}() экземпляра AuditEvent")
+        elif isinstance(node, ast.Assign | ast.AugAssign | ast.AnnAssign | ast.Delete):
+            targets = node.targets if isinstance(node, ast.Assign | ast.Delete) else [node.target]
+            for target in targets:
+                if isinstance(target, ast.Attribute) and target.attr in AUDIT_FIELDS:
+                    base = target.value
+                    if (
+                        isinstance(base, ast.Name) and (base.id in tainted or base.id == "AuditEvent")
+                    ) or _is_audit_event_expr(base):
+                        add(node, f"изменение поля {target.attr} у AuditEvent")
+        elif isinstance(node, ast.Constant) and isinstance(node.value, str) and RAW_SQL_MUTATION.search(node.value):
+            add(node, "сырой SQL, изменяющий audit_event")
+    return findings
+
+
+def scan_directory(root: Path) -> list[str]:
+    findings: list[str] = []
+    for path in sorted(root.rglob("*.py")):
+        findings.extend(find_audit_event_mutations(path.read_text(encoding="utf-8"), str(path.relative_to(root))))
+    return findings
+
+
+def test_src_has_no_audit_event_update_or_delete() -> None:
+    assert SRC_DIR.is_dir()
+    assert list(SRC_DIR.rglob("audit.py")), "сканер должен видеть src/services/audit.py"
+    assert scan_directory(SRC_DIR) == []
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "from sqlalchemy import update\nstmt = update(AuditEvent).values(comment='x')",
+        "import sqlalchemy as sa\nstmt = sa.delete(AuditEvent).where(AuditEvent.id == 1)",
+        "stmt = update(AuditEvent.__table__).values(result='x')",
+        "stmt = delete(models.audit_ex.AuditEvent)",
+        "async def f(session):\n    event = AuditEvent(object_type='x')\n    await session.delete(event)",
+        "def f(event: AuditEvent):\n    event.result = 'changed'",
+        "def f(s):\n    e = AuditEvent(object_type='x')\n    e.comment = 'x'",
+        "def f(s):\n    e = AuditEvent(object_type='x')\n    del e.comment",
+        "AuditEvent.comment = 'x'",
+        "stmt = text('UPDATE dkb.audit_event SET comment = 1')",
+        "stmt = text('delete from audit_event')",
+        "stmt = text('TRUNCATE TABLE dkb.audit_event')",
+    ],
+)
+def test_scanner_detects_violation(source: str) -> None:
+    assert find_audit_event_mutations(source), f"сканер пропустил нарушение: {source!r}"
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "stmt = select(AuditEvent).where(AuditEvent.id == 1)",
+        "event = AuditEvent(object_type='x')\nsession.add(event)",
+        "stmt = update(RegionalOffice).values(name='x')",
+        "async def f(session, office):\n    await session.delete(office)",
+        "def f(office):\n    office.comment = 'x'",
+        "stmt = text('select * from audit_event')",
+    ],
+)
+def test_scanner_passes_allowed_code(source: str) -> None:
+    assert find_audit_event_mutations(source) == []
+
+
+def test_scan_directory_fails_on_synthetic_violation(tmp_path: Path) -> None:
+    (tmp_path / "ok.py").write_text("x = 1\n", encoding="utf-8")
+    sub = tmp_path / "services"
+    sub.mkdir()
+    (sub / "bad.py").write_text("from sqlalchemy import update\nstmt = update(AuditEvent)\n", encoding="utf-8")
+    findings = scan_directory(tmp_path)
+    assert len(findings) == 1
+    assert findings[0].startswith("services/bad.py:2:")
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+# Append-only и валидация (без БД)
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+def test_service_exposes_only_record() -> None:
+    public = {name for name in dir(AuditService) if not name.startswith("_")}
+    assert public == {"record"}
+
+
+async def test_record_rejects_empty_object_type() -> None:
+    service = AuditService(AsyncMock(spec=AsyncSession))
+    for bad in ("", "   "):
+        with pytest.raises(ValueError, match="object_type"):
+            await service.record(action=AuditAction.LOGIN, section=AuditSection.SESSION, object_type=bad)
+
+
+async def test_record_rejects_string_enums() -> None:
+    session = AsyncMock(spec=AsyncSession)
+    service = AuditService(session)
+    with pytest.raises(TypeError, match="action"):
+        await service.record(action="LOGIN", section=AuditSection.SESSION, object_type="app_user")  # type: ignore[arg-type]
+    with pytest.raises(TypeError, match="section"):
+        await service.record(action=AuditAction.LOGIN, section="SESSION", object_type="app_user")  # type: ignore[arg-type]
+    session.add.assert_not_called()
+
+
+async def test_record_rejects_non_mapping_payload_and_unknown_types() -> None:
+    service = AuditService(AsyncMock(spec=AsyncSession))
+    with pytest.raises(TypeError, match="before_value"):
+        await service.record(
+            action=AuditAction.LOGIN,
+            section=AuditSection.SESSION,
+            object_type="app_user",
+            before_value=["x"],  # type: ignore[arg-type]
+        )
+    with pytest.raises(TypeError, match="object"):
+        await service.record(
+            action=AuditAction.LOGIN,
+            section=AuditSection.SESSION,
+            object_type="app_user",
+            after_value={"x": object()},
+        )
+
+
+def test_record_does_not_accept_occurred_at() -> None:
+    assert "occurred_at" not in inspect.signature(AuditService.record).parameters
+
+
+def test_to_jsonable_types() -> None:
+    uid = uuid.UUID("12345678-1234-5678-1234-567812345678")
+    moment = datetime.datetime(2026, 10, 3, 12, 0, tzinfo=datetime.UTC)
+    result = to_jsonable(
+        {
+            "sum": Decimal("106311116.46"),
+            "tiny": Decimal("1E+2"),
+            "id": uid,
+            "at": moment,
+            "day": datetime.date(2026, 10, 3),
+            "action": AuditAction.LOGIN,
+            "nested": [{"k": Decimal("0.10")}, (1, 2)],
+            3: None,
+        }
+    )
+    assert result == {
+        "sum": "106311116.46",
+        "tiny": "100",
+        "id": str(uid),
+        "at": moment.isoformat(),
+        "day": "2026-10-03",
+        "action": "LOGIN",
+        "nested": [{"k": "0.10"}, [1, 2]],
+        "3": None,
+    }
+    with pytest.raises(ValueError, match="NaN"):
+        to_jsonable({"x": float("nan")})
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+# Интеграция: запись в текущую транзакцию
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+async def _count_events(session: AsyncSession, object_id: uuid.UUID) -> int:
+    return (
+        await session.scalar(select(func.count()).select_from(AuditEvent).where(AuditEvent.object_id == object_id)) or 0
+    )
+
+
+async def _count_offices(session: AsyncSession, name: str) -> int:
+    return (
+        await session.scalar(select(func.count()).select_from(RegionalOffice).where(RegionalOffice.name == name)) or 0
+    )
+
+
+@pytest.mark.integration
+async def test_system_event_without_user(db_session: AsyncSession) -> None:
+    before = datetime.datetime.now(datetime.UTC)
+    event = await AuditService(db_session).record(
+        action=AuditAction.RECOGNIZED_DATA_EDIT,
+        section=AuditSection.RECOGNITION,
+        object_type="recognition_job",
+        user_id=None,
+        before_value={"status": "queued"},
+        after_value={"status": "published"},
+    )
+    assert event.id is not None
+    await db_session.refresh(event)  # перечитать строку из БД (jsonb round-trip)
+
+    row = (await db_session.execute(select(AuditEvent).where(AuditEvent.id == event.id))).scalar_one()
+    assert row.user_id is None
+    assert row.action is AuditAction.RECOGNIZED_DATA_EDIT
+    assert row.section is AuditSection.RECOGNITION
+    assert row.before_value == {"status": "queued"}
+    assert row.after_value == {"status": "published"}
+    assert row.occurred_at.tzinfo is not None
+    assert row.occurred_at.utcoffset() == datetime.timedelta(0)
+    assert (
+        before - datetime.timedelta(seconds=5)
+        <= row.occurred_at
+        <= datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=5)
+    )
+
+
+@pytest.mark.integration
+async def test_payload_decimal_uuid_datetime_stored_as_strings(db_session: AsyncSession) -> None:
+    uid = uuid.uuid4()
+    moment = datetime.datetime(2026, 10, 3, 9, 30, 15, tzinfo=datetime.UTC)
+    event = await AuditService(db_session).record(
+        action=AuditAction.CORRECTION_CREATE,
+        section=AuditSection.CONSOLIDATION,
+        object_type="correction",
+        object_id=uid,
+        after_value={"amount": Decimal("106311116.46"), "ref": uid, "at": moment, "kind": AuditSection.KFO},
+        changed_field="amount",
+        comment="ручная корректировка",
+        result="ok",
+        ip_address="10.0.0.1",
+        user_agent="pytest",
+        description="тест",
+    )
+    await db_session.refresh(event)
+    assert event.after_value == {
+        "amount": "106311116.46",
+        "ref": str(uid),
+        "at": moment.isoformat(),
+        "kind": "KFO",
+    }
+    assert isinstance(event.after_value["amount"], str)
+    assert (event.object_id, event.changed_field, event.comment, event.result) == (
+        uid,
+        "amount",
+        "ручная корректировка",
+        "ok",
+    )
+    assert (event.ip_address, event.user_agent, event.description) == ("10.0.0.1", "pytest", "тест")
+
+
+@pytest.mark.integration
+async def test_rollback_removes_audit_with_data(db_session: AsyncSession) -> None:
+    office_name = f"Филиал-аудит-{uuid.uuid4().hex[:8]}"
+    object_id = uuid.uuid4()
+
+    with pytest.raises(RuntimeError, match="отмена"):
+        async with db_session.begin_nested():  # SAVEPOINT: откат при исключении, как откат транзакции запроса
+            db_session.add(RegionalOffice(name=office_name))
+            event = await AuditService(db_session).record(
+                action=AuditAction.DIRECTORY_IMPORT,
+                section=AuditSection.DIRECTORIES,
+                object_type="regional_office",
+                object_id=object_id,
+            )
+            # внутри транзакции запись и данные видны
+            assert event.occurred_at is not None
+            seen = await _count_events(db_session, object_id)
+            assert seen == 1
+            assert await _count_offices(db_session, office_name) == 1
+            raise RuntimeError("отмена")
+
+    assert await _count_events(db_session, object_id) == 0
+    assert await _count_offices(db_session, office_name) == 0
+
+
+@pytest.mark.integration
+async def test_record_does_not_commit_and_external_rollback_discards(db_session: AsyncSession) -> None:
+    object_id = uuid.uuid4()
+    await AuditService(db_session).record(
+        action=AuditAction.LOGIN, section=AuditSection.SESSION, object_type="app_user", object_id=object_id
+    )
+    assert db_session.in_transaction()  # record не закрыл транзакцию (нет commit)
+    assert await _count_events(db_session, object_id) == 1
+
+    await db_session.rollback()  # внешний откат (в фикстуре — к SAVEPOINT) уничтожает запись
+    assert await _count_events(db_session, object_id) == 0