feat(wiki): persist configurable multimodal knowledge bundles

codenib/web/config.py

@@ -147,6 +147,15 @@ class QAConfig:
     wiki_media_api_base: Optional[str] = None
     wiki_media_api_key: Optional[str] = field(default=None, repr=False)
     wiki_media_options: Dict[str, Any] = field(default_factory=dict)
+    # Optional OpenAI-compatible VLM endpoint for extracting structured visual
+    # facts from repository-owned images/diagrams before grounding them to code.
+    # Disabled by default so local/offline builds keep using deterministic
+    # metadata extraction.
+    wiki_visual_facts_enabled: bool = False
+    wiki_visual_facts_model: Optional[str] = None
+    wiki_visual_facts_api_base: Optional[str] = None
+    wiki_visual_facts_api_key: Optional[str] = field(default=None, repr=False)
+    wiki_visual_facts_options: Dict[str, Any] = field(default_factory=dict)
     # Optional OpenAI-compatible endpoint for the Ask agent. Provider-native
     # models (for example Vertex or Anthropic) normally leave these unset.
     model_api_base: Optional[str] = None
@@ -224,6 +233,10 @@ def __post_init__(self) -> None:
             self.wiki_media_options,
             source="wiki_media_options",
         )
+        self.wiki_visual_facts_options = validate_model_options(
+            self.wiki_visual_facts_options,
+            source="wiki_visual_facts_options",
+        )
 
     def index_types(self) -> List[str]:
         return ["bm25", "vector"] if self.mode == "hybrid" else ["bm25"]
@@ -297,6 +310,16 @@ def wiki_media_generation_enabled(self) -> bool:
             return True
         return bool(self.wiki_media_model and self.wiki_media_api_base)
 
+    @property
+    def wiki_visual_fact_extraction_enabled(self) -> bool:
+        """Whether repository media should be sent to a configured VLM."""
+
+        return bool(
+            self.wiki_visual_facts_enabled
+            and self.wiki_visual_facts_model
+            and self.wiki_visual_facts_api_base
+        )
+
 
 def load_config(path: Optional[str] = None) -> QAConfig:
     """Load a layered demo config from YAML, then apply env overrides."""
@@ -316,6 +339,19 @@ def load_config(path: Optional[str] = None) -> QAConfig:
             data.get("wiki_media_options"),
             source="wiki_media_options",
         ),
+        wiki_visual_facts_enabled=bool(
+            data.get(
+                "wiki_visual_facts_enabled",
+                defaults.wiki_visual_facts_enabled,
+            )
+        ),
+        wiki_visual_facts_model=data.get("wiki_visual_facts_model"),
+        wiki_visual_facts_api_base=data.get("wiki_visual_facts_api_base"),
+        wiki_visual_facts_api_key=data.get("wiki_visual_facts_api_key"),
+        wiki_visual_facts_options=validate_model_options(
+            data.get("wiki_visual_facts_options"),
+            source="wiki_visual_facts_options",
+        ),
         model_api_base=data.get("model_api_base"),
         model_api_key=data.get("model_api_key"),
         model_options=validate_model_options(
@@ -374,6 +410,18 @@ def load_config(path: Optional[str] = None) -> QAConfig:
         cfg.wiki_media_api_base = os.environ["CODENIB_WIKI_MEDIA_API_BASE"]
     if os.environ.get("CODENIB_WIKI_MEDIA_API_KEY"):
         cfg.wiki_media_api_key = os.environ["CODENIB_WIKI_MEDIA_API_KEY"]
+    if os.environ.get("CODENIB_WIKI_VISUAL_FACTS_ENABLED") is not None:
+        cfg.wiki_visual_facts_enabled = os.environ[
+            "CODENIB_WIKI_VISUAL_FACTS_ENABLED"
+        ].strip().lower() in ("1", "true", "yes", "on")
+    if os.environ.get("CODENIB_WIKI_VISUAL_FACTS_MODEL"):
+        cfg.wiki_visual_facts_model = os.environ["CODENIB_WIKI_VISUAL_FACTS_MODEL"]
+    if os.environ.get("CODENIB_WIKI_VISUAL_FACTS_API_BASE"):
+        cfg.wiki_visual_facts_api_base = os.environ[
+            "CODENIB_WIKI_VISUAL_FACTS_API_BASE"
+        ]
+    if os.environ.get("CODENIB_WIKI_VISUAL_FACTS_API_KEY"):
+        cfg.wiki_visual_facts_api_key = os.environ["CODENIB_WIKI_VISUAL_FACTS_API_KEY"]
     if os.environ.get("CODENIB_DEMO_API_BASE"):
         cfg.model_api_base = os.environ["CODENIB_DEMO_API_BASE"]
     if os.environ.get("CODENIB_DEMO_API_KEY"):
@@ -402,6 +450,14 @@ def load_config(path: Optional[str] = None) -> QAConfig:
                 source="CODENIB_WIKI_MEDIA_OPTIONS",
             ),
         )
