From eca8a11e23617b288ff0bbdad9753f5729e8c832 Mon Sep 17 00:00:00 2001 From: n1kt4r Date: Thu, 15 Oct 2026 22:06:30 +0500 Subject: [PATCH] feat(pegas-memory): Tier-1 self-improvement tools (commit + staleness) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - memory_commit: closes the 'commit hole' — writes verified answers from closed gaps back into atoms (mandatory dedup); leaves gap open if no verified answer exists. Reads verified status from PostgreSQL (source of truth), not the Qdrant projection. - memory_stale: staleness guard with two signals — content-staleness (re-fetch + LLM verdict, needs proxy for foreign sites) and age-staleness (proxy-free fallback by pm_document.fetched_at, max_age_days). - db: populate pm_document.fetched_at on upsert (freshness signal). - docs: README/AGENTS/ARCHITECTURE/EXAMPLES updated (18 tools). - self_improve.py: new module (227 lines). --- AGENTS.md | 2 + ARCHITECTURE.md | 4 +- EXAMPLES.md | 38 ++++++ README.md | 21 +++- ROADMAP.md | 69 ++++++++++- mcp_server.py | 80 ++++++++++++ pegas_memory/db.py | 4 +- pegas_memory/self_improve.py | 227 +++++++++++++++++++++++++++++++++++ 8 files changed, 439 insertions(+), 6 deletions(-) create mode 100644 pegas_memory/self_improve.py diff --git a/AGENTS.md b/AGENTS.md index 3be9aec..ae5cd36 100644 --- a/AGENTS.md +++ b/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). diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 0fbba7d..466aa41 100644 --- a/ARCHITECTURE.md +++ b/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 ┌────────────────────────────┼────────────────────────────┐ diff --git a/EXAMPLES.md b/EXAMPLES.md index cc7d6b1..ee1144b 100644 --- a/EXAMPLES.md +++ b/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 | diff --git a/README.md b/README.md index 83a66c3..23e46df 100644 --- a/README.md +++ b/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` — изменения конфига подхватываются без рестарта (проверено: сервер спавнится автоматически). @@ -94,3 +95,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`. diff --git a/ROADMAP.md b/ROADMAP.md index 466ffd6..59b83ea 100644 --- a/ROADMAP.md +++ b/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` + за квартал, доля предложенных → принято правил инференции (ревью-соотношение). diff --git a/mcp_server.py b/mcp_server.py index a2550ac..ae4a9db 100644 --- a/mcp_server.py +++ b/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) diff --git a/pegas_memory/db.py b/pegas_memory/db.py index e31f91c..2bcda61 100644 --- a/pegas_memory/db.py +++ b/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)), diff --git a/pegas_memory/self_improve.py b/pegas_memory/self_improve.py new file mode 100644 index 0000000..c76cc92 --- /dev/null +++ b/pegas_memory/self_improve.py @@ -0,0 +1,227 @@ +"""Самообучение (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: + 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