diff --git a/benchmarks/uko_analyzers_bench.py b/benchmarks/uko_analyzers_bench.py new file mode 100644 index 000000000..a3d760611 --- /dev/null +++ b/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") diff --git a/features/steps/uko_analyzers_steps.py b/features/steps/uko_analyzers_steps.py new file mode 100644 index 000000000..c58e91eea --- /dev/null +++ b/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}" + ) diff --git a/features/uko_analyzers.feature b/features/uko_analyzers.feature new file mode 100644 index 000000000..df6dffb2a --- /dev/null +++ b/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 diff --git a/robot/helper_uko_analyzers.py b/robot/helper_uko_analyzers.py new file mode 100644 index 000000000..fca921cf8 --- /dev/null +++ b/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 +""" + +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 ") + 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()) diff --git a/robot/uko_analyzers.robot b/robot/uko_analyzers.robot new file mode 100644 index 000000000..b64087a4a --- /dev/null +++ b/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 diff --git a/src/cleveragents/domain/models/acms/__init__.py b/src/cleveragents/domain/models/acms/__init__.py index 0309e74c7..7ab34236e 100644 --- a/src/cleveragents/domain/models/acms/__init__.py +++ b/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", ] diff --git a/src/cleveragents/domain/models/acms/analyzers.py b/src/cleveragents/domain/models/acms/analyzers.py new file mode 100644 index 000000000..9d2b2a66d --- /dev/null +++ b/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) diff --git a/src/cleveragents/domain/models/acms/markdown_analyzer.py b/src/cleveragents/domain/models/acms/markdown_analyzer.py new file mode 100644 index 000000000..e4286ffc0 --- /dev/null +++ b/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 diff --git a/src/cleveragents/domain/models/acms/python_analyzer.py b/src/cleveragents/domain/models/acms/python_analyzer.py new file mode 100644 index 000000000..2b7954722 --- /dev/null +++ b/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