+    if os.environ.get("CODENIB_WIKI_VISUAL_FACTS_OPTIONS"):
+        cfg.wiki_visual_facts_options = merge_model_options(
+            cfg.wiki_visual_facts_options,
+            parse_model_options_json(
+                os.environ["CODENIB_WIKI_VISUAL_FACTS_OPTIONS"],
+                source="CODENIB_WIKI_VISUAL_FACTS_OPTIONS",
+            ),
+        )
     if os.environ.get("CODENIB_DEMO_DATA_DIR"):
         cfg.data_dir = os.environ["CODENIB_DEMO_DATA_DIR"]
     if os.environ.get("CODENIB_DEMO_PREBUILT_DIR"):

codenib/wiki/__init__.py

@@ -35,16 +35,30 @@
     search_visual_context,
 )
 from .media_pipeline import build_multimodal_repository_knowledge
+from .media_storage import (
+    MULTIMODAL_KNOWLEDGE_BUNDLE_SCHEMA,
+    MULTIMODAL_KNOWLEDGE_BUNDLE_VERSION,
+    build_multimodal_knowledge_bundle,
+    load_multimodal_knowledge_bundle,
+    save_multimodal_knowledge_bundle,
+    validate_multimodal_knowledge_bundle,
+)
 from .media_tools import MULTIMODAL_TOOL_SCHEMAS, MultimodalKnowledgeToolRouter
-from .media_vlm import OpenAICompatibleVisualFactExtractor
+from .media_vlm import (
+    OpenAICompatibleVisualFactExtractor,
+    visual_fact_extractor_from_config,
+)
 
 __all__ = [
     "MULTIMODAL_TOOL_SCHEMAS",
+    "MULTIMODAL_KNOWLEDGE_BUNDLE_SCHEMA",
+    "MULTIMODAL_KNOWLEDGE_BUNDLE_VERSION",
     "MultimodalKnowledgeToolRouter",
     "WikiBuilder",
     "OpenAICompatibleVisualFactExtractor",
     "build_media_evidence_pack",
     "build_multimodal_knowledge_view",
+    "build_multimodal_knowledge_bundle",
     "build_multimodal_repository_knowledge",
     "build_visual_facts_manifest",
     "deterministic_visual_facts",
@@ -59,5 +73,9 @@
     "ground_visual_facts_to_sources",
     "merge_incremental_visual_facts",
     "plan_incremental_visual_fact_update",
+    "load_multimodal_knowledge_bundle",
+    "save_multimodal_knowledge_bundle",
     "search_visual_context",
+    "validate_multimodal_knowledge_bundle",
+    "visual_fact_extractor_from_config",
 ]

codenib/wiki/media_pipeline.py

@@ -20,6 +20,7 @@
     ground_visual_facts_to_sources,
 )
 from .media_knowledge import build_multimodal_knowledge_view
