Skip to content

codenib.wiki

Index-derived wiki generation for the DeepWiki-style demo.

Produces a per-repo page tree and source-grounded page content from the already-loaded BM25 / vector indexes — no LLM required. When an LLM is configured it can later refine the prose, but every code anchor here resolves to a real symbol span pulled from the indexes (no fabricated lines).

Modules:

Name Description
agent_wiki

Agent wiki pipeline: a DeepWiki-style, high-level conceptual wiki.
builder

Build a repo wiki (page tree + source-grounded page content) from indexes.
cache_audit

Read-only coverage and quality audit for persisted AgentWiki pages.
evidence

Grounding primitives for generated repository Wiki pages.
media_artifacts

Discovery manifest for repository-native multimodal artifacts.
media_eval

Evaluation helpers for multimodal repository knowledge views.
media_evidence

Server-side evidence packs for VLM-ready wiki media generation.
media_facts

Structured visual facts extracted from repository media artifacts.
media_generation

Provider-neutral materialization for planned wiki media slots.
media_grounding

Ground structured visual facts to repository files and symbols.
media_incremental

Incremental update planning for multimodal repository knowledge.
media_knowledge

Queryable multimodal repository knowledge view for wiki media.
media_pipeline

End-to-end construction for multimodal repository knowledge.
media_storage

Stable storage helpers for multimodal repository knowledge bundles.
media_tools

MCP-compatible query surface for multimodal repository knowledge.
media_vlm

OpenAI-compatible VLM extraction for repository media artifacts.
multimodal

Deterministic multimodal planning hooks for wiki pages.
narrator

LLM-authored narrative layer for the wiki (DeepWiki-style prose).
outline

Stage 1 of the agent wiki pipeline: a high-level conceptual outline.
prewarm

Bounded Wiki cache prewarming with machine-readable results.
quality

Deterministic quality reports for generated repository Wiki pages.
sqlite_store

SQLite WAL implementation of the narrow Wiki store contract.
store

Domain contract for persisted AgentWiki cache entries.

Classes:

Name Description
WikiBuilder

Build wiki structure + content for one loaded repo bundle.
MultimodalKnowledgeToolRouter

Small tool router that mirrors the future MCP surface.
OpenAICompatibleVisualFactExtractor

Extract structured visual facts through an OpenAI-compatible chat API.

Functions:

Name Description
discover_media_manifest

Discover repository-native visual artifacts and return a stable manifest.
evaluate_mmwiki_predictions

Evaluate visual fact extraction and visual-code grounding together.
evaluate_visual_code_grounding

Evaluate visual entity to source binding accuracy.
evaluate_visual_fact_extraction

Evaluate extracted visual entities against an MMWiki-style gold file.
build_media_evidence_pack

Build a bounded, provider-neutral evidence pack for one media slot.
build_visual_facts_manifest

Extract visual fact packs for every artifact in a media manifest.
deterministic_visual_facts

Return a conservative local fact pack from artifact metadata only.
discover_source_symbol_candidates

Return a bounded, deterministic source-symbol inventory for grounding.
ground_visual_facts_to_sources

Ground visual entities to a source inventory.
diff_media_manifests

Return a stable path/hash diff between two media manifests.
merge_incremental_visual_facts

Merge reused and newly extracted fact packs for the current media manifest.
plan_incremental_visual_fact_update

Plan reusable facts for the explicitly selected extraction policy.
build_multimodal_knowledge_view

Join media artifacts, visual facts, and source bindings into one view.
find_visual_code_links

Return visual entries that ground to a file, optionally a symbol.
get_visual_evidence

Return one visual knowledge entry by artifact path.
search_visual_context

Search multimodal entries using a deterministic lexical scorer.
build_multimodal_repository_knowledge

Build the deterministic multimodal repository knowledge bundle.
build_multimodal_knowledge_bundle

Wrap multimodal pipeline outputs in a versioned, hashable bundle.
load_multimodal_knowledge_bundle

Load and validate a persisted multimodal knowledge bundle.
save_multimodal_knowledge_bundle

Atomically write a multimodal knowledge bundle as stable JSON.
validate_multimodal_knowledge_bundle

Return a normalized bundle or raise ValueError for invalid input.
multimodal_tool_schemas

Return independent mutable copies of the stable tool schemas.
visual_fact_extractor_from_config

Build a visual-fact extractor from QAConfig-shaped settings.

WikiBuilder 

WikiBuilder(
    bundle,
    narrator: Narrator | None = None,
    *,
    source_reader: RepositorySourceReader | None = None
)

Build wiki structure + content for one loaded repo bundle.

Source code in codenib/wiki/builder.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def __init__(
    self,
    bundle,
    narrator: Optional[Narrator] = None,
    *,
    source_reader: Optional[RepositorySourceReader] = None,
) -> None:
    # bundle: codenib.web.repo_registry.RepoBundle
    self._bundle = bundle
    self._entry = bundle.entry
    self._language = bundle.entry.language
    self._narrator = narrator or Narrator(enabled=False)
    self._source_reader = (
        source_reader
        if source_reader is not None
        else getattr(bundle, "source_reader", None)
    )
    manifest = getattr(bundle, "manifest", None)
    if (
        self._source_reader is None
        and type(getattr(manifest, "source_selection", None))
        is RepositorySourceSelection
        and bool(getattr(manifest, "source_fingerprint", ""))
    ):
        raise RuntimeError(
            "manifest-selected Wiki requires an authenticated source reader"
        )
    self._symbols_cache: Optional[tuple] = None
    self._project_summary_cache: Optional[str] = None

MultimodalKnowledgeToolRouter dataclass 

MultimodalKnowledgeToolRouter(view: Mapping[str, Any])

Small tool router that mirrors the future MCP surface.

OpenAICompatibleVisualFactExtractor 

OpenAICompatibleVisualFactExtractor(
    *,
    model: str,
    api_base: str,
    api_key: str | None = None,
    timeout: float = 120.0,
    urlopen: Callable[..., Any] | None = None,
    provider: str = "openai-compatible",
    repo_path: str | Path | None = None
)

Extract structured visual facts through an OpenAI-compatible chat API.

Methods:

Name Description
extract

Extract one canonical visual fact pack for artifact.
Source code in codenib/wiki/media_vlm.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    *,
    model: str,
    api_base: str,
    api_key: str | None = None,
    timeout: float = 120.0,
    urlopen: Callable[..., Any] | None = None,
    provider: str = "openai-compatible",
    repo_path: str | Path | None = None,
) -> None:
    self.model = str(model or "").strip()
    self.api_base = str(api_base or "").strip()
    self.api_key = _validated_api_key(api_key)
    self.timeout = _validated_timeout(timeout)
    self.provider = str(provider or "").strip()
    self.repo_path = (
        Path(repo_path).expanduser().resolve() if repo_path is not None else None
    )
    self._urlopen = urlopen or urllib.request.urlopen
    if not self.model:
        raise ValueError("visual fact model is required")
    if len(self.model) > _MAX_MODEL_LENGTH:
        raise ValueError("visual fact model is too long")
    if not self.provider or len(self.provider) > _MAX_PROVIDER_LENGTH:
        raise ValueError("visual fact provider is invalid")
    self._endpoint = _chat_completions_endpoint(self.api_base)

