benchmarks/uko_analyzers_bench.py

@@ -0,0 +1,157 @@
+"""ASV benchmarks for UKO Analyzer Plugin Framework.
+
+Measures the performance of:
+- UKOTriple construction overhead
+- AnalyzerRegistry registration and lookup
+- PythonAnalyzer.analyze() on various source sizes
+- MarkdownAnalyzer.analyze() on various document sizes
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Force-reload so ASV picks up the source tree version.
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.domain.models.acms.analyzers import (  # noqa: E402
+    AnalyzerRegistry,
+    UKOTriple,
+)
+from cleveragents.domain.models.acms.markdown_analyzer import (  # noqa: E402
+    MarkdownAnalyzer,
+)
+from cleveragents.domain.models.acms.python_analyzer import (  # noqa: E402
+    PythonAnalyzer,
+)
+
+# ---------------------------------------------------------------------------
+# Sample content
+# ---------------------------------------------------------------------------
+
+_SMALL_PYTHON = 'import os\n\ndef hello():\n    """Say hello."""\n    pass\n'
+
+_MEDIUM_PYTHON = (
+    '"""Module docstring."""\n\nimport os\nfrom pathlib import Path\n\n'
+    + "\n\n".join(
+        f'class Cls{i}:\n    """Class {i}."""\n'
+        f"    def method_{i}(self):\n"
+        f'        """Method {i}."""\n        pass\n'
+        for i in range(20)
+    )
+)
+
+_SMALL_MARKDOWN = "# Hello\n\nWorld\n"
+
+_MEDIUM_MARKDOWN = "# Document\n\n" + "\n".join(
+    f"## Section {i}\n\nParagraph text for section {i}.\n\n"
+    f"```python\nprint({i})\n```\n\n"
+    f"[Link {i}](https://example.com/{i})\n"
+    for i in range(20)
+)
+
+
+# ---------------------------------------------------------------------------
+# UKOTriple benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TripleConstructionSuite:
+    """Benchmark UKOTriple object creation overhead."""
+
+    timeout = 60
+
+    def time_create_object_property_triple(self) -> None:
+        UKOTriple(
+            subject_uri="uko://code/module/foo",
+            predicate="rdf:type",
+            object_uri="uko-code:Module",
+        )
+
+    def time_create_data_property_triple(self) -> None:
+        UKOTriple(
+            subject_uri="uko://code/class/Bar",
+            predicate="rdfs:label",
+            object_value="Bar",
+            confidence=0.9,
+        )
+
+
+# ---------------------------------------------------------------------------
+# AnalyzerRegistry benchmarks
+# ---------------------------------------------------------------------------
+
+
+class RegistrySuite:
+    """Benchmark AnalyzerRegistry operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.registry = AnalyzerRegistry()
+        self.py = PythonAnalyzer()
+        self.md = MarkdownAnalyzer()
+        self.registry.register(self.py)
+        self.registry.register(self.md)
+
+    def time_lookup_registered_extension(self) -> None:
+        self.registry.get_for_extension(".py")
+
+    def time_lookup_unregistered_extension(self) -> None:
+        self.registry.get_for_extension(".rs")
+
+    def time_list_extensions(self) -> None:
+        self.registry.list_extensions()
+
+    def time_list_analyzers(self) -> None:
+        self.registry.list_analyzers()
+
+
+# ---------------------------------------------------------------------------
+# PythonAnalyzer benchmarks
+# ---------------------------------------------------------------------------
+
+
+class PythonAnalyzerSuite:
+    """Benchmark PythonAnalyzer.analyze() performance."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.analyzer = PythonAnalyzer()
+
+    def time_analyze_small_python(self) -> None:
+        self.analyzer.analyze(_SMALL_PYTHON, "src/small.py")
+
+    def time_analyze_medium_python(self) -> None:
+        self.analyzer.analyze(_MEDIUM_PYTHON, "src/medium.py")
+
+
+# ---------------------------------------------------------------------------
+# MarkdownAnalyzer benchmarks
+# ---------------------------------------------------------------------------
+
+
+class MarkdownAnalyzerSuite:
+    """Benchmark MarkdownAnalyzer.analyze() performance."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.analyzer = MarkdownAnalyzer()
+
+    def time_analyze_small_markdown(self) -> None:
+        self.analyzer.analyze(_SMALL_MARKDOWN, "docs/small.md")
+
+    def time_analyze_medium_markdown(self) -> None:
+        self.analyzer.analyze(_MEDIUM_MARKDOWN, "docs/medium.md")

features/steps/uko_analyzers_steps.py

