musehub_cross_repo.py
python
sha256:e519738f2102d625d47828a17d113283cd744577b43551013cb668fd04f1c3fe
chore(mwp8/phase5): tick all RG checkboxes; close-out doc (…
Sonnet 4.6
18 days ago
| 1 | """Cross-repo symbol intelligence — Phase 6. |
| 2 | |
| 3 | Service functions that operate across all repos owned by a given user, |
| 4 | building a workspace-level view of symbol coupling and blast-radius. |
| 5 | |
| 6 | All operations are read-only and work against the existing MusehubSymbolIndex |
| 7 | msgpack blobs — no new DB models are required. The "workspace" is implicitly |
| 8 | defined as the set of non-deleted repos whose `owner` field equals the |
| 9 | requesting user's username. |
| 10 | |
| 11 | Key functions |
| 12 | ------------- |
| 13 | search_symbol_across_repos — find a symbol address in all owner repos |
| 14 | cross_repo_impact — blast radius that crosses repo boundaries |
| 15 | workspace_blast_risk_top_n — highest co-change symbols across workspace |
| 16 | build_deps_graph — derive file-level dependency graph from co-change |
| 17 | """ |
| 18 | |
| 19 | import logging |
| 20 | from collections import defaultdict |
| 21 | from dataclasses import dataclass, field |
| 22 | |
| 23 | from sqlalchemy import select |
| 24 | from sqlalchemy.ext.asyncio import AsyncSession |
| 25 | |
| 26 | from musehub.db.musehub_repo_models import MusehubRepo |
| 27 | from musehub.services.musehub_symbol_indexer import load_symbol_history |
| 28 | from musehub.types.json_types import IntDict |
| 29 | |
| 30 | type CommitToPrefixesDict = dict[str, set[str]] |
| 31 | type CrossNodeMetaDict = dict[str, tuple[str, str]] |
| 32 | |
| 33 | logger = logging.getLogger(__name__) |
| 34 | |
| 35 | # ── Data structures ─────────────────────────────────────────────────────────── |
| 36 | |
| 37 | @dataclass |
| 38 | class CrossRepoMatch: |
| 39 | """A symbol address found in one of the owner's repos.""" |
| 40 | repo_id: str |
| 41 | repo_slug: str |
| 42 | address: str |
| 43 | last_op: str |
| 44 | co_change_count: int |
| 45 | |
| 46 | @dataclass |
| 47 | class ExternalImpact: |
| 48 | """Cross-repo impact entry: symbols from another repo that co-change with |
| 49 | the target address within a shared time window.""" |
| 50 | repo_id: str |
| 51 | repo_slug: str |
| 52 | matches: list[dict] # [{address, shared_commits}] |
| 53 | |
| 54 | @dataclass |
| 55 | class CrossRepoImpact: |
| 56 | """Result of cross_repo_impact().""" |
| 57 | address: str |
| 58 | source_repo_id: str |
| 59 | source_repo_slug: str |
| 60 | local_co_changed: list[dict] # [{address, shared_commits}] |
| 61 | local_commit_count: int |
| 62 | external: list[ExternalImpact] |
| 63 | |
| 64 | @dataclass |
| 65 | class WorkspaceRiskEntry: |
| 66 | """One symbol entry in the workspace blast-risk ranking.""" |
| 67 | address: str |
| 68 | repo_id: str |
| 69 | repo_slug: str |
| 70 | co_change_count: int |
| 71 | commit_count: int |
| 72 | |
| 73 | @dataclass |
| 74 | class DepsNode: |
| 75 | """One node in the dependency graph (file-level module prefix).""" |
| 76 | id: str # e.g. "musehub.services.musehub_wire" |
| 77 | label: str # short label (last two segments) |
| 78 | type: str # "local" | "cross_repo" |
| 79 | repo_id: str |
| 80 | repo_slug: str |
| 81 | address_count: int |
| 82 | |
| 83 | @dataclass |
| 84 | class DepsEdge: |
| 85 | """One edge in the dependency graph.""" |
| 86 | source: str # node id |
| 87 | target: str # node id |
| 88 | weight: int # shared-commit count |
| 89 | type: str # "co_change" | "cross_repo" |
| 90 | |
| 91 | @dataclass |
| 92 | class DepsGraph: |
| 93 | nodes: list[DepsNode] = field(default_factory=list) |
| 94 | edges: list[DepsEdge] = field(default_factory=list) |
| 95 | |
| 96 | @dataclass |
| 97 | class WorkspaceForecast: |
| 98 | owner: str |
| 99 | repos: list[dict] |
| 100 | cross_repo_risk_symbols: list[dict] |
| 101 | |
| 102 | # ── Helpers ─────────────────────────────────────────────────────────────────── |
| 103 | |
| 104 | async def _load_owner_repos( |
| 105 | session: AsyncSession, |
| 106 | owner: str, |
| 107 | visible_to_user: str | None = None, |
| 108 | ) -> list[MusehubRepo]: |
| 109 | """Return non-deleted repos for `owner` filtered by visibility. |
| 110 | |
| 111 | When `visible_to_user` equals `owner`, all repos (public and private) are |
| 112 | returned. Any other value — including ``None`` (unauthenticated) — restricts |
| 113 | results to public repos only. |
| 114 | """ |
| 115 | stmt = ( |
| 116 | select(MusehubRepo) |
| 117 | .where( |
| 118 | MusehubRepo.owner == owner, |
| 119 | ) |
| 120 | ) |
| 121 | if visible_to_user != owner: |
| 122 | stmt = stmt.where(MusehubRepo.visibility == "public") |
| 123 | result = await session.execute(stmt) |
| 124 | return list(result.scalars().all()) |
| 125 | |
| 126 | def _module_prefix(address: str, depth: int = 3) -> str: |
| 127 | """Return the first `depth` dot-separated segments of an address. |
| 128 | |
| 129 | "musehub.services.musehub_wire.wire_push" → "musehub.services.musehub_wire" |
| 130 | """ |
| 131 | parts = address.split(".") |
| 132 | return ".".join(parts[:depth]) if len(parts) >= depth else address |
| 133 | |
| 134 | def _short_label(address: str) -> str: |
| 135 | parts = address.rsplit(".", 2) |
| 136 | return ".".join(parts[-2:]) if len(parts) >= 2 else address |
| 137 | |
| 138 | # ── Public API ──────────────────────────────────────────────────────────────── |
| 139 | |
| 140 | async def search_symbol_across_repos( |
| 141 | session: AsyncSession, |
| 142 | owner: str, |
| 143 | query: str, |
| 144 | *, |
| 145 | limit: int = 30, |
| 146 | visible_to_user: str | None = None, |
| 147 | ) -> list[CrossRepoMatch]: |
| 148 | """Search for symbol addresses matching `query` across all owner repos. |
| 149 | |
| 150 | Performs a case-insensitive substring match against the address strings |
| 151 | stored in each repo's MusehubSymbolIndex. Only repos visible to |
| 152 | `visible_to_user` are searched (see `_load_owner_repos`). |
| 153 | """ |
| 154 | repos = await _load_owner_repos(session, owner, visible_to_user=visible_to_user) |
| 155 | results: list[CrossRepoMatch] = [] |
| 156 | query_lower = query.lower() |
| 157 | |
| 158 | for repo in repos: |
| 159 | if len(results) >= limit: |
| 160 | break |
| 161 | history = await load_symbol_history(session, repo.repo_id) |
| 162 | if not history: |
| 163 | continue |
| 164 | for address, entries in history.items(): |
| 165 | if query_lower not in address.lower(): |
| 166 | continue |
| 167 | last_op = entries[-1].get("op", "modify") if entries else "modify" |
| 168 | results.append(CrossRepoMatch( |
| 169 | repo_id=repo.repo_id, |
| 170 | repo_slug=repo.name, |
| 171 | address=address, |
| 172 | last_op=last_op, |
| 173 | co_change_count=len(entries), |
| 174 | )) |
| 175 | if len(results) >= limit: |
| 176 | break |
| 177 | |
| 178 | results.sort(key=lambda r: r.co_change_count, reverse=True) |
| 179 | return results |
| 180 | |
| 181 | async def cross_repo_impact( |
| 182 | session: AsyncSession, |
| 183 | owner: str, |
| 184 | source_repo_id: str, |
| 185 | address: str, |
| 186 | *, |
| 187 | top_local: int = 20, |
| 188 | top_external: int = 10, |
| 189 | visible_to_user: str | None = None, |
| 190 | ) -> CrossRepoImpact | None: |
| 191 | """Compute blast radius that crosses repo boundaries. |
| 192 | |
| 193 | 1. Local impact: same co-change algorithm used by the single-repo |
| 194 | /impact endpoint. |
| 195 | 2. External impact: for each other repo in the workspace, look for |
| 196 | symbols that appear in commits with timestamps overlapping the |
| 197 | commit timestamps in source_repo's entries for `address`. |
| 198 | This is a heuristic — it does not prove import coupling but |
| 199 | surfaces symbols that changed in the same time window. |
| 200 | |
| 201 | Only repos visible to `visible_to_user` are considered (see `_load_owner_repos`). |
| 202 | """ |
| 203 | all_repos = await _load_owner_repos(session, owner, visible_to_user=visible_to_user) |
| 204 | source_repo = next((r for r in all_repos if r.repo_id == source_repo_id), None) |
| 205 | if source_repo is None: |
| 206 | return None |
| 207 | |
| 208 | source_history = await load_symbol_history(session, source_repo_id) |
| 209 | entries = source_history.get(address) |
| 210 | if entries is None: |
| 211 | return None |
| 212 | |
| 213 | # Local co-change |
| 214 | target_commits: set[str] = {e["commit_id"] for e in entries} |
| 215 | target_times: set[str] = {e.get("committed_at", "") for e in entries if e.get("committed_at")} |
| 216 | |
| 217 | local_co: IntDict = {} |
| 218 | for other_addr, other_entries in source_history.items(): |
| 219 | if other_addr == address: |
| 220 | continue |
| 221 | shared = sum(1 for e in other_entries if e["commit_id"] in target_commits) |
| 222 | if shared > 0: |
| 223 | local_co[other_addr] = shared |
| 224 | |
| 225 | local_ranked = sorted(local_co.items(), key=lambda x: x[1], reverse=True)[:top_local] |
| 226 | |
| 227 | # External: look in other repos for commits with overlapping timestamps |
| 228 | external: list[ExternalImpact] = [] |
| 229 | for repo in all_repos: |
| 230 | if repo.repo_id == source_repo_id: |
| 231 | continue |
| 232 | other_history = await load_symbol_history(session, repo.repo_id) |
| 233 | if not other_history: |
| 234 | continue |
| 235 | |
| 236 | # Time-window heuristic: find symbols in other repo that were |
| 237 | # modified during the same timestamps as the source symbol's commits. |
| 238 | ext_co: IntDict = {} |
| 239 | for other_addr, other_entries in other_history.items(): |
| 240 | shared_times = sum( |
| 241 | 1 for e in other_entries |
| 242 | if e.get("committed_at", "") in target_times |
| 243 | ) |
| 244 | if shared_times > 0: |
| 245 | ext_co[other_addr] = shared_times |
| 246 | |
| 247 | if not ext_co: |
| 248 | continue |
| 249 | |
| 250 | ranked_ext = sorted(ext_co.items(), key=lambda x: x[1], reverse=True)[:top_external] |
| 251 | external.append(ExternalImpact( |
| 252 | repo_id=repo.repo_id, |
| 253 | repo_slug=repo.name, |
| 254 | matches=[{"address": a, "shared_commits": c} for a, c in ranked_ext], |
| 255 | )) |
| 256 | |
| 257 | return CrossRepoImpact( |
| 258 | address=address, |
| 259 | source_repo_id=source_repo_id, |
| 260 | source_repo_slug=source_repo.name, |
| 261 | local_co_changed=[{"address": a, "shared_commits": c} for a, c in local_ranked], |
| 262 | local_commit_count=len(target_commits), |
| 263 | external=external, |
| 264 | ) |
| 265 | |
| 266 | async def workspace_blast_risk_top_n( |
| 267 | session: AsyncSession, |
| 268 | owner: str, |
| 269 | top_n: int = 20, |
| 270 | *, |
| 271 | visible_to_user: str | None = None, |
| 272 | ) -> list[WorkspaceRiskEntry]: |
| 273 | """Return the top-N symbols by total commit-touch-count across the workspace. |
| 274 | |
| 275 | A high touch count across many commits means the symbol changes often |
| 276 | and its blast radius is wide. This is an approximation of risk. |
| 277 | |
| 278 | Only repos visible to `visible_to_user` are considered (see `_load_owner_repos`). |
| 279 | """ |
| 280 | repos = await _load_owner_repos(session, owner, visible_to_user=visible_to_user) |
| 281 | all_entries: list[WorkspaceRiskEntry] = [] |
| 282 | |
| 283 | for repo in repos: |
| 284 | history = await load_symbol_history(session, repo.repo_id) |
| 285 | for address, entries in history.items(): |
| 286 | all_entries.append(WorkspaceRiskEntry( |
| 287 | address=address, |
| 288 | repo_id=repo.repo_id, |
| 289 | repo_slug=repo.name, |
| 290 | co_change_count=len(entries), |
| 291 | commit_count=len({e["commit_id"] for e in entries}), |
| 292 | )) |
| 293 | |
| 294 | all_entries.sort(key=lambda e: e.co_change_count, reverse=True) |
| 295 | return all_entries[:top_n] |
| 296 | |
| 297 | async def build_deps_graph( |
| 298 | session: AsyncSession, |
| 299 | owner: str, |
| 300 | source_repo_id: str, |
| 301 | *, |
| 302 | visible_to_user: str | None = None, |
| 303 | co_change_threshold: int = 2, |
| 304 | max_nodes: int = 60, |
| 305 | ) -> DepsGraph: |
| 306 | """Derive a file-level dependency graph from co-change patterns. |
| 307 | |
| 308 | Nodes are module prefixes (first 3 segments of address). |
| 309 | Edges are drawn between prefixes whose symbols co-change above threshold. |
| 310 | Cross-repo edges use symbols found in other repos' histories. |
| 311 | |
| 312 | Only repos visible to `visible_to_user` are considered (see `_load_owner_repos`). |
| 313 | """ |
| 314 | all_repos = await _load_owner_repos(session, owner, visible_to_user=visible_to_user) |
| 315 | source_repo = next((r for r in all_repos if r.repo_id == source_repo_id), None) |
| 316 | if source_repo is None: |
| 317 | return DepsGraph() |
| 318 | |
| 319 | source_history = await load_symbol_history(session, source_repo_id) |
| 320 | |
| 321 | # Build commit → set of module prefixes index |
| 322 | commit_to_prefixes: CommitToPrefixesDict = defaultdict(set) |
| 323 | for address, entries in source_history.items(): |
| 324 | prefix = _module_prefix(address) |
| 325 | for e in entries: |
| 326 | commit_to_prefixes[e["commit_id"]].add(prefix) |
| 327 | |
| 328 | # Count co-occurrence of prefix pairs within same commits |
| 329 | edge_weights: dict[tuple[str, str], int] = defaultdict(int) |
| 330 | for commit_id, prefixes in commit_to_prefixes.items(): |
| 331 | prefix_list = sorted(prefixes) |
| 332 | for i in range(len(prefix_list)): |
| 333 | for j in range(i + 1, len(prefix_list)): |
| 334 | key = (prefix_list[i], prefix_list[j]) |
| 335 | edge_weights[key] += 1 |
| 336 | |
| 337 | # Build local node set |
| 338 | local_prefixes: set[str] = set() |
| 339 | for address in source_history: |
| 340 | local_prefixes.add(_module_prefix(address)) |
| 341 | |
| 342 | # Collect node address counts |
| 343 | prefix_addr_counts: IntDict = defaultdict(int) |
| 344 | for address in source_history: |
| 345 | prefix_addr_counts[_module_prefix(address)] += 1 |
| 346 | |
| 347 | # Cross-repo: find which other-repo symbols appear in same commits |
| 348 | cross_edges: dict[tuple[str, str], int] = defaultdict(int) |
| 349 | cross_node_meta: CrossNodeMetaDict = {} # prefix → (repo_id, repo_slug) |
| 350 | |
| 351 | all_source_commits: set[str] = { |
| 352 | e["commit_id"] for entries in source_history.values() for e in entries |
| 353 | } |
| 354 | all_source_times: set[str] = { |
| 355 | e.get("committed_at", "") |
| 356 | for entries in source_history.values() |
| 357 | for e in entries |
| 358 | if e.get("committed_at") |
| 359 | } |
| 360 | |
| 361 | for repo in all_repos: |
| 362 | if repo.repo_id == source_repo_id: |
| 363 | continue |
| 364 | other_history = await load_symbol_history(session, repo.repo_id) |
| 365 | for other_address, other_entries in other_history.items(): |
| 366 | shared_times = sum( |
| 367 | 1 for e in other_entries |
| 368 | if e.get("committed_at", "") in all_source_times |
| 369 | ) |
| 370 | if shared_times < co_change_threshold: |
| 371 | continue |
| 372 | # Find which local prefix co-changes most with this external symbol |
| 373 | other_prefix = _module_prefix(other_address) |
| 374 | external_key = f"{repo.name}.{other_prefix}" |
| 375 | # find the most co-changed local prefix for this external symbol |
| 376 | best_local_prefix: str | None = None |
| 377 | best_count = 0 |
| 378 | for address, entries in source_history.items(): |
| 379 | src_prefix = _module_prefix(address) |
| 380 | overlap = sum(1 for e in entries if e.get("committed_at", "") in all_source_times) |
| 381 | if overlap > best_count: |
| 382 | best_count = overlap |
| 383 | best_local_prefix = src_prefix |
| 384 | if best_local_prefix: |
| 385 | cross_edges[(best_local_prefix, external_key)] += shared_times |
| 386 | cross_node_meta[external_key] = (repo.repo_id, repo.name) |
| 387 | |
| 388 | # Assemble nodes (cap at max_nodes) |
| 389 | nodes: list[DepsNode] = [] |
| 390 | seen_ids: set[str] = set() |
| 391 | |
| 392 | top_local = sorted(local_prefixes, key=lambda p: prefix_addr_counts.get(p, 0), reverse=True) |
| 393 | for prefix in top_local[:max_nodes]: |
| 394 | if prefix in seen_ids: |
| 395 | continue |
| 396 | nodes.append(DepsNode( |
| 397 | id=prefix, |
| 398 | label=_short_label(prefix), |
| 399 | type="local", |
| 400 | repo_id=source_repo_id, |
| 401 | repo_slug=source_repo.name, |
| 402 | address_count=prefix_addr_counts.get(prefix, 0), |
| 403 | )) |
| 404 | seen_ids.add(prefix) |
| 405 | |
| 406 | for ext_prefix, (ext_repo_id, ext_repo_slug) in cross_node_meta.items(): |
| 407 | if ext_prefix in seen_ids or len(nodes) >= max_nodes: |
| 408 | continue |
| 409 | nodes.append(DepsNode( |
| 410 | id=ext_prefix, |
| 411 | label=_short_label(ext_prefix), |
| 412 | type="cross_repo", |
| 413 | repo_id=ext_repo_id, |
| 414 | repo_slug=ext_repo_slug, |
| 415 | address_count=0, |
| 416 | )) |
| 417 | seen_ids.add(ext_prefix) |
| 418 | |
| 419 | # Assemble edges |
| 420 | edges: list[DepsEdge] = [] |
| 421 | for (src, tgt), weight in sorted(edge_weights.items(), key=lambda x: x[1], reverse=True): |
| 422 | if weight < co_change_threshold: |
| 423 | continue |
| 424 | if src not in seen_ids or tgt not in seen_ids: |
| 425 | continue |
| 426 | edges.append(DepsEdge(source=src, target=tgt, weight=weight, type="co_change")) |
| 427 | |
| 428 | for (src, tgt), weight in sorted(cross_edges.items(), key=lambda x: x[1], reverse=True): |
| 429 | if src not in seen_ids or tgt not in seen_ids: |
| 430 | continue |
| 431 | edges.append(DepsEdge(source=src, target=tgt, weight=weight, type="cross_repo")) |
| 432 | |
| 433 | return DepsGraph(nodes=nodes, edges=edges) |
File History
2 commits
sha256:e519738f2102d625d47828a17d113283cd744577b43551013cb668fd04f1c3fe
chore(mwp8/phase5): tick all RG checkboxes; close-out doc (…
Sonnet 4.6
18 days ago
sha256:0e5174da777684df43b2db71f282079030ef28bacffb93e19c29ee712b2a8bc4
test(mwp8/phase0): audit — repair+force-push leave stale ca…
Sonnet 4.6
20 days ago