""" repo_card_enrichment.py ======================= Single-pass enrichment service for the enriched repo card component. All five signals are derived from existing postgres tables — no engine subprocess calls, no per-card HTTP round trips. The public entry point ``enrich_repo_cards()`` issues five batched queries (one per signal) and merges the results into a ``RepoCardEnrichment`` per repo_id. Signal inventory ---------------- pulse_buckets list[PulseBucket] 30 daily commit-count buckets, normalised to SVG height autonomy_pct int 0-100, % commits with a non-empty agent_id hottest_symbol SymbolStat | None symbol with highest churn_30d in musehub_symbol_intel blast_leader SymbolStat | None symbol with highest blast score in musehub_symbol_intel dead_count int high-confidence dead symbols in musehub_intel_dead error_count int breakage errors from musehub_intel_breakage_meta warning_count int breakage warnings from musehub_intel_breakage_meta health_status 'clean' | 'warn' | 'risk' derived property — no extra query Spectral colour scheme ---------------------- Pulse bar colours are interpolated along the spectral gradient defined in the CSS design token ``--gradient-spectral``: blue #60a5fa → indigo #818cf8 → violet #a78bfa → pink #c084fc Bar i of N is assigned the hex colour at position i/(N-1) along this gradient. The interpolation is pure Python so the colour ends up as a static ``fill`` attribute in the inline SVG — no CSS gradient needed on the bars. Usage ----- enrichments = await enrich_repo_cards(db, repo_ids) card_data = enrichments.get(repo_id) # RepoCardEnrichment | None """ from __future__ import annotations import colorsys from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone import sqlalchemy as sa from sqlalchemy import case, func, select from sqlalchemy.ext.asyncio import AsyncSession from musehub.db.musehub_intel_models import ( MusehubIntelBreakageMeta, MusehubIntelDead, MusehubSymbolIntel, ) from musehub.db.musehub_repo_models import MusehubCommit, MusehubCommitRef from musehub.types.json_types import IntDict # --------------------------------------------------------------------------- # Spectral palette — four stops from the --gradient-spectral CSS token # --------------------------------------------------------------------------- _SPECTRAL_STOPS: list[tuple[int, int, int]] = [ (0x60, 0xA5, 0xFA), # #60a5fa blue (0x81, 0x8C, 0xF8), # #818cf8 indigo (0xA7, 0x8B, 0xFA), # #a78bfa violet (0xC0, 0x84, 0xFC), # #c084fc pink ] def _lerp(a: float, b: float, t: float) -> float: """Linear interpolation between a and b at position t ∈ [0.0, 1.0].""" return a + (b - a) * t def spectral_color(index: int, total: int) -> str: """ Return the hex colour for bar ``index`` of ``total`` bars along the spectral gradient. The gradient is divided into ``len(_SPECTRAL_STOPS) - 1`` equal segments. ``index`` is first mapped to a position t ∈ [0.0, 1.0] across the full gradient, then the surrounding two stops are interpolated. Parameters ---------- index: 0-based bar index total: total number of bars (must be ≥ 1) Returns ------- Hex colour string, e.g. ``"#60a5fa"``. """ if total <= 1: r, g, b = _SPECTRAL_STOPS[0] return f"#{r:02x}{g:02x}{b:02x}" t = index / (total - 1) # position 0.0 → 1.0 n_segments = len(_SPECTRAL_STOPS) - 1 segment = min(int(t * n_segments), n_segments - 1) # which segment local_t = (t * n_segments) - segment # position within segment lo = _SPECTRAL_STOPS[segment] hi = _SPECTRAL_STOPS[segment + 1] r = round(_lerp(lo[0], hi[0], local_t)) g = round(_lerp(lo[1], hi[1], local_t)) b = round(_lerp(lo[2], hi[2], local_t)) return f"#{r:02x}{g:02x}{b:02x}" # --------------------------------------------------------------------------- # Data models # --------------------------------------------------------------------------- _SPARKLINE_HEIGHT = 24 # max SVG bar height in px units _PULSE_DAYS = 30 # window for pulse sparkline @dataclass class PulseBucket: """ One day's commit count, pre-normalised to an SVG bar height. Attributes ---------- date: ISO date string (YYYY-MM-DD) count: raw commit count for this day h: bar height in SVG user units, scaled so the busiest day = _SPARKLINE_HEIGHT color: hex colour — always #60a5fa (domain-code blue), matching the profile heatmap opacity: float in [0, 1] — encodes commit intensity; 0.1 for empty days, scaling from 0.25 → 1.0 proportionally for days with commits """ date: str count: int h: int color: str opacity: float = 1.0 @dataclass class SymbolStat: """ Compact symbol reference for display in a repo card intel row. Attributes ---------- address: full Muse symbol address, e.g. "musehub/services/foo.py::bar" name: short display name — the last segment after ``::`` churn_30d: times this symbol was changed in the last 30 days blast: total downstream symbol count (blast radius) """ address: str churn_30d: int blast: int @property def name(self) -> str: """Return the symbol name — the part after the last ``::``.""" return self.address.rsplit("::", 1)[-1] if "::" in self.address else self.address @dataclass class RepoCardEnrichment: """ Complete enrichment payload for one repo card. Constructed by ``enrich_repo_cards()`` in a single batched pass. All fields have safe defaults so a card always renders even when intel tables contain no rows for a given repo. The ``health_status`` property is a derived signal — it requires no additional query. Attributes ---------- repo_id: sha256-prefixed repo identifier pulse_buckets: 30 PulseBucket entries covering the last 30 days, always length 30 (days with zero commits are zero-filled) autonomy_pct: percentage of commits carrying a non-empty agent_id; 0 when the repo has no commits at all hottest_symbol: the symbol with the highest churn_30d, or None if musehub_symbol_intel has no rows for this repo blast_leader: the symbol with the highest blast score, or None if musehub_symbol_intel has no rows for this repo dead_count: count of high-confidence dead symbols error_count: count of breakage errors from musehub_intel_breakage_meta warning_count: count of breakage warnings from musehub_intel_breakage_meta """ repo_id: str pulse_buckets: list[PulseBucket] = field(default_factory=list) autonomy_pct: int = 0 hottest_symbol: SymbolStat | None = None blast_leader: SymbolStat | None = None dead_count: int = 0 error_count: int = 0 warning_count: int = 0 @property def health_status(self) -> str: """ Three-tier health classification derived from dead + breakage signals. Returns ------- 'risk' — any breakage errors are present (highest severity) 'warn' — dead symbols or breakage warnings present, no errors 'clean' — zero dead symbols and zero breakage issues """ if self.error_count > 0: return "risk" if self.dead_count > 0 or self.warning_count > 0: return "warn" return "clean" RepoCardMap = dict[str, RepoCardEnrichment] """Maps repo_id → enrichment; returned by ``enrich_repo_cards``.""" # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- _PULSE_COLOR = "#60a5fa" # domain-code blue — matches the profile activity heatmap def _make_empty_buckets(today: date) -> list[PulseBucket]: """Return 30 zero-count PulseBuckets covering today and the 29 days before.""" buckets = [] for i in range(_PULSE_DAYS): day = today - timedelta(days=_PULSE_DAYS - 1 - i) buckets.append(PulseBucket( date=day.isoformat(), count=0, h=0, color=_PULSE_COLOR, opacity=0.1, )) return buckets def _normalise_heights(buckets: list[PulseBucket]) -> None: """ Scale bar heights and opacity in-place so the busiest bar = _SPARKLINE_HEIGHT px. Opacity mirrors the profile activity heatmap — it encodes commit intensity using the same #60a5fa base at varying alpha: count == 0 → opacity 0.10 (dim placeholder, matches bg-inset feel) count > 0 → opacity 0.25 – 1.0 (linear from quietest to busiest day) A minimum height of 2 px is applied to any bucket with count > 0 so that even a single commit is visible. """ max_count = max((b.count for b in buckets), default=0) if max_count == 0: return for bucket in buckets: if bucket.count == 0: bucket.h = 0 bucket.opacity = 0.1 else: scaled = round((bucket.count / max_count) * _SPARKLINE_HEIGHT) bucket.h = max(scaled, 2) # Scale opacity 0.25 → 1.0 proportionally to count bucket.opacity = round(0.25 + 0.75 * (bucket.count / max_count), 3) def _build_pulse( raw: dict[str, int], # date-string → commit count today: date, ) -> list[PulseBucket]: """ Merge raw day-count data into a complete 30-bucket sparkline. Parameters ---------- raw: mapping of ISO date string → commit count today: the reference date for the 30-day window """ buckets = _make_empty_buckets(today) for bucket in buckets: bucket.count = raw.get(bucket.date, 0) _normalise_heights(buckets) return buckets def _short_name(address: str) -> str: """Extract the symbol name from a full address (part after last ``::``)..""" return address.rsplit("::", 1)[-1] if "::" in address else address # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- async def enrich_repo_cards( db: AsyncSession, repo_ids: list[str], ) -> RepoCardMap: """ Batch-enrich a list of repo_ids for display in the enriched repo card component. Issues five SQL queries regardless of the number of repo_ids — one per signal. All queries use ``= ANY(:ids)`` for batching. Returns a dict mapping repo_id → RepoCardEnrichment. Repos with no intel data get a safe zero-value enrichment so the caller never needs to guard against missing keys. Parameters ---------- db: async SQLAlchemy session repo_ids: list of sha256-prefixed repo identifiers to enrich Returns ------- dict[str, RepoCardEnrichment] — one entry per requested repo_id """ if not repo_ids: return {} today = datetime.now(tz=timezone.utc).date() window_start = datetime.now(tz=timezone.utc) - timedelta(days=_PULSE_DAYS) # Initialise result map with zero-value enrichments results: dict[str, RepoCardEnrichment] = { rid: RepoCardEnrichment( repo_id=rid, pulse_buckets=_build_pulse({}, today), ) for rid in repo_ids } # ── Query 1: pulse — daily commit counts ───────────────────────────────── day_col = func.cast( func.date_trunc("day", MusehubCommit.timestamp.op("AT TIME ZONE")("UTC")), sa.Date, ).label("day") pulse_rows = (await db.execute( select( MusehubCommitRef.repo_id, day_col, func.count().label("cnt"), ) .join(MusehubCommit, MusehubCommit.commit_id == MusehubCommitRef.commit_id) .where( MusehubCommitRef.repo_id.in_(repo_ids), MusehubCommit.timestamp >= window_start, ) .group_by(MusehubCommitRef.repo_id, day_col) )).fetchall() pulse_raw: dict[str, IntDict] = {rid: {} for rid in repo_ids} for row in pulse_rows: pulse_raw[row.repo_id][row.day.isoformat()] = row.cnt for rid, raw in pulse_raw.items(): results[rid].pulse_buckets = _build_pulse(raw, today) # ── Query 2: autonomy — agent vs human commit ratio ─────────────────────── is_agent = case( ( (MusehubCommit.agent_id.isnot(None)) & (MusehubCommit.agent_id != ""), 1, ), else_=0, ) autonomy_rows = (await db.execute( select( MusehubCommitRef.repo_id, func.sum(is_agent).label("agent_commits"), func.count().label("total_commits"), ) .join(MusehubCommit, MusehubCommit.commit_id == MusehubCommitRef.commit_id) .where(MusehubCommitRef.repo_id.in_(repo_ids)) .group_by(MusehubCommitRef.repo_id) )).fetchall() for row in autonomy_rows: if row.total_commits > 0: results[row.repo_id].autonomy_pct = round( (row.agent_commits / row.total_commits) * 100 ) # ── Query 3: hottest symbol — highest churn_30d per repo ───────────────── # Use row_number() to pick the top symbol per repo without DISTINCT ON. hottest_rn = ( select( MusehubSymbolIntel.repo_id, MusehubSymbolIntel.address, MusehubSymbolIntel.churn_30d, MusehubSymbolIntel.blast, func.row_number().over( partition_by=MusehubSymbolIntel.repo_id, order_by=MusehubSymbolIntel.churn_30d.desc().nulls_last(), ).label("rn"), ) .where( MusehubSymbolIntel.repo_id.in_(repo_ids), MusehubSymbolIntel.churn_30d > 0, ) .subquery() ) hottest_rows = (await db.execute( select(hottest_rn).where(hottest_rn.c.rn == 1) )).fetchall() for row in hottest_rows: results[row.repo_id].hottest_symbol = SymbolStat( address=row.address, churn_30d=row.churn_30d, blast=row.blast, ) # ── Query 4: blast leader — highest blast score per repo ───────────────── blast_rn = ( select( MusehubSymbolIntel.repo_id, MusehubSymbolIntel.address, MusehubSymbolIntel.churn_30d, MusehubSymbolIntel.blast, func.row_number().over( partition_by=MusehubSymbolIntel.repo_id, order_by=MusehubSymbolIntel.blast.desc().nulls_last(), ).label("rn"), ) .where( MusehubSymbolIntel.repo_id.in_(repo_ids), MusehubSymbolIntel.blast > 0, ) .subquery() ) blast_rows = (await db.execute( select(blast_rn).where(blast_rn.c.rn == 1) )).fetchall() for row in blast_rows: results[row.repo_id].blast_leader = SymbolStat( address=row.address, churn_30d=row.churn_30d, blast=row.blast, ) # ── Query 5: health — dead count + breakage errors/warnings ────────────── dead_rows = (await db.execute( select( MusehubIntelDead.repo_id, func.count().label("dead_count"), ) .where( MusehubIntelDead.repo_id.in_(repo_ids), MusehubIntelDead.confidence == "high", ) .group_by(MusehubIntelDead.repo_id) )).fetchall() for row in dead_rows: results[row.repo_id].dead_count = row.dead_count breakage_rows = (await db.execute( select( MusehubIntelBreakageMeta.repo_id, MusehubIntelBreakageMeta.error_count, MusehubIntelBreakageMeta.warning_count, ) .where(MusehubIntelBreakageMeta.repo_id.in_(repo_ids)) )).fetchall() for row in breakage_rows: results[row.repo_id].error_count = row.error_count results[row.repo_id].warning_count = row.warning_count return results