@@ -0,0 +1,551 @@
+"""Step definitions for the UKO Analyzer Plugin Framework feature."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from behave import given, then, when
+
+from cleveragents.domain.models.acms.analyzers import (
+    AnalyzerProtocol,
+    AnalyzerRegistry,
+    UKOTriple,
+)
+from cleveragents.domain.models.acms.markdown_analyzer import MarkdownAnalyzer
+from cleveragents.domain.models.acms.python_analyzer import PythonAnalyzer
+
+__all__: list[str] = []
+
+
+# ---------------------------------------------------------------------------
+# UKOTriple steps
+# ---------------------------------------------------------------------------
+
+
+@given('a UKOTriple with subject "{subj}" predicate "{pred}" and object_uri "{obj}"')
+def step_given_triple_with_obj_uri(
+    context: Any, subj: str, pred: str, obj: str
+) -> None:
+    context.triple = UKOTriple(subject_uri=subj, predicate=pred, object_uri=obj)
+
+
+@given('a UKOTriple with subject "{subj}" predicate "{pred}" and object_value "{val}"')
+def step_given_triple_with_obj_value(
+    context: Any, subj: str, pred: str, val: str
+) -> None:
+    context.triple = UKOTriple(subject_uri=subj, predicate=pred, object_value=val)
+
+
+@then('the triple subject_uri should be "{expected}"')
+def step_then_triple_subject(context: Any, expected: str) -> None:
+    assert context.triple.subject_uri == expected
+
+
+@then('the triple predicate should be "{expected}"')
+def step_then_triple_predicate(context: Any, expected: str) -> None:
+    assert context.triple.predicate == expected
+
+
+@then('the triple object_uri should be "{expected}"')
+def step_then_triple_object_uri(context: Any, expected: str) -> None:
+    assert context.triple.object_uri == expected
+
+
+@then('the triple object_value should be "{expected}"')
+def step_then_triple_object_value(context: Any, expected: str) -> None:
+    assert context.triple.object_value == expected
+
+
+@then("the triple confidence should be {expected:g}")
+def step_then_triple_confidence(context: Any, expected: float) -> None:
+    assert context.triple.confidence == expected
+
+
+@then("creating a UKOTriple with empty subject_uri should raise ValueError")
+def step_then_triple_empty_subject(context: Any) -> None:
+    try:
+        UKOTriple(subject_uri="", predicate="rdf:type", object_uri="uko:X")
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+@then("creating a UKOTriple with empty predicate should raise ValueError")
+def step_then_triple_empty_predicate(context: Any) -> None:
+    try:
+        UKOTriple(subject_uri="uko://a", predicate="", object_uri="uko:X")
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+@then("creating a UKOTriple with confidence {score:g} should raise ValueError")
+def step_then_triple_bad_confidence(context: Any, score: float) -> None:
+    try:
+        UKOTriple(
+            subject_uri="uko://a",
+            predicate="rdf:type",
+            object_uri="uko:X",
+            confidence=score,
+        )
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+@then("modifying the triple confidence should raise an error")
+def step_then_triple_immutable(context: Any) -> None:
+    try:
+        context.triple.confidence = 0.5  # type: ignore[misc]
+        raise AssertionError("Expected frozen error")
+    except (TypeError, ValueError, AttributeError):
+        pass
+
+
+# ---------------------------------------------------------------------------
+# AnalyzerRegistry steps
+# ---------------------------------------------------------------------------
+
+
+@given("an AnalyzerRegistry with a PythonAnalyzer registered")
+def step_given_registry_with_python(context: Any) -> None:
+    context.registry = AnalyzerRegistry()
+    context.registry.register(PythonAnalyzer())
+
+
+@given("an empty AnalyzerRegistry")
+def step_given_empty_registry(context: Any) -> None:
+    context.registry = AnalyzerRegistry()
+
+
+@given("an AnalyzerRegistry with both Python and Markdown analyzers")
+def step_given_registry_with_both(context: Any) -> None:
+    context.registry = AnalyzerRegistry()
+    context.registry.register(PythonAnalyzer())
+    context.registry.register(MarkdownAnalyzer())
+
+
+@when('I look up the analyzer for extension "{ext}"')
+def step_when_lookup_extension(context: Any, ext: str) -> None:
+    context.lookup_result = context.registry.get_for_extension(ext)
+
+
+@when('I register another analyzer for "{ext}"')
+def step_when_register_another(context: Any, ext: str) -> None:
+    """Register a second MarkdownAnalyzer that also claims .py."""
+
+    class _DummyAnalyzer:
+        @property
+        def supported_extensions(self) -> frozenset[str]:
+            return frozenset({ext})
+
+        @property
+        def domain(self) -> str:
+            return "dummy"
+
+        def analyze(self, content: str, resource_uri: str) -> list[UKOTriple]:
+            return []
+
+    context.registry.register(_DummyAnalyzer())
+
+
+@then('the analyzer domain should be "{expected}"')
+def step_then_analyzer_domain(context: Any, expected: str) -> None:
+    assert context.lookup_result is not None
+    assert context.lookup_result.domain == expected
+
+
+@then("the lookup result should be None")
+def step_then_lookup_none(context: Any) -> None:
+    assert context.lookup_result is None
+
+
+@then("registering a plain object should raise TypeError")
+def step_then_register_plain_object(context: Any) -> None:
+    try:
+        context.registry.register("not an analyzer")  # type: ignore[arg-type]
+        raise AssertionError("Expected TypeError")
+    except TypeError:
+        pass
+
+
+@then("registering an analyzer with no extensions should raise ValueError")
+def step_then_register_no_extensions(context: Any) -> None:
+    class _Empty:
+        @property
+        def supported_extensions(self) -> frozenset[str]:
+            return frozenset()
+
+        @property
+        def domain(self) -> str:
+            return "empty"
+
+        def analyze(self, content: str, resource_uri: str) -> list[UKOTriple]:
+            return []
+
+    try:
+        context.registry.register(_Empty())
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+@then('the registry should list extensions including "{ext1}" and "{ext2}"')
+def step_then_list_extensions(context: Any, ext1: str, ext2: str) -> None:
+    exts = context.registry.list_extensions()
+    assert ext1 in exts, f"{ext1} not in {exts}"
+    assert ext2 in exts, f"{ext2} not in {exts}"
+
+
+@then("the registry length should be {expected:d}")
+def step_then_registry_length(context: Any, expected: int) -> None:
+    assert len(context.registry) == expected
+
+
+# ---------------------------------------------------------------------------
+# PythonAnalyzer steps
+# ---------------------------------------------------------------------------
+
+
+@given("a PythonAnalyzer instance")
+def step_given_python_analyzer(context: Any) -> None:
+    context.analyzer = PythonAnalyzer()
+
+
+@then("it should satisfy the AnalyzerProtocol")
+def step_then_satisfies_protocol(context: Any) -> None:
+    assert isinstance(context.analyzer, AnalyzerProtocol)
+
+
+@then('its supported extensions should include "{ext1}" and "{ext2}"')
+def step_then_supported_extensions(context: Any, ext1: str, ext2: str) -> None:
+    exts = context.analyzer.supported_extensions
+    assert ext1 in exts, f"{ext1} not in {exts}"
+    assert ext2 in exts, f"{ext2} not in {exts}"
+
+
+@when('I analyze Python content "{content}" with resource URI "{uri}"')
+def step_when_analyze_python(context: Any, content: str, uri: str) -> None:
+    context.triples = context.analyzer.analyze(content.replace("\\n", "\n"), uri)
+
+
+@when('I analyze Python content with a class "{cls_name}"')
+def step_when_analyze_python_class(context: Any, cls_name: str) -> None:
+    code = f"class {cls_name}:\n    pass\n"
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@when('I analyze Python content with a function "{func_name}"')
+def step_when_analyze_python_function(context: Any, func_name: str) -> None:
+    code = f"def {func_name}():\n    pass\n"
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@when('I analyze Python content with class "{cls_name}" and method "{method_name}"')
+def step_when_analyze_python_method(
+    context: Any, cls_name: str, method_name: str
+) -> None:
+    code = f"class {cls_name}:\n    def {method_name}(self):\n        pass\n"
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@when('I analyze Python content with import "{mod_name}"')
+def step_when_analyze_python_import(context: Any, mod_name: str) -> None:
+    code = f"import {mod_name}\n"
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@when('I analyze Python content with from-import "{mod_name}"')
+def step_when_analyze_python_from_import(context: Any, mod_name: str) -> None:
+    code = f"from {mod_name} import Path\n"
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@when('I analyze Python content with docstring "{docstring}"')
+def step_when_analyze_python_docstring(context: Any, docstring: str) -> None:
+    code = f'"""{docstring}"""\nx = 1\n'
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@when("I analyze Python content with class docstring")
+def step_when_analyze_python_class_docstring(context: Any) -> None:
+    code = 'class Foo:\n    """Class doc."""\n    pass\n'
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@when("I analyze malformed Python content")
+def step_when_analyze_malformed_python(context: Any) -> None:
+    context.triples = context.analyzer.analyze("def (\n    broken", "src/bad.py")
+
+
+@when('I analyze Python content with a class inheriting from "{base}"')
+def step_when_analyze_python_inheritance(context: Any, base: str) -> None:
+    code = f"class Child({base}):\n    pass\n"
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@then(
+    'the triples should contain a triple with predicate "{pred}" and object_uri "{obj}"'
+)
+def step_then_triples_contain_pred_obj(context: Any, pred: str, obj: str) -> None:
+    found = any(t.predicate == pred and t.object_uri == obj for t in context.triples)
+    assert found, (
+        f"No triple with predicate={pred!r} object_uri={obj!r} in {context.triples}"
+    )
+
+
+@then(
+    'the triples should contain a triple with predicate "{pred}" and object_value "{val}"'
+)
+def step_then_triples_contain_pred_val(context: Any, pred: str, val: str) -> None:
+    found = any(t.predicate == pred and t.object_value == val for t in context.triples)
+    assert found, (
+        f"No triple with predicate={pred!r} object_value={val!r} in {context.triples}"
+    )
+
+
+@then(
+    "the triples should contain a triple with predicate"
+    ' "uko:contains" linking module to class'
+)
+def step_then_triples_contain_module_contains_class(
+    context: Any,
+) -> None:
+    found = any(
+        t.predicate == "uko:contains"
+        and "module" in t.subject_uri
+        and "class" in t.object_uri
+        for t in context.triples
+    )
+    assert found, f"No module-contains-class triple in {context.triples}"
+
+
+@then(
+    'the triples should contain a method triple with predicate "{pred}" and object_value "{val}"'
+)
+def step_then_triples_contain_method(context: Any, pred: str, val: str) -> None:
+    found = any(t.predicate == pred and t.object_value == val for t in context.triples)
+    assert found, (
+        f"No method triple with predicate={pred!r} val={val!r} in {context.triples}"
+    )
+
+
+@then(
+    'the triples should contain a triple with predicate "uko:references" referencing "{mod_name}"'
+)
+def step_then_triples_contain_import_ref(context: Any, mod_name: str) -> None:
+    found = any(
+        t.predicate == "uko:references" and mod_name in t.object_uri
+        for t in context.triples
+    )
+    assert found, f"No import reference to {mod_name!r} in {context.triples}"
+
+
+@then("the triples should contain a docstring triple for the class")
+def step_then_triples_contain_class_docstring(context: Any) -> None:
+    found = any(
+        t.predicate == "uko-doc:hasDocstring" and "class" in t.subject_uri.lower()
+        for t in context.triples
+    )
+    assert found, f"No class docstring triple in {context.triples}"
+
+
+@then("the analyzer result should be an empty triple list")
+def step_then_result_empty(context: Any) -> None:
+    assert context.triples == []
+
+
+@then("analyzing empty content should raise ValueError")
+def step_then_analyze_empty_content(context: Any) -> None:
+    try:
+        context.analyzer.analyze("", "src/test.py")
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+@then("analyzing with empty resource_uri should raise ValueError")
+def step_then_analyze_empty_uri(context: Any) -> None:
+    try:
+        context.analyzer.analyze("x = 1\n", "")
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# MarkdownAnalyzer steps
+# ---------------------------------------------------------------------------
+
+
+@given("a MarkdownAnalyzer instance")
+def step_given_markdown_analyzer(context: Any) -> None:
+    context.analyzer = MarkdownAnalyzer()
+
+
+@when('I analyze Markdown content "{content}" with resource URI "{uri}"')
+def step_when_analyze_markdown(context: Any, content: str, uri: str) -> None:
+    context.triples = context.analyzer.analyze(content.replace("\\n", "\n"), uri)
+
+
+@when('I analyze Markdown content with heading "{heading}"')
+def step_when_analyze_markdown_heading(context: Any, heading: str) -> None:
+    content = f"# {heading}\n\nSome text here.\n"
+    context.triples = context.analyzer.analyze(content, "docs/test.md")
+
+
+@when("I analyze Markdown with nested headings")
+def step_when_analyze_markdown_nested(context: Any) -> None:
+    content = "# Top\n\n## Sub\n\nBody text.\n"
+    context.triples = context.analyzer.analyze(content, "docs/test.md")
+
+
+@when("I analyze Markdown content with a Python code block")
+def step_when_analyze_markdown_codeblock(context: Any) -> None:
+    content = "# Code\n\n```python\nprint('hello')\n```\n"
+    context.triples = context.analyzer.analyze(content, "docs/test.md")
+
+
+@when('I analyze Markdown content with a link to "{url}"')
+def step_when_analyze_markdown_link(context: Any, url: str) -> None:
+    content = f"See [example]({url}) for details.\n"
+    context.triples = context.analyzer.analyze(content, "docs/test.md")
+
+
+@when("I analyze Markdown content with a bare code fence")
+def step_when_analyze_markdown_bare_fence(context: Any) -> None:
+    content = "# Title\n\n```\nsome code\n```\n"
+    context.triples = context.analyzer.analyze(content, "docs/test.md")
+
+
+@then("the triples should contain section containment for nested sections")
+def step_then_nested_containment(context: Any) -> None:
+    # Document should contain Top section, Top section should contain Sub
+    doc_contains = [
+        t
+        for t in context.triples
+        if t.predicate == "uko:contains"
+        and "document" in t.subject_uri
+        and "section" in t.object_uri
+    ]
+    section_contains = [
+        t
+        for t in context.triples
+        if t.predicate == "uko:contains"
+        and "section" in t.subject_uri
+        and "section" in t.object_uri
+    ]
+    assert len(doc_contains) >= 1, f"Expected doc-contains-section, got {doc_contains}"
+    assert len(section_contains) >= 1, (
+        f"Expected section-contains-section, got {section_contains}"
+    )
+
+
+@then("the triples should contain a code block triple")
+def step_then_contains_codeblock(context: Any) -> None:
+    found = any(
+        t.predicate == "rdf:type" and t.object_uri == "uko-code:CodeBlock"
+        for t in context.triples
+    )
+    assert found, f"No code block triple in {context.triples}"
+
+
+@then("analyzing empty markdown content should raise ValueError")
+def step_then_md_analyze_empty_content(context: Any) -> None:
+    try:
+        context.analyzer.analyze("", "docs/test.md")
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+@then("analyzing markdown with empty resource_uri should raise ValueError")
+def step_then_md_analyze_empty_uri(context: Any) -> None:
+    try:
+        context.analyzer.analyze("# Title\n", "")
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Coverage-gap steps
+# ---------------------------------------------------------------------------
+
+
+@when("I look up the analyzer for an empty extension")
+def step_when_lookup_empty_extension(context: Any) -> None:
+    context.lookup_result = context.registry.get_for_extension("")
+
+
+@then("the registry list_analyzers should return {expected:d} analyzers")
+def step_then_list_analyzers(context: Any, expected: int) -> None:
+    result = context.registry.list_analyzers()
+    assert len(result) == expected, f"Expected {expected} analyzers, got {len(result)}"
+
+
+@when("I analyze Python content with a function with docstring")
+def step_when_analyze_python_func_docstring(context: Any) -> None:
+    code = 'def greet():\n    """Say hello."""\n    pass\n'
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@then(
+    'the triples should contain a triple with predicate "uko-doc:hasDocstring" for a function'
+)
+def step_then_triples_func_docstring(context: Any) -> None:
+    found = any(
+        t.predicate == "uko-doc:hasDocstring" and "function" in t.subject_uri
+        for t in context.triples
+    )
+    assert found, f"No function docstring triple in {context.triples}"
+
+
+@when("I analyze Python content with class and method with docstring")
+def step_when_analyze_python_method_docstring(context: Any) -> None:
+    code = (
+        "class Svc:\n"
+        "    def run(self):\n"
+        '        """Execute the service."""\n'
+        "        pass\n"
+    )
+    context.triples = context.analyzer.analyze(code, "src/test.py")
+
+
+@then(
+    'the triples should contain a triple with predicate "uko-doc:hasDocstring" for a method'
+)
+def step_then_triples_method_docstring(context: Any) -> None:
+    found = any(
+        t.predicate == "uko-doc:hasDocstring"
+        and "function" in t.subject_uri
+        and "Svc" in t.subject_uri
+        for t in context.triples
+    )
+    assert found, f"No method docstring triple in {context.triples}"
+
+
+@then('the analyzer domain property should be "{expected}"')
+def step_then_analyzer_domain_prop(context: Any, expected: str) -> None:
+    assert context.analyzer.domain == expected
+
+
+@when("I analyze Markdown with sibling headings at the same level")
+def step_when_analyze_md_sibling_headings(context: Any) -> None:
+    content = "# A\n\n## B\n\n## C\n\nText\n"
+    context.triples = context.analyzer.analyze(content, "docs/test.md")
+
+
+@then("each sibling section should be contained by the document")
+def step_then_sibling_sections_contained_by_doc(context: Any) -> None:
+    # Sections B and C are both at level 2 under A; the second ## pops the
+    # first ## off the stack.  Both should be contained by the # A section
+    # (which is contained by the document).
+    section_contains = [
+        t
+        for t in context.triples
+        if t.predicate == "uko:contains" and "section" in t.object_uri
+    ]
+    assert len(section_contains) >= 3, (
+        f"Expected at least 3 contains-section triples, got {section_contains}"
+    )

features/uko_analyzers.feature

@@ -0,0 +1,238 @@
+Feature: UKO Analyzer Plugin Framework
+  As a developer
+  I want pluggable domain analyzers that produce UKO triples from resources
+  So that the ACMS can index code and documents into the UKO knowledge graph
+
+  # ---- UKOTriple model ----
+
+  Scenario: Create a valid UKOTriple with object_uri
+    Given a UKOTriple with subject "uko://code/module/foo" predicate "rdf:type" and object_uri "uko-code:Module"
+    Then the triple subject_uri should be "uko://code/module/foo"
+    And the triple predicate should be "rdf:type"
+    And the triple object_uri should be "uko-code:Module"
+    And the triple confidence should be 1.0
+
+  Scenario: Create a UKOTriple with object_value
+    Given a UKOTriple with subject "uko://code/class/foo/Bar" predicate "rdfs:label" and object_value "Bar"
+    Then the triple object_value should be "Bar"
+
+  Scenario: UKOTriple rejects empty subject_uri
+    Then creating a UKOTriple with empty subject_uri should raise ValueError
+
+  Scenario: UKOTriple rejects empty predicate
+    Then creating a UKOTriple with empty predicate should raise ValueError
+
+  Scenario: UKOTriple rejects confidence above 1.0
+    Then creating a UKOTriple with confidence 1.5 should raise ValueError
+
+  Scenario: UKOTriple rejects confidence below 0.0
+    Then creating a UKOTriple with confidence -0.1 should raise ValueError
+
+  Scenario: UKOTriple is immutable
+    Given a UKOTriple with subject "uko://a" predicate "rdf:type" and object_uri "uko:B"
+    Then modifying the triple confidence should raise an error
+
+  # ---- AnalyzerRegistry ----
+
+  Scenario: Register and look up an analyzer by extension
+    Given an AnalyzerRegistry with a PythonAnalyzer registered
+    When I look up the analyzer for extension ".py"
+    Then the analyzer domain should be "python"
+
+  Scenario: Registry returns None for unregistered extension
+    Given an empty AnalyzerRegistry
+    When I look up the analyzer for extension ".rs"
+    Then the lookup result should be None
+
+  Scenario: Registry rejects non-protocol objects
+    Given an empty AnalyzerRegistry
+    Then registering a plain object should raise TypeError
+
+  Scenario: Registry rejects analyzer with no extensions
+    Given an empty AnalyzerRegistry
+    Then registering an analyzer with no extensions should raise ValueError
+
+  Scenario: Registry lists all extensions
+    Given an AnalyzerRegistry with both Python and Markdown analyzers
+    Then the registry should list extensions including ".py" and ".md"
+
+  Scenario: Registry counts analyzers
+    Given an AnalyzerRegistry with both Python and Markdown analyzers
+    Then the registry length should be 2
+
+  Scenario: Registry first-registered-wins for duplicate extensions
+    Given an AnalyzerRegistry with a PythonAnalyzer registered
+    When I register another analyzer for ".py"
+    And I look up the analyzer for extension ".py"
+    Then the analyzer domain should be "python"
+
+  # ---- PythonAnalyzer ----
+
+  Scenario: PythonAnalyzer satisfies AnalyzerProtocol
+    Given a PythonAnalyzer instance
+    Then it should satisfy the AnalyzerProtocol
+
+  Scenario: PythonAnalyzer supports .py and .pyi extensions
+    Given a PythonAnalyzer instance
+    Then its supported extensions should include ".py" and ".pyi"
+
+  Scenario: PythonAnalyzer extracts module declaration
+    Given a PythonAnalyzer instance
+    When I analyze Python content "x = 1\n" with resource URI "src/foo.py"
+    Then the triples should contain a triple with predicate "rdf:type" and object_uri "uko-code:Module"
+
+  Scenario: PythonAnalyzer extracts class definition
+    Given a PythonAnalyzer instance
+    When I analyze Python content with a class "MyClass"
+    Then the triples should contain a triple with predicate "rdf:type" and object_uri "uko-py:Class"
+    And the triples should contain a triple with predicate "rdfs:label" and object_value "MyClass"
+
+  Scenario: PythonAnalyzer extracts class containment
+    Given a PythonAnalyzer instance
+    When I analyze Python content with a class "MyClass"
+    Then the triples should contain a triple with predicate "uko:contains" linking module to class
+
+  Scenario: PythonAnalyzer extracts function definition
+    Given a PythonAnalyzer instance
+    When I analyze Python content with a function "do_work"
+    Then the triples should contain a triple with predicate "rdf:type" and object_uri "uko-py:Function"
+    And the triples should contain a triple with predicate "rdfs:label" and object_value "do_work"
+
+  Scenario: PythonAnalyzer extracts method inside class
+    Given a PythonAnalyzer instance
+    When I analyze Python content with class "Cls" and method "run"
+    Then the triples should contain a method triple with predicate "rdfs:label" and object_value "run"
+
+  Scenario: PythonAnalyzer extracts import statement
+    Given a PythonAnalyzer instance
+    When I analyze Python content with import "os"
+    Then the triples should contain a triple with predicate "uko:references" referencing "os"
+
+  Scenario: PythonAnalyzer extracts from-import statement
+    Given a PythonAnalyzer instance
+    When I analyze Python content with from-import "pathlib"
+    Then the triples should contain a triple with predicate "uko:references" referencing "pathlib"
+
+  Scenario: PythonAnalyzer extracts module docstring
+    Given a PythonAnalyzer instance
+    When I analyze Python content with docstring "Module docs."
+    Then the triples should contain a triple with predicate "uko-doc:hasDocstring" and object_value "Module docs."
+
+  Scenario: PythonAnalyzer extracts class docstring
+    Given a PythonAnalyzer instance
+    When I analyze Python content with class docstring
+    Then the triples should contain a docstring triple for the class
+
+  Scenario: PythonAnalyzer handles syntax errors gracefully
+    Given a PythonAnalyzer instance
+    When I analyze malformed Python content
+    Then the analyzer result should be an empty triple list
+
+  Scenario: PythonAnalyzer rejects empty content
+    Given a PythonAnalyzer instance
+    Then analyzing empty content should raise ValueError
+
+  Scenario: PythonAnalyzer rejects empty resource_uri
+    Given a PythonAnalyzer instance
+    Then analyzing with empty resource_uri should raise ValueError
+
+  Scenario: PythonAnalyzer extracts base class inheritance
+    Given a PythonAnalyzer instance
+    When I analyze Python content with a class inheriting from "BaseClass"
+    Then the triples should contain a triple with predicate "uko-py:inheritsFrom" and object_value "BaseClass"
+
+  # ---- MarkdownAnalyzer ----
+
+  Scenario: MarkdownAnalyzer satisfies AnalyzerProtocol
+    Given a MarkdownAnalyzer instance
+    Then it should satisfy the AnalyzerProtocol
+
+  Scenario: MarkdownAnalyzer supports .md and .markdown extensions
+    Given a MarkdownAnalyzer instance
+    Then its supported extensions should include ".md" and ".markdown"
+
+  Scenario: MarkdownAnalyzer extracts document declaration
+    Given a MarkdownAnalyzer instance
+    When I analyze Markdown content "# Hello\nWorld\n" with resource URI "docs/readme.md"
+    Then the triples should contain a triple with predicate "rdf:type" and object_uri "uko-doc:Document"
+
+  Scenario: MarkdownAnalyzer extracts headings as sections
+    Given a MarkdownAnalyzer instance
+    When I analyze Markdown content with heading "Introduction"
+    Then the triples should contain a triple with predicate "rdf:type" and object_uri "uko-doc:Section"
+    And the triples should contain a triple with predicate "uko-doc:headingText" and object_value "Introduction"
+    And the triples should contain a triple with predicate "uko-doc:headingLevel" and object_value "1"
+
+  Scenario: MarkdownAnalyzer extracts nested heading hierarchy
+    Given a MarkdownAnalyzer instance
+    When I analyze Markdown with nested headings
+    Then the triples should contain section containment for nested sections
+
+  Scenario: MarkdownAnalyzer extracts fenced code blocks
+    Given a MarkdownAnalyzer instance
+    When I analyze Markdown content with a Python code block
+    Then the triples should contain a triple with predicate "rdf:type" and object_uri "uko-code:CodeBlock"
+    And the triples should contain a triple with predicate "uko-code:language" and object_value "python"
+
+  Scenario: MarkdownAnalyzer extracts links
+    Given a MarkdownAnalyzer instance
+    When I analyze Markdown content with a link to "https://example.com"
+    Then the triples should contain a triple with predicate "uko:references" and object_value "https://example.com"
+
+  Scenario: MarkdownAnalyzer rejects empty content
+    Given a MarkdownAnalyzer instance
+    Then analyzing empty markdown content should raise ValueError
+
+  Scenario: MarkdownAnalyzer rejects empty resource_uri
+    Given a MarkdownAnalyzer instance
+    Then analyzing markdown with empty resource_uri should raise ValueError
+
+  Scenario: MarkdownAnalyzer handles bare code fences
+    Given a MarkdownAnalyzer instance
+    When I analyze Markdown content with a bare code fence
+    Then the triples should contain a code block triple
+
+  # ---- Coverage gap: AnalyzerRegistry.get_for_extension("") ----
+
+  Scenario: Registry returns None for empty extension
+    Given an AnalyzerRegistry with a PythonAnalyzer registered
+    When I look up the analyzer for an empty extension
+    Then the lookup result should be None
+
+  Scenario: Registry list_analyzers returns all registered analyzers
+    Given an AnalyzerRegistry with both Python and Markdown analyzers
+    Then the registry list_analyzers should return 2 analyzers
+
+  # ---- Coverage gap: PythonAnalyzer function docstring ----
+
+  Scenario: PythonAnalyzer extracts function docstring
+    Given a PythonAnalyzer instance
+    When I analyze Python content with a function with docstring
+    Then the triples should contain a triple with predicate "uko-doc:hasDocstring" for a function
+
+  # ---- Coverage gap: PythonAnalyzer method docstring ----
+
+  Scenario: PythonAnalyzer extracts method docstring
+    Given a PythonAnalyzer instance
+    When I analyze Python content with class and method with docstring
+    Then the triples should contain a triple with predicate "uko-doc:hasDocstring" for a method
+
+  # ---- Coverage gap: PythonAnalyzer dotted base class (ast.Attribute) ----
+
+  Scenario: PythonAnalyzer extracts dotted base class inheritance
+    Given a PythonAnalyzer instance
+    When I analyze Python content with a class inheriting from "collections.abc.Mapping"
+    Then the triples should contain a triple with predicate "uko-py:inheritsFrom" and object_value "collections.abc.Mapping"
+
+  # ---- Coverage gap: MarkdownAnalyzer domain property ----
+
+  Scenario: MarkdownAnalyzer domain is "markdown"
+    Given a MarkdownAnalyzer instance
+    Then the analyzer domain property should be "markdown"
+
+  # ---- Coverage gap: MarkdownAnalyzer sibling heading pops stack ----
+
+  Scenario: MarkdownAnalyzer handles sibling headings at same level
+    Given a MarkdownAnalyzer instance
+    When I analyze Markdown with sibling headings at the same level
+    Then each sibling section should be contained by the document

robot/helper_uko_analyzers.py

@@ -0,0 +1,241 @@
+"""Robot Framework helper for UKO Analyzer Plugin Framework smoke tests.
+
+Provides a CLI-style interface for Robot to invoke analyzer creation,
+protocol compliance, registry operations, and triple extraction.
+Exit code 0 = success, 1 = failure.
+
+Usage:
+    python robot/helper_uko_analyzers.py <command>
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure the src directory is on the import path.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.domain.models.acms.analyzers import (  # noqa: E402
+    AnalyzerProtocol,
+    AnalyzerRegistry,
+    UKOTriple,
+)
+from cleveragents.domain.models.acms.markdown_analyzer import (  # noqa: E402
+    MarkdownAnalyzer,
+)
+from cleveragents.domain.models.acms.python_analyzer import (  # noqa: E402
+    PythonAnalyzer,
+)
+
+
+def main() -> int:
+    """Entry point called by Robot Framework ``Run Process``."""
+    if len(sys.argv) < 2:
+        print("Usage: helper_uko_analyzers.py <command>")
+        return 1
+
+    command: str = sys.argv[1]
+
+    if command == "python-protocol":
+        try:
+            analyzer = PythonAnalyzer()
+            assert isinstance(analyzer, AnalyzerProtocol)
+            assert ".py" in analyzer.supported_extensions
+            assert analyzer.domain == "python"
+            print("uko-python-protocol-ok")
+            return 0
+        except Exception as exc:
+            print(f"uko-python-protocol-fail: {exc}")
+            return 1
+
+    if command == "markdown-protocol":
+        try:
+            analyzer = MarkdownAnalyzer()
+            assert isinstance(analyzer, AnalyzerProtocol)
+            assert ".md" in analyzer.supported_extensions
+            assert analyzer.domain == "markdown"
+            print("uko-markdown-protocol-ok")
+            return 0
+        except Exception as exc:
+            print(f"uko-markdown-protocol-fail: {exc}")
+            return 1
+
+    if command == "registry":
+        try:
+            registry = AnalyzerRegistry()
+            py_analyzer = PythonAnalyzer()
+            md_analyzer = MarkdownAnalyzer()
+            registry.register(py_analyzer)
+            registry.register(md_analyzer)
+            assert registry.get_for_extension(".py") is py_analyzer
+            assert registry.get_for_extension(".md") is md_analyzer
+            assert registry.get_for_extension(".rs") is None
+            assert len(registry) == 2
+            print("uko-registry-ok")
+            return 0
+        except Exception as exc:
+            print(f"uko-registry-fail: {exc}")
+            return 1
+
+    if command == "python-analyze":
+        try:
+            analyzer = PythonAnalyzer()
+            code = (
+                '"""Module doc."""\n'
+                "import os\n"
+                "from pathlib import Path\n\n"
+                "class MyClass:\n"
+                '    """Class doc."""\n'
+                "    def method(self):\n"
+                "        pass\n\n"
+                "def top_func():\n"
+                '    """Function doc."""\n'
+                "    pass\n"
+            )
+            triples = analyzer.analyze(code, "src/sample.py")
+            assert len(triples) > 0
+            # Check module exists
+            assert any(
+                t.predicate == "rdf:type" and t.object_uri == "uko-code:Module"
+                for t in triples
+            )
+            # Check class exists
+            assert any(
+                t.predicate == "rdf:type" and t.object_uri == "uko-py:Class"
+                for t in triples
+            )
+            # Check function exists
+            assert any(
+                t.predicate == "rdf:type" and t.object_uri == "uko-py:Function"
+                for t in triples
+            )
+            # Check import reference
+            assert any(
+                t.predicate == "uko:references" and "os" in t.object_uri
+                for t in triples
+            )
+            print("uko-python-analyze-ok")
+            return 0
+        except Exception as exc:
+            print(f"uko-python-analyze-fail: {exc}")
+            return 1
+
+    if command == "markdown-analyze":
+        try:
+            analyzer = MarkdownAnalyzer()
+            content = (
+                "# Introduction\n\n"
+                "Some text with [link](https://example.com).\n\n"
+                "## Details\n\n"
+                "```python\nprint('hello')\n```\n"
+            )
+            triples = analyzer.analyze(content, "docs/readme.md")
+            assert len(triples) > 0
+            # Check document exists
+            assert any(
+                t.predicate == "rdf:type" and t.object_uri == "uko-doc:Document"
+                for t in triples
+            )
+            # Check section exists
+            assert any(
+                t.predicate == "rdf:type" and t.object_uri == "uko-doc:Section"
+                for t in triples
+            )
+            # Check code block
+            assert any(
+                t.predicate == "rdf:type" and t.object_uri == "uko-code:CodeBlock"
+                for t in triples
+            )
+            # Check link reference
+            assert any(
+                t.predicate == "uko:references"
+                and "https://example.com" in t.object_value
+                for t in triples
+            )
+            print("uko-markdown-analyze-ok")
+            return 0
+        except Exception as exc:
+            print(f"uko-markdown-analyze-fail: {exc}")
+            return 1
+
+    if command == "triple-model":
+        try:
+            # Valid object-property triple
+            t1 = UKOTriple(
+                subject_uri="uko://code/module/foo",
+                predicate="rdf:type",
+                object_uri="uko-code:Module",
+            )
+            assert t1.confidence == 1.0
+            # Valid data-property triple
+            t2 = UKOTriple(
+                subject_uri="uko://code/class/Bar",
+                predicate="rdfs:label",
+                object_value="Bar",
+                confidence=0.9,
+            )
+            assert t2.object_value == "Bar"
+            # Reject empty subject
+            try:
+                UKOTriple(
+                    subject_uri="",
+                    predicate="rdf:type",
+                    object_uri="uko:X",
+                )
+                print("uko-triple-model-fail: no ValueError for empty subject")
+                return 1
+            except ValueError:
+                pass
+            # Reject bad confidence
+            try:
+                UKOTriple(
+                    subject_uri="uko://x",
+                    predicate="rdf:type",
+                    object_uri="uko:X",
+                    confidence=2.0,
+                )
+                print("uko-triple-model-fail: no ValueError for bad confidence")
+                return 1
+            except ValueError:
+                pass
+            print("uko-triple-model-ok")
+            return 0
+        except Exception as exc:
+            print(f"uko-triple-model-fail: {exc}")
+            return 1
+
+    if command == "malformed":
+        try:
+            py = PythonAnalyzer()
+            # Syntax error => empty list
+            result = py.analyze("def (\n    broken", "src/bad.py")
+            assert result == []
+            # Empty content => ValueError
+            try:
+                py.analyze("", "src/test.py")
+                print("uko-malformed-fail: no ValueError for empty content")
+                return 1
+            except ValueError:
+                pass
+            md = MarkdownAnalyzer()
+            try:
+                md.analyze("", "docs/test.md")
+                print("uko-malformed-fail: no ValueError for empty md content")
+                return 1
+            except ValueError:
+                pass
+            print("uko-malformed-ok")
+            return 0
+        except Exception as exc:
+            print(f"uko-malformed-fail: {exc}")
+            return 1
+
+    print(f"Unknown command: {command}")
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