extract 

extract(
    artifact: Mapping[str, Any], *, repo_path: str | Path | None = None
) -> dict[str, Any]

Extract one canonical visual fact pack for artifact.

Source code in codenib/wiki/media_vlm.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def extract(
    self,
    artifact: Mapping[str, Any],
    *,
    repo_path: str | Path | None = None,
) -> dict[str, Any]:
    """Extract one canonical visual fact pack for *artifact*."""

    prompt = build_visual_fact_extraction_prompt(artifact)
    content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
    source_repo = repo_path if repo_path is not None else self.repo_path
    if source_repo is not None:
        content.append(
            {
                "type": "image_url",
                "image_url": {
                    "url": _artifact_data_url(source_repo, artifact),
                },
            }
        )
    payload = {
        "model": self.model,
        "messages": [
            {
                "role": "system",
                "content": (
                    "You extract structured repository visual facts. "
                    "Return JSON only."
                ),
            },
            {"role": "user", "content": content},
        ],
        "temperature": 0,
        "response_format": {"type": "json_object"},
    }
    response = self._post_json(payload)
    extracted = _response_content_json(response)
    extracted["extractor"] = self.provider
    extracted["metadata"] = {"model": self.model, "provider": self.provider}
    return normalize_visual_fact_pack(extracted, artifact=artifact)

discover_media_manifest 

discover_media_manifest(
    repo_path: str | Path,
    *,
    commit: str | None = None,
    exclude_roots: Iterable[str | Path] = (),
    selection: RepositorySourceSelection = DEFAULT_REPOSITORY_SOURCE_SELECTION,
    max_artifacts: int = _MAX_MEDIA_ARTIFACTS
) -> dict[str, Any]

Discover repository-native visual artifacts and return a stable manifest.

Source code in codenib/wiki/media_artifacts.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def discover_media_manifest(
    repo_path: str | Path,
    *,
    commit: str | None = None,
    exclude_roots: Iterable[str | Path] = (),
    selection: RepositorySourceSelection = DEFAULT_REPOSITORY_SOURCE_SELECTION,
    max_artifacts: int = _MAX_MEDIA_ARTIFACTS,
) -> dict[str, Any]:
    """Discover repository-native visual artifacts and return a stable manifest."""

    root = Path(repo_path).expanduser().resolve()
    selected = RepositorySourceSelection(selection.exclude_subtrees)
    artifact_limit = _validated_limit(
        max_artifacts,
        name="max_artifacts",
        maximum=_MAX_MEDIA_ARTIFACTS,
    )
    excluded = tuple(exclude_roots)
    discovered: list[tuple[Path, str, str, int]] = []
    for path in walk_repository_files(
        root,
        exclude_roots=excluded,
        selection=selected,
    ):
        if len(discovered) >= artifact_limit:
            break
        if path.is_symlink() or path.suffix.lower() not in SUPPORTED_MEDIA_EXTENSIONS:
            continue
        payload = read_regular_bytes(path, max_bytes=_MAX_MEDIA_BYTES)
        if payload is None:
            continue
        relative = path.relative_to(root).as_posix()
        discovered.append(
            (path, relative, hashlib.sha256(payload).hexdigest(), len(payload))
        )

    references = _discover_markdown_references(
        root,
        artifact_paths=frozenset(relative for _, relative, _, _ in discovered),
        exclude_roots=excluded,
        selection=selected,
    )
    artifacts: list[MediaArtifact] = []
    for path, relative, digest, size in discovered:
        artifact_references = tuple(references.get(relative, ()))
        caption = _caption(artifact_references)
        artifacts.append(
            MediaArtifact(
                path=relative,
                media_type=_media_type(path),
                mime_type=_mime_type(path),
                sha256=digest,
                size_bytes=size,
                role_hint=_role_hint(relative, artifact_references),
                references=artifact_references,
                caption=caption,
                surrounding_text=_surrounding_text(artifact_references),
            )
        )

    manifest = MediaManifest(
        schema=MEDIA_MANIFEST_SCHEMA,
        version=MEDIA_MANIFEST_VERSION,
        commit=commit if commit is not None else _git_commit(root),
        source_selection_digest=selected.digest,
        artifacts=tuple(sorted(artifacts, key=lambda artifact: artifact.path)),
        metadata={
            "supported_extensions": sorted(SUPPORTED_MEDIA_EXTENSIONS),
            "max_media_bytes": _MAX_MEDIA_BYTES,
        },
    )
    return manifest.to_dict()

evaluate_mmwiki_predictions 

evaluate_mmwiki_predictions(
    visual_facts_manifest: Mapping[str, Any],
    grounding_manifest: Mapping[str, Any],
    gold: Mapping[str, Any],
    *,
    k: int = 5
) -> dict[str, Any]

Evaluate visual fact extraction and visual-code grounding together.

Source code in codenib/wiki/media_eval.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def evaluate_mmwiki_predictions(
    visual_facts_manifest: Mapping[str, Any],
    grounding_manifest: Mapping[str, Any],
    gold: Mapping[str, Any],
    *,
    k: int = 5,
) -> dict[str, Any]:
    """Evaluate visual fact extraction and visual-code grounding together."""

    instances = _gold_instances(gold)
    stable_gold = {"instances": instances}
    facts = evaluate_visual_fact_extraction(visual_facts_manifest, stable_gold)
    grounding = evaluate_visual_code_grounding(
        grounding_manifest,
        stable_gold,
        k=k,
    )
    return {
        "task": "mmwiki",
        "artifact_count": len(instances),
        "visual_fact_extraction": facts,
        "visual_code_grounding": grounding,
    }

evaluate_visual_code_grounding 

evaluate_visual_code_grounding(
    grounding_manifest: Mapping[str, Any], gold: Mapping[str, Any], *, k: int = 5
) -> dict[str, Any]

Evaluate visual entity to source binding accuracy.

