diff --git a/README.md b/README.md index 3ce09cd..80d0202 100644 --- a/README.md +++ b/README.md @@ -669,8 +669,9 @@ docker compose exec zp-agent-api \ ## Reindexovanie -Celý proces načíta dokumenty, vytvorí chunky, obnoví FTS5 index a -vytvorí embedding pre každý chunk: +Spoločná synchronizácia pripraví dokumenty, chunky, SQLite/FTS5, embeddingy +a Neo4j graf z jedného snapshotu wiki. Nová generácia sa aktivuje až po +úspešnej príprave oboch indexov: ```bash docker compose run --rm zp-agent-api python -m tools.rag.rebuild_index @@ -682,12 +683,17 @@ Vzniknú súbory: data/documents.json data/chunks.json data/zp_index.sqlite +data/index_version.json ``` Databáza obsahuje dokumenty, chunky, FTS5 index, metadata a embeddingy. -Pri automatickom reindexovaní sa používa zámok proti súbežnému spusteniu -a atomická výmena databázy po úspešnom vytvorení nového indexu. +Manuálny `/sync`, push webhook a uvedený CLI príkaz používajú rovnakú pipeline. +SQLite, Neo4j a odpovede nástrojov evidujú spoločnú verziu dát a Git revíziu. +Pri krátkom prepnutí sa nové vyhľadávanie dočasne odmietne; chyba grafu +vyvolá rollback. Pri nedokončenom prepnutí po páde zostane vyhľadávanie +nedostupné do úspešného opakovania synchronizácie. Podrobnosti a zotavenie: +[Spoločná synchronizácia indexov](docs/synchronizacia_indexov.md). ## Hybridné vyhľadávanie @@ -837,8 +843,9 @@ docker compose exec zp-agent-api \ ``` Builder vytvára constraints, zapisuje uzly a vzťahy a pri štandardnom -spustení obnoví iba časť grafu spravovanú ZP Agentom. Opakované -spustenie preto nevytvára duplicitné uzly. +spustení obnoví iba časť grafu spravovanú ZP Agentom. Zmazanie aj zápis +prebiehajú v jednej transakcii s kontrolou počtov, takže chyba zachová +pôvodný graf. Opakované spustenie nevytvára duplicitné uzly. Graf je možné vizuálne kontrolovať v Neo4j Browseri. Samostatné grafové vyhľadávanie je implementované v `tools/graphrag/graph_search.py`. diff --git a/app/routes.py b/app/routes.py index ebebb7b..f0be588 100644 --- a/app/routes.py +++ b/app/routes.py @@ -46,6 +46,7 @@ from tools.rag.rebuild_index import ( rebuild_index, ) from tools.rag.search_utils import search_database +from tools.index_state import index_snapshot, IndexUnavailableError logger = logging.getLogger(__name__) @@ -182,6 +183,7 @@ class HealthResponse(BaseModel): zpwiki_exists: bool security_configured: bool + index_version: dict[str, Any] | None = None class RagSource(BaseModel): @@ -231,6 +233,7 @@ class RagResponse(BaseModel): sources: list[ RagSource ] + index_version: dict[str, Any] | None = None class SearchResponse(BaseModel): @@ -246,6 +249,7 @@ class SearchResponse(BaseModel): results: list[ dict[str, Any] ] + index_version: dict[str, Any] | None = None class SyncResponse(BaseModel): @@ -261,6 +265,8 @@ class SyncResponse(BaseModel): str, int, ] + index_version: dict[str, Any] | None = None + graph_counts: dict[str, int] | None = None class WebhookSuccessResponse( @@ -283,6 +289,8 @@ class WebhookSuccessResponse( str, int, ] + index_version: dict[str, Any] | None = None + graph_counts: dict[str, int] | None = None class WebhookIgnoredResponse( @@ -529,10 +537,17 @@ def health() -> dict[str, Any]: and zpwiki_exists and security_configured ) + version = None + try: + with index_snapshot(DB_FILE.parent, db_file=DB_FILE) as version: + pass + except (IndexUnavailableError, sqlite3.Error): + ready = False return { "status": "ok", "ready": ready, + "index_version": version, "database_exists": ( database_exists ), @@ -708,6 +723,8 @@ def sync( "pull_git": ( request.pull_git ), + "index_version": result.get('index_version'), + "graph_counts": result.get('graph_counts'), "duration_seconds": ( result[ "duration_seconds" @@ -890,6 +907,8 @@ async def gitea_webhook( "verified_by": ( "hmac_sha256" ), + "index_version": result.get('index_version'), + "graph_counts": result.get('graph_counts'), "duration_seconds": ( result[ "duration_seconds" diff --git a/docs/synchronizacia_indexov.md b/docs/synchronizacia_indexov.md new file mode 100644 index 0000000..5700544 --- /dev/null +++ b/docs/synchronizacia_indexov.md @@ -0,0 +1,93 @@ +# Spoločná synchronizácia RAG a GraphRAG + +`POST /sync`, overený push webhook aj `python -m tools.rag.rebuild_index` +používajú spoločnú pipeline v `tools/sync_indexes.py`. Obnova vyžaduje dostupné +Neo4j a jeho konfiguráciu v prostredí. Pôvodné evaluačné výsledky sa nemenia. + +## Priebeh + +1. Procesový `.reindex.lock` pripustí iba jednu synchronizáciu nad data volume. +2. Voliteľný `git pull --ff-only` zachováva pôvodné nastavenie synchronizácie. +3. V `data/.sync-stage-*` vznikne kópia pôvodných README súborov. Dokumenty, + chunky, SQLite/FTS5 a embeddingy sa pripravujú z tejto jednej kópie. Aktívne + súbory sa počas prípravy nemenia a existujúce vyhľadávanie môže pokračovať. +4. Overí sa, že sa zdrojové súbory ani Git revízia počas prípravy nezmenili. +5. Pred aktiváciou vzniknú zálohy pôvodných súborov a trvalý marker + `.sync-pending.json`. Exkluzívny publikačný zámok počká na rozbehnutých + čitateľov. Nové požiadavky počas aktivácie dostanú dočasnú nedostupnosť. +6. Jedna explicitná Neo4j transakcia odstráni starý spravovaný graf, zapíše + nový graf a skontroluje počty uzlov/vzťahov oproti pripravenému payloadu. + Verzia grafu sa zapisuje v tej istej transakcii. +7. Pod publikačným zámkom sa aktivujú pripravené lokálne súbory a potvrdí sa + grafová transakcia. Po úspechu sa odstráni marker a sprístupní nová verzia. + +SQLite a Neo4j netvoria distribuovanú ACID transakciu. Konzistenciu pri tejto +implementácii zaisťujú spoločný zámok pre podporované vyhľadávacie vstupy, +verzie, zálohy a marker nedokončeného prepnutia. Používanie samotného nízkoúrovňového +`search_graph_with_session` mimo publikačnej ochrany je určené pre kontrolované +interné evaluácie/testy; produkčný volajúci používa chránené `search_graph`. + +## Verzia a dohľadateľnosť + +Každá generácia má identifikátor `version`, Git revíziu zdrojovej wiki, +UTC čas prípravy, hash pôvodných README súborov, hashe dokumentov/chunkov +a názov embeddingového modelu. Hash zdrojov zachytáva aj necommitnuté zmeny; +samotný Git HEAD preto nie je jediným identifikátorom obsahu. + +Rovnaký objekt `index_version` je uložený v: + +- `data/index_version.json`, +- SQLite tabuľke `index_state`, +- operatívnom uzle Neo4j `IndexState` s ID `zpwiki`. + +Operatívny uzol nie je súčasťou znalostného grafu označeného `graph_scope=zpwiki`. +Verziu vracajú `/sync`, webhook, `/health`, RAG vyhľadávanie a verejná Python +funkcia grafového vyhľadávania. Synchronizácia vracia aj `graph_counts`. + +## Chyby a zotavenie + +- Chyba prípravy, embeddingov alebo nedostupné Neo4j pred aktiváciou: + pôvodné indexy ostávajú aktívne. +- Chyba pri zápise grafu: celá grafová transakcia sa zruší. +- Chyba pri aktivácii lokálneho súboru: obnovia sa pôvodné súbory a grafová + transakcia sa zruší. Pri neúspešnej obnove marker zostáva. +- Stratené potvrdenie COMMIT: najprv sa kontroluje skutočná verzia v Neo4j. + Ak bola nová verzia potvrdená, staré lokálne dáta sa naslepo neobnovujú. +- Pád procesu alebo neoveriteľný výsledok transakcie: marker a obnovovacie + súbory ostávajú na disku. Vyhľadávanie odmietne nejasný snapshot; RAG API + vracia 503 a `/health` má `ready=false`. + +Po odstránení príčiny spusti znova autorizovaný `/sync` alebo: + +```bash +docker compose exec zp-agent-api python -m tools.rag.rebuild_index +``` + +Úspešná synchronizácia vytvorí kompletnú novú generáciu oboch indexov a marker +odstráni. Marker sa nemá ručne mazať ako náhrada opravy. Staršie adresáre +prerušených behov môžu zostať ako podklady na obnovu; bežný úspešný staging +sa automaticky odstraňuje. + +## Samostatná obnova Neo4j + +`python -m tools.graphrag.build_graphrag` používa rovnaký transakčný zápis. +Zmena grafu sa potvrdí celá alebo sa zruší celá. Ak JSON vstupy zodpovedajú +aktívnej generácii, jej proveniencia sa zachová. Iné vstupy alebo režim +`--no-replace` nemôžu predstierať synchronizovanú verziu; verejný grafový +vstup potom pri aktívnom manifeste vyžaduje spoločnú synchronizáciu. + +## Overenie + +Regresie pokrývajú chybu uprostred grafového zápisu, chybu po čiastočnej +aktivácii súborov, stratenú odpoveď COMMIT, neznámy výsledok COMMIT, následné +zotavenie, zámok pre čitateľov, nesúlad verzií, zmenu zdroja počas prípravy, +odpovede `/sync` aj podpísaného webhooku a API 503 pri nejasnej generácii. +Skutočné Neo4j testy overujú rollback po odstránení pôvodných uzlov a zápise +časti náhrady v izolovanom scope. Pozitívny prípad overuje aj opakovateľnosť. + +Záverečná sada po dokončení opravy: **464 passed, 4 skipped**. Reálny `/sync` +v Docker API obnovil 114 dokumentov, 887 chunkov a 447 grafových uzlov / +879 vzťahov za 87,72 s. Overené sú rovnaké verzie v manifeste, SQLite, Neo4j, +health, skutočnej RAG odpovedi aj grafovom vyhľadávaní. Obsah znalostného +grafu zodpovedá pôvodnej v4; nové meranie presnosti LLM sa nevykonávalo. +Podrobné overenie: [záznam](synchronizacia_indexov_overenie.json). diff --git a/docs/synchronizacia_indexov_overenie.json b/docs/synchronizacia_indexov_overenie.json new file mode 100644 index 0000000..978d744 --- /dev/null +++ b/docs/synchronizacia_indexov_overenie.json @@ -0,0 +1,52 @@ +{ + "date": "2026-10-25", + "changes": [ + "Coordinated RAG/Neo4j snapshot publication for /sync, push webhook and CLI.", + "Atomic Neo4j replacement with rollback, count validation and data-version metadata." + ], + "backup_directory": "/tmp/zp-agent-sync-backup-20261025T023515Z", + "automated_tests": { + "passed": 464, + "skipped": 4 + }, + "live_sync": { + "status": "ok", + "pull_git": false, + "duration_seconds": 87.72, + "counts": { + "documents": 114, + "chunks": 887, + "fts_chunks": 887, + "embedding_chunks": 887, + "tags": 2189, + "categories": 1193 + }, + "index_version": { + "version": "d4a6fbaa03b44a0a8aabb3cba2ddff36", + "git_revision": "3e892a1d49e9bce118cd03478ce15e01c2bebb4e", + "created_at_utc": "2026-10-25T02:50:27.976522+00:00", + "source_sha256": "007792c12f3467dd3203bb4951ccc197fdc0eef669580f1b1f893baf609ca363", + "documents_sha256": "0542af2a373771cf7c4a95f93d14f4f687f6a0187846892c959aa6efc4bade52", + "chunks_sha256": "437dde589be46e0c30842dd330de22d96a0173d2dbaff9c5f129568cb2bca4c9", + "embedding_model": "intfloat/multilingual-e5-small" + }, + "graph_counts": { + "nodes": 447, + "relationships": 879 + } + }, + "live_verification": { + "ready": true, + "rag_sources": 3, + "graph_status": "ok", + "versions_equal_in_sqlite_neo4j_manifest_and_responses": true, + "knowledge_graph_unchanged_against_frozen_v4": true + }, + "verification_scope": [ + "Full existing and new regression suite, including real isolated Neo4j rollback.", + "Signed webhook and manual sync response tests.", + "Actual /sync on deployed Docker API without Git pull.", + "Actual authenticated /rag and public graph lookup on the same data generation." + ], + "new_llm_accuracy_measurement": false +} diff --git a/test/test_index_sync.py b/test/test_index_sync.py new file mode 100644 index 0000000..8b05ff7 --- /dev/null +++ b/test/test_index_sync.py @@ -0,0 +1,238 @@ +"""Failure-injection tests for coordinated publication, not only happy paths.""" +from __future__ import annotations + +import hashlib +import hmac +import json +import sqlite3 +from pathlib import Path + +import pytest + +from tools import index_state as state +from tools import sync_indexes as sync + + +class Result: + def __init__(self, value): + self.value = value + def single(self): + return self.value + + +class Session: + def __init__(self, version): + self.version = version; self.fail_commit = None; self.unavailable = False + def run(self, query, **params): + if self.unavailable: + raise ConnectionError('lost database connection') + return Result({'metadata': json.dumps(self.version)} if self.version else None) + def begin_transaction(self): + return Transaction(self) + + +class Transaction: + def __init__(self, session): + self.session = session; self.version = session.version + def __enter__(self): + return self + def __exit__(self, *args): + pass # Uncommitted local state is discarded, as in the Neo4j driver. + def commit(self): + if self.session.fail_commit == 'unknown': + self.session.unavailable = True + raise ConnectionError('commit outcome unknown') + self.session.version = self.version + if self.session.fail_commit == 'response_lost': + raise ConnectionError('committed but response lost') + + +def snapshot(directory: Path, version: dict): + directory.mkdir(parents=True, exist_ok=True) + (directory / 'documents.json').write_text(json.dumps([version])) + (directory / 'chunks.json').write_text(json.dumps([version])) + with sqlite3.connect(directory / 'zp_index.sqlite') as conn: + conn.execute('CREATE TABLE index_state (id INTEGER PRIMARY KEY, metadata_json TEXT)') + conn.execute('INSERT INTO index_state VALUES (1,?)', (json.dumps(version),)) + state.durable_json(directory / state.VERSION_FILE, version) + + +@pytest.fixture +def publication(tmp_path, monkeypatch): + data = tmp_path / 'data'; stage = data / '.sync-stage-test' + old = {'version': 'old', 'git_revision': 'commit-old'} + new = {'version': 'new', 'git_revision': 'commit-new'} + snapshot(data, old); snapshot(stage, new) + monkeypatch.setattr(sync, 'write_graph_transaction', lambda tx, payload, version: setattr(tx, 'version', version)) + monkeypatch.setattr(sync, 'graph_counts', lambda tx: ([('Person', 1)], [('HAS_DOCUMENT', 1)])) + return data, stage, old, new, Session(old) + + +def test_publish_makes_all_versions_agree(publication): + data, stage, old, new, session = publication + result = sync.publish_snapshot(data, stage, new, {}, session) + assert result == {'nodes': 1, 'relationships': 1} + assert state.read_version(data) == state.sqlite_version(data / 'zp_index.sqlite') == session.version == new + with state.index_snapshot(data, db_file=data / 'zp_index.sqlite') as version: + assert version == new + assert not (data / state.PENDING_FILE).exists() + + +def test_graph_write_failure_keeps_original_files_and_graph(publication, monkeypatch): + data, stage, old, new, session = publication + before = {name: (data / name).read_bytes() for name in sync.FILES} + def fail(tx, payload, version): + tx.version = new + raise RuntimeError('failure halfway through graph write') + monkeypatch.setattr(sync, 'write_graph_transaction', fail) + with pytest.raises(RuntimeError, match='halfway'): + sync.publish_snapshot(data, stage, new, {}, session) + assert session.version == old + assert before == {name: (data / name).read_bytes() for name in sync.FILES} + assert not (data / state.PENDING_FILE).exists() + + +def test_partial_file_publication_rolls_back_files_and_graph(publication, monkeypatch): + data, stage, old, new, session = publication + before = {name: (data / name).read_bytes() for name in sync.FILES} + replace = sync.replace_file + def fail(source, target): + if target.name == 'zp_index.sqlite': + raise OSError('disk write failed') + replace(source, target) + monkeypatch.setattr(sync, 'replace_file', fail) + with pytest.raises(OSError, match='disk write'): + sync.publish_snapshot(data, stage, new, {}, session) + assert session.version == old + assert before == {name: (data / name).read_bytes() for name in sync.FILES} + assert not (data / state.PENDING_FILE).exists() + + +def test_lost_commit_response_is_verified_before_restoring_any_files(publication): + data, stage, old, new, session = publication + session.fail_commit = 'response_lost' + sync.publish_snapshot(data, stage, new, {}, session) + assert session.version == state.sqlite_version(data / 'zp_index.sqlite') == new + assert not (data / state.PENDING_FILE).exists() + + +def test_unknown_commit_blocks_reads_until_next_successful_sync(publication): + data, stage, old, new, session = publication + session.fail_commit = 'unknown' + with pytest.raises(ConnectionError): + sync.publish_snapshot(data, stage, new, {}, session) + assert (stage / 'previous/zp_index.sqlite').exists() + with pytest.raises(state.IndexUnavailableError, match='nebola dokončená'): + with state.index_snapshot(data): + pytest.fail('Mixed generation must never be readable') + session.unavailable = False; session.fail_commit = None + repaired = {'version': 'repaired'}; next_stage = data / '.sync-stage-repaired' + snapshot(next_stage, repaired) + sync.publish_snapshot(data, next_stage, repaired, {}, session) + assert session.version == state.sqlite_version(data / 'zp_index.sqlite') == state.read_version(data) == repaired + assert not (data / state.PENDING_FILE).exists() + + +def test_read_during_publication_is_unavailable(publication): + data, *_ = publication + with state.publication_lock(data, exclusive=True): + with pytest.raises(state.IndexUnavailableError, match='aktivácia'): + with state.index_snapshot(data): + pytest.fail('Publication barrier bypassed') + + +def test_reader_rejects_a_replaced_sqlite_database(publication): + data, stage, old, new, session = publication + state.durable_json(data / state.VERSION_FILE, new) + with pytest.raises(state.IndexUnavailableError, match='SQLite'): + with state.index_snapshot(data, db_file=data / 'zp_index.sqlite'): + pass + + +def test_staging_failure_never_touches_live_indices(tmp_path, monkeypatch): + data = tmp_path / 'data'; snapshot(data, {'version': 'old'}) + before = {name: (data / name).read_bytes() for name in sync.FILES} + def fail(wiki, stage): + (stage / 'documents.json').write_text('new incomplete data') + raise RuntimeError('embedding failed') + monkeypatch.setattr(sync, 'prepare_snapshot', fail) + with pytest.raises(RuntimeError, match='embedding failed'): + sync.synchronize_indexes(tmp_path / 'wiki', data) + assert before == {name: (data / name).read_bytes() for name in sync.FILES} + assert not list(data.glob('.sync-stage-*')) + + +def test_prepare_uses_one_frozen_source_and_records_versions(tmp_path, monkeypatch): + import numpy as np + import tools.rag.build_sqlite_index as indexer + wiki = tmp_path / 'wiki'; page = wiki / 'pages/students/2020/eva/README.md' + page.parent.mkdir(parents=True) + page.write_text('---\ntitle: Eva Nová\npublished: true\ntaxonomy:\n category: [bp2024]\n---\n# Bakalárska práca 2024\n\nNázov: Overovanie synchronizácie indexov\n') + monkeypatch.setattr(indexer, 'embed_passages', lambda texts: np.ones((len(texts), 4), dtype=np.float32)) + stage = tmp_path / 'stage'; stage.mkdir() + version, counts, payload = sync.prepare_snapshot(wiki, stage) + assert counts['documents'] == 1 and counts['chunks'] > 0 + assert version == state.sqlite_version(stage / 'zp_index.sqlite') == state.read_version(stage) + assert version['documents_sha256'] == sync.digest(stage / 'documents.json') + assert version['chunks_sha256'] == sync.digest(stage / 'chunks.json') + assert payload['people'][0]['name'] == 'Eva Nová' + assert payload['theses'][0]['year'] == 2024 + assert payload['documents'][0]['path'] == 'pages/students/2020/eva/README.md' + + +def test_changed_source_is_not_activated(tmp_path, monkeypatch): + import numpy as np + import tools.rag.build_sqlite_index as indexer + wiki = tmp_path / 'wiki'; page = wiki / 'pages/example/README.md' + page.parent.mkdir(parents=True); page.write_text('---\ntitle: Source\n---\nInitial source text.') + def embed(texts): + page.write_text('Source changed during indexing.') + return np.ones((len(texts), 4), dtype=np.float32) + monkeypatch.setattr(indexer, 'embed_passages', embed) + stage = tmp_path / 'stage'; stage.mkdir() + with pytest.raises(RuntimeError, match='počas indexovania'): + sync.prepare_snapshot(wiki, stage) + + +def test_api_returns_503_while_publication_is_unresolved(tmp_path, monkeypatch): + from fastapi.testclient import TestClient + import app.main as main + import app.routes as routes + monkeypatch.setattr(main, 'embed_query', lambda q: None) + for name, value in [('WEBHOOK_SECRET', 'w'), ('SEARCH_API_KEY', 'a'), ('SYNC_API_KEY', 's')]: + monkeypatch.setenv(name, value * 64) + data = tmp_path / 'data'; snapshot(data, {'version': 'old'}) + state.durable_json(data / state.PENDING_FILE, {'target_version': 'new'}) + monkeypatch.setattr(routes, 'DB_FILE', data / 'zp_index.sqlite') + with TestClient(main.app) as client: + response = client.post('/rag', json={'query': 'Aká práca?'}, + headers={'Authorization': 'Bearer ' + 'a' * 64}) + assert response.status_code == 503 + assert client.get('/health').json()['ready'] is False + + +@pytest.mark.parametrize('endpoint', ['sync', 'webhook/gitea']) +def test_sync_api_exposes_common_version_and_graph_counts(monkeypatch, endpoint): + from fastapi.testclient import TestClient + import app.main as main + import app.routes as routes + monkeypatch.setattr(main, 'embed_query', lambda q: None) + for name, value in [('WEBHOOK_SECRET', 'w'), ('SEARCH_API_KEY', 'a'), ('SYNC_API_KEY', 's')]: + monkeypatch.setenv(name, value * 64) + monkeypatch.setenv('EXPECTED_GITEA_REPOSITORY', 'KEMT/zpwiki') + monkeypatch.setenv('WEBHOOK_PULL_GIT', 'false') + version = {'version': 'new', 'git_revision': 'commit'} + monkeypatch.setattr(routes, 'rebuild_index', lambda **kwargs: { + 'duration_seconds': 0.1, 'counts': {'documents': 1}, 'index_version': version, + 'graph_counts': {'nodes': 2, 'relationships': 1}}) + with TestClient(main.app) as client: + if endpoint == 'sync': + response = client.post('/sync', json={}, headers={'X-API-Key': 's' * 64}) + else: + body = json.dumps({'repository': {'full_name': 'KEMT/zpwiki'}}).encode() + signature = hmac.new(('w' * 64).encode(), body, hashlib.sha256).hexdigest() + response = client.post('/webhook/gitea', content=body, + headers={'X-Gitea-Event': 'push', 'X-Gitea-Signature': signature, 'Content-Type': 'application/json'}) + assert response.status_code == 200 + assert response.json()['index_version'] == version + assert response.json()['graph_counts'] == {'nodes': 2, 'relationships': 1} diff --git a/tools/graphrag/build_graphrag.py b/tools/graphrag/build_graphrag.py index b8a8fba..540e3e2 100644 --- a/tools/graphrag/build_graphrag.py +++ b/tools/graphrag/build_graphrag.py @@ -13,6 +13,7 @@ from .graphrag import ( GRAPH_SCOPE, build_graph_payload, ) +from tools.index_state import DATA_DIR, PENDING_FILE, publication_lock, read_version PROJECT_ROOT = Path( @@ -33,6 +34,7 @@ CHUNKS_FILE = ( CONSTRAINTS = ( + ('zpwiki_index_state_unique', 'IndexState', 'id'), ( "zpwiki_document_path_unique", "Document", @@ -81,6 +83,45 @@ LEGACY_CONSTRAINTS = ( ) +def graph_version(session) -> dict | None: + record = session.run('MATCH (s:IndexState {id:$scope}) RETURN s.metadata_json AS metadata', + scope=GRAPH_SCOPE).single() + return json.loads(record['metadata']) if record and record['metadata'] else None + + +def write_graph_transaction(tx, payload: dict, *, replace: bool = True, + version: dict | None = None) -> None: + """All destructive data changes belong to the caller's single transaction.""" + if replace: + clear_managed_graph(tx) + for key, query in NODE_QUERIES.items(): + write_rows(tx, query, payload[key]) + for key, query in RELATIONSHIP_QUERIES.items(): + write_rows(tx, query, payload[key]) + if replace: + nodes, relationships = graph_counts(tx) + labels = {'people': 'Person', 'document_authors': 'DocumentAuthor', 'documents': 'Document', + 'categories': 'Category', 'topics': 'Topic', 'keywords': 'Keyword', + 'theses': 'Thesis', 'works': 'Work'} + types = {'person_documents': 'HAS_DOCUMENT', 'document_author_documents': 'AUTHORED_DOCUMENT', + 'document_categories': 'IN_CATEGORY', 'document_keywords': 'HAS_KEYWORD', + 'document_describes_topics': 'DESCRIBES', 'person_theses': 'AUTHORED', + 'person_works': 'HAS_WORK'} + expected_nodes = {labels[key]: len(payload[key]) for key in labels if payload[key]} + expected_relations = {types[key]: len(payload[key]) for key in types if payload[key]} + evidenced = len(payload['thesis_documents']) + len(payload['work_documents']) + if evidenced: + expected_relations['EVIDENCED_BY'] = evidenced + if dict(nodes) != expected_nodes or dict(relationships) != expected_relations: + raise RuntimeError('Neo4j počty nezodpovedajú pripravenému grafu; transakcia sa zruší.') + if version is None: + tx.run('MATCH (s:IndexState {id:$scope}) DELETE s', scope=GRAPH_SCOPE).consume() + else: + # Operational state is deliberately outside the knowledge graph scope. + tx.run('MERGE (s:IndexState {id:$scope}) SET s.metadata_json=$metadata', + scope=GRAPH_SCOPE, metadata=json.dumps(version, sort_keys=True)).consume() + + NODE_QUERIES = { "people": """ UNWIND $rows AS row @@ -508,23 +549,6 @@ def main() -> None: args = parser.parse_args() - documents = load_json( - DOCUMENTS_FILE - ) - - chunks = load_json( - CHUNKS_FILE - ) - - payload = build_graph_payload( - documents, - chunks, - ) - - print_payload_summary( - payload - ) - settings = get_neo4j_settings() with create_neo4j_driver( @@ -543,28 +567,21 @@ def main() -> None: session ) - if args.replace: - clear_managed_graph( - session - ) - - for key, query in ( - NODE_QUERIES.items() - ): - write_rows( - session, - query, - payload[key], - ) - - for key, query in ( - RELATIONSHIP_QUERIES.items() - ): - write_rows( - session, - query, - payload[key], - ) + import hashlib + with publication_lock(DATA_DIR, exclusive=True): + if (DATA_DIR / PENDING_FILE).exists(): + raise RuntimeError('Nedokončená spoločná synchronizácia; použite tools.rag.rebuild_index.') + payload = build_graph_payload(load_json(DOCUMENTS_FILE), load_json(CHUNKS_FILE)) + print_payload_summary(payload) + version = read_version(DATA_DIR) + if not args.replace: + version = None # An incremental merge cannot certify removed source records. + if version and any(hashlib.sha256(path.read_bytes()).hexdigest() != version[key] + for path, key in [(DOCUMENTS_FILE, 'documents_sha256'), (CHUNKS_FILE, 'chunks_sha256')]): + version = None # A standalone rebuild must not claim stale provenance. + with session.begin_transaction() as tx: + write_graph_transaction(tx, payload, replace=args.replace, version=version) + tx.commit() print_database_summary( session diff --git a/tools/graphrag/graph_search.py b/tools/graphrag/graph_search.py index ded3f4f..bf6cef8 100644 --- a/tools/graphrag/graph_search.py +++ b/tools/graphrag/graph_search.py @@ -2600,30 +2600,22 @@ def search_graph( get_neo4j_settings, ) + from tools.index_state import DATA_DIR, IndexUnavailableError, index_snapshot + from .build_graphrag import graph_version settings = ( get_neo4j_settings() ) - with create_neo4j_driver( - settings - ) as driver: - driver.verify_connectivity() - - with driver.session( - database=( - settings.database - ) - ) as session: - return ( - search_graph_with_session( - session, - query, - limit=limit, - max_entities=( - max_entities - ), - ) - ) + with index_snapshot(DATA_DIR) as version: + with create_neo4j_driver(settings) as driver: + driver.verify_connectivity() + with driver.session(database=settings.database) as session: + if version is not None and graph_version(session) != version: + raise IndexUnavailableError('Neo4j graf nezodpovedá aktívnej verzii dát.') + result = search_graph_with_session(session, query, limit=limit, max_entities=max_entities) + if version is not None: + result['index_version'] = version + return result def main() -> None: diff --git a/tools/graphrag/tests/test_graph_rebuild_atomic.py b/tools/graphrag/tests/test_graph_rebuild_atomic.py new file mode 100644 index 0000000..6153de2 --- /dev/null +++ b/tools/graphrag/tests/test_graph_rebuild_atomic.py @@ -0,0 +1,61 @@ +"""Actual Neo4j rollback after DELETE and partial replacement; isolated scopes.""" +import os +import uuid + +import pytest + +from tools.graphrag import build_graphrag as builder +from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings + +pytestmark = pytest.mark.skipif(os.getenv('RUN_NEO4J_TESTS') != '1', reason='Requires local Neo4j.') + + +@pytest.fixture +def managed_graph(monkeypatch): + scope = 'atomic-rebuild-' + uuid.uuid4().hex + monkeypatch.setattr(builder, 'GRAPH_SCOPE', scope) + settings = get_neo4j_settings() + with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: + session.run('CREATE (:Person {id:$id, name:"Original", graph_scope:$scope})', + id=scope+'old', scope=scope).consume() + session.run('CREATE (:IndexState {id:$scope, metadata_json:$metadata})', + scope=scope, metadata='{"version":"old"}').consume() + try: + yield session, scope + finally: + session.run('MATCH (n) WHERE n.graph_scope=$scope DETACH DELETE n', scope=scope).consume() + session.run('MATCH (s:IndexState {id:$scope}) DELETE s', scope=scope).consume() + + +def replacement(scope): + payload = {key: [] for key in [*builder.NODE_QUERIES, *builder.RELATIONSHIP_QUERIES]} + payload['people'] = [{'id': scope+'new', 'name': 'Replacement', 'graph_scope': scope}] + payload['documents'] = [{'path': scope+'/p.md', 'graph_scope': scope}] + payload['person_documents'] = [{'person_id': scope+'new', 'document_path': scope+'/p.md'}] + return payload + + +def test_failed_replacement_restores_deleted_graph_and_version(managed_graph): + session, scope = managed_graph + payload = replacement(scope) + payload['documents'].append({'path': None, 'graph_scope': scope}) + with pytest.raises(Exception): + with session.begin_transaction() as tx: + builder.write_graph_transaction(tx, payload, version={'version': 'new'}) + tx.commit() + rows = session.run('MATCH (n) WHERE n.graph_scope=$scope RETURN n.name AS name', scope=scope).data() + assert rows == [{'name': 'Original'}] + assert builder.graph_version(session) == {'version': 'old'} + + +def test_successful_replacement_commits_graph_and_version_together(managed_graph): + session, scope = managed_graph + payload = replacement(scope); version = {'version': 'new', 'git_revision': 'new-commit'} + for _ in range(2): + with session.begin_transaction() as tx: + builder.write_graph_transaction(tx, payload, version=version) + tx.commit() + assert builder.graph_version(session) == version + nodes, relationships = builder.graph_counts(session) + assert dict(nodes) == {'Person': 1, 'Document': 1} + assert dict(relationships) == {'HAS_DOCUMENT': 1} diff --git a/tools/index_state.py b/tools/index_state.py new file mode 100644 index 0000000..ec02268 --- /dev/null +++ b/tools/index_state.py @@ -0,0 +1,104 @@ +"""Publication barrier and provenance shared by SQLite and Neo4j. + +Builders work in a staging directory. Readers hold a shared process lock while +using a snapshot; publishers hold its exclusive counterpart. A durable pending +marker makes a process crash fail closed until the next successful sync. +""" +from __future__ import annotations + +import fcntl +import json +import os +import sqlite3 +from contextlib import closing, contextmanager +from functools import wraps +from pathlib import Path + +from tools.rag.common import DATA_DIR + +VERSION_FILE = 'index_version.json' +PENDING_FILE = '.sync-pending.json' +LOCK_FILE = '.index-publication.lock' + + +class IndexUnavailableError(RuntimeError): + pass + + +def fsync_directory(path: Path) -> None: + descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def durable_json(path: Path, value: dict) -> None: + temporary = path.with_suffix(path.suffix + '.tmp') + with temporary.open('w', encoding='utf-8') as handle: + json.dump(value, handle, ensure_ascii=False, indent=2) + handle.write('\n'); handle.flush(); os.fsync(handle.fileno()) + os.replace(temporary, path) + fsync_directory(path.parent) + + +@contextmanager +def publication_lock(data_dir: Path, *, exclusive: bool = False): + data_dir.mkdir(parents=True, exist_ok=True) + with (data_dir / LOCK_FILE).open('a+') as handle: + try: + flags = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH | fcntl.LOCK_NB + fcntl.flock(handle.fileno(), flags) + except BlockingIOError as error: + raise IndexUnavailableError('Prebieha aktivácia novej verzie indexov.') from error + try: + yield + finally: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + + +def read_version(data_dir: Path = DATA_DIR) -> dict | None: + path = data_dir / VERSION_FILE + if not path.exists(): + return None # Pre-migration index; first coordinated sync adds metadata. + try: + value = json.loads(path.read_text()) + if not isinstance(value, dict) or not value.get('version'): + raise ValueError('Missing version') + return value + except (ValueError, OSError) as error: + raise IndexUnavailableError('Neplatná verzia vyhľadávacieho indexu.') from error + + +def sqlite_version(db_file: Path) -> dict | None: + with closing(sqlite3.connect(db_file.resolve().as_uri() + '?mode=ro', uri=True)) as conn: + if not conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='index_state'").fetchone(): + return None + row = conn.execute('SELECT metadata_json FROM index_state WHERE id=1').fetchone() + try: + return json.loads(row[0]) if row else None + except ValueError as error: + raise IndexUnavailableError('Neplatná verzia SQLite indexu.') from error + + +@contextmanager +def index_snapshot(data_dir: Path = DATA_DIR, *, db_file: Path | None = None): + with publication_lock(data_dir): + if (data_dir / PENDING_FILE).exists(): + raise IndexUnavailableError('Synchronizácia indexov nebola dokončená; zopakujte synchronizáciu.') + version = read_version(data_dir) + if db_file is not None and version is not None and sqlite_version(db_file) != version: + raise IndexUnavailableError('SQLite index nezodpovedá aktívnej verzii dát.') + yield version + + +def rag_snapshot(function): + @wraps(function) + def wrapped(*args, **kwargs): + db_file = Path(args[0] if args else kwargs.get('db_path', kwargs.get('db_file'))) + with index_snapshot(db_file.parent, db_file=db_file) as version: + result = function(*args, **kwargs) + if version is not None: + result['index_version'] = version + return result + return wrapped diff --git a/tools/rag/build_chunks.py b/tools/rag/build_chunks.py index 4f35f3e..a277ad6 100644 --- a/tools/rag/build_chunks.py +++ b/tools/rag/build_chunks.py @@ -920,19 +920,20 @@ def content_hash( return hashlib.sha256(value).hexdigest()[:16] -def build_chunks() -> list[dict]: - if not PAGES_ROOT.exists(): +def build_chunks(*, wiki_root: Path | None = None, output: Path | None = None) -> list[dict]: + pages = wiki_root / 'pages' if wiki_root is not None else PAGES_ROOT + if not pages.exists(): raise SystemExit( - f"Neexistuje priečinok: {PAGES_ROOT}" + f"Neexistuje priečinok: {pages}" ) all_chunks: list[dict] = [] document_count = 0 for file_path in sorted( - PAGES_ROOT.glob("**/README.md") + pages.glob("**/README.md") ): - document = load_zpwiki_page(file_path) + document = load_zpwiki_page(file_path, wiki_root=wiki_root) document_count += 1 drafts = chunk_markdown( @@ -968,7 +969,7 @@ def build_chunks() -> list[dict]: ) write_json( - CHUNKS_FILE, + output or CHUNKS_FILE, all_chunks, ) @@ -1008,7 +1009,7 @@ def build_chunks() -> list[dict]: ) print( f"[green]Výstup uložený do:[/green] " - f"{CHUNKS_FILE}" + f"{output or CHUNKS_FILE}" ) if all_chunks: diff --git a/tools/rag/build_sqlite_index.py b/tools/rag/build_sqlite_index.py index 432d154..aeb26d7 100644 --- a/tools/rag/build_sqlite_index.py +++ b/tools/rag/build_sqlite_index.py @@ -626,23 +626,25 @@ def temporary_database_path( ) -def build_database() -> dict[str, int]: +def build_database(*, documents_file: Path | None = None, chunks_file: Path | None = None, + db_file: Path | None = None) -> dict[str, int]: + target = db_file or DB_FILE documents = read_json( - DOCUMENTS_FILE + documents_file or DOCUMENTS_FILE ) chunks = read_json( - CHUNKS_FILE + chunks_file or CHUNKS_FILE ) - DB_FILE.parent.mkdir( + target.parent.mkdir( parents=True, exist_ok=True, ) temporary_file = ( temporary_database_path( - DB_FILE + target ) ) @@ -698,7 +700,7 @@ def build_database() -> dict[str, int]: os.replace( temporary_file, - DB_FILE, + target, ) except Exception: @@ -710,7 +712,7 @@ def build_database() -> dict[str, int]: print( "[green]SQLite index " "vytvorený:[/green] " - f"{DB_FILE}" + f"{target}" ) print( diff --git a/tools/rag/common.py b/tools/rag/common.py index 675ead0..0bf3565 100644 --- a/tools/rag/common.py +++ b/tools/rag/common.py @@ -328,6 +328,8 @@ def resolve_page_title( def load_zpwiki_page( file_path: Path, + *, + wiki_root: Path | None = None, ) -> dict[str, Any]: post = frontmatter.load(file_path) @@ -366,7 +368,7 @@ def load_zpwiki_page( return { "path": str( - file_path.relative_to(ZPWIKI_ROOT) + file_path.relative_to(wiki_root or ZPWIKI_ROOT) ), "title": title, "categories": categories, diff --git a/tools/rag/rag_utils.py b/tools/rag/rag_utils.py index 37d9ef5..f65131d 100644 --- a/tools/rag/rag_utils.py +++ b/tools/rag/rag_utils.py @@ -5,6 +5,7 @@ import json import sqlite3 from pathlib import Path from typing import Any +from tools.index_state import rag_snapshot from tools.rag.rag_document_expansion import ( expand_results_with_exact_document_section, @@ -691,6 +692,7 @@ def _expand_exact_document( return expand_results_with_exact_document_section(**kwargs) +@rag_snapshot def build_rag_context( db_path: Path, query: str, diff --git a/tools/rag/rebuild_index.py b/tools/rag/rebuild_index.py index 5e1e956..4fede02 100644 --- a/tools/rag/rebuild_index.py +++ b/tools/rag/rebuild_index.py @@ -21,10 +21,7 @@ if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) -from tools.rag.build_chunks import build_chunks -from tools.rag.build_sqlite_index import build_database -from tools.rag.common import DATA_DIR, DB_FILE, ZPWIKI_ROOT -from tools.rag.scan_zpwiki import scan_pages +from tools.rag.common import DATA_DIR, ZPWIKI_ROOT REINDEX_LOCK_FILE = DATA_DIR / ".reindex.lock" @@ -144,22 +141,25 @@ def git_pull( def rebuild_index( pull_git: bool = False, *, - lock_file: Path = REINDEX_LOCK_FILE, + lock_file: Path | None = None, + wiki_root: Path | None = None, + data_dir: Path | None = None, ) -> dict: - with acquire_reindex_lock(lock_file): + root = wiki_root or ZPWIKI_ROOT + target = data_dir or DATA_DIR + with acquire_reindex_lock(lock_file or target / '.reindex.lock'): start = time.monotonic() print( f"[green]ZPWIKI_ROOT:[/green] " - f"{ZPWIKI_ROOT}" + f"{root}" ) if pull_git: - git_pull() + git_pull(root) - documents = scan_pages() - chunks = build_chunks() - counts = build_database() + from tools.sync_indexes import synchronize_indexes + result = synchronize_indexes(root, target) duration = round( time.monotonic() - start, @@ -168,10 +168,7 @@ def rebuild_index( return { "duration_seconds": duration, - "documents_scanned": len(documents), - "chunks_created": len(chunks), - "counts": counts, - "database_path": str(DB_FILE), + **result, } @@ -179,7 +176,7 @@ def main() -> None: parser = argparse.ArgumentParser( description=( "Obnoví JSON súbory " - "a SQLite FTS5 index." + "a koordinovane SQLite aj Neo4j index." ) ) diff --git a/tools/rag/scan_zpwiki.py b/tools/rag/scan_zpwiki.py index 7e9ebc7..3598e4e 100644 --- a/tools/rag/scan_zpwiki.py +++ b/tools/rag/scan_zpwiki.py @@ -16,9 +16,10 @@ if str(PROJECT_ROOT) not in sys.path: from tools.rag.common import DOCUMENTS_FILE, PAGES_ROOT, ZPWIKI_ROOT, load_zpwiki_page, write_json -def scan_pages() -> list[dict]: - if not PAGES_ROOT.exists(): - raise SystemExit(f"Neexistuje priečinok: {PAGES_ROOT}") +def scan_pages(*, wiki_root: Path | None = None, output: Path | None = None) -> list[dict]: + pages = wiki_root / 'pages' if wiki_root is not None else PAGES_ROOT + if not pages.exists(): + raise SystemExit(f"Neexistuje priečinok: {pages}") documents = [] metadata_keys = Counter() @@ -26,8 +27,8 @@ def scan_pages() -> list[dict]: tags = Counter() authors = Counter() - for file_path in sorted(PAGES_ROOT.glob("**/README.md")): - page = load_zpwiki_page(file_path) + for file_path in sorted(pages.glob("**/README.md")): + page = load_zpwiki_page(file_path, wiki_root=wiki_root) content = page.pop("content", "") for key in page["metadata"]: @@ -50,7 +51,7 @@ def scan_pages() -> list[dict]: } ) - write_json(DOCUMENTS_FILE, documents) + write_json(output or DOCUMENTS_FILE, documents) print_summary(documents, metadata_keys, categories, tags, authors) return documents diff --git a/tools/rag/search_utils.py b/tools/rag/search_utils.py index ca1ffb9..8001759 100644 --- a/tools/rag/search_utils.py +++ b/tools/rag/search_utils.py @@ -3,6 +3,7 @@ from __future__ import annotations import sqlite3 from pathlib import Path from typing import Any +from tools.index_state import rag_snapshot # Tieto importy sú zámerne verejné. # Zachovávajú spätnú kompatibilitu pre testy @@ -43,6 +44,7 @@ from tools.rag.search_core import ( ) +@rag_snapshot def search_database( db_file: Path, query: str, diff --git a/tools/rag/tests/test_rebuild_index.py b/tools/rag/tests/test_rebuild_index.py index 0d8d53e..5729ff6 100644 --- a/tools/rag/tests/test_rebuild_index.py +++ b/tools/rag/tests/test_rebuild_index.py @@ -54,40 +54,39 @@ def test_rebuild_runs_pipeline_in_correct_order( monkeypatch.setattr( rebuild, "git_pull", - lambda: calls.append("git_pull"), - ) - monkeypatch.setattr( - rebuild, - "scan_pages", - lambda: calls.append("scan") or [{"path": "a"}], - ) - monkeypatch.setattr( - rebuild, - "build_chunks", - lambda: calls.append("chunks") or [{"chunk_id": "a::0"}], - ) - monkeypatch.setattr( - rebuild, - "build_database", - lambda: calls.append("database") - or { - "documents": 1, - "chunks": 1, - "fts_chunks": 1, - "tags": 0, - "categories": 0, - }, + lambda root: calls.append("git_pull"), ) + import tools.sync_indexes as sync + monkeypatch.setattr(sync, 'synchronize_indexes', lambda wiki, data: + calls.append('sqlite_and_neo4j') or {'documents_scanned': 1, 'chunks_created': 1, + 'counts': {'fts_chunks': 1}, 'index_version': {'version': 'both'}}) result = rebuild.rebuild_index( pull_git=True, lock_file=tmp_path / "lock", ) - assert calls == ["git_pull", "scan", "chunks", "database"] + assert calls == ["git_pull", "sqlite_and_neo4j"] assert result["documents_scanned"] == 1 assert result["chunks_created"] == 1 assert result["counts"]["fts_chunks"] == 1 + assert result['index_version'] == {'version': 'both'} + + +def test_rebuild_pulls_and_locks_the_requested_snapshot_paths(tmp_path, monkeypatch): + import tools.sync_indexes as sync + wiki = tmp_path / 'alternate-wiki' + data = tmp_path / 'alternate-data' + seen = [] + monkeypatch.setattr(rebuild, 'git_pull', lambda root: seen.append(('pull', root))) + def synchronized(root, target): + assert (data / '.reindex.lock').exists() + seen.append(('sync', root, target)) + return {'index_version': {'version': 'alternate'}} + monkeypatch.setattr(sync, 'synchronize_indexes', synchronized) + result = rebuild.rebuild_index(pull_git=True, wiki_root=wiki, data_dir=data) + assert seen == [('pull', wiki), ('sync', wiki, data)] + assert result['index_version']['version'] == 'alternate' def test_git_pull_uses_fast_forward_only( diff --git a/tools/sync_indexes.py b/tools/sync_indexes.py new file mode 100644 index 0000000..9f454fb --- /dev/null +++ b/tools/sync_indexes.py @@ -0,0 +1,167 @@ +"""Stage SQLite/JSON and publish them with a single Neo4j data transaction. + +There is no distributed transaction between a filesystem and Neo4j. Cooperating +readers therefore use the publication barrier. Ordinary failures restore local +files; ambiguous commit/crash leaves a durable marker and readers fail closed. +Re-running sync creates a complete fresh generation and clears that marker. +""" +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import sqlite3 +import subprocess +import tempfile +import uuid +from contextlib import closing +from datetime import datetime, timezone +from pathlib import Path + +from tools.index_state import (PENDING_FILE, VERSION_FILE, durable_json, fsync_directory, + publication_lock, sqlite_version) +from tools.rag.scan_zpwiki import scan_pages +from tools.rag.build_chunks import build_chunks +from tools.rag.build_sqlite_index import build_database +from tools.rag.embedding_utils import embedding_model_name +from tools.graphrag.graphrag import build_graph_payload +from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings +from tools.graphrag.build_graphrag import (create_constraints, graph_version, + write_graph_transaction, graph_counts) + +FILES = ('documents.json', 'chunks.json', 'zp_index.sqlite', VERSION_FILE) + + +def digest(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def source_files(wiki_root: Path) -> dict[str, str]: + if not (wiki_root / 'pages').is_dir(): + raise RuntimeError(f'Chýba wiki priečinok pages: {wiki_root}') + return {str(p.relative_to(wiki_root)): digest(p) + for p in sorted((wiki_root / 'pages').rglob('README.md'))} + + +def git_revision(wiki_root: Path) -> str | None: + if not (wiki_root / '.git').exists(): + return None + result = subprocess.run(['git', '-C', str(wiki_root), 'rev-parse', 'HEAD'], + capture_output=True, text=True, timeout=15, check=True) + return result.stdout.strip() + + +def prepare_snapshot(wiki_root: Path, stage: Path) -> tuple[dict, dict, dict]: + revision = git_revision(wiki_root); sources = source_files(wiki_root) + frozen_wiki = stage / 'wiki' + for relative in sources: + target = frozen_wiki / relative; target.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(wiki_root / relative, target) + if source_files(frozen_wiki) != sources: + raise RuntimeError('Wiki sa počas vytvárania snapshotu zmenila.') + documents = scan_pages(wiki_root=frozen_wiki, output=stage / 'documents.json') + chunks = build_chunks(wiki_root=frozen_wiki, output=stage / 'chunks.json') + counts = build_database(documents_file=stage / 'documents.json', chunks_file=stage / 'chunks.json', + db_file=stage / 'zp_index.sqlite') + version = {'version': uuid.uuid4().hex, 'git_revision': revision, + 'created_at_utc': datetime.now(timezone.utc).isoformat(), + 'source_sha256': hashlib.sha256(json.dumps(sources, sort_keys=True).encode()).hexdigest(), + 'documents_sha256': digest(stage / 'documents.json'), 'chunks_sha256': digest(stage / 'chunks.json'), + 'embedding_model': embedding_model_name()} + with closing(sqlite3.connect(stage / 'zp_index.sqlite')) as conn, conn: + conn.execute('CREATE TABLE index_state (id INTEGER PRIMARY KEY CHECK(id=1), metadata_json TEXT NOT NULL)') + conn.execute('INSERT INTO index_state VALUES (1, ?)', (json.dumps(version, sort_keys=True),)) + durable_json(stage / VERSION_FILE, version) + if source_files(wiki_root) != sources or git_revision(wiki_root) != revision: + raise RuntimeError('Wiki sa počas indexovania zmenila; pripravené indexy sa neaktivujú.') + return version, counts, build_graph_payload(documents, chunks) + + +def replace_file(source: Path, target: Path) -> None: + """Separate operation so tests can inject a publication I/O failure.""" + os.replace(source, target) + + +def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, session) -> dict: + with publication_lock(data_dir, exclusive=True): + had_pending = (data_dir / PENDING_FILE).exists() + old_graph_version = graph_version(session) + originals = [] + backup = stage / 'previous'; backup.mkdir() + for name in FILES: + current = data_dir / name + if current.exists(): + shutil.copy2(current, backup / name) + with (backup / name).open('rb') as handle: + os.fsync(handle.fileno()) + originals.append(name) + fsync_directory(backup) + durable_json(data_dir / PENDING_FILE, {'target_version': version['version'], + 'staging_directory': stage.name, 'original_files': originals}) + commit_attempted = False + try: + with session.begin_transaction() as tx: + write_graph_transaction(tx, payload, version=version) + node_counts, relation_counts = graph_counts(tx) + graph_summary = {'nodes': sum(n for _, n in node_counts), + 'relationships': sum(n for _, n in relation_counts)} + for name in FILES: + # Flush staged contents before the durable directory switch. + with (stage / name).open('rb') as handle: + os.fsync(handle.fileno()) + replace_file(stage / name, data_dir / name) + fsync_directory(data_dir) + commit_attempted = True + tx.commit() + except BaseException: + if commit_attempted: + # A lost COMMIT response is not proof of rollback. Never restore + # old SQLite against a possibly committed new graph. + try: + current_graph_version = graph_version(session) + except Exception: + raise # Keep marker + backups; reads stay unavailable. + if current_graph_version == version and sqlite_version(data_dir / 'zp_index.sqlite') == version: + (data_dir / PENDING_FILE).unlink(); fsync_directory(data_dir) + return graph_summary + if current_graph_version != old_graph_version: + raise # Unknown external writer; fail closed. + for name in FILES: + if name in originals: + os.replace(backup / name, data_dir / name) + else: + (data_dir / name).unlink(missing_ok=True) + fsync_directory(data_dir) + if not had_pending: + (data_dir / PENDING_FILE).unlink(); fsync_directory(data_dir) + raise + (data_dir / PENDING_FILE).unlink() + fsync_directory(data_dir) + return graph_summary + + +def synchronize_indexes(wiki_root: Path, data_dir: Path) -> dict: + data_dir.mkdir(parents=True, exist_ok=True) + stage = Path(tempfile.mkdtemp(prefix='.sync-stage-', dir=data_dir)) + try: + version, counts, payload = prepare_snapshot(wiki_root, stage) + settings = get_neo4j_settings() + with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: + driver.verify_connectivity() + create_constraints(session) + graph_summary = publish_snapshot(data_dir, stage, version, payload, session) + return {'counts': counts, 'documents_scanned': counts['documents'], 'chunks_created': counts['chunks'], + 'database_path': str(data_dir / 'zp_index.sqlite'), 'index_version': version, + 'graph_counts': graph_summary} + finally: + # Keep recovery material when a crash/ambiguous commit is unresolved. + pending = data_dir / PENDING_FILE + keep_stage = False + if pending.exists(): + try: + keep_stage = json.loads(pending.read_text()).get('staging_directory') == stage.name + except (ValueError, OSError): + keep_stage = True + if not keep_stage: + shutil.rmtree(stage)