QMD Cache

akms.graph.qmd_cache

qmd_cache.py — QMD query cache layer (§2.6 of system design).

Caches qmd search results by (graph_version, query_hash). Cache is invalidated when graph_version changes.

Graph version = SHA256 of graph.json contents, minus the generated_at timestamp, so recompiling identical inputs keeps the same version. Query hash = SHA256 of (tags, role, depth) — see query_subgraph.compute_query_hash.

Cache storage: /knowledge/graph/.qmd_cache/ Each entry: {graph_version}_{query_hash}.json

compute_graph_version

compute_graph_version(graph_json_path: str | Path) -> str

Compute the graph version: a SHA256 of graph.json's content.

The graph.generated_at timestamp is left out, so recompiling the same inputs yields the same version (and the same resolution fingerprint). The rest of the document is hashed as canonical JSON (sorted keys, compact separators), so whitespace differences do not change the version either. A file that is not a JSON object is hashed byte for byte.

Parameters:

Name	Type	Description	Default
graph_json_path	str | Path	Path to graph.json.	required

Returns:

Type	Description
str	SHA256 hex digest, or "no-graph" when the file is missing.

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

def compute_graph_version(graph_json_path: str | Path) -> str:
    """Compute the graph version: a SHA256 of graph.json's content.

    The ``graph.generated_at`` timestamp is left out, so recompiling the same
    inputs yields the same version (and the same resolution fingerprint). The
    rest of the document is hashed as canonical JSON (sorted keys, compact
    separators), so whitespace differences do not change the version either.
    A file that is not a JSON object is hashed byte for byte.

    Args:
        graph_json_path: Path to graph.json.

    Returns:
        SHA256 hex digest, or ``"no-graph"`` when the file is missing.
    """
    path = Path(graph_json_path)
    if not path.exists():
        return "no-graph"

    content = path.read_bytes()
    try:
        data = json.loads(content)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return hashlib.sha256(content).hexdigest()
    if not isinstance(data, dict):
        return hashlib.sha256(content).hexdigest()

    graph_meta = data.get("graph")
    if isinstance(graph_meta, dict) and "generated_at" in graph_meta:
        stable_meta = {k: v for k, v in graph_meta.items() if k != "generated_at"}
        data = {**data, "graph": stable_meta}
    canonical = json.dumps(
        data, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

get_cached

get_cached(
    repo_root: str | Path,
    graph_version: str,
    query_hash: str,
) -> list | None

Retrieve cached query results if available.

Parameters:

Name	Type	Description	Default
repo_root	str | Path	Repository root path.	required
graph_version	str	SHA256 of current graph.json.	required
query_hash	str	SHA256 of query parameters.	required

Returns:

Type	Description
list | None	Cached result list, or None on cache miss.

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

def get_cached(
    repo_root: str | Path,
    graph_version: str,
    query_hash: str,
) -> list | None:
    """Retrieve cached query results if available.

    Args:
        repo_root: Repository root path.
        graph_version: SHA256 of current graph.json.
        query_hash: SHA256 of query parameters.

    Returns:
        Cached result list, or None on cache miss.
    """
    cache = _cache_dir(repo_root)
    key = _cache_key(graph_version, query_hash)
    cache_file = cache / key

    if not cache_file.exists():
        return None

    try:
        with open(cache_file) as f:
            data = json.load(f)

        # Validate the stored graph version matches
        if data.get("graph_version") != graph_version:
            logger.info("Cache version mismatch, invalidating %s", key)
            cache_file.unlink(missing_ok=True)
            return None

        logger.info("Cache hit: %s", key)
        return data.get("results")
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Cache read error for %s: %s", key, e)
        cache_file.unlink(missing_ok=True)
        return None

put_cached

put_cached(
    repo_root: str | Path,
    graph_version: str,
    query_hash: str,
    results: list,
) -> Path

Store query results in cache.

Parameters:

Name	Type	Description	Default
repo_root	str | Path	Repository root path.	required
graph_version	str	SHA256 of current graph.json.	required
query_hash	str	SHA256 of query parameters.	required
results	list	The query results to cache (list of node_id strings).	required

Returns:

Type	Description
Path	Path to the cache file.

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

def put_cached(
    repo_root: str | Path,
    graph_version: str,
    query_hash: str,
    results: list,
) -> Path:
    """Store query results in cache.

    Args:
        repo_root: Repository root path.
        graph_version: SHA256 of current graph.json.
        query_hash: SHA256 of query parameters.
        results: The query results to cache (list of node_id strings).

    Returns:
        Path to the cache file.
    """
    cache = _cache_dir(repo_root)
    cache.mkdir(parents=True, exist_ok=True)

    key = _cache_key(graph_version, query_hash)
    cache_file = cache / key

    data = {
        "graph_version": graph_version,
        "query_hash": query_hash,
        "results": results,
    }

    with open(cache_file, "w") as f:
        json.dump(data, f, sort_keys=True, indent=2)

    logger.info("Cache put: %s (%d results)", key, len(results))
    return cache_file

invalidate_cache

invalidate_cache(repo_root: str | Path) -> int

Remove all cached entries.

Called when graph.json is rebuilt to avoid stale results.

Returns:

Type	Description
int	Number of cache files removed.

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

def invalidate_cache(repo_root: str | Path) -> int:
    """Remove all cached entries.

    Called when graph.json is rebuilt to avoid stale results.

    Returns:
        Number of cache files removed.
    """
    cache = _cache_dir(repo_root)
    if not cache.exists():
        return 0

    count = 0
    for f in cache.glob("*.json"):
        f.unlink()
        count += 1

    logger.info("Invalidated %d cache entries", count)
    return count

is_qmd_available

is_qmd_available() -> bool

Check if qmd CLI is installed and accessible.

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

def is_qmd_available() -> bool:
    """Check if qmd CLI is installed and accessible."""
    return shutil.which("qmd") is not None