Source code in codenib/wiki/media_eval.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def evaluate_visual_code_grounding(
    grounding_manifest: Mapping[str, Any],
    gold: Mapping[str, Any],
    *,
    k: int = 5,
) -> dict[str, Any]:
    """Evaluate visual entity to source binding accuracy."""

    grounding_manifest = _require_mapping(
        grounding_manifest,
        label="grounding manifest",
    )
    result_limit = _validated_k(k)
    predicted_by_key: dict[tuple[str, str], list[dict[str, Any]]] = {}
    for binding in _mapping_items(
        grounding_manifest.get("bindings"),
        limit=_MAX_PREDICTED_BINDINGS,
    ):
        prediction = _prediction(binding)
        if prediction is None:
            continue
        key = (
            prediction["artifact_path"],
            _normalize(prediction["entity_name"]),
        )
        if not key[1]:
            continue
        predicted_by_key.setdefault(key, []).append(prediction)
    for values in predicted_by_key.values():
        values.sort(
            key=lambda item: (
                -item["score"],
                item["source_path"],
                item["symbol"],
            )
        )

    total = 0
    path_hits = 0
    symbol_hits = 0
    per_binding = []
    for instance in _gold_instances(gold):
        artifact_path = _gold_artifact_path(instance)
        for expected in _mapping_items(
            instance.get("gold_bindings"),
            limit=_MAX_GOLD_BINDINGS_PER_ARTIFACT,
        ):
            total += 1
            expected_entity = _required_text(
                expected.get("entity_name"),
                label="gold entity name",
            )
            entity_name = _normalize(expected_entity)
            expected_path = _required_relative_path(
                expected.get("source_path"),
                label="gold source path",
            )
            expected_symbol = _safe_text(expected.get("symbol"))
            predictions = predicted_by_key.get(
                (artifact_path, entity_name),
                [],
            )[:result_limit]
            path_hit = any(
                prediction["source_path"] == expected_path for prediction in predictions
            )
            symbol_hit = any(
                prediction["source_path"] == expected_path
                and (not expected_symbol or prediction["symbol"] == expected_symbol)
                for prediction in predictions
            )
            path_hits += int(path_hit)
            symbol_hits += int(symbol_hit)
            per_binding.append(
                {
                    "artifact_path": artifact_path,
                    "entity_name": expected_entity,
                    "path_hit_at_k": path_hit,
                    "symbol_hit_at_k": symbol_hit,
                    "predicted": predictions,
                }
            )
    return {
        "k": result_limit,
        "binding_count": total,
        "path_hit_at_k": _safe_div(path_hits, total),
        "symbol_hit_at_k": _safe_div(symbol_hits, total),
        "path_hits": path_hits,
        "symbol_hits": symbol_hits,
        "per_binding": per_binding,
    }

evaluate_visual_fact_extraction 

evaluate_visual_fact_extraction(
    visual_facts_manifest: Mapping[str, Any], gold: Mapping[str, Any]
) -> dict[str, Any]

Evaluate extracted visual entities against an MMWiki-style gold file.

Source code in codenib/wiki/media_eval.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def evaluate_visual_fact_extraction(
    visual_facts_manifest: Mapping[str, Any],
    gold: Mapping[str, Any],
) -> dict[str, Any]:
    """Evaluate extracted visual entities against an MMWiki-style gold file."""

    visual_facts_manifest = _require_mapping(
        visual_facts_manifest,
        label="visual facts manifest",
    )
    predicted_by_artifact: dict[str, Mapping[str, Any]] = {}
    for fact in _mapping_items(
        visual_facts_manifest.get("facts"),
        limit=_MAX_INSTANCES,
    ):
        path = _safe_relative_path(fact.get("artifact_path"))
        if path:
            predicted_by_artifact.setdefault(path, fact)

    true_positive = 0
    predicted_total = 0
    gold_total = 0
    per_artifact = []
    for instance in _gold_instances(gold):
        artifact_path = _gold_artifact_path(instance)
        predicted = {
            key
            for entity in _mapping_items(
                (predicted_by_artifact.get(artifact_path) or {}).get("entities"),
                limit=_MAX_ENTITIES_PER_ARTIFACT,
            )
            if (key := _entity_key(entity))
        }
        expected = {
            _gold_entity_key(entity)
            for entity in _mapping_items(
                instance.get("gold_entities"),
                limit=_MAX_ENTITIES_PER_ARTIFACT,
            )
        }
        hits = predicted & expected
        true_positive += len(hits)
        predicted_total += len(predicted)
        gold_total += len(expected)
        per_artifact.append(
            {
                "artifact_path": artifact_path,
                "entity_precision": _safe_div(len(hits), len(predicted)),
                "entity_recall": _safe_div(len(hits), len(expected)),
                "matched_entities": sorted(hits),
            }
        )
    precision = _safe_div(true_positive, predicted_total)
    recall = _safe_div(true_positive, gold_total)
    return {
        "entity_precision": precision,
        "entity_recall": recall,
        "entity_f1": _f1(precision, recall),
        "entity_true_positive": true_positive,
        "entity_predicted": predicted_total,
        "entity_gold": gold_total,
        "per_artifact": per_artifact,
    }

build_media_evidence_pack 

build_media_evidence_pack(
    slot: Mapping[str, Any],
    *,
    page_id: str = "",
    page_title: str = "",
    page_markdown: str = "",
    citations: Iterable[Mapping[str, Any]] = (),
    relations: Iterable[Mapping[str, Any]] = (),
    source_reader: SourceReader | None = None,
    max_sources: int = _DEFAULT_MAX_SOURCES,
    max_relations: int = _DEFAULT_MAX_RELATIONS,
    max_snippet_bytes: int = _DEFAULT_MAX_SNIPPET_BYTES
) -> dict[str, Any]

Build a bounded, provider-neutral evidence pack for one media slot.

This helper does not read repository files by itself. Callers may pass a source_reader that returns small snippets for already-selected citations, which keeps source exposure explicitly server-side and bounded.

