diff --git a/README.md b/README.md
index 2c8881c0..b4cf1213 100644
--- a/README.md
+++ b/README.md
@@ -37,8 +37,9 @@ ClickHouse, могут опираться на устаревшие рекоме
 
 - CLI-команда `chadvisor analyze`;
 - MCP-сервер `chadvisor mcp-server`;
-- 62 rewrite/advisory-правил `R-*`;
-- 15 детекторов антипаттернов `D-*`;
+- 74 rewrite/advisory-правила `R-*`;
+- 25 детекторов антипаттернов `D-*`;
+- 20 environment-правил `E-*`;
 - version detection через ClickHouse HTTP API;
 - console / JSON / Markdown отчёты;
 - режим объяснений `--mode explain`;
@@ -47,8 +48,8 @@ ClickHouse, могут опираться на устаревшие рекоме
 - synthetic benchmark для проверки срабатывания правил.
 
 В каталоге `/docs/rules/cards/` описаны 119 валидируемых карточек правил:
-77 из них уже имеют реализацию и тесты, 42 остаются backlog-карточками для
-следующих итераций.
+119 имеют зарегистрированную реализацию и прямое тестовое покрытие, backlog
+по карточкам закрыт.
 
 ## Быстрый старт
 
@@ -171,19 +172,20 @@ poetry run chadvisor analyze --sql query.sql \
 пользовательские данные не читаются. Используется только оценка планировщика
 ClickHouse (`rows`, `marks`).
 
-### Schema и EXPLAIN как дополнительные входы
+### Schema, EXPLAIN и environment как дополнительные входы
 
 CLI уже принимает опциональные файлы:
 
 ```bash
 poetry run chadvisor analyze --sql query.sql \
   --schema schema.sql \
+  --environment environment.json \
   --explain explain.json
 ```
 
-На текущем этапе основная часть реализованных правил работает по SQL AST.
-Схема и EXPLAIN зарезервированы для правил, которым нужен дополнительный
-контекст.
+`environment.json` передаёт настройки, hardware, cluster/workload facts и
+system metrics для `E-*` и части Tier 2 advisory rules. Если файл окружения не
+передан, environment-правила не срабатывают и SQL-only анализ остаётся прежним.
 
 ## Knowledge base и retrieval advisory
 
@@ -258,15 +260,13 @@ poetry run chadvisor mcp-server
 
 | Surface | Карточки | Реализовано и протестировано | Backlog |
 |---|---:|---:|---:|
-| `R-*` rewrite/advisory rules | 74 | 62 | 12 |
-| `D-*` detectors | 25 | 15 | 10 |
-| `E-*` environment cards | 20 | 0 | 20 |
-| Всего | 119 | 77 | 42 |
+| `R-*` rewrite/advisory rules | 74 | 74 | 0 |
+| `D-*` detectors | 25 | 25 | 0 |
+| `E-*` environment cards | 20 | 20 | 0 |
+| Всего | 119 | 119 | 0 |
 
-Реализованные диапазоны: `R-001`…`R-062`,
-`D-003`, `D-004`, `D-007`, `D-014`…`D-025`. Карточки `R-101`…`R-112`,
-`D-001`, `D-002`, `D-005`…`D-013` и `E-001`…`E-020`
-пока описывают backlog.
+Реализованные диапазоны: `R-001`…`R-062`, `R-101`…`R-112`,
+`D-001`…`D-025` и `E-001`…`E-020`.
 
 Полный список карточек хранится в [`docs/rules/cards/`](docs/rules/cards/),
 а фактическая регистрация правил — в
@@ -278,13 +278,14 @@ poetry run chadvisor mcp-server
 - `1B` — приближённые или opt-in рекомендации;
 - `1C` — условные рекомендации, зависящие от схемы или контекста;
 - `detector` — диагностика антипаттерна без автоматического rewrite.
+- `env` — environment-aware настройки по hardware, settings, cluster и workload.
 
 ## Метрики качества
 
 Текущие воспроизводимые метрики на 2026-06-30:
 
-- rule detection on synthetic held-out: `36/36` cases, strict precision
-  `1.000`, recall `1.000`, F1 `1.000`;
+- rule detection on expanded synthetic benchmark: `222/222` cases, strict
+  precision `1.000`, recall `1.000`, F1 `1.000`;
 - classifier on `synthetic_expanded_v1` test split: best test macro F1
   `0.691`, best test micro F1 `0.988`;
 - retrieval MRR@3: `0.517` on `20` explicit query-doc pairs with MiniLM-L6