+from .media_storage import build_multimodal_knowledge_bundle
 
 
 def build_multimodal_repository_knowledge(
@@ -60,13 +61,13 @@ def build_multimodal_repository_knowledge(
         visual_facts_manifest,
         grounding_manifest,
     )
-    return {
-        "media_manifest": media_manifest,
-        "visual_facts_manifest": visual_facts_manifest,
-        "source_candidate_count": len(source_candidates),
-        "grounding_manifest": grounding_manifest,
-        "knowledge_view": knowledge_view,
-    }
+    return build_multimodal_knowledge_bundle(
+        media_manifest=media_manifest,
+        visual_facts_manifest=visual_facts_manifest,
+        source_candidate_count=len(source_candidates),
+        grounding_manifest=grounding_manifest,
+        knowledge_view=knowledge_view,
+    )
 
 
 __all__ = ["build_multimodal_repository_knowledge"]

codenib/wiki/media_storage.py

@@ -0,0 +1,159 @@
+# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Stable storage helpers for multimodal repository knowledge bundles."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from hashlib import sha256
+from pathlib import Path
+from typing import Any, Mapping
+
+MULTIMODAL_KNOWLEDGE_BUNDLE_SCHEMA = "codenib.multimodal-knowledge-bundle.v1"
+MULTIMODAL_KNOWLEDGE_BUNDLE_VERSION = 1
+_MAX_BUNDLE_BYTES = 128 * 1024 * 1024
+
+
+def build_multimodal_knowledge_bundle(
+    *,
+    media_manifest: Mapping[str, Any],
+    visual_facts_manifest: Mapping[str, Any],
+    source_candidate_count: int,
+    grounding_manifest: Mapping[str, Any],
+    knowledge_view: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Wrap multimodal pipeline outputs in a versioned, hashable bundle."""
+
+    bundle: dict[str, Any] = {
+        "schema": MULTIMODAL_KNOWLEDGE_BUNDLE_SCHEMA,
+        "schema_version": MULTIMODAL_KNOWLEDGE_BUNDLE_VERSION,
+        "media_manifest": dict(media_manifest),
+        "visual_facts_manifest": dict(visual_facts_manifest),
+        "source_candidate_count": int(source_candidate_count),
+        "grounding_manifest": dict(grounding_manifest),
+        "knowledge_view": dict(knowledge_view),
+        "component_sha256": {
+            "media_manifest": str(media_manifest.get("manifest_sha256") or ""),
+            "visual_facts_manifest": str(
+                visual_facts_manifest.get("manifest_sha256") or ""
+            ),
+            "grounding_manifest": str(grounding_manifest.get("manifest_sha256") or ""),
+            "knowledge_view": str(knowledge_view.get("view_sha256") or ""),
+        },
+    }
+    bundle["bundle_sha256"] = _stable_sha256(
+        {key: value for key, value in bundle.items() if key != "bundle_sha256"}
+    )
+    return bundle
+
+
+def save_multimodal_knowledge_bundle(
+    bundle: Mapping[str, Any],
+    path: str | Path,
+) -> None:
+    """Atomically write a multimodal knowledge bundle as stable JSON."""
+
+    validated = validate_multimodal_knowledge_bundle(bundle)
+    destination = Path(path)
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    fd, temp_name = tempfile.mkstemp(
+        prefix=f".{destination.name}.",
+        suffix=".tmp",
+        dir=str(destination.parent),
+        text=True,
+    )
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(validated, handle, ensure_ascii=False, indent=2, sort_keys=True)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temp_name, destination)
+    finally:
+        try:
+            os.unlink(temp_name)
+        except FileNotFoundError:
+            pass
+
+
+def load_multimodal_knowledge_bundle(path: str | Path) -> dict[str, Any]:
+    """Load and validate a persisted multimodal knowledge bundle."""
+
+    source = Path(path)
+    size = source.stat().st_size
+    if size > _MAX_BUNDLE_BYTES:
+        raise ValueError("multimodal knowledge bundle exceeds the byte limit")
+    with source.open("rb") as handle:
+        raw = handle.read(_MAX_BUNDLE_BYTES + 1)
+    if len(raw) > _MAX_BUNDLE_BYTES:
+        raise ValueError("multimodal knowledge bundle exceeds the byte limit")
+    data = json.loads(raw.decode("utf-8"))
+    return validate_multimodal_knowledge_bundle(data)
+
+
+def validate_multimodal_knowledge_bundle(bundle: Mapping[str, Any]) -> dict[str, Any]:
+    """Return a normalized bundle or raise ``ValueError`` for invalid input."""
+
+    if not isinstance(bundle, Mapping):
+        raise ValueError("multimodal knowledge bundle must be an object")
+    data = dict(bundle)
+    if data.get("schema") != MULTIMODAL_KNOWLEDGE_BUNDLE_SCHEMA:
+        raise ValueError("multimodal knowledge bundle schema is unsupported")
+    if data.get("schema_version") != MULTIMODAL_KNOWLEDGE_BUNDLE_VERSION:
+        raise ValueError("multimodal knowledge bundle version is unsupported")
+    for key in (
+        "media_manifest",
+        "visual_facts_manifest",
+        "grounding_manifest",
+        "knowledge_view",
+        "component_sha256",
+    ):
+        if not isinstance(data.get(key), Mapping):
+            raise ValueError(f"multimodal knowledge bundle field {key!r} is invalid")
+        data[key] = dict(data[key])
+    source_candidate_count = data.get("source_candidate_count")
+    if isinstance(source_candidate_count, bool) or not isinstance(
+        source_candidate_count, int
+    ):
+        raise ValueError(
+            "multimodal knowledge bundle source_candidate_count is invalid"
+        )
+    if source_candidate_count < 0:
+        raise ValueError(
+            "multimodal knowledge bundle source_candidate_count is invalid"
+        )
+    expected_hash = _stable_sha256(
+        {key: value for key, value in data.items() if key != "bundle_sha256"}
+    )
+    recorded_hash = data.get("bundle_sha256")
+    if recorded_hash:
+        if recorded_hash != expected_hash:
+            raise ValueError("multimodal knowledge bundle hash does not match")
+    else:
+        data["bundle_sha256"] = expected_hash
+    return data
+
+
+def _stable_sha256(payload: Mapping[str, Any]) -> str:
+    return sha256(
+        json.dumps(
+            payload,
+            ensure_ascii=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode("utf-8")
+    ).hexdigest()
+
+
+__all__ = [
+    "MULTIMODAL_KNOWLEDGE_BUNDLE_SCHEMA",
+    "MULTIMODAL_KNOWLEDGE_BUNDLE_VERSION",
+    "build_multimodal_knowledge_bundle",
+    "load_multimodal_knowledge_bundle",
+    "save_multimodal_knowledge_bundle",
+    "validate_multimodal_knowledge_bundle",
+]

codenib/wiki/media_vlm.py

@@ -135,6 +135,27 @@ def _post_json(self, payload: Mapping[str, Any]) -> dict[str, Any]:
         return data
 
 
+def visual_fact_extractor_from_config(
+    config: Any,
+) -> OpenAICompatibleVisualFactExtractor | None:
+    """Build a visual-fact extractor from ``QAConfig``-shaped settings."""
+
+    if not bool(getattr(config, "wiki_visual_fact_extraction_enabled", False)):
+        return None
+    model = str(getattr(config, "wiki_visual_facts_model", None) or "").strip()
+    api_base = str(getattr(config, "wiki_visual_facts_api_base", None) or "").strip()
+    options = dict(getattr(config, "wiki_visual_facts_options", {}) or {})
+    timeout = options.get("timeout", 120.0)
+    provider = str(options.get("provider") or "openai-compatible")
+    return OpenAICompatibleVisualFactExtractor(
+        model=model,
+        api_base=api_base,
+        api_key=getattr(config, "wiki_visual_facts_api_key", None),
+        timeout=timeout,
+        provider=provider,
+    )
+
+
 def _response_content_json(response: Mapping[str, Any]) -> dict[str, Any]:
     choices = response.get("choices")
     if not isinstance(choices, list) or not choices:
@@ -240,4 +261,4 @@ def _chat_completions_endpoint(api_base: str) -> str:
     return parsed._replace(path=path, fragment="").geturl()
 
 
-__all__ = ["OpenAICompatibleVisualFactExtractor"]
+__all__ = ["OpenAICompatibleVisualFactExtractor", "visual_fact_extractor_from_config"]