Source code in codenib/wiki/media_evidence.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def build_media_evidence_pack(
    slot: Mapping[str, Any],
    *,
    page_id: str = "",
    page_title: str = "",
    page_markdown: str = "",
    citations: Iterable[Mapping[str, Any]] = (),
    relations: Iterable[Mapping[str, Any]] = (),
    source_reader: SourceReader | None = None,
    max_sources: int = _DEFAULT_MAX_SOURCES,
    max_relations: int = _DEFAULT_MAX_RELATIONS,
    max_snippet_bytes: int = _DEFAULT_MAX_SNIPPET_BYTES,
) -> dict[str, Any]:
    """Build a bounded, provider-neutral evidence pack for one media slot.

    This helper does not read repository files by itself. Callers may pass a
    ``source_reader`` that returns small snippets for already-selected
    citations, which keeps source exposure explicitly server-side and bounded.
    """

    max_sources = _validated_limit(
        max_sources, name="max_sources", maximum=_DEFAULT_MAX_SOURCES
    )
    max_relations = _validated_limit(
        max_relations, name="max_relations", maximum=_DEFAULT_MAX_RELATIONS
    )
    max_snippet_bytes = _validated_limit(
        max_snippet_bytes,
        name="max_snippet_bytes",
        maximum=_DEFAULT_MAX_SNIPPET_BYTES,
    )
    pack = WikiMediaEvidencePack(
        slot_id=_safe_text(slot.get("id")),
        kind=_safe_text(slot.get("kind")),
        title=_safe_text(slot.get("title")),
        purpose=_safe_text(slot.get("purpose")),
        page={
            "id": _safe_text(page_id),
            "title": _safe_text(page_title),
            "summary": _page_summary(page_markdown),
        },
        sources=_source_evidence(
            slot,
            citations,
            source_reader=source_reader,
            max_sources=max_sources,
            max_snippet_bytes=max_snippet_bytes,
        ),
        relations=_relation_evidence(relations, max_relations=max_relations),
        human_prior=_human_prior(slot.get("human_prior")),
    )
    data = pack.to_dict()
    try:
        encoded = json.dumps(
            data,
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    except (RecursionError, TypeError, ValueError) as exc:
        raise ValueError("wiki media evidence pack must contain bounded JSON") from exc
    if len(encoded) > MAX_MEDIA_EVIDENCE_PACK_BYTES:
        raise ValueError("wiki media evidence pack exceeds the byte limit")
    return data

build_visual_facts_manifest 

build_visual_facts_manifest(
    media_manifest: Mapping[str, Any],
    *,
    extractor: VisualFactExtractor = deterministic_visual_facts
) -> dict[str, Any]

Extract visual fact packs for every artifact in a media manifest.

Source code in codenib/wiki/media_facts.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def build_visual_facts_manifest(
    media_manifest: Mapping[str, Any],
    *,
    extractor: VisualFactExtractor = deterministic_visual_facts,
) -> dict[str, Any]:
    """Extract visual fact packs for every artifact in a media manifest."""

    facts = []
    for artifact in _mapping_items(
        media_manifest.get("artifacts"),
        limit=_MAX_ARTIFACTS,
    ):
        artifact_path = _safe_relative_path(artifact.get("path"))
        if not artifact_path:
            continue
        pack = extractor(artifact)
        if not isinstance(pack, Mapping):
            raise ValueError("visual fact extractor must return a mapping")
        normalized = _fact_pack_from_mapping(
            pack,
            artifact_path=artifact_path,
            artifact_sha256=_safe_text(artifact.get("sha256")),
            role_hint=_safe_text(artifact.get("role_hint") or "repository_image"),
        )
        if _json_size(normalized.to_dict()) > _MAX_FACT_PACK_BYTES:
            raise ValueError("visual fact pack exceeds the byte limit")
        facts.append(normalized)
    manifest = VisualFactsManifest(
        schema=MEDIA_FACTS_SCHEMA,
        version=MEDIA_FACTS_VERSION,
        media_manifest_sha256=_safe_text(media_manifest.get("manifest_sha256")),
        facts=tuple(sorted(facts, key=lambda fact: fact.artifact_path)),
    )
    return manifest.to_dict()

deterministic_visual_facts 

deterministic_visual_facts(artifact: Mapping[str, Any]) -> dict[str, Any]

Return a conservative local fact pack from artifact metadata only.

Source code in codenib/wiki/media_facts.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def deterministic_visual_facts(artifact: Mapping[str, Any]) -> dict[str, Any]:
    """Return a conservative local fact pack from artifact metadata only."""

    path = _safe_relative_path(artifact.get("path"))
    sha256 = _safe_text(artifact.get("sha256"))
    role = _safe_text(artifact.get("role_hint") or "repository_image")
    caption = _safe_text(artifact.get("caption"))
    surrounding = _safe_text(artifact.get("surrounding_text"))
    references = artifact.get("references") or ()
    entities = _metadata_entities(path, role, caption, surrounding)
    claims = _metadata_claims(path, role, caption, surrounding, references)
    pack = VisualFactPack(
        artifact_path=path,
        artifact_sha256=sha256,
        role_hint=role,
        extractor="local/metadata",
        entities=tuple(entities[:_MAX_FACTS_PER_ARTIFACT]),
        claims=tuple(claims[:_MAX_FACTS_PER_ARTIFACT]),
        metadata={"source": "artifact-path-caption-surrounding-markdown"},
    )
    return pack.to_dict()

discover_source_symbol_candidates 

discover_source_symbol_candidates(
    repo_path: str | Path,
    *,
    exclude_roots: Iterable[str | Path] = (),
    selection: RepositorySourceSelection = DEFAULT_REPOSITORY_SOURCE_SELECTION,
    max_candidates: int = _MAX_CANDIDATES
) -> list[dict[str, Any]]

Return a bounded, deterministic source-symbol inventory for grounding.

Source code in codenib/wiki/media_grounding.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def discover_source_symbol_candidates(
    repo_path: str | Path,
    *,
    exclude_roots: Iterable[str | Path] = (),
    selection: RepositorySourceSelection = DEFAULT_REPOSITORY_SOURCE_SELECTION,
    max_candidates: int = _MAX_CANDIDATES,
) -> list[dict[str, Any]]:
    """Return a bounded, deterministic source-symbol inventory for grounding."""

    root = Path(repo_path).expanduser().resolve()
    selected = RepositorySourceSelection(selection.exclude_subtrees)
    candidate_limit = _validated_limit(
        max_candidates,
        name="max_candidates",
        maximum=_MAX_CANDIDATES,
    )
    candidates: list[SourceSymbolCandidate] = []
    seen: set[tuple[str, str, int]] = set()
    for path in walk_repository_files(
        root,
        exclude_roots=exclude_roots,
        selection=selected,
    ):
        if len(candidates) >= candidate_limit:
            break
        if path.is_symlink() or path.suffix.lower() not in _SOURCE_EXTENSIONS:
            continue
        payload = read_regular_bytes(path, max_bytes=_MAX_SOURCE_BYTES)
        if payload is None:
            continue
        try:
            text = payload.decode("utf-8")
        except UnicodeDecodeError:
            continue
        relative = path.relative_to(root).as_posix()
        candidates.append(SourceSymbolCandidate(path=relative))
        if len(candidates) >= candidate_limit:
            break
        for symbol, line in _symbols(text):
            key = (relative, symbol, line)
            if key in seen:
                continue
            seen.add(key)
            candidates.append(
                SourceSymbolCandidate(
                    path=relative,
                    symbol=symbol,
                    kind="symbol",
                    line=line,
                )
            )
            if len(candidates) >= candidate_limit:
                break
    return [candidate.to_dict() for candidate in candidates[:candidate_limit]]

ground_visual_facts_to_sources 

ground_visual_facts_to_sources(
    visual_facts_manifest: Mapping[str, Any],
    source_candidates: Iterable[Mapping[str, Any]],
    *,
    max_bindings_per_entity: int = _MAX_BINDINGS_PER_ENTITY,
    scorer: VisualGroundingScorer | None = None
) -> dict[str, Any]

Ground visual entities to a source inventory.

The default scorer is deterministic and lexical. Callers can pass a scorer backed by BM25, embeddings, CodeGraph, LSP facts, or FactQueryIndex without changing the visual-code binding manifest schema.

Source code in codenib/wiki/media_grounding.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def ground_visual_facts_to_sources(
    visual_facts_manifest: Mapping[str, Any],
    source_candidates: Iterable[Mapping[str, Any]],
    *,
    max_bindings_per_entity: int = _MAX_BINDINGS_PER_ENTITY,
    scorer: VisualGroundingScorer | None = None,
) -> dict[str, Any]:
    """Ground visual entities to a source inventory.

    The default scorer is deterministic and lexical. Callers can pass a scorer
    backed by BM25, embeddings, CodeGraph, LSP facts, or FactQueryIndex without
    changing the visual-code binding manifest schema.
    """

    binding_limit = _validated_limit(
        max_bindings_per_entity,
        name="max_bindings_per_entity",
        maximum=_MAX_BINDINGS_PER_ENTITY,
    )
    if scorer is not None and not callable(scorer):
        raise ValueError("scorer must be callable")
    candidates_by_key: dict[tuple[str, str, str, int], SourceSymbolCandidate] = {}
    for value in _mapping_items(source_candidates, limit=_MAX_CANDIDATES):
        candidate = _candidate_from_mapping(value)
        if not candidate.path:
            continue
        key = (candidate.path, candidate.symbol, candidate.kind, candidate.line)
        candidates_by_key.setdefault(key, candidate)
    candidates = list(candidates_by_key.values())
    candidate_payloads = (
        {candidate: candidate.to_dict() for candidate in candidates}
        if scorer is not None
        else {}
    )
    bindings: list[VisualCodeBinding] = []
    entity_count = 0
    for fact_pack in _mapping_items(
        visual_facts_manifest.get("facts"),
        limit=_MAX_FACT_PACKS,
    ):
        if entity_count >= _MAX_TOTAL_ENTITIES:
            break
        artifact_path = _safe_relative_path(fact_pack.get("artifact_path"))
        if not artifact_path:
            continue
        for entity in _mapping_items(
            fact_pack.get("entities"),
            limit=_MAX_ENTITIES_PER_ARTIFACT,
        ):
            if entity_count >= _MAX_TOTAL_ENTITIES:
                break
            entity_count += 1
            entity_name = _safe_text(entity.get("name"))
            if not entity_name:
                continue
            hints = [
                entity_name,
                *[
                    _safe_text(candidate)
                    for candidate in islice(
                        _non_string_iterable(entity.get("grounding_candidates")),
                        _MAX_GROUNDING_HINTS,
                    )
                ],
            ]
            scored = [
                binding
                for binding in (
                    _score_with_optional_scorer(
                        artifact_path=artifact_path,
                        entity=entity,
                        entity_name=entity_name,
                        hints=hints,
                        candidate=candidate,
                        candidate_payload=candidate_payloads.get(candidate),
                        scorer=scorer,
                    )
                    for candidate in candidates
                )
                if binding is not None
            ]
            scored.sort(
                key=lambda binding: (
                    -binding.score,
                    binding.source_path,
                    binding.symbol,
                    binding.line,
                )
            )
            bindings.extend(scored[:binding_limit])
    manifest = VisualGroundingManifest(
        schema=MEDIA_GROUNDING_SCHEMA,
        version=MEDIA_GROUNDING_VERSION,
        visual_facts_manifest_sha256=_safe_text(
            visual_facts_manifest.get("manifest_sha256")
        ),
        bindings=tuple(
            sorted(
                _dedupe_bindings(bindings),
                key=lambda binding: (
                    binding.artifact_path,
                    binding.entity_name,
                    -binding.score,
                    binding.source_path,
                    binding.symbol,
                ),
            )
        ),
    )
    return manifest.to_dict()

diff_media_manifests 

diff_media_manifests(
    previous: Mapping[str, Any], current: Mapping[str, Any]
) -> dict[str, Any]

Return a stable path/hash diff between two media manifests.

Source code in codenib/wiki/media_incremental.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def diff_media_manifests(
    previous: Mapping[str, Any],
    current: Mapping[str, Any],
) -> dict[str, Any]:
    """Return a stable path/hash diff between two media manifests."""

    previous_artifacts = _artifacts_by_path(previous)
    current_artifacts = _artifacts_by_path(current)
    return _diff_artifact_maps(
        previous,
        current,
        previous_artifacts=previous_artifacts,
        current_artifacts=current_artifacts,
    )

merge_incremental_visual_facts 

merge_incremental_visual_facts(
    current_media_manifest: Mapping[str, Any],
    reusable_fact_packs: Iterable[Mapping[str, Any]],
    new_fact_packs: Iterable[Mapping[str, Any]],
) -> dict[str, Any]

Merge reused and newly extracted fact packs for the current media manifest.

Source code in codenib/wiki/media_incremental.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def merge_incremental_visual_facts(
    current_media_manifest: Mapping[str, Any],
    reusable_fact_packs: Iterable[Mapping[str, Any]],
    new_fact_packs: Iterable[Mapping[str, Any]],
) -> dict[str, Any]:
    """Merge reused and newly extracted fact packs for the current media manifest."""

    current_artifacts = _artifacts_by_path(current_media_manifest)
    by_path: dict[str, dict[str, Any]] = {}
    packs = chain(
        _mapping_items(reusable_fact_packs, limit=_MAX_FACT_PACKS),
        _mapping_items(new_fact_packs, limit=_MAX_FACT_PACKS),
    )
    for pack in packs:
        path = _safe_relative_path(pack.get("artifact_path"))
        artifact = current_artifacts.get(path)
        if artifact is None or _safe_text(pack.get("artifact_sha256")) != _safe_text(
            artifact.get("sha256")
        ):
            continue
        normalized = normalize_visual_fact_pack(pack, artifact=artifact)
        by_path[path] = normalized
    facts = [by_path[path] for path in sorted(by_path)]
    payload = {
        "schema": MEDIA_FACTS_SCHEMA,
        "version": MEDIA_FACTS_VERSION,
        "media_manifest_sha256": _safe_text(
            current_media_manifest.get("manifest_sha256") or ""
        ),
        "fact_count": len(facts),
        "facts": facts,
    }
    payload["manifest_sha256"] = compute_visual_facts_manifest_sha256(
        schema=MEDIA_FACTS_SCHEMA,
        version=MEDIA_FACTS_VERSION,
        media_manifest_sha256=payload["media_manifest_sha256"],
        facts=facts,
    )
    return payload

plan_incremental_visual_fact_update 

plan_incremental_visual_fact_update(
    previous_media_manifest: Mapping[str, Any],
    current_media_manifest: Mapping[str, Any],
    previous_visual_facts_manifest: Mapping[str, Any],
    *,
    expected_extractor: str | None = None
) -> dict[str, Any]

Plan reusable facts for the explicitly selected extraction policy.

Reuse is disabled unless expected_extractor names the extractor that will be used for new work. Callers should change that identifier whenever their model or extraction policy changes.

Source code in codenib/wiki/media_incremental.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def plan_incremental_visual_fact_update(
    previous_media_manifest: Mapping[str, Any],
    current_media_manifest: Mapping[str, Any],
    previous_visual_facts_manifest: Mapping[str, Any],
    *,
    expected_extractor: str | None = None,
) -> dict[str, Any]:
    """Plan reusable facts for the explicitly selected extraction policy.

    Reuse is disabled unless ``expected_extractor`` names the extractor that
    will be used for new work. Callers should change that identifier whenever
    their model or extraction policy changes.
    """

    previous_artifacts = _artifacts_by_path(previous_media_manifest)
    current_artifacts = _artifacts_by_path(current_media_manifest)
    diff = _diff_artifact_maps(
        previous_media_manifest,
        current_media_manifest,
        previous_artifacts=previous_artifacts,
        current_artifacts=current_artifacts,
    )
    previous_visual_facts_manifest = _require_mapping(
        previous_visual_facts_manifest,
        label="previous visual facts manifest",
    )
    if expected_extractor is not None and not isinstance(expected_extractor, str):
        raise ValueError("expected_extractor must be a non-empty string")
    extractor = _safe_text(expected_extractor)
    if expected_extractor is not None and not extractor:
        raise ValueError("expected_extractor must be a non-empty string")
    previous_fact_items = _bounded_fact_items(
        previous_visual_facts_manifest.get("facts")
    )
    previous_facts = (
        _trusted_previous_facts(
            previous_visual_facts_manifest,
            previous_media_manifest_sha256=_safe_text(
                previous_media_manifest.get("manifest_sha256")
            ),
            previous_artifacts=previous_artifacts,
            facts=previous_fact_items,
        )
        if previous_fact_items is not None
        else {}
    )
    reusable_fact_packs = []
    extract_artifact_paths = []
    removed_artifact_paths = []
    for change in diff["changes"]:
        path = change["path"]
        if change["status"] == "unchanged":
            artifact = current_artifacts[path]
            fact = previous_facts.get(path)
            if (
                not extractor
                or fact is None
                or _safe_text(fact.get("extractor")) != extractor
                or _safe_text(fact.get("artifact_sha256"))
                != _safe_text(artifact.get("sha256"))
            ):
                extract_artifact_paths.append(path)
                continue
            try:
                reusable_fact_packs.append(
                    normalize_visual_fact_pack(fact, artifact=artifact)
                )
            except ValueError:
                extract_artifact_paths.append(path)
        elif change["status"] in {"added", "changed"}:
            extract_artifact_paths.append(path)
        elif change["status"] == "removed":
            removed_artifact_paths.append(path)
    payload = {
        "schema": MEDIA_INCREMENTAL_SCHEMA,
        "version": MEDIA_INCREMENTAL_VERSION,
        "media_diff": diff,
        "current_media_manifest_sha256": _safe_text(
            current_media_manifest.get("manifest_sha256") or ""
        ),
        "previous_visual_facts_manifest_sha256": _safe_text(
            previous_visual_facts_manifest.get("manifest_sha256") or ""
        ),
        "expected_extractor": extractor,
        "reusable_fact_packs": reusable_fact_packs,
        "extract_artifact_paths": sorted(extract_artifact_paths),
        "removed_artifact_paths": sorted(removed_artifact_paths),
    }
    payload["plan_sha256"] = _sha256_json(payload)
    return payload

build_multimodal_knowledge_view 

build_multimodal_knowledge_view(
    media_manifest: Mapping[str, Any],
    visual_facts_manifest: Mapping[str, Any],
    grounding_manifest: Mapping[str, Any],
) -> dict[str, Any]

Join media artifacts, visual facts, and source bindings into one view.

Source code in codenib/wiki/media_knowledge.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def build_multimodal_knowledge_view(
    media_manifest: Mapping[str, Any],
    visual_facts_manifest: Mapping[str, Any],
    grounding_manifest: Mapping[str, Any],
) -> dict[str, Any]:
    """Join media artifacts, visual facts, and source bindings into one view."""

    artifacts = {
        str(artifact.get("path") or ""): dict(artifact)
        for artifact in media_manifest.get("artifacts") or ()
        if isinstance(artifact, Mapping) and artifact.get("path")
    }
    facts = {
        str(fact.get("artifact_path") or ""): dict(fact)
        for fact in visual_facts_manifest.get("facts") or ()
        if isinstance(fact, Mapping) and fact.get("artifact_path")
    }
    bindings_by_artifact: dict[str, list[dict[str, Any]]] = {}
    for binding in grounding_manifest.get("bindings") or ():
        if not isinstance(binding, Mapping):
            continue
        artifact_path = str(binding.get("artifact_path") or "")
        if artifact_path:
            bindings_by_artifact.setdefault(artifact_path, []).append(dict(binding))

    entries = []
    for path in sorted(set(artifacts) | set(facts) | set(bindings_by_artifact)):
        artifact = artifacts.get(path, {})
        fact = facts.get(path, {})
        bindings = sorted(
            bindings_by_artifact.get(path, []),
            key=lambda item: (
                str(item.get("source_path") or ""),
                str(item.get("symbol") or ""),
                str(item.get("entity_name") or ""),
            ),
        )
        entries.append(
            {
                "artifact": artifact,
                "facts": fact,
                "bindings": bindings,
                "search_text": _entry_search_text(artifact, fact, bindings),
            }
        )
    payload = {
        "schema": MULTIMODAL_KNOWLEDGE_SCHEMA,
        "version": MULTIMODAL_KNOWLEDGE_VERSION,
        "media_manifest_sha256": str(media_manifest.get("manifest_sha256") or ""),
        "visual_facts_manifest_sha256": str(
            visual_facts_manifest.get("manifest_sha256") or ""
        ),
        "grounding_manifest_sha256": str(
            grounding_manifest.get("manifest_sha256") or ""
        ),
        "entry_count": len(entries),
        "entries": entries,
    }
    payload["view_sha256"] = _sha256_json(
        {key: value for key, value in payload.items() if key != "view_sha256"}
    )
    return payload

find_visual_code_links(
    view: Mapping[str, Any], source_path: str, *, symbol: str = ""
) -> list[dict[str, Any]]

Return visual entries that ground to a file, optionally a symbol.

Source code in codenib/wiki/media_knowledge.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def find_visual_code_links(
    view: Mapping[str, Any],
    source_path: str,
    *,
    symbol: str = "",
) -> list[dict[str, Any]]:
    """Return visual entries that ground to a file, optionally a symbol."""

    links = []
    for entry in view.get("entries") or ():
        if not isinstance(entry, Mapping):
            continue
        for binding in entry.get("bindings") or ():
            if not isinstance(binding, Mapping):
                continue
            if binding.get("source_path") != source_path:
                continue
            if symbol and binding.get("symbol") != symbol:
                continue
            links.append(
                {
                    "artifact_path": ((entry.get("artifact") or {}).get("path") or ""),
                    "binding": dict(binding),
                    "artifact": dict(entry.get("artifact") or {}),
                    "facts": dict(entry.get("facts") or {}),
                }
            )
    links.sort(
        key=lambda item: (
            str(item["artifact_path"]),
            str(item["binding"].get("entity_name") or ""),
            str(item["binding"].get("symbol") or ""),
        )
    )
    return links

get_visual_evidence 

get_visual_evidence(
    view: Mapping[str, Any], artifact_path: str
) -> dict[str, Any] | None

Return one visual knowledge entry by artifact path.

Source code in codenib/wiki/media_knowledge.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def get_visual_evidence(
    view: Mapping[str, Any],
    artifact_path: str,
) -> dict[str, Any] | None:
    """Return one visual knowledge entry by artifact path."""

    for entry in view.get("entries") or ():
        if not isinstance(entry, Mapping):
            continue
        artifact = entry.get("artifact") or {}
        if isinstance(artifact, Mapping) and artifact.get("path") == artifact_path:
            return {
                "artifact": dict(artifact),
                "facts": dict(entry.get("facts") or {}),
                "bindings": list(entry.get("bindings") or ()),
            }
    return None

search_visual_context 

search_visual_context(
    view: Mapping[str, Any], query: str, *, limit: int = 5
) -> list[dict[str, Any]]

Search multimodal entries using a deterministic lexical scorer.

Source code in codenib/wiki/media_knowledge.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def search_visual_context(
    view: Mapping[str, Any],
    query: str,
    *,
    limit: int = 5,
) -> list[dict[str, Any]]:
    """Search multimodal entries using a deterministic lexical scorer."""

    tokens = _tokens(query)
    if not tokens:
        return []
    results = []
    for entry in view.get("entries") or ():
        if not isinstance(entry, Mapping):
            continue
        haystack = str(entry.get("search_text") or "").lower()
        score = sum(1 for token in tokens if token in haystack)
        if score:
            results.append(
                {
                    "artifact_path": ((entry.get("artifact") or {}).get("path") or ""),
                    "score": score,
                    "artifact": dict(entry.get("artifact") or {}),
                    "facts": dict(entry.get("facts") or {}),
                    "bindings": list(entry.get("bindings") or ()),
                }
            )
    results.sort(key=lambda item: (-item["score"], item["artifact_path"]))
    return results[: max(0, limit)]

build_multimodal_repository_knowledge 

build_multimodal_repository_knowledge(
    repo_path: str | Path,
    *,
    commit: str | None = None,
    exclude_roots: Iterable[str | Path] = (),
    selection: RepositorySourceSelection = DEFAULT_REPOSITORY_SOURCE_SELECTION,
    extractor: VisualFactExtractor | None = None,
    scorer: VisualGroundingScorer | None = None,
    max_artifacts: int = 4096,
    max_source_candidates: int = 8192
) -> dict[str, Any]

Build the deterministic multimodal repository knowledge bundle.

Source code in codenib/wiki/media_pipeline.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def build_multimodal_repository_knowledge(
    repo_path: str | Path,
    *,
    commit: str | None = None,
    exclude_roots: Iterable[str | Path] = (),
    selection: RepositorySourceSelection = DEFAULT_REPOSITORY_SOURCE_SELECTION,
    extractor: VisualFactExtractor | None = None,
    scorer: VisualGroundingScorer | None = None,
    max_artifacts: int = 4096,
    max_source_candidates: int = 8192,
) -> dict[str, Any]:
    """Build the deterministic multimodal repository knowledge bundle."""

    root = _repository_root(repo_path)
    excluded = tuple(exclude_roots)
    media_manifest = discover_media_manifest(
        root,
        commit=commit,
        exclude_roots=excluded,
        selection=selection,
        max_artifacts=max_artifacts,
    )
    facts_kwargs: dict[str, Any] = {}
    if extractor is not None:
        facts_kwargs["extractor"] = extractor
    visual_facts_manifest = build_visual_facts_manifest(media_manifest, **facts_kwargs)
    source_candidates = discover_source_symbol_candidates(
        root,
        exclude_roots=excluded,
        selection=selection,
        max_candidates=max_source_candidates,
    )
    grounding_manifest = ground_visual_facts_to_sources(
        visual_facts_manifest,
        source_candidates,
        scorer=scorer,
    )
    knowledge_view = build_multimodal_knowledge_view(
        media_manifest,
        visual_facts_manifest,
        grounding_manifest,
    )
    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,
    )

