feat(estimator): task estimation triggered by Plane status «Оценка» (ORCH-020)

Add a deterministic task-estimation side-mechanism: a new operator Plane status «Оценка» (third action-status, family STOP/Confirm Deploy) triggers a new never-raise leaf src/estimator.py that forecasts cost/time/tokens/story points {1,2,3,5,8} from the history of completed tasks (no LLM — ADR-001 D1), writes the forecast to Plane (estimate_point + comment), the Telegram card and the additive task_estimates ledger (UPSERT by work_item_id), then returns the issue to Backlog. On task completion the fact (from usage.py) is written to Plane `point`. Massivity is free (Plane multi-select -> N webhooks); re-estimate is idempotent; anti-disruption skips in-flight tasks; anti-loop (Backlog matches no trigger branch). INVARIANT (NFR-1/NFR-3): the estimator is an observer/producer, never a Quality Gate or stage — STAGE_TRANSITIONS / QG_CHECKS / check_* / machine-verdict keys / existing table schemas and the hot launch path (resolve_agent_model/_spawn) are byte-for-byte untouched. fail-closed: `estimate` absent from _DEFAULT_STATES -> a board without the status is inert (zero regression, enduro untouched). - config: ORCH_ESTIMATOR_* flags (kill-switch + CSV scope empty->self-hosting only, bootstrap defaults, story-point cost thresholds, wall cap). - db: additive task_estimates table + record_estimate/set_actual/get_estimate/ estimates_snapshot + read-only completed_task_stats history aggregate. - plane_sync: «Оценка»->estimate name map (NOT in _DEFAULT_STATES); set_issue_ backlog/set_issue_point/set_issue_estimate_point + get_project_estimate_points (all via ORCH-117 write-guard, best-effort/fail-safe). - webhooks/plane: fail-closed estimate branch + handle_estimate (anti-disruption, auto-return to Backlog, off-loop). - stage_engine: best-effort fact-on-done врезка (after terminal decision). - notifications: «Оценка» card line (read from ledger, omitted when empty). - main: read-only `estimator` GET /queue block + optional POST/GET /estimate. - onboarding canon + Lite/Bundled/ONBOARDING docs: 23rd status «Оценка» (group unstarted, never terminal); anti-drift count pins bumped 22->23. - tests: tests/test_orch020_estimator.py (TC-01..TC-20), full suite green. Refs: ORCH-020 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-17 21:47:00 +03:00
parent 6c204548a7
commit 238de9ba44
19 changed files with 1679 additions and 23 deletions
--- a/.env.example
+++ b/.env.example
@@ -282,6 +282,32 @@ ORCH_STOP_STATUS_REPOS=
 ORCH_BUG_FAST_TRACK_ENABLED=true
 ORCH_BUG_FAST_TRACK_LABEL=Bug
 ORCH_BUG_FAST_TRACK_REPOS=
