917 lines
36 KiB
Python
917 lines
36 KiB
Python
"""Best-effort periodic full scans for the DeerFlow-local Wiki index."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Coroutine
|
|
from datetime import UTC, datetime
|
|
from time import monotonic
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
_CONFIG_REFRESH_SECONDS = 10.0
|
|
_MUTATION_SYNC_DELAYS_SECONDS = (1.0, 8.0, 30.0, 90.0, 300.0)
|
|
_MUTATION_SYNC_MAX_SECONDS = 10 * 60.0
|
|
_MUTATION_SYNC_CONFIRM_GRACE_SECONDS = 30.0
|
|
_FULL_SCAN_FALLBACK_SECONDS = 10 * 60.0
|
|
_WIKI_CONFIG_UPGRADE_RETRY_SECONDS = 5 * 60.0
|
|
|
|
|
|
def _has_focus_constraint(mapping: dict[str, Any]) -> bool:
|
|
"""Return whether this mapping needs strict focus reconciliation.
|
|
|
|
Both current ``wiki_directory_paths`` and legacy
|
|
``wiki_parent_directories`` values are prompt-only references. They never
|
|
trigger post-generation folder mutations.
|
|
"""
|
|
|
|
return bool(
|
|
str(mapping.get("wiki_focus_instructions") or "").strip()
|
|
)
|
|
|
|
|
|
def _wiki_config_schema_version(mapping: dict[str, Any]) -> int:
|
|
try:
|
|
return int(mapping.get("wiki_config_schema_version") or 0)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
async def _upgrade_wiki_generation_configs_once(
|
|
app: FastAPI,
|
|
*,
|
|
client: Any | None = None,
|
|
) -> dict[str, int]:
|
|
"""Replace legacy effective prompts with the current reference semantics.
|
|
|
|
Raw user inputs in the DeerFlow database are authoritative. A failed
|
|
provider write leaves the old schema version in place and records an error,
|
|
so a later scheduler pass retries safely. Each mapping is isolated: one bad
|
|
knowledge base never blocks application startup or the remaining upgrades.
|
|
"""
|
|
|
|
from app.gateway.wiki_directory_constraints import (
|
|
normalize_directory_paths,
|
|
paths_from_parent_directories,
|
|
)
|
|
from app.gateway.wiki_generation_config import (
|
|
WIKI_CONFIG_SCHEMA_VERSION,
|
|
compile_wiki_instructions,
|
|
normalize_focus_instructions,
|
|
)
|
|
|
|
mapping_store = getattr(app.state, "llmwiki_store", None)
|
|
list_visible = getattr(mapping_store, "list_visible", None)
|
|
update_mapping = getattr(mapping_store, "update_mapping", None)
|
|
if not callable(list_visible) or not callable(update_mapping):
|
|
return {"eligible": 0, "upgraded": 0, "failed": 0}
|
|
try:
|
|
rows = await list_visible("system", scope="all", is_admin=True)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.warning("Could not list Wiki mappings for prompt-schema upgrade", exc_info=True)
|
|
return {"eligible": 0, "upgraded": 0, "failed": 0}
|
|
|
|
targets = [
|
|
row
|
|
for row in rows
|
|
if isinstance(row, dict)
|
|
and row.get("wiki_generation_managed") is True
|
|
# Schema 2 is the first version that guarantees complete raw prompt
|
|
# fields. Older/unknown rows may only have provider-side instructions;
|
|
# recompiling those from missing raw values could erase user content.
|
|
and _wiki_config_schema_version(row) == 2
|
|
]
|
|
if not targets:
|
|
return {"eligible": 0, "upgraded": 0, "failed": 0}
|
|
|
|
remote_client = client
|
|
if remote_client is None:
|
|
service = getattr(app.state, "llmwiki_sync_service", None)
|
|
remote_client = getattr(service, "client", None)
|
|
if remote_client is None:
|
|
from deerflow.integrations.weknora.runtime import (
|
|
build_weknora_client,
|
|
get_resolved_llmwiki_runtime,
|
|
)
|
|
|
|
runtime = get_resolved_llmwiki_runtime(
|
|
getattr(app.state, "config", None)
|
|
)
|
|
if not runtime.weknora_enabled:
|
|
return {"eligible": len(targets), "upgraded": 0, "failed": 0}
|
|
try:
|
|
remote_client = build_weknora_client(runtime)
|
|
except Exception:
|
|
logger.warning("Wiki prompt-schema upgrade client is unavailable", exc_info=True)
|
|
return {"eligible": len(targets), "upgraded": 0, "failed": 0}
|
|
|
|
upgraded = 0
|
|
failed = 0
|
|
semaphore = asyncio.Semaphore(4)
|
|
|
|
async def run(mapping: dict[str, Any]) -> bool:
|
|
mapping_id = str(mapping.get("id") or "")
|
|
try:
|
|
remote_id = str(mapping.get("weknora_id") or "").strip()
|
|
if not mapping_id or not remote_id:
|
|
raise ValueError("Wiki mapping or WeKnora knowledge base ID is missing")
|
|
raw_paths = mapping.get("wiki_directory_paths")
|
|
directory_paths = (
|
|
normalize_directory_paths(raw_paths)
|
|
if raw_paths is not None
|
|
else paths_from_parent_directories(
|
|
mapping.get("wiki_parent_directories")
|
|
)
|
|
)
|
|
compiled = compile_wiki_instructions(
|
|
content_instructions=str(
|
|
mapping.get("wiki_content_instructions_raw") or ""
|
|
),
|
|
extraction_instructions=str(
|
|
mapping.get("wiki_extraction_instructions_raw") or ""
|
|
),
|
|
focus_instructions=normalize_focus_instructions(
|
|
str(mapping.get("wiki_focus_instructions") or "")
|
|
),
|
|
directory_paths=directory_paths,
|
|
)
|
|
async with semaphore:
|
|
await remote_client.update_knowledge_base(
|
|
remote_id,
|
|
{
|
|
"wiki_config": {
|
|
"content_instructions": compiled.content_instructions,
|
|
"extraction_instructions": compiled.extraction_instructions,
|
|
}
|
|
},
|
|
)
|
|
await update_mapping(
|
|
mapping_id,
|
|
wiki_config_schema_version=WIKI_CONFIG_SCHEMA_VERSION,
|
|
wiki_config_revision=int(
|
|
mapping.get("wiki_config_revision") or 0
|
|
)
|
|
+ 1,
|
|
wiki_config_apply_status="active",
|
|
wiki_config_apply_error=None,
|
|
)
|
|
return True
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Wiki prompt-schema upgrade failed mapping_id=%s",
|
|
mapping_id,
|
|
exc_info=True,
|
|
)
|
|
if mapping_id:
|
|
try:
|
|
await update_mapping(
|
|
mapping_id,
|
|
wiki_config_apply_status="error",
|
|
wiki_config_apply_error=str(exc)[:4000],
|
|
)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.warning(
|
|
"Could not record Wiki prompt-schema upgrade failure mapping_id=%s",
|
|
mapping_id,
|
|
exc_info=True,
|
|
)
|
|
return False
|
|
|
|
results = await asyncio.gather(*(run(mapping) for mapping in targets))
|
|
upgraded = sum(result is True for result in results)
|
|
failed = sum(result is False for result in results)
|
|
if upgraded or failed:
|
|
logger.info(
|
|
"Wiki prompt-schema upgrade finished eligible=%s upgraded=%s failed=%s",
|
|
len(targets),
|
|
upgraded,
|
|
failed,
|
|
)
|
|
return {"eligible": len(targets), "upgraded": upgraded, "failed": failed}
|
|
|
|
|
|
def _has_directory_constraints(mapping: dict[str, Any]) -> bool:
|
|
return bool(
|
|
mapping.get("wiki_directory_paths")
|
|
or mapping.get("wiki_parent_directories")
|
|
)
|
|
|
|
|
|
def _knowledge_base_is_processing(remote: dict[str, Any]) -> bool:
|
|
"""Normalize processing markers across old/new WeKnora response shapes."""
|
|
|
|
raw_flag = remote.get("is_processing", remote.get("isProcessing"))
|
|
if isinstance(raw_flag, str):
|
|
is_processing = raw_flag.strip().lower() in {"1", "true", "yes", "on"}
|
|
else:
|
|
is_processing = bool(raw_flag)
|
|
raw_count = remote.get("processing_count", remote.get("processingCount"))
|
|
try:
|
|
processing_count = int(raw_count or 0)
|
|
except (TypeError, ValueError):
|
|
processing_count = 0
|
|
return is_processing or processing_count > 0
|
|
|
|
|
|
def _document_is_processing(document: dict[str, Any]) -> bool:
|
|
for key in (
|
|
"pending_subtasks_count",
|
|
"pendingSubtasksCount",
|
|
"pending_subtasks",
|
|
):
|
|
try:
|
|
if int(document.get(key) or 0) > 0:
|
|
return True
|
|
except (TypeError, ValueError):
|
|
continue
|
|
active = {
|
|
"pending",
|
|
"queued",
|
|
"processing",
|
|
"running",
|
|
"finalizing",
|
|
"summarizing",
|
|
"extracting",
|
|
"generating",
|
|
"in_progress",
|
|
"in-progress",
|
|
}
|
|
return any(
|
|
str(document.get(key) or "").strip().casefold() in active
|
|
for key in (
|
|
"processing_status",
|
|
"processingStatus",
|
|
"parse_status",
|
|
"parseStatus",
|
|
"summary_status",
|
|
"summaryStatus",
|
|
"wiki_status",
|
|
"wikiStatus",
|
|
)
|
|
)
|
|
|
|
|
|
async def _remote_generation_is_processing(
|
|
client: Any,
|
|
mapping: dict[str, Any],
|
|
*,
|
|
remote: dict[str, Any] | None = None,
|
|
) -> bool | None:
|
|
"""Check document Wiki subtasks as well as the coarse KB parse flag.
|
|
|
|
WeKnora can clear ``knowledge_base.is_processing`` while Wiki ingest still
|
|
has pages in draft/finalizing. The per-document pending-subtask count is
|
|
the reliable completion signal for generation constraints.
|
|
"""
|
|
|
|
remote_id = str(mapping.get("weknora_id") or "")
|
|
if not remote_id:
|
|
return None
|
|
if remote is None:
|
|
get_remote = getattr(client, "get_knowledge_base", None)
|
|
if callable(get_remote):
|
|
try:
|
|
remote = await get_remote(remote_id)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
return None
|
|
if isinstance(remote, dict) and _knowledge_base_is_processing(remote):
|
|
return True
|
|
|
|
list_documents = getattr(client, "list_documents", None)
|
|
if not callable(list_documents):
|
|
return False if isinstance(remote, dict) else None
|
|
page = 1
|
|
page_size = 100
|
|
try:
|
|
while True:
|
|
payload = await list_documents(
|
|
remote_id,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
rows = payload.get("items")
|
|
if not isinstance(rows, list):
|
|
return None
|
|
if any(
|
|
_document_is_processing(row)
|
|
for row in rows
|
|
if isinstance(row, dict)
|
|
):
|
|
return True
|
|
total = int(payload.get("total") or len(rows))
|
|
if page * page_size >= total or len(rows) < page_size:
|
|
return False
|
|
page += 1
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def spawn_local_wiki_index_job(app: FastAPI, coroutine: Coroutine[Any, Any, Any]) -> asyncio.Task:
|
|
"""Keep fire-and-forget index jobs strongly referenced until completion."""
|
|
|
|
jobs: set[asyncio.Task] = getattr(app.state, "llmwiki_index_jobs", None)
|
|
if jobs is None:
|
|
jobs = set()
|
|
app.state.llmwiki_index_jobs = jobs
|
|
task = asyncio.create_task(coroutine)
|
|
jobs.add(task)
|
|
task.add_done_callback(jobs.discard)
|
|
return task
|
|
|
|
|
|
def schedule_local_wiki_mutation_sync(
|
|
app: FastAPI,
|
|
mapping: dict[str, Any],
|
|
*,
|
|
delays: tuple[float, ...] | None = None,
|
|
client: Any | None = None,
|
|
sync_local: bool = True,
|
|
) -> asyncio.Task | None:
|
|
"""Debounce one mapping's mutation until WeKnora finishes Wiki work.
|
|
|
|
Document ingestion is asynchronous upstream, so a single immediate scan can
|
|
legitimately observe the old Wiki snapshot. One coordinator per mapping
|
|
performs a few inexpensive metadata-first scans. A newer write only bumps
|
|
the generation and resets that coordinator; it never cancels an active
|
|
sync. Database leases remain the cross-process source of truth.
|
|
"""
|
|
|
|
service = getattr(app.state, "llmwiki_sync_service", None)
|
|
constrained = _has_focus_constraint(mapping)
|
|
local_sync_enabled = bool(sync_local and service is not None)
|
|
mapping_id = str(mapping.get("id") or "")
|
|
if not mapping_id:
|
|
return None
|
|
retry_delays = tuple(max(0.0, float(value)) for value in (delays or _MUTATION_SYNC_DELAYS_SECONDS))
|
|
if not retry_delays:
|
|
return None
|
|
|
|
plans: dict[str, dict[str, Any]] = getattr(app.state, "llmwiki_mutation_sync_plans", None)
|
|
if plans is None:
|
|
plans = {}
|
|
app.state.llmwiki_mutation_sync_plans = plans
|
|
tasks: dict[str, asyncio.Task] = getattr(app.state, "llmwiki_mutation_sync_tasks", None)
|
|
if tasks is None:
|
|
tasks = {}
|
|
app.state.llmwiki_mutation_sync_tasks = tasks
|
|
|
|
prior = plans.get(mapping_id) or {}
|
|
remote_client = (
|
|
client
|
|
or getattr(service, "client", None)
|
|
or prior.get("client")
|
|
)
|
|
running = tasks.get(mapping_id)
|
|
# Clearing a focus must supersede an already-running settle plan even when
|
|
# the optional local index is disabled. With no prior plan there is no work
|
|
# to schedule.
|
|
if not local_sync_enabled and not constrained:
|
|
if running is None or running.done() or not prior:
|
|
return None
|
|
wakeup = prior.get("wakeup")
|
|
if not isinstance(wakeup, asyncio.Event):
|
|
wakeup = asyncio.Event()
|
|
plans[mapping_id] = {
|
|
"generation": int(prior.get("generation") or 0) + 1,
|
|
"mapping": dict(mapping),
|
|
"wakeup": wakeup,
|
|
"client": remote_client,
|
|
"sync_local": False,
|
|
}
|
|
wakeup.set()
|
|
return running
|
|
if constrained and remote_client is None:
|
|
return None
|
|
wakeup = prior.get("wakeup")
|
|
if not isinstance(wakeup, asyncio.Event):
|
|
wakeup = asyncio.Event()
|
|
plans[mapping_id] = {
|
|
"generation": int(prior.get("generation") or 0) + 1,
|
|
"mapping": dict(mapping),
|
|
"wakeup": wakeup,
|
|
"client": remote_client,
|
|
"sync_local": local_sync_enabled,
|
|
}
|
|
if running is not None and not running.done():
|
|
# Interrupt a long backoff, but never cancel an in-flight sync. The
|
|
# coordinator will observe the newer generation at its next await
|
|
# boundary and restart the bounded settle sequence from one second.
|
|
wakeup.set()
|
|
return running
|
|
|
|
async def _remote_processing(
|
|
current_mapping: dict[str, Any],
|
|
current_client: Any,
|
|
) -> bool | None:
|
|
"""Read KB and per-document generation status.
|
|
|
|
``None`` deliberately means "unknown": a transient status-read error
|
|
must not be mistaken for a completed Wiki generation.
|
|
"""
|
|
|
|
status = await _remote_generation_is_processing(
|
|
current_client,
|
|
current_mapping,
|
|
)
|
|
if status is None:
|
|
logger.debug(
|
|
"Could not read WeKnora generation status mapping_id=%s",
|
|
mapping_id,
|
|
)
|
|
return status
|
|
|
|
def _sync_changed(result: Any) -> bool:
|
|
if not isinstance(result, dict):
|
|
return False
|
|
if str(result.get("status") or "").lower() not in {"completed", "idle"}:
|
|
return False
|
|
return any(
|
|
int(result.get(key) or 0) > 0
|
|
for key in (
|
|
"added_pages",
|
|
"updated_pages",
|
|
"deleted_pages",
|
|
)
|
|
)
|
|
|
|
async def _run() -> None:
|
|
observed_generation = -1
|
|
delay_index = 0
|
|
quiet_scans = 0
|
|
observed_processing = False
|
|
post_processing_confirmations = 0
|
|
deadline = monotonic() + _MUTATION_SYNC_MAX_SECONDS
|
|
try:
|
|
while True:
|
|
plan = plans.get(mapping_id)
|
|
if plan is None:
|
|
return
|
|
generation = int(plan["generation"])
|
|
if generation != observed_generation:
|
|
observed_generation = generation
|
|
delay_index = 0
|
|
quiet_scans = 0
|
|
observed_processing = False
|
|
post_processing_confirmations = 0
|
|
deadline = monotonic() + _MUTATION_SYNC_MAX_SECONDS
|
|
if not observed_processing and quiet_scans >= len(retry_delays):
|
|
return
|
|
remaining = deadline - monotonic()
|
|
if remaining <= 0:
|
|
return
|
|
delay = min(retry_delays[min(delay_index, len(retry_delays) - 1)], remaining)
|
|
wakeup = plan.get("wakeup")
|
|
if isinstance(wakeup, asyncio.Event):
|
|
wakeup.clear()
|
|
# Re-read after clearing: a mutation scheduled from
|
|
# another callback/thread between the earlier generation
|
|
# read and ``clear()`` must not be hidden for a long
|
|
# backoff interval.
|
|
latest_after_clear = plans.get(mapping_id)
|
|
if (
|
|
latest_after_clear is None
|
|
or int(latest_after_clear["generation"])
|
|
!= observed_generation
|
|
):
|
|
continue
|
|
try:
|
|
await asyncio.wait_for(wakeup.wait(), timeout=delay)
|
|
except TimeoutError:
|
|
pass
|
|
else:
|
|
# A newer mutation resets the generation and the
|
|
# backoff on the next loop iteration.
|
|
continue
|
|
else:
|
|
await asyncio.sleep(delay)
|
|
latest = plans.get(mapping_id)
|
|
if latest is None:
|
|
return
|
|
if int(latest["generation"]) != observed_generation:
|
|
continue
|
|
current_mapping = dict(latest["mapping"])
|
|
current_client = latest.get("client")
|
|
current_sync_local = bool(latest.get("sync_local"))
|
|
delay_index = min(delay_index + 1, len(retry_delays) - 1)
|
|
|
|
# Do not repeatedly enumerate the whole Wiki while WeKnora is
|
|
# still parsing the uploaded document. Processing polls do
|
|
# not consume one of the reserved quiet-scan opportunities,
|
|
# so a final post-processing scan is always retained within
|
|
# the bounded ten-minute settle window.
|
|
processing = await _remote_processing(
|
|
current_mapping,
|
|
current_client,
|
|
)
|
|
if processing is True:
|
|
observed_processing = True
|
|
post_processing_confirmations = 0
|
|
# Upstream Wiki generation can legitimately take longer
|
|
# than the bounded quiet-snapshot window. Keep the
|
|
# coordinator alive while the provider explicitly reports
|
|
# active work, and start the timeout from the latest
|
|
# positive processing observation instead of the upload.
|
|
deadline = monotonic() + _MUTATION_SYNC_MAX_SECONDS
|
|
continue
|
|
if processing is None and _has_focus_constraint(current_mapping):
|
|
observed_processing = True
|
|
post_processing_confirmations = 0
|
|
continue
|
|
if (
|
|
observed_processing
|
|
and _has_focus_constraint(current_mapping)
|
|
and post_processing_confirmations == 0
|
|
):
|
|
# Focus reconciliation can delete out-of-scope generated
|
|
# pages. Require a second consecutive completed status
|
|
# before that destructive phase.
|
|
post_processing_confirmations = 1
|
|
delay_index = min(1, len(retry_delays) - 1)
|
|
deadline = max(
|
|
deadline,
|
|
monotonic() + _MUTATION_SYNC_CONFIRM_GRACE_SECONDS,
|
|
)
|
|
continue
|
|
try:
|
|
if _has_focus_constraint(current_mapping):
|
|
from app.gateway.wiki_directory_constraints import (
|
|
reconcile_fixed_parent_directories,
|
|
)
|
|
|
|
await reconcile_fixed_parent_directories(
|
|
current_client,
|
|
current_mapping,
|
|
)
|
|
if not current_sync_local or service is None:
|
|
# Strict focus filtering remains useful even when the
|
|
# optional DeerFlow-local vector mirror is disabled.
|
|
latest = plans.get(mapping_id)
|
|
if (
|
|
latest is not None
|
|
and int(latest["generation"])
|
|
!= observed_generation
|
|
):
|
|
continue
|
|
if observed_processing:
|
|
post_processing = await _remote_processing(
|
|
current_mapping,
|
|
current_client,
|
|
)
|
|
latest = plans.get(mapping_id)
|
|
if (
|
|
latest is not None
|
|
and int(latest["generation"])
|
|
!= observed_generation
|
|
):
|
|
continue
|
|
if post_processing is not False:
|
|
post_processing_confirmations = 0
|
|
continue
|
|
post_processing_confirmations += 1
|
|
if post_processing_confirmations < 2:
|
|
delay_index = min(1, len(retry_delays) - 1)
|
|
deadline = max(
|
|
deadline,
|
|
monotonic()
|
|
+ _MUTATION_SYNC_CONFIRM_GRACE_SECONDS,
|
|
)
|
|
continue
|
|
return
|
|
result = await service.sync_mapping(current_mapping, force=False)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.warning(
|
|
"Mutation-triggered local Wiki sync failed mapping_id=%s attempt=%s",
|
|
mapping_id,
|
|
quiet_scans + 1,
|
|
exc_info=True,
|
|
)
|
|
continue
|
|
|
|
# A periodic/manual worker may temporarily own the mapping.
|
|
# That is not a completed settle scan; retry after its lease
|
|
# releases instead of silently dropping this mutation.
|
|
if isinstance(result, dict) and result.get("status") == "already_running":
|
|
continue
|
|
# The sync service already owns the provider-wide exponential
|
|
# breaker. A mutation settle loop must not burn all of its
|
|
# delayed attempts while that breaker is open.
|
|
if isinstance(result, dict) and result.get("status") == "provider_backoff":
|
|
latest = plans.get(mapping_id)
|
|
if (
|
|
latest is not None
|
|
and int(latest["generation"]) != observed_generation
|
|
):
|
|
continue
|
|
return
|
|
|
|
post_processing = await _remote_processing(current_mapping, current_client)
|
|
latest = plans.get(mapping_id)
|
|
if latest is None:
|
|
return
|
|
if int(latest["generation"]) != observed_generation:
|
|
continue
|
|
if post_processing is True:
|
|
observed_processing = True
|
|
post_processing_confirmations = 0
|
|
deadline = monotonic() + _MUTATION_SYNC_MAX_SECONDS
|
|
continue
|
|
if post_processing is None and _has_focus_constraint(current_mapping):
|
|
observed_processing = True
|
|
post_processing_confirmations = 0
|
|
continue
|
|
if isinstance(result, dict) and result.get("deferred_deletion"):
|
|
# The service's destructive-phase fence observed a race
|
|
# that the cheap pre-check missed (or could not verify).
|
|
# Treat it exactly like an explicit processing window and
|
|
# retain the post-processing confirmation scans.
|
|
observed_processing = True
|
|
post_processing_confirmations = 0
|
|
delay_index = min(1, len(retry_delays) - 1)
|
|
deadline = max(
|
|
deadline,
|
|
monotonic() + _MUTATION_SYNC_CONFIRM_GRACE_SECONDS,
|
|
)
|
|
continue
|
|
|
|
quiet_scans += 1
|
|
# Once WeKnora has explicitly transitioned out of processing,
|
|
# require one short, quiet confirmation when the first scan
|
|
# sees no delta: some WeKnora builds clear ``is_processing``
|
|
# just before publishing the final Wiki directory snapshot.
|
|
# A synchronous edit or a post-processing delta can finish
|
|
# immediately.
|
|
if _sync_changed(result):
|
|
return
|
|
if observed_processing:
|
|
post_processing_confirmations += 1
|
|
if post_processing_confirmations >= 2:
|
|
return
|
|
delay_index = min(1, len(retry_delays) - 1)
|
|
deadline = max(
|
|
deadline,
|
|
monotonic() + _MUTATION_SYNC_CONFIRM_GRACE_SECONDS,
|
|
)
|
|
finally:
|
|
current = asyncio.current_task()
|
|
if tasks.get(mapping_id) is current:
|
|
tasks.pop(mapping_id, None)
|
|
plan = plans.get(mapping_id)
|
|
if plan is not None and int(plan.get("generation") or 0) == observed_generation:
|
|
plans.pop(mapping_id, None)
|
|
|
|
task = spawn_local_wiki_index_job(app, _run())
|
|
tasks[mapping_id] = task
|
|
return task
|
|
|
|
|
|
def _as_utc(value: Any) -> datetime | None:
|
|
if isinstance(value, datetime):
|
|
parsed = value
|
|
else:
|
|
raw = str(value or "").strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=UTC)
|
|
return parsed.astimezone(UTC)
|
|
|
|
|
|
def _mapping_is_due(
|
|
mapping: dict[str, Any],
|
|
state: dict[str, Any] | None,
|
|
remote: dict[str, Any] | None,
|
|
*,
|
|
force_scan: bool,
|
|
) -> bool:
|
|
# Even the low-frequency reconciliation pass must not enumerate a Wiki
|
|
# directory while WeKnora says it is transiently processing. ``force``
|
|
# bypasses only the timestamp preflight; the destructive-phase fence in
|
|
# the sync service remains the final race/data-loss guard.
|
|
if remote is not None and _knowledge_base_is_processing(remote):
|
|
return False
|
|
if force_scan or state is None or int(state.get("index_revision") or 0) <= 0:
|
|
return True
|
|
if str(state.get("state") or "") in {"failed", "partial", "rebuild_required"}:
|
|
return True
|
|
if str(state.get("state") or "") in {"running", "rebuilding"}:
|
|
return False
|
|
if remote is None:
|
|
# Preserve the local snapshot on a transient/incomplete directory read;
|
|
# the low-frequency forced scan remains the deletion/error authority.
|
|
return False
|
|
remote_updated = _as_utc(remote.get("updated_at") or remote.get("updatedAt"))
|
|
last_success = _as_utc(state.get("last_success_at"))
|
|
if remote_updated is None or last_success is None:
|
|
return True
|
|
return remote_updated > last_success
|
|
|
|
|
|
async def _refresh_local_wiki_config(app: FastAPI):
|
|
from app.gateway.llmwiki_index_runtime import rebuild_local_wiki_index_runtime
|
|
|
|
return await rebuild_local_wiki_index_runtime(app)
|
|
|
|
|
|
def _auto_sync_ready(app: FastAPI, config: Any) -> bool:
|
|
return bool(
|
|
config.enabled
|
|
and config.auto_sync
|
|
and getattr(app.state, "llmwiki_sync_service", None) is not None
|
|
and getattr(app.state, "llmwiki_store", None) is not None
|
|
)
|
|
|
|
|
|
async def _sync_once(
|
|
app: FastAPI,
|
|
*,
|
|
config: Any | None = None,
|
|
force_scan: bool | None = None,
|
|
) -> None:
|
|
config = config or await _refresh_local_wiki_config(app)
|
|
if force_scan is None:
|
|
force_scan = bool(
|
|
getattr(app.state, "llmwiki_force_full_scan_once", False)
|
|
)
|
|
service = getattr(app.state, "llmwiki_sync_service", None)
|
|
mapping_store = getattr(app.state, "llmwiki_store", None)
|
|
if not _auto_sync_ready(app, config) or service is None or mapping_store is None:
|
|
return
|
|
mappings = [
|
|
row
|
|
for row in await mapping_store.list_visible(
|
|
"system",
|
|
scope="all",
|
|
is_admin=True,
|
|
)
|
|
if row.get("wiki_index_enabled", True)
|
|
or _has_focus_constraint(row)
|
|
]
|
|
index_store = getattr(app.state, "llmwiki_index_store", None)
|
|
states: dict[str, dict[str, Any]] = {}
|
|
if index_store is not None:
|
|
states = {
|
|
str(row.get("knowledge_base_mapping_id") or ""): row
|
|
for row in await index_store.list_index_status(
|
|
[str(mapping["id"]) for mapping in mappings]
|
|
)
|
|
if isinstance(row, dict)
|
|
}
|
|
remote_by_id: dict[str, dict[str, Any]] = {}
|
|
client = getattr(service, "client", None)
|
|
list_remote = getattr(client, "list_knowledge_bases", None)
|
|
if callable(list_remote):
|
|
try:
|
|
remote_rows = await list_remote(refresh=True)
|
|
except TypeError:
|
|
remote_rows = await list_remote()
|
|
remote_by_id = {
|
|
str(row.get("id") or ""): row
|
|
for row in remote_rows
|
|
if isinstance(row, dict) and row.get("id")
|
|
}
|
|
if remote_by_id:
|
|
def should_run(mapping: dict[str, Any]) -> bool:
|
|
remote = remote_by_id.get(str(mapping.get("weknora_id") or ""))
|
|
if not mapping.get("wiki_index_enabled", True):
|
|
return bool(
|
|
force_scan
|
|
and _has_focus_constraint(mapping)
|
|
and remote is not None
|
|
and not _knowledge_base_is_processing(remote)
|
|
)
|
|
return _mapping_is_due(
|
|
mapping,
|
|
states.get(str(mapping["id"])),
|
|
remote,
|
|
force_scan=bool(force_scan),
|
|
)
|
|
|
|
mappings = [
|
|
mapping
|
|
for mapping in mappings
|
|
if should_run(mapping)
|
|
]
|
|
elif not force_scan:
|
|
mappings = [
|
|
mapping
|
|
for mapping in mappings
|
|
if mapping.get("wiki_index_enabled", True)
|
|
]
|
|
semaphore = asyncio.Semaphore(config.sync_concurrency)
|
|
|
|
async def run(mapping):
|
|
async with semaphore:
|
|
try:
|
|
if _has_focus_constraint(mapping):
|
|
processing = await _remote_generation_is_processing(
|
|
client,
|
|
mapping,
|
|
remote=remote_by_id.get(
|
|
str(mapping.get("weknora_id") or "")
|
|
),
|
|
)
|
|
if processing is not False:
|
|
return
|
|
from app.gateway.wiki_directory_constraints import (
|
|
reconcile_fixed_parent_directories,
|
|
)
|
|
|
|
await reconcile_fixed_parent_directories(client, mapping)
|
|
if mapping.get("wiki_index_enabled", True):
|
|
await service.sync_mapping(mapping)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.warning("Periodic local Wiki sync failed mapping_id=%s", mapping.get("id"), exc_info=True)
|
|
|
|
await asyncio.gather(*(run(mapping) for mapping in mappings))
|
|
|
|
|
|
async def local_wiki_index_scheduler_loop(app: FastAPI) -> None:
|
|
logger.info(
|
|
"Wiki background coordinator started; runtime config refresh interval=%s seconds",
|
|
_CONFIG_REFRESH_SECONDS,
|
|
)
|
|
last_sync_started_at: float | None = None
|
|
last_synced_runtime_signature: str | None = None
|
|
last_full_scan_started_at: float | None = None
|
|
last_prompt_schema_upgrade_started_at: float | None = None
|
|
while True:
|
|
try:
|
|
# Rebuild checks both config.yaml mtime and persisted runtime
|
|
# settings every poll, independently from the much longer sync
|
|
# cadence. Provider/URL changes therefore take effect promptly.
|
|
config = await _refresh_local_wiki_config(app)
|
|
now = monotonic()
|
|
auto_sync_ready = _auto_sync_ready(app, config)
|
|
if not auto_sync_ready:
|
|
# With automatic sync off, the scheduler must remain local-only:
|
|
# no periodic WeKnora page scan, prompt rewrite, or directory
|
|
# reconciliation. Explicit user/API calls remain responsible
|
|
# for remote refreshes.
|
|
last_sync_started_at = None
|
|
last_synced_runtime_signature = None
|
|
else:
|
|
if (
|
|
last_prompt_schema_upgrade_started_at is None
|
|
or now - last_prompt_schema_upgrade_started_at
|
|
>= _WIKI_CONFIG_UPGRADE_RETRY_SECONDS
|
|
):
|
|
last_prompt_schema_upgrade_started_at = now
|
|
await _upgrade_wiki_generation_configs_once(app)
|
|
interval = float(config.sync_interval_seconds)
|
|
runtime_signature = getattr(
|
|
app.state,
|
|
"llmwiki_runtime_signature",
|
|
None,
|
|
)
|
|
if (
|
|
last_sync_started_at is None
|
|
or runtime_signature != last_synced_runtime_signature
|
|
or now - last_sync_started_at >= interval
|
|
):
|
|
sync_started_at = now
|
|
force_scan = bool(
|
|
last_full_scan_started_at is None
|
|
or runtime_signature != last_synced_runtime_signature
|
|
or now - last_full_scan_started_at >= max(
|
|
_FULL_SCAN_FALLBACK_SECONDS,
|
|
interval,
|
|
)
|
|
)
|
|
app.state.llmwiki_force_full_scan_once = force_scan
|
|
try:
|
|
await _sync_once(app, config=config)
|
|
finally:
|
|
app.state.llmwiki_force_full_scan_once = False
|
|
last_sync_started_at = sync_started_at
|
|
last_synced_runtime_signature = runtime_signature
|
|
if force_scan:
|
|
last_full_scan_started_at = sync_started_at
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("Local Wiki auto-sync scheduler tick failed (non-fatal)")
|
|
await asyncio.sleep(_CONFIG_REFRESH_SECONDS)
|