fix(pegas-memory): web fetch SOCKS5h proxy fallback + stale_guard supported double-count fix

- ingest.parse_web: direct fetch; on block/timeout retry via socks5h://127.0.0.1:1080 (SOCKS_FALLBACK_PROXY); raises ValueError if both fail.
- config: add SOCKS_FALLBACK_PROXY default (socks5h://127.0.0.1:1080).
- self_improve.stale_guard: supported LLM verdict now sets status_change='supported' (authoritative), no fallthrough to age fallback -> no double-checked/skipped_no_source.
- README: document auto-fallback.

AGENTS.md

@@ -45,6 +45,8 @@ Read this file before answering any question about knowledge, memory, or researc
 | "Show one gap" | `memory_get_gap` |
 | "Delete a gap" | `memory_delete_gap` |
 | "Run logical inference" | `memory_infer` |
+| "Commit verified answers from closed gaps into atoms" | `memory_commit(gap_ids, dedup?)` — закрывает «дыра commit»: answer пробела пишется в атомы (dedup обязательный); без верифицированного ответа пробел не трогает |
+| "Mark stale atoms" | `memory_stale(atom_id?/all?, limit?, sample?, min_confidence?, age_only?, max_age_days?)` — гвард сталкности: content-staleness (рефетч + LLM, нужен прокси для заруб.) или age-staleness (proxy-free по `fetched_at`) |
 
 **Ambiguity rule:** if the question is about *knowledge content* → memory tools.
 If it is about *the outside world right now* → `web_search` (built-in).

ARCHITECTURE.md

@@ -1,7 +1,7 @@
 # Pegas Memory — Architecture
 
 Pegas Memory is the long-term memory and knowledge base for the **Pegas** research agent.
-It is a Python service exposed to the agent as an **MCP server** (16 tools).
+It is a Python service exposed to the agent as an **MCP server** (18 tools).
 
 ---
 
@@ -14,7 +14,7 @@ It is a Python service exposed to the agent as an **MCP server** (16 tools).
                  └───────────────────┬──────────────────────┘
                                      │  JSON-RPC 2.0 / stdio
                  ┌───────────────────▼──────────────────────┐
-                 │  mcp_server.py  (tool layer, 16 tools)   │
+                 │  mcp_server.py  (tool layer, 18 tools)   │
                  └───────────────────┬──────────────────────┘
                                      │  Python API
         ┌────────────────────────────┼────────────────────────────┐

EXAMPLES.md

@@ -246,6 +246,44 @@ memory_query_graph(kind = "depends_on", top_k = 10)
 
 ---
 
+## Example 9 — Self-improvement (commit + staleness)
+
+Commit a verified answer from a closed gap into atoms (closes the «commit hole»):
+
+```
+memory_commit(gap_ids = [3], dedup = true)
+```
+
+**Result** (actual, abridged):
+
+```json
+{"committed": 1, "resolved": 1, "skipped_empty": 0, "skipped_duplicate": 0,
+ "atoms": [218]}
+```
+
+Mark stale atoms — content-staleness (re-fetch + LLM, needs proxy for foreign sites)
+or age-staleness (proxy-free, by `fetched_at`):
+
+```
+memory_stale(atom_id = 126)
+memory_stale(all = true, limit = 20, sample = true,
+             min_confidence = 0.5, age_only = true, max_age_days = 90)
+```
+
+**Result** (abridged):
+
+```json
+{"checked": 0, "stale": 1, "supported": 0, "skipped_no_source": 0,
+ "atom_ids": [126], "reasons": {"126": "content_age 10d"}}
+```
+
+> `age_only = true` skips the network re-fetch and flags atoms whose source
+> `fetched_at` is older than `max_age_days` (default 90). For foreign sites
+> (memgraph.com, cloudflare/geo-blocked) set `PROXY_URL` or the re-fetch
+> returns `ConnectTimeout` and falls back to age-staleness.
+
+---
+
 ## Anti-patterns (do not do this)
 
 | Anti-pattern | Why it is wrong | Do instead |

README.md

@@ -45,7 +45,8 @@ schema.sql         — DDL PostgreSQL
   `memory_search`, `memory_store`, `memory_ingest`, `memory_stats`, `memory_get_context`,
   `memory_verify`, `memory_store_code`, `memory_search_code`, `memory_code_graph`,
   `memory_infer`, `memory_query_graph`,
-  `memory_add_gap`, `memory_list_gaps`, `memory_get_gap`, `memory_update_gap`, `memory_delete_gap`.
+  `memory_add_gap`, `memory_list_gaps`, `memory_get_gap`, `memory_update_gap`, `memory_delete_gap`,
+  `memory_commit`, `memory_stale`.
 - **Регистрация**: в `~/.dsh/profiles/web/cordis.patch.yml` через `@deepseek-ai/dsh-mcp-client`.
 - **Live-reload**: профиль web имеет `patchReload: live` — изменения конфига
   подхватываются без рестарта (проверено: сервер спавнится автоматически).
@@ -75,7 +76,10 @@ Cloudflare anti-bot (neo4j.com — 403). Поведение research-цикла:
 ```bash
 export PROXY_URL="socks5://127.0.0.1:1080"   # hysteria2
 ```
-Когда прокси задан, `parse_web` ходит через него; иначе — напрямую.
+
+**Авто-повор:** `parse_web` сначала грузит URL напрямую; при таймауте/блокировке повторяет запрос через `socks5h://127.0.0.1:1080` (дефолт,
+`SOCKS_FALLBACK_PROXY`). Если и проки недоступен — падает с
+`ValueError: no main content ... (direct + ... fallback failed)`.
 Поддержка SOCKS5 уже установлена (`httpx[socks]`).
 
 ## Управление пробелами (gaps)
@@ -94,3 +98,21 @@ export PROXY_URL="socks5://127.0.0.1:1080"   # hysteria2
 Причины: `low_coverage`, `low_confidence`, `stale`, `contested`, `user_query`.
 
 Работает целиком на PostgreSQL — без веб-поиска. Поиск выполняет модель встроенным `web_search`.
+
+## Самообучение (self-improvement)
+
+Два инструмента закрывают петлю commit/maintain (см. `ROADMAP.md`):
+
+| Операция | Инструмент |
+|---|---|
+| Записать верифицированный answer закрытого пробела в атомы | `memory_commit(gap_ids, dedup?)` |
+| Пометить устаревшие атомы в `stale` | `memory_stale(atom_id?/all?, limit?, sample?, min_confidence?, age_only?, max_age_days?)` |
+
+- **`memory_commit`** — для каждого `gap_id` ищет верифицированные атомы-ответы, дистиллирует
+  канонический атом (обязательный `dedup`) и закрывает пробел. Если верифицированного ответа нет —
+  пробел не трогает (`skipped_empty`).
+- **`memory_stale`** — два сигнала: **content-staleness** (рефетч источника + LLM-вердикт, для
+  зарубежных сайтов нужен `PROXY_URL`) и **age-staleness** (proxy-free, по `fetched_at`, порог
+  `max_age_days`, default 90). Лимит LLM-вызовов — `limit` + `sample`.
+
+Источники: `web`/`text`/`epub`/`pdf`. Проверено: рефетч memgraph.com без прокси → `ConnectTimeout`.

ROADMAP.md

@@ -92,6 +92,73 @@ outbox        — CDC-журнал для синка в Qdrant + Memgraph
 - Memgraph in-memory: объём графа ограничен RAM.
 - Дистилляция зависит от качества локальной LLM — нужна верификация.
 
-## Как проверять
+## Самообучение (self-improvement) — статус и планы
+
+### Что работает (4 механизма, но НЕ замкнутый loop)
+
+1. **Верификация** (`verify_atom`/`memory_verify` + `verify.py`). LLM self-critique
+   (`duplicate | corroborates | contradicts`) + **межисточная конфирмация** (≥1 другой
+   `doc_id`, cosine ≥ 0.90) → ставка растёт, статус `verified`. Самое зрелое звено.
+2. **Логическая инференция** (`memory_infer` + `inference.py`). Применение правил к рёбрам
+   графа до fixpoint (`imports → depends_on` и т.д.). ⚠️ Даст **только связи**, не новые факты.
+   Правила **статичные/вручную написанные**, новых не выводит сама.
+3. **Синтез контекста** (`memory_get_context`). Ищет Qdrant, кластеризует + суммирует атомы.
+   Может **создавать gap** из исходного текста → есть *реактивное* появление вопроса из источника.
+4. **Эпизодические следы** (`pm_episodic_trace`). План/действие/наблюдение/рефлексия —
+   отдельно от БЗ. Самый «рефлексивный», но **условный**, не привязанный к gap.
+
+### Что не хватает (дыры в петлях, а не в примитивах)
+
+| Пробел | Почему критично |
+|---|---|
+| **① Цепочка gap↛answer («дыра commit»)** | При закрытии gap **ответ не автоматически
+   пишется обратно в атомы** → знания улетучиваются, приходится пере-исследовать. |
+| **② Самообучение схемы инференции** | Ничего не *предлагает новые правила* из повторяющихся
+   `verified`-атомов. Это как раз слабое место Hermes-аналога. |
+| **③ Сталкнутость/забвение (staleness guard)** | `verify` ставит `verified`, но **ничто не
+   помечает `stale`** пере-проверкой по свежему источнику. Путь →`stale` не идёт автоматически. |
+| **④ Поверхность противоречий** | `verify` ловит конфликт только в окрестности атома. Ничего
+   не сканирует весь граф → противоречия не переходят в gap автоматически. |
+| **⑤ Процедурная/рецепт-память** | Повторяющиеся успешные воркфлоу не ловятся как переиспользуемые атомы. |
+| **⑥ Пассивное создание gap** | Gap создаются через `add_gap` (вручную) или из `memory_get_context`.
+   Нет авто-обнаружения из не-верифицированного/переносного контента. |
+
+### Приоритет
+
+Hermes-анализ показал: Pegas сильна в **«detect»** (verify, infer), слаба в **«commit»** и
+**«maintain»**. Слабые места — **feedback-петли**, а не примитивы. Поэтому максимальный выигрыш:
+① дыра «commit» × ③ гвард сталкнутости × ② самообучение правил инференции — это **закрывает loop**.
+
+### Планируемые улучшения (Tier)
+
+- **Tier 1 — высокий приоритет, безопасно / small**
+  - ① **Закрыть дыру** — после `verify`, подняв атомы из gap, запустить синтез обратно в
+    `memory_ingest/synthesis` (письмо верифицированного ответа в атомы).
+  - ③ **Гвард сталкнутости** — `update_status`: пере-`web_search` по URL/date атома;
+    противоречит/устарел → `stale`. Внедряет путь →`stale`, который сейчас не ходит.
+- **Tier 2 — средний приоритет**
+  - ② **Самообучение правил инференции** — `memory_infer` **предлагает кандидаты**
+    `logical_rule` на ревью (никогда не применяет автоматически).
+  - ⑤ **Атомные паттерны действий** — агент действует (это я через `mcp__pegas-memory__*`),
+    поэтому повторы успешных `/запрос` или воркфлоу ≥N раз с `verified`-результатом →
+    **атомный** паттерн-рецепт (не файлы Hermes, а атомно с dedup).
+- **Tier 3 — структурные**
+  - ④ **Скунт-скан** — периодический бэч-проход графа на противоречия → авто gap.
+  - ⑥ **Авто-обнаружение gap** — `verify` открывает gap на конфликт/дыру в источнике.
+
+### Atom-efficiency guardrail (обязательное)
+
+Чтобы self-improvement **не скатился в гермесовский файловый балласт**, все записи и
+обновления подчинены лимитам эффективности атомной памяти (преимущество Peras над Hermes):
+
+- **Dedup обязателен** при записи (`memory_store`/synthesis): cosine ≥ 0.90, без дублей.
+- **Лимит web/пере-записей** (staleness-③): batchами + сэмплом, не per-atom без порога.
+- **Bound на рост графа/атомов**: Memgraph in-memory → RAM-лимит, авто-`stale`/архив старых.
+- **Атомно, а не файльно**: паттерны-рецепты хранятся как типизированные атомы, не как
+  накопитель markdown-файлов (как у Hermes) → no prompt-bloat.
+
+### Как проверять
 - recall@k на реальных запросах против существующей `knowledge_base` (bge-m3 384-dim).
 - Latency эмбеддинга + потребление GPU параллельно с LLM (a2000 12 GB).
+- **Самоеобучение**: доля авто-закрытых gap «commit»-петлёй, доля атомов, ушедших в `stale`
+  за квартал, доля предложенных → принято правил инференции (ревью-соотношение).

mcp_server.py

@@ -289,6 +289,55 @@ def _tool_verify(args):
     return json.dumps(result, ensure_ascii=False, indent=2), False
 
 
+def _tool_commit(args):
+    """Tier 1 (commit): закрыть дыру — answer закрытого gap пишется в атомы."""
+    from pegas_memory import self_improve
+    gap_ids = args.get("gap_ids")
+    try:
+        gap_ids = [int(x) for x in (gap_ids or [])]
+    except (TypeError, ValueError):
+        return json.dumps({"error": "gap_ids должен быть списком целых id"}, ensure_ascii=False), True
+    if not gap_ids:
+        return json.dumps({"error": "нужен gap_ids (список id закрытых пробелов)"}, ensure_ascii=False), True
+    dedup = bool(args.get("dedup", True))
+    try:
+        result = self_improve.commit_gaps(_conn(), gap_ids=gap_ids, dedup=dedup)
+    except ValueError as e:
+        return json.dumps({"error": str(e)}, ensure_ascii=False), True
+    return json.dumps(result, ensure_ascii=False, indent=2), False
+
+
+def _tool_stale(args):
+    """Tier 1 (maintain): гвард сталкнутости — атомы по свежему источнику → stale."""
+    from pegas_memory import self_improve
+    atom_id = args.get("atom_id")
+    all_ = bool(args.get("all", False))
+    if atom_id is not None:
+        try:
+            result = self_improve.stale_guard(
+                _conn(), atom_id=int(atom_id),
+                age_only=bool(args.get("age_only", False)),
+                max_age_days=int(args.get("max_age_days") or 90),
+            )
+        except ValueError as e:
+            return json.dumps({"error": str(e)}, ensure_ascii=False), True
+    elif all_:
+        try:
+            result = self_improve.stale_guard(
+                _conn(), all=True,
+                limit=min(int(args.get("limit") or 20), 200),
+                sample=bool(args.get("sample", True)),
+                min_confidence=float(args.get("min_confidence") or 0.0),
+                age_only=bool(args.get("age_only", False)),
+                max_age_days=int(args.get("max_age_days") or 90),
+            )
+        except ValueError as e:
+            return json.dumps({"error": str(e)}, ensure_ascii=False), True
+    else:
+        return json.dumps({"error": "укажи atom_id или all=true"}, ensure_ascii=False), True
+    return json.dumps(result, ensure_ascii=False, indent=2), False
+
+
 TOOLS = [
     {
         "name": "memory_search",
@@ -492,6 +541,35 @@ TOOLS = [
 
 
 
+    {
+        "name": "memory_commit",
+        "description": "Tier 1 (commit): закрыть «дыру commit» — верифицированный ответ закрытого пробела пишется обратно в атомы (не даёт знанию слетывать). Берет id пробелов, ищет верифицированные атомы-ответы, дистиллирует канонический атом с обязательным dedup и закрывает пробел. Atom-efficiency: dedup по умолчанию.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "gap_ids": {"type": "array", "items": {"type": "integer"}, "description": "Список id закрытых пробелов, answer которых надо записать в атомы."},
+                "dedup": {"type": "boolean", "description": "Семантическая дедупляция при записи (default true)."},
+            },
+            "required": ["gap_ids"],
+        },
+    },
+    {
+        "name": "memory_stale",
+        "description": "Tier 1 (maintain): гвард сталкнутости. Пере-проверяет атом(ы) по свежу источнику: противоречит/устарел → status='stale'. Лимит LLM-вызовов через limit/sample. atom_id — один атом; all — пачка (с limit/sample).",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "atom_id": {"type": "integer", "description": "ID атома для проверки на актуальность."},
+                "all": {"type": "boolean", "description": "Прогнать пачку атомов (с limit/sample)."},
+                "limit": {"type": "integer", "description": "Лимит атомов в пачке (1-200)."},
+                "sample": {"type": "boolean", "description": "Сэмпл вместо проверки всех (по умолчанию true при all)."},
+                "min_confidence": {"type": "number", "description": "Мин. уверенность атомов в пачке (default 0.0)."},
+                "age_only": {"type": "boolean", "description": "Без рефечя: только age-staleness по fetched_at (proxy-free)."},
+                "max_age_days": {"type": "integer", "description": "Возраст контента в днях, старше которого → stale (default 90)."},
+            },
+            "required": [],
+        },
+    },
     {
         "name": "memory_get_context",
         "description": "Собрать релевантный контекст из твоей базы для ответа (поиск по атомам + форматирование). "
@@ -551,6 +629,8 @@ def handle(msg):
             "memory_infer": _tool_infer,
             "memory_query_graph": _tool_query_graph,
             "memory_verify": _tool_verify,
+            "memory_commit": _tool_commit,
+            "memory_stale": _tool_stale,
             "memory_get_context": _tool_get_context,
         }
         fn = handlers.get(tool)

