# ADR-001: Sweeper/reconciler потерянных webhook (реконсиляция застрявших стадий)

- **Статус:** Proposed
- **Дата:** 2026-06-06
- **Задача:** ORCH-053
- **Сквозной ADR:** `docs/architecture/adr/adr-0007-reconciler.md`
- **Связи:** adr-0001 (реестр проектов), adr-0002 (очередь / `available_at`),
  adr-0003 (условный staging-гейт — образец условности), adr-0006 (merge-gate как
  под-гейт ребра), ORCH-5 (events de-dup), ORCH-045 (`ci_poll`).

## Контекст

Продвижение задач по конвейеру завязано **исключительно** на входящие webhook
(Plane status / Gitea CI/PR). Потерянное событие (502 на ребилдящемся инстансе,
Plane/Gitea не ретраят, `sha→branch` не разрезолвился) → источник истины (Plane /
зелёный CI) изменился, а задача в оркестраторе застряла молча (живой инцидент
ORCH-044). Ни один существующий механизм resilience (`requeue_running_jobs`,
orphan-recovery, events de-dup, `ci_poll`) не реконсилирует рассинхрон
**«источник истины ≠ стадия задачи»** — все они работают на уровне jobs/agent_runs,
а не stage-перехода.

ТЗ (`02-trz.md`) фиксирует объём; данный ADR фиксирует архитектурные развилки,
явно отданные архитектору: (1) потокобезопасность и подавление спама нотификаций,
(2) способ вызова `async`-обработчиков `plane.py` из sync-потока, (3) atomic-claim
на создании задачи (анти-дубль), (4) критерий «потерян, а не задержан» (grace),
(5) отсутствие изменений схемы БД.

## Решение

### 1. Компонент: `src/reconciler.py` — фоновый daemon-поток

Новый модуль по образцу `queue_worker.py`: класс `Reconciler` + module-singleton
`reconciler`. Plain `threading.Thread(daemon=True)` + `threading.Event` для
остановки. Стартует в `main.lifespan` **после** `worker.start()`, останавливается в
`finally` **перед** `worker.stop()`. Цикл:

```
while not stop:
    try:
        if settings.reconcile_enabled:
            reconcile_gate_once()                       # F-1
            if settings.reconcile_plane_enabled:
                reconcile_plane_once()                  # F-2
    except Exception: log.error(...)                    # outer never-raise
    stop.wait(settings.reconcile_interval_s)
```

`start()` идемпотентен (как `QueueWorker.start`: если поток жив — no-op), что
покрывает AC-11 (повторный start не плодит второй поток). Никакого критичного
состояния в памяти — всё перечитывается из БД/Plane на каждом тике; метрики
наблюдаемости (`last_run_ts`, `unblocked_total`) — best-effort, теряются при
рестарте (AC-11 это явно допускает).

### 2. Источник истины — гейт, не событие. Продвижение строго через `advance_stage`

F-1 НЕ дублирует логику advance. Вводится тонкий хелпер в `stage_engine.py`:

```python
def advance_if_gate_passed(task_id, stage, repo, work_item_id, branch) -> AdvanceResult | None
```

Алгоритм:
1. `stage == "analysis"` → немедленный возврат `None` (см. §6, AC-16).
2. `qg = get_qg_for_stage(stage)`; если `None` (created/done) → возврат `None`.
3. **Read-only пред-оценка гейта** тем же диспетчером, что использует webhook-путь:
   `passed, reason = _run_qg(qg, repo, work_item_id, branch)`.
4. **passed** → вызвать `advance_stage(task_id, stage, repo, work_item_id, branch,
   finished_agent=None)` — это **тот же** путь, которым продвигает Plane
   Approved-webhook (`webhooks/plane._try_advance_stage`). Он повторно прогонит
   гейт (гейты идемпотентны/read-only), продвинет стадию, отправит **штатные**
   advance-нотификации и поставит следующего агента.
5. **not passed** → **тишина**: `logger.debug(...)`, возврат `None`. Никаких
   `notify_qg_failure` / `plane_notify_qg`.

Это даёт оба контракта одновременно:
- **AC-2 / TC-02:** в `reconciler.py` нет собственного `update_task_stage` +
  `enqueue_job` для advance — продвижение исключительно через `advance_stage(...,
  finished_agent=None)`.
- **AC-9 / TC-06:** на застрявшей-но-красной задаче `advance_stage` **не
  вызывается вовсе**, поэтому ветка нотификации провала (`agent is None` →
  `notify_qg_failure`+`plane_notify_qg`, `stage_engine.py:228-230`) не
  срабатывает ни на одном тике. Спам структурно невозможен.

