diff --git a/README.md b/README.md
index 25559b2..f45bfc5 100644
--- a/README.md
+++ b/README.md
@@ -44,6 +44,31 @@
 cd frontend && npm install && npm run dev   # http://localhost:5174
 ```
 
+### Проверка Ingestion (#31)
+
+```bash
+# Сидинг конфигурации (в проде — через админку):
+docker compose exec api python -m app.cli create-source monitoring
+docker compose exec api python -m app.cli add-type monitoring container down
+docker compose exec api python -m app.cli add-target telegram infra-alerts --config '{"chat_id":"@infra-alerts"}'
+docker compose exec api python -m app.cli add-rule "контейнеры вниз" --source monitoring \
+  --subjects container --actions down --priority high \
+  --template "Контейнер {{ payload.container }} упал" \
+  --action telegram:infra-alerts --action internal_log
+docker compose exec api python -m app.cli create-key monitoring   # токен печатается один раз
+
+# Событие (202 мгновенно) и его статус:
+curl -X POST http://localhost:8013/api/v1/events \
+  -H "Authorization: Bearer syn_..." -H "Content-Type: application/json" \
+  -d '{"source":"monitoring","subject":"container","action":"down","priority":"critical","payload":{"container":"api","host":"melody"},"dedup_key":"demo-1"}'
+curl http://localhost:8013/api/v1/events/<id> -H "Authorization: Bearer syn_..."
+
+# Коды защиты: 401 — ключа нет/битый, 403 — source не совпадает с ключом,
+# 422 — тройка (source, subject, action) не зарегистрирована.
+# Пачка: POST /api/v1/events/batch — список конвертов, честные ошибки по индексам.
+# Дедуп: тот же dedup_key в окне (DEDUP_WINDOW_SECONDS, дефолт 24ч) → id первого.
+```
+
 ### Проверка воркера
 
 ```bash