pegas_memory/config.py

@@ -19,5 +19,6 @@ ATOM_STATUSES = ("candidate", "verified", "stale", "contested")
 
 DEDUP_THRESHOLD = float(os.getenv("DEDUP_THRESHOLD", "0.90"))  # порог семантической дедупликации
 
-PROXY_URL = os.getenv("PROXY_URL", "")  # опционально: socks5://127.0.0.1:1080 или http://...
+PROXY_URL = os.getenv("PROXY_URL", "")  # явный прокси: используется всегда (socks5://… или http://…)
+SOCKS_FALLBACK_PROXY = os.getenv("SOCKS_FALLBACK_PROXY", "socks5h://127.0.0.1:1080")  # авто-повор через прокси при ошибке прямого запроса
 USER_AGENT = os.getenv("USER_AGENT", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36")

pegas_memory/db.py

@@ -30,8 +30,8 @@ def upsert_document(conn, *, kind: str, uri: str | None, title: str | None,
             if row:
                 return row["id"]
         cur.execute(
-            """INSERT INTO pm_document (kind, uri, title, content_hash, authority, meta)
-               VALUES (%s, %s, %s, %s, %s, %s)
+            """INSERT INTO pm_document (kind, uri, title, content_hash, authority, meta, fetched_at)
+               VALUES (%s, %s, %s, %s, %s, %s, now())
                ON CONFLICT (uri) DO UPDATE SET title = EXCLUDED.title
                RETURNING id""",
             (kind, uri, title, content_hash, authority, json.dumps(meta or {}, ensure_ascii=False)),

pegas_memory/ingest.py

@@ -59,26 +59,49 @@ def parse_epub(path: str) -> ParsedDoc:
     doc.kind = "epub"
     return doc
 
-def _http_client():
-    """httpx.Client с опциональным прокси и единым User-Agent."""
+def _http_client(proxy=None, timeout: int = 30):
+    """httpx.Client с единым User-Agent.
+
+    proxy: явный прокси (None = прямой запрос). Если проси не передан, но
+    config.PROXY_URL задан — используется он.
+    """
     import httpx
     from . import config
-    kwargs = dict(timeout=30, follow_redirects=True,
+    kwargs = dict(timeout=timeout, follow_redirects=True,
                   headers={"User-Agent": config.USER_AGENT})
-    if config.PROXY_URL:
-        kwargs["proxy"] = config.PROXY_URL
+    use_proxy = proxy or (config.PROXY_URL or None)
+    if use_proxy:
+        kwargs["proxy"] = use_proxy
     return httpx.Client(**kwargs)
 
-def parse_web(url: str) -> ParsedDoc:
+def _fetch_text(url: str, proxy: str | None = None) -> str | None:
+    """Fetch + extract main content. Tries direct; on error/empty retries `proxy`.
+
+    trafilatura.extract has no timeout -> fetch via httpx with timeout.
+    """
     import trafilatura
-    # fetch через httpx с таймаутом + прокси (trafilatura.fetch_url не имеет таймаута и виснет)
-    with _http_client() as client:
+    for _lbl, _prx in (("direct", None), ("proxy", proxy)):
+        try:
+            with _http_client(proxy=_prx, timeout=30) as client:
                 r = client.get(url)
             r.raise_for_status()
-    text = trafilatura.extract(r.text, include_comments=False, include_tables=True,
-                               url=url)
+            text = trafilatura.extract(r.text, include_comments=False,
+                                       include_tables=True, url=url)
+            if text:
+                return text
+        except Exception:
+            continue
+    return None
+
+def parse_web(url: str) -> ParsedDoc:
+    from . import config
+    # direct fetch + socks5h fallback on block/timeout
+    text = _fetch_text(url, proxy=config.SOCKS_FALLBACK_PROXY)
     if not text:
-        raise ValueError(f"trafilatura: no main content for {url}")
+        raise ValueError(
+            f"no main content for {url} "
+            f"(direct + {config.SOCKS_FALLBACK_PROXY} fallback failed)"
+        )
     doc = parse_text(text, uri=url, title=url)
     doc.kind = "web"
     return doc

pegas_memory/self_improve.py

@@ -0,0 +1,229 @@
+"""Самообучение (Tier 1): закрытие «дыры commit» + гвард сталкнутости.
+
+Два механизма, закрывающих петлю commit/maintain (см. ROADMAP.md):
+  1. commit_gaps — после закрытия gap пишет верифицированный answer обратно
+                   в атомы, чтобы знание не слетивало и не пере-исследовалось.
+                   Atom-efficiency: обязательный dedup.
+  2. stale_guard — пере-проверяет атомы по свежу источнику; противоречит/
+                   устарел → status='stale'. Два сигнала:
+                     - content-staleness: рефетч источника + LLM-вердикт (нужен proxy
+                       для зарубежных сайтов);
+                     - age-staleness: fallback по `fetched_at` (proxy-free).
+                   Лимит LLM-вызовов (limit/sample).
+
+Лимиты: LLM-вызовы дорогие — каждый вызов инструмента лимитирован.
+"""
+from __future__ import annotations
+import json
+import random
+from datetime import datetime
+from . import config, db, research, qdrant_store
+from .ingest import parse_web
+
+
+def _fresh_source(uri: str) -> str | None:
+    """Свежный контент источника (прокси/UA/таймаут через ingest.parse_web)."""
+    try:
+        doc = parse_web(uri)
+    except Exception:
+        return None
+    return "\n\n".join(c.content for c in doc.chunks)
+
+
+def _stale_verdict(atom: dict, fresh_text: str, model: str | None) -> dict:
+    """LLM: поддерживает ли свежный источник утверждение атома?
+
+    Вернит {"verdict": "supported|contradicts|outdated", "confidence", "reason"}.
+    """
+    import httpx
+    m = model or config.LLM_MODEL
+    ctx = (atom.get("claim", "") or "") + "\n" + (atom.get("content", "") or "")[:600]
+    prompt = (
+        f"Свежный источник:\n{fresh_text[:4000]}\n\n"
+        f"Утверждение атома ({atom.get('type')}):\n{ctx[:500]}\n\n"
+        f"Оцени, свежный источник всё ещё ПОДДЕРЖИВАЕТ утверждение. "
+        f"Верни ТОЛЬКО JSON:\n"
+        f'{{"verdict": "supported|contradicts|outdated", "confidence": 0.0-1.0, '
+        f'"reason": "кратко"}}\n'
+        f"- supported: источник подтверждает/сохраняет утверждение\n"
+        f"- contradicts: источник теперь противоречит\n"
+        f"- outdated: информация устарела в источнике"
+    )
+    try:
+        with httpx.Client(timeout=120) as client:
+            r = client.post(f"{config.OLLAMA_URL}/api/chat", json={
+                "model": m,
+                "messages": [{"role": "user", "content": prompt}],
+                "stream": False,
+                "options": {"temperature": 0.0},
+            })
+            r.raise_for_status()
+            out = r.json()["message"]["content"].strip()
+        start, end = out.find("{"), out.rfind("}")
+        if start == -1 or end == -1:
+            return {"verdict": "supported", "confidence": 0.5, "reason": "parse error"}
+        v = json.loads(out[start:end + 1])
+    except Exception:
+        return {"verdict": "supported", "confidence": 0.5, "reason": "llm error"}
+    if v.get("verdict") not in ("supported", "contradicts", "outdated"):
+        v["verdict"] = "supported"
+    return v
+
+
+def commit_gaps(conn, *, gap_ids=None, dedup=True, model: str | None = None) -> dict:
+    """Закрывает дыра «commit»: answer закрытого gap пишется в атомы.
+
+    Для каждого gap из gap_ids: ищет verified-атомы, отвечающие на вопрос,
+    дистиллирует канонический атом-ответ, сохраняет с dedup, resolves gap.
+    Возвращает статистику. dedup — atom-efficiency guardrail (по умолчанию True).
+    """
+    from . import distill, dedup as _dedup
+
+    out = {"committed": 0, "skipped_duplicate": 0, "resolved": 0,
+           "skipped_empty": 0, "atoms": []}
+    if not gap_ids:
+        raise ValueError("нужен gap_ids (список id)")
+    for gid in gap_ids:
+        gid = int(gid)
+        gap = research.get_gap(conn, gid)
+        if not gap:
+            continue
+        q = gap["question"]
+        # релевантность — из Qdrant (рейтинг), статус — из Postgres (истина):
+        # verify меняет статус в PG, но не пушит projection, поэтому читаем status оттуда.
+        hits = qdrant_store.search(q, top_k=8)
+        if not hits:
+            out["skipped_empty"] += 1
+            continue
+        hit_ids = [int(h["id"]) for h in hits]
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT id, claim, content, status FROM pm_atom WHERE id = ANY(%s)",
+                (hit_ids,),
+            )
+            rows = {r["id"]: r for r in cur.fetchall()}
+        verified = [rows[hid] for hid in hit_ids if rows[hid]["status"] == "verified"]
+        if not verified:
+            # нет верифицированного ответа — не резать gap, агент должен его получить
+            out["skipped_empty"] += 1
+            continue
+        text = f"ВОПРОС: {q}\n\nОТВЕТ (верифицированные знания):\n" + "\n".join(
+            f"- {v.get('claim', '')}\n  {str(v.get('content', ''))[:300]}" for v in verified
+        )
+        atoms = distill.distill_chunk(text, model=model)
+        for a in atoms:
+            if a.get("confidence", 0.5) < 0.4:
+                continue
+            if dedup:
+                dup_id = _dedup.find_duplicate(
+                    (a.get("claim") or "") + " " + str(a["content"])[:800],
+                    threshold=config.DEDUP_THRESHOLD,
+                )
+                if dup_id is not None:
+                    out["skipped_duplicate"] += 1
+                    continue
+            atom_id, created = db.upsert_atom(
+                conn, type=a["type"], claim=a.get("claim"), content=a["content"],
+                confidence=a.get("confidence", 0.5), status="verified", doc_id=None,
+                metadata={"committed_from_gap": gid, "source_question": q},
+            )
+            if created:
+                out["committed"] += 1
+                out["atoms"].append(atom_id)
+                db.enqueue_outbox(conn, entity_type="atom", entity_id=atom_id)
+        research.update_gap(conn, gid, status="resolved")
+        out["resolved"] += 1
+    conn.commit()
+    return out
+
+
+def stale_guard(conn, *, atom_id=None, all: bool = False, limit: int = 20,
+                sample: bool = False, min_confidence: float = 0.0,
+                age_only: bool = False, max_age_days: int = 90,
+                model: str | None = None) -> dict:
+    """Гвард сталкнутости: пере-проверка атомов по свежему источнику.
+
+    atom_id — один атом (1 fetch + 1 LLM-вызов).
+    all   — пачка атомов; limit + sample лимитируют число LLM-вызовов.
+    Противоречит/устарел → status='stale'. Возвращает сводку.
+    """
+    ids = []
+    if atom_id is not None:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT id, type, claim, content, confidence, status, doc_id "
+                "FROM pm_atom WHERE id = %s",
+                (int(atom_id),),
+            )
+            row = cur.fetchone()
+        if row:
+            ids = [row]
+    elif all:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT id, type, claim, content, confidence, status, doc_id "
+                "FROM pm_atom WHERE confidence >= %s",
+                (min_confidence,),
+            )
+            rows = cur.fetchall()
+        if sample and rows:
+            n = min(int(limit), len(rows))
+            ids = random.sample(rows, n)
+        else:
+            ids = rows[:int(limit)]
+    else:
+        raise ValueError("укажи atom_id или all=true")
+
+    summary = {"checked": 0, "stale": 0, "supported": 0,
+               "skipped_no_source": 0, "atom_ids": [], "reasons": {}}
+    for atom in ids:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT d.uri, d.fetched_at FROM pm_document d WHERE id = %s",
+                (atom["doc_id"],),
+            )
+            doc = cur.fetchone()
+        if not doc or not doc.get("uri"):
+            summary["skipped_no_source"] += 1
+            continue
+
+        status_change = None
+        reason = None
+        if not age_only:
+            fresh = _fresh_source(doc["uri"])
+            if fresh:
+                v = _stale_verdict(atom, fresh, model)
+                summary["checked"] += 1
+                if v["verdict"] in ("contradicts", "outdated"):
+                    status_change = "stale"
+                    reason = f"source_{v['verdict']}: {v.get('reason', '')}"
+                else:
+                    # content-staleness re-fetch succeeded -> verdict is authoritative
+                    status_change = "supported"
+                    reason = "supported"
+        if status_change is None:
+            # proxy-free fallback: возраст контента
+            fa = doc.get("fetched_at")
+            if fa is None:
+                summary["skipped_no_source"] += 1
+                continue
+            tz = fa.tzinfo or datetime.timezone.utc
+            age_days = max(0, (datetime.now(tz=tz) - fa).days)
+            if age_days >= max_age_days:
+                status_change = "stale"
+                reason = f"content_age {age_days}d"
+            else:
+                reason = f"fresh {age_days}d"
+        if status_change == "stale":
+            with conn.cursor() as cur:
+                cur.execute(
+                    "UPDATE pm_atom SET status = 'stale', updated_at = now() WHERE id = %s",
+                    (atom["id"],),
+                )
+            summary["stale"] += 1
+            summary["atom_ids"].append(atom["id"])
+        else:
+            summary["supported"] += 1
+        summary["reasons"][atom["id"]] = reason
+    conn.commit()
+    return summary