From 76d440fbaf18c6f4c5b6213878434e79bed7b4ea Mon Sep 17 00:00:00 2001 From: Crespin Alexandre <114501859+kunail0804@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:58:12 +0200 Subject: [PATCH] feat(resolver): implement the deterministic Illustrious Resolver (#12) (#56) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add compiler/resolver/illustrious_resolver.py: resolve_scene(scene, knowledge_base, config, logger=None) -> CompilerResult, per MASTER_SPEC §17. No LLM; no concept or tag is invented. - normalize_concept (§17.3): NFKC + lowercase, unifying underscores/spaces so natural concepts, snake_case ids, and spaced aliases share one lookup key. The resolver builds its own normalized index over the Knowledge Base. - Per-concept pipeline in scene discovery order (§17.8): normalize -> direct alias resolution (no chains; skipped when allow_aliases is false) -> canonical lookup -> recursive expansion -> tag generation. - Expansion (§17.5): own tags first, then depth-first in list order per expand target, inserted immediately after; capped by max_expansion_depth; skipped when expansion_enabled is false; defensive path-based cycle detection -> SC0003. - Deduplication after expansion keeps the first occurrence (SC0007). - Unknown concept -> SC0001 warning (continue); deprecated -> SC0006 (continue); invalid category -> SC0008 (stop); circular expansion -> SC0003 (stop). - Every ResolvedTag keeps its entry's category and full traceability (source_concept = originating scene concept, knowledge_base_entry = generating entry). Verbose logging of each decision. Tests (no Ollama): normalization, alias resolution + disable, expansion order, nested/depth-capped/disabled expansion, traceability, dedup, unknown/deprecated, circular/invalid-category stops, category preservation, and determinism. Co-authored-by: kunail0804 Co-authored-by: Claude Opus 4.8 --- compiler/resolver/illustrious_resolver.py | 250 ++++++++++++++++++++ tests/resolver/test_illustrious_resolver.py | 229 ++++++++++++++++++ 2 files changed, 479 insertions(+) create mode 100644 compiler/resolver/illustrious_resolver.py create mode 100644 tests/resolver/test_illustrious_resolver.py diff --git a/compiler/resolver/illustrious_resolver.py b/compiler/resolver/illustrious_resolver.py new file mode 100644 index 0000000..808fdda --- /dev/null +++ b/compiler/resolver/illustrious_resolver.py @@ -0,0 +1,250 @@ +"""The Illustrious Resolver (MASTER_SPEC §17). + +Deterministically converts the semantic concepts of a validated Scene into +Illustrious Resolved Tags using the Knowledge Base. No language model is involved +and no concept or tag is ever invented. + +Per-concept pipeline (§17.2): normalize -> alias resolution (direct only) -> +canonical lookup -> expansion -> tag generation. Concepts are processed in scene +discovery order (§17.8) and every generated tag keeps the category of its +Knowledge Base Entry. +""" + +from __future__ import annotations + +import unicodedata +from collections.abc import Iterator + +from compiler.common.categories import is_valid_category +from compiler.common.config import Config +from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry +from compiler.common.log import StructuredLogger +from compiler.common.message_codes import message +from compiler.common.result import CompilerResult, Message +from schemas.models import ResolvedTag, Scene + +_CHARACTER_CONCEPT_FIELDS = ( + "identity", + "appearance", + "clothing", + "accessories", + "pose", + "expression", + "actions", +) + + +def normalize_concept(name: str) -> str: + """Normalize a concept for lookup (§17.3). + + Applies Unicode (NFKC) normalization and lowercasing, and unifies underscores + with spaces before collapsing whitespace, so natural-language concepts + ("holding hands"), snake_case ids ("holding_hands"), and spaced aliases all + map to the same key. + """ + text = unicodedata.normalize("NFKC", name).replace("_", " ").lower() + return " ".join(text.split()) + + +def resolve_scene( + scene: Scene, + knowledge_base: KnowledgeBase, + config: Config, + logger: StructuredLogger | None = None, +) -> CompilerResult: + """Resolve a validated Scene into an ordered tuple of Resolved Tags. + + Returns: + A CompilerResult whose ``data`` is the tuple of :class:`ResolvedTag` on + success, or ``None`` with error messages when an error condition + (circular expansion SC0003, invalid category SC0008) stops resolution. + """ + index = _build_normalized_index(knowledge_base) + resolver = _ConceptResolver(knowledge_base, index, config, logger) + + tags: list[ResolvedTag] = [] + warnings: list[Message] = [] + errors: list[Message] = [] + + for concept_name in _iter_scene_concepts(scene): + resolved = resolver.resolve(concept_name) + tags.extend(resolved.tags) + warnings.extend(resolved.warnings) + errors.extend(resolved.errors) + + if errors: + result = CompilerResult() + for error in errors: + result = result.add_error(error) + return result + + deduped, dedup_warnings = _deduplicate(tags) + warnings.extend(dedup_warnings) + + result = CompilerResult(data=tuple(deduped)) + for warning in warnings: + result = result.add_warning(warning) + if logger is not None: + logger.basic("scene_resolved", tags=len(deduped), warnings=len(warnings)) + return result + + +def _build_normalized_index( + knowledge_base: KnowledgeBase, +) -> dict[str, KnowledgeBaseEntry]: + """Map every normalized canonical id and alias to its entry (aliases direct).""" + index: dict[str, KnowledgeBaseEntry] = {} + for entry in knowledge_base.by_id.values(): + index[normalize_concept(entry.id)] = entry + for alias in entry.aliases: + index[normalize_concept(alias)] = entry + return index + + +def _iter_scene_concepts(scene: Scene) -> Iterator[str]: + """Yield concept names in scene discovery order (§17.8).""" + for character in scene.characters: + for field in _CHARACTER_CONCEPT_FIELDS: + for concept in getattr(character, field): + yield concept.name + for interaction in scene.interactions: + yield interaction.concept + for field in ("objects", "environment", "camera", "lighting"): + for concept in getattr(scene, field): + yield concept.name + + +class _Resolution: + """The outcome of resolving one concept.""" + + __slots__ = ("tags", "warnings", "errors") + + def __init__(self) -> None: + self.tags: list[ResolvedTag] = [] + self.warnings: list[Message] = [] + self.errors: list[Message] = [] + + +class _ConceptResolver: + """Resolves a single concept (with expansion) against a normalized index.""" + + def __init__( + self, + knowledge_base: KnowledgeBase, + index: dict[str, KnowledgeBaseEntry], + config: Config, + logger: StructuredLogger | None, + ) -> None: + self._kb = knowledge_base + self._index = index + self._allow_aliases = config.resolver.allow_aliases + self._expansion_enabled = config.resolver.expansion_enabled + self._max_depth = config.resolver.max_expansion_depth + self._logger = logger + + def resolve(self, concept_name: str) -> _Resolution: + outcome = _Resolution() + key = normalize_concept(concept_name) + entry = self._lookup(key) + if entry is None: + outcome.warnings.append( + message( + "SC0001", + f"Unknown concept '{concept_name}' has no Knowledge Base entry; ignored.", + concept=concept_name, + ) + ) + if self._logger is not None: + self._logger.verbose("concept_unknown", concept=concept_name, normalized=key) + return outcome + + self._expand(entry, concept_name, depth=0, path=(), outcome=outcome) + return outcome + + def _lookup(self, key: str) -> KnowledgeBaseEntry | None: + entry = self._index.get(key) + if entry is None: + return None + # A key that matches an alias but not the entry's own id is an alias hit. + if not self._allow_aliases and key != normalize_concept(entry.id): + return None + return entry + + def _expand( + self, + entry: KnowledgeBaseEntry, + source_concept: str, + depth: int, + path: tuple[str, ...], + outcome: _Resolution, + ) -> None: + if entry.id in path: + outcome.errors.append( + message( + "SC0003", + f"Circular expansion detected at '{entry.id}' (path: {' -> '.join(path)}).", + entry=entry.id, + ) + ) + return + if not is_valid_category(entry.category): + outcome.errors.append( + message( + "SC0008", + f"Entry '{entry.id}' has invalid category '{entry.category}'.", + entry=entry.id, + ) + ) + return + if entry.deprecated: + outcome.warnings.append( + message( + "SC0006", + f"Concept '{entry.id}' is deprecated; migration is encouraged.", + entry=entry.id, + ) + ) + + for tag in entry.tags: + outcome.tags.append( + ResolvedTag( + tag=tag, + category=entry.category, + source_concept=source_concept, + knowledge_base_entry=entry.id, + ) + ) + if self._logger is not None: + self._logger.verbose( + "concept_resolved", + concept=source_concept, + canonical=entry.id, + tags=list(entry.tags), + ) + + if not self._expansion_enabled or depth >= self._max_depth: + return + for target_id in entry.expand: + target = self._kb.get(target_id) + if target is not None: + self._expand(target, source_concept, depth + 1, (*path, entry.id), outcome) + + +def _deduplicate(tags: list[ResolvedTag]) -> tuple[list[ResolvedTag], list[Message]]: + """Remove duplicate tags after expansion, keeping the first occurrence (§17.7).""" + seen: set[str] = set() + kept: list[ResolvedTag] = [] + warnings: list[Message] = [] + for resolved in tags: + if resolved.tag in seen: + warnings.append( + message( + "SC0007", + f"Duplicate tag '{resolved.tag}' removed after expansion.", + tag=resolved.tag, + ) + ) + continue + seen.add(resolved.tag) + kept.append(resolved) + return kept, warnings diff --git a/tests/resolver/test_illustrious_resolver.py b/tests/resolver/test_illustrious_resolver.py new file mode 100644 index 0000000..a3d63f8 --- /dev/null +++ b/tests/resolver/test_illustrious_resolver.py @@ -0,0 +1,229 @@ +"""Tests for the Illustrious Resolver (issue #12, MASTER_SPEC §17). No Ollama.""" + +from __future__ import annotations + +from compiler.common.config import Config +from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry +from compiler.resolver.illustrious_resolver import resolve_scene +from schemas.models import Scene + + +def kb(*entries: KnowledgeBaseEntry) -> KnowledgeBase: + return KnowledgeBase(list(entries)) + + +def entry(id_, tags, category="hair", aliases=(), expand=(), deprecated=False): + return KnowledgeBaseEntry( + id=id_, + tags=tuple(tags), + category=category, + aliases=tuple(aliases), + expand=tuple(expand), + deprecated=deprecated, + ) + + +def scene_with(identity=(), **fields) -> Scene: + character = { + "id": 0, + "identity": list(identity), + "appearance": list(fields.get("appearance", [])), + "clothing": list(fields.get("clothing", [])), + "accessories": [], + "pose": [], + "expression": [], + "actions": [], + } + return Scene.from_json( + { + "characters": [character], + "interactions": [], + "objects": list(fields.get("objects", [])), + "environment": list(fields.get("environment", [])), + "camera": [], + "lighting": [], + "metadata": {"schema_version": "1.0"}, + } + ) + + +def default_config() -> Config: + return Config() + + +def tags(result) -> list[str]: + return [t.tag for t in result.data] + + +def codes(messages) -> list[str]: + return [m.code for m in messages] + + +# --- normalization --------------------------------------------------------- + + +def test_normalization_matches_snake_case_id() -> None: + knowledge = kb(entry("long_hair", ["long hair"])) + for written in ["long hair", "Long Hair", " LONG_HAIR ", "long_hair"]: + result = resolve_scene(scene_with(identity=[written]), knowledge, default_config()) + assert tags(result) == ["long hair"], written + + +# --- alias resolution ------------------------------------------------------ + + +def test_alias_resolves_directly() -> None: + knowledge = kb(entry("female", ["1girl"], category="character", aliases=["girl", "woman"])) + result = resolve_scene(scene_with(identity=["girl"]), knowledge, default_config()) + assert tags(result) == ["1girl"] + assert result.data[0].knowledge_base_entry == "female" + assert result.data[0].source_concept == "girl" + + +def test_alias_ignored_when_disabled() -> None: + knowledge = kb(entry("female", ["1girl"], category="character", aliases=["girl"])) + config = Config.from_json({"resolver": {"allow_aliases": False}}) + result = resolve_scene(scene_with(identity=["girl"]), knowledge, config) + assert tags(result) == [] + assert "SC0001" in codes(result.warnings) + + +# --- expansion order ------------------------------------------------------- + + +def test_expansion_order_is_normative() -> None: + knowledge = kb( + entry( + "school_uniform", ["school uniform"], category="clothing", expand=["blazer", "skirt"] + ), + entry("blazer", ["blazer"], category="clothing"), + entry("skirt", ["pleated skirt"], category="clothing"), + ) + result = resolve_scene(scene_with(clothing=["school uniform"]), knowledge, default_config()) + assert tags(result) == ["school uniform", "blazer", "pleated skirt"] + + +def test_nested_expansion_is_depth_first() -> None: + knowledge = kb( + entry("a", ["ta"], expand=["b"]), + entry("b", ["tb"], expand=["c"]), + entry("c", ["tc"]), + ) + result = resolve_scene(scene_with(identity=["a"]), knowledge, default_config()) + assert tags(result) == ["ta", "tb", "tc"] + + +def test_expansion_disabled_emits_only_own_tags() -> None: + knowledge = kb( + entry("a", ["ta"], expand=["b"]), + entry("b", ["tb"]), + ) + config = Config.from_json({"resolver": {"expansion_enabled": False}}) + result = resolve_scene(scene_with(identity=["a"]), knowledge, config) + assert tags(result) == ["ta"] + + +def test_max_expansion_depth_caps_recursion() -> None: + # depth 1 allows one expansion level (a -> b) but not the next (b -> c). + knowledge = kb( + entry("a", ["ta"], expand=["b"]), + entry("b", ["tb"], expand=["c"]), + entry("c", ["tc"]), + ) + config = Config.from_json({"resolver": {"max_expansion_depth": 1}}) + result = resolve_scene(scene_with(identity=["a"]), knowledge, config) + assert tags(result) == ["ta", "tb"] + + +def test_expanded_tag_traceability() -> None: + knowledge = kb( + entry("school_uniform", ["school uniform"], category="clothing", expand=["blazer"]), + entry("blazer", ["blazer"], category="clothing"), + ) + result = resolve_scene(scene_with(clothing=["school uniform"]), knowledge, default_config()) + blazer_tag = result.data[1] + assert blazer_tag.tag == "blazer" + assert blazer_tag.knowledge_base_entry == "blazer" + assert blazer_tag.source_concept == "school uniform" # originating scene concept + + +# --- deduplication --------------------------------------------------------- + + +def test_duplicate_tags_removed_keeping_first() -> None: + knowledge = kb( + entry("a", ["shared"]), + entry("b", ["shared"]), + ) + result = resolve_scene(scene_with(identity=["a", "b"]), knowledge, default_config()) + assert tags(result) == ["shared"] + assert "SC0007" in codes(result.warnings) + + +# --- unknown / deprecated -------------------------------------------------- + + +def test_unknown_concept_warns_and_continues() -> None: + knowledge = kb(entry("female", ["1girl"], category="character")) + result = resolve_scene(scene_with(identity=["dragon", "female"]), knowledge, default_config()) + assert tags(result) == ["1girl"] + assert "SC0001" in codes(result.warnings) + + +def test_deprecated_concept_warns_but_emits_tags() -> None: + knowledge = kb(entry("old_style", ["old style"], deprecated=True)) + result = resolve_scene(scene_with(identity=["old_style"]), knowledge, default_config()) + assert tags(result) == ["old style"] + assert "SC0006" in codes(result.warnings) + + +# --- error conditions stop ------------------------------------------------- + + +def test_circular_expansion_stops_with_sc0003() -> None: + knowledge = kb( + entry("a", ["ta"], expand=["b"]), + entry("b", ["tb"], expand=["a"]), + ) + result = resolve_scene(scene_with(identity=["a"]), knowledge, default_config()) + assert not result.success + assert result.data is None + assert "SC0003" in codes(result.errors) + + +def test_invalid_category_stops_with_sc0008() -> None: + knowledge = kb(entry("a", ["ta"], category="not_a_category")) + result = resolve_scene(scene_with(identity=["a"]), knowledge, default_config()) + assert not result.success + assert "SC0008" in codes(result.errors) + + +# --- category preservation & determinism ----------------------------------- + + +def test_tag_keeps_entry_category() -> None: + knowledge = kb(entry("blue_eyes", ["blue eyes"], category="eyes")) + result = resolve_scene(scene_with(appearance=["blue_eyes"]), knowledge, default_config()) + assert result.data[0].category == "eyes" + + +def test_discovery_order_preserved_across_scene() -> None: + knowledge = kb( + entry("female", ["1girl"], category="character"), + entry("forest", ["forest"], category="environment"), + ) + scene = scene_with(identity=["female"], environment=["forest"]) + result = resolve_scene(scene, knowledge, default_config()) + assert tags(result) == ["1girl", "forest"] + + +def test_output_is_deterministic() -> None: + knowledge = kb( + entry("school_uniform", ["school uniform"], category="clothing", expand=["blazer"]), + entry("blazer", ["blazer"], category="clothing"), + ) + scene = scene_with(clothing=["school uniform", "school uniform"]) + a = resolve_scene(scene, knowledge, default_config()) + b = resolve_scene(scene, knowledge, default_config()) + assert tags(a) == tags(b) + assert codes(a.warnings) == codes(b.warnings)