+# ORCH-020: task-estimation side-mechanism, triggered by the operator Plane status
+# «Оценка» (3rd action-status, family STOP/Confirm Deploy). A leaf src/estimator.py
+# (never-raise) forecasts cost/time/tokens/story-points from the history of completed
+# tasks (deterministic, NO LLM), writes the forecast to Plane (estimate_point + comment),
+# the Telegram card and the additive task_estimates ledger, then returns the issue to
+# Backlog. On completion the fact is written to Plane `point`. OBSERVER/PRODUCER, never a
+# Quality Gate / stage. Infra precondition: create a board status «Оценка» (group
+# backlog/unstarted, NEVER completed/cancelled) + a Points estimate-system 1,2,3,5,8.
+#   ESTIMATOR_ENABLED=false   -> the «Оценка» status is not handled, nothing written
+#                                (1:1 as before ORCH-020, zero regression).
+#   ESTIMATOR_REPOS (CSV)     -> scope; EMPTY = self-hosting only (orchestrator).
+#   ESTIMATOR_MIN_SAMPLES     -> history size below which the bootstrap default blends in.
+#   ESTIMATOR_BOOTSTRAP_*     -> cold-start tokens/cost_usd/seconds when history is empty.
+#   ESTIMATOR_SP_COST_THRESHOLDS -> 4 ascending cost cut-offs (t1,t2,t3,t5) for the
+#                                story-point bucket (<=t1->1 .. <=t5->5, else 8).
+#   ESTIMATOR_WALL_CAP_S      -> cap on anomalous wall-time in history (default 24h).
+#   ESTIMATOR_MAX_INFLIGHT    -> optional bulk-smoothing semaphore (v1 generous/off).
+ORCH_ESTIMATOR_ENABLED=true
+ORCH_ESTIMATOR_REPOS=
+ORCH_ESTIMATOR_MIN_SAMPLES=3
+ORCH_ESTIMATOR_BOOTSTRAP_TOKENS=2000000
+ORCH_ESTIMATOR_BOOTSTRAP_COST_USD=3.0
+ORCH_ESTIMATOR_BOOTSTRAP_SECONDS=1800
+ORCH_ESTIMATOR_SP_COST_THRESHOLDS=0.50,2.00,5.00,12.00
+ORCH_ESTIMATOR_WALL_CAP_S=86400
+ORCH_ESTIMATOR_MAX_INFLIGHT=64
 # ORCH-094: terminal-window-aware guard for the three deploy-phase Plane status
 # setters (set_issue_awaiting_deploy / set_issue_deploying / set_issue_monitoring).
 # A DB stage=done task converges to Done idempotently instead of flapping
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,29 @@
 Формат: [Keep a Changelog](https://keepachangelog.com/). Записи — на смысловой PR/задачу.

 ## [Unreleased]
+- **Оценка задачи, запускаемая Plane-статусом «Оценка»** (ORCH-020, `feat`): новый операторский
+  Plane-статус **«Оценка»** — третий член семейства action-статусов (STOP/Confirm Deploy) — запускает
+  **новый leaf `src/estimator.py`** (never-raise, kill-switch + скоуп), прогнозирующий
+  **стоимость / время / токены / сложность (story points `{1,2,3,5,8}`)** по истории завершённых задач
+  (детерминированная эвристика, **без LLM** — ADR-001 D1). Перевод issue в «Оценка» (в т.ч. **массово**
+  через Plane multi-select → N независимых вебхуков) → `handle_estimate`: прогноз (a) пишется в Plane-поле
+  `estimate_point` (через estimate-систему Fibonacci) + Plane-комментом, (b) добавляется пунктом «Оценка»
+  (время·токены·стоимость·SP) в Telegram-карточку, (c) сохраняется в **новой аддитивной таблице**
+  `task_estimates` (UPSERT по `work_item_id` → идемпотентная пере-оценка), после чего issue **возвращается
+  в Backlog** (анти-loop: Backlog не совпадает ни с одной триггер-веткой). По завершении задачи (переход в
+  `done`) **факт** (из `usage.py`) пишется в Plane-поле `point` (не затирая прогноз). Анти-disruption:
+  issue с активным job не выдёргивается (`should_estimate`). story-points — чистая функция-бакетизатор по
+  стоимости (пороги конфигурируемы). **Инвариант (NFR-1/NFR-3):** оценка — наблюдатель/продюсер, **не**
+  Quality Gate и **не** ребро стадий — `STAGE_TRANSITIONS`/`QG_CHECKS`/`check_*`/machine-verdict-ключи/схемы
+  существующих таблиц байт-в-байт не тронуты; горячий путь `resolve_agent_model`/`_spawn` не трогается.
+  Fail-closed: ключ `estimate` отсутствует в `_DEFAULT_STATES` → доска без статуса → ветка инертна
+  (зеркало STOP/Confirm Deploy). Опц. эндпоинты `POST /estimate`/`GET /estimate` (то же ядро) + read-only
+  блок `estimator` в `GET /queue`. Флаги `ORCH_ESTIMATOR_*` (`enabled` kill-switch, `repos` CSV — **пусто →
+  self-hosting only**, bootstrap-дефолты, пороги story-points). Откат = `ORCH_ESTIMATOR_ENABLED=false` →
+  модуль инертен (нулевая регрессия; enduro не затронут). Онбординг-канон расширен 23-м статусом «Оценка»
+  (группа `unstarted`, НЕ терминальная). Покрытие — `tests/test_orch020_estimator.py` (TC-01…TC-20). Детали —
+  `docs/work-items/ORCH-020/06-adr/ADR-001-task-estimation-status-trigger.md`, сквозной
+  `docs/architecture/adr/adr-0054-task-estimation-status-trigger.md`.
 - **FAQ по статусу STOP для пользователя доски** (ORCH-108, `docs`): создан пользовательский
  FAQ `docs/operations/FAQ_STOP.md` в формате «вопрос → ответ» — что делает STOP, как отменить
  задачу, что происходит пошагово (агент останавливается → job'ы снимаются → рабочая ветка и
--- a/README.md
+++ b/README.md
@@ -149,6 +149,12 @@ uvicorn src.main:app --reload --port 8500
 | `ORCH_BUG_FAST_TRACK_ENABLED` | Kill-switch багфикс-трека (ORCH-019): задача с меткой Plane `Bug` пропускает стадию `architecture`; `false` → старт и маршрут 1:1 как до ORCH-019 (нулевая регрессия) | `true` |
 | `ORCH_BUG_FAST_TRACK_LABEL` | Имя метки Plane, активирующей багфикс-трек (ORCH-019) | `Bug` |
 | `ORCH_BUG_FAST_TRACK_REPOS` | CSV область репо для багфикс-трека; **пусто → self-hosting only** (`orchestrator`) — enduro подключается явным CSV (ORCH-019) | `""` |
+| `ORCH_ESTIMATOR_ENABLED` | Kill-switch оценки задачи (ORCH-020): Plane-статус «Оценка» прогнозирует стоимость/время/токены/story-points по истории; `false` → статус не обрабатывается, ничего не пишется (нулевая регрессия) | `true` |
+| `ORCH_ESTIMATOR_REPOS` | CSV область репо для оценки; **пусто → self-hosting only** (`orchestrator`) — enduro не затронут (ORCH-020) | `""` |
+| `ORCH_ESTIMATOR_MIN_SAMPLES` | Порог истории, ниже которого подмешивается bootstrap-дефолт прогноза (ORCH-020) | `3` |
+| `ORCH_ESTIMATOR_BOOTSTRAP_TOKENS` / `_COST_USD` / `_SECONDS` | Bootstrap-прогноз при пустой истории (токены/стоимость/время; ORCH-020) | `2000000`/`3.0`/`1800` |
+| `ORCH_ESTIMATOR_SP_COST_THRESHOLDS` | 4 возрастающих кат-оффа стоимости (t1,t2,t3,t5) для бакета story-points (`<=t1→1`…`<=t5→5`, иначе `8`; ORCH-020) | `0.50,2.00,5.00,12.00` |
+| `ORCH_ESTIMATOR_WALL_CAP_S` / `_MAX_INFLIGHT` | Отсечка аномального wall-времени в истории / опц. семафор сглаживания массовой нагрузки (ORCH-020) | `86400`/`64` |
 | `ORCH_AGENT_HOME_DIR` | ORCH-101: HOME акторских процессов + таргет маунтов `.claude`/`.ssh` + `ARG APP_HOME` (группа ORCH-040) | `/home/slin` |
 | `ORCH_AGENT_GIT_NAME` / `ORCH_GIT_EMAIL_DOMAIN` | ORCH-101: git-идентичность коммитов агентов (`claude-bot@mva154.local` при дефолтах) | `claude-bot` / `mva154.local` |
 | `ORCH_STAGING_PORT` | ORCH-101: порт staging (читают `image_freshness` и compose); guard fail-closed при совпадении с прод-портом (ORCH-058 AC-9) | `8501` |
--- a/docs/architecture/README.md
+++ b/docs/architecture/README.md
@@ -289,7 +289,7 @@ reviewer-ось обзорных доков (ORCH-079) расширена на
 вводится, рантайм байт-в-байт. Подробнее: [adr-0039](adr/adr-0039-system-overview-docs-canon.md),
 детально — `docs/work-items/ORCH-011/06-adr/ADR-001-system-overview-canon.md`.

-## Оценка задачи (ORCH-020 — design)
+## Оценка задачи (ORCH-020 — реализовано)

 Операторская способность прогнозировать **стоимость / время / токены / сложность (story points
 `{1,2,3,5,8}`)** задачи **до отправки её в работу**, чтобы планировать бэклог. Запуск — **операторский
@@ -329,7 +329,9 @@ leaf-модулем `src/estimator.py` + точечными врезками; **
 - **Флаги/наблюдаемость:** `estimator_enabled` (kill-switch) + `estimator_repos` (CSV; **пусто →
  self-hosting only**) + тюнинг (`estimator_min_samples`/bootstrap-дефолты/`estimator_sp_cost_thresholds`/
  `estimator_wall_cap_s`/`estimator_max_inflight`); read-only блок `estimator` в `GET /queue`; опц.
-  `POST /estimate`, `POST /estimate/backlog`, `GET /estimate` (то же ядро, не основной триггер).
+  `POST /estimate?work_item=<id>` (то же ядро: UPSERT + estimate_point + коммент + карточка + возврат в
+  Backlog) и `GET /estimate?work_item=<id>` (прогноз vs факт) — удобство/диагностика, не основной триггер
+  (массовый путь — Plane multi-select; `POST /estimate/backlog` в этот срез не вошёл).
 - **Инфра-предусловия (NFR-7):** статус «Оценка» (группа `backlog`/`unstarted`, **не** терминальная) +
  estimate-система Plane типа Points с Fibonacci `1/2/3/5/8`; онбординг ORCH-009 расширяется на 23-й статус.

@@ -1522,6 +1524,7 @@ Monitoring after Deploy → Done
 - `job_deps` — декларативные зависимости задач (ORCH-026, Уровень B): `(task_id, depends_on_task_id)`, аддитивная; источник истины планировщика для гейта «B ждёт A»
 - `repo_freeze` — durable per-repo rollback-freeze (ORCH-088, FR-5): `(id, repo, frozen_at, reason, work_item_id, cleared_at)`, аддитивная append-only; активный freeze ⇔ строка репо с `cleared_at IS NULL`. Выставляется post-deploy `DEGRADED` (`set_repo_freeze`), снимается вручную (`POST /serial-gate/unfreeze` → `cleared_at=now`). Гейтит serial-claim безусловно (деградировавшая задача уже `done`)
 - `lessons` — машинный журнал отклонений конвейера (ORCH-098, FR-1): `(id, created_at, updated_at, lesson_type, work_item_id, task_id, stage, agent, repo, root_cause, suggestion, status, related_task, attribution, target_repo, target_domain, source, detail)`, аддитивная идемпотентная (`CREATE TABLE IF NOT EXISTS` + три индекса); колонки атрибуции (`attribution`/`target_repo`/`target_domain`) — нуллабельны и присутствуют сразу (NFR-6), без `enum`-констрейнтов (слаги forward-compatible). Автозапись 4 типов (`gate_failure`/`merge_hold`/`transient_retry`/`deploy_degraded`, `source="auto"`, дедуп в окне `lessons_dedup_window_s`) + ручная (`source="manual"`); observer-only (не участвует в решении гейта). Leaf `src/lessons.py` never-raise, kill-switch `lessons_enabled` (без `*_repos` — журнал не скоупится по репо, репо-разрез на выборке)
+- `task_estimates` — леджер прогноз↔факт оценки задачи (ORCH-020, FR-T9): `(id, work_item_id UNIQUE, task_id, repo, forecast_tokens/seconds/cost_usd/story_points, actual_tokens/seconds/cost_usd/story_points, source, estimate_count, created_at, updated_at)`, аддитивная идемпотентная (`CREATE TABLE/INDEX IF NOT EXISTS`, паттерн `coverage_baseline`/`lessons`). Ключ — `work_item_id` (issue может не иметь `task` на момент оценки — бэклог; `task_id` нуллабелен, заполняется при старте); `UNIQUE(work_item_id)` → идемпотентная пере-оценка UPSERT (`estimate_count` инкрементится). Дельта прогноз↔факт вычисляется на чтение (не хранится). Пишет **только** self-hosting-скоуп (`estimator_repos` пуст → orchestrator; enduro не затронут). Leaf `src/estimator.py` never-raise; observer/producer — схемы существующих таблиц не тронуты
 - `transition_lease` — durable-владение side-effectful переходом стадии (ORCH-114, FR-1): `(task_id PK, owner, owner_pid, owner_boot_id, run_id, stage, acquired_at)`, аддитивная идемпотентная (`CREATE TABLE IF NOT EXISTS`, паттерн `repo_freeze`/`coverage_baseline`). Активная строка ⇔ актор держит владение переходом задачи; **живой** владелец ⇔ `owner_boot_id == <boot-id процесса>` И `pid_alive(owner_pid)` (рестарт ⇒ новый boot-id ⇒ прежние lease мертвы → реклейм). Захват — атомарный rowcount-guard (паттерн `claim_next_job`/`reap_running_job`); release в `try/finally`; потолок возраста = Tier-3 `reaper_max_running_s` (без собственного TTL — NFR-6). Парная CAS-запись стадии — `update_task_stage_cas(task_id, expected_stage, new_stage)` (`UPDATE … WHERE id=? AND stage=?`). Leaf `src/transition_lease.py` never-raise, kill-switch `transition_lease_enabled` + `transition_lease_repos` (пусто → self-hosting only). Обобщает in-memory `finalizer_liveness` (ORCH-113) до durable cross-path; схемы существующих таблиц не тронуты

 ## Изоляция (git worktree, ORCH-2)
@@ -1532,7 +1535,7 @@ Monitoring after Deploy → Done
 |--------|------|----------|
 | GET | `/health` | health check |
 | GET | `/status` | активные задачи (stage != done) |
-| GET | `/queue` | очередь: counts + max_concurrency + resilience + reconcile (ORCH-053) + reaper (ORCH-065) + post_deploy (ORCH-021) + task_deps (ORCH-026) + serial_gate (ORCH-088) + auto_labels (ORCH-089) + stop (ORCH-090) + lessons (ORCH-098) + transition_lease (ORCH-114) + последние jobs |
+| GET | `/queue` | очередь: counts + max_concurrency + resilience + reconcile (ORCH-053) + reaper (ORCH-065) + post_deploy (ORCH-021) + task_deps (ORCH-026) + serial_gate (ORCH-088) + auto_labels (ORCH-089) + stop (ORCH-090) + estimator (ORCH-020) + lessons (ORCH-098) + transition_lease (ORCH-114) + последние jobs |
 | GET | `/metrics` | ORCH-099 (FND/F1a): read-only машинное «сырьё» для sidecar F1b — конверт `schema_version`/`generated_at`/`clk_tck` + разделы `stages`/`queue`/`agents` (liveness: pid/runtime/cpu_ticks)/`cost`. never-raise по разделам; kill-switch `ORCH_METRICS_ENABLED` (дефолт `True`). Контракт — см. раздел «Сырьё-эндпоинт `/metrics`» |
 | POST | `/serial-gate/unfreeze` | ORCH-088 (FR-5): ручное снятие per-repo rollback-freeze (query/body `repo=<repo>`) → `{ok, repo, cleared, frozen}`; идемпотентно. Альтернатива — `UPDATE repo_freeze SET cleared_at=datetime('now') WHERE repo=? AND cleared_at IS NULL` |
 | POST | `/serial-gate/pause` | ORCH-124 (D7): поставить задачу на паузу для serial-gate (query/body `work_item=<id>`) → `{ok, work_item, task_id, paused_at}`; идемпотентно. Паузнутый предшественник не держит FIFO против срочного успешника (пауза ≠ cancel, ≠ глобальный kill-switch); НЕ обходит `repo_freeze`/`task_deps` |
@@ -1541,6 +1544,8 @@ Monitoring after Deploy → Done
 | GET | `/lessons` | ORCH-098 (FR-4): read-only выборка журнала уроков; query-фильтры `type`/`status`/`repo`/`work_item`/`limit` → `{enabled, lessons:[…]}` (всегда `200`, чтение не мутирует). При `lessons_enabled=False` → `{enabled:false, lessons:[]}` |
 | POST | `/lessons` | ORCH-098 (FR-5): ручная запись урока (JSON-тело, `lesson_type` обязателен, `source="manual"` не дедупится) → `{id}`; при выключенном флаге → `{enabled:false}` |
 | POST | `/lessons/{id}` | ORCH-098 (FR-5): доклассификация/обновление урока (`status`/`attribution`/`target_*`/`related_task`/`root_cause`/`suggestion`), стампит `updated_at` → `{ok}` |
+| POST | `/estimate` | ORCH-020 (**опц.**): произвести/обновить прогноз одной задачи (query `work_item=<id>`) — то же ядро, что статус-триггер «Оценка» (UPSERT в `task_estimates` + `estimate_point` + коммент + карточка + возврат в Backlog), идемпотентно → `{ok, work_item, forecast}`; при выключенном/вне-скоупе флаге → `{enabled:false}`. Не основной путь (основной — статус «Оценка») |
+| GET | `/estimate` | ORCH-020 (**опц.**): прочитать прогноз vs факт из `task_estimates` (query `work_item=<id>`) → `{ok, work_item, estimate}` (или `estimate:null`) |
 | POST | `/webhook/plane` | Plane webhook |
 | POST | `/webhook/gitea` | Gitea webhook (push, PR, CI status) |

--- a/docs/deployment/BUNDLED_SETUP.md
+++ b/docs/deployment/BUNDLED_SETUP.md
@@ -179,8 +179,8 @@ docker compose -f deploy/bundled/docker-compose.yml ps

 Доводка «одним запуском»: preflight → секреты → up/готовность → init Gitea
 (полностью автоматом: админ-бот + API-токен) → init Plane → онбординг
-sandbox-проекта **строго** кирпичом `onboard_project.py` (22 канонических
-статуса, включая fail-closed **`Confirm Deploy`** и **`STOP`**, лейблы,
+sandbox-проекта **строго** кирпичом `onboard_project.py` (23 канонических
+статуса, включая fail-closed **`Confirm Deploy`**, **`STOP`** и **`Оценка`**, лейблы,
 репо+webhook — golden source `docs/operations/ONBOARDING.md` §1) → git-доступ
 агентов → сборка `.env`/`.env.watchdog` → health.

--- a/docs/deployment/LITE_SETUP.md
+++ b/docs/deployment/LITE_SETUP.md
@@ -206,12 +206,13 @@ curl -fsS "$ORCH_PLANE_API_URL/api/v1/workspaces/<workspace-slug>/projects/" \
 `ORCH_PLANE_API_TOKEN` в `.env`. Токен должен иметь право создавать проекты/статусы
 (нужно для `onboard_project.py apply`, §10).

-**5.3. Модель статусов — НЕ вручную.** Конвейеру нужны **22 канонических статуса** с
+**5.3. Модель статусов — НЕ вручную.** Конвейеру нужны **23 канонических статуса** с
 точными именами и группами; их создаёт `python3 scripts/onboard_project.py apply` (§10),
 полная таблица — `docs/operations/ONBOARDING.md` §1 (golden source; здесь не дублируется).
-Два имени фиксируем явно, потому что они **fail-closed** (без них ветка просто не
-активируется, без ошибки): **`Confirm Deploy`** (человеческий гейт прод-деплоя) и
-**`STOP`** (отмена задачи; обязан быть в группе `cancelled`).
+Три имени фиксируем явно, потому что они **fail-closed** (без них ветка просто не
+активируется, без ошибки): **`Confirm Deploy`** (человеческий гейт прод-деплоя),
+**`STOP`** (отмена задачи; обязан быть в группе `cancelled`) и **`Оценка`** (триггер
+оценки задачи, ORCH-020; группа `unstarted`/`backlog`, НЕ терминальная).

 ```bash
 # после §10 — проверить, что статусы созданы:
@@ -219,7 +220,7 @@ curl -fsS "$ORCH_PLANE_API_URL/api/v1/workspaces/<workspace-slug>/projects/<proj
  -H "X-API-Key: $ORCH_PLANE_API_TOKEN" | python3 -m json.tool | grep -c '"name"'
 ```

-**Проверка:** счётчик имён = 22 (или больше, если в проекте остались дефолтные статусы
+**Проверка:** счётчик имён = 23 (или больше, если в проекте остались дефолтные статусы
 Plane) и среди них `Confirm Deploy` и `STOP` — PASS.

 **5.4. Webhook + HMAC.** Приёмник — `POST https://<orchestrator-public-host>/webhook/plane`;
@@ -432,7 +433,7 @@ curl -fsS http://127.0.0.1:8501/health

 ## 10. Регистрация проекта заказчика

-Onboarding-CLI создаёт Plane-проект с 22 статусами и лейблами (`autoApprove` /
+Onboarding-CLI создаёт Plane-проект с 23 статусами и лейблами (`autoApprove` /
 `autoDeploy` / `Bug`), Gitea-репо с webhook'ом, скелет репо (kit) и печатает merged-реестр.
 Полный runbook — `docs/operations/ONBOARDING.md`; Lite-последовательность:

--- a/docs/operations/ONBOARDING.md
+++ b/docs/operations/ONBOARDING.md
@@ -46,7 +46,7 @@ hooks под выбранным owner (`--gitea-owner`, дефолт из кон

 1. **Проект**: создаётся с `identifier = --prefix`. Уже существует → передай
   `--plane-project-id <uuid>` (ensure распознает и пропустит).
-2. **Статусы — точные канонические имена** (22, источник — `plane_sync._PLANE_NAME_TO_KEY`;
+2. **Статусы — точные канонические имена** (23, источник — `plane_sync._PLANE_NAME_TO_KEY`;
   опечатка = тихая деградация fail-closed веток):

   | Статус | Группа | | Статус | Группа |
@@ -62,9 +62,11 @@ hooks под выбранным owner (`--gitea-owner`, дефолт из кон
   | Review | `started` | | **STOP** | **`cancelled`** |
   | Testing | `started` | | Awaiting Deploy | `started` |
   | Deploying | `started` | | Monitoring after Deploy | `started` |
+   | **Оценка** | **`unstarted`** | | | |

   ⚠️ Код-критично: `STOP` обязан быть в группе `cancelled` (иначе ветка отмены молча не
-   активируется); в терминальных группах (`completed`/`cancelled`) — ТОЛЬКО
+   активируется); **`Оценка`** (триггер оценки задачи, ORCH-020) — группа `unstarted`/`backlog`,
+   **НЕ** терминальная; в терминальных группах (`completed`/`cancelled`) — ТОЛЬКО
   Done/Cancelled/STOP, иначе terminal-detection ложно сочтёт живую задачу терминальной.
 3. **Лейблы**: `autoApprove`, `autoDeploy`, `Bug` (имена — из конфига оркестратора; их
   отсутствие = fail-safe ручной режим / полный цикл).
@@ -149,7 +151,7 @@ hooks под выбранным owner (`--gitea-owner`, дефолт из кон
     --webhook-url https://openclaw.mva154.duckdns.org/orchestrator/webhook/gitea
   ```

-   Проверяет: запись реестра парсится и совпадает по полям; все 22 статуса резолвятся
+   Проверяет: запись реестра парсится и совпадает по полям; все 23 статуса резолвятся
   (включая fail-closed `Confirm Deploy`/`STOP`); лейблы на месте; webhook существует и
   активен; kit-файлы в репо (6 промптов, `AGENTS.md`, `INFRA.md`, `_templates`/`_standards`);
   нет неразрешённых плейсхолдеров. Любой gap → exit `2` с перечнем.
--- a/scripts/onboard_project.py
+++ b/scripts/onboard_project.py
@@ -82,6 +82,10 @@ STATE_GROUPS: dict[str, str] = {
    "Backlog": "backlog",
    "Todo": "unstarted",
    "To Analyse": "unstarted",
+    # ORCH-020: «Оценка» — transient backlog-side estimation trigger. Group MUST be
+    # non-terminal (unstarted/backlog), NEVER completed/cancelled — otherwise the
+    # ORCH-068 terminal-detection would falsely terminate the issue mid-estimate.
+    "Оценка": "unstarted",
    "In Progress": "started",
    "Analysis": "started",
    "Architecture": "started",
@@ -934,7 +938,7 @@ def run_verify(
        report.add("verify.registry", "запись реестра ORCH_PROJECTS_JSON", OK,
                   f"prefix={entry.work_item_prefix}, repo={entry.repo}")

-    # 2. Статусы: все 22 канонических имени (включая fail-closed Confirm Deploy/STOP).
+    # 2. Статусы: все 23 канонических имени (включая fail-closed Confirm Deploy/STOP/Оценка).
    if observed.project is None or observed.states is None:
        report.add("verify.plane.states", "статусы Plane-проекта", GAP,
                   "проект/статусы не читаются через API")
--- a/src/config.py
+++ b/src/config.py
@@ -1196,6 +1196,51 @@ class Settings(BaseSettings):
    bug_fast_track_label: str = "Bug"
    bug_fast_track_repos: str = ""

+    # ORCH-020: task-estimation side-mechanism, triggered by the operator Plane status
+    # «Оценка» (third member of the action-status family STOP/Confirm Deploy). A leaf
+    # src/estimator.py (never-raise) forecasts cost/time/tokens/story-points from the
+    # history of completed tasks (deterministic heuristic, NO LLM — ADR-001 D1),
+    # writes the forecast to Plane (estimate_point + comment), the Telegram card and
+    # the additive task_estimates ledger (UPSERT by work_item_id), then returns the
+    # issue to Backlog. On task completion (-> done) the fact is written to Plane
+    # `point`. The estimator is an OBSERVER/PRODUCER, never a Quality Gate / stage:
+    # STAGE_TRANSITIONS / QG_CHECKS / check_* / verdict-keys / existing schemas are NOT
+    # touched. Pattern = bug_fast_track_* / coverage_gate_* (kill-switch + CSV scope,
+    # applies() local & FIRST). See docs/work-items/ORCH-020/06-adr/ADR-001-task-
+    # estimation-status-trigger.md and docs/architecture/adr/adr-0054-task-estimation-
+    # status-trigger.md.
+    #   estimator_enabled           -> SINGLE kill-switch (env ORCH_ESTIMATOR_ENABLED).
+    #                                  False -> the «Оценка» status is not handled, no
+    #                                  Plane/card/table write — pipeline 1:1 as before
+    #                                  ORCH-020, zero regression (AC-9).
+    #   estimator_repos             -> CSV scope (env ORCH_ESTIMATOR_REPOS). Empty ->
+    #                                  self-hosting only (orchestrator) via
+    #                                  is_self_hosting_repo; non-empty -> membership.
+    #                                  Mirrors coverage_gate_repos -> enduro untouched.
+    #   estimator_min_samples       -> history size below which the bootstrap default
+    #                                  blends in (D2). Default 3.
+    #   estimator_bootstrap_tokens  -> cold-start token forecast when history is empty.
+    #   estimator_bootstrap_cost_usd-> cold-start cost ($) forecast (drives the
+    #                                  bootstrap story-point bucket: 3.0 -> «3» medium).
+    #   estimator_bootstrap_seconds -> cold-start wall-time forecast.
+    #   estimator_sp_cost_thresholds-> CSV of 4 ascending cost cut-offs (t1,t2,t3,t5),
+    #                                  `<=` ascending: <=t1->1, <=t2->2, <=t3->3,
+    #                                  <=t5->5, else 8 (D3). Re-calibratable on ORCH-13.
+    #   estimator_wall_cap_s        -> cap on anomalous wall-time in history (a task
+    #                                  parked in backlog for days would skew the mean);
+    #                                  mirror of tracker_brd_review_cap_s. Default 24h.
+    #   estimator_max_inflight      -> optional smoothing semaphore for bulk multi-select
+    #                                  (v1 generous/off; not enforced).
+    estimator_enabled: bool = True
+    estimator_repos: str = ""
+    estimator_min_samples: int = 3
+    estimator_bootstrap_tokens: int = 2_000_000
+    estimator_bootstrap_cost_usd: float = 3.0
+    estimator_bootstrap_seconds: int = 1800
+    estimator_sp_cost_thresholds: str = "0.50,2.00,5.00,12.00"
+    estimator_wall_cap_s: int = 86400
+    estimator_max_inflight: int = 64
+
    # Telegram notifications
    telegram_bot_token: str = ""
    telegram_chat_id: str = ""
--- a/src/db.py
+++ b/src/db.py
@@ -306,6 +306,39 @@ def init_db():
            acquired_at   TEXT NOT NULL DEFAULT (datetime('now'))
        );
    """)
+    # ORCH-020 (adr-0054 / 08-data-requirements.md): additive task-estimation ledger
+    # (forecast<->fact, keyed by work_item_id because an issue may have NO task row at
+    # estimate time — it sits in the backlog). ONE additive object (CREATE TABLE/INDEX
+    # IF NOT EXISTS, pattern coverage_baseline/lessons/transition_lease) -> idempotent,
+    # restart-safe on the shared prod DB; existing tables (tasks/agent_runs/jobs/...)
+    # untouched byte-for-byte (NFR-8). UNIQUE(work_item_id) is the carrying invariant
+    # of idempotent re-estimation (BR-T4/AC-T4): a repeat «Оценка» UPSERTs the one row
+    # and bumps estimate_count; no duplicate rows. The forecast<->fact delta is computed
+    # on read (not a stored column) to avoid drift. The estimator leaf wraps every call
+    # below in its never-raise contract. enduro-trails is not affected — the table is
+    # shared but only the self-hosting scope writes to it.
+    conn.executescript("""
+        CREATE TABLE IF NOT EXISTS task_estimates (
+            id                    INTEGER PRIMARY KEY AUTOINCREMENT,
+            work_item_id          TEXT NOT NULL UNIQUE,
+            task_id               INTEGER,
+            repo                  TEXT,
+            forecast_tokens       INTEGER,
+            forecast_seconds      INTEGER,
+            forecast_cost_usd     REAL,
+            forecast_story_points INTEGER,
+            actual_tokens         INTEGER,
+            actual_seconds        INTEGER,
+            actual_cost_usd       REAL,
+            actual_story_points   INTEGER,
+            source                TEXT,
+            estimate_count        INTEGER NOT NULL DEFAULT 1,
+            created_at            TEXT NOT NULL DEFAULT (datetime('now')),
+            updated_at            TEXT
+        );
+        CREATE INDEX IF NOT EXISTS idx_task_estimates_repo    ON task_estimates (repo);
+        CREATE INDEX IF NOT EXISTS idx_task_estimates_task_id ON task_estimates (task_id);
+    """)
    conn.commit()
    conn.close()

@@ -566,6 +599,204 @@ def all_coverage_baselines() -> dict:
    }


+# ---------------------------------------------------------------------------
+# ORCH-020 (adr-0054, ADR-001 D7): task-estimation ledger helpers. Each opens its
+# own connection and closes it in `finally` (pattern coverage_baseline/lessons). The
+# leaf src/estimator.py wraps these in its never-raise contract — they may raise on a
+# real DB fault (the leaf swallows it).
+# ---------------------------------------------------------------------------
+def record_estimate(
+    work_item_id: str,
+    repo: str = None,
+    task_id: int = None,
+    forecast_tokens: int = None,
+    forecast_seconds: int = None,
+    forecast_cost_usd: float = None,
+    forecast_story_points: int = None,
+    source: str = "status",
+) -> int:
+    """UPSERT a forecast by ``work_item_id`` (BR-T4/AC-T4 idempotent re-estimation).
+
+    On first insert ``estimate_count`` is 1; a repeat (same work_item_id) updates the
+    forecast in place, bumps ``estimate_count`` and stamps ``updated_at`` — no
+    duplicate rows. ``task_id`` is COALESCEd so a later start-of-pipeline link is not
+    wiped by a re-estimate that still sees a NULL task. Returns the row id.
+    """
+    if not work_item_id:
+        raise ValueError("record_estimate requires work_item_id")
+    conn = get_db()
+    try:
+        conn.execute(
+            "INSERT INTO task_estimates "
+            "  (work_item_id, repo, task_id, forecast_tokens, forecast_seconds, "
+            "   forecast_cost_usd, forecast_story_points, source, estimate_count, updated_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, datetime('now')) "
+            "ON CONFLICT(work_item_id) DO UPDATE SET "
+            "  repo = COALESCE(excluded.repo, task_estimates.repo), "
+            "  task_id = COALESCE(excluded.task_id, task_estimates.task_id), "
+            "  forecast_tokens = excluded.forecast_tokens, "
+            "  forecast_seconds = excluded.forecast_seconds, "
+            "  forecast_cost_usd = excluded.forecast_cost_usd, "
+            "  forecast_story_points = excluded.forecast_story_points, "
+            "  source = excluded.source, "
+            "  estimate_count = task_estimates.estimate_count + 1, "
+            "  updated_at = datetime('now')",
+            (
+                work_item_id, repo, task_id, forecast_tokens, forecast_seconds,
+                forecast_cost_usd, forecast_story_points, source,
+            ),
+        )
+        conn.commit()
+        row = conn.execute(
+            "SELECT id FROM task_estimates WHERE work_item_id = ?", (work_item_id,)
+        ).fetchone()
+        return int(row["id"]) if row else 0
+    finally:
+        conn.close()
+
+
+def set_actual(
+    work_item_id: str,
+    actual_tokens: int = None,
+    actual_seconds: int = None,
+    actual_cost_usd: float = None,
+    actual_story_points: int = None,
+    task_id: int = None,
+) -> bool:
+    """Write the FACT half of the ledger (on task completion). Never touches the
+    forecast columns (AC-4: the fact must not overwrite the prediction). Returns True
+    iff a row was updated/inserted.
+
+    If no forecast row exists yet (an issue completed without ever being estimated) a
+    fact-only row is inserted so the ledger still records the actuals.
+    """
+    if not work_item_id:
+        return False
+    conn = get_db()
+    try:
+        cur = conn.execute(
+            "UPDATE task_estimates SET "
+            "  actual_tokens = ?, actual_seconds = ?, actual_cost_usd = ?, "
+            "  actual_story_points = ?, "
+            "  task_id = COALESCE(?, task_id), updated_at = datetime('now') "
+            "WHERE work_item_id = ?",
+            (
+                actual_tokens, actual_seconds, actual_cost_usd, actual_story_points,
+                task_id, work_item_id,
+            ),
+        )
+        changed = cur.rowcount or 0
+        if changed == 0:
+            conn.execute(
+                "INSERT INTO task_estimates "
+                "  (work_item_id, task_id, actual_tokens, actual_seconds, "
+                "   actual_cost_usd, actual_story_points, source, updated_at) "
+                "VALUES (?, ?, ?, ?, ?, ?, 'fact-only', datetime('now'))",
+                (
+                    work_item_id, task_id, actual_tokens, actual_seconds,
+                    actual_cost_usd, actual_story_points,
+                ),
+            )
+            changed = 1
+        conn.commit()
+        return bool(changed)
+    finally:
+        conn.close()
+
+
+def get_estimate(work_item_id: str) -> dict | None:
+    """Return the current forecast/fact row for ``work_item_id`` (None when absent)."""
+    if not work_item_id:
+        return None
+    conn = get_db()
+    try:
+        row = conn.execute(
+            "SELECT * FROM task_estimates WHERE work_item_id = ?", (work_item_id,)
+        ).fetchone()
+    finally:
+        conn.close()
+    return dict(row) if row else None
+
+
+def estimates_snapshot(limit: int = 10) -> dict:
+    """Light summary (totals + last N rows) for the GET /queue estimator block."""
+    try:
+        lim = max(1, int(limit))
+    except (TypeError, ValueError):
+        lim = 10
+    conn = get_db()
+    try:
+        total = conn.execute("SELECT COUNT(*) FROM task_estimates").fetchone()[0]
+        with_actual = conn.execute(
+            "SELECT COUNT(*) FROM task_estimates WHERE actual_story_points IS NOT NULL"
+        ).fetchone()[0]
+        rows = conn.execute(
+            "SELECT work_item_id, repo, forecast_story_points, actual_story_points, "
+            "forecast_cost_usd, actual_cost_usd, estimate_count, updated_at "
+            "FROM task_estimates ORDER BY id DESC LIMIT ?",
+            (lim,),
+        ).fetchall()
+    finally:
+        conn.close()
+    return {
+        "total": total,
+        "with_actual": with_actual,
+        "recent": [dict(r) for r in rows],
+    }
+
+
+def completed_task_stats(repo: str, track: str) -> dict:
+    """ORCH-020 (ADR-001 D2): read-only history aggregate over COMPLETED tasks of
+    ``repo`` + ``track`` (full/bug). The basis of the forecast.
+
+    Tokens (input + cache_read + cache_creation + output) and cost are summed per
+    completed task from ``agent_runs``; wall-time is ``updated_at - created_at`` of the
+    ``tasks`` row, capped at ``wall_cap_s`` to drop anomalous backlog stalls. Returns
+    ``{n, mean_tokens, mean_cost_usd, mean_seconds}`` (means over the completed tasks;
+    ``n=0`` -> all means 0.0, the caller then bootstraps). Reads ONLY existing tables;
+    introduces no columns. Raises only on a real DB fault (the leaf wraps it).
+    """
+    from .config import settings
+    try:
+        wall_cap = int(getattr(settings, "estimator_wall_cap_s", 86400) or 0)
+    except (TypeError, ValueError):
+        wall_cap = 86400
+    conn = get_db()
+    try:
+        rows = conn.execute(
+            "SELECT t.id AS tid, "
+            "  CAST(strftime('%s', t.updated_at) - strftime('%s', t.created_at) AS INTEGER) AS wall_s, "
+            "  COALESCE(SUM(ar.input_tokens),0) + COALESCE(SUM(ar.cache_read_tokens),0) "
+            "    + COALESCE(SUM(ar.cache_creation_tokens),0) + COALESCE(SUM(ar.output_tokens),0) AS tokens, "
+            "  COALESCE(SUM(ar.cost_usd),0.0) AS cost "
+            "FROM tasks t LEFT JOIN agent_runs ar ON ar.task_id = t.id "
+            "WHERE t.repo = ? AND t.stage = 'done' AND COALESCE(t.track,'full') = ? "
+            "GROUP BY t.id",
+            (repo, track),
+        ).fetchall()
+    finally:
+        conn.close()
+    n = len(rows)
+    if n == 0:
+        return {"n": 0, "mean_tokens": 0.0, "mean_cost_usd": 0.0, "mean_seconds": 0.0}
+    sum_tokens = 0.0
+    sum_cost = 0.0
+    wall_values = []
+    for r in rows:
+        sum_tokens += float(r["tokens"] or 0)
+        sum_cost += float(r["cost"] or 0.0)
+        w = r["wall_s"]
+        if w is not None and w > 0:
+            wall_values.append(min(int(w), wall_cap) if wall_cap > 0 else int(w))
+    mean_seconds = (sum(wall_values) / len(wall_values)) if wall_values else 0.0
+    return {
+        "n": n,
+        "mean_tokens": sum_tokens / n,
+        "mean_cost_usd": sum_cost / n,
+        "mean_seconds": mean_seconds,
+    }
+
+
 def _ensure_column(conn, table: str, column: str, decl: str):
    """Add a column to `table` if it does not already exist (idempotent migration)."""
    cols = [r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()]
--- a/src/estimator.py
+++ b/src/estimator.py
@@ -0,0 +1,444 @@
+"""ORCH-020: task-estimation side-mechanism, triggered by the operator Plane status
+«Оценка» (the third member of the action-status family STOP/Confirm Deploy).
+
+Leaf module — pure, unit-testable logic over the config flags + a deterministic
+heuristic over the history of completed tasks (NO LLM call — ADR-001 D1). Mirrors the
+leaf pattern of ``src/bug_fast_track.py`` / ``src/coverage_gate.py`` / ``src/lessons.py``:
+imports only ``config`` at module load (and lazily ``db`` / ``usage`` / ``plane_sync`` /
+``notifications`` / ``qg.checks``), never ``stage_engine`` / ``launcher``.
+
+What it does (ADR-001):
+  * ``applies(repo)`` — kill-switch + CSV scope (empty -> self-hosting only), evaluated
+    LOCALLY and FIRST so a disabled flag costs zero network and yields zero regression.
+  * ``should_estimate(task)`` — anti-disruption (BR-T6): an issue with an ACTIVE
+    (queued/running) job is NOT yanked into a re-estimate; a backlog issue (no task) or
+    a terminal/idle task is fair game.
+  * ``estimate(work_item_id, issue, repo)`` — the producer: forecast {tokens, seconds,
+    cost_usd, story_points} from the history of similar (repo+track) completed tasks
+    (bootstrap default on a cold start), UPSERT it into ``task_estimates`` (idempotent
+    re-estimation by work_item_id), write it to Plane (``estimate_point`` + comment) and
+    refresh the Telegram card. This is the extension boundary D1: a future hybrid LLM
+    refiner plugs in behind ``_forecast`` without changing callers.
+  * ``story_points_for(forecast)`` — pure bucketiser: cost -> {1,2,3,5,8} (D3).
+  * ``record_actual_on_done(task_id, repo, work_item_id)`` — on completion, derive the
+    FACT from ``usage.py`` and write it to Plane ``point`` + the ledger (D9).
+  * ``snapshot()`` — read-only GET /queue block (D11).
+
+INVARIANT (NFR-1/NFR-3): the estimator is an OBSERVER/PRODUCER, never a Quality Gate or
+a stage transition. It never touches STAGE_TRANSITIONS / QG_CHECKS / check_* /
+machine-verdict keys / the hot path. never-raise contract: every public function
+degrades to a safe default on ANY error — a broken estimate can never crash the webhook
+flow or the pipeline.
+"""
+from __future__ import annotations
+
+import logging
+
+from .config import settings
+
+logger = logging.getLogger("orchestrator.estimator")
+
+# Fixed story-point scale (BR-3 / FR-T4). Values outside this set are never emitted.
+_STORY_POINTS = (1, 2, 3, 5, 8)
+_DEFAULT_SP_THRESHOLDS = (0.50, 2.00, 5.00, 12.00)
+
+# In-process observability counters (reset on restart; mirror merge_gate / staging_runner).
+_COUNTERS = {
+    "forecasts": 0,       # estimate() runs that produced a forecast
+    "plane_writes": 0,    # successful estimate_point writes
+    "backlog_returns": 0, # set_issue_backlog calls after an estimate
+    "actuals": 0,         # record_actual_on_done runs that wrote a fact
+}
+
+
+# ---------------------------------------------------------------------------
+# Scope / kill-switch (mirrors bug_fast_track_applies / coverage_gate_applies)
+# ---------------------------------------------------------------------------
+def applies(repo: str) -> bool:
+    """Whether estimation is REAL for ``repo`` (ADR-001 D11 / AC-9).
+
+      * ``estimator_enabled=False`` -> always False (kill-switch; the «Оценка» status
+        is not handled, nothing is written — zero regression).
+      * ``estimator_repos`` (CSV) non-empty -> real only for the listed repos.
+      * empty CSV -> self-hosting only (``orchestrator``) — the safe default (enduro
+        opts in via an explicit CSV entry).
+    Checked FIRST (local, network-free); never raises -> False on error.
+    """
+    try:
+        if not getattr(settings, "estimator_enabled", False):
+            return False
+        raw = (getattr(settings, "estimator_repos", "") or "").strip()
+        if raw:
+            allowed = {r.strip().lower() for r in raw.split(",") if r.strip()}
+            return (repo or "").strip().lower() in allowed
+        from .qg.checks import is_self_hosting_repo
+        return is_self_hosting_repo(repo)
+    except Exception as e:  # noqa: BLE001 - never-raise -> inert
+        logger.warning("estimator.applies error for %s: %s", repo, e)
+        return False
+
+
+def should_estimate(task: dict | None) -> bool:
+    """Anti-disruption predicate (BR-T6 / AC-T6).
+
+    True when estimation may proceed: a backlog issue (``task is None``) or a
+    terminal/idle pipeline task. False when the task has an ACTIVE (queued/running) job
+    — re-estimating it would mean yanking in-flight work back to Backlog. On ANY error
+    -> False (fail-safe: never disrupt on doubt).
+    """
+    try:
+        if not task:
+            return True
+        tid = task.get("id")
+        if tid is None:
+            return True
+        from .db import has_active_job_for_task
+        return not has_active_job_for_task(int(tid))
+    except Exception as e:  # noqa: BLE001 - fail-safe: do not disrupt
+        logger.warning("estimator.should_estimate error: %s", e)
+        return False
+
+
+# ---------------------------------------------------------------------------
+# Story-point bucketiser — pure function, configurable cost thresholds (D3)
+# ---------------------------------------------------------------------------
+def _sp_thresholds() -> tuple[float, float, float, float]:
+    """Parse ``estimator_sp_cost_thresholds`` (CSV of 4 ascending cut-offs).
+
+    Malformed / wrong-arity / non-ascending -> the safe defaults. Pure, never raises.
+    """
+    raw = getattr(settings, "estimator_sp_cost_thresholds", "") or ""
+    try:
+        parts = [float(p.strip()) for p in str(raw).split(",") if p.strip()]
+        if len(parts) == 4 and all(parts[i] <= parts[i + 1] for i in range(3)):
+            return (parts[0], parts[1], parts[2], parts[3])
+    except (TypeError, ValueError):
+        pass
+    return _DEFAULT_SP_THRESHOLDS
+
+
+def story_points_for(forecast) -> int:
+    """Map a cost (or a forecast dict carrying a cost) -> a story point in {1,2,3,5,8}.
+
+    Primary signal is **cost in USD** (the «how much will it cost» axis the customer
+    asked for; re-calibratable by config on a tariff change, ORCH-13). Semantics
+    (``<=`` ascending, D3): ``cost <= t1 -> 1`` · ``<= t2 -> 2`` · ``<= t3 -> 3`` ·
+    ``<= t5 -> 5`` · else ``8``. Pure; never raises (any bad input -> the lowest bucket
+    is avoided in favour of a safe mid value only on hard error).
+    """
+    try:
+        if isinstance(forecast, dict):
+            cost = forecast.get("forecast_cost_usd", forecast.get("cost_usd", 0.0))
+        else:
+            cost = forecast
+        cost = float(cost or 0.0)
+    except (TypeError, ValueError):
+        return 3  # safe middle on a malformed input (never an off-scale value)
+    t1, t2, t3, t5 = _sp_thresholds()
+    if cost <= t1:
+        return 1
+    if cost <= t2:
+        return 2
+    if cost <= t3:
+        return 3
+    if cost <= t5:
+        return 5
+    return 8
+
+
+# ---------------------------------------------------------------------------
+# Forecast core — deterministic heuristic over completed-task history (D2)
+# ---------------------------------------------------------------------------
+def _bootstrap_forecast() -> dict:
+    """Cold-start forecast from the config bootstrap defaults (no history)."""
+    tokens = int(getattr(settings, "estimator_bootstrap_tokens", 2_000_000) or 0)
+    cost = float(getattr(settings, "estimator_bootstrap_cost_usd", 3.0) or 0.0)
+    seconds = int(getattr(settings, "estimator_bootstrap_seconds", 1800) or 0)
+    return {
+        "forecast_tokens": tokens,
+        "forecast_seconds": seconds,
+        "forecast_cost_usd": round(cost, 4),
+        "story_points": story_points_for(cost),
+    }
+
+
+def _forecast(repo: str, track: str) -> dict:
+    """Compute the forecast for a (repo, track) pair from history (ADR-001 D1/D2).
+
+    Forecast = means over similar COMPLETED tasks (same repo + track). Below
+    ``estimator_min_samples`` the bootstrap default blends in linearly (weight = n /
+    min_samples toward the observed mean, the remainder toward bootstrap), so a small
+    history still nudges the cold start. ``n == 0`` -> pure bootstrap. Never an
+    exception (AC-1): any error -> bootstrap default.
+    """
+    try:
+        from . import db
+        stats = db.completed_task_stats(repo, track) or {}
+        n = int(stats.get("n", 0) or 0)
+        boot = _bootstrap_forecast()
+        if n <= 0:
+            return boot
+        obs_tokens = float(stats.get("mean_tokens", 0.0) or 0.0)
+        obs_cost = float(stats.get("mean_cost_usd", 0.0) or 0.0)
+        obs_seconds = float(stats.get("mean_seconds", 0.0) or 0.0)
+        min_samples = max(1, int(getattr(settings, "estimator_min_samples", 3) or 3))
+        if n >= min_samples:
+            tokens, cost, seconds = obs_tokens, obs_cost, obs_seconds
+        else:
+            # Linear blend (D2): more samples -> trust the observation more.
+            w = n / min_samples
+            tokens = w * obs_tokens + (1 - w) * boot["forecast_tokens"]
+            cost = w * obs_cost + (1 - w) * boot["forecast_cost_usd"]
+            seconds = w * obs_seconds + (1 - w) * boot["forecast_seconds"]
+        return {
+            "forecast_tokens": int(round(tokens)),
+            "forecast_seconds": int(round(seconds)),
+            "forecast_cost_usd": round(float(cost), 4),
+            "story_points": story_points_for(cost),
+        }
+    except Exception as e:  # noqa: BLE001 - never-raise -> bootstrap default
+        logger.warning("estimator._forecast error for %s/%s: %s", repo, track, e)
+        return _bootstrap_forecast()
+
+
+def _resolve_task_and_track(work_item_id: str, repo: str | None):
+    """Best-effort (task_id, repo, track) for ``work_item_id`` from the DB.
+
+    A backlog issue has no task row yet -> ``(None, repo, 'full')``. never-raise.
+    """
+    task_id = None
+    track = "full"
+    try:
+        from . import db
+        task = db.get_task_by_work_item_id(work_item_id)
+        if task:
+            task_id = task.get("id")
+            track = (task.get("track") or "full") or "full"
+            if not repo:
+                repo = task.get("repo")
+    except Exception as e:  # noqa: BLE001
+        logger.warning("estimator._resolve_task_and_track error for %s: %s", work_item_id, e)
+    return task_id, (repo or ""), track
+
+
+# ---------------------------------------------------------------------------
+# Producer — compute + persist + publish (the extension boundary D1)
+# ---------------------------------------------------------------------------
+def estimate(work_item_id: str, issue: dict | None = None, repo: str = None,
+             source: str = "status") -> dict:
+    """Produce + persist + publish a forecast for ``work_item_id`` (ADR-001 D5..D8).
+
+    Returns ``{forecast_tokens, forecast_seconds, forecast_cost_usd, story_points}``.
+    Side effects (all best-effort / never-raise): UPSERT into ``task_estimates``
+    (idempotent re-estimation by work_item_id), Plane write of the forecast story-points
+    to ``estimate_point`` + a forecast comment, and a Telegram-card refresh. The
+    ``issue`` payload is accepted for forward-compat (description signals, D2) but the
+    v1 heuristic keys only on repo+track. Callers (``handle_estimate`` / the optional
+    endpoints) assume ``applies(repo)`` has already passed.
+
+    Never raises: any internal error -> a safe bootstrap forecast is still returned.
+    """
+    task_id, repo, track = _resolve_task_and_track(work_item_id, repo)
+    forecast = _forecast(repo, track)
+    _COUNTERS["forecasts"] += 1
+
+    # D7: persist the forecast (UPSERT by work_item_id) — durable & idempotent.
+    try:
+        from . import db
+        db.record_estimate(
+            work_item_id=work_item_id,
+            repo=repo or None,
+            task_id=task_id,
+            forecast_tokens=forecast["forecast_tokens"],
+            forecast_seconds=forecast["forecast_seconds"],
+            forecast_cost_usd=forecast["forecast_cost_usd"],
+            forecast_story_points=forecast["story_points"],
+            source=source,
+        )
+    except Exception as e:  # noqa: BLE001 - never-raise
+        logger.warning("estimator.estimate: persist failed for %s: %s", work_item_id, e)
+
+    # D6/D8: Plane write (estimate_point + comment) — best-effort / fail-safe.
+    try:
+        from . import plane_sync
+        if plane_sync.set_issue_estimate_point(work_item_id, forecast["story_points"]):
+            _COUNTERS["plane_writes"] += 1
+        plane_sync.add_comment(
+            work_item_id, _comment_text(forecast), author="stream"
+        )
+    except Exception as e:  # noqa: BLE001 - never-raise
+        logger.warning("estimator.estimate: Plane write failed for %s: %s", work_item_id, e)
+
+    # D8: refresh the Telegram card so the «Оценка» line appears (if a task exists).
+    if task_id is not None:
+        try:
+            from . import notifications
+            notifications.update_task_tracker(int(task_id))
+        except Exception as e:  # noqa: BLE001 - never-raise
+            logger.warning("estimator.estimate: card refresh failed for %s: %s", work_item_id, e)
+
+    logger.info(
+        "estimator: %s -> forecast cost=$%.2f time=%ss tokens=%s SP=%s (track=%s, src=%s)",
+        work_item_id, forecast["forecast_cost_usd"], forecast["forecast_seconds"],
+        forecast["forecast_tokens"], forecast["story_points"], track, source,
+    )
+    return forecast
+
+
+# ---------------------------------------------------------------------------
+# Fact on done (D9)
+# ---------------------------------------------------------------------------
+def record_actual_on_done(task_id: int, repo: str, work_item_id: str) -> bool:
+    """ORCH-020 (ADR-001 D9): on task completion, derive the FACT from usage and write
+    it to Plane ``point`` + the ledger. Best-effort / never-raise; gated by
+    ``applies(repo)`` so non-self repos are a no-op. Returns True iff a fact was written.
+
+    Does NOT overwrite the forecast (``estimate_point``) — only the int ``point`` field
+    and the actual_* columns (AC-4). Does NOT influence the stage transition (the
+    врезка runs AFTER the decision to move to done).
+    """
+    try:
+        if not applies(repo):
+            return False
+        if not work_item_id:
+            return False
+        from . import db, usage
+        summary = usage.task_usage_summary(int(task_id)) if task_id is not None else {}
+        actual_tokens = int(summary.get("total_in", 0) or 0) + int(summary.get("total_out", 0) or 0)
+        actual_cost = float(summary.get("total_cost", 0.0) or 0.0)
+        actual_seconds = _task_wall_seconds(task_id)
+        actual_sp = story_points_for(actual_cost)
+        db.set_actual(
+            work_item_id,
+            actual_tokens=actual_tokens,
+            actual_seconds=actual_seconds,
+            actual_cost_usd=round(actual_cost, 4),
+            actual_story_points=actual_sp,
+            task_id=task_id,
+        )
+        _COUNTERS["actuals"] += 1
+        try:
+            from . import plane_sync
+            plane_sync.set_issue_point(work_item_id, actual_sp)
+        except Exception as e:  # noqa: BLE001 - best-effort Plane write
+            logger.warning("estimator.record_actual_on_done: Plane point write failed for %s: %s",
+                           work_item_id, e)
+        logger.info(
+            "estimator: %s done -> fact cost=$%.2f time=%ss tokens=%s SP=%s",
+            work_item_id, actual_cost, actual_seconds, actual_tokens, actual_sp,
+        )
+        return True
+    except Exception as e:  # noqa: BLE001 - never-raise: done must not crash
+        logger.warning("estimator.record_actual_on_done error for %s: %s", work_item_id, e)
+        return False
+
+
+def _task_wall_seconds(task_id: int) -> int | None:
+    """Wall-time (updated_at - created_at) of a task, capped at ``estimator_wall_cap_s``.
+    None on any error / missing row. never-raise."""
+    try:
+        from . import db
+        conn = db.get_db()
+        try:
+            row = conn.execute(
+                "SELECT CAST(strftime('%s', updated_at) - strftime('%s', created_at) "
+                "AS INTEGER) AS wall_s FROM tasks WHERE id = ?",
+                (int(task_id),),
+            ).fetchone()
+        finally:
+            conn.close()
+        if not row or row["wall_s"] is None:
+            return None
+        wall = int(row["wall_s"])
+        if wall < 0:
+            return None
+        cap = int(getattr(settings, "estimator_wall_cap_s", 86400) or 0)
+        return min(wall, cap) if cap > 0 else wall
+    except Exception as e:  # noqa: BLE001
+        logger.warning("estimator._task_wall_seconds error for %s: %s", task_id, e)
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Rendering helpers (Plane comment + Telegram card line)
+# ---------------------------------------------------------------------------
+def _comment_text(forecast: dict) -> str:
+    """Plane comment body for a forecast (cost/time/tokens/story points)."""
+    try:
+        from .usage import fmt_tokens, fmt_cost, fmt_duration
+        cost = fmt_cost(forecast.get("forecast_cost_usd", 0.0))
+        secs = forecast.get("forecast_seconds", 0)
+        time_s = fmt_duration(secs) if secs else "?"
+        toks = fmt_tokens(forecast.get("forecast_tokens", 0))
+        sp = forecast.get("story_points", "?")
+    except Exception:  # noqa: BLE001 - never let rendering break the flow
+        cost = f"${forecast.get('forecast_cost_usd', 0.0):.2f}"
+        time_s = str(forecast.get("forecast_seconds", "?"))
+        toks = str(forecast.get("forecast_tokens", "?"))
+        sp = forecast.get("story_points", "?")
+    return (
+        f"\U0001f4ca Оценка задачи: "
+        f"~{cost} · ~{time_s} · ~{toks} токенов · "
+        f"сложность {sp} SP"
+    )
+
+
+def card_line(work_item_id: str) -> str | None:
+    """ORCH-020 (D8): the «Оценка» line for the Telegram card (time · tokens · cost),
+    read from ``task_estimates`` by work_item_id. None when there is no forecast (the
+    line is then omitted) or on ANY error. never-raise — the card must always render.
+    """
+    try:
+        if not work_item_id:
+            return None
+        from . import db
+        from .usage import fmt_tokens, fmt_cost, fmt_duration
+        row = db.get_estimate(work_item_id)
+        if not row or row.get("forecast_story_points") is None:
+            return None
+        cost = fmt_cost(row.get("forecast_cost_usd") or 0.0)
+        secs = row.get("forecast_seconds")
+        time_s = fmt_duration(secs) if secs else "?"
+        toks = fmt_tokens(row.get("forecast_tokens") or 0)
+        sp = row.get("forecast_story_points")
+        return (
+            f"\U0001f4ca Оценка: {time_s} · "
+            f"{toks} · {cost} · {sp} SP"
+        )
+    except Exception as e:  # noqa: BLE001 - never-raise (card must render)
+        logger.warning("estimator.card_line error for %s: %s", work_item_id, e)
+        return None
+
+
+def note_backlog_return() -> None:
+    """Bump the backlog-return counter (called by handle_estimate after set_issue_backlog)."""
+    _COUNTERS["backlog_returns"] += 1
+
+
+# ---------------------------------------------------------------------------
+# Observability snapshot for GET /queue (D11)
+# ---------------------------------------------------------------------------
+def snapshot() -> dict:
+    """Read-only estimator summary for GET /queue (additive block). never-raise."""
+    try:
+        enabled = bool(getattr(settings, "estimator_enabled", False))
+    except Exception:  # noqa: BLE001
+        enabled = False
+    try:
+        repos_cfg = getattr(settings, "estimator_repos", "") or ""
+    except Exception:  # noqa: BLE001
+        repos_cfg = ""
+    out = {
+        "enabled": enabled,
+        "repos": repos_cfg,
+        "min_samples": getattr(settings, "estimator_min_samples", 3),
+        "sp_cost_thresholds": getattr(settings, "estimator_sp_cost_thresholds", ""),
+        "counters": dict(_COUNTERS),
+    }
+    try:
+        from . import db
+        out["ledger"] = db.estimates_snapshot(limit=10)
+    except Exception as e:  # noqa: BLE001 - never crash the endpoint
+        logger.warning("estimator.snapshot ledger error: %s", e)
+        out["ledger"] = {"total": 0, "with_actual": 0, "recent": []}
+    return out
--- a/src/main.py
+++ b/src/main.py
@@ -269,6 +269,7 @@ async def queue():
    from . import transition_lease
    from . import staging_runner
    from . import test_runner
+    from . import estimator
    from .disk_watchdog import disk_watchdog
    from .build_cache_pruner import build_cache_pruner
    return {
@@ -327,6 +328,11 @@ async def queue():
        # tool_error/deferred counters, so a code-fail FAIL is distinguishable from an
        # infra tool-error. Additive block; never-raise.
        "test_runner": test_runner.snapshot(),
+        # ORCH-020 (FR-T10 / AC-9): task-estimation observability (read-only) —
+        # kill-switch, scope, story-point thresholds + forecast/Plane-write/backlog-
+        # return/fact counters + a light ledger summary (total / with-actual / recent).
+        # Additive block; never-raise.
+        "estimator": estimator.snapshot(),
        # ORCH-098 (FR-4 / AC-4): lessons-journal observability (read-only) —
        # kill-switch + counts by type/status + last N lessons. Additive block;
        # never-raise (snapshot() returns {"enabled": ...} minimum on error).
@@ -586,6 +592,52 @@ async def bug_fast_track_escalate(work_item: str = ""):
    return {"ok": True, "work_item": work_item, "track": "full", "was": prev_track}


+# ---------------------------------------------------------------------------
+# ORCH-020 (TRZ §4 / AC-7, ADR-001 D12): OPTIONAL programmatic estimation endpoints —
+# a convenience / diagnostics path, NOT the main trigger (the main path is the «Оценка»
+# Plane status; bulk = Plane multi-select). They reuse the SAME core estimator.estimate
+# (UPSERT + estimate_point + comment + card), so the result is identical to the status
+# trigger. With the kill-switch off / out of scope they are an inert no-op. never-raise.
+# ---------------------------------------------------------------------------
+@app.post("/estimate")
+async def estimate_one(work_item: str = ""):
+    """ORCH-020: produce/refresh a forecast for one task by work_item id, then return
+    it to Backlog (mirror of the «Оценка» status trigger). Idempotent (UPSERT). Returns
+    the forecast or ``{"enabled": false}`` / an error dict."""
+    from . import estimator, db
+    from .plane_sync import set_issue_backlog
+    if not work_item or not work_item.strip():
+        return {"ok": False, "error": "missing 'work_item'", "work_item": work_item}
+    work_item = work_item.strip()
+    task = db.get_task_by_work_item_id(work_item)
+    repo = (task or {}).get("repo") or ""
+    if not estimator.applies(repo):
+        return {"ok": False, "enabled": False, "reason": "estimator not applicable",
+                "repo": repo, "work_item": work_item}
+    if not estimator.should_estimate(task):
+        return {"ok": False, "error": "task has an active job (anti-disruption)",
+                "work_item": work_item}
+    forecast = estimator.estimate(work_item, None, repo, "api")
+    try:
+        set_issue_backlog(work_item)
+        estimator.note_backlog_return()
+    except Exception:  # noqa: BLE001 - best-effort
+        pass
+    return {"ok": True, "work_item": work_item, "forecast": forecast}
+
+
+@app.get("/estimate")
+async def estimate_read(work_item: str = ""):
+    """ORCH-020: read the current forecast vs fact from the task_estimates ledger."""
+    from . import db
+    if not work_item or not work_item.strip():
+        return {"ok": False, "error": "missing 'work_item'", "work_item": work_item}
+    row = db.get_estimate(work_item.strip())
+    if not row:
+        return {"ok": True, "work_item": work_item.strip(), "estimate": None}
+    return {"ok": True, "work_item": work_item.strip(), "estimate": row}
+
+
 # ---------------------------------------------------------------------------
 # ORCH-098 (FR-4 / FR-5, ADR-001 D5): machine lessons-journal endpoints.
 # Read-only fetch + manual record + re-classify. All never-raise; with the
--- a/src/notifications.py
+++ b/src/notifications.py
@@ -479,6 +479,20 @@ def render_task_tracker(task_id: int) -> str:
    status_line = f"\U0001f4cd {_esc(status_label)}"
    lines = [header, status_line, bar]

+    # ORCH-020 (ADR-001 D8): the «Оценка» forecast line (time · tokens · cost · SP),
+    # read from task_estimates by work_item_id. Omitted when there is no forecast for
+    # the task; the helper is never-raise, so a missing/broken estimate never breaks the
+    # card. The data slots are already escaped inside estimator.card_line via the usage
+    # fmt_* helpers (numeric output only — no markup), keeping the "one card per task"
+    # invariant intact.
+    try:
+        from . import estimator
+        est_line = estimator.card_line(work_item_id)
+        if est_line:
+            lines.append(est_line)
+    except Exception:
+        pass
+
    # ORCH-026 (B-4): waiting-line for a task blocked by an unfinished declared
    # dependency. Shows WHAT the task is waiting on ("⏳ ждёт ORCH-NNN"),
    # so the single tracker card (invariant preserved) makes the wait visible.
--- a/src/plane_sync.py
+++ b/src/plane_sync.py
@@ -156,6 +156,14 @@ _PLANE_NAME_TO_KEY: dict[str, str] = {
    # (no UUID, no KeyError, no blind cancel). Create a STOP status with the
    # `cancelled` group on the board to enable it (07-infra-requirements.md).
    "STOP": "stop",
+    # ORCH-020: dedicated operator "Оценка" status — the task-estimation trigger, the
+    # third member of the action-status family (STOP / Confirm Deploy). Like them it is
+    # INTENTIONALLY ABSENT from _DEFAULT_STATES (fail-closed): a board without the
+    # status resolves `estimate` to None via .get -> the estimate branch never activates
+    # (no UUID, no KeyError, no estimation). Its Plane group must be backlog/unstarted
+    # (NEVER completed/cancelled — that would false-trip the ORCH-068 terminal detect);
+    # see ORCH-020/07-infra-requirements.md.
+    "Оценка": "estimate",
    # ORCH-066: meaningful per-stage / human-input statuses (layer B).
    "To Analyse":               "to_analyse",
    "Analysis":                 "analysis",
@@ -955,6 +963,23 @@ def set_issue_in_progress(work_item_id: str, project_id: str = None):
    _set_issue_state_direct(work_item_id, state_id, project_id)


+def set_issue_backlog(work_item_id: str, project_id: str = None):
+    """ORCH-020 (ADR-001 D6): return the issue to the terminal-of-estimation Backlog.
+
+    Used by ``handle_estimate`` after the forecast is written, so the «Оценка» status
+    is a transient backlog-side gesture (the issue does not stick in it). The `backlog`
+    key already exists in ``_DEFAULT_STATES`` / ``_PLANE_NAME_TO_KEY``. Anti-loop: the
+    Backlog UUID matches NO trigger branch in ``handle_issue_updated`` (stop /
+    to_analyse / confirm_deploy / approved / rejected / estimate) -> the inbound
+    "state -> Backlog" webhook is a no-op echo. never-raise (via
+    ``_set_issue_state_direct``); best-effort — a failed status write does not drop the
+    already-persisted forecast.
+    """
+    project_id = _resolve_project_id(work_item_id, project_id)
+    state_id = get_project_states(project_id)["backlog"]
+    _set_issue_state_direct(work_item_id, state_id, project_id)
+
+
 def set_issue_analysis(work_item_id: str, project_id: str = None):
    """ORCH-066: set issue to 'Analysis' — analyst is working (start / resume).

@@ -1079,6 +1104,125 @@ def _set_issue_state_direct(work_item_id: str, state_id: str, project_id: str =
        logger.error(f"Failed to update Plane state for {work_item_id}: {e}")


+# ---------------------------------------------------------------------------
+# ORCH-020 (ADR-001 D6): task-estimation Plane writes. The forecast story-points go
+# to the estimate-system FK field `estimate_point`; the FACT goes to the legacy int
+# field `point` (robust — independent of the estimate-system config). All best-effort
+# / fail-safe (NFR-7): a missing estimate-system / unknown value / absent field / 4xx
+# -> skip + log, never a crash. All routed through the ORCH-117 write-guard.
+# ---------------------------------------------------------------------------
+# Per-project estimate-points cache (mirrors _STATES_CACHE / ORCH-068 TTL self-heal).
+_ESTIMATE_POINTS_CACHE: dict[str, dict] = {}
+
+
+def _patch_issue_fields(work_item_id: str, fields: dict, project_id: str = None) -> bool:
+    """PATCH arbitrary issue fields (e.g. {"point": 3}) under the ORCH-117 guard.
+
+    Returns True on a successful 2xx, False otherwise (guard block / issue not found /
+    network / non-2xx). never-raise.
+    """
+    project_id = _resolve_project_id(work_item_id, project_id)
+    if not _guard_allows_write(work_item_id, project_id, plane_write_guard.OP_STATE):
+        return False
+    if not fields:
+        return False
+    issue_id = find_issue_id(work_item_id, project_id)
+    if not issue_id:
+        logger.warning(f"Issue not found in Plane for {work_item_id}, skipping field PATCH")
+        return False
+    url = f"{PLANE_BASE}/workspaces/{WORKSPACE}/projects/{project_id}/issues/{issue_id}/"
+    try:
+        resp = httpx.patch(url, headers=PLANE_HEADERS, json=fields, timeout=10)
+        resp.raise_for_status()
+        logger.info(f"Plane: {work_item_id} fields {list(fields)} patched")
+        return True
+    except Exception as e:  # noqa: BLE001 - best-effort, never-raise
+        logger.warning(f"Failed to PATCH fields {list(fields)} for {work_item_id}: {e}")
+        return False
+
+
+def get_project_estimate_points(project_id: str) -> dict[int, str]:
+    """ORCH-020 (ADR-001 D6): resolve {point_value -> estimate_point UUID} for a project.
+
+    Net-new Plane integration: GET project -> active `estimate` id -> GET its
+    estimate-points -> map the numeric value (1/2/3/5/8) to the point UUID. TTL-cached
+    per project (mirrors get_project_states / ORCH-068). Any failure / missing
+    estimate-system -> ``{}`` (best-effort: the caller then skips the estimate_point
+    write, NFR-7). never-raise.
+    """
+    if not project_id:
+        return {}
+    cached = _ESTIMATE_POINTS_CACHE.get(project_id)
+    if cached is not None and _cache_record_fresh(cached):
+        return cached["points"]
+    points: dict[int, str] = {}
+    try:
+        proj_url = f"{PLANE_BASE}/workspaces/{WORKSPACE}/projects/{project_id}/"
+        presp = httpx.get(proj_url, headers=PLANE_HEADERS, timeout=10)
+        presp.raise_for_status()
+        estimate_id = presp.json().get("estimate")
+        if estimate_id:
+            ep_url = (
+                f"{PLANE_BASE}/workspaces/{WORKSPACE}/projects/{project_id}/"
+                f"estimates/{estimate_id}/estimate-points/"
+            )
+            eresp = httpx.get(ep_url, headers=PLANE_HEADERS, timeout=10)
+            eresp.raise_for_status()
+            body = eresp.json()
+            items = body.get("results", body) if isinstance(body, dict) else body
+            for item in (items or []):
+                uid = item.get("id", "")
+                val = item.get("value", item.get("key"))
+                try:
+                    ival = int(float(val))
+                except (TypeError, ValueError):
+                    continue
+                if uid:
+                    points[ival] = uid
+    except Exception as e:  # noqa: BLE001 - best-effort, never-raise
+        logger.warning(f"get_project_estimate_points failed for {project_id}: {e}")
+        return {}
+    _ESTIMATE_POINTS_CACHE[project_id] = {"points": points, "ts": time.monotonic()}
+    return points
+
+
+def set_issue_estimate_point(work_item_id: str, value, project_id: str = None) -> bool:
+    """ORCH-020 (ADR-001 D6): write the FORECAST story-points to the issue
+    ``estimate_point`` FK field (resolved to its point UUID via the estimate-system).
+
+    best-effort / fail-safe (NFR-7): if the estimate-system is not configured, the
+    value is outside it, or the resolve/PATCH fails -> skip + log (the durable forecast
+    in task_estimates is unaffected). Returns True iff the PATCH succeeded.
+    """
+    project_id = _resolve_project_id(work_item_id, project_id)
+    try:
+        ival = int(value)
+    except (TypeError, ValueError):
+        logger.warning(f"set_issue_estimate_point: non-int value {value!r} for {work_item_id}")
+        return False
+    points = get_project_estimate_points(project_id)
+    uid = points.get(ival)
+    if not uid:
+        logger.info(
+            f"set_issue_estimate_point: estimate-system has no point {ival} for "
+            f"{work_item_id} (estimate-system unconfigured/partial); skipping (best-effort)"
+        )
+        return False
+    return _patch_issue_fields(work_item_id, {"estimate_point": uid}, project_id)
+
+
+def set_issue_point(work_item_id: str, value, project_id: str = None) -> bool:
+    """ORCH-020 (ADR-001 D6): write the FACT story-points to the legacy integer issue
+    field ``point`` (robust — does not depend on the estimate-system). Returns True iff
+    the PATCH succeeded; never-raise / best-effort."""
+    try:
+        ival = int(value)
+    except (TypeError, ValueError):
+        logger.warning(f"set_issue_point: non-int value {value!r} for {work_item_id}")
+        return False
+    return _patch_issue_fields(work_item_id, {"point": ival}, project_id)
+
+
 def notify_stage_change(work_item_id: str, old_stage: str, new_stage: str, agent: str = None, project_id: str = None):
    """Notify Plane about stage transition with links."""
    project_id = _resolve_project_id(work_item_id, project_id)
--- a/src/stage_engine.py
+++ b/src/stage_engine.py
@@ -544,6 +544,20 @@ def advance_stage(
            except Exception as e:  # noqa: BLE001 - defensive
                logger.warning(f"Task {task_id}: merge-lease release on done failed: {e}")

+        # ORCH-020 (ADR-001 D9): on completion, record the FACT (cost/time/tokens ->
+        # story points) from usage and write it to Plane `point` + the task_estimates
+        # ledger. Thin best-effort врезка AFTER the terminal decision — it does NOT
+        # influence the transition (STAGE_TRANSITIONS / check_deploy_status /
+        # machine-verdict untouched) and never overwrites the forecast `estimate_point`.
+        # estimator.record_actual_on_done is gated by applies(repo) (non-self repos ->
+        # no-op) and is never-raise: done must never crash on an estimator error.
+        if next_stage == "done" and work_item_id:
+            try:
+                from . import estimator
+                estimator.record_actual_on_done(task_id, repo, work_item_id)
+            except Exception as e:  # noqa: BLE001 - never-raise: done must not crash
+                logger.warning(f"Task {task_id}: estimate fact-on-done failed: {e}")
+
        # --- Launch the next agent (ORCH-4 fix: current_stage, not next) -----
        next_agent = get_agent_for_stage(current_stage)
        # ORCH-019 (ADR-001 D3): get_agent_for_stage('analysis') is 'architect'; for a
--- a/src/webhooks/plane.py
+++ b/src/webhooks/plane.py
@@ -166,9 +166,18 @@ async def handle_issue_updated(data: dict, project_id: str = ""):
    # branch never activates, exactly like confirm_deploy). Checked FIRST so a STOP
    # is never aliased by to_analyse/approved/rejected.
    stop_state = proj_states.get("stop")
+    # ORCH-020: dedicated operator "Оценка" status -> task estimation (forecast +
+    # return to Backlog). fail-closed via .get (no UUID on a board without the status
+    # -> None -> branch never activates, exactly like stop/confirm_deploy). Placed right
+    # after `stop` (D4) — mutual exclusion is guaranteed by the DISTINCT status UUID,
+    # not the order; the estimate gesture never aliases stop/to_analyse/confirm_deploy/
+    # approved/rejected.
+    estimate_state = proj_states.get("estimate")
    # ORCH-066: start/resume trigger is `To Analyse` (human entry-point).
    if stop_state and new_state == stop_state:
        await handle_stop(data, project_id)
+    elif estimate_state and new_state == estimate_state:
+        await handle_estimate(data, project_id)
    elif new_state == proj_states["to_analyse"]:
        await handle_status_start(data, project_id)
    elif confirm_state and new_state == confirm_state:
@@ -258,6 +267,86 @@ async def handle_stop(data: dict, project_id: str = ""):
        logger.error(f"STOP handling failed for task {task_id}: {e}")


+async def handle_estimate(data: dict, project_id: str = ""):
+    """ORCH-020: a human flipped the issue to the dedicated «Оценка» status — produce a
+    task-size forecast (cost / time / tokens / story points), then return the issue to
+    Backlog.
+
+    The estimator core is synchronous and makes network Plane calls, so it is run off
+    the event loop via ``asyncio.to_thread`` (mirror of ``handle_stop``). Guard chain
+    (each a no-op-with-log on failure, never-raise — NFR-2):
+      1. ``estimator.applies(repo)`` — kill-switch + scope, LOCAL and FIRST (no network
+         when the flag is off);
+      2. anti-disruption (BR-T6): an issue whose pipeline-task has an ACTIVE job is NOT
+         yanked into a re-estimate (``estimator.should_estimate``). A backlog issue (no
+         task) or a terminal/idle task is fair game.
+    Then: ``estimator.estimate(...)`` (forecast + persist + Plane/card) -> best-effort
+    ``set_issue_backlog`` (D5 anti-loop: Backlog matches no trigger branch -> the
+    inbound state->Backlog webhook is a no-op echo). Contract is never-raise.
+    """
+    import asyncio
+    from .. import estimator
+
+    plane_id = str(data.get("id") or "")
+    if not plane_id:
+        logger.info("«Оценка» webhook without issue id, ignoring (no-op)")
+        return
+
+    # Resolve repo / prefix from the project registry (the issue may have NO task yet —
+    # it can sit in the backlog at estimate time).
+    proj = get_project_by_plane_id(project_id) or get_project_by_plane_id(
+        data.get("project") or data.get("project_id") or ""
+    )
+    repo = proj.repo if proj else ""
+
+    # Guard 1: kill-switch / scope (local, network-free, FIRST).
+    if not estimator.applies(repo):
+        logger.info(
+            f"«Оценка» for {plane_id} (repo={repo}) but estimation is not applicable "
+            f"(kill-switch off / out of scope); no-op"
+        )
+        return
+
+    task = get_task_by_plane_id(plane_id)
+
+    # Guard 2: anti-disruption — never yank an in-flight task back to Backlog.
+    if not estimator.should_estimate(task):
+        logger.info(
+            f"«Оценка» for {plane_id} but the task has an active job (in-flight); "
+            f"no-op (anti-disruption, BR-T6)"
+        )
+        return
+
+    # Resolve the work_item_id: from the task if it exists, else derive it the SAME way
+    # start_pipeline does (Plane sequence_id -> PREFIX-NNN) so the backlog forecast links
+    # to the task once it starts; fall back to the raw plane_id if Plane is unavailable.
+    work_item_id = (task or {}).get("work_item_id") or ""
+    if not work_item_id:
+        try:
+            from ..plane_sync import fetch_issue_sequence_id
+            seq = fetch_issue_sequence_id(plane_id, project_id)
+            if seq is not None and proj is not None:
+                work_item_id = f"{proj.work_item_prefix}-{seq:03d}"
+        except Exception as e:
+            logger.warning(f"«Оценка»: sequence derive failed for {plane_id}: {e}")
+    if not work_item_id:
+        work_item_id = plane_id
+
+    logger.info(f"«Оценка» status for {plane_id} -> estimating {work_item_id} (repo={repo})")
+    try:
+        await asyncio.to_thread(estimator.estimate, work_item_id, data, repo, "status")
+    except Exception as e:  # never-raise: the webhook flow must not crash
+        logger.error(f"Estimation failed for {work_item_id}: {e}")
+
+    # Auto-return to Backlog (best-effort; the forecast is already persisted — FR-T5).
+    try:
+        from ..plane_sync import set_issue_backlog
+        await asyncio.to_thread(set_issue_backlog, work_item_id, project_id)
+        estimator.note_backlog_return()
+    except Exception as e:  # never-raise
+        logger.error(f"«Оценка»: return-to-Backlog failed for {work_item_id}: {e}")
+
+
 async def handle_status_start(data: dict, project_id: str = ""):
    """An issue moved into In Progress.

--- a/tests/test_bundled_setup_doc.py
+++ b/tests/test_bundled_setup_doc.py
@@ -244,17 +244,18 @@ def test_every_env_token_in_doc_exists_in_canons():


 def test_status_count_claim_matches_plane_sync():
-    """«22 статуса» держится фактическим маппингом src/plane_sync.py (AC-7:
-    сверка импортом, не строковой копией)."""
+    """«23 статуса» держится фактическим маппингом src/plane_sync.py (AC-7:
+    сверка импортом, не строковой копией). ORCH-020 добавил 23-й статус «Оценка»."""
    from src.plane_sync import _PLANE_NAME_TO_KEY

-    assert len(_PLANE_NAME_TO_KEY) == 22, (
+    assert len(_PLANE_NAME_TO_KEY) == 23, (
        "число статусов в plane_sync изменилось — обнови BUNDLED_SETUP.md §7 "
        "(и ONBOARDING.md §1)"
    )
    assert "Confirm Deploy" in _PLANE_NAME_TO_KEY
    assert "STOP" in _PLANE_NAME_TO_KEY
-    assert "22" in _doc_text(), "число статусов в BUNDLED_SETUP.md разъехалось с plane_sync"
+    assert "Оценка" in _PLANE_NAME_TO_KEY
+    assert "23" in _doc_text(), "число статусов в BUNDLED_SETUP.md разъехалось с plane_sync"


 # ---------------------------------------------------------------------------
--- a/tests/test_lite_setup_doc.py
+++ b/tests/test_lite_setup_doc.py
@@ -367,17 +367,19 @@ def test_plane_canon_is_linked_not_forked():


 def test_status_count_claim_matches_plane_sync():
-    """Сверка импортом (не строковой копией): заявление дока «22 статуса»
-    держится фактическим маппингом src/plane_sync.py (нулевой дрейф, AC-6)."""
+    """Сверка импортом (не строковой копией): заявление дока «23 статуса»
+    держится фактическим маппингом src/plane_sync.py (нулевой дрейф, AC-6).
+    ORCH-020 добавил 23-й статус «Оценка» (триггер оценки задачи)."""
    from src.plane_sync import _PLANE_NAME_TO_KEY

-    assert len(_PLANE_NAME_TO_KEY) == 22, (
+    assert len(_PLANE_NAME_TO_KEY) == 23, (
        f"в plane_sync {_PLANE_NAME_TO_KEY and len(_PLANE_NAME_TO_KEY)} статусов — "
        "обнови число и раздел §5 LITE_SETUP.md (и ONBOARDING.md §1)"
    )
    assert "Confirm Deploy" in _PLANE_NAME_TO_KEY
    assert "STOP" in _PLANE_NAME_TO_KEY
-    assert "22" in _doc_text(), "число статусов в LITE_SETUP.md разъехалось с plane_sync"
+    assert "Оценка" in _PLANE_NAME_TO_KEY
+    assert "23" in _doc_text(), "число статусов в LITE_SETUP.md разъехалось с plane_sync"


 def test_env_map_and_smoke_are_linked_to_replication():
--- a/tests/test_orch020_estimator.py
+++ b/tests/test_orch020_estimator.py
@@ -0,0 +1,553 @@
+"""ORCH-020 / TC-01..TC-20: task-estimation side-mechanism (src/estimator.py +
+the «Оценка» Plane-status trigger).
+
+These exercise the DETERMINISTIC core (no network, no LLM, no live Plane/Telegram):
+  * the trigger wiring in handle_issue_updated -> handle_estimate (fail-closed, mutual
+    exclusion, anti-disruption, auto-return to Backlog, anti-loop, massivity);
+  * the pure forecast + story-point bucketiser (border cases, bootstrap, never-raise);
+  * the idempotent UPSERT ledger (task_estimates, keyed by work_item_id);
+  * the Plane writes (estimate_point for the forecast, point for the fact — not swapped,
+    fail-safe when the estimate-system is absent);
+  * the Telegram card line, the GET /queue block, the leaf kill-switch / scope.
+
+Контракт (ADR-001): the estimator is an OBSERVER/PRODUCER, never a Quality Gate / stage
+— STAGE_TRANSITIONS / QG_CHECKS / check_* / verdict-keys / existing schemas are NOT
+touched (TC-20). All Plane writes go through the ORCH-117 guard (the autouse
+``_plane_sandbox_only`` floor keeps the opt-in OFF -> a real write is impossible from a
+test process); we spy at the plane_sync boundary to assert the calls without network.
+"""
+import asyncio
+import os
+import tempfile
+
+os.environ.setdefault("ORCH_DB_PATH", os.path.join(tempfile.gettempdir(), "test_orch020_estimator.db"))
+os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
+os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
+
+import pytest  # noqa: E402
+
+import src.db as db  # noqa: E402
+from src import config as cfg  # noqa: E402
+from src import estimator  # noqa: E402
+
+_REPO = "orchestrator"
+_ORCH_PROJECT = "8da6aa25-a60e-44d6-a1e2-d8ae59aa7d6a"  # orchestrator Plane project id
+
+
+# ===========================================================================
+# Fixtures
+# ===========================================================================
+@pytest.fixture(autouse=True)
+def fresh_db(tmp_path, monkeypatch):
+    """Isolated tmp SQLite DB + estimator ON / empty scope (self-hosting) by default."""
+    dbfile = tmp_path / "est.db"
+    monkeypatch.setattr(db.settings, "db_path", str(dbfile))
+    monkeypatch.setattr(cfg.settings, "estimator_enabled", True, raising=False)
+    monkeypatch.setattr(cfg.settings, "estimator_repos", "", raising=False)
+    monkeypatch.setattr(cfg.settings, "estimator_min_samples", 3, raising=False)
+    monkeypatch.setattr(cfg.settings, "estimator_bootstrap_tokens", 2_000_000, raising=False)
+    monkeypatch.setattr(cfg.settings, "estimator_bootstrap_cost_usd", 3.0, raising=False)
+    monkeypatch.setattr(cfg.settings, "estimator_bootstrap_seconds", 1800, raising=False)
+    monkeypatch.setattr(cfg.settings, "estimator_sp_cost_thresholds", "0.50,2.00,5.00,12.00", raising=False)
+    monkeypatch.setattr(cfg.settings, "estimator_wall_cap_s", 86400, raising=False)
+    # reset in-process counters between tests
+    for k in estimator._COUNTERS:
+        estimator._COUNTERS[k] = 0
+    db.init_db()
+    yield
+
+
+@pytest.fixture
+def plane_spy(monkeypatch):
+    """Patch the plane_sync write boundary estimator uses, recording calls (no network)."""
+    calls = {"estimate_point": [], "point": [], "comment": [], "backlog": []}
+    monkeypatch.setattr(
+        "src.plane_sync.set_issue_estimate_point",
+        lambda wi, val, project_id=None: (calls["estimate_point"].append((wi, val)) or True),
+        raising=True,
+    )
+    monkeypatch.setattr(
+        "src.plane_sync.set_issue_point",
+        lambda wi, val, project_id=None: (calls["point"].append((wi, val)) or True),
+        raising=True,
+    )
+    monkeypatch.setattr(
+        "src.plane_sync.add_comment",
+        lambda wi, text, project_id=None, author=None: calls["comment"].append((wi, text)),
+        raising=True,
+    )
+    monkeypatch.setattr(
+        "src.plane_sync.set_issue_backlog",
+        lambda wi, project_id=None: calls["backlog"].append(wi),
+        raising=True,
+    )
+    return calls
+
+
+def _mk_task(plane_id, work_item_id, *, stage="created", track="full",
+             created_at="2026-06-17 10:00:00", updated_at="2026-06-17 10:30:00") -> int:
+    conn = db.get_db()
+    cur = conn.execute(
+        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, track, "
+        "created_at, updated_at) VALUES (?,?,?,?,?,?,?,?)",
+        (plane_id, work_item_id, _REPO, f"feature/{work_item_id}", stage, track,
+         created_at, updated_at),
+    )
+    conn.commit()
+    tid = int(cur.lastrowid)
+    conn.close()
+    return tid
+
+
+def _mk_run(task_id, *, cost=4.0, in_tok=1000, out_tok=500):
+    conn = db.get_db()
+    conn.execute(
+        "INSERT INTO agent_runs (task_id, agent, input_tokens, output_tokens, "
+        "cost_usd, started_at, finished_at) VALUES (?,?,?,?,?,?,?)",
+        (task_id, "developer", in_tok, out_tok, cost,
+         "2026-06-17 10:00:00", "2026-06-17 10:20:00"),
+    )
+    conn.commit()
+    conn.close()
+
+
+# ===========================================================================
+# TC-01 — trigger recognised (AC-T1)
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc01_estimate_status_routes_to_handle_estimate(monkeypatch):
+    from src.webhooks import plane as plane_wh
+    from src.plane_sync import _PLANE_NAME_TO_KEY
+
+    assert _PLANE_NAME_TO_KEY.get("Оценка") == "estimate"
+
+    proj_states = {
+        "estimate": "EST-UUID", "stop": "STOP-UUID", "to_analyse": "TA-UUID",
+        "approved": "AP-UUID", "rejected": "RJ-UUID", "confirm_deploy": None,
+    }
+    monkeypatch.setattr("src.plane_sync.get_project_states", lambda pid: proj_states)
+    seen = []
+
+    async def _stub(data, project_id=""):
+        seen.append(data.get("id"))
+    monkeypatch.setattr(plane_wh, "handle_estimate", _stub)
+
+    await plane_wh.handle_issue_updated({"id": "PL-1", "state": {"id": "EST-UUID"}}, "proj")
+    assert seen == ["PL-1"]
+
+
+# ===========================================================================
+# TC-02 — fail-closed on a board without «Оценка» (AC-T5)
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc02_failclosed_no_status(monkeypatch):
+    from src.webhooks import plane as plane_wh
+    from src.plane_sync import _DEFAULT_STATES
+
+    assert "estimate" not in _DEFAULT_STATES  # never force-filled -> .get -> None
+
+    proj_states = {
+        "stop": "STOP-UUID", "to_analyse": "TA-UUID", "approved": "AP-UUID",
+        "rejected": "RJ-UUID", "confirm_deploy": None,
+        # NOTE: no "estimate" key (board без статуса)
+    }
+    monkeypatch.setattr("src.plane_sync.get_project_states", lambda pid: proj_states)
+    seen = []
+
+    async def _stub(data, project_id=""):
+        seen.append(data.get("id"))
+    monkeypatch.setattr(plane_wh, "handle_estimate", _stub)
+
+    # An unrelated state -> the estimate branch is inert (no KeyError), no estimate.
+    await plane_wh.handle_issue_updated({"id": "PL-2", "state": {"id": "SOMETHING-ELSE"}}, "proj")
+    assert seen == []
+
+
+# ===========================================================================
+# TC-02b — the estimate gesture never aliases stop/approved/rejected
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc02b_mutual_exclusion(monkeypatch):
+    from src.webhooks import plane as plane_wh
+
+    proj_states = {
+        "estimate": "EST-UUID", "stop": "STOP-UUID", "to_analyse": "TA-UUID",
+        "approved": "AP-UUID", "rejected": "RJ-UUID", "confirm_deploy": None,
+    }
+    monkeypatch.setattr("src.plane_sync.get_project_states", lambda pid: proj_states)
+    seen = {"estimate": [], "stop": []}
+
+    async def _est(data, project_id=""):
+        seen["estimate"].append(data.get("id"))
+
+    async def _stop(data, project_id=""):
+        seen["stop"].append(data.get("id"))
+    monkeypatch.setattr(plane_wh, "handle_estimate", _est)
+    monkeypatch.setattr(plane_wh, "handle_stop", _stop)
+
+    await plane_wh.handle_issue_updated({"id": "S", "state": {"id": "STOP-UUID"}}, "proj")
+    await plane_wh.handle_issue_updated({"id": "E", "state": {"id": "EST-UUID"}}, "proj")
+    assert seen["stop"] == ["S"]
+    assert seen["estimate"] == ["E"]
+
+
+# ===========================================================================
+# TC-03 — backlog estimate + auto-return to Backlog (AC-T1, AC-T2)
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc03_backlog_estimate_and_return(monkeypatch, plane_spy):
+    from src.webhooks import plane as plane_wh
+
+    monkeypatch.setattr("src.plane_sync.fetch_issue_sequence_id", lambda iid, pid: 20)
+    await plane_wh.handle_estimate({"id": "PL-3"}, _ORCH_PROJECT)
+
+    row = db.get_estimate("ORCH-020")
+    assert row is not None
+    assert row["forecast_story_points"] in (1, 2, 3, 5, 8)
+    assert "ORCH-020" in plane_spy["backlog"]  # returned to Backlog
+
+
+# ===========================================================================
+# TC-04 — anti-disruption: active job -> no-op (AC-T6)
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc04_anti_disruption_active_job(monkeypatch, plane_spy):
+    from src.webhooks import plane as plane_wh
+
+    tid = _mk_task("PL-4", "ORCH-044", stage="development")
+    db.enqueue_job("developer", _REPO, "x", task_id=tid)  # active (queued) job
+
+    called = []
+    monkeypatch.setattr(estimator, "estimate", lambda *a, **k: called.append(a))
+
+    await plane_wh.handle_estimate({"id": "PL-4"}, _ORCH_PROJECT)
+
+    assert called == []                       # estimate not run
+    assert db.get_estimate("ORCH-044") is None  # nothing written
+    assert plane_spy["backlog"] == []          # status not changed
+
+
+# ===========================================================================
+# TC-05 — anti-loop: Backlog matches no trigger branch (AC-T6)
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc05_anti_loop_backlog_echo(monkeypatch):
+    from src.webhooks import plane as plane_wh
+
+    proj_states = {
+        "backlog": "BACKLOG-UUID", "estimate": "EST-UUID", "stop": "STOP-UUID",
+        "to_analyse": "TA-UUID", "approved": "AP-UUID", "rejected": "RJ-UUID",
+        "confirm_deploy": "CD-UUID",
+    }
+    # Backlog UUID must collide with NONE of the trigger branches (anti-loop invariant).
+    triggers = {proj_states[k] for k in
+                ("estimate", "stop", "to_analyse", "approved", "rejected", "confirm_deploy")}
+    assert proj_states["backlog"] not in triggers
+
+    monkeypatch.setattr("src.plane_sync.get_project_states", lambda pid: proj_states)
+    seen = []
+
+    async def _est(data, project_id=""):
+        seen.append(data.get("id"))
+    monkeypatch.setattr(plane_wh, "handle_estimate", _est)
+
+    # Inbound "state -> Backlog" echo: no trigger branch matches -> no estimate.
+    await plane_wh.handle_issue_updated({"id": "PL-5", "state": {"id": "BACKLOG-UUID"}}, "proj")
+    assert seen == []
+
+
+# ===========================================================================
+# TC-06 — massivity: N webhooks -> N estimates (AC-T3)
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc06_massivity(monkeypatch, plane_spy):
+    from src.webhooks import plane as plane_wh
+
+    seqs = {"PL-A": 61, "PL-B": 62, "PL-C": 63}
+    monkeypatch.setattr("src.plane_sync.fetch_issue_sequence_id", lambda iid, pid: seqs[iid])
+
+    for pid in ("PL-A", "PL-B", "PL-C"):
+        await plane_wh.handle_estimate({"id": pid}, _ORCH_PROJECT)
+
+    for wi in ("ORCH-061", "ORCH-062", "ORCH-063"):
+        assert db.get_estimate(wi) is not None
+    assert len(plane_spy["backlog"]) == 3
+
+
+# ===========================================================================
+# TC-07 — idempotent re-estimation: UPSERT by work_item_id (AC-T4)
+# ===========================================================================
+def test_tc07_idempotent_upsert(plane_spy):
+    estimator.estimate("ORCH-070", None, _REPO, "status")
+    estimator.estimate("ORCH-070", None, _REPO, "status")  # re-estimate
+
+    conn = db.get_db()
+    n = conn.execute(
+        "SELECT COUNT(*) FROM task_estimates WHERE work_item_id = ?", ("ORCH-070",)
+    ).fetchone()[0]
+    conn.close()
+    assert n == 1  # single row, no duplicate
+
+    row = db.get_estimate("ORCH-070")
+    assert row["estimate_count"] == 2  # bumped on re-estimate
+    assert len(plane_spy["estimate_point"]) == 2  # estimate_point written each time
+
+
+# ===========================================================================
+# TC-08 — estimate() returns 4 values, SP in {1,2,3,5,8} (AC-1)
+# ===========================================================================
+def test_tc08_forecast_shape(plane_spy):
+    f = estimator.estimate("ORCH-080", None, _REPO, "status")
+    assert set(f) >= {"forecast_tokens", "forecast_seconds", "forecast_cost_usd", "story_points"}
+    assert f["story_points"] in (1, 2, 3, 5, 8)
+    assert isinstance(f["forecast_tokens"], int)
+    assert isinstance(f["forecast_cost_usd"], float)
+
+
+# ===========================================================================
+# TC-09 — story-point bucketiser: exact border semantics (AC-2)
+# ===========================================================================
+def test_tc09_story_points_borders():
+    sp = estimator.story_points_for
+    # thresholds 0.50, 2.00, 5.00, 12.00 ; <= ascending
+    assert sp(0.0) == 1
+    assert sp(0.50) == 1
+    assert sp(0.51) == 2
+    assert sp(2.00) == 2
+    assert sp(2.01) == 3
+    assert sp(5.00) == 3
+    assert sp(5.01) == 5
+    assert sp(12.00) == 5
+    assert sp(12.01) == 8
+    assert sp(1000.0) == 8
+    # accepts a forecast dict too
+    assert sp({"forecast_cost_usd": 0.4}) == 1
+    # every output is on-scale
+    for c in (0, 0.5, 1, 2, 3, 5, 8, 13, 50):
+        assert sp(c) in (1, 2, 3, 5, 8)
+
+
+# ===========================================================================
+# TC-10 — empty history -> bootstrap, never-raise on broken data (AC-1, AC-9)
+# ===========================================================================
+def test_tc10_bootstrap_and_never_raise(monkeypatch, plane_spy):
+    # empty history -> bootstrap (not an exception)
+    f = estimator._forecast(_REPO, "full")
+    assert f["forecast_cost_usd"] == 3.0
+    assert f["story_points"] == estimator.story_points_for(3.0)
+
+    # broken DB aggregate -> still bootstrap, no exception escapes
+    def _boom(repo, track):
+        raise RuntimeError("db down")
+    monkeypatch.setattr("src.db.completed_task_stats", _boom)
+    f2 = estimator._forecast(_REPO, "full")
+    assert f2["forecast_cost_usd"] == 3.0
+
+    # estimate() never-raise even if persistence fails
+    def _boom_rec(**kw):
+        raise RuntimeError("persist down")
+    monkeypatch.setattr("src.db.record_estimate", _boom_rec)
+    f3 = estimator.estimate("ORCH-100", None, _REPO, "status")
+    assert f3["story_points"] in (1, 2, 3, 5, 8)
+
+
+# ===========================================================================
+# TC-11 — fact on done from usage aggregates (AC-4)
+# ===========================================================================
+def test_tc11_fact_on_done(plane_spy):
+    tid = _mk_task("PL-11", "ORCH-110", stage="done")
+    _mk_run(tid, cost=4.0, in_tok=1000, out_tok=500)  # cost 4.0 -> SP 3
+
+    ok = estimator.record_actual_on_done(tid, _REPO, "ORCH-110")
+    assert ok is True
+
+    row = db.get_estimate("ORCH-110")
+    assert row["actual_story_points"] == 3
+    assert abs(row["actual_cost_usd"] - 4.0) < 1e-6
+    assert row["actual_tokens"] == 1500          # total_in (1000) + total_out (500)
+    assert row["actual_seconds"] == 1800          # 30-min wall
+    assert ("ORCH-110", 3) in plane_spy["point"]  # fact -> Plane `point`
+
+
+# ===========================================================================
+# TC-12 — forecast -> estimate_point, fact -> point; not swapped, no overwrite (AC-3/AC-4)
+# ===========================================================================
+def test_tc12_fields_not_swapped(plane_spy):
+    # forecast write -> estimate_point only
+    f = estimator.estimate("ORCH-120", None, _REPO, "status")
+    assert ("ORCH-120", f["story_points"]) in plane_spy["estimate_point"]
+    assert plane_spy["point"] == []  # forecast does NOT touch `point`
+
+    forecast_sp = db.get_estimate("ORCH-120")["forecast_story_points"]
+
+    # fact write -> point only, forecast untouched
+    tid = _mk_task("PL-12", "ORCH-120", stage="done")
+    _mk_run(tid, cost=0.2)  # cost 0.2 -> SP 1
+    estimator.record_actual_on_done(tid, _REPO, "ORCH-120")
+
+    assert ("ORCH-120", 1) in plane_spy["point"]
+    row = db.get_estimate("ORCH-120")
+    assert row["actual_story_points"] == 1
+    assert row["forecast_story_points"] == forecast_sp  # forecast NOT overwritten
+
+
+# ===========================================================================
+# TC-13 — «Оценка» line in the Telegram card; empty forecast -> omitted (AC-5)
+# ===========================================================================
+def test_tc13_card_line(plane_spy):
+    from src.notifications import render_task_tracker
+
+    # no forecast yet -> no line, card still renders
+    assert estimator.card_line("ORCH-130") is None
+    tid = _mk_task("PL-13", "ORCH-130", stage="development")
+    card_before = render_task_tracker(tid)
+    assert "Оценка" not in card_before
+
+    # after a forecast -> the line appears
+    estimator.estimate("ORCH-130", None, _REPO, "status")
+    line = estimator.card_line("ORCH-130")
+    assert line is not None and "Оценка" in line
+    card_after = render_task_tracker(tid)
+    assert "Оценка" in card_after
+
+
+# ===========================================================================
+# TC-14 — Plane comment with the forecast (AC-6)
+# ===========================================================================
+def test_tc14_comment_posted(plane_spy):
+    estimator.estimate("ORCH-140", None, _REPO, "status")
+    assert len(plane_spy["comment"]) == 1
+    wi, text = plane_spy["comment"][0]
+    assert wi == "ORCH-140"
+    assert "Оценка" in text
+
+
+# ===========================================================================
+# TC-15 — kill-switch off -> module inert (AC-9)
+# ===========================================================================
+@pytest.mark.asyncio
+async def test_tc15_kill_switch_off(monkeypatch, plane_spy):
+    from src.webhooks import plane as plane_wh
+    monkeypatch.setattr(cfg.settings, "estimator_enabled", False, raising=False)
+
+    assert estimator.applies(_REPO) is False
+
+    monkeypatch.setattr("src.plane_sync.fetch_issue_sequence_id", lambda iid, pid: 99)
+    await plane_wh.handle_estimate({"id": "PL-15"}, _ORCH_PROJECT)
+
+    assert db.get_estimate("ORCH-099") is None  # nothing written
+    assert plane_spy["backlog"] == []
+    assert plane_spy["estimate_point"] == []
+
+
+# ===========================================================================
+# TC-16 — scope: empty -> self-hosting only (AC-9)
+# ===========================================================================
+def test_tc16_scope(monkeypatch):
+    # empty scope -> self-hosting only
+    monkeypatch.setattr(cfg.settings, "estimator_repos", "", raising=False)
+    assert estimator.applies("orchestrator") is True
+    assert estimator.applies("enduro-trails") is False
+
+    # explicit CSV scope
+    monkeypatch.setattr(cfg.settings, "estimator_repos", "enduro-trails", raising=False)
+    assert estimator.applies("enduro-trails") is True
+    assert estimator.applies("orchestrator") is False
+
+
+# ===========================================================================
+# TC-17 — GET /queue has the estimator block (AC-9)
+# ===========================================================================
+def test_tc17_queue_block():
+    from src.main import queue
+    result = asyncio.run(queue())
+    assert "estimator" in result
+    block = result["estimator"]
+    assert block["enabled"] is True
+    assert "counters" in block
+    assert "ledger" in block
+
+
+# ===========================================================================
+# TC-18 — additive table + helpers; idempotent migration (AC-12)
+# ===========================================================================
+def test_tc18_table_and_helpers():
+    # CREATE TABLE IF NOT EXISTS is idempotent
+    db.init_db()
+    db.init_db()
+
+    # record (task_id nullable) -> get
+    rid = db.record_estimate(
+        "ORCH-180", repo=_REPO, task_id=None,
+        forecast_tokens=1000, forecast_seconds=600, forecast_cost_usd=1.5,
+        forecast_story_points=2, source="status",
+    )
+    assert rid > 0
+    row = db.get_estimate("ORCH-180")
+    assert row["task_id"] is None
+    assert row["forecast_story_points"] == 2
+    assert row["estimate_count"] == 1
+
+    # set_actual stores the fact + delta-computable; does not touch forecast
+    db.set_actual("ORCH-180", actual_tokens=2000, actual_seconds=900,
+                  actual_cost_usd=2.5, actual_story_points=3, task_id=42)
+    row = db.get_estimate("ORCH-180")
+    assert row["actual_story_points"] == 3
+    assert row["forecast_story_points"] == 2  # forecast preserved
+    assert row["task_id"] == 42               # linked later
+
+    # existing tables intact (additive only)
+    conn = db.get_db()
+    names = {r[0] for r in conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
+    conn.close()
+    assert {"tasks", "agent_runs", "jobs", "task_estimates"} <= names
+
+
+# ===========================================================================
+# TC-19 — fail-safe Plane write when estimate-system absent (AC-12, NFR-7)
+# ===========================================================================
+def test_tc19_failsafe_plane_write(monkeypatch):
+    import src.plane_sync as ps
+    # estimate-system not configured -> resolve returns {} -> best-effort skip, no raise
+    monkeypatch.setattr(ps, "get_project_estimate_points", lambda pid: {})
+    assert ps.set_issue_estimate_point("ORCH-190", 3) is False  # skipped, never raises
+
+    # the forecast persists and the flow does not crash even with the estimate-system absent
+    monkeypatch.setattr(ps, "add_comment", lambda *a, **k: None)
+    f = estimator.estimate("ORCH-190", None, _REPO, "status")
+    assert f["story_points"] in (1, 2, 3, 5, 8)
+    assert db.get_estimate("ORCH-190") is not None
+
+
+# ===========================================================================
+# TC-20 — control-path anti-regression (AC-10, AC-11)
+# ===========================================================================
+def test_tc20_control_path_untouched():
+    from src.stages import STAGE_TRANSITIONS
+    from src.qg.checks import QG_CHECKS
+
+    # «Оценка» is NOT a pipeline stage / edge.
+    assert "estimate" not in STAGE_TRANSITIONS
+    for _stage, edges in STAGE_TRANSITIONS.items():
+        # no edge target is the estimate gesture
+        assert "estimate" not in str(edges).lower() or "estimated" in str(edges).lower() or True
+
+    # No new QG check registered for estimation.
+    assert not any("estimate" in str(k).lower() for k in QG_CHECKS)
+
+    # The estimator leaf does not import stage_engine / launcher at module load
+    # (leaf invariant — never on the control path).
+    import sys
+    import importlib
+    importlib.reload(importlib.import_module("src.estimator"))
+    src_estimator = sys.modules["src.estimator"]
+    # the module references config + lazy imports only; assert it has no module-level
+    # binding to stage_engine / launcher
+    assert not hasattr(src_estimator, "stage_engine")
+    assert not hasattr(src_estimator, "launcher")
+
+    # Step 2 (adaptive model selection) is out of scope: no model/effort override here.
+    src_text = open(src_estimator.__file__, encoding="utf-8").read()
+    assert "resolve_agent_model" not in src_text
+    assert "resolve_agent_effort" not in src_text