diff --git a/.dockerignore b/.dockerignore index e64104f..4254a13 100644 --- a/.dockerignore +++ b/.dockerignore @@ -11,7 +11,9 @@ htmlcov/ data/ docs/ integrations/ -tools/git_runner/ +tools/git_runner/* +!tools/git_runner/sync_fork.sh +!tools/git_runner/ssh_known_hosts **/*.md experiments/ test/ diff --git a/README.md b/README.md index d5ad5e2..520f8d4 100644 --- a/README.md +++ b/README.md @@ -16,8 +16,12 @@ odpovede o prácach, projektoch, rokoch a súvisiacich vzťahoch so zdrojovými 5. Backend skontroluje hodnoty, väzby, citácie a pokrytie požadovaných údajov. 6. Z overených podkladov zostaví odpoveď; chýbajúce informácie výslovne označí. -Aktualizáciu možno spustiť manuálne alebo webhookom z Gitea. Oba indexy -sa obnovujú z rovnakého snapshotu a používajú spoločnú verziu dát. +Aktualizáciu možno spustiť manuálne alebo webhookom z Gitea. Docker nasadenie +ju plánuje aj každých 24 hodín (`WIKI_SYNC_INTERVAL_SECONDS=86400`). +`SYNC_UPSTREAM_FORK=1` bezpečne dopĺňa fork `jp170na/zpwiki` z `KEMT/zpwiki`; +do originálu nezapisuje. Vyžaduje bežiaci backend a SSH prístup do forku. +Textový index, metadátový graf a už zapnutá obsahová extrakcia sa pripravujú +z rovnakého snapshotu a aktivujú spoločne. Pri chybe zostáva predchádzajúca verzia. ## Vyhľadávanie @@ -62,7 +66,10 @@ v Docker sieti. Dočasná verejná URL vyžaduje zodpovedajúce OAuth nastavenie RAG, GraphRAG, kontrolovaná cesta odpovedí a pripojenie cez Claude sú funkčné. Systém obsahuje embeddingovú cache, kontroly dostupnosti a persistentnú -synchronizačnú frontu. Historické evaluácie sú oddelené od prevádzkového kódu. +synchronizačnú frontu. `/health` samostatne uvádza vyhľadávanie, obsahové +poznatky a generovanie. Dostupnosť modelu eviduje po jeho použití alebo po +explicitnom autentifikovanom `POST /health/generation`; samotná kontrola +`GET /health` model nevolá. Historické evaluácie sú oddelené od prevádzkového kódu. Samostatný hlavný agent, CRZP a spracovanie plných textov prác sú ďalšie etapy. ## Organizácia projektu diff --git a/app/main.py b/app/main.py index 1b96d00..92758e3 100644 --- a/app/main.py +++ b/app/main.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import logging +import os import sys from contextlib import asynccontextmanager from pathlib import Path @@ -68,7 +69,8 @@ async def lifespan( application.state.embedding_ready = True return result - application.state.sync_jobs = SyncJobs(DATA_DIR, runner=run_sync) + application.state.sync_jobs = SyncJobs(DATA_DIR, runner=run_sync, + periodic_interval_seconds=float(os.getenv("WIKI_SYNC_INTERVAL_SECONDS", "0"))) application.state.sync_jobs.start() try: yield diff --git a/app/routes.py b/app/routes.py index d490007..78d4e63 100644 --- a/app/routes.py +++ b/app/routes.py @@ -209,6 +209,8 @@ class HealthResponse(BaseModel): security_configured: bool index_version: dict[str, Any] | None = None tools: dict[str, Any] + capabilities: dict[str, Any] + answer_ready: dict[str, bool] class RagSource(RetrievalSource): @@ -444,12 +446,23 @@ def health(request: Request) -> dict[str, Any]: embedding_ready=getattr(request.app.state, "embedding_ready", False), ) ready = tools["rag"]["ready"] and tools["graphrag"]["ready"] + from tools.runtime.answer_health import answer_capabilities + capabilities = answer_capabilities(DB_FILE, tools, version) + content = capabilities["knowledge"] + generation = capabilities["generation"]["ready"] + answer_ready = { + "rag": tools["rag"]["ready"] and content["ready"] and generation, + "graphrag": tools["graphrag"]["ready"] and content["ready"] + and content.get("graph", {}).get("ready", False) and generation, + } return { "status": "ok", "ready": ready, "index_version": version, "tools": tools, + "capabilities": capabilities, + "answer_ready": answer_ready, "database_exists": (database_exists), "search_engine": ("hybrid_fts5_embeddings"), "rag_enabled": True, @@ -458,6 +471,14 @@ def health(request: Request) -> dict[str, Any]: } +@router.post("/health/generation", include_in_schema=False, + dependencies=[Depends(require_search_api_key)]) +def probe_answer_generation(): + """Explicit authenticated model probe; GET /health never incurs model calls.""" + from tools.answers.model_client import probe_generation + return probe_generation() + + @router.post( "/rag", operation_id=("retrieve_zpwiki_context"), diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index 000f8d5..dfbdd8c 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -19,6 +19,9 @@ services: environment: ZPWIKI_ROOT: /zpwiki + WIKI_SYNC_INTERVAL_SECONDS: ${WIKI_SYNC_INTERVAL_SECONDS:-86400} + SYNC_UPSTREAM_FORK: ${SYNC_UPSTREAM_FORK:-1} + ZPWIKI_SYNC_KEY_FILE: /run/secrets/zpwiki_ssh_key CHUNK_MAX_TOKENS: "450" CHUNK_OVERLAP_TOKENS: "70" CHUNK_MIN_TOKENS: "80" diff --git a/packages/claude/client.py b/packages/claude/client.py index a50366f..2f6362e 100644 --- a/packages/claude/client.py +++ b/packages/claude/client.py @@ -336,6 +336,8 @@ def backend_health(config=None): "status": "ok" if value["ready"] else "error", "ready": value["ready"], "index_version": value.get("index_version"), + **({"capabilities": value["capabilities"]} if isinstance(value.get("capabilities"), dict) else {}), + **({"answer_ready": value["answer_ready"]} if isinstance(value.get("answer_ready"), dict) else {}), "tools": { engine: { k: v diff --git a/packages/claude/dist/packages.json b/packages/claude/dist/packages.json index 5c56b4d..6b33982 100644 --- a/packages/claude/dist/packages.json +++ b/packages/claude/dist/packages.json @@ -3,12 +3,12 @@ { "archive": "zpwiki-rag.zip", "name": "rag", - "sha256": "c2b0902370aaa43a972edf0efa38a1211fe97575c4a8a68b0eb7cea1bbec5ccf" + "sha256": "c7a207660004c1e4e7ae7d507006df231c666d13c53067bd5896a022ef44b64d" }, { "archive": "zpwiki-graphrag.zip", "name": "graphrag", - "sha256": "e127fb2c97b5ed8279904c754ae394a8fc3be9fa210a6f7b13fff9499d8888a6" + "sha256": "1f340a0298625b7dd5f427e8501dde8ec781cc8d72e7a4f50b9ad4b21de6b1fa" } ], "schema_version": 1 diff --git a/packages/claude/dist/zpwiki-graphrag.zip b/packages/claude/dist/zpwiki-graphrag.zip index 77eb8db..ea83b5b 100644 Binary files a/packages/claude/dist/zpwiki-graphrag.zip and b/packages/claude/dist/zpwiki-graphrag.zip differ diff --git a/packages/claude/dist/zpwiki-rag.zip b/packages/claude/dist/zpwiki-rag.zip index 1651217..2f66ed6 100644 Binary files a/packages/claude/dist/zpwiki-rag.zip and b/packages/claude/dist/zpwiki-rag.zip differ diff --git a/test/test_answer_health.py b/test/test_answer_health.py new file mode 100644 index 0000000..2343272 --- /dev/null +++ b/test/test_answer_health.py @@ -0,0 +1,54 @@ +import json + +import pytest + +from tools.answers import model_client as model +from tools.runtime import answer_health as health +from tools.knowledge.publication import checkpoint_state, PENDING_FILE +from tools.sync.index_state import durable_json, VERSION_FILE +from test_index_sync import snapshot + + +def test_generation_poll_never_calls_provider_and_reports_unknown(monkeypatch): + monkeypatch.setenv("ANSWER_API_KEY", "private-secret") + monkeypatch.setenv("ANSWER_MODEL", "test") + monkeypatch.setenv("ANSWER_LLM_URL", "https://provider.example/v1/chat/completions") + monkeypatch.setattr(model, "_status", {}) + def forbidden(*args, **kwargs): + pytest.fail("Health polling must not call LLM") + monkeypatch.setattr(model, "_complete_request", forbidden) + assert model.generation_status()["status"] == "unknown" + assert "private-secret" not in json.dumps(model.generation_status()) + + +def test_probe_records_success_failure_and_configuration_change(monkeypatch): + monkeypatch.setenv("ANSWER_API_KEY", "test-secret") + monkeypatch.setenv("ANSWER_MODEL", "test") + monkeypatch.setenv("ANSWER_LLM_URL", "https://provider.example/v1/chat/completions") + monkeypatch.setattr(model, "_status", {}) + monkeypatch.setattr(model, "_complete_request", lambda *a, **kw: {"choices": [{"message": {"content": "OK"}}]}) + assert model.probe_generation()["ready"] is True + def fail(*args, **kwargs): + raise ConnectionError("do not expose private-secret") + monkeypatch.setattr(model, "_complete_request", fail) + result = model.probe_generation() + assert not result["ready"] and result["reason"] == "model_probe_failed" + assert "private-secret" not in json.dumps(result) + monkeypatch.setenv("ANSWER_MODEL", "new-model") + assert model.generation_status()["status"] == "unknown" + + +def test_content_health_distinguishes_current_stale_corrupt_and_pending(tmp_path): + version = {"version": "old"} + snapshot(tmp_path, version) + index = {"version": "content", "index_version": version, "graph_ready": False, "units": []} + checkpoint_state(tmp_path, index) + db = tmp_path / "zp_index.sqlite" + assert health.content_readiness(db, version, graph_available=False)["ready"] + path = tmp_path / "knowledge/index.json" + path.write_text(path.read_text() + " ") + assert health.content_readiness(db, version, graph_available=False)["reason"] == "content_checkpoint_invalid" + checkpoint_state(tmp_path, {**index, "index_version": {"version": "new"}}) + assert health.content_readiness(db, version, graph_available=False)["reason"] == "content_version_mismatch" + durable_json(tmp_path / PENDING_FILE, {}) + assert health.content_readiness(db, version, graph_available=False)["reason"] == "content_publication_incomplete" diff --git a/test/test_content_publication.py b/test/test_content_publication.py new file mode 100644 index 0000000..8f3e004 --- /dev/null +++ b/test/test_content_publication.py @@ -0,0 +1,145 @@ +"""Content is published with the source generation, including failure recovery.""" +import json +from pathlib import Path + +import pytest + +from tools.knowledge import publication as content +from tools.knowledge import store +from tools.sync import index_state as state +from tools.sync import sync_indexes as sync +from test_index_sync import Session, snapshot + + +def checkpoint(version, label="content"): + return {"version": label, "index_version": version, "graph_ready": True, + "units": [], "profiles": 1} + + +class ContentSession(Session): + def __init__(self, version, content_state): + super().__init__(version) + self.content = content_state + + def begin_transaction(self): + session = self + class Tx: + def __enter__(self): + self.version, self.content = session.version, session.content + return self + def __exit__(self, *args): + pass + def commit(self): + if session.fail_commit == "unknown": + session.unavailable = True + raise ConnectionError("unknown commit") + session.version, session.content = self.version, self.content + if session.fail_commit == "response_lost": + raise ConnectionError("lost response") + return Tx() + + +def graph_state(value): + return {"version": value["version"], "checkpoint_sha256": value["checkpoint_sha256"], + "index_version_json": json.dumps(value["index_version"], sort_keys=True)} + + +@pytest.fixture +def publication(tmp_path, monkeypatch): + old, new = {"version": "old"}, {"version": "new"} + data, stage = tmp_path / "data", tmp_path / "stage" + snapshot(data, old) + snapshot(stage, new) + old_state = content.checkpoint_state(data, checkpoint(old, "old-content")) + content.checkpoint_state(stage, checkpoint(new, "new-content")) + session = ContentSession(old, graph_state(old_state)) + monkeypatch.setattr(sync, "write_graph_transaction", lambda tx, payload, version: setattr(tx, "version", version)) + monkeypatch.setattr(sync, "graph_counts", lambda tx: ([("Person", 1)], [])) + def write_content(tx, index, *, checkpoint_sha256=None): + tx.content = {"version": index["version"], "checkpoint_sha256": checkpoint_sha256, + "index_version_json": json.dumps(index["index_version"], sort_keys=True)} + monkeypatch.setattr(store, "write_graph_transaction", write_content) + def read_content(session): + if session.unavailable: + raise ConnectionError("database offline") + return session.content + monkeypatch.setattr(store, "graph_state", read_content) + return data, stage, old, new, session + + +@pytest.mark.parametrize("commit", [None, "response_lost"]) +def test_publish_all_sources_and_content_in_same_transaction(publication, commit): + data, stage, old, new, session = publication + session.fail_commit = commit + sync.publish_snapshot(data, stage, new, {}, session, knowledge=checkpoint(new, "new-content")) + assert state.read_version(data) == state.sqlite_version(data / "zp_index.sqlite") == session.version == new + assert store.load_index(data, new)["version"] == "new-content" + assert content.state_matches_graph(content.read_state(data), session.content) + assert not (data / state.PENDING_FILE).exists() + + +def test_failed_content_file_switch_restores_entire_generation(publication, monkeypatch): + data, stage, old, new, session = publication + files = (*sync.FILES, *content.FILES) + before = {f: (data / f).read_bytes() for f in files} + replace = sync.replace_file + def fail(source, target): + if target == data / "knowledge/state.json": + raise OSError("disk failed") + replace(source, target) + monkeypatch.setattr(sync, "replace_file", fail) + with pytest.raises(OSError): + sync.publish_snapshot(data, stage, new, {}, session, knowledge=checkpoint(new, "new-content")) + assert before == {f: (data / f).read_bytes() for f in files} + assert session.version == old + assert content.state_matches_graph(content.read_state(data), session.content) + + +def test_unknown_commit_blocks_metadata_and_content(publication): + data, stage, old, new, session = publication + session.fail_commit = "unknown" + with pytest.raises(ConnectionError): + sync.publish_snapshot(data, stage, new, {}, session, knowledge=checkpoint(new, "new-content")) + with pytest.raises(state.IndexUnavailableError): + with state.index_snapshot(data): + pass + assert (stage / "previous/knowledge/index.json").exists() + + +@pytest.mark.parametrize("commit", [None, "response_lost", "unknown"]) +def test_standalone_content_publication_failure_barrier(publication, commit): + data, stage, old, new, session = publication + session.fail_commit = commit + if commit == "unknown": + with pytest.raises(ConnectionError): + content.publish_checkpoint(data, stage, checkpoint(new, "new-content"), session) + with pytest.raises(state.IndexUnavailableError): + store.load_index(data, new) + assert (data / content.PENDING_FILE).exists() + else: + content.publish_checkpoint(data, stage, checkpoint(new, "new-content"), session) + assert content.state_matches_graph(content.read_state(data), session.content) + assert store.load_index(data, new)["version"] == "new-content" + + +def test_content_staging_failure_leaves_old_primary_and_content(publication, monkeypatch): + from tools.knowledge import build + data, stage, old, new, session = publication + before = {f: (data / f).read_bytes() for f in (*sync.FILES, *content.FILES)} + monkeypatch.setattr(sync, "prepare_snapshot", lambda *args: (new, {}, {})) + monkeypatch.setattr(build, "configured_extractor", lambda: (None, {})) + def fail(*args, **kwargs): + raise RuntimeError("LLM unavailable") + monkeypatch.setattr(build, "prepare_index", fail) + with pytest.raises(RuntimeError, match="LLM unavailable"): + sync.synchronize_indexes(Path("unused-wiki"), data) + assert before == {f: (data / f).read_bytes() for f in before} + assert not (data / state.PENDING_FILE).exists() + + +def test_corrupt_content_fails_integrity_check(publication): + data, *_ = publication + path = data / "knowledge/index.json" + path.write_text(path.read_text() + " ") + with pytest.raises(RuntimeError, match="invalid"): + store.load_index(data, {"version": "old"}) diff --git a/test/test_git_runner.py b/test/test_git_runner.py index 2c8a07a..7aefcd1 100644 --- a/test/test_git_runner.py +++ b/test/test_git_runner.py @@ -305,3 +305,16 @@ def test_runner_credentials_and_state_are_ignored_by_git(private_path): text=True, ) assert result.returncode == 0 + + +def test_existing_key_file_updates_only_fork(repositories, tmp_path): + source, fork, _ = repositories + upstream_head = commit(source, "pending-with-local-key") + key = tmp_path / "private-key" + key.write_text("not-an-actual-key-local-transport-only") + key.chmod(0o600) + result = sync(repositories, "--key-file", str(key), token="") + assert result.returncode == 0, result.stderr + assert git(fork, "rev-parse", "master") == upstream_head + assert git(source, "rev-parse", "HEAD") == upstream_head + assert key.read_text() == "not-an-actual-key-local-transport-only" diff --git a/test/test_knowledge.py b/test/test_knowledge.py index 8ff5af9..e9ae820 100644 --- a/test/test_knowledge.py +++ b/test/test_knowledge.py @@ -529,32 +529,18 @@ def test_build_is_cached_and_source_changes_cannot_activate_it(tmp_path): assert activated == (data / "knowledge/index.json").read_bytes() -@pytest.mark.parametrize("enabled", [False, True]) -def test_index_sync_refreshes_content_only_after_it_has_been_enabled( - tmp_path, monkeypatch, enabled -): +def test_rebuild_delegates_content_to_coordinated_publication(tmp_path, monkeypatch): import tools.rag.rebuild_index as rebuild import tools.sync.sync_indexes as sync from tools.knowledge import build as knowledge - target = tmp_path / "data" - target.mkdir() - if enabled: - (target / "knowledge").mkdir() - (target / "knowledge/index.json").write_text("{}") - calls = [] - phases = [] - monkeypatch.setattr(sync, "synchronize_indexes", lambda *a, **kw: {"counts": {"documents": 1}}) - monkeypatch.setattr( - knowledge, "refresh_existing", lambda *a: calls.append(a) or {"cached_batches": 1} - ) - result = rebuild.rebuild_index( - wiki_root=tmp_path / "wiki", data_dir=target, progress=phases.append - ) - assert bool(calls) == enabled - assert ("knowledge" in result) == enabled - if enabled: - assert phases[-1] == "extracting_knowledge" + expected = {"counts": {"documents": 1}, "knowledge": {"cached_batches": 1}} + monkeypatch.setattr(sync, "synchronize_indexes", lambda *a, **kw: expected) + def forbidden(*args): + pytest.fail("Content must not be refreshed after publishing primary indexes") + monkeypatch.setattr(knowledge, "refresh_existing", forbidden) + result = rebuild.rebuild_index(wiki_root=tmp_path / "wiki", data_dir=tmp_path / "data") + assert result["knowledge"] == expected["knowledge"] def test_explicit_work_year_is_not_misused_as_an_event_date(): diff --git a/test/test_publication_health.py b/test/test_publication_health.py index 75b985c..f85bf9b 100644 --- a/test/test_publication_health.py +++ b/test/test_publication_health.py @@ -221,3 +221,16 @@ def test_connection_readiness_is_distinct_from_graph_readiness(monkeypatch, actu expected = {"version": "expected"} if reason == "version_mismatch" else actual neo, graph = health.neo4j_readiness(expected) assert neo["ready"] is True and graph["ready"] is False and graph["reason"] == reason + + +def test_model_probe_requires_authentication_and_returns_sanitized_status(corpus, monkeypatch): + import tools.answers.model_client as model + seen = [] + monkeypatch.setattr(model, "probe_generation", lambda: seen.append(1) or + {"ready": True, "status": "ready", "reason": None}) + with TestClient(main.app) as client: + assert client.post("/health/generation").status_code == 401 + assert not seen + response = client.post("/health/generation", headers={"X-API-Key": "a" * 64}) + assert response.status_code == 200 and response.json()["ready"] is True + assert len(seen) == 1 diff --git a/test/test_sync_jobs.py b/test/test_sync_jobs.py index 5c9fb19..45cea73 100644 --- a/test/test_sync_jobs.py +++ b/test/test_sync_jobs.py @@ -176,3 +176,20 @@ def test_other_branches_and_deletions_do_not_queue(api, extra, reason): def test_malformed_push_is_rejected(api, extra): assert post_push(api, **extra).status_code == 400 assert main.app.state.sync_jobs.overview()["active"] == [] + + +def test_daily_schedule_is_durable_and_duplicate_workers_do_not_enqueue_twice(tmp_path, monkeypatch): + monkeypatch.setattr(time, "time", lambda: 1000) + jobs = SyncJobs(tmp_path, periodic_interval_seconds=86400) + assert jobs.schedule_if_due() + restarted = SyncJobs(tmp_path, periodic_interval_seconds=86400) + assert not restarted.schedule_if_due() + assert len(jobs.overview()["active"]) == 1 + assert jobs.overview()["schedule"]["next_due"] == 87400 + # Manual webhook gets the same queued job, without losing its Git update. + assert jobs.enqueue(pull_git=False)["coalesced"] + assert jobs.overview()["active"][0]["pull_git"] + monkeypatch.setattr(time, "time", lambda: 87401) + assert restarted.schedule_if_due() + assert len(jobs.overview()["active"]) == 1 + assert not SyncJobs(tmp_path).schedule_if_due() diff --git a/test/test_wiki_git.py b/test/test_wiki_git.py index 6ddbf99..7d9100d 100644 --- a/test/test_wiki_git.py +++ b/test/test_wiki_git.py @@ -93,3 +93,20 @@ def test_real_git_fetch_fast_forward_and_local_change_protection(tmp_path, monke wiki_git.update_wiki(checkout) assert git(checkout, "rev-parse", "HEAD") == local_head assert (checkout / "local.md").read_text() == "local commit" + + +def test_upstream_sync_rejects_wrong_destination_before_any_write(tmp_path, monkeypatch): + root = tmp_path / "wiki" + (root / ".git").mkdir(parents=True) + monkeypatch.setenv("SYNC_UPSTREAM_FORK", "1") + seen = [] + def run(args, **kwargs): + seen.append(args) + output = {"remote": "git@git.kemt.fei.tuke.sk:KEMT/zpwiki.git", + "branch": "master", "status": ""}.get(args[1], "") + return subprocess.CompletedProcess(args, 0, stdout=output, stderr="") + monkeypatch.setattr(wiki_git.subprocess, "run", run) + with pytest.raises(RuntimeError, match="restricted"): + wiki_git.update_wiki(root) + assert len(seen) == 3 + assert all(args[0] == "git" for args in seen) diff --git a/tools/answers/model_client.py b/tools/answers/model_client.py index cc39a00..9274265 100644 --- a/tools/answers/model_client.py +++ b/tools/answers/model_client.py @@ -6,6 +6,10 @@ import json import os import urllib.parse import urllib.request +import threading +import time +import hashlib +from datetime import datetime, timezone class NoRedirect(urllib.request.HTTPRedirectHandler): @@ -13,14 +17,77 @@ class NoRedirect(urllib.request.HTTPRedirectHandler): return None -def complete_with_configured_model(messages: list[dict], *, max_tokens: int | None = None) -> dict: - if max_tokens is not None and (type(max_tokens) is not int or not 1 <= max_tokens <= 16384): - raise ValueError("Invalid model output budget") +def model_configuration() -> dict: provider = os.getenv("ANSWER_PROVIDER", "openai_compatible").strip() model = os.getenv("ANSWER_MODEL", "").strip() if provider == "anthropic": key = (os.getenv("ANSWER_API_KEY") or os.getenv("ANTHROPIC_API_KEY", "")).strip() endpoint = os.getenv("ANSWER_LLM_URL", "https://api.anthropic.com/v1/messages").strip() + elif provider == "openai_compatible": + legacy = os.getenv("OPENWEBUI_API_KEY", "").strip() + key = os.getenv("ANSWER_API_KEY", "").strip() or legacy + endpoint = os.getenv("ANSWER_LLM_URL", "https://ui.tukekemt.xyz/api/chat/completions" if legacy else "").strip() + model = model or ("model120-fast" if legacy else "") + else: + raise RuntimeError("Unsupported answer provider") + parsed = urllib.parse.urlsplit(endpoint) + if (not key or not model or any(c in key for c in "\r\n") or parsed.scheme != "https" + or not parsed.hostname or parsed.username is not None or parsed.password is not None + or parsed.fragment or parsed.query or any(c.isspace() for c in endpoint)): + raise RuntimeError("Answer model is not configured") + return {'provider':provider, 'model':model, 'endpoint':endpoint, 'key':key} + + +_status_lock = threading.Lock() +_status = {} +HEALTH_MAX_AGE_SECONDS = 900 + + +def configuration_key(config: dict) -> str: + return hashlib.sha256(json.dumps(config,sort_keys=True).encode()).hexdigest() + + +def generation_status() -> dict: + try: + config = model_configuration() + except (RuntimeError,ValueError): + return {'ready':False,'status':'unavailable','reason':'model_not_configured','configured':False} + public={'configured':True,'provider':config['provider'],'model':config['model']} + with _status_lock: + status = dict(_status) + if status.get('configuration') != configuration_key(config) or time.time()-status.get('checked_at',0)>HEALTH_MAX_AGE_SECONDS: + return {**public,'ready':False,'status':'unknown','reason':'model_availability_not_verified'} + return {**public,'ready':status['success'],'status':'ready' if status['success'] else 'unavailable', + 'reason':None if status['success'] else 'model_probe_failed', + 'checked_at_utc':datetime.fromtimestamp(status['checked_at'],timezone.utc).isoformat()} + + +def complete_with_configured_model(messages: list[dict], *, max_tokens: int | None = None) -> dict: + config = model_configuration() + try: + result = _complete_request(messages,config,max_tokens=max_tokens) + except Exception: + with _status_lock: + _status.update(configuration=configuration_key(config), checked_at=time.time(),success=False) + raise + with _status_lock: + _status.update(configuration=configuration_key(config),checked_at=time.time(),success=True) + return result + + +def probe_generation() -> dict: + try: + complete_with_configured_model([{'role':'user','content':'Reply only with OK.'}],max_tokens=32) + except Exception: + pass # The caller gets only sanitized configuration/availability, never raw exceptions. + return generation_status() + + +def _complete_request(messages: list[dict], config: dict, *, max_tokens: int | None = None) -> dict: + if max_tokens is not None and (type(max_tokens) is not int or not 1 <= max_tokens <= 16384): + raise ValueError("Invalid model output budget") + provider,model,key,endpoint = (config[k] for k in ('provider','model','key','endpoint')) + if provider == "anthropic": system = "\n\n".join(row["content"] for row in messages if row["role"] == "system") body = { "model": model, @@ -32,33 +99,12 @@ def complete_with_configured_model(messages: list[dict], *, max_tokens: int | No body["system"] = system headers = {"x-api-key": key, "anthropic-version": "2023-06-01"} elif provider == "openai_compatible": - # Compatibility for historical experiments; new installations use ANSWER_API_KEY. - legacy = os.getenv("OPENWEBUI_API_KEY", "").strip() - key = os.getenv("ANSWER_API_KEY", "").strip() or legacy - endpoint = os.getenv( - "ANSWER_LLM_URL", "https://ui.tukekemt.xyz/api/chat/completions" if legacy else "" - ).strip() - model = model or ("model120-fast" if legacy else "") body = {"model": model, "messages": messages, "temperature": 0, "stream": False} if max_tokens is not None: body["max_tokens"] = max_tokens headers = {"Authorization": "Bearer " + key} else: raise RuntimeError("Unsupported answer provider") - parsed = urllib.parse.urlsplit(endpoint) - if ( - not key - or not model - or any(c in key for c in "\r\n") - or parsed.scheme != "https" - or not parsed.hostname - or parsed.username is not None - or parsed.password is not None - or parsed.fragment - or parsed.query - or any(c.isspace() for c in endpoint) - ): - raise RuntimeError("Answer model is not configured") request = urllib.request.Request( endpoint, data=json.dumps(body, ensure_ascii=False).encode(), @@ -78,4 +124,9 @@ def complete_with_configured_model(messages: list[dict], *, max_tokens: int | No if not text: raise RuntimeError("Missing answer model output") return {"choices": [{"message": {"content": text}}]} + if not isinstance(value,dict) or not value.get('choices'): + raise RuntimeError('Invalid answer model response') + choice=value['choices'][0] + if choice.get('finish_reason') not in (None,'stop') or not choice.get('message',{}).get('content'): + raise RuntimeError('Incomplete answer model output') return value diff --git a/tools/claude/build_claude_packages.py b/tools/claude/build_claude_packages.py index 58bb459..319debc 100644 --- a/tools/claude/build_claude_packages.py +++ b/tools/claude/build_claude_packages.py @@ -12,7 +12,7 @@ import zipfile ROOT = Path(__file__).resolve().parents[2] BASE = ROOT / "packages/claude" ENGINES = ("rag", "graphrag") -PACKAGE_VERSION = "1.8.2" +PACKAGE_VERSION = "1.8.3" def json_bytes(value): diff --git a/tools/git_runner/sync_fork.sh b/tools/git_runner/sync_fork.sh index 2309c6a..3d08fb0 100644 --- a/tools/git_runner/sync_fork.sh +++ b/tools/git_runner/sync_fork.sh @@ -8,26 +8,34 @@ fork_url='git@git.kemt.fei.tuke.sk:jp170na/zpwiki.git' upstream_url='git@git.kemt.fei.tuke.sk:KEMT/zpwiki.git' script_root="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" check_only=false +local_key='' case "${1:-}" in --check) check_only=true ;; + --key-file) + [[ $# -eq 2 && -f "$2" ]] || { echo "Chyba platny lokalny SSH kluc." >&2; exit 64; } + local_key="$2" + ;; '') ;; -h|--help) - echo "Pouzitie: bash tools/git_runner/sync_fork.sh [--check]" + echo "Pouzitie: bash tools/git_runner/sync_fork.sh [--check | --key-file /cesta/ku/klucu]" echo "--check iba porovna vetvy bez zapisu; pouzije dostupny lokalny SSH pristup." exit 0 ;; *) echo "Neznamy parameter." >&2; exit 64 ;; esac -if [[ $# -gt 1 ]]; then +if [[ $# -gt 1 && -z "$local_key" ]]; then echo "Prilis vela parametrov." >&2 exit 64 fi -if ! "$check_only"; then +if ! "$check_only" && [[ -z "$local_key" ]]; then : "${ZPWIKI_SYNC_SSH_KEY:?Chyba Actions secret ZPWIKI_SYNC_SSH_KEY}" fi export GIT_TERMINAL_PROMPT=0 LC_ALL=C +# Ignore URL rewrites and hooks in machine-wide/user Git configuration. +# The only write transport below is the explicit, fixed fork URL. +export GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null export ZPWIKI_KNOWN_HOSTS_FILE="$script_root/ssh_known_hosts" if [[ ! -s "$ZPWIKI_KNOWN_HOSTS_FILE" ]]; then @@ -38,7 +46,9 @@ fi work="$(mktemp -d /tmp/zpwiki-fork-sync.XXXXXXXX)" trap 'rm -rf "$work"' EXIT -if [[ -n "${ZPWIKI_SYNC_SSH_KEY:-}" ]]; then +if [[ -n "$local_key" ]]; then + export ZPWIKI_SSH_KEY_FILE="$local_key" +elif [[ -n "${ZPWIKI_SYNC_SSH_KEY:-}" ]]; then key_text="${ZPWIKI_SYNC_SSH_KEY//$'\r'/}" printf '%s\n' "${key_text%$'\n'}" > "$work/id_ed25519" unset key_text @@ -91,5 +101,5 @@ if "$check_only"; then fi # Normal push also rejects a concurrent divergent update on the server. -git_cmd push origin "$upstream:refs/heads/master" +git_cmd push "$fork_url" "$upstream:refs/heads/master" echo "Fork bol aktualizovany na $upstream." diff --git a/tools/knowledge/build.py b/tools/knowledge/build.py index 18fab1f..4b66e3b 100644 --- a/tools/knowledge/build.py +++ b/tools/knowledge/build.py @@ -55,9 +55,43 @@ def build( def _build(wiki_root, data_dir, complete, model_identity, *, graph, vectors, workers, progress): + import tempfile + import shutil + from tools.sync.index_state import index_snapshot + from tools.knowledge.publication import checkpoint_state, publish_checkpoint, PENDING_FILE + from tools.knowledge.store import create_constraints + from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings + from tools.sync.sync_indexes import source_files + with index_snapshot(data_dir) as version: + if version is None: + raise RuntimeError("Build the shared source indexes before content extraction.") + index, summary = prepare_index(wiki_root, data_dir, complete, model_identity, + graph=graph, vectors=vectors, workers=workers, progress=progress, source_version=version) + stage = Path(tempfile.mkdtemp(prefix=".knowledge-stage-", dir=data_dir)) + try: + checkpoint_state(stage, index) + with publication_lock(data_dir, exclusive=True): + source_digest=hashlib.sha256(json.dumps(source_files(wiki_root),sort_keys=True).encode()).hexdigest() + if read_version(data_dir) != version or source_digest != version['source_sha256']: + raise RuntimeError("Source changed during extraction; checkpoint not activated.") + if graph: + settings = get_neo4j_settings() + with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: + driver.verify_connectivity() + create_constraints(session) + publish_checkpoint(data_dir, stage, index, session) + else: + publish_checkpoint(data_dir, stage, index) + return summary + finally: + if not (data_dir / PENDING_FILE).exists(): + shutil.rmtree(stage) + + +def prepare_index(wiki_root, data_dir, complete, model_identity, *, graph=True, vectors=True, workers=2, progress=None, source_version=None): from tools.sync.sync_indexes import source_files - version = read_version(data_dir) + version = source_version or read_version(data_dir) if not version: raise RuntimeError("Build the shared source indexes before content extraction.") sources = source_files(wiki_root) @@ -210,15 +244,10 @@ def _build(wiki_root, data_dir, complete, model_identity, *, graph, vectors, wor if vectors: index.update(embedding_index(extracted, data_dir)) index["version"] = identifier(json.dumps(index, ensure_ascii=False, sort_keys=True)) - with publication_lock(data_dir, exclusive=True): - if read_version(data_dir) != version or source_files(wiki_root) != sources: - raise RuntimeError("Source changed during extraction; checkpoint not activated.") - if graph: - from tools.knowledge.store import write_graph + if source_files(wiki_root) != sources: + raise RuntimeError("Source changed during extraction; checkpoint not activated.") + return index, { - write_graph(index) - durable_json(data_dir / "knowledge/index.json", index) - return { "profiles": len(documents), "evidence_units": len(extracted), "llm_batches": calls, diff --git a/tools/knowledge/publication.py b/tools/knowledge/publication.py new file mode 100644 index 0000000..25e9096 --- /dev/null +++ b/tools/knowledge/publication.py @@ -0,0 +1,119 @@ +"""Recoverable publication of a content checkpoint and its Neo4j projection.""" +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import shutil + +from tools.sync.index_state import durable_json, fsync_directory, IndexUnavailableError + +FILES = ("knowledge/index.json", "knowledge/state.json") +PENDING_FILE = "knowledge/.publication-pending.json" + + +def checkpoint_state(stage: Path, index: dict) -> dict: + directory = stage / "knowledge" + directory.mkdir(parents=True, exist_ok=True) + durable_json(directory / "index.json", index) + content = (directory / "index.json").read_bytes() + state = { + "schema": 1, + "version": index["version"], + "index_version": index["index_version"], + "model": index.get("model", {}), + "graph_ready": index.get("graph_ready", False), + "embedding_fingerprint": index.get("embedding_fingerprint"), + "checkpoint_sha256": hashlib.sha256(content).hexdigest(), + "checkpoint_bytes": len(content), + "profiles": index.get("profiles"), + "units": len(index["units"]), + } + durable_json(directory / "state.json", state) + return state + + +def read_state(data_dir: Path) -> dict | None: + from tools.sync.index_state import PENDING_FILE as SOURCE_PENDING + if (data_dir / PENDING_FILE).exists() or (data_dir / SOURCE_PENDING).exists(): + raise IndexUnavailableError("Publikovanie obsahovej extrakcie nebolo dokončené.") + path = data_dir / "knowledge/state.json" + return json.loads(path.read_text()) if path.exists() else None + + +def state_matches_graph(state: dict, actual: dict | None) -> bool: + return bool(actual and actual.get("version") == state["version"] + and actual.get("checkpoint_sha256") == state["checkpoint_sha256"] + and actual.get("index_version_json") == json.dumps(state["index_version"], sort_keys=True)) + + +def replace_file(source: Path, target: Path) -> None: + os.replace(source, target) + + +def publish_checkpoint(data_dir: Path, stage: Path, index: dict, session=None) -> None: + """Caller holds the exclusive publication lock and content build lock. + + A failed/unknown commit leaves a durable marker; readers refuse that + checkpoint until a successful rebuild. A known rollback restores both files. + """ + from tools.knowledge.store import graph_state, write_graph_transaction + + state = json.loads((stage / "knowledge/state.json").read_text()) + old_graph = graph_state(session) if session is not None else None + backup = stage / "previous/knowledge" + backup.mkdir(parents=True) + originals = [] + for name in FILES: + old = data_dir / name + if old.exists(): + target = stage / "previous" / name + shutil.copy2(old, target) + with target.open("rb") as stream: + os.fsync(stream.fileno()) + originals.append(name) + fsync_directory(backup) + had_pending = (data_dir / PENDING_FILE).exists() + durable_json(data_dir / PENDING_FILE, + {"staging_directory": stage.name, "target_version": index["version"]}) + commit_attempted = False + try: + if session is None: + for name in FILES: + replace_file(stage / name, data_dir / name) + fsync_directory(data_dir / "knowledge") + else: + with session.begin_transaction() as tx: + write_graph_transaction(tx, index, checkpoint_sha256=state["checkpoint_sha256"]) + for name in FILES: + replace_file(stage / name, data_dir / name) + fsync_directory(data_dir / "knowledge") + commit_attempted = True + tx.commit() + except BaseException: + if commit_attempted: + # Never equate a lost acknowledgement with a rolled-back commit. + actual = graph_state(session) + if state_matches_graph(state, actual): + if json.loads((data_dir / "knowledge/state.json").read_text()) != state: + raise + if hashlib.sha256((data_dir / "knowledge/index.json").read_bytes()).hexdigest() != state["checkpoint_sha256"]: + raise + (data_dir / PENDING_FILE).unlink() + fsync_directory(data_dir / "knowledge") + return + if actual != old_graph: + raise + for name in FILES: + if name in originals: + replace_file(stage / "previous" / name, data_dir / name) + else: + (data_dir / name).unlink(missing_ok=True) + fsync_directory(data_dir / "knowledge") + if not had_pending: + (data_dir / PENDING_FILE).unlink() + fsync_directory(data_dir / "knowledge") + raise + (data_dir / PENDING_FILE).unlink() + fsync_directory(data_dir / "knowledge") diff --git a/tools/knowledge/query.py b/tools/knowledge/query.py index 9fec79d..90722d5 100644 --- a/tools/knowledge/query.py +++ b/tools/knowledge/query.py @@ -974,7 +974,10 @@ def deliver_content(question: str, *, engine: str, limit: int, db_file, complete from tools.knowledge.sources import source_lifecycle_notes from tools.rag.common import ZPWIKI_ROOT - notes = source_lifecycle_notes(ZPWIKI_ROOT, index["index_version"].get("source_sha256")) + # Current checkpoints already carry lifecycle evidence. A changed local + # clone must never supplement a published generation with newer notes. + notes = ([] if index.get("academic_schema") == 1 else + source_lifecycle_notes(ZPWIKI_ROOT, index["index_version"].get("source_sha256"))) existing = {(row["source_path"], row.get("work_label"), row["text"]) for row in units} units = [ *units, diff --git a/tools/knowledge/store.py b/tools/knowledge/store.py index 0ba0028..f8ec1e2 100644 --- a/tools/knowledge/store.py +++ b/tools/knowledge/store.py @@ -11,8 +11,26 @@ from tools.answers.context_policy import normalized SCOPE = "zpwiki-content-v1" -def write_graph(index: dict) -> None: - settings = get_neo4j_settings() +def create_constraints(session) -> None: + for name, label, field in [ + ("evidence", "KnowledgeEvidence", "id"), + ("concept", "KnowledgeConcept", "id"), + ("person", "KnowledgePerson", "source_path"), + ("work", "KnowledgeWork", "id"), + ("advisor", "KnowledgeAdvisor", "id"), + ]: + session.run( + f"CREATE CONSTRAINT zp_content_{name} IF NOT EXISTS " + f"FOR (n:{label}) REQUIRE (n.graph_scope,n.{field}) IS UNIQUE" + ).consume() + session.run( + "CREATE CONSTRAINT zp_content_state IF NOT EXISTS " + "FOR (n:KnowledgeState) REQUIRE n.scope IS UNIQUE" + ).consume() + + +def write_graph_transaction(tx, index: dict, *, checkpoint_sha256: str | None = None) -> None: + """The caller owns the transaction and filesystem publication barrier.""" rows = [] for unit in index["units"]: rows.append( @@ -24,162 +42,155 @@ def write_graph(index: dict) -> None: "relations_json": json.dumps(unit["relations"], ensure_ascii=False), } ) - with ( - create_neo4j_driver(settings) as driver, - driver.session(database=settings.database) as session, - ): - driver.verify_connectivity() - for name, label, field in [ - ("evidence", "KnowledgeEvidence", "id"), - ("concept", "KnowledgeConcept", "id"), - ("person", "KnowledgePerson", "source_path"), - ("work", "KnowledgeWork", "id"), - ("advisor", "KnowledgeAdvisor", "id"), - ]: - session.run( - f"CREATE CONSTRAINT zp_content_{name} IF NOT EXISTS " - f"FOR (n:{label}) REQUIRE (n.graph_scope,n.{field}) IS UNIQUE" - ).consume() - session.run( - "CREATE CONSTRAINT zp_content_state IF NOT EXISTS " - "FOR (n:KnowledgeState) REQUIRE n.scope IS UNIQUE" + tx.run( + "MATCH (n:KnowledgeRelation {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE + ).consume() + tx.run( + "MATCH (n:KnowledgeEvidence {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE + ).consume() + tx.run( + "MATCH (n:KnowledgeConcept {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE + ).consume() + tx.run( + "MATCH (n:KnowledgePerson {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE + ).consume() + evidence_rows = [ + { + "evidence": {k: v for k, v in row.items() if k != "concepts"} + | {"graph_scope": SCOPE}, + "concepts": row["concepts"], + } + for row in rows + ] + tx.run( + """UNWIND $rows AS row + CREATE (e:KnowledgeEvidence) SET e=row.evidence + MERGE (p:KnowledgePerson {source_path:row.evidence.source_path,graph_scope:$scope}) + SET p.name=row.evidence.person + MERGE (p)-[:HAS_EVIDENCE]->(e) + FOREACH (concept IN row.concepts | + MERGE (c:KnowledgeConcept {id:concept.id,graph_scope:$scope}) + SET c.name=concept.name MERGE (e)-[:ABOUT]->(c))""", + rows=evidence_rows, + scope=SCOPE, + ).consume() + relations = [ + { + "left": identifier(normalized(r["subject"])), + "right": identifier(normalized(r["object"])), + "evidence": row["id"], + "predicate": r["predicate"], + } + for row in rows + for r in json.loads(row["relations_json"]) + ] + tx.run( + """UNWIND $rows AS row + MATCH (a:KnowledgeConcept {id:row.left,graph_scope:$scope}), + (b:KnowledgeConcept {id:row.right,graph_scope:$scope}), + (e:KnowledgeEvidence {id:row.evidence,graph_scope:$scope}) + CREATE (r:KnowledgeRelation {graph_scope:$scope,predicate:row.predicate,origin:'llm'}) + CREATE (r)-[:SUBJECT]->(a), (r)-[:OBJECT]->(b), (r)-[:EVIDENCED_BY]->(e)""", + rows=relations, + scope=SCOPE, + ).consume() + from tools.knowledge.sources import advisor_name + tx.run( + "MATCH (n:KnowledgeWork {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE + ).consume() + tx.run( + "MATCH (n:KnowledgeAdvisor {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE + ).consume() + works = [ + { + "id": identifier(u["source_path"], u["work_label"]), + **{ + k: u.get(k) for k in ("source_path", "work_label", "work_type", "work_year") + }, + "evidence": u["id"], + } + for u in index["units"] + if u["kind"] == "work_record" + ] + tx.run( + """UNWIND $rows AS row + MATCH (p:KnowledgePerson {source_path:row.source_path,graph_scope:$scope}), + (e:KnowledgeEvidence {id:row.evidence,graph_scope:$scope}) + MERGE (w:KnowledgeWork {id:row.id,graph_scope:$scope}) + SET w.label=row.work_label,w.type=row.work_type,w.year=row.work_year + MERGE (p)-[:HAS_WORK]->(w) MERGE (w)-[:EVIDENCED_BY]->(e)""", + rows=works, + scope=SCOPE, + ).consume() + for role, edge in [("supervisor", "SUPERVISED_BY"), ("consultant", "CONSULTED_BY")]: + advisors = [ + { + "work": identifier(u["source_path"], u["work_label"]), + "id": identifier(normalized(name)), + "name": name, + "evidence": u["id"], + } + for u in index["units"] + if u.get("work_label") and (name := advisor_name(u, role)) + ] + tx.run( + f"""UNWIND $rows AS row + MATCH (w:KnowledgeWork {{id:row.work,graph_scope:$scope}}), + (e:KnowledgeEvidence {{id:row.evidence,graph_scope:$scope}}) + MERGE (a:KnowledgeAdvisor {{id:row.id,graph_scope:$scope}}) SET a.name=row.name + MERGE (w)-[:{edge}]->(a) MERGE (e)-[:DOCUMENTS_ROLE]->(a)""", + rows=advisors, + scope=SCOPE, ).consume() - with session.begin_transaction() as tx: - tx.run( - "MATCH (n:KnowledgeRelation {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE - ).consume() - tx.run( - "MATCH (n:KnowledgeEvidence {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE - ).consume() - tx.run( - "MATCH (n:KnowledgeConcept {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE - ).consume() - tx.run( - "MATCH (n:KnowledgePerson {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE - ).consume() - # Copy only scalar evidence properties. Dynamic labels/Cypher from - # the model are never executed; all relationships have fixed types. - evidence_rows = [ - { - "evidence": {k: v for k, v in row.items() if k != "concepts"} - | {"graph_scope": SCOPE}, - "concepts": row["concepts"], - } - for row in rows - ] - tx.run( - """UNWIND $rows AS row - CREATE (e:KnowledgeEvidence) SET e=row.evidence - MERGE (p:KnowledgePerson {source_path:row.evidence.source_path,graph_scope:$scope}) - SET p.name=row.evidence.person - MERGE (p)-[:HAS_EVIDENCE]->(e) - FOREACH (concept IN row.concepts | - MERGE (c:KnowledgeConcept {id:concept.id,graph_scope:$scope}) - SET c.name=concept.name MERGE (e)-[:ABOUT]->(c))""", - rows=evidence_rows, - scope=SCOPE, - ).consume() - relations = [ - { - "left": identifier(normalized(r["subject"])), - "right": identifier(normalized(r["object"])), - "evidence": row["id"], - "predicate": r["predicate"], - } - for row in rows - for r in json.loads(row["relations_json"]) - ] - tx.run( - """UNWIND $rows AS row - MATCH (a:KnowledgeConcept {id:row.left,graph_scope:$scope}), - (b:KnowledgeConcept {id:row.right,graph_scope:$scope}), - (e:KnowledgeEvidence {id:row.evidence,graph_scope:$scope}) - CREATE (r:KnowledgeRelation {graph_scope:$scope,predicate:row.predicate,origin:'llm'}) - CREATE (r)-[:SUBJECT]->(a), (r)-[:OBJECT]->(b), (r)-[:EVIDENCED_BY]->(e)""", - rows=relations, - scope=SCOPE, - ).consume() - from tools.knowledge.sources import advisor_name + tx.run( + """MATCH (e:KnowledgeEvidence {graph_scope:$scope}) + WHERE e.meeting_id IS NOT NULL AND e.kind <> 'meeting' + MATCH (m:KnowledgeEvidence {graph_scope:$scope,kind:'meeting',meeting_id:e.meeting_id}) + MERGE (e)-[:RECORDED_AT]->(m)""", + scope=SCOPE, + ).consume() + tx.run( + """MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_WORK]->(w), + (p)-[:HAS_EVIDENCE]->(e) + WHERE e.work_label=w.label AND e.kind='meeting' + MERGE (w)-[:HAS_MEETING]->(e)""", + scope=SCOPE, + ).consume() + tx.run( + """MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_WORK]->(w), + (p)-[:HAS_EVIDENCE]->(e) + WHERE e.work_label=w.label AND e.completion_status IS NOT NULL + MERGE (w)-[:HAS_STATUS_RECORD]->(e)""", + scope=SCOPE, + ).consume() + tx.run( + "MERGE (s:KnowledgeState {scope:$scope}) SET s.version=$version, s.index_version_json=$index_version_json, s.checkpoint_sha256=$checkpoint_sha256", + scope=SCOPE, + version=index["version"], + index_version_json=json.dumps(index.get("index_version"), sort_keys=True), + checkpoint_sha256=checkpoint_sha256, + ).consume() - # Roles are explicit, source-backed edges attached to an exact work. - # Profile-level roles remain unassigned; they are never spread over - # every work on that profile. - tx.run( - "MATCH (n:KnowledgeWork {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE - ).consume() - tx.run( - "MATCH (n:KnowledgeAdvisor {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE - ).consume() - works = [ - { - "id": identifier(u["source_path"], u["work_label"]), - **{ - k: u.get(k) for k in ("source_path", "work_label", "work_type", "work_year") - }, - "evidence": u["id"], - } - for u in index["units"] - if u["kind"] == "work_record" - ] - tx.run( - """UNWIND $rows AS row - MATCH (p:KnowledgePerson {source_path:row.source_path,graph_scope:$scope}), - (e:KnowledgeEvidence {id:row.evidence,graph_scope:$scope}) - MERGE (w:KnowledgeWork {id:row.id,graph_scope:$scope}) - SET w.label=row.work_label,w.type=row.work_type,w.year=row.work_year - MERGE (p)-[:HAS_WORK]->(w) MERGE (w)-[:EVIDENCED_BY]->(e)""", - rows=works, - scope=SCOPE, - ).consume() - for role, edge in [("supervisor", "SUPERVISED_BY"), ("consultant", "CONSULTED_BY")]: - advisors = [ - { - "work": identifier(u["source_path"], u["work_label"]), - "id": identifier(normalized(name)), - "name": name, - "evidence": u["id"], - } - for u in index["units"] - if u.get("work_label") and (name := advisor_name(u, role)) - ] - tx.run( - f"""UNWIND $rows AS row - MATCH (w:KnowledgeWork {{id:row.work,graph_scope:$scope}}), - (e:KnowledgeEvidence {{id:row.evidence,graph_scope:$scope}}) - MERGE (a:KnowledgeAdvisor {{id:row.id,graph_scope:$scope}}) SET a.name=row.name - MERGE (w)-[:{edge}]->(a) MERGE (e)-[:DOCUMENTS_ROLE]->(a)""", - rows=advisors, - scope=SCOPE, - ).consume() - tx.run( - """MATCH (e:KnowledgeEvidence {graph_scope:$scope}) - WHERE e.meeting_id IS NOT NULL AND e.kind <> 'meeting' - MATCH (m:KnowledgeEvidence {graph_scope:$scope,kind:'meeting',meeting_id:e.meeting_id}) - MERGE (e)-[:RECORDED_AT]->(m)""", - scope=SCOPE, - ).consume() - tx.run( - """MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_WORK]->(w), - (p)-[:HAS_EVIDENCE]->(e) - WHERE e.work_label=w.label AND e.kind='meeting' - MERGE (w)-[:HAS_MEETING]->(e)""", - scope=SCOPE, - ).consume() - tx.run( - """MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_WORK]->(w), - (p)-[:HAS_EVIDENCE]->(e) - WHERE e.work_label=w.label AND e.completion_status IS NOT NULL - MERGE (w)-[:HAS_STATUS_RECORD]->(e)""", - scope=SCOPE, - ).consume() - tx.run( - "MERGE (s:KnowledgeState {scope:$scope}) SET s.version=$version", - scope=SCOPE, - version=index["version"], - ).consume() + +def graph_state(session) -> dict | None: + record = session.run( + "MATCH (s:KnowledgeState {scope:$scope}) RETURN properties(s) AS state", scope=SCOPE + ).single() + return dict(record["state"]) if record else None + + +def write_graph(index: dict) -> None: + """Low-level graph import; the production builder publishes through a journal.""" + settings = get_neo4j_settings() + with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: + driver.verify_connectivity() + create_constraints(session) + with session.begin_transaction() as tx: + write_graph_transaction(tx, index) tx.commit() + def read_graph(index: dict) -> list[dict]: settings = get_neo4j_settings() with ( @@ -188,10 +199,13 @@ def read_graph(index: dict) -> list[dict]: ): with session.begin_transaction() as tx: state = tx.run( - "MATCH (s:KnowledgeState {scope:$scope}) RETURN s.version AS version", scope=SCOPE + "MATCH (s:KnowledgeState {scope:$scope}) RETURN s.version AS version, " + "s.checkpoint_sha256 AS checkpoint_sha256", scope=SCOPE ).single() if not state or state["version"] != index["version"]: raise RuntimeError("Content graph and extraction checkpoint differ.") + if index.get("_checkpoint_sha256") and state["checkpoint_sha256"] != index["_checkpoint_sha256"]: + raise RuntimeError("Content graph and checkpoint checksum differ.") rows = tx.run( """MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_EVIDENCE]->(e) OPTIONAL MATCH (e)-[:ABOUT]->(c:KnowledgeConcept) @@ -204,10 +218,22 @@ def read_graph(index: dict) -> list[dict]: def load_index(data_dir: Path, version: dict | None) -> dict | None: + from tools.knowledge.publication import read_state + import hashlib + + state = read_state(data_dir) path = data_dir / "knowledge/index.json" if not path.exists(): return None - index = json.loads(path.read_text()) + content = path.read_bytes() + index = json.loads(content) if not version or index.get("index_version") != version: raise RuntimeError("Content extraction belongs to another source snapshot. Rebuild it.") + if state is not None: + if not isinstance(state, dict) or state.get("schema") != 1: + raise RuntimeError("Content extraction publication state is invalid.") + if (state.get("version") != index.get("version") or state.get("index_version") != version + or hashlib.sha256(content).hexdigest() != state.get("checkpoint_sha256")): + raise RuntimeError("Content extraction checkpoint is invalid. Rebuild it.") + index["_checkpoint_sha256"] = state["checkpoint_sha256"] return index diff --git a/tools/rag/rebuild_index.py b/tools/rag/rebuild_index.py index 75e611f..6935d67 100644 --- a/tools/rag/rebuild_index.py +++ b/tools/rag/rebuild_index.py @@ -123,13 +123,6 @@ def rebuild_index( else: result = synchronize_indexes(root, target) - from tools.knowledge.build import refresh_existing - - if (target / "knowledge/index.json").exists(): - if progress: - progress("extracting_knowledge") - result["knowledge"] = refresh_existing(root, target) - duration = round( time.monotonic() - start, 2, diff --git a/tools/runtime/answer_health.py b/tools/runtime/answer_health.py new file mode 100644 index 0000000..6443b74 --- /dev/null +++ b/tools/runtime/answer_health.py @@ -0,0 +1,83 @@ +"""Read-only readiness of versioned content and the configured answer model.""" +from __future__ import annotations + +import hashlib +import json +from functools import lru_cache +from pathlib import Path + +from neo4j import Query + +from tools.answers.model_client import generation_status +from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings +from tools.knowledge.publication import read_state, state_matches_graph +from tools.knowledge.store import SCOPE +from tools.runtime.tool_health import PROBE_TIMEOUT_SECONDS, readiness +from tools.sync.index_state import IndexUnavailableError, index_snapshot + + +@lru_cache(maxsize=2) +def _checkpoint(path: str, inode: int, size: int, modified_ns: int, changed_ns: int): + content = Path(path).read_bytes() + index = json.loads(content) + return {"version": index.get("version"), "index_version": index.get("index_version"), + "sha256": hashlib.sha256(content).hexdigest(), "bytes": len(content)} + + +def content_readiness(db_file: Path, version: dict | None, *, graph_available: bool) -> dict: + try: + # tools_readiness released its lock: recheck the generation before reading. + with index_snapshot(db_file.parent) as current: + if not version or current != version: + return readiness(False, "snapshot_changed") + state = read_state(db_file.parent) + path = db_file.parent / "knowledge/index.json" + if not path.is_file(): + return readiness(False, "content_index_missing") + if not isinstance(state, dict) or state.get("schema") != 1: + return readiness(False, "content_publication_required") + st = path.stat() + actual = _checkpoint(str(path), st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns) + if state.get("index_version") != version or actual["index_version"] != version: + return readiness(False, "content_version_mismatch") + if (actual["version"] != state.get("version") + or actual["sha256"] != state.get("checkpoint_sha256") + or actual["bytes"] != state.get("checkpoint_bytes")): + return readiness(False, "content_checkpoint_invalid") + details = {"index_version": version, "content_version": state["version"], + "checkpoint_sha256": actual["sha256"], "profiles": state.get("profiles"), + "units": state.get("units")} + graph = readiness(False, "content_graph_disabled") + if state.get("graph_ready"): + graph = _graph_readiness(state) if graph_available else readiness(False, "neo4j_unavailable") + return {**readiness(True, **details), "graph": graph} + except IndexUnavailableError: + return readiness(False, "content_publication_incomplete") + except (OSError, ValueError, TypeError, KeyError): + return readiness(False, "content_checkpoint_invalid") + + +def _graph_readiness(state): + try: + settings = get_neo4j_settings() + with create_neo4j_driver(settings, connection_timeout=PROBE_TIMEOUT_SECONDS, + connection_acquisition_timeout=PROBE_TIMEOUT_SECONDS, + connection_write_timeout=PROBE_TIMEOUT_SECONDS, + max_transaction_retry_time=0, max_connection_pool_size=1) as driver: + with driver.session(database=settings.database) as session: + row = session.run(Query( + "MATCH (s:KnowledgeState {scope:$scope}) RETURN properties(s) AS state", + timeout=PROBE_TIMEOUT_SECONDS), scope=SCOPE).single() + actual = dict(row["state"]) if row else None + if not state_matches_graph(state, actual): + return readiness(False, "content_graph_version_mismatch") + return readiness(True, content_version=state["version"]) + except Exception: + return readiness(False, "content_graph_unavailable") + + +def answer_capabilities(db_file, tools, version): + content = content_readiness(db_file, version, graph_available=tools["neo4j"]["ready"]) + generation = generation_status() + return {"text_search": tools["rag"], "metadata_graph": tools["graphrag"], + "knowledge": content, "generation": generation} diff --git a/tools/sync/sync_indexes.py b/tools/sync/sync_indexes.py index e72273b..a8aba61 100644 --- a/tools/sync/sync_indexes.py +++ b/tools/sync/sync_indexes.py @@ -16,7 +16,7 @@ import sqlite3 import subprocess import tempfile import uuid -from contextlib import closing +from contextlib import closing, ExitStack from datetime import datetime, timezone from pathlib import Path @@ -121,21 +121,28 @@ def replace_file(source: Path, target: Path) -> None: os.replace(source, target) -def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, session) -> dict: +def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, session, *, knowledge=None) -> dict: with publication_lock(data_dir, exclusive=True): + from tools.knowledge import publication as content_publication + from tools.knowledge import store as content_store + files = (*FILES, *content_publication.FILES) if knowledge is not None else FILES + content_state = (json.loads((stage / "knowledge/state.json").read_text()) + if knowledge is not None else None) had_pending = (data_dir / PENDING_FILE).exists() old_graph_version = graph_version(session) originals = [] backup = stage / "previous" backup.mkdir() - for name in FILES: + for name in files: current = data_dir / name if current.exists(): + (backup / name).parent.mkdir(parents=True, exist_ok=True) shutil.copy2(current, backup / name) with (backup / name).open("rb") as handle: os.fsync(handle.fileno()) originals.append(name) - fsync_directory(backup) + for directory in {backup, *((backup / name).parent for name in originals)}: + fsync_directory(directory) durable_json( data_dir / PENDING_FILE, { @@ -148,17 +155,22 @@ def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, try: with session.begin_transaction() as tx: write_graph_transaction(tx, payload, version=version) + if knowledge is not None: + content_store.write_graph_transaction(tx, knowledge, + checkpoint_sha256=content_state["checkpoint_sha256"]) 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: + for name in files: # Flush staged contents before the durable directory switch. with (stage / name).open("rb") as handle: os.fsync(handle.fileno()) + (data_dir / name).parent.mkdir(parents=True, exist_ok=True) replace_file(stage / name, data_dir / name) - fsync_directory(data_dir) + for parent in { (data_dir / name).parent for name in files }: + fsync_directory(parent) commit_attempted = True tx.commit() except BaseException: @@ -173,22 +185,35 @@ def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, current_graph_version == version and sqlite_version(data_dir / "zp_index.sqlite") == version ): + if knowledge is not None: + if not content_publication.state_matches_graph(content_state, content_store.graph_state(session)): + raise + if (json.loads((data_dir / "knowledge/state.json").read_text()) != content_state + or digest(data_dir / "knowledge/index.json") != content_state["checkpoint_sha256"]): + raise (data_dir / PENDING_FILE).unlink() + if knowledge is not None: + (data_dir / content_publication.PENDING_FILE).unlink(missing_ok=True) + fsync_directory(data_dir / 'knowledge') fsync_directory(data_dir) return graph_summary if current_graph_version != old_graph_version: raise # Unknown external writer; fail closed. - for name in FILES: + 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) + for directory in {(data_dir / name).parent for name in files}: + fsync_directory(directory) if not had_pending: (data_dir / PENDING_FILE).unlink() fsync_directory(data_dir) raise (data_dir / PENDING_FILE).unlink() + if knowledge is not None: + (data_dir / content_publication.PENDING_FILE).unlink(missing_ok=True) + fsync_directory(data_dir / 'knowledge') fsync_directory(data_dir) return graph_summary @@ -198,16 +223,24 @@ def synchronize_indexes(wiki_root: Path, data_dir: Path, *, progress=None) -> di stage = Path(tempfile.mkdtemp(prefix=".sync-stage-", dir=data_dir)) try: version, counts, payload = prepare_snapshot(wiki_root, stage) - if progress: - progress("publishing_indexes") - 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) + knowledge = None + knowledge_summary = None + with ExitStack() as locks: + if (data_dir / "knowledge/index.json").exists(): + from tools.rag.rebuild_index import acquire_reindex_lock + from tools.knowledge.build import prepare_index, configured_extractor + from tools.knowledge.publication import checkpoint_state + locks.enter_context(acquire_reindex_lock(data_dir / "knowledge/.build.lock")) + if progress: + progress("extracting_knowledge") + complete, identity = configured_extractor() + knowledge, knowledge_summary = prepare_index(stage / 'wiki', data_dir, complete, identity, + source_version=version) + checkpoint_state(stage, knowledge) + if git_revision(wiki_root) != version['git_revision'] or hashlib.sha256( + json.dumps(source_files(wiki_root),sort_keys=True).encode()).hexdigest() != version['source_sha256']: + raise RuntimeError("Wiki sa počas extrakcie zmenila; generácia nebola aktivovaná.") + graph_summary = _publish_prepared(data_dir, stage, version, payload, knowledge, progress) return { "counts": counts, "documents_scanned": counts["documents"], @@ -215,6 +248,7 @@ def synchronize_indexes(wiki_root: Path, data_dir: Path, *, progress=None) -> di "database_path": str(data_dir / "zp_index.sqlite"), "index_version": version, "graph_counts": graph_summary, + **({"knowledge": knowledge_summary} if knowledge_summary is not None else {}), } finally: # Keep recovery material when a crash/ambiguous commit is unresolved. @@ -227,3 +261,16 @@ def synchronize_indexes(wiki_root: Path, data_dir: Path, *, progress=None) -> di keep_stage = True if not keep_stage: shutil.rmtree(stage) + + +def _publish_prepared(data_dir, stage, version, payload, knowledge, progress): + if progress: + progress("publishing_indexes") + settings = get_neo4j_settings() + with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: + driver.verify_connectivity() + create_constraints(session) + if knowledge is not None: + from tools.knowledge.store import create_constraints as create_content_constraints + create_content_constraints(session) + return publish_snapshot(data_dir, stage, version, payload, session, knowledge=knowledge) diff --git a/tools/sync/sync_jobs.py b/tools/sync/sync_jobs.py index 619c4dd..cda614f 100644 --- a/tools/sync/sync_jobs.py +++ b/tools/sync/sync_jobs.py @@ -18,10 +18,13 @@ logger = logging.getLogger(__name__) class SyncJobs: - def __init__(self, data_dir: Path, *, runner: Callable | None = None): + def __init__(self, data_dir: Path, *, runner: Callable | None = None, periodic_interval_seconds: float = 0): self.data_dir = Path(data_dir) self.data_dir.mkdir(parents=True, exist_ok=True) self.path = self.data_dir / "sync_jobs.sqlite" + if periodic_interval_seconds < 0: + raise ValueError("Synchronization interval must not be negative") + self.periodic_interval_seconds = periodic_interval_seconds self.runner = runner or rebuild_index self.stopping = threading.Event() self.thread: threading.Thread | None = None @@ -34,6 +37,7 @@ class SyncJobs: request_count INTEGER NOT NULL DEFAULT 1, attempts INTEGER NOT NULL DEFAULT 0, available_at REAL NOT NULL DEFAULT 0, reason TEXT, result_json TEXT); CREATE INDEX IF NOT EXISTS jobs_status ON jobs(status, created_at); + CREATE TABLE IF NOT EXISTS schedule (id INTEGER PRIMARY KEY CHECK(id=1), next_due REAL NOT NULL); CREATE TABLE IF NOT EXISTS deliveries ( delivery_key TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES jobs(id)); """) @@ -64,45 +68,62 @@ class SyncJobs: now = time.time() with closing(self.connect()) as conn, conn: conn.execute("BEGIN IMMEDIATE") - if delivery_key: - duplicate = conn.execute( - "SELECT job_id FROM deliveries WHERE delivery_key=?", (delivery_key,) - ).fetchone() - if duplicate: - return { - "job": self.public( - conn.execute( - "SELECT * FROM jobs WHERE id=?", (duplicate["job_id"],) - ).fetchone() - ), - "coalesced": False, - "duplicate": True, - } - # Never merge a new request into a running job: its source snapshot - # may already exist. One queued follow-up absorbs further pushes. - queued = conn.execute( - "SELECT * FROM jobs WHERE status='queued' ORDER BY created_at LIMIT 1" + return self._enqueue(conn, now, pull_git, delivery_key, requested_revision) + + def _enqueue(self, conn, now, pull_git, delivery_key=None, requested_revision=None): + if delivery_key: + duplicate = conn.execute( + "SELECT job_id FROM deliveries WHERE delivery_key=?", (delivery_key,) ).fetchone() - if queued: - job_id = queued["id"] - conn.execute( - """UPDATE jobs SET pull_git=MAX(pull_git,?), updated_at=?, - requested_revision=COALESCE(?,requested_revision), request_count=request_count+1 - WHERE id=?""", - (int(pull_git), now, requested_revision, job_id), - ) - else: - job_id = uuid.uuid4().hex - conn.execute( - """INSERT INTO jobs(id,status,phase,pull_git,created_at,updated_at,requested_revision) - VALUES(?,'queued','queued',?,?,?,?)""", - (job_id, int(pull_git), now, now, requested_revision), - ) - if delivery_key: - conn.execute("INSERT INTO deliveries VALUES(?,?)", (delivery_key, job_id)) - job = self.public(conn.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()) + if duplicate: + return { + "job": self.public( + conn.execute( + "SELECT * FROM jobs WHERE id=?", (duplicate["job_id"],) + ).fetchone() + ), + "coalesced": False, + "duplicate": True, + } + # Never merge a new request into a running job: its source snapshot + # may already exist. One queued follow-up absorbs further pushes. + queued = conn.execute( + "SELECT * FROM jobs WHERE status='queued' ORDER BY created_at LIMIT 1" + ).fetchone() + if queued: + job_id = queued["id"] + conn.execute( + """UPDATE jobs SET pull_git=MAX(pull_git,?), updated_at=?, + requested_revision=COALESCE(?,requested_revision), request_count=request_count+1 + WHERE id=?""", + (int(pull_git), now, requested_revision, job_id), + ) + else: + job_id = uuid.uuid4().hex + conn.execute( + """INSERT INTO jobs(id,status,phase,pull_git,created_at,updated_at,requested_revision) + VALUES(?,'queued','queued',?,?,?,?)""", + (job_id, int(pull_git), now, now, requested_revision), + ) + if delivery_key: + conn.execute("INSERT INTO deliveries VALUES(?,?)", (delivery_key, job_id)) + job = self.public(conn.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()) return {"job": job, "coalesced": queued is not None, "duplicate": False} + def schedule_if_due(self) -> bool: + if not self.periodic_interval_seconds: + return False + now = time.time() + with closing(self.connect()) as conn, conn: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute("SELECT next_due FROM schedule WHERE id=1").fetchone() + if row and row["next_due"] > now: + return False + self._enqueue(conn, now, True) + conn.execute("INSERT OR REPLACE INTO schedule VALUES(1,?)", + (now + self.periodic_interval_seconds,)) + return True + def get(self, job_id: str): with closing(self.connect()) as conn: return self.public(conn.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()) @@ -125,7 +146,11 @@ class SyncJobs: "SELECT * FROM jobs WHERE status='failed' ORDER BY finished_at DESC LIMIT 1" ).fetchone() ) - return {"active": active, "last_success": success, "last_failure": failure} + schedule = conn.execute("SELECT next_due FROM schedule WHERE id=1").fetchone() + return {"active": active, "last_success": success, "last_failure": failure, + "schedule": {"enabled": bool(self.periodic_interval_seconds), + "interval_seconds": self.periodic_interval_seconds, + "next_due": schedule["next_due"] if schedule else None}} def phase(self, job_id: str, phase: str): with closing(self.connect()) as conn, conn: @@ -208,6 +233,7 @@ class SyncJobs: def _loop(self): while not self.stopping.is_set(): try: + self.schedule_if_due() worked = self.run_once() except Exception: logger.exception("Chyba pracovníka synchronizačnej fronty.") diff --git a/tools/sync/wiki_git.py b/tools/sync/wiki_git.py index 59b70be..6f49245 100644 --- a/tools/sync/wiki_git.py +++ b/tools/sync/wiki_git.py @@ -82,6 +82,16 @@ def update_wiki(repo_path: Path) -> None: raise RuntimeError("Aktívna Git vetva nezodpovedá EXPECTED_GITEA_BRANCH") if git("status", "--porcelain", "--untracked-files=normal"): raise RuntimeError("Wiki obsahuje lokálne zmeny; automatická aktualizácia ich neprepíše") + if os.getenv("SYNC_UPSTREAM_FORK", "0") == "1": + if expected != "jp170na/zpwiki" or branch != "master": + raise RuntimeError("Upstream fork synchronization is restricted to jp170na/zpwiki master") + key_file = os.getenv("ZPWIKI_SYNC_KEY_FILE", "/run/secrets/zpwiki_ssh_key") + script = Path(__file__).resolve().parents[1] / "git_runner/sync_fork.sh" + result = subprocess.run(["bash", str(script), "--key-file", key_file], + env=environment, capture_output=True, text=True, timeout=GIT_TIMEOUT_SECONDS, + check=False) + if result.returncode: + raise RuntimeError("Fork synchronization failed; local indexes were not changed") git( "fetch", "--no-tags",