build_multimodal_knowledge_bundle 

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.

Source code in codenib/wiki/media_storage.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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": source_candidate_count,
        "grounding_manifest": dict(grounding_manifest),
        "knowledge_view": dict(knowledge_view),
        "component_sha256": {
            "media_manifest": media_manifest.get("manifest_sha256"),
            "visual_facts_manifest": visual_facts_manifest.get("manifest_sha256"),
            "grounding_manifest": grounding_manifest.get("manifest_sha256"),
            "knowledge_view": knowledge_view.get("view_sha256"),
        },
    }
    bundle["bundle_sha256"] = _stable_sha256(
        {key: value for key, value in bundle.items() if key != "bundle_sha256"}
    )
    return validate_multimodal_knowledge_bundle(bundle)

load_multimodal_knowledge_bundle 

load_multimodal_knowledge_bundle(path: str | Path) -> dict[str, Any]

Load and validate a persisted multimodal knowledge bundle.

Source code in codenib/wiki/media_storage.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def load_multimodal_knowledge_bundle(path: str | Path) -> dict[str, Any]:
    """Load and validate a persisted multimodal knowledge bundle."""

    source = Path(path).expanduser()
    raw = read_regular_bytes(source, max_bytes=_MAX_BUNDLE_BYTES)
    if raw is None:
        raise ValueError(
            "multimodal knowledge bundle must be a stable regular file "
            "within the byte limit"
        )
    validate_bounded_json_stream(
        io.BytesIO(raw),
        label="multimodal knowledge bundle",
        max_bytes=_MAX_BUNDLE_BYTES,
        max_nodes=_MAX_BUNDLE_NODES,
        max_lexical_tokens=_MAX_BUNDLE_TOKENS,
    )
    try:
        data = json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=_reject_duplicate_object,
            parse_constant=_reject_nonfinite_number,
            parse_float=_finite_float,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("multimodal knowledge bundle contains invalid JSON") from exc
    return validate_multimodal_knowledge_bundle(data)

