diff --git a/codenib/wiki/__init__.py b/codenib/wiki/__init__.py index fb723c3c..89ba32d0 100644 --- a/codenib/wiki/__init__.py +++ b/codenib/wiki/__init__.py @@ -13,11 +13,21 @@ from .builder import WikiBuilder from .media_artifacts import discover_media_manifest from .media_evidence import build_media_evidence_pack +from .media_eval import ( + evaluate_mmwiki_predictions, + evaluate_visual_code_grounding, + evaluate_visual_fact_extraction, +) from .media_facts import build_visual_facts_manifest, deterministic_visual_facts from .media_grounding import ( discover_source_symbol_candidates, ground_visual_facts_to_sources, ) +from .media_incremental import ( + diff_media_manifests, + merge_incremental_visual_facts, + plan_incremental_visual_fact_update, +) from .media_knowledge import ( build_multimodal_knowledge_view, find_visual_code_links, @@ -38,10 +48,16 @@ "build_multimodal_repository_knowledge", "build_visual_facts_manifest", "deterministic_visual_facts", + "diff_media_manifests", "discover_media_manifest", "discover_source_symbol_candidates", + "evaluate_mmwiki_predictions", + "evaluate_visual_code_grounding", + "evaluate_visual_fact_extraction", "find_visual_code_links", "get_visual_evidence", "ground_visual_facts_to_sources", + "merge_incremental_visual_facts", + "plan_incremental_visual_fact_update", "search_visual_context", ] diff --git a/codenib/wiki/media_eval.py b/codenib/wiki/media_eval.py new file mode 100644 index 00000000..e818798a --- /dev/null +++ b/codenib/wiki/media_eval.py @@ -0,0 +1,187 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +"""Evaluation helpers for multimodal repository knowledge views.""" + +from __future__ import annotations + +import re +from typing import Any, Mapping + + +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.""" + + predicted_by_artifact = { + str(fact.get("artifact_path") or ""): fact + for fact in visual_facts_manifest.get("facts") or () + if isinstance(fact, Mapping) + } + true_positive = 0 + predicted_total = 0 + gold_total = 0 + per_artifact = [] + for instance in _gold_instances(gold): + artifact_path = str(instance.get("artifact_path") or "") + predicted = { + _entity_key(entity) + for entity in (predicted_by_artifact.get(artifact_path) or {}).get( + "entities" + ) + or () + if isinstance(entity, Mapping) and _entity_key(entity) + } + expected = { + _entity_key(entity) + for entity in instance.get("gold_entities") or () + if isinstance(entity, Mapping) and _entity_key(entity) + } + 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, + } + + +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.""" + + predicted_by_key: dict[tuple[str, str], list[Mapping[str, Any]]] = {} + for binding in grounding_manifest.get("bindings") or (): + if not isinstance(binding, Mapping): + continue + key = ( + str(binding.get("artifact_path") or ""), + _normalize(str(binding.get("entity_name") or "")), + ) + predicted_by_key.setdefault(key, []).append(binding) + for values in predicted_by_key.values(): + values.sort( + key=lambda item: ( + -float(item.get("score") or 0.0), + str(item.get("source_path") or ""), + str(item.get("symbol") or ""), + ) + ) + + total = 0 + path_hits = 0 + symbol_hits = 0 + per_binding = [] + for instance in _gold_instances(gold): + artifact_path = str(instance.get("artifact_path") or "") + for expected in instance.get("gold_bindings") or (): + if not isinstance(expected, Mapping): + continue + total += 1 + entity_name = _normalize(str(expected.get("entity_name") or "")) + predictions = predicted_by_key.get((artifact_path, entity_name), [])[ + : max(0, k) + ] + expected_path = str(expected.get("source_path") or "") + expected_symbol = str(expected.get("symbol") or "") + path_hit = any( + prediction.get("source_path") == expected_path + for prediction in predictions + ) + symbol_hit = any( + prediction.get("source_path") == expected_path + and (not expected_symbol or prediction.get("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.get("entity_name") or "", + "path_hit_at_k": path_hit, + "symbol_hit_at_k": symbol_hit, + "predicted": [dict(prediction) for prediction in predictions], + } + ) + return { + "k": k, + "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, + } + + +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.""" + + facts = evaluate_visual_fact_extraction(visual_facts_manifest, gold) + grounding = evaluate_visual_code_grounding(grounding_manifest, gold, k=k) + return { + "task": "mmwiki", + "artifact_count": len(list(_gold_instances(gold))), + "visual_fact_extraction": facts, + "visual_code_grounding": grounding, + } + + +def _gold_instances(gold: Mapping[str, Any]) -> list[Mapping[str, Any]]: + instances = gold.get("instances") or () + return [instance for instance in instances if isinstance(instance, Mapping)] + + +def _entity_key(entity: Mapping[str, Any]) -> str: + name = _normalize(str(entity.get("name") or "")) + kind = _normalize(str(entity.get("type") or "unknown")) + return f"{name}:{kind}" if name else "" + + +def _normalize(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "", str(value or "").lower()) + + +def _safe_div(numerator: int | float, denominator: int | float) -> float: + return float(numerator) / float(denominator) if denominator else 0.0 + + +def _f1(precision: float, recall: float) -> float: + return 2 * precision * recall / (precision + recall) if precision + recall else 0.0 + + +__all__ = [ + "evaluate_mmwiki_predictions", + "evaluate_visual_code_grounding", + "evaluate_visual_fact_extraction", +] diff --git a/codenib/wiki/media_incremental.py b/codenib/wiki/media_incremental.py new file mode 100644 index 00000000..f82a0410 --- /dev/null +++ b/codenib/wiki/media_incremental.py @@ -0,0 +1,174 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +"""Incremental update planning for multimodal repository knowledge.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any, Iterable, Mapping + +from .media_facts import ( + MEDIA_FACTS_SCHEMA, + MEDIA_FACTS_VERSION, + normalize_visual_fact_pack, +) + +MEDIA_INCREMENTAL_SCHEMA = "codenib.media-incremental-plan.v1" +MEDIA_INCREMENTAL_VERSION = 1 + + +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) + changes = [] + for path in sorted(set(previous_artifacts) | set(current_artifacts)): + before = previous_artifacts.get(path) + after = current_artifacts.get(path) + if before is None: + status = "added" + elif after is None: + status = "removed" + elif before.get("sha256") == after.get("sha256"): + status = "unchanged" + else: + status = "changed" + changes.append( + { + "path": path, + "status": status, + "previous_sha256": str((before or {}).get("sha256") or ""), + "current_sha256": str((after or {}).get("sha256") or ""), + } + ) + return { + "schema": MEDIA_INCREMENTAL_SCHEMA, + "version": MEDIA_INCREMENTAL_VERSION, + "previous_media_manifest_sha256": str(previous.get("manifest_sha256") or ""), + "current_media_manifest_sha256": str(current.get("manifest_sha256") or ""), + "counts": { + "added": sum(1 for change in changes if change["status"] == "added"), + "removed": sum(1 for change in changes if change["status"] == "removed"), + "changed": sum(1 for change in changes if change["status"] == "changed"), + "unchanged": sum( + 1 for change in changes if change["status"] == "unchanged" + ), + }, + "changes": changes, + } + + +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], +) -> dict[str, Any]: + """Plan which visual facts can be reused and which artifacts need VLM work.""" + + diff = diff_media_manifests(previous_media_manifest, current_media_manifest) + current_artifacts = _artifacts_by_path(current_media_manifest) + previous_facts = { + str(fact.get("artifact_path") or ""): dict(fact) + for fact in previous_visual_facts_manifest.get("facts") or () + if isinstance(fact, Mapping) and fact.get("artifact_path") + } + reusable_fact_packs = [] + extract_artifact_paths = [] + removed_artifact_paths = [] + for change in diff["changes"]: + path = change["path"] + if change["status"] == "unchanged" and path in previous_facts: + fact = previous_facts[path] + if fact.get("artifact_sha256") == current_artifacts[path].get("sha256"): + reusable_fact_packs.append(fact) + else: + 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) + return { + "schema": MEDIA_INCREMENTAL_SCHEMA, + "version": MEDIA_INCREMENTAL_VERSION, + "media_diff": diff, + "current_media_manifest_sha256": str( + current_media_manifest.get("manifest_sha256") or "" + ), + "previous_visual_facts_manifest_sha256": str( + previous_visual_facts_manifest.get("manifest_sha256") or "" + ), + "reusable_fact_packs": reusable_fact_packs, + "extract_artifact_paths": sorted(extract_artifact_paths), + "removed_artifact_paths": sorted(removed_artifact_paths), + } + + +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]] = {} + for pack in list(reusable_fact_packs or ()) + list(new_fact_packs or ()): + if not isinstance(pack, Mapping): + continue + normalized = normalize_visual_fact_pack(pack) + path = normalized.get("artifact_path") + artifact = current_artifacts.get(path) + if artifact is None or normalized.get("artifact_sha256") != artifact.get( + "sha256" + ): + continue + 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": str( + current_media_manifest.get("manifest_sha256") or "" + ), + "fact_count": len(facts), + "facts": facts, + } + payload["manifest_sha256"] = _sha256_json( + {key: value for key, value in payload.items() if key != "manifest_sha256"} + ) + return payload + + +def _artifacts_by_path(manifest: Mapping[str, Any]) -> dict[str, dict[str, Any]]: + return { + str(artifact.get("path") or ""): dict(artifact) + for artifact in manifest.get("artifacts") or () + if isinstance(artifact, Mapping) and artifact.get("path") + } + + +def _sha256_json(payload: Mapping[str, Any]) -> str: + encoded = json.dumps( + payload, + allow_nan=False, + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +__all__ = [ + "MEDIA_INCREMENTAL_SCHEMA", + "MEDIA_INCREMENTAL_VERSION", + "diff_media_manifests", + "merge_incremental_visual_facts", + "plan_incremental_visual_fact_update", +] diff --git a/docs/experiments/multimodal_repository_knowledge.md b/docs/experiments/multimodal_repository_knowledge.md index 46ef7576..c2daa1b4 100644 --- a/docs/experiments/multimodal_repository_knowledge.md +++ b/docs/experiments/multimodal_repository_knowledge.md @@ -77,6 +77,55 @@ a queryable view. It exposes three functions that future MCP tools can wrap: surface as an MCP-compatible tool router with stable tool schemas and bounded input validation. This keeps the query surface testable before wiring it into a server-specific MCP registration path. + +### Incremental updates + +`codenib.wiki.media_incremental` provides deterministic update planning for +multimodal views. It compares two media manifests by path and content hash, +marks artifacts as added, removed, changed, or unchanged, and identifies which +visual fact packs can be reused without another VLM call. + +This is the first step toward incremental multimodal maintenance: + +```text +media unchanged -> reuse existing VisualFactPack +media changed -> rerun VLM/extractor for that artifact +media removed -> drop stale visual facts and bindings +``` + +### MMWiki-style evaluation + +`codenib.wiki.media_eval` defines a small evaluation protocol for the first +benchmark seed. It does not try to replace SWE-bench Multimodal or MM-IssueLoc. +Instead, it measures whether repository visuals can be compiled into persistent +wiki knowledge: + +- visual entity extraction precision / recall / F1; +- visual-code grounding path hit@k; +- visual-code grounding symbol hit@k. + +Gold instances use this shape: + +```json +{ + "instances": [ + { + "artifact_path": "docs/architecture.svg", + "gold_entities": [ + {"name": "IndexCompiler", "type": "component"} + ], + "gold_bindings": [ + { + "entity_name": "IndexCompiler", + "source_path": "codenib/compiler/index_compiler.py", + "symbol": "IndexCompiler" + } + ] + } + ] +} +``` + ## Why evidence stays server-side Media generation may use bounded source snippets inside provider prompts. Those diff --git a/test/wiki/test_media_eval.py b/test/wiki/test_media_eval.py new file mode 100644 index 00000000..79029107 --- /dev/null +++ b/test/wiki/test_media_eval.py @@ -0,0 +1,111 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from codenib.wiki.media_eval import ( + evaluate_mmwiki_predictions, + evaluate_visual_code_grounding, + evaluate_visual_fact_extraction, +) + + +def _facts(): + return { + "facts": [ + { + "artifact_path": "docs/architecture.svg", + "entities": [ + {"name": "IndexCompiler", "type": "component"}, + {"name": "VectorStore", "type": "component"}, + {"name": "Noise", "type": "component"}, + ], + } + ] + } + + +def _grounding(): + return { + "bindings": [ + { + "artifact_path": "docs/architecture.svg", + "entity_name": "IndexCompiler", + "source_path": "codenib/compiler/index_compiler.py", + "symbol": "IndexCompiler", + "score": 1.0, + }, + { + "artifact_path": "docs/architecture.svg", + "entity_name": "VectorStore", + "source_path": "codenib/index/embedding/vector_store.py", + "symbol": "VectorStore", + "score": 0.9, + }, + ] + } + + +def _gold(): + return { + "instances": [ + { + "artifact_path": "docs/architecture.svg", + "gold_entities": [ + {"name": "IndexCompiler", "type": "component"}, + {"name": "VectorStore", "type": "component"}, + ], + "gold_bindings": [ + { + "entity_name": "IndexCompiler", + "source_path": "codenib/compiler/index_compiler.py", + "symbol": "IndexCompiler", + }, + { + "entity_name": "VectorStore", + "source_path": "codenib/index/embedding/vector_store.py", + "symbol": "VectorStore", + }, + ], + } + ] + } + + +def test_evaluate_visual_fact_extraction_reports_precision_recall_f1(): + metrics = evaluate_visual_fact_extraction(_facts(), _gold()) + + assert metrics["entity_true_positive"] == 2 + assert metrics["entity_predicted"] == 3 + assert metrics["entity_gold"] == 2 + assert round(metrics["entity_precision"], 3) == 0.667 + assert metrics["entity_recall"] == 1.0 + assert round(metrics["entity_f1"], 3) == 0.8 + + +def test_evaluate_visual_code_grounding_reports_hits_at_k(): + metrics = evaluate_visual_code_grounding(_grounding(), _gold(), k=1) + + assert metrics["binding_count"] == 2 + assert metrics["path_hit_at_k"] == 1.0 + assert metrics["symbol_hit_at_k"] == 1.0 + assert metrics["path_hits"] == 2 + assert metrics["symbol_hits"] == 2 + + +def test_evaluate_mmwiki_predictions_combines_tasks(): + metrics = evaluate_mmwiki_predictions(_facts(), _grounding(), _gold(), k=1) + + assert metrics["task"] == "mmwiki" + assert metrics["artifact_count"] == 1 + assert metrics["visual_fact_extraction"]["entity_true_positive"] == 2 + assert metrics["visual_code_grounding"]["symbol_hit_at_k"] == 1.0 + + +def test_evaluate_visual_code_grounding_handles_missing_predictions(): + metrics = evaluate_visual_code_grounding({"bindings": []}, _gold(), k=5) + + assert metrics["path_hit_at_k"] == 0.0 + assert metrics["symbol_hit_at_k"] == 0.0 + assert metrics["binding_count"] == 2 diff --git a/test/wiki/test_media_incremental.py b/test/wiki/test_media_incremental.py new file mode 100644 index 00000000..654e21de --- /dev/null +++ b/test/wiki/test_media_incremental.py @@ -0,0 +1,144 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from codenib.wiki.media_incremental import ( + diff_media_manifests, + merge_incremental_visual_facts, + plan_incremental_visual_fact_update, +) + + +def _artifact(path, sha): + return { + "path": path, + "sha256": sha, + "role_hint": "repository_image", + "mime_type": "image/png", + } + + +def _fact(path, sha, name="Component"): + return { + "artifact_path": path, + "artifact_sha256": sha, + "role_hint": "repository_image", + "extractor": "local/metadata", + "entities": [ + { + "name": name, + "type": "component", + "evidence": path, + "confidence": 0.5, + "grounding_candidates": [name], + } + ], + "relations": [], + "claims": [], + "metadata": {}, + } + + +def test_diff_media_manifests_reports_added_removed_changed_unchanged(): + previous = { + "manifest_sha256": "previous", + "artifacts": [ + _artifact("unchanged.png", "same"), + _artifact("changed.png", "old"), + _artifact("removed.png", "gone"), + ], + } + current = { + "manifest_sha256": "current", + "artifacts": [ + _artifact("unchanged.png", "same"), + _artifact("changed.png", "new"), + _artifact("added.png", "fresh"), + ], + } + + diff = diff_media_manifests(previous, current) + + assert diff["counts"] == { + "added": 1, + "removed": 1, + "changed": 1, + "unchanged": 1, + } + statuses = {change["path"]: change["status"] for change in diff["changes"]} + assert statuses == { + "added.png": "added", + "changed.png": "changed", + "removed.png": "removed", + "unchanged.png": "unchanged", + } + + +def test_plan_incremental_visual_fact_update_reuses_only_matching_facts(): + previous_media = { + "manifest_sha256": "previous-media", + "artifacts": [ + _artifact("unchanged.png", "same"), + _artifact("changed.png", "old"), + _artifact("removed.png", "gone"), + ], + } + current_media = { + "manifest_sha256": "current-media", + "artifacts": [ + _artifact("unchanged.png", "same"), + _artifact("changed.png", "new"), + _artifact("added.png", "fresh"), + ], + } + previous_facts = { + "manifest_sha256": "previous-facts", + "facts": [ + _fact("unchanged.png", "same"), + _fact("changed.png", "old"), + _fact("removed.png", "gone"), + ], + } + + plan = plan_incremental_visual_fact_update( + previous_media, + current_media, + previous_facts, + ) + + assert [fact["artifact_path"] for fact in plan["reusable_fact_packs"]] == [ + "unchanged.png" + ] + assert plan["extract_artifact_paths"] == ["added.png", "changed.png"] + assert plan["removed_artifact_paths"] == ["removed.png"] + + +def test_merge_incremental_visual_facts_keeps_current_artifacts_only(): + current_media = { + "manifest_sha256": "current-media", + "artifacts": [ + _artifact("unchanged.png", "same"), + _artifact("added.png", "fresh"), + ], + } + + merged = merge_incremental_visual_facts( + current_media, + reusable_fact_packs=[ + _fact("unchanged.png", "same", name="Reused"), + _fact("removed.png", "gone", name="Removed"), + _fact("changed.png", "old", name="Stale"), + ], + new_fact_packs=[_fact("added.png", "fresh", name="New")], + ) + + assert merged["schema"] == "codenib.media-facts.v1" + assert merged["media_manifest_sha256"] == "current-media" + assert merged["fact_count"] == 2 + assert [fact["artifact_path"] for fact in merged["facts"]] == [ + "added.png", + "unchanged.png", + ] + assert merged["manifest_sha256"]