feat: Add Runtime Signals to the knowledge graph: Cobertura test coverage report processing, associated new entity and relationship types, and integrate into graph building.

Author: deepakdgupta1

src/knowcode/cli.py

@@ -11,17 +11,17 @@
 from knowcode.context_synthesizer import ContextSynthesizer
 from knowcode.graph_builder import GraphBuilder
 from knowcode.knowledge_store import KnowledgeStore
-from knowcode.models import EntityKind
+from knowcode.models import EntityKind, RelationshipKind
 
 
 @click.group()
 @click.version_option(version=__version__)
-def main() -> None:
+def cli() -> None:
     """KnowCode - Transform your codebase into an effective knowledge base."""
     pass
 
 
-@main.command()
+@cli.command()
 @click.argument("directory", type=click.Path(exists=True, file_okay=False))
 @click.option(
     "--output", "-o",
@@ -39,20 +39,28 @@ def main() -> None:
     default=False,
     help="Analyze git history and add temporal context.",
 )
-def analyze(directory: str, output: str, ignore: tuple[str, ...], temporal: bool) -> None:
+@click.option(
+    "--coverage",
+    type=click.Path(exists=True, dir_okay=False),
+    help="Path to Cobertura XML coverage report.",
+)
+def analyze(directory: str, output: str, ignore: tuple[str, ...], temporal: bool, coverage: Optional[str]) -> None:
     """Scan and analyze a codebase.
 
     DIRECTORY: Path to the codebase to analyze.
     """
     click.echo(f"Analyzing: {directory}")
     click.echo(f"Temporal analysis: {'Enabled' if temporal else 'Disabled'}")
+    if coverage:
+        click.echo(f"Coverage report: {coverage}")
 
     # Build graph
     builder = GraphBuilder()
     builder.build_from_directory(
         root_dir=directory,
         additional_ignores=list(ignore),
         analyze_temporal=temporal,
+        coverage_path=Path(coverage) if coverage else None,
     )
 
     # Create store and save