save_multimodal_knowledge_bundle 

save_multimodal_knowledge_bundle(bundle: Mapping[str, Any], path: str | Path) -> None

Atomically write a multimodal knowledge bundle as stable JSON.

Source code in codenib/wiki/media_storage.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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).expanduser()
    destination.parent.mkdir(parents=True, exist_ok=True)
    payload = (
        json.dumps(
            validated,
            allow_nan=False,
            ensure_ascii=False,
            indent=2,
            sort_keys=True,
        ).encode("utf-8")
        + b"\n"
    )
    if len(payload) > _MAX_BUNDLE_BYTES:
        raise ValueError("multimodal knowledge bundle exceeds the byte limit")
    try:
        existing = os.stat(destination, follow_symlinks=False)
    except FileNotFoundError:
        existing_mode = None
    else:
        existing_mode = (
            stat.S_IMODE(existing.st_mode) if stat.S_ISREG(existing.st_mode) else None
        )
    fd, temp_name = tempfile.mkstemp(
        prefix=f".{destination.name}.",
        suffix=".tmp",
        dir=str(destination.parent),
    )
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(payload)
            handle.flush()
            if existing_mode is not None:
                os.fchmod(handle.fileno(), existing_mode)
            os.fsync(handle.fileno())
        os.replace(temp_name, destination)
    finally:
        try:
            os.unlink(temp_name)
        except FileNotFoundError:
            pass