robot/uko_analyzers.robot

@@ -0,0 +1,65 @@
+*** Settings ***
+Documentation    Smoke tests for UKO Analyzer Plugin Framework
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_uko_analyzers.py
+
+*** Test Cases ***
+AnalyzerProtocol Compliance For PythonAnalyzer
+    [Documentation]    Verify PythonAnalyzer satisfies AnalyzerProtocol
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    python-protocol    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    uko-python-protocol-ok
+
+AnalyzerProtocol Compliance For MarkdownAnalyzer
+    [Documentation]    Verify MarkdownAnalyzer satisfies AnalyzerProtocol
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    markdown-protocol    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    uko-markdown-protocol-ok
+
+AnalyzerRegistry Registration And Lookup
+    [Documentation]    Verify registry registers and looks up analyzers
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    registry    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    uko-registry-ok
+
+PythonAnalyzer Produces Triples
+    [Documentation]    Verify PythonAnalyzer produces valid UKO triples from Python source
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    python-analyze    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    uko-python-analyze-ok
+
+MarkdownAnalyzer Produces Triples
+    [Documentation]    Verify MarkdownAnalyzer produces valid UKO triples from Markdown
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    markdown-analyze    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    uko-markdown-analyze-ok
+
+UKOTriple Construction
+    [Documentation]    Verify UKOTriple model construction and validation
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    triple-model    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    uko-triple-model-ok
+
+Malformed Input Handling
+    [Documentation]    Verify graceful handling of malformed inputs
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    malformed    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    uko-malformed-ok