@@ -72,7 +80,7 @@ def analyze(directory: str, output: str, ignore: tuple[str, ...], temporal: bool
     click.echo(f"\n  Saved to: {save_path}")
 
 
-@main.command()
+@cli.command()
 @click.argument("query_type", type=click.Choice(["callers", "callees", "deps", "search"]))
 @click.argument("target")
 @click.option(
@@ -179,7 +187,7 @@ def query(query_type: str, target: str, store: str, as_json: bool) -> None:
                 click.echo(f"  • {name}{extra}")
 
 
-@main.command()
+@cli.command()
 @click.argument("target")
 @click.option(
     "--store", "-s",
@@ -227,7 +235,7 @@ def context(target: str, store: str, max_tokens: int) -> None:
             click.echo("(truncated)", err=True)
 
 
-@main.command()
+@cli.command()
 @click.option(
     "--store", "-s",
     type=click.Path(exists=True),
@@ -280,7 +288,7 @@ def export(store: str, output: str) -> None:
     click.echo(f"  Index: {index_path}")
 
 
-@main.command()
+@cli.command()
 @click.option(
     "--store", "-s",
     type=click.Path(exists=True),
@@ -320,9 +328,8 @@ def stats(store: str) -> None:
         click.echo(f"  {kind}: {count}")
 
 
-if __name__ == "__main__":
-    main()
-@main.command()
+
+@cli.command()
 @click.argument("target", required=False)
 @click.option(
     "--store", "-s",
@@ -406,3 +413,6 @@ def history(target: Optional[str], store: str, limit: int) -> None:
             date = commit.metadata.get("date", "")
             click.echo(f"  {date} {commit.name} {stats}: {commit.docstring.splitlines()[0]}")
 
+
+if __name__ == "__main__":
+    cli()

src/knowcode/graph_builder.py

@@ -8,6 +8,7 @@
 from knowcode.parsers.javascript_parser import JavaScriptParser
 from knowcode.parsers.java_parser import JavaParser
 from knowcode.scanner import FileInfo, Scanner
+from knowcode.signals import CoverageProcessor
 from knowcode.temporal import TemporalAnalyzer
 
 
@@ -31,6 +32,7 @@ def build_from_directory(
         root_dir: str | Path,
         additional_ignores: Optional[list[str]] = None,
         analyze_temporal: bool = False,
+        coverage_path: Optional[Path] = None,
     ) -> "GraphBuilder":
         """Build graph by scanning and parsing a directory.
 
@@ -58,6 +60,12 @@ def build_from_directory(
             result = temporal_analyzer.analyze_history()
             self._merge_result(result)
 
+        # Coverage Analysis
+        if coverage_path:
+            coverage_processor = CoverageProcessor(root_dir)
+            result = coverage_processor.process_cobertura(coverage_path)
+            self._merge_result(result)
+            
         return self
 
     def build_from_files(self, files: list[FileInfo]) -> "GraphBuilder":

src/knowcode/models.py

@@ -21,6 +21,9 @@ class EntityKind(str, Enum):
     # Temporal entities
     COMMIT = "commit"
     AUTHOR = "author"
+    # Runtime entities
+    TEST_RUN = "test_run"
+    COVERAGE_REPORT = "coverage_report"
 
 
 class RelationshipKind(str, Enum):
@@ -35,6 +38,9 @@ class RelationshipKind(str, Enum):
     CHANGED_BY = "changed_by"  # Entity -> Commit
     AUTHORED = "authored"      # Author -> Commit
     MODIFIED = "modified"      # Commit -> Entity
+    # Runtime relationships
+    COVERS = "covers"           # Report/Test -> Entity
+    EXECUTED_BY = "executed_by" # Entity -> Report/Test
 
 
 @dataclass

src/knowcode/parsers/yaml_parser.py

@@ -105,11 +105,12 @@ def _extract_keys(
     ) -> None:
         """Recursively extract configuration keys."""
         for key, value in data.items():
-            qualified_name = f"{prefix}{key}" if prefix else key
+            str_key = str(key)
+            qualified_name = f"{prefix}{str_key}" if prefix else str_key
             key_id = f"{file_path}::{qualified_name}"
 
             # Try to find line number for this key
-            line_num = self._find_key_line(key, prefix, lines)
+            line_num = self._find_key_line(str_key, prefix, lines)
 
             # Determine value representation
             if isinstance(value, dict):
@@ -122,7 +123,7 @@ def _extract_keys(
             key_entity = Entity(
                 id=key_id,
                 kind=EntityKind.CONFIG_KEY,
-                name=key,
+                name=str_key,
                 qualified_name=qualified_name,
                 location=Location(
                     file_path=str(file_path),

src/knowcode/signals.py

@@ -0,0 +1,131 @@
+"""Runtime signal processing (e.g., coverage, traces)."""
+
+import xml.etree.ElementTree as ET
+from pathlib import Path
+from typing import Optional
+
+from knowcode.models import (
+    Entity,
+    EntityKind,
+    Location,
+    ParseResult,
+    Relationship,
+    RelationshipKind,
+)
+
+
+class CoverageProcessor:
+    """Process coverage reports."""
+
+    def __init__(self, root_dir: str | Path) -> None:
+        """Initialize coverage processor.
+
+        Args:
+            root_dir: Root directory of the codebase (for relative path resolution).
+        """
+        self.root_dir = Path(root_dir).resolve()
+
+    def process_cobertura(self, xml_path: str | Path) -> ParseResult:
+        """Process a Cobertura XML coverage report.
+
+        Args:
+            xml_path: Path to coverage.xml.
+
+        Returns:
+            ParseResult containing COVERAGE_REPORT entity and COVERS relationships.
+        """
+        xml_path = Path(xml_path)
+        if not xml_path.exists():
+            return ParseResult(
+                file_path=str(xml_path),
+                entities=[],
+                relationships=[],
+                errors=[f"Coverage file not found: {xml_path}"],
+            )
+
+        entities: list[Entity] = []
+        relationships: list[Relationship] = []
+        errors: list[str] = []
+
+        try:
+            tree = ET.parse(xml_path)
+            root = tree.getroot()
+
+            # Create Report Entity
+            report_id = f"coverage::{xml_path.name}"
+            # Extract timestamp if available available in root attributes usually 'timestamp'
+            timestamp = root.get("timestamp", str(xml_path.stat().st_mtime))
+            
+            report_entity = Entity(
+                id=report_id,
+                kind=EntityKind.COVERAGE_REPORT,
+                name=f"Coverage Report ({xml_path.name})",
+                qualified_name=xml_path.name,
+                location=Location(str(xml_path), 0, 0),
+                metadata={
+                    "timestamp": timestamp,
+                    "line-rate": root.get("line-rate", "0"),
+                    "branch-rate": root.get("branch-rate", "0"),
+                },
+            )
+            entities.append(report_entity)
+
+            # Traverse packages -> classes -> lines
+            # Structure: coverage -> packages -> package -> classes -> class -> lines -> line
+            
+            # We want to map files/classes to the report.
+            # <class name="knowcode.models" filename="src/knowcode/models.py" line-rate="1.0" ...>
+            
+            for cls in root.findall(".//class"):
+                filename = cls.get("filename")
+                if not filename:
+                    continue
+                
+                # Resolve file path to simple module ID
+                # We assume standard module ID: /abs/path/to/file::filename_stem
+                # filename in coverage.xml is usually relative to root
+                abs_file_path = (self.root_dir / filename).resolve()
+                module_name = abs_file_path.stem
+                module_id = f"{abs_file_path}::{module_name}"
+                
+                line_rate = cls.get("line-rate", "0")
+                
+                # Relationship: REPORT -> COVERS -> MODULE
+                relationships.append(
+                    Relationship(
+                        source_id=report_id,
+                        target_id=module_id,
+                        kind=RelationshipKind.COVERS,
+                        metadata={
+                            "line-rate": line_rate,
+                            "hits": cls.get("lines-covered", "0") + "/" + cls.get("lines-valid", "0")
+                        }
+                    )
+                )
+                
+                # We could map specific lines to entities if we had line ranges of entities loaded.
+                # Since CoverageProcessor runs independently or after graph build, 
+                # we usually just link to the File/Module level for MVP.
+                # Detailed line mapping requires access to the full graph to find which entity covers line X.
+                # For v1.4 MVP, linking to Module is sufficient.
+                
+                # Note: We can also add "EXECUTED_BY" from Module to Report
+                relationships.append(
+                    Relationship(
+                        source_id=module_id,
+                        target_id=report_id,
+                        kind=RelationshipKind.EXECUTED_BY
+                    )
+                )
+
+        except ET.ParseError as e:
+            errors.append(f"Invalid XML format: {e}")
+        except Exception as e:
+            errors.append(f"Error processing coverage: {e}")
+
+        return ParseResult(
+            file_path=str(xml_path),
+            entities=entities,
+            relationships=relationships,
+            errors=errors,
+        )

tests/test_signals.py

@@ -0,0 +1,67 @@
+"""Tests for Signal Ingestion."""
+
+import pytest
+from pathlib import Path
+from knowcode.signals import CoverageProcessor
+from knowcode.models import EntityKind, RelationshipKind
+
+@pytest.fixture
+def coverage_xml(tmp_path):
+    """Create a sample coverage.xml."""
+    content = """<?xml version="1.0" ?>
+    <coverage line-rate="0.5" branch-rate="0.0" lines-covered="10" lines-valid="20" timestamp="123456789">
+        <packages>
+            <package name="my_package" line-rate="0.5" branch-rate="0.0" complexity="0.0">
+                <classes>
+                    <class name="module_a" filename="module_a.py" line-rate="1.0" branch-rate="0.0" complexity="0.0" lines-covered="5" lines-valid="5">
+                        <lines>
+                            <line hits="1" number="1"/>
+                            <line hits="1" number="2"/>
+                        </lines>
+                    </class>
+                    <class name="module_b" filename="module_b.py" line-rate="0.0" branch-rate="0.0" complexity="0.0" lines-covered="0" lines-valid="5">
+                         <lines>
+                            <line hits="0" number="1"/>
+                        </lines>
+                    </class>
+                </classes>
+            </package>
+        </packages>
+    </coverage>
+    """
+    path = tmp_path / "coverage.xml"
+    path.write_text(content, encoding="utf-8")
+    return path
+
+def test_process_cobertura(tmp_path, coverage_xml):
+    """Test Cobertura XML processing."""
+    # Create dummy files so they resolve
+    (tmp_path / "module_a.py").touch()
+    (tmp_path / "module_b.py").touch()
+    
+    processor = CoverageProcessor(tmp_path)
+    result = processor.process_cobertura(coverage_xml)
+    
+    assert not result.errors
+    
+    # Check Report Entity
+    reports = [e for e in result.entities if e.kind == EntityKind.COVERAGE_REPORT]
+    assert len(reports) == 1
+    report = reports[0]
+    assert report.name == "Coverage Report (coverage.xml)"
+    assert report.metadata["line-rate"] == "0.5"
+    
+    # Check Relationships
+    # Report -> COVERS -> Module A
+    covers_rels = [r for r in result.relationships if r.kind == RelationshipKind.COVERS]
+    assert len(covers_rels) == 2 # module_a and module_b
+    
+    # Verify module IDs are correct (absolute path)
+    targets = {r.target_id for r in covers_rels}
+    assert any("module_a.py" in t for t in targets)
+    assert any("module_b.py" in t for t in targets)
+    
+    # Check metadata on relationships
+    rel_a = next(r for r in result.relationships if "module_a.py" in r.target_id and r.kind == RelationshipKind.COVERS)
+    assert rel_a.metadata["line-rate"] == "1.0"
+    assert rel_a.metadata["hits"] == "5/5"