`memory.dashboard.backend.routes.code`¶

memory.dashboard.backend.routes.code ¶

Code-structure lens route.

Serves GET /code: a muted code flow-graph backdrop (files → symbols, with inter-file CALLS / IMPORTS as the layout edges) with the memory tether overlay lit on top (each tethered anchor's rolling tip + provenance chain).

Mirrors :mod:.cluster: a signature-keyed module cache guarded by :func:_conditional_etag, and a cold-start that serves a cheap first response (backdrop + only already-warm anchors, refining: true) while the expensive tether invert warms in a background thread and the next poll upgrades to the full overlay (refining: false). The provisional response is NEVER cached as final.

The dashboard backend is a SEPARATE process from the memory MCP server, so it builds its OWN code graph (via kglite._kglite_code_tree.build), cached keyed on git HEAD — mirroring memory/server.py::_ensure_code_graph.

See documents/260721_code-structure-lens-design.md for the full design.

get_code ¶

get_code(project: Optional[str] = Query(None), request: Request = None, response: Response = None)

Code-structure lens: the muted code flow-graph backdrop + tether overlay.

See the module docstring and documents/260721_code-structure-lens-design.md for the response shape and the cold-start/refining contract.

Source code in memory/dashboard/backend/routes/code.py

@router.get("/code")
def get_code(
    project: Optional[str] = Query(None),
    request: Request = None,
    response: Response = None,
):
    """Code-structure lens: the muted code flow-graph backdrop + tether overlay.

    See the module docstring and ``documents/260721_code-structure-lens-design.md``
    for the response shape and the cold-start/refining contract.
    """
    empty = {
        "nodes": [], "edges": [], "regions": [],
        "code_head": None, "refining": False, "latest_activity": None,
    }

    # Prime the loader with the exact local signature BEFORE reading its graph
    # (a delete/older-mtime swap the loader's mtime gate misses would otherwise
    # serve — and cache — a stale overlay). Same discipline as cluster.py/tree.py.
    local_sig = _local_fileset_signature()
    _ensure_loader_local_fresh(local_sig)
    try:
        mem_graph = loader.graph
    except FileNotFoundError:
        return empty

    proj = _get_project(project)
    project_id = proj["id"]

    cgraph, code_head = _ensure_code_graph()
    if cgraph is None:
        return {**empty, "error": "code graph unavailable"}

    # Cache key: code HEAD (backdrop invalidation) + the memory fileset signature
    # (overlay invalidation) + project. The concatenation is hashed to a fixed
    # width (mirrors cluster.py's local_sig_key) so a path-bearing signature never
    # collides with the ':'-delimited key fields.
    mem_sig = storage.source_fileset_signature()
    sig_key = hashlib.blake2b(
        f"{code_head}\x00{mem_sig}".encode("utf-8", "surrogatepass"), digest_size=16
    ).hexdigest()
    cache_key = f"{project_id}:{sig_key}"

    if cache_key in _code_cache:
        # A cached entry is always the settled (non-refining) overlay, so its ETag
        # carries ":final" — distinct from the ":refining" ETag served while the
        # invert warms, which is what lets a client that last saw the provisional
        # overlay get a 200 upgrade (different ETag) rather than a spurious 304.
        not_modified = _conditional_etag(request, response, f"{cache_key}:final")
        if not_modified is not None:
            return not_modified
        return _code_cache[cache_key]

    # Backdrop is memoized per cache_key (so the cold-start refining polls, which
    # never cache their response, don't re-scan the whole graph each poll) and
    # deep-copied so the in-place overlay never corrupts the cached copy.
    backdrop = _get_backdrop(cgraph, code_head)
    root = _git_repo_root()
    entries = _collect_pinned_entries(mem_graph, project_id)
    tethers, refining = _get_or_start_tethers(cgraph, root, entries, code_head, cache_key)
    latest_activity = _apply_overlay(backdrop, tethers)

    result = {
        "nodes": backdrop["nodes"],
        "edges": backdrop["edges"],
        "regions": backdrop["regions"],
        "code_head": (code_head or "")[:12] or None,
        "refining": refining,
        # Only surface latest-activity on the FINAL (settled) response, over the
        # full pinned set — during refining the overlay is a partial warm subset,
        # so a ring computed then would jump when the full set lands.
        "latest_activity": latest_activity if not refining else None,
        "project": project_id,
    }

    # Cache the FINAL overlay only. Never cache the provisional (refining) result:
    # it would pin until the inputs change, so the background warm would never
    # surface (mirrors cluster.py's PCA rule).
    if not refining:
        _code_cache.clear()
        _code_cache[cache_key] = result

    etag_state = "refining" if refining else "final"
    not_modified = _conditional_etag(request, response, f"{cache_key}:{etag_state}")
    if not_modified is not None:
        return not_modified
    return result

memory.dashboard.backend.routes.code¶