**Подавление спама = «не вызывать advance_stage на красном гейте»**, а не флаг
внутри `advance_stage`. Это сохраняет унифицированный критический путь
(`advance_stage`) **без изменений** — минимальный blast-radius для self-hosting.

> **Цена (осознанная):** на «зелёной» задаче гейт оценивается дважды (пред-оценка
> в хелпере + повтор внутри `advance_stage`). Гейты — чистые read-only проверки
> (`check_ci_green`, `check_*_status` из `12/13/14/15`), на реально-застрявшей-но-
> готовой задаче (целевой кейс ORCH-044) возвращаются быстро (CI уже зелёный →
> `ci_poll` отдаёт результат на первой итерации). Двойная оценка приемлема ради
> неизменности `advance_stage`.

#### Отклонённая альтернатива: флаг `suppress_qg_failure_notify` в `advance_stage`
Однократная оценка гейта, но изменяет сигнатуру и поведение общего
критического пути (риск для self-hosting, обслуживающего все проекты). Отклонено
в пользу неизменности `advance_stage` (Option A выше).

### 3. F-2: вызов `async`-обработчиков `plane.py` из sync-потока

Reconciler — sync daemon-поток; `handle_status_start` / `handle_verdict` —
`async`. Решение: вызывать через **`asyncio.run(coro)`** на каждую единицу работы
внутри per-issue `try/except`. `asyncio.run` создаёт свежий event loop на вызов,
что необходимо, т.к. `handle_verdict → _try_advance_stage` использует
`asyncio.to_thread` (требует running loop). Логику **не дублировать** —
переиспользуются ровно `handle_status_start` / `handle_verdict` /
`list_issues_by_state`.

`issue_data` собирается в форму, ожидаемую обработчиками (`{"id", "state":{"id":..},
"project", "name", "description_stripped"}`); недостающие name/description
обработчики сами дотянут через `fetch_issue_fields` (как для status-only webhook).

### 4. Идемпотентность создания (анти-дубль, AC-4) — atomic-claim в БД

Гонка: F-2 видит `In Progress` + нет задачи; одновременно реальный webhook тоже
видит `In Progress` + нет задачи → оба проходят `get_task_by_plane_id() is None`
→ два `start_pipeline` → два task-row / branch / worktree / стартовых analyst-job
(events de-dup тут НЕ помогает: reconciler — не webhook-доставка).

Решение: **atomic-claim создания, защищённый process-wide `threading.Lock`**.
Новый хелпер `db.create_task_atomic(plane_id, ...)` выполняет
`SELECT-exists → INSERT` под module-level `Lock`, возвращая `(row, created: bool)`:
только победитель (`created=True`) продолжает branch/docs/analyst; проигравший
видит существующую задачу и выходит. `start_pipeline` рефакторится так, чтобы
**первым** DB-действием был этот claim; reconciler идёт тем же путём через
`handle_status_start` → `start_pipeline`.

**Обоснование выбора Lock, а не UNIQUE-индекса:**
- Прод — **один процесс** uvicorn на одну БД (staging/prod изолированы своими БД);
  webhook исполняется в asyncio-треде uvicorn, reconciler — в своём треде того же
  процесса → `threading.Lock` покрывает обе стороны гонки.
- **Без миграции схемы** (соответствует §6 ТЗ и образцу merge-gate ORCH-043).
  `CREATE UNIQUE INDEX` на `tasks.plane_id` рискует упасть на проде, если там уже
  существуют дубли `plane_id` (исторические) — а проверить это вживую нельзя.
- Дешёвый fast-path `get_task_by_plane_id` сохраняется до claim.

> **Граница применимости:** гарантия верна для single-process деплоя (текущая
> топология). Многопроцессный запуск (`uvicorn --workers N`) потребовал бы
> DB-native UNIQUE-индекса — задокументировано как будущее упрочнение в
> `08-data-requirements.md`. Очередь (`queue_worker`) уже опирается на ту же
> single-process-singleton модель, так что допущение не новое.

### 5. Анти-гонка с живым webhook (AC-3) — active-job guard + grace

- **Active-job guard:** `has_active_job_for_task(task.id) == True` → задача
  легитимно в работе или живой webhook только что поставил job → **skip** (ни
  пред-оценки гейта, ни advance, ни нотификаций). И в F-1, и в F-2.
