@router.get("/cluster")
def get_cluster(
granularity: int = Query(50, ge=10, le=90, description="Cluster granularity (10=few large, 90=many small)"),
project: Optional[str] = Query(None),
max_nodes: int = Query(0, ge=0, description="LOD node budget: 0=full graph, >0 collapses to super-nodes"),
bbox: Optional[str] = Query(None, description="Viewport bbox 'x0,y0,x1,y1' in t-SNE space for viewport-LOD detail"),
q: Optional[str] = Query(None, description="Search query: matching entries are force-kept as real nodes under LOD"),
request: Request = None,
response: Response = None,
):
"""Return the 2D t-SNE projection of entry embeddings for the cluster view.
When ``max_nodes > 0`` the payload is reduced to a level-of-detail view: the
lowest-ranked nodes collapse into semantic super-nodes (see
:func:`_apply_lod_to_result`) and an ``lod`` metadata block is included. The
full projection/clustering is computed and cached once; the LOD collapse is a
cheap per-request pass keyed additionally on ``max_nodes`` + quantized ``bbox``.
"""
# When invoked directly (internal callers / tests) rather than through
# FastAPI's dependency resolution, the ``Query(...)`` defaults arrive as
# sentinel objects instead of resolved values, so a bare ``max_nodes > 0``
# would raise ``TypeError``. Coerce any unresolved parameter back to its
# declared default so this route is safe to call as a plain function.
if not isinstance(granularity, int):
granularity = 50
if not isinstance(max_nodes, int):
max_nodes = 0
if not isinstance(project, str):
project = None
if not isinstance(bbox, str):
bbox = None
if not isinstance(q, str):
q = None
# Quantize the viewport bbox to a coarse tile so a smooth pan reuses a handful
# of stable ETag keys instead of a fresh one every pixel. ``lod_key`` folds the
# LOD terms into the ETag so tiers/tiles are cached distinctly from one another
# and from the full (max_nodes=0) payload.
qbbox = quantize_bbox(parse_bbox(bbox)) if max_nodes and max_nodes > 0 else None
# A search query only matters under LOD (it force-keeps matches as real nodes);
# fold it into the ETag key so a search change is served fresh instead of a 304
# for the non-search payload. Normalized so trivial whitespace/case reuse cache.
qnorm = (q or "").strip().lower() if max_nodes and max_nodes > 0 else ""
lod_key = f"{max_nodes}:{qbbox if qbbox is not None else '-'}:{qnorm or '-'}"
# Compute the exact LOCAL fileset signature ONCE and prime the loader with it
# BEFORE reading ``loader.graph``, so a delete-of-a-non-newest entry or a
# same-count/older-mtime swap rebuilds the graph immediately (not bounded by
# ``_SCAN_MIN_INTERVAL``). The same signature keys ``_cluster_cache`` below.
local_sig = _local_fileset_signature()
_ensure_loader_local_fresh(local_sig)
try:
graph = loader.graph
except FileNotFoundError:
return {"entries": [], "edges": [], "clusters": []}
proj = _get_project(project)
project_id = proj["id"]
# Compute EVERY cache-key term up front from cheap stats, then check the cache
# and RETURN early on a hit — BEFORE running ``graph.cypher(...)``,
# ``_federated_projection(...)``, or reading the embeddings sidecar. This keeps
# the cheap-stat fast path intact: a cache hit costs only a handful of stats,
# never a cypher scan / federation walk / full sidecar read+parse (which
# would otherwise run on every request, defeating the cache and its 200k-node
# performance goal). The heavy work below runs ONLY on a miss.
#
# The LOCAL fileset signature replaces the former newest-mtime term
# (``MEMORY_FILE.stat().st_mtime``): the newest mtime is blind to a
# delete-of-a-non-newest entry or a same-count/older-mtime swap, so a
# local delete/swap would otherwise keep serving a stale cached cluster. The
# federation + embeddings + granularity terms are UNCHANGED (federation
# delete-robustness is a separate concern, out of scope here).
config_mtime = str((Path(".memory") / "config.yaml").stat().st_mtime) if (Path(".memory") / "config.yaml").exists() else ""
federated_mtime = str(federation.latest_mtime())
federated_emb_mtime = str(federation.latest_embeddings_mtime())
# Stat the embeddings sidecar mtime BEFORE reading it (below, on a miss), so a
# cache hit skips the sidecar read+parse entirely. Stat-before-read is correct
# (no stale serve): a concurrent write landing BETWEEN this stat and the later
# read caches NEWER content under the OLDER mtime, so the NEXT request stats
# the newer mtime and MISSES (a spurious recompute) — never a false hit on
# stale content. Integer ``st_mtime_ns`` (not float ``st_mtime``) matches the
# precision used by ``tree._local_fileset_signature`` and will not collapse two
# sidecar writes within one coarse (NFS/SMB) mtime tick to the same key.
#
# Accepted residual limitation: an mtime-keyed cache cannot observe a content
# change that does NOT advance the sidecar's mtime — a coarse-FS same-tick
# re-embed, or an mtime-preserving restore / backward clock. Closing it would
# require hashing the sidecar's contents on EVERY request (including hits),
# defeating the cheap-stat fast path this cache exists for, so it is
# deliberately out of scope.
from memory.storage import EMBEDDINGS_FILE
emb_mtime = str(EMBEDDINGS_FILE.stat().st_mtime_ns) if EMBEDDINGS_FILE.exists() else ""
# ``local_sig`` is ``"<path>|<mtime_ns>"`` components joined by ``\x00``; on
# Windows a path carries a drive-letter ``':'`` (``C:\...``) that would
# collide with this ``':'``-delimited key's field boundaries (the old numeric
# mtime term could not). Hash it to fixed-width hex first (mirrors tree.py's
# ``_tree_cache_key``) so no raw path-bearing string lands inside the key.
local_sig_key = hashlib.blake2b(
local_sig.encode("utf-8", "surrogatepass"), digest_size=16
).hexdigest()
cache_key = f"{project_id}:{local_sig_key}:{emb_mtime}:{config_mtime}:{federated_mtime}:{federated_emb_mtime}:{granularity}"
if cache_key in _cluster_cache:
# ETag / conditional-GET on the FINAL cached layout. A cached entry is
# always the settled (non-refining) projection, so its ETag carries the
# ":final" state — distinct from the ":refining" ETag served while PCA is
# provisional, which is exactly what lets a client that last saw the
# provisional layout receive a 200 upgrade (different ETag) instead of a
# spurious 304. The ``lod_key`` distinguishes each LOD tier/tile so a client
# on one zoom level never gets a 304 for another's payload. A matching
# If-None-Match short-circuits to a bodyless 304, skipping the multi-MB
# re-serialize on the server and re-parse on the client.
not_modified = _conditional_etag(request, response, f"{cache_key}:{lod_key}:final")
if not_modified is not None:
return not_modified
return _apply_lod_to_result(_cluster_cache[cache_key], cache_key, max_nodes, qbbox, qnorm)
entries = list(graph.cypher(
"MATCH (e:entry) WHERE e.project = $proj AND e.status = $st "
"RETURN e.id AS id, e.type AS type, e.title AS title, "
"e.node_label AS node_label, e.parent_id AS parent_id, "
"e.success AS success, e.created_at AS created_at, "
"e.tags AS tags",
params={"proj": project_id, "st": "active"},
))
fed = _federated_projection(project_id)
entries.extend(fed["entries"])
if not entries:
return {"entries": [], "edges": [], "clusters": []}
try:
embs = graph.embeddings("entry", "search_text")
except Exception:
embs = {}
if not embs:
from memory import storage as _storage
embs = _storage.load_embeddings_sidecar()
# Merge federated repos' embeddings (namespaced to <repo_id>:<entry_id>) so
# federated entries participate in the t-SNE/clustering projection instead of
# falling through to the "no embeddings → cluster -1 (other)" path.
fed_embeddings = fed.get("embeddings") or {}
if fed_embeddings:
embs = {**embs, **fed_embeddings}
entry_ids = []
vectors = []
entry_map = {}
# t-SNE requires every vector to share the same dimension. All repos use the
# same embed model today, but a federated repo on a different model would
# otherwise break np.array() and take down the whole view — so we lock onto
# the first dimension seen and skip any mismatches (they fall through to the
# "no embedding → cluster -1" path like an un-embedded entry).
expected_dim: Optional[int] = None
for e in entries:
entry_map[e["id"]] = e
vec = embs.get(e["id"])
if vec is None:
continue
if expected_dim is None:
expected_dim = len(vec)
if len(vec) != expected_dim:
continue
entry_ids.append(e["id"])
vectors.append(vec)
# Report repos whose entries are present but have no usable embeddings, so the
# UI can explain "no entries to cluster" instead of silently dropping them.
# repo_id is None/absent for the local repo; federated entries carry repo_id.
embedded_repo_ids = {entry_map[eid].get("repo_id") for eid in entry_ids}
repos_present: dict[str, str] = {}
local_present = False
for e in entries:
rid = e.get("repo_id")
if rid:
repos_present[rid] = e.get("repo_name") or rid
else:
local_present = True
repos_without_embeddings = [
{"id": rid, "name": name}
for rid, name in repos_present.items()
if rid not in embedded_repo_ids
]
if local_present and None not in embedded_repo_ids:
repos_without_embeddings.insert(
0, {"id": "local", "name": proj.get("name") or "this repo"}
)
if len(vectors) < 2:
result_entries = []
for i, e in enumerate(entries):
result_entries.append({
"id": e["id"],
"type": e["type"],
"title": e["title"],
"node_label": e.get("node_label") or e["title"],
"parent_id": e.get("parent_id"),
"success": e.get("success"),
"repo_id": e.get("repo_id"),
"repo_name": e.get("repo_name"),
"repo_path": e.get("repo_path"),
"read_only": bool(e.get("read_only", False)),
"x": float(i * 100),
"y": 0.0,
"cluster": -1,
"cluster_label": "other",
})
return {
"entries": result_entries,
"edges": [],
"clusters": [],
"max_clusters": 0,
"reposWithoutEmbeddings": repos_without_embeddings,
}
X = np.array(vectors, dtype=np.float32)
# Cold-start responsiveness: serve an instant PCA projection now and refine to
# t-SNE in the background (see the module-level note). ``projection_kind`` is
# "tsne" once the cached coords are ready, "pca" while refining.
emb_sig = _embedding_signature(project_id, emb_mtime, federated_emb_mtime, entry_ids)
coords, projection_kind = _get_or_start_projection(X, entry_ids, emb_sig)
# Map the granularity slider monotonically to a target cluster count so the
# control behaves predictably: "Few" (10) → 2 clusters, "Many" (90) → k_max.
# DBSCAN's eps made the cluster count peak in the middle of the slider (a tiny
# eps shatters everything into noise — a whole-graph sea of grey — while a huge
# eps merges it into one blob), so we use a target-count clustering
# (``_cluster_labels``: agglomerative under CLUSTER_MAX_NODES, MiniBatchKMeans
# above it) whose count is monotonic in the slider by construction and never
# collapses to all-noise. Grey now means only "no embedding for this entry".
n_pts = len(vectors)
# Cap the cluster ceiling at 30 (or n_pts-1). Past the color-palette length
# colors repeat, but that's an accepted trade-off for finer granularity.
k_max = max(2, min(n_pts - 1, 30))
frac = min(1.0, max(0.0, (granularity - 10) / 80.0))
n_clusters = int(round(2 + frac * (k_max - 2)))
n_clusters = max(2, min(n_clusters, n_pts - 1))
labels = _cluster_labels(coords, n_clusters)
# Auto-label clusters from member tags and titles
cluster_members: dict[int, list[int]] = {}
for i, label in enumerate(labels):
cluster_members.setdefault(int(label), []).append(i)
cluster_labels: dict[int, str] = {-1: "other"}
skip_tags = {
"checkpoint", "decision", "experiment", "note", "annotation",
"plan", "todo", "subproject", "phase", "phase1", "phase2",
"phase3", "phase4", "phase5", "phase6", "session-end",
"milestone", "complete", "phase1-complete", "phase2-complete",
}
for cid, member_indices in cluster_members.items():
if cid == -1:
continue
tag_counter: Counter = Counter()
for idx in member_indices:
eid = entry_ids[idx]
e = entry_map[eid]
raw_tags = e.get("tags") or ""
for tag in raw_tags.split(","):
tag = tag.strip().lower()
if tag and tag not in skip_tags:
tag_counter[tag] += 1
if tag_counter:
top_tags = [t for t, _ in tag_counter.most_common(2)]
cluster_labels[cid] = " + ".join(top_tags)
else:
titles = [entry_map[entry_ids[i]].get("node_label") or entry_map[entry_ids[i]]["title"] for i in member_indices[:3]]
cluster_labels[cid] = titles[0][:30] if titles else f"cluster {cid}"
id_to_cluster: dict[str, int] = {}
for i, eid in enumerate(entry_ids):
id_to_cluster[eid] = int(labels[i])
result_entries = []
id_to_pos = {}
for i, eid in enumerate(entry_ids):
e = entry_map[eid]
x, y = float(coords[i, 0]), float(coords[i, 1])
id_to_pos[eid] = (x, y)
cid = id_to_cluster[eid]
result_entries.append({
"id": eid,
"type": e["type"],
"title": e["title"],
"node_label": e.get("node_label") or e["title"],
"parent_id": e.get("parent_id"),
"success": e.get("success"),
"repo_id": e.get("repo_id"),
"repo_name": e.get("repo_name"),
"repo_path": e.get("repo_path"),
"read_only": bool(e.get("read_only", False)),
"x": x,
"y": y,
"cluster": cid,
"cluster_label": cluster_labels.get(cid, "other"),
})
# Entries without embeddings get placed near their parent
for e in entries:
if e["id"] not in id_to_pos:
parent_pos = id_to_pos.get(e.get("parent_id", ""), (500.0, 500.0))
x = parent_pos[0] + np.random.uniform(-30, 30)
y = parent_pos[1] + np.random.uniform(-30, 30)
id_to_pos[e["id"]] = (x, y)
parent_cluster = id_to_cluster.get(e.get("parent_id", ""), -1)
id_to_cluster[e["id"]] = parent_cluster
result_entries.append({
"id": e["id"],
"type": e["type"],
"title": e["title"],
"node_label": e.get("node_label") or e["title"],
"parent_id": e.get("parent_id"),
"success": e.get("success"),
"repo_id": e.get("repo_id"),
"repo_name": e.get("repo_name"),
"repo_path": e.get("repo_path"),
"read_only": bool(e.get("read_only", False)),
"x": float(x),
"y": float(y),
"cluster": parent_cluster,
"cluster_label": cluster_labels.get(parent_cluster, "other"),
})
positioned_ids = set(id_to_pos.keys())
# Related edges (first-class in cluster view)
related_edges = list(graph.cypher(
"MATCH (a:entry)-[:related]->(b:entry) "
"WHERE a.project = $proj "
"RETURN a.id AS source, b.id AS target",
params={"proj": project_id},
))
edges = [
{"source": e["source"], "target": e["target"], "type": "related"}
for e in related_edges
if e["source"] in positioned_ids and e["target"] in positioned_ids
]
edges.extend([
{"source": e["source"], "target": e["target"], "type": "related"}
for e in fed["edges"]
if e["type"] == "related" and e["source"] in positioned_ids and e["target"] in positioned_ids
])
# Parent-child edges (faint in cluster view) — includes secondary parents
contains_edges_cluster = list(graph.cypher(
"MATCH (src:entry)-[:contains]->(tgt:entry) "
"WHERE src.project = $proj "
"RETURN src.id AS source, tgt.id AS target",
params={"proj": project_id},
))
seen_pc: set[tuple[str, str]] = set()
for e in entries:
if e.get("parent_id") and e["parent_id"] in positioned_ids and e["id"] in positioned_ids:
key = (e["parent_id"], e["id"])
if key not in seen_pc:
seen_pc.add(key)
edges.append({
"source": e["parent_id"],
"target": e["id"],
"type": "parent_child",
})
for ce in contains_edges_cluster:
if ce["source"] in positioned_ids and ce["target"] in positioned_ids:
key = (ce["source"], ce["target"])
if key not in seen_pc:
seen_pc.add(key)
edges.append({
"source": ce["source"],
"target": ce["target"],
"type": "parent_child",
})
clusters_summary = [
{"id": cid, "label": cluster_labels.get(cid, "other"), "count": len(members)}
for cid, members in sorted(cluster_members.items()) if cid != -1
]
result = {
"entries": result_entries,
"edges": edges,
"clusters": clusters_summary,
"max_clusters": k_max,
"reposWithoutEmbeddings": repos_without_embeddings,
# True ONLY for the provisional "pca" layout (t-SNE computing in the
# background); the cluster view polls every 5s and upgrades automatically.
# "pca_final" (t-SNE capped out by size) and "tsne" are both final → False,
# so the "Refining layout…" pill never spins forever on huge graphs.
"refining": projection_kind == "pca",
}
# Cache the FINAL layout only ("tsne", or "pca_final" when t-SNE is capped).
# Never cache the provisional "pca" result — it would pin until the inputs
# change, so the background refine would never surface.
if projection_kind != "pca":
_cluster_cache.clear()
# The fine cluster assignment is derived from the base layout, so it is
# invalidated in lockstep — a stale fine map would drill against the wrong
# positions after a re-projection.
_fine_cluster_cache.clear()
if cache_key:
_cluster_cache[cache_key] = result
# ETag / conditional-GET. The ETag folds in the projection STATE
# (":refining" vs ":final") on top of the input-derived cache_key, so:
# • unchanged inputs while still refining → same ETag → 304 (client keeps
# the provisional PCA layout; the refining pill stays up, no re-ship);
# • the PCA→t-SNE (or →pca_final) upgrade → state flips to ":final" → new
# ETag → 200 carrying the settled layout (the refine surfaces);
# • changed inputs → new cache_key → new ETag → 200.
# Keying on cache_key+state (not a body hash) mirrors the tree route and keeps
# the provisional PCA frozen client-side (the un-embedded-node jitter from
# np.random positions never re-ships), which is the desirable behaviour.
etag_state = "refining" if result["refining"] else "final"
not_modified = _conditional_etag(request, response, f"{cache_key}:{lod_key}:{etag_state}")
if not_modified is not None:
return not_modified
# Only memoize the fine clusters against a FINAL (cached) base layout: a
# provisional PCA layout is transient and its positions will shift when t-SNE
# lands under the same key, so recompute fine clusters each poll while refining
# (a falsy cache_key skips the memo) rather than pin a stale map.
fine_cache_key = cache_key if projection_kind != "pca" else ""
return _apply_lod_to_result(result, fine_cache_key, max_nodes, qbbox, qnorm)