src/cleveragents/domain/models/acms/__init__.py

@@ -31,11 +31,25 @@ Tier types (from :mod:`~cleveragents.domain.models.acms.tiers`):
 - ``TierMetrics`` -- Hit/miss counters for cache monitoring
 - ``ScopedBackendView`` -- Project-scoped resource isolation
 
+Analyzer types (from :mod:`~cleveragents.domain.models.acms.analyzers`):
+- ``UKOTriple`` -- Immutable subject-predicate-object triple
+- ``AnalyzerProtocol`` -- Protocol for domain analyzers
+- ``AnalyzerRegistry`` -- Extension-to-analyzer registration
+
+Concrete analyzers:
+- ``PythonAnalyzer`` -- AST-based Python source analyzer
+- ``MarkdownAnalyzer`` -- Heading/code-block/link Markdown analyzer
+
 Based on ``docs/specification.md`` ACMS / CRP sections and ADR-014.
 """
 
 from __future__ import annotations
 
+from cleveragents.domain.models.acms.analyzers import (
+    AnalyzerProtocol,
+    AnalyzerRegistry,
+    UKOTriple,
+)
 from cleveragents.domain.models.acms.backends import (
     GraphBackend,
     GraphResult,
@@ -52,6 +66,8 @@ from cleveragents.domain.models.acms.crp import (
     DetailLevelMap,
     FragmentProvenance,
 )
+from cleveragents.domain.models.acms.markdown_analyzer import MarkdownAnalyzer
+from cleveragents.domain.models.acms.python_analyzer import PythonAnalyzer
 from cleveragents.domain.models.acms.stubs import (
     InMemoryGraphBackend,
     InMemoryTextBackend,
@@ -70,6 +86,8 @@ from cleveragents.domain.models.acms.tiers import (
 __all__: list[str] = [
     "ActorContextView",
     "ActorRole",
+    "AnalyzerProtocol",
+    "AnalyzerRegistry",
     "AssembledContext",
     "ContextBudget",
     "ContextFragment",
@@ -82,12 +100,15 @@ __all__: list[str] = [
     "InMemoryGraphBackend",
     "InMemoryTextBackend",
     "InMemoryVectorBackend",
+    "MarkdownAnalyzer",
+    "PythonAnalyzer",
     "ScopedBackendView",
     "TextBackend",
     "TextResult",
     "TierBudget",
     "TierMetrics",
     "TieredFragment",
+    "UKOTriple",
     "VectorBackend",
     "VectorResult",
 ]

src/cleveragents/domain/models/acms/analyzers.py

@@ -0,0 +1,219 @@
+"""UKO analyzer plugin framework — protocol, triple model, and registry.
+
+Defines the core types for domain analyzers that parse resources into
+UKO (Universal Knowledge Ontology) triples:
+
+- ``UKOTriple`` — Immutable Pydantic model representing a single
+  subject-predicate-object statement with optional confidence score.
+- ``AnalyzerProtocol`` — ``typing.Protocol`` that all analyzers must
+  satisfy (``supported_extensions``, ``domain``, ``analyze``).
+- ``AnalyzerRegistry`` — In-memory registry mapping file extensions to
+  analyzer instances with registration, lookup, and listing.
+
+Analyzers are registered by file extension and invoked by the
+``UKOIndexer`` to produce triples from resource content.  Custom
+analyzers are registered via ``config.toml``
+``context.uko.analyzers.custom.*``.
+
+Based on ``docs/specification.md`` ACMS Extensions — Domain and
+Language-Specific Analyzers, and configuration key
+``context.uko.default-analyzers``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Protocol, runtime_checkable
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# UKOTriple
+# ---------------------------------------------------------------------------
+
+
+class UKOTriple(BaseModel):
+    """A single subject-predicate-object statement in the UKO graph.
+
+    Represents either an object-property triple (``object_uri`` set) or
+    a data-property triple (``object_value`` set).  At least one of the
+    two must be provided; both may be set when the object is
+    simultaneously a URI-identified resource with a literal label.
+
+    Attributes:
+        subject_uri: URI of the subject node (e.g.
+            ``uko://code/module/mypackage``).
+        predicate: URI or prefixed name of the predicate (e.g.
+            ``uko:contains``, ``rdf:type``).
+        object_uri: URI of the object node (for object-property triples).
+        object_value: Literal value (for data-property triples).
+        confidence: Confidence score in ``[0.0, 1.0]``.  Defaults to
+            ``1.0`` for deterministic extractions.
+    """
+
+    subject_uri: str = Field(..., min_length=1, description="URI of the subject node.")
+    predicate: str = Field(
+        ..., min_length=1, description="Predicate URI or prefixed name."
+    )
+    object_uri: str = Field(
+        default="", description="URI of the object node (object-property)."
+    )
+    object_value: str = Field(default="", description="Literal value (data-property).")
+    confidence: float = Field(
+        default=1.0,
+        ge=0.0,
+        le=1.0,
+        description="Confidence in [0.0, 1.0].",
+    )
+
+    model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
+
+    # -- Validators -----------------------------------------------------------
+
+    @field_validator("subject_uri")
+    @classmethod
+    def _validate_subject_uri(cls, value: str) -> str:
+        if not value:
+            raise ValueError("subject_uri must not be empty.")
+        return value
+
+    @field_validator("predicate")
+    @classmethod
+    def _validate_predicate(cls, value: str) -> str:
+        if not value:
+            raise ValueError("predicate must not be empty.")
+        return value
+
+
+# ---------------------------------------------------------------------------
+# AnalyzerProtocol
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class AnalyzerProtocol(Protocol):
+    """Protocol for domain analyzers that parse resources into UKO triples.
+
+    Implementations must expose:
+
+    * ``supported_extensions`` — frozen set of file extensions (with
+      leading dot, e.g. ``{".py"}``).
+    * ``domain`` — human-readable domain label (e.g. ``"python"``).
+    * ``analyze(content, resource_uri)`` — produce a list of
+      ``UKOTriple`` instances from the given content.
+    """
+
+    @property
+    def supported_extensions(self) -> frozenset[str]:
+        """File extensions this analyzer handles (e.g. ``{".py"}``)."""
+        ...  # pragma: no cover
+
+    @property
+    def domain(self) -> str:
+        """Human-readable domain label (e.g. ``"python"``)."""
+        ...  # pragma: no cover
+
+    def analyze(self, content: str, resource_uri: str) -> list[UKOTriple]:
+        """Parse *content* and return UKO triples.
+
+        Args:
+            content: Raw textual content of the resource.
+            resource_uri: Canonical URI for the resource being analyzed.
+
+        Returns:
+            List of ``UKOTriple`` instances extracted from *content*.
+        """
+        ...  # pragma: no cover
+
+
+# ---------------------------------------------------------------------------
+# AnalyzerRegistry
+# ---------------------------------------------------------------------------
+
+
+class AnalyzerRegistry:
+    """In-memory registry mapping file extensions to analyzer instances.
+
+    Analyzers are registered via :meth:`register` and looked up by file
+    extension via :meth:`get_for_extension`.  Multiple analyzers may
+    handle the same extension; the *first* registered wins.
+
+    Example::
+
+        registry = AnalyzerRegistry()
+        registry.register(PythonAnalyzer())
+        analyzer = registry.get_for_extension(".py")
+    """
+
+    def __init__(self) -> None:
+        self._by_extension: dict[str, AnalyzerProtocol] = {}
+        self._all: list[AnalyzerProtocol] = []
+
+    # -- Registration ---------------------------------------------------------
+
+    def register(self, analyzer: AnalyzerProtocol) -> None:
+        """Register an analyzer for all its supported extensions.
+
+        Args:
+            analyzer: An object satisfying ``AnalyzerProtocol``.
+
+        Raises:
+            TypeError: If *analyzer* does not satisfy the protocol.
+            ValueError: If *analyzer* declares no supported extensions.
+        """
+        if not isinstance(analyzer, AnalyzerProtocol):
+            raise TypeError(f"Expected AnalyzerProtocol, got {type(analyzer).__name__}")
+        extensions = analyzer.supported_extensions
+        if not extensions:
+            raise ValueError(
+                f"Analyzer '{type(analyzer).__name__}' declares no "
+                "supported extensions."
+            )
+        for ext in extensions:
+            if ext not in self._by_extension:
+                self._by_extension[ext] = analyzer
+                logger.debug(
+                    "Registered analyzer %s for extension '%s'",
+                    type(analyzer).__name__,
+                    ext,
+                )
+            else:
+                logger.debug(
+                    "Extension '%s' already handled by %s; skipping %s",
+                    ext,
+                    type(self._by_extension[ext]).__name__,
+                    type(analyzer).__name__,
+                )
+        self._all.append(analyzer)
+
+    # -- Lookup ---------------------------------------------------------------
+
+    def get_for_extension(self, extension: str) -> AnalyzerProtocol | None:
+        """Return the analyzer registered for *extension*, or ``None``.
+
+        Args:
+            extension: File extension including leading dot (e.g.
+                ``".py"``).
+
+        Returns:
+            The registered ``AnalyzerProtocol`` or ``None``.
+        """
+        if not extension:
+            return None
+        return self._by_extension.get(extension)
+
+    # -- Listing --------------------------------------------------------------
+
+    def list_extensions(self) -> frozenset[str]:
+        """Return all registered file extensions."""
+        return frozenset(self._by_extension.keys())
+
+    def list_analyzers(self) -> list[AnalyzerProtocol]:
+        """Return all registered analyzers (in registration order)."""
+        return list(self._all)
+
+    def __len__(self) -> int:
+        """Return the number of registered analyzers."""
+        return len(self._all)