@@ -322,7 +323,7 @@ poetry run python scripts/eval/ablation_embeddings.py
 ## Архитектура
 
 ```text
-SQL + optional Schema / EXPLAIN / CH version
+SQL + optional Schema / EXPLAIN / Environment / CH version
         ↓
    SQL Parser (sqlglot, ClickHouse dialect)
         ↓
@@ -380,7 +381,6 @@ CLI/MCP формируются rule engine, ML evaluation surface и локал
 - автоматический анализ `query_log`;
 - автоматические DDL-изменения;
 - выполнение `ANALYZE` или реальный replay запросов на данных;
-- полноценные environment rules по hardware/config.
+- автоматическое применение Tier 2 design/storage рекомендаций без проверки DBA.
 
-Эти направления описаны в ADR и backlog, но README отражает именно то, что
-поддерживает текущий код репозитория.
+README отражает именно то, что поддерживает текущий код репозитория.
diff --git a/benchmark/README.md b/benchmark/README.md
index 87fb99ba..a2e56dcc 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -26,14 +26,17 @@ poetry run python scripts/eval/run_benchmark.py
 
 The repository also includes an expanded generated synthetic dataset:
 
-- `benchmark/cases/synthetic_expanded/`: 180 deterministic cases
+- `benchmark/cases/synthetic_expanded/`: 222 deterministic cases, including
+  SQL-only, schema-aware, and env-aware rule probes
 - `benchmark/splits/synthetic_expanded_v1.yaml`: fixed 80/20 train/test split
+  for the original 180-case ML dataset
 - `scripts/benchmark/generate_synthetic_dataset.py`: reproducible generator
 
 The expanded dataset exists to avoid presenting a suspicious metric on only 20
 hand-authored probes. It contains positive variations for implemented rules,
-multi-label overlaps where the current analyzer really emits several findings,
-and negative queries where no implemented rule should fire.
+targeted backlog-closure probes for `D-*`, `E-*`, and `R-101+`, multi-label
+overlaps where the current analyzer really emits several findings, and
+negative queries where no implemented rule should fire.
 
 Run the expanded benchmark:
 
@@ -101,6 +104,9 @@ Core fields:
 - `source`
 - `sql`
 - `schema_files`
+- `schema_ddl` for inline schema-aware cases
+- `environment` for env-aware cases
+- `context_type`: `sql-only`, `schema-aware`, `env-aware`, or `explain-aware`
 - `known_issues`
 - `expected_rules_to_fire`
 - `expected_improvement`
@@ -113,8 +119,9 @@ Core fields:
 - `cases/clickbench/`: ClickBench query seeds
 - `cases/job/`: selected JOIN-heavy JOB seeds
 - `cases/synthetic/`: 20 validated targeted rule cases
-- `cases/synthetic_expanded/`: 180 generated rule-regression cases with
-  negative examples and a fixed split
+- `cases/synthetic_expanded/`: 222 rule-regression cases with negative examples,
+  fixed split metadata for the original generated subset, and targeted
+  schema/env-aware probes for newly implemented rules
 - `cases/github-issues/`: placeholder for manually curated issue-derived cases
 
 ## Validation
diff --git a/benchmark/SCHEMA.yaml b/benchmark/SCHEMA.yaml
index e2ec6741..8b398b12 100644
--- a/benchmark/SCHEMA.yaml
+++ b/benchmark/SCHEMA.yaml
@@ -29,6 +29,13 @@ properties:
     type: array
     items:
       type: string
+  schema_ddl:
+    type: ["string", "null"]
+  environment:
+    type: ["object", "null"]
+  context_type:
+    type: string
+    enum: ["sql-only", "schema-aware", "env-aware", "explain-aware"]
   known_issues:
     type: array
     items:
diff --git a/benchmark/TEMPLATE.yaml b/benchmark/TEMPLATE.yaml
index 249d4bf1..0291f4e4 100644
--- a/benchmark/TEMPLATE.yaml
+++ b/benchmark/TEMPLATE.yaml
@@ -8,6 +8,9 @@ source: "TODO: tpch | clickbench | job | synthetic | github-issues"
 sql: |
   -- TODO: paste SQL here
 schema_files: [] # TODO: add relative schema file paths if relevant
+schema_ddl: null # TODO: inline DDL for schema-aware cases when no schema file is needed
+environment: null # TODO: structured environment context for env-aware cases
+context_type: sql-only # sql-only | schema-aware | env-aware | explain-aware
 known_issues: [] # TODO: list of {issue_id, type, detected_by_rule, severity, description}
 expected_rules_to_fire: [] # TODO: expert-only labeling with DBA review
 expected_findings_count: 0 # TODO: expected number of findings in the final report
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d001_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d001_001.yaml
new file mode 100644
index 00000000..79bfa80d
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d001_001.yaml
@@ -0,0 +1,19 @@
+status: validated
+case_id: synthetic_expanded_d001_001
+source: synthetic
+sql: SELECT count() FROM events WHERE user_id = 42
+schema_files: []
+schema_ddl: CREATE TABLE events (ts DateTime, user_id UInt64) ENGINE = MergeTree PARTITION BY toYYYYMM(ts) PRIMARY KEY user_id ORDER BY user_id
+context_type: schema-aware
+known_issues:
+- issue_id: I-1
+  type: full_scan_on_partitioned_table
+  detected_by_rule: D-001
+  severity: high
+  description: Partitioned table query does not filter the partition key.
+expected_rules_to_fire:
+- D-001
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-001.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d002_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d002_001.yaml
new file mode 100644
index 00000000..8e6f230f
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d002_001.yaml
@@ -0,0 +1,18 @@
+status: validated
+case_id: synthetic_expanded_d002_001
+source: synthetic
+sql: SELECT count() FROM events CROSS JOIN users
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: cross_join_risk
+  detected_by_rule: D-002
+  severity: high
+  description: CROSS JOIN can create a large Cartesian product.
+expected_rules_to_fire:
+- D-002
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-002.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d005_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d005_001.yaml
new file mode 100644
index 00000000..bd3dc7fb
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d005_001.yaml
@@ -0,0 +1,24 @@
+status: validated
+case_id: synthetic_expanded_d005_001
+source: synthetic
+sql: SELECT count() FROM logs WHERE message LIKE '%timeout%'
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: leading_wildcard_like
+  detected_by_rule: D-005
+  severity: medium
+  description: Leading wildcard LIKE cannot use ordered pruning efficiently.
+- issue_id: I-2
+  type: skip_index_recommendation
+  detected_by_rule: R-102
+  severity: medium
+  description: Substring search may need token or ngram skip indexes.
+expected_rules_to_fire:
+- D-005
+- R-102
+expected_findings_count: 2
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-005 with the related R-102 advisory.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d006_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d006_001.yaml
new file mode 100644
index 00000000..4044e76e
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d006_001.yaml
@@ -0,0 +1,18 @@
+status: validated
+case_id: synthetic_expanded_d006_001
+source: synthetic
+sql: SELECT arrayJoin(tags) AS tag FROM events WHERE user_id = 42 LIMIT 10
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: arrayjoin_before_filter
+  detected_by_rule: D-006
+  severity: high
+  description: arrayJoin appears before WHERE and may expand rows too early.
+expected_rules_to_fire:
+- D-006
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-006.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d008_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d008_001.yaml
new file mode 100644
index 00000000..2ff9426a
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d008_001.yaml
@@ -0,0 +1,19 @@
+status: validated
+case_id: synthetic_expanded_d008_001
+source: synthetic
+sql: SELECT count() FROM events SAMPLE 0.1
+schema_files: []
+schema_ddl: CREATE TABLE events (id UInt64) ENGINE = MergeTree ORDER BY id
+context_type: schema-aware
+known_issues:
+- issue_id: I-1
+  type: sample_without_sample_by
+  detected_by_rule: D-008
+  severity: low
+  description: Query uses SAMPLE but table DDL has no SAMPLE BY expression.
+expected_rules_to_fire:
+- D-008
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-008.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d009_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d009_001.yaml
new file mode 100644
index 00000000..85efdf01
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d009_001.yaml
@@ -0,0 +1,22 @@
+status: validated
+case_id: synthetic_expanded_d009_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+schema_ddl: CREATE TABLE events (status Nullable(String)) ENGINE = MergeTree ORDER BY status
+environment:
+  null_counts:
+    status: 0
+context_type: schema-aware
+known_issues:
+- issue_id: I-1
+  type: nullable_without_need
+  detected_by_rule: D-009
+  severity: low
+  description: Nullable column has no observed NULL values in supplied context.
+expected_rules_to_fire:
+- D-009
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-009.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d010_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d010_001.yaml
new file mode 100644
index 00000000..f6c8239c
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d010_001.yaml
@@ -0,0 +1,18 @@
+status: validated
+case_id: synthetic_expanded_d010_001
+source: synthetic
+sql: SELECT id FROM (SELECT id, payload FROM events) LIMIT 10
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: unused_columns_in_select
+  detected_by_rule: D-010
+  severity: low
+  description: Inner subquery projects a column unused by the outer query.
+expected_rules_to_fire:
+- D-010
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-010.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d011_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d011_001.yaml
new file mode 100644
index 00000000..e98c55f7
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d011_001.yaml
@@ -0,0 +1,30 @@
+status: validated
+case_id: synthetic_expanded_d011_001
+source: synthetic
+sql: SELECT count() FROM events e JOIN users u ON toUInt64(e.user_id) = u.id
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: implicit_type_coercion_in_join
+  detected_by_rule: D-011
+  severity: high
+  description: JOIN key comparison contains an explicit cast.
+- issue_id: I-2
+  type: redundant_cast_on_filter_removal
+  detected_by_rule: R-008
+  severity: medium
+  description: Cast around a column can block index use.
+- issue_id: I-3
+  type: throwing_cast_to_or_default_cast
+  detected_by_rule: R-020
+  severity: medium
+  description: Throwing toType conversion on a column can interrupt the query.
+expected_rules_to_fire:
+- D-011
+- R-008
+- R-020
+expected_findings_count: 3
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-011; explicit JOIN cast also exercises cast advisories.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d012_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d012_001.yaml
new file mode 100644
index 00000000..4883f229
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d012_001.yaml
@@ -0,0 +1,18 @@
+status: validated
+case_id: synthetic_expanded_d012_001
+source: synthetic
+sql: SELECT row_number() OVER (ORDER BY ts) FROM events LIMIT 10
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: window_function_without_partition
+  detected_by_rule: D-012
+  severity: medium
+  description: Window function has no PARTITION BY clause.
+expected_rules_to_fire:
+- D-012
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-012.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_d013_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_d013_001.yaml
new file mode 100644
index 00000000..bd569ee1
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_d013_001.yaml
@@ -0,0 +1,18 @@
+status: validated
+case_id: synthetic_expanded_d013_001
+source: synthetic
+sql: SELECT id FROM (SELECT id FROM (SELECT id FROM (SELECT id FROM (SELECT id FROM events)))) LIMIT 10
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: deeply_nested_subqueries
+  detected_by_rule: D-013
+  severity: low
+  description: Query contains deeply nested subqueries.
+expected_rules_to_fire:
+- D-013
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for D-013.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e001_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e001_001.yaml
new file mode 100644
index 00000000..a2280d9a
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e001_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e001_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  settings:
+    max_threads: 64
+  hardware:
+    cpu_cores: 8
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: max_threads_vs_cpu_cores
+  detected_by_rule: E-001
+  severity: medium
+  description: max_threads is high relative to CPU cores.
+expected_rules_to_fire:
+- E-001
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-001.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e002_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e002_001.yaml
new file mode 100644
index 00000000..3e530856
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e002_001.yaml
@@ -0,0 +1,24 @@
+status: validated
+case_id: synthetic_expanded_e002_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  settings:
+    max_memory_usage: 90
+    max_bytes_before_external_group_by: 60
+  hardware:
+    ram_bytes: 100
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: max_memory_usage_vs_total_ram
+  detected_by_rule: E-002
+  severity: high
+  description: max_memory_usage leaves too little RAM headroom.
+expected_rules_to_fire:
+- E-002
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-002.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e003_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e003_001.yaml
new file mode 100644
index 00000000..0376216e
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e003_001.yaml
@@ -0,0 +1,21 @@
+status: validated
+case_id: synthetic_expanded_e003_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  settings:
+    max_memory_usage: 1000
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: external_groupby_threshold
+  detected_by_rule: E-003
+  severity: medium
+  description: External aggregation threshold is absent or too high.
+expected_rules_to_fire:
+- E-003
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-003.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e004_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e004_001.yaml
new file mode 100644
index 00000000..627d2069
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e004_001.yaml
@@ -0,0 +1,22 @@
+status: validated
+case_id: synthetic_expanded_e004_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  caches:
+    mark_cache_size: 100
+    active_marks_bytes: 1000
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: mark_cache_sizing
+  detected_by_rule: E-004
+  severity: medium
+  description: Mark cache is small relative to active marks.
+expected_rules_to_fire:
+- E-004
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-004.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e005_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e005_001.yaml
new file mode 100644
index 00000000..67808620
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e005_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e005_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    repeated_reads: true
+  caches:
+    use_uncompressed_cache: false
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: uncompressed_cache_sizing
+  detected_by_rule: E-005
+  severity: low
+  description: Repeated reads run without uncompressed cache.
+expected_rules_to_fire:
+- E-005
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-005.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e006_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e006_001.yaml
new file mode 100644
index 00000000..446c53e7
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e006_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e006_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  hardware:
+    disk_type: hdd
+  settings:
+    background_pool_size: 32
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: merge_settings_for_disk_type
+  detected_by_rule: E-006
+  severity: medium
+  description: Merge concurrency is high for HDD storage.
+expected_rules_to_fire:
+- E-006
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-006.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e007_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e007_001.yaml
new file mode 100644
index 00000000..f5ec0736
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e007_001.yaml
@@ -0,0 +1,21 @@
+status: validated
+case_id: synthetic_expanded_e007_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  storage:
+    has_hot_cold_data: true
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: storage_policy_recommendation
+  detected_by_rule: E-007
+  severity: low
+  description: Hot/cold data is present without a storage policy.
+expected_rules_to_fire:
+- E-007
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-007.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e008_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e008_001.yaml
new file mode 100644
index 00000000..3682038b
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e008_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e008_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    complex_expressions: true
+  settings:
+    compile_expressions: false
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: jit_compilation_toggle
+  detected_by_rule: E-008
+  severity: low
+  description: Complex expressions run without JIT compilation.
+expected_rules_to_fire:
+- E-008
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-008.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e009_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e009_001.yaml
new file mode 100644
index 00000000..bfb7d751
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e009_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e009_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    repeated_queries: true
+  settings:
+    use_query_cache: false
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: query_cache_for_repeated_workload
+  detected_by_rule: E-009
+  severity: low
+  description: Repeated queries run without query cache.
+expected_rules_to_fire:
+- E-009
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-009.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e010_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e010_001.yaml
new file mode 100644
index 00000000..772aad40
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e010_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e010_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  system_metrics:
+    active_parts: 900
+  settings:
+    parts_to_throw_insert: 1000
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: parts_to_throw_insert_tuning
+  detected_by_rule: E-010
+  severity: high
+  description: Active part count is close to parts_to_throw_insert.
+expected_rules_to_fire:
+- E-010
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-010.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e011_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e011_001.yaml
new file mode 100644
index 00000000..083ffa7b
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e011_001.yaml
@@ -0,0 +1,24 @@
+status: validated
+case_id: synthetic_expanded_e011_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  cluster:
+    shards: 4
+    replicas: 2
+  settings:
+    max_distributed_connections: 4
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: distributed_connections_sizing
+  detected_by_rule: E-011
+  severity: medium
+  description: max_distributed_connections is below cluster fan-out.
+expected_rules_to_fire:
+- E-011
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-011.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e012_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e012_001.yaml
new file mode 100644
index 00000000..8f836af4
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e012_001.yaml
@@ -0,0 +1,21 @@
+status: validated
+case_id: synthetic_expanded_e012_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  settings:
+    join_use_nulls: true
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: join_use_nulls_overhead
+  detected_by_rule: E-012
+  severity: medium
+  description: join_use_nulls adds Nullable overhead to JOIN output.
+expected_rules_to_fire:
+- E-012
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-012.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e013_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e013_001.yaml
new file mode 100644
index 00000000..f0b04508
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e013_001.yaml
@@ -0,0 +1,22 @@
+status: validated
+case_id: synthetic_expanded_e013_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  settings:
+    async_insert: true
+    async_insert_busy_timeout_ms: 10
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: async_insert_busy_timeout_tuning
+  detected_by_rule: E-013
+  severity: low
+  description: async_insert busy timeout is too low for batching.
+expected_rules_to_fire:
+- E-013
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-013.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e014_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e014_001.yaml
new file mode 100644
index 00000000..38807c78
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e014_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e014_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    large_parts: true
+  settings:
+    max_bytes_to_merge_at_max_space_in_pool: 0
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: merge_max_bytes_for_level
+  detected_by_rule: E-014
+  severity: medium
+  description: Large-parts workload lacks explicit merge max bytes tuning.
+expected_rules_to_fire:
+- E-014
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-014.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e015_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e015_001.yaml
new file mode 100644
index 00000000..b4b09576
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e015_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e015_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    group_by_order_key_prefix: true
+  settings:
+    optimize_aggregation_in_order: false
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: optimize_aggregation_in_order
+  detected_by_rule: E-015
+  severity: medium
+  description: Workload can use aggregation in order but the setting is disabled.
+expected_rules_to_fire:
+- E-015
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-015.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e016_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e016_001.yaml
new file mode 100644
index 00000000..bac86076
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e016_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e016_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    large_join: true
+  settings:
+    join_algorithm: hash
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: join_algorithm_for_large_tables
+  detected_by_rule: E-016
+  severity: high
+  description: Large JOIN workload uses a non-adaptive join algorithm.
+expected_rules_to_fire:
+- E-016
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-016.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e017_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e017_001.yaml
new file mode 100644
index 00000000..1fb0d345
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e017_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e017_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    distributed_aggregation: true
+  settings:
+    distributed_aggregation_memory_efficient: false
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: distributed_aggregation_memory_efficient
+  detected_by_rule: E-017
+  severity: medium
+  description: Distributed aggregation runs without memory efficient mode.
+expected_rules_to_fire:
+- E-017
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-017.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e018_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e018_001.yaml
new file mode 100644
index 00000000..5b37a645
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e018_001.yaml
@@ -0,0 +1,22 @@
+status: validated
+case_id: synthetic_expanded_e018_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    bulk_inserts: true
+    insert_format: JSONEachRow
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: input_format_for_bulk_inserts
+  detected_by_rule: E-018
+  severity: medium
+  description: Bulk insert workload uses a less efficient text format.
+expected_rules_to_fire:
+- E-018
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-018.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e019_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e019_001.yaml
new file mode 100644
index 00000000..3b502a9c
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e019_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e019_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  cluster:
+    has_local_replica: true
+  settings:
+    prefer_localhost_replica: false
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: prefer_localhost_replica
+  detected_by_rule: E-019
+  severity: low
+  description: Local replica exists but prefer_localhost_replica is disabled.
+expected_rules_to_fire:
+- E-019
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-019.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_e020_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_e020_001.yaml
new file mode 100644
index 00000000..0e5f29f9
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_e020_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_e020_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  workload:
+    interactive_queries: true
+  settings:
+    max_execution_time: 0
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: max_execution_time_for_protection
+  detected_by_rule: E-020
+  severity: medium
+  description: Interactive workload has no max_execution_time guardrail.
+expected_rules_to_fire:
+- E-020
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for E-020.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r101_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r101_001.yaml
new file mode 100644
index 00000000..36309b06
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r101_001.yaml
@@ -0,0 +1,19 @@
+status: validated
+case_id: synthetic_expanded_r101_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+schema_ddl: CREATE TABLE events (id UInt64) ENGINE = MergeTree ORDER BY tuple()
+context_type: schema-aware
+known_issues:
+- issue_id: I-1
+  type: order_by_key_redesign
+  detected_by_rule: R-101
+  severity: high
+  description: MergeTree table uses ORDER BY tuple().
+expected_rules_to_fire:
+- R-101
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-101.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r102_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r102_001.yaml
new file mode 100644
index 00000000..eaf9b887
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r102_001.yaml
@@ -0,0 +1,24 @@
+status: validated
+case_id: synthetic_expanded_r102_001
+source: synthetic
+sql: SELECT count() FROM logs WHERE message LIKE '%timeout%'
+schema_files: []
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: leading_wildcard_like
+  detected_by_rule: D-005
+  severity: medium
+  description: Leading wildcard LIKE cannot use ordered pruning efficiently.
+- issue_id: I-2
+  type: skip_index_recommendation
+  detected_by_rule: R-102
+  severity: medium
+  description: Substring search may need token or ngram skip indexes.
+expected_rules_to_fire:
+- D-005
+- R-102
+expected_findings_count: 2
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-102.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r103_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r103_001.yaml
new file mode 100644
index 00000000..44bb4c21
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r103_001.yaml
@@ -0,0 +1,21 @@
+status: validated
+case_id: synthetic_expanded_r103_001
+source: synthetic
+sql: SELECT user_id, count() FROM events GROUP BY user_id
+schema_files: []
+environment:
+  workload:
+    repeated_group_by: true
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: projection_recommendation
+  detected_by_rule: R-103
+  severity: medium
+  description: Repeated GROUP BY query shape may benefit from a projection.
+expected_rules_to_fire:
+- R-103
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-103.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r104_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r104_001.yaml
new file mode 100644
index 00000000..221ff1e0
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r104_001.yaml
@@ -0,0 +1,19 @@
+status: validated
+case_id: synthetic_expanded_r104_001
+source: synthetic
+sql: SELECT count() FROM events WHERE user_id = 42
+schema_files: []
+schema_ddl: CREATE TABLE events (user_id UInt64, ts DateTime) ENGINE = MergeTree ORDER BY (user_id, ts)
+context_type: schema-aware
+known_issues:
+- issue_id: I-1
+  type: primary_key_refinement
+  detected_by_rule: R-104
+  severity: medium
+  description: Table has ORDER BY but no explicit PRIMARY KEY for selective filters.
+expected_rules_to_fire:
+- R-104
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-104.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r105_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r105_001.yaml
new file mode 100644
index 00000000..e52e49d7
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r105_001.yaml
@@ -0,0 +1,22 @@
+status: validated
+case_id: synthetic_expanded_r105_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+environment:
+  system_metrics:
+    active_parts: 1000
+    active_partitions: 5
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: partition_strategy_adjustment
+  detected_by_rule: R-105
+  severity: high
+  description: Table has too many active parts per partition.
+expected_rules_to_fire:
+- R-105
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-105.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r106_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r106_001.yaml
new file mode 100644
index 00000000..84395b34
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r106_001.yaml
@@ -0,0 +1,19 @@
+status: validated
+case_id: synthetic_expanded_r106_001
+source: synthetic
+sql: SELECT country FROM events LIMIT 10
+schema_files: []
+schema_ddl: CREATE TABLE events (country String, id UInt32) ENGINE = MergeTree ORDER BY id
+context_type: schema-aware
+known_issues:
+- issue_id: I-1
+  type: lowcardinality_conversion
+  detected_by_rule: R-106
+  severity: medium
+  description: Low-cardinality String column may benefit from LowCardinality.
+expected_rules_to_fire:
+- R-106
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-106.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r107_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r107_001.yaml
new file mode 100644
index 00000000..ab6d33ad
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r107_001.yaml
@@ -0,0 +1,22 @@
+status: validated
+case_id: synthetic_expanded_r107_001
+source: synthetic
+sql: SELECT count() FROM events
+schema_files: []
+schema_ddl: CREATE TABLE events (ts DateTime, value UInt64) ENGINE = MergeTree ORDER BY ts
+environment:
+  workload:
+    compression_sensitive: true
+context_type: schema-aware
+known_issues:
+- issue_id: I-1
+  type: codec_recommendation
+  detected_by_rule: R-107
+  severity: low
+  description: Compressible columns have no explicit CODEC.
+expected_rules_to_fire:
+- R-107
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-107.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r108_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r108_001.yaml
new file mode 100644
index 00000000..0ea0e3d9
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r108_001.yaml
@@ -0,0 +1,27 @@
+status: validated
+case_id: synthetic_expanded_r108_001
+source: synthetic
+sql: SELECT * FROM events WHERE user_id = 42 LIMIT 10
+schema_files: []
+environment:
+  workload:
+    wide_table: true
+context_type: sql-only
+known_issues:
+- issue_id: I-1
+  type: select_star_on_wide_table
+  detected_by_rule: D-003
+  severity: medium
+  description: SELECT * reads all columns.
+- issue_id: I-2
+  type: prewhere_injection
+  detected_by_rule: R-108
+  severity: medium
+  description: Wide table query filters in WHERE only.
+expected_rules_to_fire:
+- D-003
+- R-108
+expected_findings_count: 2
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-108 with existing D-003 overlap.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r109_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r109_001.yaml
new file mode 100644
index 00000000..1bbfe446
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r109_001.yaml
@@ -0,0 +1,21 @@
+status: validated
+case_id: synthetic_expanded_r109_001
+source: synthetic
+sql: SELECT count() FROM small s JOIN large l ON s.id = l.id
+schema_files: []
+environment:
+  workload:
+    right_table_larger_than_left: true
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: join_reorder_small_on_right
+  detected_by_rule: R-109
+  severity: medium
+  description: JOIN appears to put the larger table on the right side.
+expected_rules_to_fire:
+- R-109
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-109.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r110_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r110_001.yaml
new file mode 100644
index 00000000..fbc7b51b
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r110_001.yaml
@@ -0,0 +1,23 @@
+status: validated
+case_id: synthetic_expanded_r110_001
+source: synthetic
+sql: SELECT count() FROM events e JOIN users u ON e.user_id = u.id
+schema_files: []
+environment:
+  workload:
+    large_join: true
+  settings:
+    join_algorithm: auto
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: join_algorithm_choice
+  detected_by_rule: R-110
+  severity: medium
+  description: Large JOIN needs explicit algorithm review.
+expected_rules_to_fire:
+- R-110
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-110.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r111_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r111_001.yaml
new file mode 100644
index 00000000..6e7aaf39
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r111_001.yaml
@@ -0,0 +1,21 @@
+status: validated
+case_id: synthetic_expanded_r111_001
+source: synthetic
+sql: SELECT user_id, count() FROM events GROUP BY user_id
+schema_files: []
+environment:
+  workload:
+    repeated_expensive_query: true
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: materialized_view_recommendation
+  detected_by_rule: R-111
+  severity: medium
+  description: Repeated expensive aggregation may benefit from a materialized view.
+expected_rules_to_fire:
+- R-111
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-111.
diff --git a/benchmark/cases/synthetic_expanded/synthetic_expanded_r112_001.yaml b/benchmark/cases/synthetic_expanded/synthetic_expanded_r112_001.yaml
new file mode 100644
index 00000000..a818b90e
--- /dev/null
+++ b/benchmark/cases/synthetic_expanded/synthetic_expanded_r112_001.yaml
@@ -0,0 +1,21 @@
+status: validated
+case_id: synthetic_expanded_r112_001
+source: synthetic
+sql: SELECT user_id, count() FROM events GROUP BY user_id
+schema_files: []
+environment:
+  workload:
+    high_cardinality_group_by: true
+context_type: env-aware
+known_issues:
+- issue_id: I-1
+  type: groupby_multipass_split
+  detected_by_rule: R-112
+  severity: medium
+  description: High-cardinality GROUP BY may need staged aggregation.
+expected_rules_to_fire:
+- R-112
+expected_findings_count: 1
+expected_improvement: null
+synthetic_explain_path: null
+notes: Synthetic positive case for R-112.
diff --git a/clickadvisor/cli/main.py b/clickadvisor/cli/main.py
index 2fb6ba01..685eaa2b 100644
--- a/clickadvisor/cli/main.py
+++ b/clickadvisor/cli/main.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from pathlib import Path
 from typing import Annotated
 
@@ -32,6 +33,18 @@ def _read_optional(path: Path | None) -> str | None:
     return path.read_text(encoding="utf-8")
 
 
+def _read_environment(path: Path | None) -> dict[str, object] | None:
+    if path is None:
+        return None
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as error:
+        raise typer.BadParameter(f"invalid environment JSON: {error}") from error
+    if not isinstance(payload, dict):
+        raise typer.BadParameter("environment JSON must be an object")
+    return payload
+
+
 @app.command()
 def version() -> None:
     console.print(f"ClickAdvisor {__version__}")
@@ -50,6 +63,7 @@ def analyze(
     sql: Annotated[Path, typer.Option(help="SQL файл")],
     explain: Annotated[Path | None, typer.Option()] = None,
     schema: Annotated[Path | None, typer.Option()] = None,
+    environment: Annotated[Path | None, typer.Option(help="Environment JSON file")] = None,
     connect: Annotated[str | None, typer.Option(help="clickhouse://host:9000")] = None,
     ch_version: Annotated[str | None, typer.Option(help="24.3")] = None,
     ch_user: Annotated[str, typer.Option(help="ClickHouse user")] = "default",
@@ -82,6 +96,7 @@ def analyze(
     sql_text = sql.read_text(encoding="utf-8")
     explain_text = _read_optional(explain)
     schema_text = _read_optional(schema)
+    environment_data = _read_environment(environment)
 
     version = ch_version
     if connect and not ch_version:
@@ -92,6 +107,7 @@ def analyze(
         explain_output=explain_text,
         schema_ddl=schema_text,
         ch_version=version,
+        environment=environment_data,
     )
 
     rules = get_all_rules()
diff --git a/clickadvisor/core/models.py b/clickadvisor/core/models.py
index 33a5f638..f07f486f 100644
--- a/clickadvisor/core/models.py
+++ b/clickadvisor/core/models.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Optional
+from typing import Any, Optional
 
 
 @dataclass(slots=True)
@@ -10,6 +10,7 @@ class QueryContext:
     explain_output: Optional[str] = None
     schema_ddl: Optional[str] = None
     ch_version: Optional[str] = None
+    environment: dict[str, Any] | None = None
 
 
 @dataclass(slots=True)
diff --git a/clickadvisor/rules/detectors/__init__.py b/clickadvisor/rules/detectors/__init__.py
index f92ccf42..840b546b 100644
--- a/clickadvisor/rules/detectors/__init__.py
+++ b/clickadvisor/rules/detectors/__init__.py
@@ -1,3 +1,15 @@
+from clickadvisor.rules.detectors.backlog import (
+    D001FullScanOnPartitionedTable,
+    D002CrossJoinRisk,
+    D005LeadingWildcardLike,
+    D006ArrayJoinBeforeFilter,
+    D008SampleWithoutSampleBy,
+    D009NullableWithoutNeed,
+    D010UnusedColumnsInSelect,
+    D011ImplicitTypeCoercionInJoin,
+    D012WindowFunctionWithoutPartition,
+    D013DeeplyNestedSubqueries,
+)
 from clickadvisor.rules.detectors.D003_select_star import D003SelectStar
 from clickadvisor.rules.detectors.D004_missing_limit import D004MissingLimit
 from clickadvisor.rules.detectors.D007_final_modifier import D007FinalModifier
@@ -20,6 +32,16 @@
 
 __all__ = [
     "D003SelectStar",
+    "D001FullScanOnPartitionedTable",
+    "D002CrossJoinRisk",
+    "D005LeadingWildcardLike",
+    "D006ArrayJoinBeforeFilter",
+    "D008SampleWithoutSampleBy",
+    "D009NullableWithoutNeed",
+    "D010UnusedColumnsInSelect",
+    "D011ImplicitTypeCoercionInJoin",
+    "D012WindowFunctionWithoutPartition",
+    "D013DeeplyNestedSubqueries",
     "D004MissingLimit",
     "D007FinalModifier",
     "D014AsyncInsertNoWait",
diff --git a/clickadvisor/rules/detectors/backlog.py b/clickadvisor/rules/detectors/backlog.py
new file mode 100644
index 00000000..01426935
--- /dev/null
+++ b/clickadvisor/rules/detectors/backlog.py
@@ -0,0 +1,246 @@
+from __future__ import annotations
+
+import re
+
+from clickadvisor.core.models import Finding, QueryContext
+from clickadvisor.rules.base import Rule
+
+
+def _finding(
+    rule: Rule,
+    severity: str,
+    description: str,
+    suggestion: str,
+    *,
+    confidence: str = "advisory",
+) -> Finding:
+    return Finding(
+        rule_id=rule.rule_id,
+        rule_name=rule.name,
+        tier=rule.tier,
+        severity=severity,
+        description=description,
+        suggestion=suggestion,
+        confidence=confidence,
+        ch_version_introduced=rule.ch_version_introduced,
+    )
+
+
+def _schema_text(context: QueryContext) -> str:
+    return context.schema_ddl or context.sql
+
+
+class D001FullScanOnPartitionedTable(Rule):
+    rule_id = "D-001"
+    name = "full_scan_on_partitioned_table"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        schema = context.schema_ddl
+        if not schema or "PARTITION BY" not in schema.upper():
+            return None
+        match = re.search(r"\bPARTITION\s+BY\s+(.+?)(?:\s+ORDER\s+BY|\s+PRIMARY\s+KEY|\s+SETTINGS|$)", schema, re.I | re.S)
+        if not match:
+            return None
+        partition_expr = match.group(1).strip()
+        identifiers = {item.lower() for item in re.findall(r"[A-Za-z_][A-Za-z0-9_]*", partition_expr)}
+        functions = {"toyyyymm", "todate", "tostartofmonth", "tostartofday", "tostartofyear"}
+        partition_cols = identifiers - functions
+        where_match = re.search(r"\bWHERE\b(.+?)(?:\bGROUP\b|\bORDER\b|\bLIMIT\b|$)", context.sql, re.I | re.S)
+        where_text = where_match.group(1).lower() if where_match else ""
+        if partition_cols and any(col in where_text for col in partition_cols):
+            return None
+        return _finding(
+            self,
+            "high",
+            f"Таблица партиционирована по {partition_expr}, но запрос не фильтрует partition key.",
+            "Добавьте фильтр по partition-колонке или временной функции партиционирования.",
+        )
+
+
+class D002CrossJoinRisk(Rule):
+    rule_id = "D-002"
+    name = "cross_join_risk"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        sql = context.sql
+        if re.search(r"\bCROSS\s+JOIN\b", sql, re.I):
+            return _finding(self, "high", "Запрос содержит CROSS JOIN.", "Проверьте, что декартово произведение намеренное.")
+        for match in re.finditer(r"\bJOIN\b(?P<body>.*?)(?=\bJOIN\b|\bWHERE\b|\bGROUP\b|\bORDER\b|\bLIMIT\b|$)", sql, re.I | re.S):
+            body = match.group("body")
+            on_match = re.search(r"\bON\b(.+)", body, re.I | re.S)
+            if not on_match or "=" not in on_match.group(1):
+                return _finding(self, "high", "JOIN не содержит equality-условия.", "Добавьте ON с equality-ключами соединения.")
+        return None
+
+
+class D005LeadingWildcardLike(Rule):
+    rule_id = "D-005"
+    name = "leading_wildcard_like"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not re.search(r"\bLIKE\s+['\"]%", context.sql, re.I):
+            return None
+        return _finding(
+            self,
+            "medium",
+            "LIKE с leading wildcard не может эффективно использовать индекс.",
+            "Рассмотрите ngram/token skip-index или другой способ поиска по подстроке.",
+        )
+
+
+class D006ArrayJoinBeforeFilter(Rule):
+    rule_id = "D-006"
+    name = "arrayjoin_before_filter"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        sql = context.sql
+        array_pos = sql.lower().find("arrayjoin")
+        where_pos = sql.lower().find("where")
+        if array_pos < 0 or where_pos < 0 or array_pos > where_pos:
+            return None
+        return _finding(
+            self,
+            "high",
+            "arrayJoin встречается до WHERE и может резко увеличить промежуточную кардинальность.",
+            "Отфильтруйте строки до arrayJoin или вынесите arrayJoin в более поздний этап.",
+        )
+
+
+class D008SampleWithoutSampleBy(Rule):
+    rule_id = "D-008"
+    name = "sample_without_sample_by"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not re.search(r"\bSAMPLE\b", context.sql, re.I):
+            return None
+        if context.schema_ddl and re.search(r"\bSAMPLE\s+BY\b", context.schema_ddl, re.I):
+            return None
+        return _finding(
+            self,
+            "low",
+            "Запрос использует SAMPLE, но schema DDL не содержит SAMPLE BY.",
+            "Добавьте SAMPLE BY в MergeTree DDL или используйте явную sampling-логику.",
+        )
+
+
+class D009NullableWithoutNeed(Rule):
+    rule_id = "D-009"
+    name = "nullable_without_need"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        text = _schema_text(context)
+        match = re.search(r"\b([A-Za-z_][A-Za-z0-9_]*)\s+Nullable\s*\(", text, re.I)
+        if not match:
+            return None
+        env = context.environment or {}
+        null_counts = env.get("null_counts")
+        if isinstance(null_counts, dict) and null_counts.get(match.group(1), 1) != 0:
+            return None
+        return _finding(
+            self,
+            "low",
+            f"Колонка {match.group(1)} объявлена Nullable; это добавляет NULL bitmap.",
+            "Если NULL-значений нет, рассмотрите non-nullable тип.",
+        )
+
+
+class D010UnusedColumnsInSelect(Rule):
+    rule_id = "D-010"
+    name = "unused_columns_in_select"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        match = re.search(
+            r"SELECT\s+(?P<outer>[A-Za-z0-9_,\s]+)\s+FROM\s*\(\s*SELECT\s+(?P<inner>[A-Za-z0-9_,\s]+)\s+FROM\b",
+            context.sql,
+            re.I | re.S,
+        )
+        if not match:
+            return None
+        outer = {part.strip().lower() for part in match.group("outer").split(",")}
+        inner = {part.strip().lower() for part in match.group("inner").split(",")}
+        unused = sorted(inner - outer)
+        if not unused:
+            return None
+        return _finding(
+            self,
+            "low",
+            f"Подзапрос проектирует неиспользуемые колонки: {', '.join(unused)}.",
+            "Оставьте во внутреннем SELECT только колонки, используемые внешним уровнем.",
+        )
+
+
+class D011ImplicitTypeCoercionInJoin(Rule):
+    rule_id = "D-011"
+    name = "implicit_type_coercion_in_join"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not re.search(r"\bJOIN\b.*\bON\b", context.sql, re.I | re.S):
+            return None
+        cast_pattern = r"\b(?:CAST|to(?:U?Int(?:8|16|32|64)?|String|UUID|Date(?:Time)?))\s*\([^)]*\)\s*="
+        if re.search(cast_pattern, context.sql, re.I):
+            return _finding(
+                self,
+                "high",
+                "JOIN ON содержит явное приведение типа, что обычно указывает на несовпадение типов ключей.",
+                "Выровняйте типы JOIN-ключей в схеме или подготовительном слое.",
+            )
+        return None
+
+
+class D012WindowFunctionWithoutPartition(Rule):
+    rule_id = "D-012"
+    name = "window_function_without_partition"
+    tier = "detector"
+    ch_version_introduced = "21.1"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not re.search(r"\bOVER\s*\((?![^)]*\bPARTITION\s+BY\b)", context.sql, re.I | re.S):
+            return None
+        return _finding(
+            self,
+            "medium",
+            "Window function без PARTITION BY может требовать память на весь набор строк.",
+            "Добавьте PARTITION BY, если расчёт можно разбить по ключу.",
+        )
+
+
+class D013DeeplyNestedSubqueries(Rule):
+    rule_id = "D-013"
+    name = "deeply_nested_subqueries"
+    tier = "detector"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        depth = 0
+        max_depth = 0
+        for char in context.sql:
+            if char == "(":
+                depth += 1
+                max_depth = max(max_depth, depth)
+            elif char == ")":
+                depth = max(0, depth - 1)
+        select_count = len(re.findall(r"\bSELECT\b", context.sql, re.I))
+        if max_depth <= 3 or select_count < 4:
+            return None
+        return _finding(
+            self,
+            "low",
+            f"Запрос содержит глубокую вложенность подзапросов: depth={max_depth}.",
+            "Рассмотрите CTE через WITH ... AS для читаемости и предсказуемости оптимизации.",
+        )
diff --git a/clickadvisor/rules/environment.py b/clickadvisor/rules/environment.py
new file mode 100644
index 00000000..41751ab3
--- /dev/null
+++ b/clickadvisor/rules/environment.py
@@ -0,0 +1,349 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from clickadvisor.core.models import Finding, QueryContext
+from clickadvisor.rules.base import Rule
+
+
+def _section(env: dict[str, Any] | None, name: str) -> Mapping[str, Any]:
+    if not env:
+        return {}
+    value = env.get(name)
+    return value if isinstance(value, Mapping) else {}
+
+
+def _num(mapping: Mapping[str, Any], key: str) -> float | None:
+    value = mapping.get(key)
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int | float):
+        return float(value)
+    if isinstance(value, str):
+        try:
+            return float(value)
+        except ValueError:
+            return None
+    return None
+
+
+def _bool(mapping: Mapping[str, Any], key: str) -> bool | None:
+    value = mapping.get(key)
+    return value if isinstance(value, bool) else None
+
+
+def _str(mapping: Mapping[str, Any], key: str) -> str | None:
+    value = mapping.get(key)
+    return value if isinstance(value, str) else None
+
+
+def _finding(rule: Rule, severity: str, description: str, suggestion: str) -> Finding:
+    return Finding(
+        rule_id=rule.rule_id,
+        rule_name=rule.name,
+        tier=rule.tier,
+        severity=severity,
+        description=description,
+        suggestion=suggestion,
+        confidence="advisory",
+        ch_version_introduced=rule.ch_version_introduced,
+    )
+
+
+class E001MaxThreadsVsCpuCores(Rule):
+    rule_id = "E-001"
+    name = "max_threads_vs_cpu_cores"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        settings = _section(context.environment, "settings")
+        hardware = _section(context.environment, "hardware")
+        max_threads = _num(settings, "max_threads")
+        cpu_cores = _num(hardware, "cpu_cores")
+        if max_threads is None or cpu_cores is None or max_threads <= cpu_cores * 2:
+            return None
+        return _finding(self, "medium", "max_threads сильно превышает число CPU cores.", "Ограничьте max_threads ближе к CPU cores или профилю workload.")
+
+
+class E002MaxMemoryUsageVsTotalRam(Rule):
+    rule_id = "E-002"
+    name = "max_memory_usage_vs_total_ram"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        settings = _section(context.environment, "settings")
+        hardware = _section(context.environment, "hardware")
+        memory = _num(settings, "max_memory_usage")
+        ram = _num(hardware, "ram_bytes")
+        if memory is None or ram is None or memory <= ram * 0.8:
+            return None
+        return _finding(self, "high", "max_memory_usage близок к общему RAM сервера.", "Оставьте запас памяти для merges, caches и параллельных запросов.")
+
+
+class E003ExternalGroupByThreshold(Rule):
+    rule_id = "E-003"
+    name = "external_groupby_threshold"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        settings = _section(context.environment, "settings")
+        max_memory = _num(settings, "max_memory_usage")
+        external = _num(settings, "max_bytes_before_external_group_by")
+        if max_memory is None or (external is not None and 0 < external <= max_memory * 0.7):
+            return None
+        return _finding(self, "medium", "external GROUP BY threshold отсутствует или слишком высок.", "Настройте max_bytes_before_external_group_by ниже max_memory_usage.")
+
+
+class E004MarkCacheSizing(Rule):
+    rule_id = "E-004"
+    name = "mark_cache_sizing"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        caches = _section(context.environment, "caches")
+        mark_cache = _num(caches, "mark_cache_size")
+        active_marks = _num(caches, "active_marks_bytes")
+        if mark_cache is None or active_marks is None or mark_cache >= active_marks * 0.5:
+            return None
+        return _finding(self, "medium", "mark_cache_size мал относительно активных marks.", "Увеличьте mark_cache_size или сократите hot working set.")
+
+
+class E005UncompressedCacheSizing(Rule):
+    rule_id = "E-005"
+    name = "uncompressed_cache_sizing"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        caches = _section(context.environment, "caches")
+        enabled = _bool(caches, "use_uncompressed_cache")
+        cache_size = _num(caches, "uncompressed_cache_size")
+        repeated = _bool(_section(context.environment, "workload"), "repeated_reads")
+        if not repeated or (enabled and cache_size and cache_size > 0):
+            return None
+        return _finding(self, "low", "Повторяющийся workload без uncompressed cache.", "Включите и настройте uncompressed cache для повторных чтений.")
+
+
+class E006MergeSettingsForDiskType(Rule):
+    rule_id = "E-006"
+    name = "merge_settings_for_disk_type"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        disk_type = (_str(_section(context.environment, "hardware"), "disk_type") or "").lower()
+        merges = _num(_section(context.environment, "settings"), "background_pool_size")
+        if disk_type != "hdd" or merges is None or merges <= 16:
+            return None
+        return _finding(self, "medium", "background_pool_size высок для HDD storage.", "Снизьте merge concurrency или перенесите hot tables на SSD/NVMe.")
+
+
+class E007StoragePolicyRecommendation(Rule):
+    rule_id = "E-007"
+    name = "storage_policy_recommendation"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        storage = _section(context.environment, "storage")
+        hot_cold = _bool(storage, "has_hot_cold_data")
+        policy = _str(storage, "storage_policy")
+        if not hot_cold or policy:
+            return None
+        return _finding(self, "low", "Есть hot/cold data без storage_policy.", "Добавьте storage policy с hot и cold volumes.")
+
+
+class E008JitCompilationToggle(Rule):
+    rule_id = "E-008"
+    name = "jit_compilation_toggle"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        workload = _section(context.environment, "workload")
+        settings = _section(context.environment, "settings")
+        if not _bool(workload, "complex_expressions") or _bool(settings, "compile_expressions"):
+            return None
+        return _finding(self, "low", "Сложные выражения выполняются без JIT compilation.", "Проверьте compile_expressions для CPU-heavy workload.")
+
+
+class E009QueryCacheForRepeatedWorkload(Rule):
+    rule_id = "E-009"
+    name = "query_cache_for_repeated_workload"
+    tier = "env"
+    ch_version_introduced = "23.1"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        workload = _section(context.environment, "workload")
+        settings = _section(context.environment, "settings")
+        if not _bool(workload, "repeated_queries") or _bool(settings, "use_query_cache"):
+            return None
+        return _finding(self, "low", "Повторяющиеся запросы выполняются без query cache.", "Рассмотрите use_query_cache для стабильных повторных запросов.")
+
+
+class E010PartsToThrowInsertTuning(Rule):
+    rule_id = "E-010"
+    name = "parts_to_throw_insert_tuning"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        metrics = _section(context.environment, "system_metrics")
+        parts = _num(metrics, "active_parts")
+        threshold = _num(_section(context.environment, "settings"), "parts_to_throw_insert")
+        if parts is None or threshold is None or parts < threshold * 0.8:
+            return None
+        return _finding(self, "high", "active parts близки к parts_to_throw_insert.", "Укрупните inserts, проверьте partitioning и merge throughput.")
+
+
+class E011DistributedConnectionsSizing(Rule):
+    rule_id = "E-011"
+    name = "distributed_connections_sizing"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        cluster = _section(context.environment, "cluster")
+        shards = _num(cluster, "shards")
+        replicas = _num(cluster, "replicas")
+        connections = _num(_section(context.environment, "settings"), "max_distributed_connections")
+        needed = (shards or 0) * (replicas or 1)
+        if needed <= 0 or (connections is not None and connections >= needed):
+            return None
+        return _finding(self, "medium", "max_distributed_connections ниже размера fan-out.", "Увеличьте max_distributed_connections или уменьшите distributed fan-out.")
+
+
+class E012JoinUseNullsOverhead(Rule):
+    rule_id = "E-012"
+    name = "join_use_nulls_overhead"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not _bool(_section(context.environment, "settings"), "join_use_nulls"):
+            return None
+        return _finding(self, "medium", "join_use_nulls=1 добавляет Nullable overhead в JOIN results.", "Отключите join_use_nulls, если SQL-совместимость NULL не требуется.")
+
+
+class E013AsyncInsertBusyTimeoutTuning(Rule):
+    rule_id = "E-013"
+    name = "async_insert_busy_timeout_tuning"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        settings = _section(context.environment, "settings")
+        if not _bool(settings, "async_insert"):
+            return None
+        timeout = _num(settings, "async_insert_busy_timeout_ms")
+        if timeout is not None and timeout >= 100:
+            return None
+        return _finding(self, "low", "async_insert включён с низким busy timeout.", "Увеличьте async_insert_busy_timeout_ms для лучшего batching.")
+
+
+class E014MergeMaxBytesForLevel(Rule):
+    rule_id = "E-014"
+    name = "merge_max_bytes_for_level"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        settings = _section(context.environment, "settings")
+        value = _num(settings, "max_bytes_to_merge_at_max_space_in_pool")
+        large_parts = _bool(_section(context.environment, "workload"), "large_parts")
+        if not large_parts or (value is not None and value > 0):
+            return None
+        return _finding(self, "medium", "Large parts workload без явной настройки merge max bytes.", "Проверьте max_bytes_to_merge_at_max_space_in_pool для merge throughput.")
+
+
+class E015OptimizeAggregationInOrder(Rule):
+    rule_id = "E-015"
+    name = "optimize_aggregation_in_order"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        workload = _section(context.environment, "workload")
+        settings = _section(context.environment, "settings")
+        if not _bool(workload, "group_by_order_key_prefix") or _bool(settings, "optimize_aggregation_in_order"):
+            return None
+        return _finding(self, "medium", "GROUP BY совпадает с ORDER BY prefix, но optimize_aggregation_in_order выключен.", "Включите optimize_aggregation_in_order для streaming aggregation.")
+
+
+class E016JoinAlgorithmForLargeTables(Rule):
+    rule_id = "E-016"
+    name = "join_algorithm_for_large_tables"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        workload = _section(context.environment, "workload")
+        settings = _section(context.environment, "settings")
+        algorithm = (_str(settings, "join_algorithm") or "").lower()
+        if not _bool(workload, "large_join") or algorithm in {"partial_merge", "grace_hash", "auto"}:
+            return None
+        return _finding(self, "high", "Large JOIN использует неподходящий join_algorithm.", "Рассмотрите join_algorithm=auto, grace_hash или partial_merge.")
+
+
+class E017DistributedAggregationMemoryEfficient(Rule):
+    rule_id = "E-017"
+    name = "distributed_aggregation_memory_efficient"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        workload = _section(context.environment, "workload")
+        settings = _section(context.environment, "settings")
+        if not _bool(workload, "distributed_aggregation") or _bool(settings, "distributed_aggregation_memory_efficient"):
+            return None
+        return _finding(self, "medium", "Distributed aggregation без memory efficient mode.", "Включите distributed_aggregation_memory_efficient для больших distributed GROUP BY.")
+
+
+class E018InputFormatForBulkInserts(Rule):
+    rule_id = "E-018"
+    name = "input_format_for_bulk_inserts"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        workload = _section(context.environment, "workload")
+        fmt = (_str(workload, "insert_format") or "").lower()
+        if not _bool(workload, "bulk_inserts") or fmt in {"native", "parquet", "arrow", "rowbinary"}:
+            return None
+        return _finding(self, "medium", "Bulk inserts используют текстовый или неэффективный формат.", "Для bulk inserts используйте Native, RowBinary, Parquet или Arrow.")
+
+
+class E019PreferLocalhostReplica(Rule):
+    rule_id = "E-019"
+    name = "prefer_localhost_replica"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        cluster = _section(context.environment, "cluster")
+        settings = _section(context.environment, "settings")
+        has_local = _bool(cluster, "has_local_replica") or _bool(cluster, "has_local_replicas")
+        if not has_local or _bool(settings, "prefer_localhost_replica"):
+            return None
+        return _finding(self, "low", "Локальная replica доступна, но prefer_localhost_replica выключен.", "Включите prefer_localhost_replica для сокращения network hop.")
+
+
+class E020MaxExecutionTimeForProtection(Rule):
+    rule_id = "E-020"
+    name = "max_execution_time_for_protection"
+    tier = "env"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        workload = _section(context.environment, "workload")
+        settings = _section(context.environment, "settings")
+        max_time = _num(settings, "max_execution_time")
+        protected_workload = _bool(workload, "interactive_queries") or _bool(workload, "user_facing")
+        if not protected_workload or (max_time is not None and 0 < max_time <= 300):
+            return None
+        return _finding(self, "medium", "Interactive queries не защищены max_execution_time.", "Задайте max_execution_time в user profile или query settings.")
diff --git a/clickadvisor/rules/registry.py b/clickadvisor/rules/registry.py
index 7957c3ba..44cfe7d8 100644
--- a/clickadvisor/rules/registry.py
+++ b/clickadvisor/rules/registry.py
@@ -7,9 +7,19 @@
 
 from clickadvisor.rules.base import Rule
 from clickadvisor.rules.detectors import (
+    D001FullScanOnPartitionedTable,
+    D002CrossJoinRisk,
     D003SelectStar,
     D004MissingLimit,
+    D005LeadingWildcardLike,
+    D006ArrayJoinBeforeFilter,
     D007FinalModifier,
+    D008SampleWithoutSampleBy,
+    D009NullableWithoutNeed,
+    D010UnusedColumnsInSelect,
+    D011ImplicitTypeCoercionInJoin,
+    D012WindowFunctionWithoutPartition,
+    D013DeeplyNestedSubqueries,
     D014AsyncInsertNoWait,
     D015OptimizeTableFinal,
     D016AlterTableMutation,
@@ -23,6 +33,28 @@
     D024MVWithPopulate,
     D025MVWithJoin,
 )
+from clickadvisor.rules.environment import (
+    E001MaxThreadsVsCpuCores,
+    E002MaxMemoryUsageVsTotalRam,
+    E003ExternalGroupByThreshold,
+    E004MarkCacheSizing,
+    E005UncompressedCacheSizing,
+    E006MergeSettingsForDiskType,
+    E007StoragePolicyRecommendation,
+    E008JitCompilationToggle,
+    E009QueryCacheForRepeatedWorkload,
+    E010PartsToThrowInsertTuning,
+    E011DistributedConnectionsSizing,
+    E012JoinUseNullsOverhead,
+    E013AsyncInsertBusyTimeoutTuning,
+    E014MergeMaxBytesForLevel,
+    E015OptimizeAggregationInOrder,
+    E016JoinAlgorithmForLargeTables,
+    E017DistributedAggregationMemoryEfficient,
+    E018InputFormatForBulkInserts,
+    E019PreferLocalhostReplica,
+    E020MaxExecutionTimeForProtection,
+)
 from clickadvisor.rules.tier1 import (
     R001CountDistinct,
     R002CountDistinctApprox,
@@ -87,6 +119,20 @@
     R061HasOrHasToHasAny,
     R062ArrayCountZeroToNotHas,
 )
+from clickadvisor.rules.tier2 import (
+    R101OrderByKeyRedesign,
+    R102SkipIndexRecommendation,
+    R103ProjectionRecommendation,
+    R104PrimaryKeyRefinement,
+    R105PartitionStrategyAdjustment,
+    R106LowCardinalityConversion,
+    R107CodecRecommendation,
+    R108PrewhereInjection,
+    R109JoinReorderSmallOnRight,
+    R110JoinAlgorithmChoice,
+    R111MaterializedViewRecommendation,
+    R112GroupByMultipassSplit,
+)
 
 DEFAULT_CARDS_DIR = Path("docs/rules/cards")
 RULES = [
@@ -122,9 +168,19 @@
     R030NotInSingleton(),
     R031StringUUIDColumn(),
     R032Int8BooleanColumn(),
+    D001FullScanOnPartitionedTable(),
+    D002CrossJoinRisk(),
     D003SelectStar(),
     D004MissingLimit(),
+    D005LeadingWildcardLike(),
+    D006ArrayJoinBeforeFilter(),
     D007FinalModifier(),
+    D008SampleWithoutSampleBy(),
+    D009NullableWithoutNeed(),
+    D010UnusedColumnsInSelect(),
+    D011ImplicitTypeCoercionInJoin(),
+    D012WindowFunctionWithoutPartition(),
+    D013DeeplyNestedSubqueries(),
     D014AsyncInsertNoWait(),
     D015OptimizeTableFinal(),
     D016AlterTableMutation(),
@@ -167,6 +223,38 @@
     R060HasAndHasToHasAll(),
     R061HasOrHasToHasAny(),
     R062ArrayCountZeroToNotHas(),
+    E001MaxThreadsVsCpuCores(),
+    E002MaxMemoryUsageVsTotalRam(),
+    E003ExternalGroupByThreshold(),
+    E004MarkCacheSizing(),
+    E005UncompressedCacheSizing(),
+    E006MergeSettingsForDiskType(),
+    E007StoragePolicyRecommendation(),
+    E008JitCompilationToggle(),
+    E009QueryCacheForRepeatedWorkload(),
+    E010PartsToThrowInsertTuning(),
+    E011DistributedConnectionsSizing(),
+    E012JoinUseNullsOverhead(),
+    E013AsyncInsertBusyTimeoutTuning(),
+    E014MergeMaxBytesForLevel(),
+    E015OptimizeAggregationInOrder(),
+    E016JoinAlgorithmForLargeTables(),
+    E017DistributedAggregationMemoryEfficient(),
+    E018InputFormatForBulkInserts(),
+    E019PreferLocalhostReplica(),
+    E020MaxExecutionTimeForProtection(),
+    R101OrderByKeyRedesign(),
+    R102SkipIndexRecommendation(),
+    R103ProjectionRecommendation(),
+    R104PrimaryKeyRefinement(),
+    R105PartitionStrategyAdjustment(),
+    R106LowCardinalityConversion(),
+    R107CodecRecommendation(),
+    R108PrewhereInjection(),
+    R109JoinReorderSmallOnRight(),
+    R110JoinAlgorithmChoice(),
+    R111MaterializedViewRecommendation(),
+    R112GroupByMultipassSplit(),
 ]
 RULE_REGISTRY: dict[str, Rule] = {rule.rule_id: rule for rule in RULES}
 
diff --git a/clickadvisor/rules/tier2.py b/clickadvisor/rules/tier2.py
new file mode 100644
index 00000000..03c3c264
--- /dev/null
+++ b/clickadvisor/rules/tier2.py
@@ -0,0 +1,206 @@
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from typing import Any
+
+from clickadvisor.core.models import Finding, QueryContext
+from clickadvisor.rules.base import Rule
+
+
+def _env(context: QueryContext, section: str) -> Mapping[str, Any]:
+    if not context.environment:
+        return {}
+    value = context.environment.get(section)
+    return value if isinstance(value, Mapping) else {}
+
+
+def _flag(context: QueryContext, section: str, key: str) -> bool:
+    value = _env(context, section).get(key)
+    return bool(value) if isinstance(value, bool) else False
+
+
+def _num(context: QueryContext, section: str, key: str) -> float | None:
+    value = _env(context, section).get(key)
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int | float):
+        return float(value)
+    return None
+
+
+def _finding(rule: Rule, severity: str, description: str, suggestion: str) -> Finding:
+    return Finding(
+        rule_id=rule.rule_id,
+        rule_name=rule.name,
+        tier=rule.tier,
+        severity=severity,
+        description=description,
+        suggestion=suggestion,
+        confidence="advisory",
+        ch_version_introduced=rule.ch_version_introduced,
+    )
+
+
+class R101OrderByKeyRedesign(Rule):
+    rule_id = "R-101"
+    name = "order_by_key_redesign"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        schema = context.schema_ddl or ""
+        if not schema or not re.search(r"\bENGINE\s*=\s*MergeTree", schema, re.I):
+            return None
+        if not re.search(r"\bORDER\s+BY\s+tuple\s*\(\s*\)", schema, re.I):
+            return None
+        return _finding(self, "high", "MergeTree table uses ORDER BY tuple(), so data skipping is weak.", "Redesign ORDER BY around frequent filters and high-selectivity dimensions.")
+
+
+class R102SkipIndexRecommendation(Rule):
+    rule_id = "R-102"
+    name = "skip_index_recommendation"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not re.search(r"\bLIKE\s+['\"]%", context.sql, re.I):
+            return None
+        return _finding(self, "medium", "Substring search can benefit from a token/ngram skip index.", "Consider tokenbf_v1 or ngrambf_v1 on the searched text column.")
+
+
+class R103ProjectionRecommendation(Rule):
+    rule_id = "R-103"
+    name = "projection_recommendation"
+    tier = "2"
+    ch_version_introduced = "21.6"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not (_flag(context, "workload", "repeated_group_by") or _flag(context, "workload", "repeated_query_shape")):
+            return None
+        if "GROUP BY" not in context.sql.upper():
+            return None
+        return _finding(self, "medium", "Repeated GROUP BY query shape may benefit from a projection.", "Evaluate a projection for this aggregation pattern and verify with EXPLAIN.")
+
+
+class R104PrimaryKeyRefinement(Rule):
+    rule_id = "R-104"
+    name = "primary_key_refinement"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        schema = context.schema_ddl or ""
+        if not schema or "PRIMARY KEY" in schema.upper():
+            return None
+        if "ORDER BY" not in schema.upper() or "WHERE" not in context.sql.upper():
+            return None
+        return _finding(self, "medium", "Table has ORDER BY but no explicit PRIMARY KEY while queries filter selectively.", "Consider a shorter PRIMARY KEY prefix aligned with frequent filters.")
+
+
+class R105PartitionStrategyAdjustment(Rule):
+    rule_id = "R-105"
+    name = "partition_strategy_adjustment"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        parts = _num(context, "system_metrics", "active_parts")
+        partitions = _num(context, "system_metrics", "active_partitions")
+        if parts is None or partitions is None or partitions <= 0 or parts / partitions <= 100:
+            return None
+        return _finding(self, "high", "Table has many active parts per partition.", "Review partition key granularity and insert batching.")
+
+
+class R106LowCardinalityConversion(Rule):
+    rule_id = "R-106"
+    name = "lowcardinality_conversion"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        schema = context.schema_ddl or context.sql
+        if " String" not in schema:
+            return None
+        if not (_flag(context, "workload", "low_cardinality_strings") or re.search(r"\b(status|type|category|country)\s+String\b", schema, re.I)):
+            return None
+        return _finding(self, "medium", "Low-cardinality String columns can be stored more efficiently.", "Consider LowCardinality(String) after checking cardinality and mutation cost.")
+
+
+class R107CodecRecommendation(Rule):
+    rule_id = "R-107"
+    name = "codec_recommendation"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        schema = context.schema_ddl or ""
+        if not schema or "CODEC" in schema.upper():
+            return None
+        if not _flag(context, "workload", "compression_sensitive"):
+            return None
+        return _finding(self, "low", "Schema has compressible columns without explicit CODEC.", "Evaluate codecs such as Delta/ZSTD for time or numeric columns.")
+
+
+class R108PrewhereInjection(Rule):
+    rule_id = "R-108"
+    name = "prewhere_injection"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if "WHERE" not in context.sql.upper() or "PREWHERE" in context.sql.upper():
+            return None
+        if not _flag(context, "workload", "wide_table"):
+            return None
+        return _finding(self, "medium", "Wide table query filters in WHERE only.", "Consider PREWHERE for selective predicates after verifying with EXPLAIN.")
+
+
+class R109JoinReorderSmallOnRight(Rule):
+    rule_id = "R-109"
+    name = "join_reorder_small_on_right"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if "JOIN" not in context.sql.upper() or not _flag(context, "workload", "right_table_larger_than_left"):
+            return None
+        return _finding(self, "medium", "JOIN appears to put the larger table on the right side.", "Place the smaller build-side table on the right where possible.")
+
+
+class R110JoinAlgorithmChoice(Rule):
+    rule_id = "R-110"
+    name = "join_algorithm_choice"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if "JOIN" not in context.sql.upper() or not _flag(context, "workload", "large_join"):
+            return None
+        return _finding(self, "medium", "Large JOIN needs explicit algorithm review.", "Evaluate join_algorithm=auto, grace_hash or partial_merge based on memory and table size.")
+
+
+class R111MaterializedViewRecommendation(Rule):
+    rule_id = "R-111"
+    name = "materialized_view_recommendation"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if not _flag(context, "workload", "repeated_expensive_query"):
+            return None
+        if "GROUP BY" not in context.sql.upper():
+            return None
+        return _finding(self, "medium", "Repeated expensive aggregation may benefit from a materialized view.", "Consider an AggregatingMergeTree materialized view after validating freshness requirements.")
+
+
+class R112GroupByMultipassSplit(Rule):
+    rule_id = "R-112"
+    name = "groupby_multipass_split"
+    tier = "2"
+    ch_version_introduced = "1.0"
+
+    def check(self, context: QueryContext) -> Finding | None:
+        if "GROUP BY" not in context.sql.upper() or not _flag(context, "workload", "high_cardinality_group_by"):
+            return None
+        return _finding(self, "medium", "High-cardinality GROUP BY may exceed memory in one pass.", "Consider splitting aggregation into stages or using external aggregation settings.")
diff --git a/docs/evaluation.md b/docs/evaluation.md
index 59455fa3..a2aaea70 100644
--- a/docs/evaluation.md
+++ b/docs/evaluation.md
@@ -20,7 +20,7 @@ Result on 2026-06-30:
 
 | Dataset | Cases | Precision | Recall | F1 | Notes |
 |---|---:|---:|---:|---:|---|
-| `synthetic_expanded` | 180 | 1.000 | 1.000 | 1.000 | Generated rule-regression set |
+| `synthetic_expanded` | 222 | 1.000 | 1.000 | 1.000 | Generated plus targeted SQL/schema/env rule-regression set |
 | held-out split only | 36 | 1.000 | 1.000 | 1.000 | `synthetic_expanded_v1.yaml` test IDs |
 
 This is a regression result for deterministic matchers. It should not be
@@ -81,6 +81,8 @@ Artifacts:
 
 - Synthetic dataset: `benchmark/cases/synthetic_expanded`
 - Split metadata: `benchmark/splits/synthetic_expanded_v1.yaml`
+- Context metadata: benchmark cases may declare `context_type`, inline
+  `schema_ddl`, and structured `environment` objects.
 - Classifier script: `scripts/eval/ablation_classifiers.py`
 - Retrieval script: `scripts/eval/ablation_embeddings.py`
 - Benchmark runner: `scripts/eval/run_benchmark.py`
diff --git a/scripts/eval/run_benchmark.py b/scripts/eval/run_benchmark.py
index 19870303..130b8c30 100644
--- a/scripts/eval/run_benchmark.py
+++ b/scripts/eval/run_benchmark.py
@@ -43,7 +43,11 @@ def load_cases(root: Path = BENCHMARK_ROOT) -> list[dict]:
 
 
 def run_case(case: dict, mode: str) -> dict:
-    context = QueryContext(sql=case["sql"])
+    context = QueryContext(
+        sql=case["sql"],
+        schema_ddl=case.get("schema_ddl"),
+        environment=case.get("environment"),
+    )
     rules = get_applicable_rules(None)
     pipeline = AnalysisPipeline(rules)
     report = pipeline.run(context)
diff --git a/tests/rules/test_backlog_detectors.py b/tests/rules/test_backlog_detectors.py
new file mode 100644
index 00000000..9ebf2584
--- /dev/null
+++ b/tests/rules/test_backlog_detectors.py
@@ -0,0 +1,115 @@
+from clickadvisor.core.models import QueryContext
+from clickadvisor.rules.detectors.backlog import (
+    D001FullScanOnPartitionedTable,
+    D002CrossJoinRisk,
+    D005LeadingWildcardLike,
+    D006ArrayJoinBeforeFilter,
+    D008SampleWithoutSampleBy,
+    D009NullableWithoutNeed,
+    D010UnusedColumnsInSelect,
+    D011ImplicitTypeCoercionInJoin,
+    D012WindowFunctionWithoutPartition,
+    D013DeeplyNestedSubqueries,
+)
+
+
+def test_d001_triggers_on_partitioned_table_without_partition_filter() -> None:
+    rule = D001FullScanOnPartitionedTable()
+    ctx = QueryContext(
+        sql="SELECT count() FROM events WHERE user_id = 42",
+        schema_ddl="CREATE TABLE events (ts DateTime, user_id UInt64) ENGINE = MergeTree PARTITION BY toYYYYMM(ts) ORDER BY user_id",
+    )
+    assert rule.check(ctx) is not None
+
+
+def test_d001_requires_schema_context() -> None:
+    assert D001FullScanOnPartitionedTable().check(QueryContext(sql="SELECT count() FROM events")) is None
+
+
+def test_d002_triggers_on_cross_join() -> None:
+    ctx = QueryContext(sql="SELECT * FROM events CROSS JOIN users")
+    assert D002CrossJoinRisk().check(ctx) is not None
+
+
+def test_d002_no_trigger_on_equality_join() -> None:
+    ctx = QueryContext(sql="SELECT * FROM events e JOIN users u ON e.user_id = u.id")
+    assert D002CrossJoinRisk().check(ctx) is None
+
+
+def test_d005_triggers_on_leading_wildcard_like() -> None:
+    assert D005LeadingWildcardLike().check(QueryContext(sql="SELECT * FROM logs WHERE msg LIKE '%error'")) is not None
+
+
+def test_d005_no_trigger_on_prefix_like() -> None:
+    assert D005LeadingWildcardLike().check(QueryContext(sql="SELECT * FROM logs WHERE msg LIKE 'error%'")) is None
+
+
+def test_d006_triggers_when_arrayjoin_before_where() -> None:
+    assert D006ArrayJoinBeforeFilter().check(QueryContext(sql="SELECT arrayJoin(tags) AS tag FROM events WHERE user_id = 1")) is not None
+
+
+def test_d006_no_trigger_when_filter_before_arrayjoin() -> None:
+    assert D006ArrayJoinBeforeFilter().check(QueryContext(sql="SELECT * FROM events WHERE id IN (SELECT arrayJoin(ids) FROM ids)")) is None
+
+
+def test_d008_triggers_on_sample_without_sample_by() -> None:
+    ctx = QueryContext(sql="SELECT * FROM events SAMPLE 0.1", schema_ddl="CREATE TABLE events (id UInt64) ENGINE = MergeTree ORDER BY id")
+    assert D008SampleWithoutSampleBy().check(ctx) is not None
+
+
+def test_d008_no_trigger_when_schema_has_sample_by() -> None:
+    ctx = QueryContext(sql="SELECT * FROM events SAMPLE 0.1", schema_ddl="CREATE TABLE events (id UInt64) ENGINE = MergeTree ORDER BY id SAMPLE BY id")
+    assert D008SampleWithoutSampleBy().check(ctx) is None
+
+
+def test_d009_triggers_on_nullable_without_nulls() -> None:
+    ctx = QueryContext(
+        sql="SELECT count() FROM events",
+        schema_ddl="CREATE TABLE events (status Nullable(String)) ENGINE = MergeTree ORDER BY tuple()",
+        environment={"null_counts": {"status": 0}},
+    )
+    assert D009NullableWithoutNeed().check(ctx) is not None
+
+
+def test_d009_no_trigger_when_nulls_are_present() -> None:
+    ctx = QueryContext(sql="CREATE TABLE events (status Nullable(String))", environment={"null_counts": {"status": 12}})
+    assert D009NullableWithoutNeed().check(ctx) is None
+
+
+def test_d010_triggers_on_unused_inner_projection() -> None:
+    ctx = QueryContext(sql="SELECT id FROM (SELECT id, payload FROM events)")
+    assert D010UnusedColumnsInSelect().check(ctx) is not None
+
+
+def test_d010_no_trigger_when_inner_projection_is_used() -> None:
+    ctx = QueryContext(sql="SELECT id, payload FROM (SELECT id, payload FROM events)")
+    assert D010UnusedColumnsInSelect().check(ctx) is None
+
+
+def test_d011_triggers_on_join_cast() -> None:
+    ctx = QueryContext(sql="SELECT * FROM events e JOIN users u ON toUInt64(e.user_id) = u.id")
+    assert D011ImplicitTypeCoercionInJoin().check(ctx) is not None
+
+
+def test_d011_no_trigger_without_join_cast() -> None:
+    ctx = QueryContext(sql="SELECT * FROM events e JOIN users u ON e.user_id = u.id")
+    assert D011ImplicitTypeCoercionInJoin().check(ctx) is None
+
+
+def test_d012_triggers_on_window_without_partition() -> None:
+    ctx = QueryContext(sql="SELECT row_number() OVER (ORDER BY ts) FROM events")
+    assert D012WindowFunctionWithoutPartition().check(ctx) is not None
+
+
+def test_d012_no_trigger_with_partition() -> None:
+    ctx = QueryContext(sql="SELECT row_number() OVER (PARTITION BY user_id ORDER BY ts) FROM events")
+    assert D012WindowFunctionWithoutPartition().check(ctx) is None
+
+
+def test_d013_triggers_on_deeply_nested_subqueries() -> None:
+    sql = "SELECT * FROM (SELECT * FROM (SELECT * FROM (SELECT * FROM (SELECT * FROM events))))"
+    assert D013DeeplyNestedSubqueries().check(QueryContext(sql=sql)) is not None
+
+
+def test_d013_no_trigger_on_shallow_query() -> None:
+    assert D013DeeplyNestedSubqueries().check(QueryContext(sql="SELECT * FROM (SELECT * FROM events)")) is None
diff --git a/tests/rules/test_environment_rules.py b/tests/rules/test_environment_rules.py
new file mode 100644
index 00000000..ac10bd97
--- /dev/null
+++ b/tests/rules/test_environment_rules.py
@@ -0,0 +1,92 @@
+from collections.abc import Callable
+
+import pytest
+
+from clickadvisor.core.models import QueryContext
+from clickadvisor.rules.base import Rule
+from clickadvisor.rules.environment import (
+    E001MaxThreadsVsCpuCores,
+    E002MaxMemoryUsageVsTotalRam,
+    E003ExternalGroupByThreshold,
+    E004MarkCacheSizing,
+    E005UncompressedCacheSizing,
+    E006MergeSettingsForDiskType,
+    E007StoragePolicyRecommendation,
+    E008JitCompilationToggle,
+    E009QueryCacheForRepeatedWorkload,
+    E010PartsToThrowInsertTuning,
+    E011DistributedConnectionsSizing,
+    E012JoinUseNullsOverhead,
+    E013AsyncInsertBusyTimeoutTuning,
+    E014MergeMaxBytesForLevel,
+    E015OptimizeAggregationInOrder,
+    E016JoinAlgorithmForLargeTables,
+    E017DistributedAggregationMemoryEfficient,
+    E018InputFormatForBulkInserts,
+    E019PreferLocalhostReplica,
+    E020MaxExecutionTimeForProtection,
+)
+
+RuleFactory = Callable[[], Rule]
+
+
+@pytest.mark.parametrize(
+    ("factory", "environment"),
+    [
+        (E001MaxThreadsVsCpuCores, {"settings": {"max_threads": 64}, "hardware": {"cpu_cores": 8}}),
+        (E002MaxMemoryUsageVsTotalRam, {"settings": {"max_memory_usage": 90}, "hardware": {"ram_bytes": 100}}),
+        (E003ExternalGroupByThreshold, {"settings": {"max_memory_usage": 1000}}),
+        (E004MarkCacheSizing, {"caches": {"mark_cache_size": 100, "active_marks_bytes": 1000}}),
+        (E005UncompressedCacheSizing, {"workload": {"repeated_reads": True}, "caches": {"use_uncompressed_cache": False}}),
+        (E006MergeSettingsForDiskType, {"hardware": {"disk_type": "hdd"}, "settings": {"background_pool_size": 32}}),
+        (E007StoragePolicyRecommendation, {"storage": {"has_hot_cold_data": True}}),
+        (E008JitCompilationToggle, {"workload": {"complex_expressions": True}, "settings": {"compile_expressions": False}}),
+        (E009QueryCacheForRepeatedWorkload, {"workload": {"repeated_queries": True}, "settings": {"use_query_cache": False}}),
+        (E010PartsToThrowInsertTuning, {"system_metrics": {"active_parts": 900}, "settings": {"parts_to_throw_insert": 1000}}),
+        (E011DistributedConnectionsSizing, {"cluster": {"shards": 4, "replicas": 2}, "settings": {"max_distributed_connections": 4}}),
+        (E012JoinUseNullsOverhead, {"settings": {"join_use_nulls": True}}),
+        (E013AsyncInsertBusyTimeoutTuning, {"settings": {"async_insert": True, "async_insert_busy_timeout_ms": 10}}),
+        (E014MergeMaxBytesForLevel, {"workload": {"large_parts": True}, "settings": {"max_bytes_to_merge_at_max_space_in_pool": 0}}),
+        (E015OptimizeAggregationInOrder, {"workload": {"group_by_order_key_prefix": True}, "settings": {"optimize_aggregation_in_order": False}}),
+        (E016JoinAlgorithmForLargeTables, {"workload": {"large_join": True}, "settings": {"join_algorithm": "hash"}}),
+        (E017DistributedAggregationMemoryEfficient, {"workload": {"distributed_aggregation": True}, "settings": {"distributed_aggregation_memory_efficient": False}}),
+        (E018InputFormatForBulkInserts, {"workload": {"bulk_inserts": True, "insert_format": "jsonEachRow"}}),
+        (E019PreferLocalhostReplica, {"cluster": {"has_local_replicas": True}, "settings": {"prefer_localhost_replica": False}}),
+        (E020MaxExecutionTimeForProtection, {"workload": {"user_facing": True}, "settings": {"max_execution_time": 0}}),
+    ],
+)
+def test_environment_rules_trigger_with_required_context(factory: RuleFactory, environment: dict[str, object]) -> None:
+    rule = factory()
+    finding = rule.check(QueryContext(sql="SELECT count() FROM events", environment=environment))
+    assert finding is not None
+    assert finding.rule_id == rule.rule_id
+    assert finding.confidence == "advisory"
+
+
+@pytest.mark.parametrize(
+    "factory",
+    [
+        E001MaxThreadsVsCpuCores,
+        E002MaxMemoryUsageVsTotalRam,
+        E003ExternalGroupByThreshold,
+        E004MarkCacheSizing,
+        E005UncompressedCacheSizing,
+        E006MergeSettingsForDiskType,
+        E007StoragePolicyRecommendation,
+        E008JitCompilationToggle,
+        E009QueryCacheForRepeatedWorkload,
+        E010PartsToThrowInsertTuning,
+        E011DistributedConnectionsSizing,
+        E012JoinUseNullsOverhead,
+        E013AsyncInsertBusyTimeoutTuning,
+        E014MergeMaxBytesForLevel,
+        E015OptimizeAggregationInOrder,
+        E016JoinAlgorithmForLargeTables,
+        E017DistributedAggregationMemoryEfficient,
+        E018InputFormatForBulkInserts,
+        E019PreferLocalhostReplica,
+        E020MaxExecutionTimeForProtection,
+    ],
+)
+def test_environment_rules_do_not_fire_without_environment(factory: RuleFactory) -> None:
+    assert factory().check(QueryContext(sql="SELECT count() FROM events")) is None
diff --git a/tests/rules/test_tier2_rules.py b/tests/rules/test_tier2_rules.py
new file mode 100644
index 00000000..b91d8d8e
--- /dev/null
+++ b/tests/rules/test_tier2_rules.py
@@ -0,0 +1,104 @@
+from clickadvisor.core.models import QueryContext
+from clickadvisor.rules.tier2 import (
+    R101OrderByKeyRedesign,
+    R102SkipIndexRecommendation,
+    R103ProjectionRecommendation,
+    R104PrimaryKeyRefinement,
+    R105PartitionStrategyAdjustment,
+    R106LowCardinalityConversion,
+    R107CodecRecommendation,
+    R108PrewhereInjection,
+    R109JoinReorderSmallOnRight,
+    R110JoinAlgorithmChoice,
+    R111MaterializedViewRecommendation,
+    R112GroupByMultipassSplit,
+)
+
+
+def assert_advisory(rule_id: str, ctx: QueryContext) -> None:
+    rule = {
+        "R-101": R101OrderByKeyRedesign,
+        "R-102": R102SkipIndexRecommendation,
+        "R-103": R103ProjectionRecommendation,
+        "R-104": R104PrimaryKeyRefinement,
+        "R-105": R105PartitionStrategyAdjustment,
+        "R-106": R106LowCardinalityConversion,
+        "R-107": R107CodecRecommendation,
+        "R-108": R108PrewhereInjection,
+        "R-109": R109JoinReorderSmallOnRight,
+        "R-110": R110JoinAlgorithmChoice,
+        "R-111": R111MaterializedViewRecommendation,
+        "R-112": R112GroupByMultipassSplit,
+    }[rule_id]()
+    finding = rule.check(ctx)
+    assert finding is not None
+    assert finding.rule_id == rule_id
+    assert finding.tier == "2"
+    assert finding.confidence == "advisory"
+
+
+def test_r101_order_by_tuple_schema() -> None:
+    assert_advisory("R-101", QueryContext(sql="SELECT count() FROM events", schema_ddl="CREATE TABLE events (id UInt64) ENGINE = MergeTree ORDER BY tuple()"))
+    assert R101OrderByKeyRedesign().check(QueryContext(sql="SELECT count() FROM events")) is None
+
+
+def test_r102_skip_index_for_substring_search() -> None:
+    assert_advisory("R-102", QueryContext(sql="SELECT * FROM logs WHERE msg LIKE '%timeout%'"))
+    assert R102SkipIndexRecommendation().check(QueryContext(sql="SELECT * FROM logs WHERE msg LIKE 'timeout%'")) is None
+
+
+def test_r103_projection_for_repeated_group_by() -> None:
+    ctx = QueryContext(sql="SELECT user_id, count() FROM events GROUP BY user_id", environment={"workload": {"repeated_group_by": True}})
+    assert_advisory("R-103", ctx)
+    assert R103ProjectionRecommendation().check(QueryContext(sql="SELECT user_id, count() FROM events GROUP BY user_id")) is None
+
+
+def test_r104_primary_key_refinement() -> None:
+    ctx = QueryContext(sql="SELECT * FROM events WHERE user_id = 42", schema_ddl="CREATE TABLE events (user_id UInt64, ts DateTime) ENGINE = MergeTree ORDER BY (user_id, ts)")
+    assert_advisory("R-104", ctx)
+    assert R104PrimaryKeyRefinement().check(QueryContext(sql="SELECT * FROM events", schema_ddl=ctx.schema_ddl)) is None
+
+
+def test_r105_partition_strategy_adjustment() -> None:
+    assert_advisory("R-105", QueryContext(sql="SELECT count() FROM events", environment={"system_metrics": {"active_parts": 1000, "active_partitions": 5}}))
+    assert R105PartitionStrategyAdjustment().check(QueryContext(sql="SELECT count() FROM events")) is None
+
+
+def test_r106_lowcardinality_conversion() -> None:
+    schema = "CREATE TABLE events (country String, ts DateTime) ENGINE = MergeTree ORDER BY ts"
+    assert_advisory("R-106", QueryContext(sql="SELECT country FROM events", schema_ddl=schema))
+    assert R106LowCardinalityConversion().check(QueryContext(sql="SELECT count() FROM events")) is None
+
+
+def test_r107_codec_recommendation() -> None:
+    schema = "CREATE TABLE events (ts DateTime, value UInt64) ENGINE = MergeTree ORDER BY ts"
+    assert_advisory("R-107", QueryContext(sql="SELECT count() FROM events", schema_ddl=schema, environment={"workload": {"compression_sensitive": True}}))
+    assert R107CodecRecommendation().check(QueryContext(sql="SELECT count() FROM events")) is None
+
+
+def test_r108_prewhere_injection() -> None:
+    assert_advisory("R-108", QueryContext(sql="SELECT * FROM events WHERE user_id = 42", environment={"workload": {"wide_table": True}}))
+    assert R108PrewhereInjection().check(QueryContext(sql="SELECT * FROM events PREWHERE user_id = 42")) is None
+
+
+def test_r109_join_reorder() -> None:
+    ctx = QueryContext(sql="SELECT * FROM small s JOIN large l ON s.id = l.id", environment={"workload": {"right_table_larger_than_left": True}})
+    assert_advisory("R-109", ctx)
+    assert R109JoinReorderSmallOnRight().check(QueryContext(sql="SELECT * FROM a JOIN b ON a.id = b.id")) is None
+
+
+def test_r110_join_algorithm_choice() -> None:
+    assert_advisory("R-110", QueryContext(sql="SELECT * FROM a JOIN b ON a.id = b.id", environment={"workload": {"large_join": True}}))
+    assert R110JoinAlgorithmChoice().check(QueryContext(sql="SELECT * FROM a")) is None
+
+
+def test_r111_materialized_view_recommendation() -> None:
+    ctx = QueryContext(sql="SELECT user_id, count() FROM events GROUP BY user_id", environment={"workload": {"repeated_expensive_query": True}})
+    assert_advisory("R-111", ctx)
+    assert R111MaterializedViewRecommendation().check(QueryContext(sql="SELECT count() FROM events")) is None
+
+
+def test_r112_groupby_multipass_split() -> None:
+    ctx = QueryContext(sql="SELECT user_id, count() FROM events GROUP BY user_id", environment={"workload": {"high_cardinality_group_by": True}})
+    assert_advisory("R-112", ctx)
+    assert R112GroupByMultipassSplit().check(QueryContext(sql="SELECT count() FROM events")) is None
diff --git a/tests/test_environment_cli.py b/tests/test_environment_cli.py
new file mode 100644
index 00000000..25fa9f8a
--- /dev/null
+++ b/tests/test_environment_cli.py
@@ -0,0 +1,46 @@
+import json
+
+from typer.testing import CliRunner
+
+from clickadvisor.cli.app import app
+
+
+def test_analyze_accepts_environment_json(tmp_path) -> None:
+    sql_path = tmp_path / "query.sql"
+    env_path = tmp_path / "environment.json"
+    sql_path.write_text("SELECT count() FROM events", encoding="utf-8")
+    env_path.write_text(json.dumps({"settings": {"join_use_nulls": True}}), encoding="utf-8")
+
+    result = CliRunner().invoke(
+        app,
+        [
+            "analyze",
+            "--sql",
+            str(sql_path),
+            "--environment",
+            str(env_path),
+            "--output-format",
+            "json",
+            "--ch-version",
+            "25.3",
+        ],
+    )
+
+    assert result.exit_code == 0
+    payload = json.loads(result.stdout)
+    assert any(finding["rule_id"] == "E-012" for finding in payload["findings"])
+
+
+def test_analyze_rejects_invalid_environment_json(tmp_path) -> None:
+    sql_path = tmp_path / "query.sql"
+    env_path = tmp_path / "environment.json"
+    sql_path.write_text("SELECT count() FROM events", encoding="utf-8")
+    env_path.write_text("[1, 2, 3]", encoding="utf-8")
+
+    result = CliRunner().invoke(
+        app,
+        ["analyze", "--sql", str(sql_path), "--environment", str(env_path)],
+    )
+
+    assert result.exit_code != 0
+    assert "environment JSON must be an object" in result.stderr