fix: harden extraction batch handling

site/config.py

@@ -1,7 +1,7 @@
 """Конфигурация приложения — все настройки в одном месте."""
 import os
 
-VERSION = "3.0.0"
+VERSION = "3.0.3"
 
 
 def _load_env():

site/db/connection.py

@@ -12,7 +12,7 @@ import time
 import os
 from contextlib import contextmanager
 
-DB_PATH = "/tmp/contracts.db"
+DB_PATH = os.getenv("CONTRACTS_DB_PATH", "/tmp/contracts.db")
 _db_session_key = None
 _local = threading.local()
 
@@ -67,6 +67,19 @@ def _create_schema(conn):
             created_at TEXT DEFAULT (datetime('now'))
         );
 
+        CREATE TABLE IF NOT EXISTS extraction_jobs (
+            batch_id TEXT PRIMARY KEY,
+            owner TEXT NOT NULL,
+            state TEXT NOT NULL,
+            total INTEGER NOT NULL,
+            done INTEGER NOT NULL DEFAULT 0,
+            failed INTEGER NOT NULL DEFAULT 0,
+            classes_json TEXT NOT NULL DEFAULT '{}',
+            result_json TEXT,
+            error TEXT,
+            updated_at REAL NOT NULL
+        );
+
         CREATE TABLE IF NOT EXISTS contracts (
             id TEXT PRIMARY KEY,
             number TEXT NOT NULL,
@@ -332,6 +345,9 @@ def execute_returning(sql, params=None):
         sql = sql[:sql.upper().rfind(" RETURNING ")]
 
     cur = conn.execute(sql, params or [])
+    # The outer transaction owns commit and rollback; committing here would make
+    # its rollback ineffective and could expose only part of a multi-write update.
+    if not getattr(_local, "in_transaction", False):
         conn.commit()
     rowid = cur.lastrowid
 

site/db/dossiers.py

@@ -7,7 +7,7 @@
 import json
 import uuid
 
-from db.connection import execute, query, transaction
+from db.connection import execute, query
 
 
 def _json_value(value, default):
@@ -123,5 +123,4 @@ def list_dossier_documents(dossier_id):
 
 def clear_batch_dossiers(batch_id):
     """Удалить все досье batch; таблицы с ON DELETE CASCADE очистятся вместе с ними."""
-    with transaction():
     return execute("DELETE FROM contract_dossiers WHERE batch_id = %s", (batch_id,))

site/db/extraction_jobs.py

@@ -0,0 +1,79 @@
+"""Persistent extraction progress and per-batch worker ownership."""
+import json
+import time
+import uuid
+
+from db.connection import execute, query
+
+LEASE_SECONDS = 300
+
+
+def claim(batch_id, total):
+    """Claim a batch atomically, allowing retry after failure or expired lease."""
+    owner = str(uuid.uuid4())
+    now = time.time()
+    changed = execute(
+        """INSERT INTO extraction_jobs (batch_id, owner, state, total, updated_at)
+           VALUES (%s, %s, 'running', %s, %s)
+           ON CONFLICT(batch_id) DO UPDATE SET owner=excluded.owner, state='running',
+             total=excluded.total, done=0, failed=0, classes_json='{}',
+             result_json=NULL, error=NULL, updated_at=excluded.updated_at
+           WHERE extraction_jobs.state != 'running' OR extraction_jobs.updated_at < %s""",
+        (batch_id, owner, total, now, now - LEASE_SECONDS),
+    )
+    return owner if changed else None
+
+
+def heartbeat(batch_id, owner):
+    return execute(
+        "UPDATE extraction_jobs SET updated_at=%s WHERE batch_id=%s AND owner=%s AND state='running'",
+        (time.time(), batch_id, owner),
+    )
+
+
+def progress(batch_id, owner, completed, result):
+    rows = query(
+        "SELECT failed, classes_json FROM extraction_jobs WHERE batch_id=%s AND owner=%s AND state='running'",
+        (batch_id, owner),
+    )
+    if not rows:
+        return
+    failed = rows[0]["failed"]
+    classes = json.loads(rows[0]["classes_json"])
+    if not result.get("ok"):
+        failed += 1
+    else:
+        doc_class = (result.get("extraction") or {}).get("doc_class", "unknown")
+        classes[doc_class] = classes.get(doc_class, 0) + 1
+    execute(
+        """UPDATE extraction_jobs SET done=%s, failed=%s, classes_json=%s, updated_at=%s
+           WHERE batch_id=%s AND owner=%s AND state='running'""",
+        (completed, failed, json.dumps(classes), time.time(), batch_id, owner),
+    )
+
+
+def finish(batch_id, owner, result=None, error=None):
+    state = "done" if result and result.get("ok") and not error else "failed"
+    return execute(
+        """UPDATE extraction_jobs SET state=%s, result_json=%s, error=%s, updated_at=%s
+           WHERE batch_id=%s AND owner=%s AND state='running'""",
+        (state, json.dumps(result, ensure_ascii=False) if result is not None else None,
+         error or ("extraction failed" if state == "failed" else None), time.time(), batch_id, owner),
+    )
+
+
+def get(batch_id):
+    """Mark an abandoned lease interrupted; callers may explicitly retry it."""
+    execute(
+        """UPDATE extraction_jobs SET state='interrupted', error='worker stopped; retry the batch'
+           WHERE batch_id=%s AND state='running' AND updated_at < %s""",
+        (batch_id, time.time() - LEASE_SECONDS),
+    )
+    rows = query("SELECT * FROM extraction_jobs WHERE batch_id=%s", (batch_id,))
+    if not rows:
+        return None
+    job = rows[0]
+    job["classes"] = json.loads(job.pop("classes_json"))
+    result_json = job.pop("result_json")
+    job["result"] = json.loads(result_json) if result_json else None
+    return job

site/routes/pipeline_bp.py

@@ -1,18 +1,26 @@
 """Pipeline blueprint — SSE-сравнение + classify."""
-import json, re, os, threading
+import hashlib, hmac, json, re, os, threading, time
 from flask import Blueprint, request, jsonify, Response, stream_with_context
+from config import API_KEY
 from services.process import run_pipeline
 from services import extractor as stateless_extractor
+from services.engine import analyze_batch
 from services.grouping import build_dossiers_for_batch
 from services.timeline_builder import build_dossier_timeline
 from llm_prompt import build_prompt
-from db import documents
+from db import documents, extraction_jobs
+from db.connection import transaction
 
 pipeline_bp = Blueprint("pipeline", __name__)
 
-# In-memory lock для classify (замена файлового lock)
+_engine_rate_lock = threading.Lock()
-_classify_locks: dict[str, threading.Thread] = {}
+_engine_rate_hits: dict[str, list[float]] = {}
-_extraction_jobs: dict[str, dict] = {}
+_ENGINE_MAX_BODY_BYTES = 10 * 1024 * 1024
+_ENGINE_MAX_DOCUMENTS = 100
+_ENGINE_MAX_ELEMENTS = 20_000
+_ENGINE_MAX_DOCUMENT_BYTES = 2 * 1024 * 1024
+_ENGINE_RATE_WINDOW_SECONDS = 60
+_ENGINE_RATE_LIMIT = 10
 
 
 @pipeline_bp.route("/process-v2", methods=["GET"])
@@ -110,85 +118,214 @@ def classify_batch_route():
     if not isinstance(batch_id, str) or not batch_id.strip():
         return jsonify(ok=False, error="batch_id required"), 400
 
-    # Guard: уже запущена?
-    if batch_id in _classify_locks:
-        t = _classify_locks[batch_id]
-        if t.is_alive():
-            return jsonify(ok=False, error="classify already running"), 409
-        else:
-            del _classify_locks[batch_id]
-
     pending = documents.list_for_extraction(batch_id)
     total = len(pending)
     if total == 0:
         return jsonify(ok=False, error="no parsed documents"), 400
+    owner = extraction_jobs.claim(batch_id, total)
+    if not owner:
+        return jsonify(ok=False, error="classify already running"), 409
 
     # Keep small batches synchronous for the existing browser workflow.
     if total <= 10:
-        result = _run_v3_extraction(batch_id, total)
+        result = _run_extraction_job(batch_id, pending, owner)
         return jsonify(result)
 
-    _extraction_jobs[batch_id] = {
+    _start_extraction_job(batch_id, pending, owner)
-        "state": "running", "total": total, "done": 0,
-        "failed": 0, "classes": {},
-    }
-
-    def _run():
-        try:
-            _run_v3_extraction(batch_id, total, _extraction_progress)
-            _extraction_jobs[batch_id]["state"] = "done"
-        except Exception as error:
-            _extraction_jobs[batch_id].update(state="failed", error=str(error))
-        finally:
-            _classify_locks.pop(batch_id, None)
-
-    t = threading.Thread(target=_run, daemon=True)
-    _classify_locks[batch_id] = t
-    t.start()
-
     return jsonify(ok=True, total=total, v3=True), 202
 
 
-def _extraction_progress(batch_id, completed, result):
+@pipeline_bp.route("/api/engine/analyze", methods=["POST"])
-    """Обновить лёгкий in-memory прогресс для крупного batch."""
+def engine_analyze_route():
-    job = _extraction_jobs[batch_id]
+    """Accept parsed documents over a versioned, authenticated JSON boundary.
-    job["done"] = completed
+
-    if not result.get("ok"):
+    Validation is completed for the whole payload before opening the write
-        job["failed"] += 1
+    transaction, so malformed input cannot leave a partial batch. The caller
+    supplies parsed elements only; extraction passports and dossier decisions
+    are always produced by the in-process Engine and persisted by its adapters.
+    """
+    if not API_KEY:
+        return jsonify(ok=False, error="engine API is not configured"), 503
+    supplied_key = request.headers.get("X-API-Key", "")
+    if not hmac.compare_digest(supplied_key.encode("utf-8"), API_KEY.encode("utf-8")):
+        return jsonify(ok=False, error="unauthorized"), 401
+    if _engine_rate_limited():
+        return jsonify(ok=False, error="rate limit exceeded"), 429, {"Retry-After": "60"}
+
+    raw_body = request.stream.read(_ENGINE_MAX_BODY_BYTES + 1)
+    if len(raw_body) > _ENGINE_MAX_BODY_BYTES:
+        return jsonify(ok=False, error="request body too large"), 413
+    try:
+        body = json.loads(raw_body) if request.is_json else None
+    except (ValueError, UnicodeDecodeError):
+        body = None
+    if not isinstance(body, dict):
+        return jsonify(ok=False, error="JSON object required"), 400
+    if type(body.get("schema_version")) is not int or body["schema_version"] != 1:
+        return jsonify(ok=False, error="unsupported schema_version"), 400
+
+    batch_id = body.get("batch_id")
+    if not isinstance(batch_id, str) or not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", batch_id):
+        return jsonify(ok=False, error="invalid batch_id"), 400
+    source_documents = body.get("documents")
+    if not isinstance(source_documents, list) or not source_documents:
+        return jsonify(ok=False, error="documents must be a non-empty array"), 400
+    if len(source_documents) > _ENGINE_MAX_DOCUMENTS:
+        return jsonify(ok=False, error="too many documents"), 413
+    validated = []
+    for index, source_document in enumerate(source_documents):
+        if not isinstance(source_document, dict):
+            return jsonify(ok=False, error=f"documents[{index}] must be an object"), 400
+        filename = source_document.get("filename")
+        elements = source_document.get("elements")
+        if not isinstance(filename, str) or not filename.strip() or len(filename) > 512:
+            return jsonify(ok=False, error=f"documents[{index}].filename is invalid"), 400
+        if not isinstance(elements, list) or len(elements) > _ENGINE_MAX_ELEMENTS:
+            return jsonify(ok=False, error=f"documents[{index}].elements is invalid"), 400
+        for element_index, element in enumerate(elements):
+            if not _valid_engine_element(element):
+                return jsonify(
+                    ok=False, error=f"documents[{index}].elements[{element_index}] is invalid",
+                ), 400
+        try:
+            elements_json = json.dumps(
+                elements, ensure_ascii=False, sort_keys=True, separators=(",", ":"),
+                allow_nan=False,
+            )
+        except (TypeError, ValueError):
+            return jsonify(ok=False, error=f"documents[{index}].elements is not valid JSON"), 400
+        encoded_elements = elements_json.encode("utf-8")
+        if len(encoded_elements) > _ENGINE_MAX_DOCUMENT_BYTES:
+            return jsonify(ok=False, error=f"documents[{index}] is too large"), 413
+        validated.append((filename.strip(), elements, elements_json, encoded_elements))
+
+    inserted = []
+    with transaction():
+        # Engine API owns a fresh batch per request. Preventing reuse inside
+        # BEGIN IMMEDIATE keeps old parsed rows out of this run and closes the
+        # race between two concurrent requests using the same batch ID.
+        if documents.list_by_batch(batch_id):
+            return jsonify(ok=False, error="batch_id already exists"), 409
+        owner = extraction_jobs.claim(batch_id, len(validated))
+        if not owner:
+            return jsonify(ok=False, error="classify already running"), 409
+        for filename, elements, elements_json, encoded_elements in validated:
+            content_hash = hashlib.sha256(encoded_elements).hexdigest()
+            document = documents.insert(
+                filename, "application/json", None, status="uploaded",
+                batch_id=batch_id, content_hash=content_hash,
+            )
+            documents.set_parsed(document["id"], elements)
+            inserted.append({"id": document["id"], "filename": filename})
+
+    pending = documents.list_for_extraction(batch_id)
+    total = len(pending)
+    if total > 10:
+        _start_extraction_job(batch_id, pending, owner)
+        return jsonify(ok=True, v3=True, total=total, documents=inserted), 202
+
+    result = _run_extraction_job(batch_id, pending, owner)
+    result["documents"] = inserted
+    return jsonify(result)
+
+
+def _valid_engine_element(element):
+    """Validate the parser fields consumed by structured-text formatting."""
+    if not isinstance(element, dict):
+        return False
+    if element.get("type") == "paragraph":
+        return isinstance(element.get("text"), str) and isinstance(element.get("style", ""), str)
+    if element.get("type") != "table":
+        return False
+    rows = element.get("rows")
+    context = element.get("preceding_paragraphs", [])
+    table_index = element.get("table_index")
+    return (
+        isinstance(rows, list)
+        and all(isinstance(row, list) and all(isinstance(cell, str) for cell in row) for row in rows)
+        and isinstance(context, list)
+        and all(isinstance(paragraph, str) for paragraph in context)
+        and (table_index is None or (type(table_index) is int and table_index > 0))
+    )
+
+
+def _start_extraction_job(batch_id, pending, owner):
+    """Run a claimed batch in the background with persistent progress."""
+    def _run():
+        try:
+            _run_extraction_job(batch_id, pending, owner)
+        except Exception:
+            pass
+
+    thread = threading.Thread(target=_run, daemon=True)
+    try:
+        thread.start()
+    except Exception as error:
+        extraction_jobs.finish(batch_id, owner, error=str(error))
+        raise
+
+
+def _engine_rate_limited():
+    """Apply a small per-source fixed-window limit before accepting LLM work."""
+    now = time.monotonic()
+    source = request.remote_addr or "unknown"
+    cutoff = now - _ENGINE_RATE_WINDOW_SECONDS
+    with _engine_rate_lock:
+        for key in list(_engine_rate_hits):
+            recent = [stamp for stamp in _engine_rate_hits[key] if stamp > cutoff]
+            if recent:
+                _engine_rate_hits[key] = recent
             else:
-        extraction = result.get("extraction") or {}
+                del _engine_rate_hits[key]
-        doc_class = extraction.get("doc_class", "unknown")
+        recent = _engine_rate_hits.get(source, [])
-        job["classes"][doc_class] = job["classes"].get(doc_class, 0) + 1
+        if len(recent) >= _ENGINE_RATE_LIMIT:
+            return True
+        recent.append(now)
+        _engine_rate_hits[source] = recent
+    return False
 
 
-def _run_v3_extraction(batch_id, total, progress_callback=None):
+def _run_extraction_job(batch_id, pending, owner):
+    """Keep the lease alive during LLM calls and persist every final outcome."""
+    stop = threading.Event()
+
+    def renew_lease():
+        while not stop.wait(15):
+            if not extraction_jobs.heartbeat(batch_id, owner):
+                return
+
+    heartbeat = threading.Thread(target=renew_lease, daemon=True)
+    heartbeat.start()
+    try:
+        result = _run_v3_extraction(
+            batch_id, pending,
+            progress_callback=lambda completed, item: extraction_jobs.progress(batch_id, owner, completed, item),
+        )
+        extraction_jobs.finish(batch_id, owner, result=result)
+        return result
+    except Exception as error:
+        extraction_jobs.finish(batch_id, owner, error=str(error))
+        raise
+    finally:
+        stop.set()
+        heartbeat.join()
+
+
+def _run_v3_extraction(batch_id, pending, progress_callback=None, llm_client=None):
     """Собрать extraction и dossier summaries в один ответ API."""
-    extraction = stateless_extractor.extract_batch(
+    parsed_documents = [{
+        "id": document["id"],
+        "filename": document["filename"],
+        "elements": stateless_extractor._decode_elements(document.get("elements_json")),
+    } for document in pending]
+    return analyze_batch(
         batch_id,
-        progress_callback=(
+        parsed_documents,
-            (lambda completed, result: progress_callback(batch_id, completed, result))
+        extractor=stateless_extractor.extract_batch,
-            if progress_callback else None
+        dossier_builder=build_dossiers_for_batch,
-        ),
+        llm_client=llm_client,
+        progress_callback=progress_callback,
     )
-    dossier_result = build_dossiers_for_batch(batch_id)
-    classes = extraction.get("classes", {})
-    math_checks_passed = sum(
-        1 for result in extraction.get("results", [])
-        if (result.get("extraction") or {}).get("math_validation", {}).get("status") == "VERIFIED"
-    )
-    return {
-        "ok": bool(extraction.get("results")) and dossier_result.get("ok", False),
-        "v3": True,
-        "done": total,
-        "total": total,
-        "failed": extraction.get("failed", 0),
-        "classes": classes,
-        "ignored_accounting": classes.get("ignored_accounting", 0),
-        "math_checks_passed": math_checks_passed,
-        "dossier_count": len(dossier_result.get("dossiers", [])),
-        "ambiguous": dossier_result.get("ambiguous", 0),
-        "unlinked": dossier_result.get("unlinked", []),
-    }
 
 
 @pipeline_bp.route("/api/classify-progress")
@@ -197,13 +334,13 @@ def classify_progress_route():
     batch_id = request.args.get("batch")
     if not batch_id:
         return jsonify(ok=False, error="batch required"), 400
-    job = _extraction_jobs.get(batch_id)
+    job = extraction_jobs.get(batch_id)
     if not job:
         return jsonify(ok=False, error="batch extraction not found"), 404
     response = {
-        "ok": job["state"] != "failed",
+        "ok": job["state"] not in {"failed", "interrupted"},
         "v3": True,
-        "finished": job["state"] in {"done", "failed"},
+        "finished": job["state"] in {"done", "failed", "interrupted"},
         "state": job["state"],
         "done": job["done"],
         "total": job["total"],
@@ -212,4 +349,6 @@ def classify_progress_route():
     }
     if job.get("error"):
         response["error"] = job["error"]
+    if job.get("result") is not None:
+        response["result"] = job["result"]
     return jsonify(response)

site/services/engine.py

@@ -0,0 +1,82 @@
+"""Shared orchestration for the contract reconstruction engine.
+
+This module deliberately accepts the batch's persistence-backed extractor and
+dossier builder as dependencies. Both the browser workflow and the JSON
+integration boundary therefore execute the same business flow in-process,
+while tests can replace either dependency without constructing HTTP calls or
+making network requests.
+"""
+
+
+def analyze_batch(batch_id, documents, extractor, dossier_builder,
+                  progress_callback=None, llm_client=None):
+    """Extract document passports, assemble dossiers, and shape the API result.
+
+    `extractor` owns LLM interaction and persistence of each extraction;
+    `dossier_builder` consumes those saved passports and persists the resulting
+    graph. Keeping the calls in this order is an invariant: grouping must never
+    observe a partially extracted batch. The callback remains optional because
+    small batches run synchronously, while the existing large-batch route uses
+    it to update persistent progress.
+    """
+    extraction = extractor(
+        batch_id,
+        llm_client=llm_client,
+        progress_callback=progress_callback,
+        documents=documents,
+    )
+    # Extractor-level `ok` means every input succeeded; dossier construction only
+    # requires at least one persisted passport and can proceed for that subset.
+    extraction_results = extraction.get("results") or []
+    total = extraction.get("total", len(documents))
+    done = extraction.get("done")
+    if done is None:
+        done = sum(1 for result in extraction_results if result.get("ok"))
+    failed = extraction.get("failed")
+    if failed is None:
+        failed = max(total - done, 0)
+    dossier_result = dossier_builder(batch_id) if done > 0 else {"ok": False}
+    dossier_ok = bool(dossier_result.get("ok"))
+    # Keep result usability separate from completeness so clients can continue
+    # with valid dossiers while still reporting failures explicitly.
+    status = (
+        "failed" if done == 0 or not dossier_ok
+        else "partial" if failed > 0
+        else "completed"
+    )
+    documents_by_id = {document.get("id"): document for document in documents}
+    failed_documents = []
+    for result in extraction_results:
+        if result.get("ok"):
+            continue
+        document_id = result.get("document_id")
+        source_document = documents_by_id.get(document_id, {})
+        failure = {
+            "document_id": document_id,
+            "filename": source_document.get("filename"),
+            "error": result.get("error") or "extraction failed",
+        }
+        if result.get("persistence_error"):
+            failure["persistence_error"] = result["persistence_error"]
+        failed_documents.append(failure)
+    classes = extraction.get("classes", {})
+    math_checks_passed = sum(
+        1 for result in extraction_results
+        if (result.get("extraction") or {}).get("math_validation", {}).get("status") == "VERIFIED"
+    )
+    return {
+        "ok": status != "failed",
+        "status": status,
+        "partial": status == "partial",
+        "v3": True,
+        "done": done,
+        "total": total,
+        "failed": failed,
+        "failed_documents": failed_documents,
+        "classes": classes,
+        "ignored_accounting": classes.get("ignored_accounting", 0),
+        "math_checks_passed": math_checks_passed,
+        "dossier_count": len(dossier_result.get("dossiers", [])),
+        "ambiguous": dossier_result.get("ambiguous", 0),
+        "unlinked": dossier_result.get("unlinked", []),
+    }

site/services/extractor.py

@@ -250,12 +250,16 @@ def _save_extraction(document_id, extracted, validation, status, raw_response="{
 
 def extract_document(document, llm_client=None, batch_id=None):
     """Извлечь паспорт одного документа, изолируя ошибки на уровне документа."""
-    document_id = document["id"]
+    document_id = document.get("id") if isinstance(document, dict) else None
-    batch_id = batch_id if batch_id is not None else document.get("batch_id")
-    elements = _decode_elements(document.get("elements_json"))
-    structured_text = format_elements_with_structure(elements)
     raw_response = "{}"
     try:
+        if not document_id:
+            raise ValueError("document id is required")
+        batch_id = batch_id if batch_id is not None else document.get("batch_id")
+        elements = _decode_elements(document.get("elements", document.get("elements_json")))
+        # Formatting is part of this document's failure boundary: corrupt stored
+        # elements must not escape and stop extraction of later batch members.
+        structured_text = format_elements_with_structure(elements)
         if _is_garbage_by_filename(document.get("filename") or "") or _is_garbage_by_header(structured_text[:2000]):
             extracted = {
                 "doc_class": "ignored_accounting", "intent_mode": "no_spec",
@@ -299,19 +303,28 @@ def extract_document(document, llm_client=None, batch_id=None):
         return {"ok": True, "document_id": document_id, "status": result["status"], "extraction": saved}
     except Exception as error:
         log.exception("Stateless extraction failed for document %s", document_id)
+        result = {"ok": False, "document_id": document_id, "error": str(error)}
+        if document_id:
+            # Failure persistence is best-effort; a secondary write failure must
+            # not escape this handler and abort processing of remaining documents.
+            try:
                 _save_extraction(document_id, {
                     "doc_class": "unknown", "intent_mode": "no_spec", "counterparty": {},
                     "referenced_contracts": [], "sections": [],
                 }, {"status": "FAILED", "error": str(error)}, "FAILED", raw_response)
-        return {"ok": False, "document_id": document_id, "error": str(error)}
+            except Exception as persistence_error:
+                log.exception("Failed to persist extraction failure for document %s", document_id)
+                result["persistence_error"] = str(persistence_error)
+        return result
 
 
-def extract_batch(batch_id, llm_client=None, progress_callback=None):
+def extract_batch(batch_id, llm_client=None, progress_callback=None, documents=None):
     """Обработать документы batch и вернуть сводку без batch-wide failure.
 
     Optional progress callback получает (completed_count, document_result) после
     каждого документа; ошибки callback не должны останавливать обработку batch.
     """
+    if documents is None:
         documents = db_docs.list_for_extraction(batch_id)
     results = []
     for completed, document in enumerate(documents, start=1):

site/services/grouping.py

@@ -15,6 +15,7 @@ from db import supplements as db_supplements
 from db import dossiers as db_dossiers
 from db import exceptions as db_exceptions
 from db import extractions as db_extractions
+from db.connection import transaction
 
 
 _HOMOGLYPHS = str.maketrans({
@@ -164,6 +165,12 @@ def _dossier_exception(batch_id, document_id, title, source_quote):
 
 
 def build_dossiers_for_batch(batch_id):
+    """Atomically rebuild the dossier graph, retaining the old graph on failure."""
+    with transaction():
+        return _build_dossiers_for_batch(batch_id)
+
+
+def _build_dossiers_for_batch(batch_id):
     """Построить и сохранить dossier graph из паспортов документов одного batch.
 
     Полные номера и ссылки образуют рёбра только при совместимом контрагенте.
@@ -177,8 +184,10 @@ def build_dossiers_for_batch(batch_id):
     records = []
     for document in documents:
         passport = passport_by_id.get(document["id"])
-        if not passport:
+        # A missing or failed passport says extraction produced no usable evidence;
-            passport = {"document_id": document["id"], "doc_class": "unknown"}
+        # do not misreport that technical failure as a missing contract number.
+        if not passport or passport.get("status") == "FAILED":
+            continue
         if passport.get("doc_class") == "ignored_accounting":
             continue
         counterparty = {

site/static/app.js

@@ -157,12 +157,15 @@ async function runClassify() {
           clearInterval(pollTimer);
           clearInterval(timer);
           classifyDone = true;
-          if (d.state === 'failed') {
+          if (d.state === 'failed' || d.state === 'interrupted') {
             btn.innerHTML = '✗ ' + escHtml(d.error || 'ошибка извлечения');
             btn.disabled = false;
             return;
           }
-          btn.innerHTML = '✓ ' + completed + '/' + total + ' документов извлечено';
+          // Partial jobs are terminal and usable, but must not look like full success.
+          var failed = d.failed || 0;
+          btn.innerHTML = (failed ? '⚠ ' : '✓ ') + completed + '/' + total + ' документов извлечено'
+            + (failed ? ' | ошибок: ' + failed : '');
           stepDone('stepClassify');
           stepActive('stepGroups');
           loadGroupsAction();
@@ -212,7 +215,9 @@ async function runClassify() {
       if (data.v3 && data.failed) summary += ' | ошибок: ' + data.failed;
       if (data.garbage) summary += ' | 🗑️ ' + data.garbage + ' мусор';
       if (data.types) summary += ' | ' + Object.entries(data.types).map(function (e) { return e[1] + '×' + e[0]; }).join(', ');
-      btn.innerHTML = '✓ ' + summary + ' (' + Math.round((Date.now() - start) / 1000) + 'с)';
+      // Continue to dossier rendering for partial success while keeping its warning visible.
+      btn.innerHTML = (data.v3 && data.partial ? '⚠ ' : '✓ ') + summary
+        + ' (' + Math.round((Date.now() - start) / 1000) + 'с)';
       stepDone('stepClassify');
       stepActive('stepGroups');
       loadGroupsAction();