src/cleveragents/domain/models/acms/markdown_analyzer.py

@@ -0,0 +1,275 @@
+"""MarkdownAnalyzer — heading/code-block/link extraction into UKO triples.
+
+Parses Markdown documents and extracts:
+
+- Document declaration (``uko-doc:Document``).
+- Sections by heading level with containment hierarchy
+  (``uko-doc:Section``).
+- Code blocks with language annotations (``uko-code:CodeBlock``).
+- Links and references (``uko:references``).
+- Heading text and level properties.
+
+All extracted elements are represented as ``UKOTriple`` instances with
+``uko://`` URI schemes following the UKO ontology hierarchy:
+
+- Layer 0 core: ``uko:contains``, ``uko:references``
+- Layer 1 doc: ``uko-doc:Document``, ``uko-doc:Section``
+- Layer 1 code: ``uko-code:CodeBlock``
+
+Uses line-by-line regex parsing; no external Markdown library required.
+
+Based on ``docs/specification.md`` ACMS Extensions — MarkdownAnalyzer.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+
+from cleveragents.domain.models.acms.analyzers import AnalyzerProtocol, UKOTriple
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Regex patterns
+# ---------------------------------------------------------------------------
+
+_HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)(?:\s*#*\s*)?$")
+_FENCED_OPEN_RE = re.compile(r"^```(\w*)")
+_FENCED_CLOSE_RE = re.compile(r"^```\s*$")
+_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
+
+# ---------------------------------------------------------------------------
+# URI helpers
+# ---------------------------------------------------------------------------
+
+_SAFE_RE = re.compile(r"[^a-zA-Z0-9_.-]")
+
+
+def _safe(text: str) -> str:
+    """Sanitise text for use in a URI path segment."""
+    return _SAFE_RE.sub("_", text).strip("_")[:120]
+
+
+def _doc_uri(resource_uri: str) -> str:
+    """Build a UKO document URI from a resource URI."""
+    return f"uko://docs/document/{_safe(resource_uri)}"
+
+
+def _section_uri(resource_uri: str, heading_text: str, index: int) -> str:
+    """Build a UKO section URI."""
+    slug = _safe(heading_text.lower())
+    return f"uko://docs/section/{_safe(resource_uri)}/{slug}_{index}"
+
+
+def _codeblock_uri(resource_uri: str, index: int) -> str:
+    """Build a UKO code-block URI."""
+    return f"uko://docs/codeblock/{_safe(resource_uri)}/{index}"
+
+
+# ---------------------------------------------------------------------------
+# MarkdownAnalyzer
+# ---------------------------------------------------------------------------
+
+
+class MarkdownAnalyzer:
+    """Markdown document analyzer producing UKO triples.
+
+    Satisfies :class:`AnalyzerProtocol`.  Handles ``.md`` and
+    ``.markdown`` files.  Performs line-by-line parsing for headings,
+    fenced code blocks, and Markdown links.
+
+    Example::
+
+        analyzer = MarkdownAnalyzer()
+        triples = analyzer.analyze("# Hello\\nWorld\\n", "docs/readme.md")
+    """
+
+    @property
+    def supported_extensions(self) -> frozenset[str]:
+        """File extensions handled by this analyzer."""
+        return frozenset({".md", ".markdown"})
+
+    @property
+    def domain(self) -> str:
+        """Human-readable domain label."""
+        return "markdown"
+
+    def analyze(self, content: str, resource_uri: str) -> list[UKOTriple]:
+        """Parse *content* as Markdown and extract UKO triples.
+
+        Args:
+            content: Raw Markdown text.
+            resource_uri: Canonical URI of the resource.
+
+        Returns:
+            List of ``UKOTriple`` instances.
+
+        Raises:
+            ValueError: If *content* or *resource_uri* is empty.
+        """
+        if not content:
+            raise ValueError("content must not be empty.")
+        if not resource_uri:
+            raise ValueError("resource_uri must not be empty.")
+
+        triples: list[UKOTriple] = []
+        doc_uri = _doc_uri(resource_uri)
+
+        # Document declaration
+        triples.append(
+            UKOTriple(
+                subject_uri=doc_uri,
+                predicate="rdf:type",
+                object_uri="uko-doc:Document",
+            )
+        )
+
+        lines = content.splitlines()
+        section_idx = 0
+        codeblock_idx = 0
+        in_code_block = False
+        # Stack of (heading_level, section_uri) for containment
+        section_stack: list[tuple[int, str]] = []
+
+        for line in lines:
+            # -- Fenced code block tracking --
+            if in_code_block:
+                if _FENCED_CLOSE_RE.match(line):
+                    in_code_block = False
+                continue
+
+            fence_match = _FENCED_OPEN_RE.match(line)
+            if fence_match and line.strip() != "```":
+                # Opening fence with or without language
+                in_code_block = True
+                codeblock_idx += 1
+                cb_uri = _codeblock_uri(resource_uri, codeblock_idx)
+                lang = fence_match.group(1)
+
+                triples.append(
+                    UKOTriple(
+                        subject_uri=cb_uri,
+                        predicate="rdf:type",
+                        object_uri="uko-code:CodeBlock",
+                    )
+                )
+
+                # Containment: inside nearest section or document
+                parent = section_stack[-1][1] if section_stack else doc_uri
+                triples.append(
+                    UKOTriple(
+                        subject_uri=parent,
+                        predicate="uko:contains",
+                        object_uri=cb_uri,
+                    )
+                )
+
+                if lang:
+                    triples.append(
+                        UKOTriple(
+                            subject_uri=cb_uri,
+                            predicate="uko-code:language",
+                            object_value=lang,
+                        )
+                    )
+                continue
+
+            # Bare ``` that opens a fence without language
+            if line.strip() == "```":
+                in_code_block = True
+                codeblock_idx += 1
+                cb_uri = _codeblock_uri(resource_uri, codeblock_idx)
+
+                triples.append(
+                    UKOTriple(
+                        subject_uri=cb_uri,
+                        predicate="rdf:type",
+                        object_uri="uko-code:CodeBlock",
+                    )
+                )
+                parent = section_stack[-1][1] if section_stack else doc_uri
+                triples.append(
+                    UKOTriple(
+                        subject_uri=parent,
+                        predicate="uko:contains",
+                        object_uri=cb_uri,
+                    )
+                )
+                continue
+
+            # -- Headings --
+            heading_match = _HEADING_RE.match(line)
+            if heading_match:
+                level = len(heading_match.group(1))
+                text = heading_match.group(2).strip()
+                section_idx += 1
+                sec_uri = _section_uri(resource_uri, text, section_idx)
+
+                triples.append(
+                    UKOTriple(
+                        subject_uri=sec_uri,
+                        predicate="rdf:type",
+                        object_uri="uko-doc:Section",
+                    )
+                )
+                triples.append(
+                    UKOTriple(
+                        subject_uri=sec_uri,
+                        predicate="uko-doc:headingLevel",
+                        object_value=str(level),
+                    )
+                )
+                triples.append(
+                    UKOTriple(
+                        subject_uri=sec_uri,
+                        predicate="uko-doc:headingText",
+                        object_value=text,
+                    )
+                )
+
+                # Pop sections at same or deeper level
+                while section_stack and section_stack[-1][0] >= level:
+                    section_stack.pop()
+
+                # Containment
+                parent = section_stack[-1][1] if section_stack else doc_uri
+                triples.append(
+                    UKOTriple(
+                        subject_uri=parent,
+                        predicate="uko:contains",
+                        object_uri=sec_uri,
+                    )
+                )
+
+                section_stack.append((level, sec_uri))
+                continue
+
+            # -- Links --
+            for link_match in _LINK_RE.finditer(line):
+                link_text = link_match.group(1)
+                link_url = link_match.group(2)
+                # Current context is nearest section or document
+                context_uri = section_stack[-1][1] if section_stack else doc_uri
+                triples.append(
+                    UKOTriple(
+                        subject_uri=context_uri,
+                        predicate="uko:references",
+                        object_value=link_url,
+                        confidence=0.9,
+                    )
+                )
+                if link_text:
+                    triples.append(
+                        UKOTriple(
+                            subject_uri=context_uri,
+                            predicate="rdfs:label",
+                            object_value=link_text,
+                        )
+                    )
+
+        return triples
+
+
+# Protocol compliance assertion (zero-cost at import time).
+_: type[AnalyzerProtocol] = MarkdownAnalyzer

src/cleveragents/domain/models/acms/python_analyzer.py

@@ -0,0 +1,365 @@
+"""PythonAnalyzer — AST-based extraction of UKO triples from Python files.
+
+Parses Python source into an ``ast`` tree and extracts:
+
+- Module-level declarations (``uko-code:Module``).
+- Class definitions with docstrings (``uko-py:Class``).
+- Function and method definitions with docstrings (``uko-py:Function``).
+- Import statements (``uko:references``).
+- Module-level docstrings (``uko-doc:hasDocstring``).
+
+All extracted elements are represented as ``UKOTriple`` instances with
+``uko://`` URI schemes following the UKO ontology hierarchy:
+
+- Layer 0 core: ``uko:contains``, ``uko:references``
+- Layer 1 code: ``uko-code:Module``
+- Layer 3 Python-specific: ``uko-py:Class``, ``uko-py:Function``
+
+Based on ``docs/specification.md`` ACMS Extensions — PythonAnalyzer.
+"""
+
+from __future__ import annotations
+
+import ast
+import logging
+import re
+
+from cleveragents.domain.models.acms.analyzers import AnalyzerProtocol, UKOTriple
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# URI helpers
+# ---------------------------------------------------------------------------
+
+_SAFE_NAME_RE = re.compile(r"[^a-zA-Z0-9_.]")
+
+
+def _safe_name(name: str) -> str:
+    """Sanitise a Python identifier for use in a URI path segment."""
+    return _SAFE_NAME_RE.sub("_", name)
+
+
+def _module_uri(resource_uri: str) -> str:
+    """Build a UKO module URI from a resource URI."""
+    return f"uko://code/module/{_safe_name(resource_uri)}"
+
+
+def _class_uri(resource_uri: str, class_name: str) -> str:
+    """Build a UKO class URI."""
+    return f"uko://code/class/{_safe_name(resource_uri)}/{_safe_name(class_name)}"
+
+
+def _function_uri(resource_uri: str, func_name: str, class_name: str = "") -> str:
+    """Build a UKO function/method URI."""
+    base = f"uko://code/function/{_safe_name(resource_uri)}"
+    if class_name:
+        return f"{base}/{_safe_name(class_name)}/{_safe_name(func_name)}"
+    return f"{base}/{_safe_name(func_name)}"
+
+
+def _import_uri(module_name: str) -> str:
+    """Build a UKO import reference URI."""
+    return f"uko://code/module/{_safe_name(module_name)}"
+
+
+# ---------------------------------------------------------------------------
+# PythonAnalyzer
+# ---------------------------------------------------------------------------
+
+
+class PythonAnalyzer:
+    """AST-based Python source analyzer producing UKO triples.
+
+    Satisfies :class:`AnalyzerProtocol`.  Handles ``.py`` and ``.pyi``
+    files.  Gracefully returns an empty list for unparsable content.
+
+    Example::
+
+        analyzer = PythonAnalyzer()
+        triples = analyzer.analyze("class Foo:\\n    pass\\n", "src/foo.py")
+    """
+
+    @property
+    def supported_extensions(self) -> frozenset[str]:
+        """File extensions handled by this analyzer."""
+        return frozenset({".py", ".pyi"})
+
+    @property
+    def domain(self) -> str:
+        """Human-readable domain label."""
+        return "python"
+
+    def analyze(self, content: str, resource_uri: str) -> list[UKOTriple]:
+        """Parse *content* as Python and extract UKO triples.
+
+        Args:
+            content: Raw Python source code.
+            resource_uri: Canonical URI of the resource.
+
+        Returns:
+            List of ``UKOTriple`` instances.  Returns an empty list on
+            parse failure.
+
+        Raises:
+            ValueError: If *content* or *resource_uri* is empty.
+        """
+        if not content:
+            raise ValueError("content must not be empty.")
+        if not resource_uri:
+            raise ValueError("resource_uri must not be empty.")
+
+        try:
+            tree = ast.parse(content, filename=resource_uri)
+        except SyntaxError:
+            logger.warning(
+                "PythonAnalyzer: syntax error in '%s'; returning empty",
+                resource_uri,
+            )
+            return []
+
+        triples: list[UKOTriple] = []
+        mod_uri = _module_uri(resource_uri)
+
+        # Module declaration
+        triples.append(
+            UKOTriple(
+                subject_uri=mod_uri,
+                predicate="rdf:type",
+                object_uri="uko-code:Module",
+            )
+        )
+
+        # Module docstring
+        docstring = ast.get_docstring(tree)
+        if docstring:
+            triples.append(
+                UKOTriple(
+                    subject_uri=mod_uri,
+                    predicate="uko-doc:hasDocstring",
+                    object_value=docstring[:500],
+                )
+            )
+
+        # Walk top-level nodes
+        for node in ast.iter_child_nodes(tree):
+            if isinstance(node, ast.ClassDef):
+                triples.extend(self._extract_class(node, resource_uri, mod_uri))
+            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                triples.extend(self._extract_function(node, resource_uri, mod_uri))
+            elif isinstance(node, ast.Import):
+                triples.extend(self._extract_import(node, mod_uri))
+            elif isinstance(node, ast.ImportFrom):
+                triples.extend(self._extract_import_from(node, mod_uri))
+
+        return triples
+
+    # -- Internal extraction helpers ------------------------------------------
+
+    def _extract_class(
+        self,
+        node: ast.ClassDef,
+        resource_uri: str,
+        mod_uri: str,
+    ) -> list[UKOTriple]:
+        """Extract triples for a class definition."""
+        triples: list[UKOTriple] = []
+        cls_uri = _class_uri(resource_uri, node.name)
+
+        # Type declaration
+        triples.append(
+            UKOTriple(
+                subject_uri=cls_uri,
+                predicate="rdf:type",
+                object_uri="uko-py:Class",
+            )
+        )
+
+        # Containment
+        triples.append(
+            UKOTriple(
+                subject_uri=mod_uri,
+                predicate="uko:contains",
+                object_uri=cls_uri,
+            )
+        )
+
+        # Label
+        triples.append(
+            UKOTriple(
+                subject_uri=cls_uri,
+                predicate="rdfs:label",
+                object_value=node.name,
+            )
+        )
+
+        # Docstring
+        docstring = ast.get_docstring(node)
+        if docstring:
+            triples.append(
+                UKOTriple(
+                    subject_uri=cls_uri,
+                    predicate="uko-doc:hasDocstring",
+                    object_value=docstring[:500],
+                )
+            )
+
+        # Base classes
+        for base in node.bases:
+            base_name = _base_name(base)
+            if base_name:
+                triples.append(
+                    UKOTriple(
+                        subject_uri=cls_uri,
+                        predicate="uko-py:inheritsFrom",
+                        object_value=base_name,
+                    )
+                )
+
+        # Methods inside the class
+        for child in ast.iter_child_nodes(node):
+            if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                triples.extend(
+                    self._extract_method(child, resource_uri, cls_uri, node.name)
+                )
+
+        return triples
+
+    def _extract_function(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef,
+        resource_uri: str,
+        mod_uri: str,
+    ) -> list[UKOTriple]:
+        """Extract triples for a module-level function."""
+        triples: list[UKOTriple] = []
+        func_uri = _function_uri(resource_uri, node.name)
+
+        triples.append(
+            UKOTriple(
+                subject_uri=func_uri,
+                predicate="rdf:type",
+                object_uri="uko-py:Function",
+            )
+        )
+        triples.append(
+            UKOTriple(
+                subject_uri=mod_uri,
+                predicate="uko:contains",
+                object_uri=func_uri,
+            )
+        )
+        triples.append(
+            UKOTriple(
+                subject_uri=func_uri,
+                predicate="rdfs:label",
+                object_value=node.name,
+            )
+        )
+
+        docstring = ast.get_docstring(node)
+        if docstring:
+            triples.append(
+                UKOTriple(
+                    subject_uri=func_uri,
+                    predicate="uko-doc:hasDocstring",
+                    object_value=docstring[:500],
+                )
+            )
+
+        return triples
+
+    def _extract_method(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef,
+        resource_uri: str,
+        cls_uri: str,
+        class_name: str,
+    ) -> list[UKOTriple]:
+        """Extract triples for a method inside a class."""
+        triples: list[UKOTriple] = []
+        method_uri = _function_uri(resource_uri, node.name, class_name)
+
+        triples.append(
+            UKOTriple(
+                subject_uri=method_uri,
+                predicate="rdf:type",
+                object_uri="uko-py:Function",
+            )
+        )
+        triples.append(
+            UKOTriple(
+                subject_uri=cls_uri,
+                predicate="uko:contains",
+                object_uri=method_uri,
+            )
+        )
+        triples.append(
+            UKOTriple(
+                subject_uri=method_uri,
+                predicate="rdfs:label",
+                object_value=node.name,
+            )
+        )
+
+        docstring = ast.get_docstring(node)
+        if docstring:
+            triples.append(
+                UKOTriple(
+                    subject_uri=method_uri,
+                    predicate="uko-doc:hasDocstring",
+                    object_value=docstring[:500],
+                )
+            )
+
+        return triples
+
+    def _extract_import(self, node: ast.Import, mod_uri: str) -> list[UKOTriple]:
+        """Extract triples for ``import X`` statements."""
+        triples: list[UKOTriple] = []
+        for alias in node.names:
+            triples.append(
+                UKOTriple(
+                    subject_uri=mod_uri,
+                    predicate="uko:references",
+                    object_uri=_import_uri(alias.name),
+                )
+            )
+        return triples
+
+    def _extract_import_from(
+        self, node: ast.ImportFrom, mod_uri: str
+    ) -> list[UKOTriple]:
+        """Extract triples for ``from X import Y`` statements."""
+        triples: list[UKOTriple] = []
+        module_name = node.module or ""
+        if module_name:
+            triples.append(
+                UKOTriple(
+                    subject_uri=mod_uri,
+                    predicate="uko:references",
+                    object_uri=_import_uri(module_name),
+                )
+            )
+        return triples
+
+
+# ---------------------------------------------------------------------------
+# ast helpers
+# ---------------------------------------------------------------------------
+
+
+def _base_name(node: ast.expr) -> str:
+    """Extract a human-readable base class name from an AST node."""
+    if isinstance(node, ast.Name):
+        return node.id
+    if isinstance(node, ast.Attribute):
+        value_name = _base_name(node.value)
+        if value_name:
+            return f"{value_name}.{node.attr}"
+        return node.attr
+    return ""
+
+
+# Protocol compliance assertion (zero-cost at import time).
+_: type[AnalyzerProtocol] = PythonAnalyzer