- **Самозатухание повторов:** `advance_stage → update_task_stage` обновляет
  `tasks.updated_at` → следующий тик увидит свежий `updated_at` и не сработает
  повторно (grace).
- `max_concurrency=1`: новый enqueued job встаёт в общую очередь — двойного
  запуска нет (atomic `claim_next_job`).

### 6. F-1 НЕ реконсилирует `analysis` (AC-16)

Гейт `check_analysis_approved` — **человеческий**. В `advance_stage` при
`finished_agent=None` он трактуется как `approved-via-status` и продвинул бы
задачу. Но при потере именно **Approved**-webhka продвигать analysis допустимо
**только** если Plane реально в статусе Approved — этого локальная БД не знает.
Поэтому advance стадии `analysis` отдан **F-2** (сверяется с реальным статусом
Plane). `advance_if_gate_passed` для `stage == "analysis"` — ранний возврат
`None`. Защита от автопродвижения неодобренного человеком BRD.

### 7. Grace: критерий «потерян, а не задержан»

- **F-1:** кандидат, если `has_active_job_for_task == False` **и**
  `age(tasks.updated_at) >= grace_for_stage(stage)`.
  `grace_for_stage(stage)` = per-stage override из `reconcile_grace_overrides_json`,
  иначе `reconcile_grace_default_s`. Невалидный JSON → дефолт (паттерн
  `agent_timeout_overrides_json`, never-raise).
- **F-2:** источник «давности» — поле `updated_at` **issue из Plane** (когда статус
  реально сменился). Реагировать только если `age(issue.updated_at) >=
  reconcile_grace_default_s` — отсекает просто задержавшийся webhook. Для
  существующей задачи дополнительно требуется отсутствие активного job.

### 8. F-3: усиление `sha→branch` в `handle_ci_status`

При неразрезолвленном branch (нет `branches`, `git branch -r --contains` пуст) —
fallback на БД: найти task'и repo со `stage='development'`; при **однозначности**
(ровно одна) использовать её branch; при неоднозначности — оставить
неразрезолвленным + `logger.info`. `logger.debug → logger.info` для видимости.
Success/failure-семантика гейта не меняется. Defense-in-depth: F-1 всё равно
подберёт такую задачу.

### 9. БД и реестры — без изменений

- Схема **не меняется** (§6 ТЗ). Стуковость — по `tasks.updated_at`/`tasks.stage`
  + `has_active_job_for_task`. Анти-дребезг колонкой `last_reconcile_at` **не
  нужен**: на красном гейте действий/нотификаций нет вовсе (§2), а после advance
  `updated_at` обновляется → повтор невозможен.
- `STAGE_TRANSITIONS` и `QG_CHECKS` **не меняются** (AC / TC-23). Новых QG нет.

### 10. Наблюдаемость (F-4)

- При **разблокировке** (произошёл advance) — явная лог-строка
  `reconciler: <work_item_id> <stage> разблокирована (потерян webhook)`; при
  `reconcile_notify_unblock=True` — `send_telegram(...)`. Только на изменении
  состояния, не на каждый тик (AC-12, не конфликтует с AC-9/AC-10).
- `/queue` расширяется блоком `"reconcile": {enabled, plane_enabled, interval,
  last_run_ts, unblocked_total, last_unblocked}` по образцу `worker.status()`.

## Альтернативы (сводно)
- **Флаг подавления нотификаций в `advance_stage`** — отклонён (§2): изменяет общий
  критический путь.
- **UNIQUE-индекс на `tasks.plane_id`** — отклонён как primary (§4): риск падения
  миграции на проде; задокументирован как будущее упрочнение для multi-process.
- **Отдельная стадия/QG для реконсиляции** — вне объёма; нарушило бы «источник
  истины — существующий гейт».
- **Реконсиляция analysis по локальным артефактам** — отклонена (§6): риск
  автопродвижения неодобренного BRD.

## Последствия
- Потерянный webhook больше не = молча застрявшая задача; ручной heartbeat-watchdog
  Стрима не нужен; резервная сетка к ORCH-51/ORCH-36.
- Плата: фоновый поток + периодический опрос Plane API (нагрузка — митигируется
  интервалом + фильтром по статусам + per-project); двойная оценка гейта на зелёной
  задаче; анти-дубль на создании опирается на single-process-допущение.
- Self-hosting: kill-switch `reconcile_enabled` обязателен; reconciler не
  рестартит/не роняет прод-контейнер; раскат поэтапный (флаги).
- Сквозной resilience-механизм → сопровождается global `adr-0007`.