validate_multimodal_knowledge_bundle 

validate_multimodal_knowledge_bundle(bundle: Mapping[str, Any]) -> dict[str, Any]

Return a normalized bundle or raise ValueError for invalid input.

Source code in codenib/wiki/media_storage.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
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 (
        type(data.get("schema_version")) is not int
        or data["schema_version"] != MULTIMODAL_KNOWLEDGE_BUNDLE_VERSION
    ):
        raise ValueError("multimodal knowledge bundle version is unsupported")
    media_manifest = _mapping(data.get("media_manifest"), label="media_manifest")
    visual_facts_manifest = _mapping(
        data.get("visual_facts_manifest"),
        label="visual_facts_manifest",
    )
    grounding_manifest = _mapping(
        data.get("grounding_manifest"),
        label="grounding_manifest",
    )
    knowledge_view = _mapping(data.get("knowledge_view"), label="knowledge_view")
    component_sha256 = _mapping(
        data.get("component_sha256"),
        label="component_sha256",
    )
    data.update(
        {
            "media_manifest": media_manifest,
            "visual_facts_manifest": visual_facts_manifest,
            "grounding_manifest": grounding_manifest,
            "knowledge_view": knowledge_view,
            "component_sha256": component_sha256,
        }
    )
    source_candidate_count = data.get("source_candidate_count")
    if (
        type(source_candidate_count) is not int
        or not 0 <= source_candidate_count <= _MAX_SOURCE_CANDIDATES
    ):
        raise ValueError(
            "multimodal knowledge bundle source_candidate_count is invalid"
        )

    component_digests = {
        "media_manifest": _validate_media_manifest(media_manifest),
        "visual_facts_manifest": _validate_visual_facts_manifest(
            visual_facts_manifest,
        ),
        "grounding_manifest": _validate_grounding_manifest(grounding_manifest),
        "knowledge_view": _validate_knowledge_view(knowledge_view),
    }
    if (
        visual_facts_manifest["media_manifest_sha256"]
        != component_digests["media_manifest"]
    ):
        raise ValueError("visual facts manifest is bound to another media manifest")
    if (
        grounding_manifest["visual_facts_manifest_sha256"]
        != component_digests["visual_facts_manifest"]
    ):
        raise ValueError("grounding manifest is bound to another visual facts manifest")
    linked_view_digests = {
        "media_manifest": knowledge_view.get("media_manifest_sha256"),
        "visual_facts_manifest": knowledge_view.get("visual_facts_manifest_sha256"),
        "grounding_manifest": knowledge_view.get("grounding_manifest_sha256"),
    }
    if any(
        linked_view_digests[key] != component_digests[key]
        for key in linked_view_digests
    ):
        raise ValueError("multimodal knowledge view component binding does not match")
    if set(component_sha256) != set(component_digests):
        raise ValueError("multimodal knowledge component hash inventory is invalid")
    for key, digest in component_sha256.items():
        if type(key) is not str or not key:
            raise ValueError("multimodal knowledge component hash key is invalid")
        _digest(digest, label=f"component_sha256.{key}")
    if any(
        component_sha256.get(key) != digest for key, digest in component_digests.items()
    ):
        raise ValueError("multimodal knowledge component hash does not match")

    expected_hash = _stable_sha256(
        {key: value for key, value in data.items() if key != "bundle_sha256"}
    )
    recorded_hash = _digest(data.get("bundle_sha256"), label="bundle_sha256")
    if not hmac.compare_digest(recorded_hash, expected_hash):
        raise ValueError("multimodal knowledge bundle hash does not match")

    normalized = copy.deepcopy(data)
    validate_json_complexity(
        normalized,
        label="multimodal knowledge bundle",
        max_nodes=_MAX_BUNDLE_NODES,
    )
    if len(_canonical_json_bytes(normalized)) > _MAX_BUNDLE_BYTES:
        raise ValueError("multimodal knowledge bundle exceeds the byte limit")
    return normalized

multimodal_tool_schemas 

multimodal_tool_schemas() -> list[dict[str, Any]]

Return independent mutable copies of the stable tool schemas.

Source code in codenib/wiki/media_tools.py 
108
109
110
111
def multimodal_tool_schemas() -> list[dict[str, Any]]:
    """Return independent mutable copies of the stable tool schemas."""

    return list(copy.deepcopy(_MULTIMODAL_TOOL_SCHEMAS))

visual_fact_extractor_from_config 

visual_fact_extractor_from_config(
    config: Any, *, repo_path: str | Path | None = None
) -> OpenAICompatibleVisualFactExtractor | None

Build a visual-fact extractor from QAConfig-shaped settings.

Source code in codenib/wiki/media_vlm.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def visual_fact_extractor_from_config(
    config: Any,
    *,
    repo_path: str | Path | None = None,
) -> 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()
    raw_options = getattr(config, "wiki_visual_facts_options", {}) or {}
    if not isinstance(raw_options, Mapping):
        raise ValueError("wiki visual fact options must be a mapping")
    options = dict(raw_options)
    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,
        repo_path=repo_path,
    )