Code Mirror Generation

akms.graph.generate_mirror

generate_mirror.py — Code Mirror Generation + Drift Check (§2.9 of system design).

Generates the code mirror — a search index that allows qmd to replace grep. Mirror nodes exist in the graph as simple existence markers (no edges, no tags).

Runs as part of the write-back cycle after update_graph.py. Only processes files modified in the current phase (from git diff).

Generation algorithm (legacy provider): 1. Parse source file with ast module 2. For each FunctionDef, AsyncFunctionDef, ClassDef: - Extract docstring → rendered as markdown (semantic search layer) - Extract full source → wrapped in `python block (literal search layer) 3. Write mirror file toknowledge/code-mirror/{module_path}.md`` 4. Write mirror node frontmatter (marker only)

Provider routing (A2-4): Public :func:generate_mirror dispatches through the mirror-provider protocol (default: legacy AST). The pure legacy body is :func:generate_mirror_legacy.

Docstring drift detection lives in :mod:akms.graph.drift and is re-exported here for backward-compatible imports.

check_docstring_drift_llm

check_docstring_drift_llm(
    definitions: list[dict], llm_fn: Any = None
) -> list[dict]

Check for semantic docstring drift using an LLM call.

Parameters:

Name	Type	Description	Default
definitions	list[dict]	Extracted definitions from extract_definitions().	required
llm_fn	Any	Callable (prompt: str) → str. If None, falls back to structural check.	None

Returns list of drift warning dicts with keys

function, type, detail, llm_response

Source code in packages/akms/src/akms/graph/drift.py

def check_docstring_drift_llm(
    definitions: list[dict],
    llm_fn: Any = None,
) -> list[dict]:
    """Check for semantic docstring drift using an LLM call.

    Args:
        definitions: Extracted definitions from extract_definitions().
        llm_fn: Callable (prompt: str) → str. If None, falls back to structural check.

    Returns list of drift warning dicts with keys:
        function, type, detail, llm_response
    """
    if llm_fn is None:
        return check_docstring_drift_structural(definitions)

    warnings: list[dict] = []

    for defn in definitions:
        if defn["type"] == "class":
            continue
        docstring = defn.get("docstring")
        if not docstring:
            continue

        params = defn.get("parameters", [])
        return_ann = defn.get("return_annotation")
        decorators = defn.get("decorators", [])

        prompt = (
            "Does this docstring accurately describe a function with these parameters "
            "and this return type? Reply YES or NO with one sentence.\n\n"
            f"Docstring:\n{docstring}\n\n"
            f"Parameters: {', '.join(params)}\n"
            f"Return type: {return_ann or 'not annotated'}\n"
            f"Decorators: {', '.join(decorators) or 'none'}"
        )

        try:
            response = llm_fn(prompt)
            response_stripped = response.strip().upper()
            if response_stripped.startswith("NO"):
                warnings.append(
                    {
                        "function": defn["name"],
                        "type": "llm_drift",
                        "detail": response.strip(),
                        "llm_response": response.strip(),
                    }
                )
        except Exception as e:
            logger.warning("LLM drift check failed for %s: %s", defn["name"], e)

    return warnings

check_docstring_drift_structural

check_docstring_drift_structural(
    definitions: list[dict],
) -> list[dict]

Check for structural docstring drift (pure heuristic, no LLM).

Catches the most obvious contradictions: - Docstring mentions parameters not in the signature - Docstring misses parameters that are in the signature (excluding self/cls) - Docstring mentions 'returns None' but function has a non-None annotation

For full semantic drift detection, use check_docstring_drift_llm() which requires an LLM API call.

Returns list of drift warning dicts.

Source code in packages/akms/src/akms/graph/drift.py

def check_docstring_drift_structural(definitions: list[dict]) -> list[dict]:
    """Check for structural docstring drift (pure heuristic, no LLM).

    Catches the most obvious contradictions:
    - Docstring mentions parameters not in the signature
    - Docstring misses parameters that are in the signature (excluding self/cls)
    - Docstring mentions 'returns None' but function has a non-None annotation

    For full semantic drift detection, use check_docstring_drift_llm()
    which requires an LLM API call.

    Returns list of drift warning dicts.
    """
    warnings: list[dict] = []

    for defn in definitions:
        if defn["type"] == "class":
            continue
        docstring = defn.get("docstring")
        if not docstring:
            continue

        params = defn.get("parameters", [])
        return_ann = defn.get("return_annotation")

        # Filter out self/cls
        real_params = [
            p for p in params if p not in ("self", "cls") and not p.startswith("*")
        ]

        doc_lower = docstring.lower()

        # Check for parameter name mismatches
        for param in real_params:
            if (
                len(param) > 2
                and param not in doc_lower
                and param.replace("_", " ") not in doc_lower
            ):
                warnings.append(
                    {
                        "function": defn["name"],
                        "type": "missing_param_in_docstring",
                        "detail": f"Parameter '{param}' not mentioned in docstring",
                    }
                )

        # Check return annotation contradiction
        if return_ann and return_ann != "None":
            if "returns none" in doc_lower or "return none" in doc_lower:
                warnings.append(
                    {
                        "function": defn["name"],
                        "type": "return_type_contradiction",
                        "detail": f"Docstring says 'returns None' but annotation is {return_ann}",
                    }
                )

    return warnings

extract_definitions

extract_definitions(source: str) -> list[dict[str, Any]]

Parse Python source and extract all top-level and class-level definitions.

Returns a list of dicts with keys

name, type (function|async_function|class), docstring, source, parameters, return_annotation, decorators

Source code in packages/akms/src/akms/graph/generate_mirror.py

def extract_definitions(source: str) -> list[dict[str, Any]]:
    """Parse Python source and extract all top-level and class-level definitions.

    Returns a list of dicts with keys:
        name, type (function|async_function|class), docstring, source,
        parameters, return_annotation, decorators
    """
    try:
        tree = ast.parse(source)
    except SyntaxError as e:
        logger.warning("SyntaxError parsing source: %s", e)
        return []

    source_lines = source.splitlines()
    definitions = []

    def _process_node(node: ast.AST, prefix: str = "") -> None:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            kind = (
                "async_function"
                if isinstance(node, ast.AsyncFunctionDef)
                else "function"
            )
            full_name = f"{prefix}{node.name}" if prefix else node.name
            definitions.append(
                {
                    "name": full_name,
                    "type": kind,
                    "docstring": _get_docstring(node),
                    "source": _get_source_segment(source_lines, node),
                    "parameters": _get_parameters(node),
                    "return_annotation": _get_return_annotation(node),
                    "decorators": _get_decorators(node),
                }
            )
        elif isinstance(node, ast.ClassDef):
            full_name = f"{prefix}{node.name}" if prefix else node.name
            definitions.append(
                {
                    "name": full_name,
                    "type": "class",
                    "docstring": _get_docstring(node),
                    "source": _get_source_segment(source_lines, node),
                    "parameters": [],
                    "return_annotation": None,
                    "decorators": _get_decorators(node),
                }
            )
            # Process methods within the class
            for child in ast.iter_child_nodes(node):
                if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    _process_node(child, prefix=f"{full_name}.")

    for node in ast.iter_child_nodes(tree):
        _process_node(node)

    return definitions

write_mirror_file

write_mirror_file(
    repo_root: Path,
    source_file: str,
    source_content: str,
    phase: int,
    generated_at: datetime | None = None,
) -> dict[str, Any] | None

Generate and write a mirror file for a single source file.

Parameters:

Name	Type	Description	Default
repo_root	Path	Path to the repository root.	required
source_file	str	Relative path to the source file (e.g., 'src/module.py').	required
source_content	str	Python source code to parse.	required
phase	int	Phase number that generated this mirror.	required
generated_at	datetime | None	Timestamp (defaults to now).	None

Returns:

Type	Description
dict[str, Any] | None	Dict with mirror info or None if no definitions found.

Source code in packages/akms/src/akms/graph/generate_mirror.py

def write_mirror_file(
    repo_root: Path,
    source_file: str,
    source_content: str,
    phase: int,
    generated_at: datetime | None = None,
) -> dict[str, Any] | None:
    """Generate and write a mirror file for a single source file.

    Args:
        repo_root: Path to the repository root.
        source_file: Relative path to the source file (e.g., 'src/module.py').
        source_content: Python source code to parse.
        phase: Phase number that generated this mirror.
        generated_at: Timestamp (defaults to now).

    Returns:
        Dict with mirror info or None if no definitions found.
    """
    if generated_at is None:
        generated_at = datetime.now()

    definitions = extract_definitions(source_content)
    if not definitions:
        logger.debug("No definitions found in %s — skipping mirror", source_file)
        return None

    # Format mirror content
    content = _format_mirror_content(source_file, definitions, phase, generated_at)

    # Generate frontmatter
    frontmatter_data = _generate_mirror_frontmatter(source_file, phase, generated_at)

    # Write mirror file
    mirror_path = (
        repo_root / "knowledge" / "code-mirror" / source_file.replace(".py", ".md")
    )
    mirror_path.parent.mkdir(parents=True, exist_ok=True)

    post = fm.Post(content)
    post.metadata = frontmatter_data
    with open(mirror_path, "wb") as f:
        fm.dump(post, f)

    logger.info("Wrote mirror: %s → %s", source_file, mirror_path)

    return {
        "source_file": source_file,
        "mirror_path": str(mirror_path),
        "node_id": frontmatter_data["id"],
        "definitions_count": len(definitions),
    }

get_changed_files

get_changed_files(
    repo_root: Path,
    parent_branch: str = "main",
    extensions: set[str] | None = None,
) -> list[str]

Get files changed between parent_branch and HEAD.

Parameters:

Name	Type	Description	Default
repo_root	Path	Repository root path.	required
parent_branch	str	Branch to diff against (default 'main').	'main'
extensions	set[str] | None	Set of extensions to include (default {'.py'}).	None

Returns:

Type	Description
list[str]	List of relative file paths.

Source code in packages/akms/src/akms/graph/generate_mirror.py

def get_changed_files(
    repo_root: Path,
    parent_branch: str = "main",
    extensions: set[str] | None = None,
) -> list[str]:
    """Get files changed between parent_branch and HEAD.

    Args:
        repo_root: Repository root path.
        parent_branch: Branch to diff against (default 'main').
        extensions: Set of extensions to include (default {'.py'}).

    Returns:
        List of relative file paths.
    """
    if extensions is None:
        extensions = {".py"}

    try:
        result = subprocess.run(
            ["git", "diff", f"{parent_branch}..HEAD", "--name-only"],
            capture_output=True,
            text=True,
            cwd=str(repo_root),
            timeout=30,
        )
        if result.returncode != 0:
            logger.warning("git diff failed: %s", result.stderr.strip())
            return []

        files = []
        for line in result.stdout.strip().splitlines():
            line = line.strip()
            if not line:
                continue
            if any(line.endswith(ext) for ext in extensions):
                files.append(line)

        return sorted(files)

    except FileNotFoundError:
        logger.warning("git not found — cannot determine changed files")
        return []
    except subprocess.TimeoutExpired:
        logger.warning("git diff timed out")
        return []
    except Exception as e:
        logger.warning("git diff error: %s", e)
        return []

generate_mirror_legacy

generate_mirror_legacy(
    repo_root: str | Path,
    phase: int,
    parent_branch: str = "main",
    source_files: list[str] | None = None,
    drift_check: bool = True,
    llm_fn: Any = None,
    generated_at: datetime | None = None,
) -> dict[str, Any]

Legacy in-process Python AST mirror generation (provider implementation body).

This is the historical algorithm, unchanged in behavior. Prefer the public :func:generate_mirror entry point which routes through the provider protocol.

Source code in packages/akms/src/akms/graph/generate_mirror.py

def generate_mirror_legacy(
    repo_root: str | Path,
    phase: int,
    parent_branch: str = "main",
    source_files: list[str] | None = None,
    drift_check: bool = True,
    llm_fn: Any = None,
    generated_at: datetime | None = None,
) -> dict[str, Any]:
    """Legacy in-process Python AST mirror generation (provider implementation body).

    This is the historical algorithm, unchanged in behavior. Prefer the public
    :func:`generate_mirror` entry point which routes through the provider protocol.
    """
    repo_root = Path(repo_root)
    if generated_at is None:
        generated_at = datetime.now()

    # Get files to process
    if source_files is None:
        source_files = get_changed_files(repo_root, parent_branch)

    mirrors = []
    drift_warnings = []
    definitions_total = 0

    for source_file in source_files:
        source_path = repo_root / source_file
        if not source_path.exists():
            logger.warning("Source file not found: %s", source_path)
            continue

        if not source_file.endswith(".py"):
            logger.debug("Skipping non-Python file: %s", source_file)
            continue

        try:
            source_content = source_path.read_text(encoding="utf-8")
        except Exception as e:
            logger.warning("Failed to read %s: %s", source_path, e)
            continue

        # Write mirror
        mirror_info = write_mirror_file(
            repo_root,
            source_file,
            source_content,
            phase,
            generated_at,
        )
        if mirror_info:
            mirrors.append(mirror_info)
            definitions_total += mirror_info["definitions_count"]

        # Drift check (provider-neutral module; structural unless llm_fn set)
        if drift_check:
            definitions = extract_definitions(source_content)
            if llm_fn is not None:
                file_warnings = check_docstring_drift_llm(definitions, llm_fn)
            else:
                file_warnings = check_docstring_drift_structural(definitions)

            for w in file_warnings:
                w["file"] = source_file
            drift_warnings.extend(file_warnings)

    summary = {
        "mirrors": mirrors,
        "drift_warnings": drift_warnings,
        "files_processed": len(source_files),
        "definitions_total": definitions_total,
        "provider": "legacy",
        "success": True,
        "fallback_used": False,
        "errors": [],
        "provider_metadata": {
            "kind": "in_process",
            "language_coverage": ["python"],
        },
    }

    logger.info(
        "generate_mirror_legacy: %d files → %d mirrors, %d definitions, %d drift warnings",
        len(source_files),
        len(mirrors),
        definitions_total,
        len(drift_warnings),
    )

    return summary

generate_mirror

generate_mirror(
    repo_root: str | Path,
    phase: int,
    parent_branch: str = "main",
    source_files: list[str] | None = None,
    drift_check: bool = True,
    llm_fn: Any = None,
    *,
    config: Any = None,
    provider_name: str | None = None,
) -> dict[str, Any]

Generate code mirror files via the configured mirror provider.

Default provider is legacy (Python AST), preserving historical behavior for callers that omit config / provider_name.

Parameters:

Name	Type	Description	Default
repo_root	str | Path	Path to the repository root.	required
phase	int	Current phase number.	required
parent_branch	str	Branch to diff against (default 'main').	'main'
source_files	list[str] | None	Explicit list of source files (overrides git diff).	None
drift_check	bool	Whether to run drift detection.	True
llm_fn	Any	Optional LLM callable for semantic drift check.	None
config	Any	Optional PropagationConfig or MirrorConfig.	None
provider_name	str | None	Override provider name (default from config / legacy).	None

Returns:

Type	Description
dict[str, Any]	Dict with keys:
dict[str, Any]	{ "mirrors": [...], "drift_warnings": [...], "files_processed": int, "definitions_total": int, additive provider fields: "provider": str, "provider_metadata": dict, "errors": list, "success": bool, "fallback_used": bool,
dict[str, Any]	}

Source code in packages/akms/src/akms/graph/generate_mirror.py

@traced("akms.generate_mirror")
def generate_mirror(
    repo_root: str | Path,
    phase: int,
    parent_branch: str = "main",
    source_files: list[str] | None = None,
    drift_check: bool = True,
    llm_fn: Any = None,
    *,
    config: Any = None,
    provider_name: str | None = None,
) -> dict[str, Any]:
    """Generate code mirror files via the configured mirror provider.

    Default provider is ``legacy`` (Python AST), preserving historical behavior
    for callers that omit *config* / *provider_name*.

    Args:
        repo_root: Path to the repository root.
        phase: Current phase number.
        parent_branch: Branch to diff against (default 'main').
        source_files: Explicit list of source files (overrides git diff).
        drift_check: Whether to run drift detection.
        llm_fn: Optional LLM callable for semantic drift check.
        config: Optional ``PropagationConfig`` or ``MirrorConfig``.
        provider_name: Override provider name (default from config / legacy).

    Returns:
        Dict with keys:
        {
            "mirrors": [...],
            "drift_warnings": [...],
            "files_processed": int,
            "definitions_total": int,
            # additive provider fields:
            "provider": str,
            "provider_metadata": dict,
            "errors": list,
            "success": bool,
            "fallback_used": bool,
        }
    """
    # Local import: mirror_provider imports providers which call back into
    # generate_mirror_legacy — keep the cycle at function scope.
    from akms.graph.mirror_provider import (
        MirrorProviderError,
        MirrorRequest,
        resolve_mirror_config,
        run_mirror_provider,
    )

    mirror_cfg = resolve_mirror_config(config)
    # Fast path: pure legacy with no config override keeps the historical
    # call shape and avoids registry overhead for the default case.
    name = (provider_name or mirror_cfg.provider or "legacy").strip().lower()
    if name == "legacy" and config is None and provider_name is None:
        return generate_mirror_legacy(
            repo_root=repo_root,
            phase=phase,
            parent_branch=parent_branch,
            source_files=source_files,
            drift_check=drift_check,
            llm_fn=llm_fn,
        )

    request = MirrorRequest(
        repo_root=Path(repo_root),
        phase=phase,
        parent_branch=parent_branch,
        source_files=source_files,
        selection_mode=mirror_cfg.selection_mode or "changed",
        drift_check=drift_check,
        llm_fn=llm_fn,
        prune=mirror_cfg.prune,
        force_lock=mirror_cfg.force_lock,
    )
    try:
        result = run_mirror_provider(
            request,
            mirror_cfg,
            provider_name=provider_name,
        )
    except MirrorProviderError:
        raise
    return result.to_dict()