diff --git a/app/api/events_routes.py b/app/api/events_routes.py
new file mode 100644
index 0000000..476b9b5
--- /dev/null
+++ b/app/api/events_routes.py
@@ -0,0 +1,225 @@
+"""Ingestion: приём событий по контракту docs/05-ingestion-api.md.
+
+Приём быстрый: жёсткая валидация конверта + дедупликация + вставка в БД +
+бросок в Celery (202 мгновенно). Никакой работы с каналами здесь.
+"""
+
+import uuid
+from datetime import UTC, datetime, timedelta
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
+from pydantic import ValidationError
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.api.schemas import (
+    AcceptedEvent,
+    BatchRejected,
+    BatchResult,
+    DeliverySummary,
+    EventEnvelope,
+    EventStatus,
+)
+from app.auth.apikeys import resolve_event_source
+from app.config import get_settings
+from app.database import get_db
+from app.models import ApiKey, ChannelTarget, Delivery, Event, NotificationType
+from app.worker.celery_app import celery_app
+
+router = APIRouter(prefix="/api/v1/events", tags=["ingestion"])
+
+
+class Deduplicated(Exception):
+    """Сигнал из accept_event: это дубль события (id и статус первого)."""
+
+    def __init__(self, event_id: uuid.UUID, event_status: str) -> None:
+        self.event_id = event_id
+        self.event_status = event_status
+
+
+def accept_event(db: Session, api_key: ApiKey, source_name: str, envelope: EventEnvelope) -> Event:
+    """Общая логика приёма (одиночный и batch): проверки + вставка.
+
+    Диспетчеризацию в Celery выполняет вызывающий код после коммита.
+    """
+    if envelope.source != source_name:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=(
+                f"Поле source='{envelope.source}' не совпадает с источником "
+                f"ключа '{source_name}'"
+            ),
+        )
+
+    known_type = db.execute(
+        select(NotificationType.id).where(
+            NotificationType.source_id == api_key.source_id,
+            NotificationType.subject == envelope.subject,
+            NotificationType.action == envelope.action,
+        )
+    ).scalar_one_or_none()
+    if known_type is None:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=(
+                f"Тип ({envelope.source}, {envelope.subject}, {envelope.action}) "
+                "не зарегистрирован — обратись к админу Synapse"
+            ),
+        )
+
+    # Дедупликация: тот же dedup_key от источника в окне времени — дубль.
+    # Точность в рантайме под нагрузкой не гарантирована (нет unique-констрейнта);
+    # race на «одновременно влетевшие» дублей в MVP не ловим.
+    if envelope.dedup_key:
+        window = get_settings().dedup_window_seconds
+        prior = db.execute(
+            select(Event)
+            .where(
+                Event.source_id == api_key.source_id,
+                Event.dedup_key == envelope.dedup_key,
+                Event.created_at > datetime.now(UTC) - timedelta(seconds=window),
+            )
+            .order_by(Event.created_at.desc())
+        ).scalars().first()
+        if prior is not None:
+            raise Deduplicated(prior.id, prior.status)
+
+    event = Event(
+        source_id=api_key.source_id,
+        subject=envelope.subject,
+        action=envelope.action,
+        priority=envelope.priority,
+        payload=envelope.payload,
+        dedup_key=envelope.dedup_key,
+        expires_at=(
+            datetime.now(UTC) + timedelta(seconds=envelope.ttl_seconds)
+            if envelope.ttl_seconds
+            else None
+        ),
+        scheduled_at=envelope.scheduled_at,
+        status="queued",
+    )
+    db.add(event)
+    # flush: uuid4-дефолт id срабатывает на INSERT; batch строит ответ до коммита.
+    db.flush()
+    return event
+
+
+def _dispatch(event_id: str) -> None:
+    celery_app.send_task("synapse.ingest", args=[event_id])
+
+
+@router.post("", status_code=202, response_model=AcceptedEvent)
+def ingest(
+    envelope: EventEnvelope,
+    request: Request,
+    db: Session = Depends(get_db),
+) -> AcceptedEvent:
+    api_key, source = resolve_event_source(db, request)
+    try:
+        event = accept_event(db, api_key, source.name, envelope)
+    except Deduplicated as dup:
+        return AcceptedEvent(id=str(dup.event_id), status=dup.event_status, deduplicated=True)
+    api_key.last_used_at = datetime.now(UTC)
+    db.commit()
+    _dispatch(str(event.id))
+    return AcceptedEvent(id=str(event.id), status=event.status, deduplicated=False)
+
+
+@router.post("/batch", response_model=BatchResult)
+def ingest_batch(
+    raw_envelopes: list[dict],
+    request: Request,
+    response: Response,
+    db: Session = Depends(get_db),
+) -> BatchResult:
+    """Пачка событий (smart-home, Navi). Частичная валидация честная:
+    ошибки по индексам, валидные приняты; 202, если принято хоть что-то,
+    иначе 422 (откат — транзакцию нельзя коммитить частично).
+    """
+    api_key, source = resolve_event_source(db, request)
+    results: list[AcceptedEvent | BatchRejected] = []
+    accepted: list[str] = []
+
+    for i, raw in enumerate(raw_envelopes):
+        try:
+            envelope = EventEnvelope.model_validate(raw)
+        except ValidationError as err:
+            results.append(BatchRejected(index=i, detail=err.errors()[0].get("msg", "невалидно")))
+            continue
+        try:
+            event = accept_event(db, api_key, source.name, envelope)
+            results.append(AcceptedEvent(id=str(event.id), status=event.status))
+            accepted.append(str(event.id))
+        except Deduplicated as dup:
+            results.append(
+                AcceptedEvent(id=str(dup.event_id), status=dup.event_status, deduplicated=True)
+            )
+        except HTTPException as err:
+            results.append(
+                BatchRejected(
+                    index=i,
+                    detail=err.detail if isinstance(err.detail, str) else "Ошибка приёма",
+                )
+            )
+
+    if not accepted:
+        db.rollback()
+        response.status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
+    else:
+        api_key.last_used_at = datetime.now(UTC)
+        db.commit()
+        response.status_code = status.HTTP_202_ACCEPTED
+        for event_id in accepted:
+            _dispatch(event_id)
+    return BatchResult(results=results)
+
+
+@router.get("/{event_id}", response_model=EventStatus)
+def event_status(
+    event_id: uuid.UUID,
+    request: Request,
+    db: Session = Depends(get_db),
+) -> EventStatus:
+    """Статус события: только его источнику, чужой id — 404."""
+    api_key, source = resolve_event_source(db, request)
+    event = db.execute(
+        select(Event).where(Event.id == event_id, Event.source_id == api_key.source_id)
+    ).scalar_one_or_none()
+    if event is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Событие не найдено")
+
+    deliveries = (
+        db.execute(select(Delivery).where(Delivery.event_id == event.id).order_by(Delivery.id))
+        .scalars()
+        .all()
+    )
+
+    out: list[DeliverySummary] = []
+    for d in deliveries:
+        target_name: str | None = None
+        if d.channel_target_id:
+            target = db.get(ChannelTarget, d.channel_target_id)
+            target_name = target.name if target else None
+        out.append(
+            DeliverySummary(
+                channel=d.channel,
+                target=target_name,
+                status=d.status,
+                attempts=d.attempts,
+                error=d.last_error,
+                rendered_message=d.rendered_message,
+            )
+        )
+
+    return EventStatus(
+        id=str(event.id),
+        source=source.name,
+        subject=event.subject,
+        action=event.action,
+        priority=event.priority,
+        status=event.status,
+        created_at=event.created_at,
+        expires_at=event.expires_at,
+        deliveries=out,
+    )
\ No newline at end of file
diff --git a/app/api/schemas.py b/app/api/schemas.py
new file mode 100644
index 0000000..43b6e8e
--- /dev/null
+++ b/app/api/schemas.py
@@ -0,0 +1,62 @@
+"""Pydantic-схемы Ingestion — конверт из docs/05-ingestion-api.md."""
+
+from datetime import datetime
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+# Нейминг источников/subject/action: slug из нижнего регистра, точки/дефис/подчерк.
+_NAME_PATTERN = r"^[a-z0-9]([a-z0-9._-]*[a-z0-9])?$"
+
+
+class EventEnvelope(BaseModel):
+    """Конверт события. Конверт валидируется жёстко, payload — мягко (воркером)."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    source: str = Field(pattern=_NAME_PATTERN, min_length=1, max_length=64)
+    subject: str = Field(pattern=_NAME_PATTERN, min_length=1, max_length=64)
+    action: str = Field(pattern=_NAME_PATTERN, min_length=1, max_length=64)
+    priority: Literal["low", "normal", "high", "critical"] = "normal"
+    payload: dict = Field(default_factory=dict)
+    dedup_key: str | None = Field(default=None, max_length=255)
+    ttl_seconds: int | None = Field(default=None, ge=1, le=7 * 24 * 3600)
+    scheduled_at: datetime | None = None  # резерв контракта, в MVP не читается
+
+
+class AcceptedEvent(BaseModel):
+    ok: Literal[True] = True
+    id: str
+    status: str
+    deduplicated: bool = False
+
+
+class BatchRejected(BaseModel):
+    ok: Literal[False] = False
+    index: int
+    detail: str
+
+
+class BatchResult(BaseModel):
+    results: list[Annotated[AcceptedEvent | BatchRejected, Field(discriminator="ok")]]
+
+
+class DeliverySummary(BaseModel):
+    channel: str
+    target: str | None
+    status: str
+    attempts: int
+    error: str | None
+    rendered_message: str | None
+
+
+class EventStatus(BaseModel):
+    id: str
+    source: str
+    subject: str
+    action: str
+    priority: str
+    status: str
+    created_at: datetime
+    expires_at: datetime | None
+    deliveries: list[DeliverySummary]
\ No newline at end of file
diff --git a/app/auth/apikeys.py b/app/auth/apikeys.py
new file mode 100644
index 0000000..411961e
--- /dev/null
+++ b/app/auth/apikeys.py
@@ -0,0 +1,60 @@
+"""API-ключи источников для Ingestion (Bearer syn_...).
+
+Ключ — удостоверение источника (docs/05-ingestion-api.md): событие обязано
+нести source, совпадающий с ключом; правила маршрутизации матчают имя
+источника, а не токен, поэтому ротация ключа на правила не влияет.
+"""
+
+import hashlib
+import secrets
+from datetime import UTC, datetime
+
+from fastapi import HTTPException, Request, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models import ApiKey, Source
+
+MISSING_MSG = "Требуется API-ключ: Authorization: Bearer syn_..."
+
+
+def generate_token() -> str:
+    """Токен показывается один раз: `syn_` + 64 hex (32 байта энтропии)."""
+    return f"syn_{secrets.token_hex(32)}"
+
+
+def token_hash_of(token: str) -> str:
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+def resolve_event_source(db: Session, request: Request) -> tuple[ApiKey, Source]:
+    """Bearer-ключ -> (ApiKey, Source) в сессии запроса.
+
+    last_used_at фиксируется транзакцией запроса (коммитит роут).
+    401, если ключа нет/битый/отозван.
+    """
+    auth = request.headers.get("Authorization", "")
+    if not auth.startswith("Bearer "):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=MISSING_MSG,
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    digest = token_hash_of(auth[len("Bearer "):].strip())
+    row = db.execute(
+        select(ApiKey).where(ApiKey.token_hash == digest, ApiKey.revoked_at.is_(None))
+    ).scalar_one_or_none()
+    if row is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="API-ключ неизвестен или отозван",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    source = db.get(Source, row.source_id)
+    if source is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="Источник ключа удалён"
+        )
+    return row, source
\ No newline at end of file
diff --git a/app/cli.py b/app/cli.py
new file mode 100644
index 0000000..bb66a7f
--- /dev/null
+++ b/app/cli.py
@@ -0,0 +1,186 @@
+"""CLI для dev/админ-посева конфигурации (креды/источники/типы/правила).
+
+В проде всё это будет управляться SPA админ-панелью (постройка UI — #32+);
+CLI нужен для локальной разработки и smoke-тестов.
+
+    python -m app.cli create-source monitoring [--label ...]
+    python -m app.cli create-key monitoring [--name ...]   # токен печатается ОДИН раз
+    python -m app.cli add-type monitoring container down
+    python -m app.cli add-target telegram infra --config '{"chat_id": "@infra"}'
+    python -m app.cli add-rule "контейнеры вниз" --source monitoring \
+        --subjects container --actions down --priority high \
+        --template "Контейнер {{ payload.container }} упал" \
+        --action telegram:infra --action internal_log
+"""
+
+import argparse
+import json
+import sys
+
+from sqlalchemy import select
+
+from app.auth.apikeys import generate_token, token_hash_of
+from app.database import SessionLocal
+from app.models import (
+    ApiKey,
+    ChannelTarget,
+    NotificationType,
+    RoutingRule,
+    RoutingRuleAction,
+    Source,
+)
+
+
+def cmd_create_source(args: argparse.Namespace) -> int:
+    with SessionLocal() as db:
+        exists = db.execute(select(Source).where(Source.name == args.name)).scalar_one_or_none()
+        if exists:
+            print(f"источник '{args.name}' уже существует (id={exists.id})")
+            return 1
+        source = Source(name=args.name, label=args.label)
+        db.add(source)
+        db.commit()
+        print(f"ok: source id={source.id} name={source.name}")
+    return 0
+
+
+def cmd_create_key(args: argparse.Namespace) -> int:
+    token = generate_token()
+    with SessionLocal() as db:
+        source = db.execute(select(Source).where(Source.name == args.source)).scalar_one_or_none()
+        if source is None:
+            print(f"источник '{args.source}' не найден (сначала create-source)")
+            return 1
+        key = ApiKey(
+            source_id=source.id,
+            name=args.name or "key",
+            token_hash=token_hash_of(token),
+            token_hint=token[-4:],
+        )
+        db.add(key)
+        db.commit()
+        print(key.token_hint)
+        print(f"ok: key id={key.id} source={args.source}")
+        print(token)  # показывается один раз; хранится только sha256
+    return 0
+
+
+def cmd_add_type(args: argparse.Namespace) -> int:
+    with SessionLocal() as db:
+        source = db.execute(select(Source).where(Source.name == args.source)).scalar_one_or_none()
+        if source is None:
+            print(f"источник '{args.source}' не найден")
+            return 1
+        schema = json.loads(args.schema) if args.schema else None
+        nt = NotificationType(
+            source_id=source.id, subject=args.subject, action=args.action, payload_schema=schema
+        )
+        db.add(nt)
+        db.commit()
+        print(f"ok: type id={nt.id} ({args.source}, {args.subject}, {args.action})")
+    return 0
+
+
+def cmd_add_target(args: argparse.Namespace) -> int:
+    with SessionLocal() as db:
+        target = ChannelTarget(
+            channel=args.channel, name=args.name, config=json.loads(args.config)
+        )
+        db.add(target)
+        db.commit()
+        print(f"ok: target id={target.id} {target.channel}/{target.name}")
+    return 0
+
+
+def cmd_add_rule(args: argparse.Namespace) -> int:
+    with SessionLocal() as db:
+        source_id = None
+        if args.source:
+            source = db.execute(select(Source).where(Source.name == args.source)).scalar_one_or_none()
+            if source is None:
+                print(f"источник '{args.source}' не найден")
+                return 1
+            source_id = source.id
+
+        targets: list[tuple[str, int | None]] = []
+        for spec in args.action:
+            channel, _, target_name = spec.partition(":")
+            target_id = None
+            if target_name:
+                target = db.execute(
+                    select(ChannelTarget).where(ChannelTarget.name == target_name)
+                ).scalar_one_or_none()
+                if target is None:
+                    print(f"цель '{target_name}' не найдена")
+                    return 1
+                target_id = target.id
+            targets.append((channel, target_id))
+
+        rule = RoutingRule(
+            name=args.name,
+            enabled=True,
+            weight=args.weight,
+            conditions={
+                "source": args.source,  # None => любой источник
+                "subjects": args.subjects or [],
+                "actions": args.actions or [],
+                "priority_min": args.priority or None,
+            },
+            template=args.template,
+        )
+        db.add(rule)
+        db.flush()
+        for channel, target_id in targets:
+            db.add(
+                RoutingRuleAction(rule_id=rule.id, channel=channel, target_id=target_id)
+            )
+        db.commit()
+        print(f"ok: rule id={rule.id} '{rule.name}' ({len(targets)} действий)")
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Synapse admin CLI (dev/seed)")
+    sub = parser.add_subparsers(required=True)
+
+    p = sub.add_parser("create-source")
+    p.add_argument("name")
+    p.add_argument("--label")
+    p.set_defaults(func=cmd_create_source)
+
+    p = sub.add_parser("create-key")
+    p.add_argument("source")
+    p.add_argument("--name")
+    p.set_defaults(func=cmd_create_key)
+
+    p = sub.add_parser("add-type")
+    p.add_argument("source")
+    p.add_argument("subject")
+    p.add_argument("action")
+    p.add_argument("--schema")  # JSON-строка схемы payload'а
+    p.set_defaults(func=cmd_add_type)
+
+    p = sub.add_parser("add-target")
+    p.add_argument("channel")
+    p.add_argument("name")
+    p.add_argument("--config", default="{}")
+    p.set_defaults(func=cmd_add_target)
+
+    p = sub.add_parser("add-rule")
+    p.add_argument("name")
+    p.add_argument("--source")
+    p.add_argument("--subjects", nargs="*")
+    p.add_argument("--actions", nargs="*")
+    p.add_argument("--priority")
+    p.add_argument("--template")
+    p.add_argument("--weight", type=int, default=0)
+    p.add_argument("--action", action="append", default=[])  # channel[:target]
+    p.set_defaults(func=cmd_add_rule)
+
+    args = parser.parse_args()
+    assert hasattr(args, "func")
+    return args.func(args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())
\ No newline at end of file
diff --git a/app/config.py b/app/config.py
index 4f3005f..49363ef 100644
--- a/app/config.py
+++ b/app/config.py
@@ -31,9 +31,13 @@
 
     # Публичный URL SPA для редиректа после callback. По умолчанию пусто =
     # тот же origin, что и API (прод: SPA раздаётся этим же контейнером).
-    # Для dev-сервера vite указать SPA_PUBLIC_URL=http://localhost:5173.
+    # Для dev-сервера vite указать SPA_PUBLIC_URL=http://localhost:5174.
     spa_public_url: str = ""
 
+    # Окно дедупликации событий по dedup_key (docs/05): дубль с тем же ключом
+    # от источника внутри окна вернёт id первого события (deduplicated: true).
+    dedup_window_seconds: int = 24 * 3600
+
     # SPA лежит в образе рядом с приложением; в dev используется vite-сервер.
     spa_dist_dir: Path = BASE_DIR / "spa_static"
 
diff --git a/app/main.py b/app/main.py
index 0e32981..2280e35 100644
--- a/app/main.py
+++ b/app/main.py
@@ -5,6 +5,7 @@
 from fastapi.staticfiles import StaticFiles
 
 from fastapi.responses import FileResponse
+from app.api import events_routes
 from app.api.auth_routes import router as auth_router
 from app.api.routes import router as api_router
 from app.config import get_settings
@@ -26,6 +27,7 @@
 
     app.include_router(api_router)
     app.include_router(auth_router)
+    app.include_router(events_routes.router)
 
     # Собранная SPA: в Docker-образе кладётся в spa_static/. Если собранной
     # статики нет (локальный запуск api без фронта) — просто пропускаем.
diff --git a/app/worker/tasks.py b/app/worker/tasks.py
index c03b46d..ca30ba2 100644
--- a/app/worker/tasks.py
+++ b/app/worker/tasks.py
@@ -1,11 +1,134 @@
-"""Задачи воркера. Пока каркасная ping — на её месте будут доставки."""
+"""Задачи воркера: ping + ingest (маршрутизация принятых событий).
+
+Приём (api) кладёт только конверт в БД; здесь Routing Engine MVP:
+первое подходящее enabled-правило по weight → создание Delivery по каждому
+действию. Исполнение каналов (Telegram/Email/S2S) — задачи #29/#28;
+internal_log исполняется сразу (это запись в эту же таблицу).
+"""
+
+import uuid
+from datetime import UTC, datetime
 
 from celery import shared_task
+from jinja2 import Template
+from sqlalchemy import select
 
+from app.database import SessionLocal
+from app.models import (
+    ChannelTarget,
+    Delivery,
+    Event,
+    RoutingRule,
+    RoutingRuleAction,
+    Source,
+)
 from app.worker.celery_app import celery_app
 
+PRIORITY_ORDER = {"low": 0, "normal": 1, "high": 2, "critical": 3}
+
+
+def _rule_matches(rule: RoutingRule, *, source_name: str, subject: str, action: str, priority: str) -> bool:
+    cond = rule.conditions or {}
+    if cond.get("source") is not None and cond["source"] != source_name:
+        return False
+    subjects = cond.get("subjects") or []
+    if subjects and subject not in subjects:
+        return False
+    actions = cond.get("actions") or []
+    if actions and action not in actions:
+        return False
+    min_priority = cond.get("priority_min")
+    if min_priority and PRIORITY_ORDER.get(priority, 0) < PRIORITY_ORDER.get(min_priority, 1):
+        return False
+    return True
+
+
+def _render(template_text: str, event: Event, source_name: str) -> str:
+    """Jinja2-рендер (`{{ payload.x }}`); сломанный шаблон не крушит доставку."""
+    try:
+        return Template(template_text).render(
+            payload=event.payload,
+            id=str(event.id),
+            source=source_name,
+            subject=event.subject,
+            action=event.action,
+            priority=event.priority,
+        )
+    except Exception:  # noqa: BLE001 — шаблон не должен ронять событие
+        return template_text
+
 
 @celery_app.task(name="synapse.ping")
 def ping() -> str:
     """Проверка связки api -> очередь -> воркер."""
-    return "pong"
\ No newline at end of file
+    return "pong"
+
+
+@celery_app.task(name="synapse.ingest")
+def ingest(event_id: str) -> str:
+    """Маршрутизация события: найти правило, создать deliveries.
+
+    Первое подошедшее правило (по weight) применяется целиком; прочие
+    условия/полнота Routing Engine уточняются в #30. Если правил нет —
+    событие просто завершается (конверт остаётся в events как история).
+    """
+    with SessionLocal() as db:
+        event = db.get(Event, uuid.UUID(event_id))
+        if event is None:
+            return f"skip: event {event_id} не найден"
+        if event.status != "queued":
+            return f"skip: event {event_id} в статусе {event.status}"
+
+        source = db.get(Source, event.source_id)
+        event.status = "processing"
+
+        rules = db.execute(
+            select(RoutingRule).where(RoutingRule.enabled.is_(True)).order_by(RoutingRule.weight)
+        ).scalars().all()
+
+        matched: RoutingRule | None = None
+        for rule in rules:
+            if _rule_matches(
+                rule,
+                source_name=source.name if source else "",
+                subject=event.subject,
+                action=event.action,
+                priority=event.priority,
+            ):
+                matched = rule
+                break
+
+        now = datetime.now(UTC)
+        if matched is not None:
+            actions = db.execute(
+                select(RoutingRuleAction).where(RoutingRuleAction.rule_id == matched.id)
+            ).scalars().all()
+            for act in actions:
+                target_name: str | None = None
+                if act.target_id:
+                    target = db.get(ChannelTarget, act.target_id)
+                    target_name = target.name if target else None
+                delivery = Delivery(
+                    event_id=event.id,
+                    rule_id=matched.id,
+                    channel=act.channel,
+                    channel_target_id=act.target_id,
+                    # Фактическая отправка канала — задача #29/#28; пока
+                    # Telegram/Email/S2S лежат pending. internal_log
+                    # исполняется сразу: delivery — и есть запись лога.
+                    status="delivered" if act.channel == "internal_log" else "pending",
+                    attempts=1,
+                    rendered_message=(
+                        _render(act.template or matched.template or "", event, source.name)
+                        if act.channel == "internal_log"
+                        else None
+                    ),
+                    delivered_at=now if act.channel == "internal_log" else None,
+                    last_error=None if act.channel == "internal_log" else "канал ещё не реализован (#29/#28)",
+                )
+                db.add(delivery)
+
+        event.status = "done"
+        db.commit()
+        rule_note = f"rule={matched.id}" if matched else "без правила"
+        return f"event {event_id}: {rule_note}"
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index d6daea7..140bde0 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,4 +9,5 @@
 pydantic-settings>=2.3
 psycopg[binary]>=3.1
 httpx>=0.27
-gnexus-gauth @ git+https://git.gnexus.space/git/root/gnexus-auth-client-py.git
\ No newline at end of file
+gnexus-gauth @ git+https://git.gnexus.space/git/root/gnexus-auth-client-py.git
+jinja2>=3.1