feat: bidirectional source↔claim linking

Forward link: claims get `sourced_from: {domain}/{filename}` at extraction time.
Reverse link: after merge, backlink_source_claims() updates source files with
`claims_extracted:` list. All disk writes happen under async_main_worktree_lock.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

diagnostics/activity_feed_api.py

@@ -0,0 +1,212 @@
+"""Activity feed API — serves contribution events from pipeline.db."""
+import re
+import sqlite3
+import math
+import time
+from aiohttp import web
+
+DB_PATH = "/opt/teleo-eval/pipeline/pipeline.db"
+_cache = {"data": None, "ts": 0}
+CACHE_TTL = 60  # 1 minute — activity should feel fresh
+
+
+def _get_conn():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA busy_timeout = 10000")
+    return conn
+
+
+def _classify_event(branch, description, commit_type):
+    if commit_type != "knowledge":
+        return None
+    if branch and branch.startswith("extract/"):
+        return "create"
+    if branch and branch.startswith("reweave/"):
+        return "enrich"
+    if branch and branch.startswith("challenge/"):
+        return "challenge"
+    if description and "challenged_by" in description.lower():
+        return "challenge"
+    if branch and branch.startswith("enrich/"):
+        return "enrich"
+    return "create"
+
+
+def _normalize_contributor(submitted_by, agent):
+    if submitted_by and submitted_by.strip():
+        name = submitted_by.strip().lstrip("@")
+        return name
+    if agent and agent.strip() and agent != "pipeline":
+        return agent.strip()
+    return "pipeline"
+
+
+def _summary_from_branch(branch):
+    if not branch:
+        return ""
+    parts = branch.split("/", 1)
+    if len(parts) < 2:
+        return ""
+    slug = parts[1]
+    slug = re.sub(r"^[\d-]+-", "", slug)  # strip date prefix
+    slug = re.sub(r"-[a-f0-9]{4}$", "", slug)  # strip hash suffix
+    return slug.replace("-", " ").strip().capitalize()
+
+
+def _extract_claim_slugs(description, branch=None):
+    if not description:
+        if branch:
+            parts = branch.split("/", 1)
+            if len(parts) > 1:
+                return [parts[1][:120]]
+        return []
+    titles = [t.strip() for t in description.split("|") if t.strip()]
+    slugs = []
+    for title in titles:
+        slug = title.lower().strip()
+        slug = "".join(c if c.isalnum() or c in (" ", "-") else "" for c in slug)
+        slug = slug.replace(" ", "-").strip("-")
+        if len(slug) > 10:
+            slugs.append(slug[:120])
+    return slugs
+
+
+def _hot_score(challenge_count, enrich_count, signal_count, hours_since):
+    numerator = challenge_count * 3 + enrich_count * 2 + signal_count
+    denominator = max(hours_since, 0.5) ** 1.5
+    return numerator / denominator
+
+
+def _build_events():
+    conn = _get_conn()
+    try:
+        rows = conn.execute("""
+            SELECT p.number, p.branch, p.domain, p.agent, p.submitted_by,
+                   p.merged_at, p.description, p.commit_type, p.cost_usd
+            FROM prs p
+            WHERE p.status = 'merged'
+              AND p.commit_type = 'knowledge'
+              AND p.merged_at IS NOT NULL
+            ORDER BY p.merged_at DESC
+            LIMIT 2000
+        """).fetchall()
+
+        events = []
+        claim_activity = {}  # slug -> {challenges, enriches, signals, first_seen}
+
+        for row in rows:
+            event_type = _classify_event(row["branch"], row["description"], row["commit_type"])
+            if not event_type:
+                continue
+
+            contributor = _normalize_contributor(row["submitted_by"], row["agent"])
+            slugs = _extract_claim_slugs(row["description"], row["branch"])
+            merged_at = row["merged_at"] or ""
+
+            ci_map = {"create": 0.35, "enrich": 0.25, "challenge": 0.40}
+            ci_earned = ci_map.get(event_type, 0)
+
+            for slug in slugs:
+                if slug not in claim_activity:
+                    claim_activity[slug] = {
+                        "challenges": 0, "enriches": 0, "signals": 0,
+                        "first_seen": merged_at,
+                    }
+                if event_type == "challenge":
+                    claim_activity[slug]["challenges"] += 1
+                elif event_type == "enrich":
+                    claim_activity[slug]["enriches"] += 1
+                else:
+                    claim_activity[slug]["signals"] += 1
+
+            summary_text = ""
+            if row["description"]:
+                first_title = row["description"].split("|")[0].strip()
+                if len(first_title) > 120:
+                    first_title = first_title[:117] + "..."
+                summary_text = first_title
+            elif row["branch"]:
+                summary_text = _summary_from_branch(row["branch"])
+
+            for slug in (slugs[:1] if slugs else [""]):
+                events.append({
+                    "type": event_type,
+                    "claim_slug": slug,
+                    "domain": row["domain"] or "unknown",
+                    "contributor": contributor,
+                    "timestamp": merged_at,
+                    "ci_earned": round(ci_earned, 2),
+                    "summary": summary_text,
+                    "pr_number": row["number"],
+                })
+
+        return events, claim_activity
+    finally:
+        conn.close()
+
+
+def _sort_events(events, claim_activity, sort_mode, now_ts):
+    if sort_mode == "recent":
+        events.sort(key=lambda e: e["timestamp"], reverse=True)
+    elif sort_mode == "hot":
+        def hot_key(e):
+            slug = e["claim_slug"]
+            ca = claim_activity.get(slug, {"challenges": 0, "enriches": 0, "signals": 0})
+            try:
+                from datetime import datetime
+                evt_time = datetime.fromisoformat(e["timestamp"].replace("Z", "+00:00"))
+                hours = (now_ts - evt_time.timestamp()) / 3600
+            except (ValueError, AttributeError):
+                hours = 9999
+            return _hot_score(ca["challenges"], ca["enriches"], ca["signals"], hours)
+        events.sort(key=hot_key, reverse=True)
+    elif sort_mode == "important":
+        type_rank = {"challenge": 0, "enrich": 1, "create": 2}
+        events.sort(key=lambda e: (type_rank.get(e["type"], 3), -len(e["summary"])))
+    return events
+
+
+async def handle_activity_feed(request):
+    sort_mode = request.query.get("sort", "recent")
+    if sort_mode not in ("hot", "recent", "important"):
+        sort_mode = "recent"
+    domain = request.query.get("domain", "")
+    contributor = request.query.get("contributor", "")
+    try:
+        limit = min(int(request.query.get("limit", "20")), 100)
+    except ValueError:
+        limit = 20
+    try:
+        offset = max(int(request.query.get("offset", "0")), 0)
+    except ValueError:
+        offset = 0
+
+    now = time.time()
+    if _cache["data"] is None or (now - _cache["ts"]) > CACHE_TTL:
+        _cache["data"] = _build_events()
+        _cache["ts"] = now
+
+    events, claim_activity = _cache["data"]
+
+    filtered = events
+    if domain:
+        filtered = [e for e in filtered if e["domain"] == domain]
+    if contributor:
+        filtered = [e for e in filtered if e["contributor"] == contributor]
+
+    sorted_events = _sort_events(list(filtered), claim_activity, sort_mode, now)
+    total = len(sorted_events)
+    page = sorted_events[offset:offset + limit]
+
+    return web.json_response({
+        "events": page,
+        "total": total,
+        "sort": sort_mode,
+        "offset": offset,
+        "limit": limit,
+    }, headers={"Access-Control-Allow-Origin": "*"})
+
+
+def register(app):
+    app.router.add_get("/api/activity-feed", handle_activity_feed)

diagnostics/contributor_profile_api.py

@@ -0,0 +1,365 @@
+"""Contributor profile API — GET /api/contributors/{handle}"""
+
+import sqlite3
+import json
+import os
+import re
+import subprocess
+from datetime import datetime
+
+DB_PATH = os.environ.get("PIPELINE_DB", "/opt/teleo-eval/pipeline/pipeline.db")
+SYSTEM_ACCOUNTS = {"pipeline", "unknown", "teleo-agents", "teleo pipeline"}
+CODEX_PATH = "/opt/teleo-eval/workspaces/main"
+
+CI_WEIGHTS = {
+    "sourcer": 0.15,
+    "extractor": 0.05,
+    "challenger": 0.35,
+    "synthesizer": 0.25,
+    "reviewer": 0.20,
+}
+
+FOUNDING_CUTOFF = "2026-03-15"
+
+BADGE_DEFS = {
+    "FOUNDING CONTRIBUTOR": {"rarity": "limited", "desc": "Contributed during pre-launch phase"},
+    "BELIEF MOVER": {"rarity": "rare", "desc": "Challenge that led to a claim revision"},
+    "KNOWLEDGE SOURCER": {"rarity": "uncommon", "desc": "Source that generated 3+ claims"},
+    "DOMAIN SPECIALIST": {"rarity": "rare", "desc": "Top 3 CI contributor in a domain"},
+    "VETERAN": {"rarity": "uncommon", "desc": "10+ accepted contributions"},
+    "FIRST BLOOD": {"rarity": "common", "desc": "First contribution of any kind"},
+    "CONTRIBUTOR": {"rarity": "common", "desc": "Account created + first accepted contribution"},
+}
+
+
+def _get_conn():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def _compute_ci(row):
+    total = 0
+    for role, weight in CI_WEIGHTS.items():
+        total += (row.get(f"{role}_count", 0) or 0) * weight
+    return round(total, 2)
+
+
+def _compute_badges(handle, row, domain_breakdown, conn):
+    badges = []
+    first = row.get("first_contribution", "")
+
+    if first and first <= FOUNDING_CUTOFF:
+        badges.append("FOUNDING CONTRIBUTOR")
+
+    claims = row.get("claims_merged", 0) or 0
+    if claims > 0:
+        badges.append("CONTRIBUTOR")
+        badges.append("FIRST BLOOD")
+
+    if claims >= 10:
+        badges.append("VETERAN")
+
+    challenger = row.get("challenger_count", 0) or 0
+    challenge_ci = row.get("_challenge_count_from_scores", 0)
+    if challenger > 0 or challenge_ci > 0:
+        badges.append("BELIEF MOVER")
+
+    sourcer = row.get("sourcer_count", 0) or 0
+    if sourcer >= 3:
+        badges.append("KNOWLEDGE SOURCER")
+
+    return badges
+
+
+def _get_domain_breakdown(handle, conn):
+    rows = conn.execute("""
+        SELECT domain, COUNT(*) as cnt
+        FROM prs
+        WHERE status='merged' AND (LOWER(agent)=LOWER(?) OR LOWER(submitted_by)=LOWER(?))
+        AND domain IS NOT NULL
+        GROUP BY domain ORDER BY cnt DESC
+    """, (handle, handle)).fetchall()
+    return {r["domain"]: r["cnt"] for r in rows}
+
+
+def _get_contribution_timeline(handle, conn, limit=20):
+    rows = conn.execute("""
+        SELECT number, domain, status, created_at, description, commit_type, source_path
+        FROM prs
+        WHERE status='merged' AND (LOWER(agent)=LOWER(?) OR LOWER(submitted_by)=LOWER(?))
+        ORDER BY created_at DESC LIMIT ?
+    """, (handle, handle, limit)).fetchall()
+
+    timeline = []
+    for r in rows:
+        desc = r["description"] or ""
+        if not desc and r["source_path"]:
+            desc = os.path.basename(r["source_path"]).replace("-", " ").replace(".md", "")
+        timeline.append({
+            "pr_number": r["number"],
+            "domain": r["domain"],
+            "date": r["created_at"][:10] if r["created_at"] else None,
+            "type": _classify_commit(r["commit_type"]),
+            "summary": desc[:200] if desc else None,
+        })
+    return timeline
+
+
+def _classify_commit(commit_type):
+    if not commit_type:
+        return "create"
+    ct = commit_type.lower()
+    if "challenge" in ct:
+        return "challenge"
+    if "enrich" in ct or "update" in ct or "reweave" in ct:
+        return "enrich"
+    return "create"
+
+
+def _get_review_stats(handle, conn):
+    rows = conn.execute("""
+        SELECT outcome, COUNT(*) as cnt
+        FROM review_records
+        WHERE LOWER(agent) = LOWER(?)
+        GROUP BY outcome
+    """, (handle,)).fetchall()
+    stats = {}
+    for r in rows:
+        stats[r["outcome"]] = r["cnt"]
+    return stats
+
+
+def _get_action_ci(handle, conn):
+    """Get action-type CI from contribution_scores table.
+
+    Checks both exact handle and common variants (with/without suffix).
+    """
+    h = handle.lower()
+    base = re.sub(r"[-_]\w+\d+$", "", h)
+    variants = list({h, base}) if base and base != h else [h]
+    try:
+        placeholders = ",".join("?" for _ in variants)
+        rows = conn.execute(f"""
+            SELECT event_type, SUM(ci_earned) as total, COUNT(*) as cnt
+            FROM contribution_scores
+            WHERE LOWER(contributor) IN ({placeholders})
+            GROUP BY event_type
+        """, variants).fetchall()
+    except Exception:
+        return None
+
+    if not rows:
+        return None
+
+    breakdown = {}
+    total = 0.0
+    for r in rows:
+        breakdown[r["event_type"]] = {
+            "count": r["cnt"],
+            "ci": round(r["total"], 4),
+        }
+        total += r["total"]
+
+    return {
+        "total": round(total, 4),
+        "breakdown": breakdown,
+    }
+
+
+def _get_git_contributor(handle):
+    """Fallback: check git log for contributors not in pipeline.db."""
+    try:
+        result = subprocess.run(
+            ["git", "log", "--all", "--format=%H|%an|%ae|%aI", "--diff-filter=A", "--", "domains/"],
+            capture_output=True, text=True, cwd=CODEX_PATH, timeout=30
+        )
+        if result.returncode != 0:
+            return None
+
+        claims = []
+        for line in result.stdout.strip().split("\n"):
+            if not line:
+                continue
+            parts = line.split("|", 3)
+            if len(parts) < 4:
+                continue
+            sha, name, email, date = parts
+            if handle.lower() in name.lower() or handle.lower() in email.lower():
+                claims.append({"sha": sha, "author": name, "email": email, "date": date[:10]})
+
+        if not claims:
+            return None
+
+        return {
+            "handle": handle,
+            "display_name": claims[0]["author"],
+            "email": claims[0]["email"],
+            "first_contribution": min(c["date"] for c in claims),
+            "last_contribution": max(c["date"] for c in claims),
+            "claims_merged": len(claims),
+            "sourcer_count": 0,
+            "extractor_count": 0,
+            "challenger_count": 0,
+            "synthesizer_count": 0,
+            "reviewer_count": 0,
+        }
+    except Exception:
+        return None
+
+
+def get_contributor_profile(handle):
+    conn = _get_conn()
+    try:
+        row = conn.execute(
+            "SELECT * FROM contributors WHERE LOWER(handle) = LOWER(?)", (handle,)
+        ).fetchone()
+
+        if row:
+            data = dict(row)
+        else:
+            git_data = _get_git_contributor(handle)
+            if git_data:
+                data = git_data
+            else:
+                return None
+
+        ci_score = _compute_ci(data)
+        action_ci = _get_action_ci(handle, conn)
+        domain_breakdown = _get_domain_breakdown(handle, conn)
+        timeline = _get_contribution_timeline(handle, conn)
+        review_stats = _get_review_stats(handle, conn)
+        if action_ci and "challenge" in action_ci.get("breakdown", {}):
+            data["_challenge_count_from_scores"] = action_ci["breakdown"]["challenge"]["count"]
+        badges = _compute_badges(handle, data, domain_breakdown, conn)
+
+        # For git-only contributors, build domain breakdown from git
+        if not domain_breakdown and not row:
+            domain_breakdown = _git_domain_breakdown(handle)
+
+        hero_badge = None
+        rarity_order = ["limited", "rare", "uncommon", "common"]
+        for rarity in rarity_order:
+            for b in badges:
+                if BADGE_DEFS.get(b, {}).get("rarity") == rarity:
+                    hero_badge = b
+                    break
+            if hero_badge:
+                break
+
+        role_breakdown = {
+            "sourcer": data.get("sourcer_count", 0) or 0,
+            "extractor": data.get("extractor_count", 0) or 0,
+            "challenger": data.get("challenger_count", 0) or 0,
+            "synthesizer": data.get("synthesizer_count", 0) or 0,
+            "reviewer": data.get("reviewer_count", 0) or 0,
+        }
+        total_roles = sum(role_breakdown.values())
+        role_pct = {}
+        for k, v in role_breakdown.items():
+            role_pct[k] = round(v / total_roles * 100) if total_roles > 0 else 0
+
+        return {
+            "handle": data.get("handle", handle),
+            "display_name": data.get("display_name"),
+            "ci_score": ci_score,
+            "action_ci": action_ci,
+            "primary_ci": action_ci["total"] if action_ci else ci_score,
+            "hero_badge": hero_badge,
+            "badges": [{"name": b, **BADGE_DEFS.get(b, {})} for b in badges],
+            "joined": data.get("first_contribution"),
+            "last_active": data.get("last_contribution"),
+            "claims_merged": data.get("claims_merged", 0) or 0,
+            "principal": data.get("principal"),
+            "role_breakdown": role_breakdown,
+            "role_percentages": role_pct,
+            "domain_breakdown": domain_breakdown,
+            "review_stats": review_stats,
+            "contribution_timeline": timeline,
+            "active_domains": list(domain_breakdown.keys()),
+        }
+    finally:
+        conn.close()
+
+
+def _git_domain_breakdown(handle):
+    """For git-only contributors, count claims by domain from file paths."""
+    try:
+        result = subprocess.run(
+            ["git", "log", "--all", "--name-only", "--format=COMMIT|%an", "--diff-filter=A", "--", "domains/"],
+            capture_output=True, text=True, cwd=CODEX_PATH, timeout=30
+        )
+        if result.returncode != 0:
+            return {}
+
+        domains = {}
+        current_match = False
+        for line in result.stdout.strip().split("\n"):
+            if line.startswith("COMMIT|"):
+                author = line.split("|", 1)[1]
+                current_match = handle.lower() in author.lower()
+            elif current_match and line.startswith("domains/"):
+                parts = line.split("/")
+                if len(parts) >= 2:
+                    domain = parts[1]
+                    domains[domain] = domains.get(domain, 0) + 1
+
+        return domains
+    except Exception:
+        return {}
+
+
+async def handle_contributor_profile(request):
+    from aiohttp import web
+    handle = request.match_info["handle"]
+    profile = get_contributor_profile(handle)
+    if profile is None:
+        return web.json_response({"error": f"Contributor '{handle}' not found"}, status=404)
+    return web.json_response(profile)
+
+
+async def handle_contributors_list(request):
+    from aiohttp import web
+    conn = _get_conn()
+    try:
+        min_claims = int(request.query.get("min_claims", "1"))
+        rows = conn.execute("""
+            SELECT handle, display_name, first_contribution, last_contribution, 
+                   sourcer_count, extractor_count, challenger_count, synthesizer_count,
+                   reviewer_count, claims_merged, principal
+            FROM contributors
+            WHERE claims_merged >= ?
+            ORDER BY claims_merged DESC
+        """, (min_claims,)).fetchall()
+
+        contributors = []
+        for r in rows:
+            data = dict(r)
+            if data["handle"].lower() in SYSTEM_ACCOUNTS:
+                continue
+            ci = _compute_ci(data)
+            action_ci = _get_action_ci(data["handle"], conn)
+            action_total = action_ci["total"] if action_ci else 0.0
+            contributors.append({
+                "handle": data["handle"],
+                "display_name": data["display_name"],
+                "ci_score": ci,
+                "action_ci": action_total,
+                "primary_ci": action_total if action_total > 0 else ci,
+                "claims_merged": data["claims_merged"],
+                "first_contribution": data["first_contribution"],
+                "last_contribution": data["last_contribution"],
+                "principal": data["principal"],
+            })
+
+        return web.json_response({
+            "contributors": contributors,
+            "total": len(contributors),
+        })
+    finally:
+        conn.close()
+
+
+def register_contributor_routes(app):
+    app.router.add_get("/api/contributors/list", handle_contributors_list)
+    app.router.add_get("/api/contributors/{handle}", handle_contributor_profile)

diagnostics/dashboard_routes.py

@@ -10,6 +10,7 @@ Endpoints:
 Owner: Argus
 """
 
+import asyncio
 import json
 import logging
 import os
@@ -17,6 +18,7 @@ import sqlite3
 import statistics
 import time
 import urllib.request
+from collections import defaultdict
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -1182,6 +1184,113 @@ async def handle_telegram_extractions(request):
         conn.close()
 
 
+# ─── GET /api/contributor-growth ─────────────────────────────────────────
+
+CODEX_WORKTREE = Path(os.environ.get("MAIN_WORKTREE", "/opt/teleo-eval/workspaces/main"))
+FOUNDING_CUTOFF = "2026-03-15"
+CONTRIBUTOR_EXCLUDE = {"Teleo Agents", "Teleo Pipeline"}
+
+_growth_cache: dict | None = None
+_growth_cache_ts: float = 0
+GROWTH_CACHE_TTL = 300
+
+
+async def handle_contributor_growth(request):
+    """Cumulative unique contributors and claims over time from git log.
+
+    Returns time-series data for Chart.js line charts.
+    Cached for 5 minutes since git log is expensive.
+    """
+    global _growth_cache, _growth_cache_ts
+    now = time.monotonic()
+    if _growth_cache is not None and (now - _growth_cache_ts) < GROWTH_CACHE_TTL:
+        return web.json_response(_growth_cache)
+
+    codex_path = str(CODEX_WORKTREE)
+    if not CODEX_WORKTREE.exists():
+        return web.json_response(
+            {"error": "codex worktree not found", "path": codex_path}, status=404
+        )
+
+    proc = await asyncio.create_subprocess_exec(
+        "git", "log", "--format=%ad|%an", "--date=format:%Y-%m-%d", "--all",
+        cwd=codex_path,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        return web.json_response(
+            {"error": "git log failed", "detail": stderr.decode()[:500]}, status=500
+        )
+
+    first_seen: dict[str, str] = {}
+    daily_commits: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
+    for line in stdout.decode().strip().split("\n"):
+        if "|" not in line:
+            continue
+        date, author = line.split("|", 1)
+        if author in CONTRIBUTOR_EXCLUDE:
+            continue
+        daily_commits[date][author] += 1
+        if author not in first_seen or date < first_seen[author]:
+            first_seen[author] = date
+
+    by_date: dict[str, list[str]] = defaultdict(list)
+    for author, date in first_seen.items():
+        by_date[date].append(author)
+
+    contributors_timeline = []
+    seen: set[str] = set()
+    for date in sorted(by_date.keys()):
+        new_authors = by_date[date]
+        seen.update(new_authors)
+        contributors_timeline.append({
+            "date": date,
+            "cumulative": len(seen),
+            "new": [{"name": a, "founding": date <= FOUNDING_CUTOFF} for a in sorted(new_authors)],
+        })
+
+    proc2 = await asyncio.create_subprocess_exec(
+        "git", "log", "--format=%ad", "--date=format:%Y-%m-%d",
+        "--all", "--diff-filter=A", "--", "domains/*.md",
+        cwd=codex_path,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout2, _ = await proc2.communicate()
+    claim_counts: dict[str, int] = defaultdict(int)
+    for line in stdout2.decode().strip().split("\n"):
+        line = line.strip()
+        if line:
+            claim_counts[line] += 1
+
+    claims_timeline = []
+    cumulative = 0
+    for date in sorted(claim_counts.keys()):
+        cumulative += claim_counts[date]
+        claims_timeline.append({"date": date, "cumulative": cumulative, "added": claim_counts[date]})
+
+    all_contributors = set(first_seen.keys())
+    founding = sorted(a for a in all_contributors if first_seen[a] <= FOUNDING_CUTOFF)
+
+    result = {
+        "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "summary": {
+            "total_contributors": len(all_contributors),
+            "founding_contributors": founding,
+            "total_claims": cumulative,
+            "days_active": (datetime.now(timezone.utc) - datetime(2026, 3, 5, tzinfo=timezone.utc)).days,
+        },
+        "cumulative_contributors": contributors_timeline,
+        "cumulative_claims": claims_timeline,
+    }
+
+    _growth_cache = result
+    _growth_cache_ts = now
+    return web.json_response(result)
+
+
 # ─── Registration ──────────────────────────────────────────────────────────
 
 def register_dashboard_routes(app: web.Application, get_conn):
@@ -1199,3 +1308,42 @@ def register_dashboard_routes(app: web.Application, get_conn):
     app.router.add_get("/api/growth", handle_growth)
     app.router.add_get("/api/pr-lifecycle", handle_pr_lifecycle)
     app.router.add_get("/api/telegram-extractions", handle_telegram_extractions)
+    app.router.add_get("/api/contributor-growth", handle_contributor_growth)
+    app.router.add_get("/api/digest/latest", handle_digest_latest)
+    app.router.add_get("/api/contributor-graph", handle_contributor_graph)
+
+
+async def handle_digest_latest(request):
+    """GET /api/digest/latest — return the most recent scoring digest."""
+    import json as _json
+    digest_path = "/opt/teleo-eval/logs/scoring-digest-latest.json"
+    try:
+        with open(digest_path) as f:
+            data = _json.load(f)
+        return web.json_response(data)
+    except FileNotFoundError:
+        return web.json_response({"error": "No digest available yet"}, status=404)
+    except Exception as e:
+        return web.json_response({"error": str(e)}, status=500)
+
+
+async def handle_contributor_graph(request):
+    """GET /api/contributor-graph — serve the PNG chart."""
+    import subprocess, os
+    png_path = "/opt/teleo-eval/static/contributor-graph.png"
+    # Regenerate if older than 1 hour or missing
+    regen = not os.path.exists(png_path)
+    if not regen:
+        age = __import__('time').time() - os.path.getmtime(png_path)
+        regen = age > 3600
+    if regen:
+        try:
+            subprocess.run(
+                ['python3', '/opt/teleo-eval/scripts/contributor-graph.py'],
+                timeout=30, capture_output=True
+            )
+        except Exception:
+            pass
+    if not os.path.exists(png_path):
+        return web.Response(text='Chart not available', status=503)
+    return web.FileResponse(png_path, headers={'Content-Type': 'image/png'})

lib/config.py

@@ -156,13 +156,13 @@ CONTRIBUTOR_TIER_RULES = {
     },
 }
 
-# Role weights for CI computation (must match schemas/contribution-weights.yaml)
+# Role weights for CI computation (must match core/contribution-architecture.md)
 CONTRIBUTION_ROLE_WEIGHTS = {
+    "challenger": 0.35,
+    "synthesizer": 0.25,
+    "reviewer": 0.20,
     "sourcer": 0.15,
-    "extractor": 0.40,
-    "challenger": 0.20,
-    "synthesizer": 0.15,
-    "reviewer": 0.10,
+    "extractor": 0.05,
 }
 
 # --- Circuit breakers ---

lib/contributor.py

@@ -38,6 +38,22 @@ def is_knowledge_pr(diff: str) -> bool:
     return False
 
 
+COMMIT_TYPE_TO_ROLE = {
+    "challenge": "challenger",
+    "enrich": "synthesizer",
+    "extract": "extractor",
+    "research": "synthesizer",
+    "entity": "extractor",
+    "reweave": "synthesizer",
+    "fix": "extractor",
+}
+
+
+def commit_type_to_role(commit_type: str) -> str:
+    """Map a refined commit_type to a contributor role."""
+    return COMMIT_TYPE_TO_ROLE.get(commit_type, "extractor")
+
+
 def refine_commit_type(diff: str, branch_commit_type: str) -> str:
     """Refine commit_type from diff content when branch prefix is ambiguous.
 
@@ -126,8 +142,9 @@ async def record_contributor_attribution(conn, pr_number: int, branch: str, git_
         for match in re.finditer(r"Pentagon-Agent:\s*(\S+)\s*<([^>]+)>", log_output):
             agent_name = match.group(1).lower()
             agent_uuid = match.group(2)
+            role = commit_type_to_role(refined_type)
             upsert_contributor(
-                conn, agent_name, agent_uuid, "extractor", today,
+                conn, agent_name, agent_uuid, role, today,
             )
             agents_found.add(agent_name)
 
@@ -167,13 +184,15 @@ async def record_contributor_attribution(conn, pr_number: int, branch: str, git_
             for author_line in author_output.strip().split("\n"):
                 author_name = author_line.strip().lower()
                 if author_name and author_name not in _BOT_AUTHORS:
-                    upsert_contributor(conn, author_name, None, "extractor", today)
+                    role = commit_type_to_role(refined_type)
+                    upsert_contributor(conn, author_name, None, role, today)
                     agents_found.add(author_name)
 
         if not agents_found:
             row = conn.execute("SELECT agent FROM prs WHERE number = ?", (pr_number,)).fetchone()
             if row and row["agent"] and row["agent"] != "external":
-                upsert_contributor(conn, row["agent"].lower(), None, "extractor", today)
+                role = commit_type_to_role(refined_type)
+                upsert_contributor(conn, row["agent"].lower(), None, role, today)
 
 
 def upsert_contributor(

lib/extract.py

@@ -229,7 +229,7 @@ def _parse_extraction_json(text: str) -> dict | None:
         return None
 
 
-def _build_claim_content(claim: dict, agent: str, source_format: str | None = None) -> str:
+def _build_claim_content(claim: dict, agent: str, source_format: str | None = None, source_file: str = "") -> str:
     """Build claim markdown file content from extraction JSON."""
     today = date.today().isoformat()
     domain = claim.get("domain", "")
@@ -281,6 +281,8 @@ def _build_claim_content(claim: dict, agent: str, source_format: str | None = No
         f"created: {today}",
         f"agent: {agent}",
     ]
+    if source_file:
+        lines.append(f"sourced_from: {source_file}")
     if scope:
         lines.append(f"scope: {scope}")
     if sourcer:
@@ -432,7 +434,7 @@ async def _extract_one_source(
         filename = Path(filename).name  # Strip directory components — LLM output may contain path traversal
         if not filename.endswith(".md"):
             filename += ".md"
-        content = _build_claim_content(c, agent_lower, source_format=source_format)
+        content = _build_claim_content(c, agent_lower, source_format=source_format, source_file=f"{domain}/{source_file}" if domain else source_file)
         claim_files.append({"filename": filename, "domain": c.get("domain", domain), "content": content})
 
     # Build entity file contents

lib/merge.py

@@ -308,7 +308,14 @@ async def _cherry_pick_onto_main(branch: str) -> tuple[bool, str]:
     rc, merge_base = await _git("merge-base", "origin/main", f"origin/{branch}")
     rc2, main_sha = await _git("rev-parse", "origin/main")
     if rc == 0 and rc2 == 0 and merge_base.strip() == main_sha.strip():
-        return True, "already up to date"
+        # Branch is descendant of main — but fork workflows (merge main into branch)
+        # create this state while still having new content. Check for actual diff.
+        rc_diff, diff_out = await _git(
+            "diff", "--stat", f"origin/main..origin/{branch}", timeout=10,
+        )
+        if rc_diff != 0 or not diff_out.strip():
+            return True, "already up to date"
+        logger.info("Branch %s is descendant of main but has new content — proceeding", branch)
 
     # Get extraction commits (oldest first), skip merge commits from fork workflows
     rc, commits_out = await _git(
@@ -429,6 +436,7 @@ from .frontmatter import (
     serialize_frontmatter,
 )
 from .post_merge import (
+    backlink_source_claims,
     embed_merged_claims,
     reciprocal_edges,
     archive_source_for_pr,
@@ -848,6 +856,12 @@ async def _merge_domain_queue(conn, domain: str) -> tuple[int, int]:
             # Archive source file (closes near-duplicate loop — Ganymede review)
             archive_source_for_pr(branch, domain)
 
+            # Backlink: update source files with claims_extracted refs
+            try:
+                await backlink_source_claims(main_sha, branch_sha, _git)
+            except Exception:
+                logger.exception("PR #%d: backlink_source_claims failed (non-fatal)", pr_num)
+
             # Embed new/changed claims into Qdrant (non-fatal)
             await embed_merged_claims(main_sha, branch_sha, _git)
 

lib/post_merge.py

@@ -13,6 +13,7 @@ import logging
 import os
 import re
 import shutil
+from pathlib import Path
 from typing import Callable
 
 from . import config
@@ -295,6 +296,139 @@ async def reciprocal_edges(main_sha: str, branch_sha: str, git_fn: Callable):
         logger.exception("reciprocal_edges: failed (non-fatal)")
 
 
+async def backlink_source_claims(main_sha: str, branch_sha: str, git_fn: Callable):
+    """After merge, update source files with claims_extracted backlinks.
+
+    Reads sourced_from from merged claim frontmatter, finds the source file,
+    and appends the claim filename to its claims_extracted list.
+    Only runs for newly added claims (diff-filter=A).
+    """
+    try:
+        rc, diff_out = await git_fn(
+            "diff", "--name-only", "--diff-filter=A",
+            main_sha, branch_sha,
+            cwd=str(config.MAIN_WORKTREE),
+            timeout=10,
+        )
+        if rc != 0:
+            logger.warning("backlink_source_claims: diff failed (rc=%d), skipping", rc)
+            return
+
+        claim_dirs = {"domains/", "core/", "foundations/"}
+        new_claims = [
+            f for f in diff_out.strip().split("\n")
+            if f.endswith(".md")
+            and any(f.startswith(d) for d in claim_dirs)
+            and not f.split("/")[-1].startswith("_")
+            and "/entities/" not in f
+            and "/decisions/" not in f
+        ]
+
+        if not new_claims:
+            return
+
+        modified_sources = {}
+        for claim_path in new_claims:
+            full_path = config.MAIN_WORKTREE / claim_path
+            if not full_path.exists():
+                continue
+
+            try:
+                content = full_path.read_text()
+            except Exception:
+                continue
+
+            fm, raw_fm, body = parse_yaml_frontmatter(content)
+            if fm is None:
+                continue
+
+            sourced_from = fm.get("sourced_from", "")
+            if not sourced_from:
+                continue
+
+            source_path = config.MAIN_WORKTREE / "inbox" / "archive" / sourced_from
+            if not source_path.exists():
+                logger.debug("backlink_source_claims: source %s not found at %s", sourced_from, source_path)
+                continue
+
+            claim_filename = claim_path.rsplit("/", 1)[-1].replace(".md", "")
+
+            try:
+                source_content = source_path.read_text()
+            except Exception:
+                continue
+
+            source_fm, source_raw_fm, source_body = parse_yaml_frontmatter(source_content)
+            if source_fm is None:
+                continue
+
+            existing_claims = source_fm.get("claims_extracted", [])
+            if isinstance(existing_claims, str):
+                existing_claims = [existing_claims]
+            if not isinstance(existing_claims, list):
+                existing_claims = []
+
+            if claim_filename in existing_claims:
+                continue
+
+            existing_claims.append(claim_filename)
+            new_block = "claims_extracted:\n" + "\n".join(f"- {c}" for c in existing_claims)
+
+            lines = source_content.split("\n")
+            if "claims_extracted:" not in source_content:
+                end_idx = None
+                for i, line in enumerate(lines):
+                    if i > 0 and line.strip() == "---":
+                        end_idx = i
+                        break
+                if end_idx is None:
+                    continue
+                lines.insert(end_idx, new_block)
+            else:
+                start_idx = None
+                end_idx = None
+                for i, line in enumerate(lines):
+                    if line.startswith("claims_extracted:"):
+                        start_idx = i
+                    elif start_idx is not None and not line.startswith("- "):
+                        end_idx = i
+                        break
+                if start_idx is None:
+                    continue
+                if end_idx is None:
+                    end_idx = len(lines)
+                lines[start_idx:end_idx] = new_block.split("\n")
+
+            modified_sources[str(source_path)] = "\n".join(lines)
+            logger.info("backlink_source_claims: added %s to %s", claim_filename, sourced_from)
+
+        if modified_sources:
+            async with async_main_worktree_lock():
+                for sp, content in modified_sources.items():
+                    Path(sp).write_text(content)
+                    await git_fn("add", sp, cwd=str(config.MAIN_WORKTREE))
+                rc, out = await git_fn(
+                    "commit", "-m", f"backlink: update claims_extracted on {len(modified_sources)} source(s)",
+                    cwd=str(config.MAIN_WORKTREE),
+                    timeout=15,
+                )
+                if rc == 0:
+                    push_rc, push_out = await git_fn(
+                        "push", "origin", "main",
+                        cwd=str(config.MAIN_WORKTREE),
+                        timeout=30,
+                    )
+                    if push_rc == 0:
+                        logger.info("backlink_source_claims: %d source(s) updated and pushed", len(modified_sources))
+                    else:
+                        logger.warning("backlink_source_claims: push failed: %s", push_out[:200])
+                else:
+                    logger.warning("backlink_source_claims: commit failed: %s", out[:200])
+
+    except Exception:
+        logger.exception("backlink_source_claims: failed (non-fatal)")
+
+
 def archive_source_for_pr(branch: str, domain: str, merged: bool = True):
     """Move source from queue/ to archive/{domain}/ after PR merge or close.
 

ops/backfill-contributor-roles.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""Backfill contributor role counts from prs.commit_type.
+
+Resets all role counts to 0, then re-derives them from the prs table's
+commit_type column using the COMMIT_TYPE_TO_ROLE mapping. This corrects
+the bug where all contributors were recorded as 'extractor' regardless
+of their actual commit_type.
+
+Usage:
+    python3 ops/backfill-contributor-roles.py [--dry-run]
+"""
+
+import argparse
+import sqlite3
+import sys
+import os
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from lib.contributor import COMMIT_TYPE_TO_ROLE, commit_type_to_role
+
+DB_PATH = os.environ.get("PIPELINE_DB", "/opt/teleo-eval/pipeline/pipeline.db")
+
+
+def backfill(db_path: str, dry_run: bool = False):
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+
+    # Get all merged PRs with commit_type and agent
+    prs = conn.execute("""
+        SELECT number, commit_type, agent, branch
+        FROM prs
+        WHERE status = 'merged' AND agent IS NOT NULL
+        ORDER BY number
+    """).fetchall()
+
+    print(f"Processing {len(prs)} merged PRs...")
+
+    # Reset all role counts
+    if not dry_run:
+        conn.execute("""
+            UPDATE contributors SET
+                extractor_count = 0,
+                challenger_count = 0,
+                synthesizer_count = 0,
+                sourcer_count = 0
+        """)
+        print("Reset all role counts to 0")
+
+    # Tally roles from commit_type
+    role_counts: dict[str, dict[str, int]] = {}
+    for pr in prs:
+        agent = pr["agent"].lower() if pr["agent"] else None
+        if not agent or agent in ("external", "pipeline"):
+            continue
+
+        commit_type = pr["commit_type"] or "extract"
+        role = commit_type_to_role(commit_type)
+
+        if agent not in role_counts:
+            role_counts[agent] = {
+                "extractor_count": 0, "challenger_count": 0,
+                "synthesizer_count": 0, "sourcer_count": 0,
+                "reviewer_count": 0,
+            }
+        role_col = f"{role}_count"
+        if role_col in role_counts[agent]:
+            role_counts[agent][role_col] += 1
+
+    # Apply tallied counts
+    for handle, counts in sorted(role_counts.items()):
+        non_zero = {k: v for k, v in counts.items() if v > 0}
+        print(f"  {handle}: {non_zero or '(no knowledge PRs)'}")
+        if not dry_run and non_zero:
+            set_clauses = ", ".join(f"{k} = {v}" for k, v in non_zero.items())
+            conn.execute(
+                f"UPDATE contributors SET {set_clauses}, updated_at = datetime('now') WHERE handle = ?",
+                (handle,),
+            )
+
+    if not dry_run:
+        conn.commit()
+        print("\nBackfill committed.")
+    else:
+        print("\n[DRY RUN] No changes made.")
+
+    # Print summary
+    print("\nRole distribution across all contributors:")
+    if not dry_run:
+        rows = conn.execute("""
+            SELECT handle, extractor_count, challenger_count, synthesizer_count,
+                   sourcer_count, reviewer_count
+            FROM contributors
+            ORDER BY (extractor_count + challenger_count + synthesizer_count) DESC
+        """).fetchall()
+        for r in rows:
+            parts = []
+            if r["extractor_count"]: parts.append(f"extract:{r['extractor_count']}")
+            if r["challenger_count"]: parts.append(f"challenge:{r['challenger_count']}")
+            if r["synthesizer_count"]: parts.append(f"synthesize:{r['synthesizer_count']}")
+            if r["sourcer_count"]: parts.append(f"source:{r['sourcer_count']}")
+            if r["reviewer_count"]: parts.append(f"review:{r['reviewer_count']}")
+            if parts:
+                print(f"  {r['handle']}: {', '.join(parts)}")
+
+    conn.close()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--db", default=DB_PATH)
+    args = parser.parse_args()
+    backfill(args.db, args.dry_run)

scripts/audit-wiki-links.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+"""Audit wiki-links across the teleo-codex knowledge base.
+
+Crawls domains/, foundations/, core/, decisions/ for [[wiki-links]].
+Resolves each link against known claim files, entity files, and _map files.
+Reports dead links, orphaned claims, and link counts.
+
+Output: JSON to stdout with dead links, orphans, and per-file link counts.
+"""
+
+import json
+import os
+import re
+import sys
+import unicodedata
+from pathlib import Path
+
+CODEX_ROOT = Path(os.environ.get("CODEX_ROOT", "/opt/teleo-eval/workspaces/main"))
+CLAIM_DIRS = ["domains", "foundations", "core", "decisions"]
+ENTITY_DIR = "entities"
+
+WIKI_LINK_RE = re.compile(r"\[\[([^\]]+)\]\]")
+
+
+def slugify(title: str) -> str:
+    """Convert a wiki-link title to the kebab-case slug used for filenames."""
+    s = title.strip().lower()
+    s = unicodedata.normalize("NFKD", s)
+    s = re.sub(r"[^\w\s-]", "", s)
+    s = re.sub(r"[\s_]+", "-", s)
+    s = re.sub(r"-+", "-", s)
+    return s.strip("-")
+
+
+def build_index(codex: Path) -> dict:
+    """Build a lookup index of all resolvable targets.
+
+    Returns dict mapping normalized slug -> file path.
+    Also maps raw stem (filename without .md) -> file path.
+    """
+    index = {}
+
+    # Index claim files across all claim directories
+    for claim_dir in CLAIM_DIRS:
+        d = codex / claim_dir
+        if not d.exists():
+            continue
+        for md in d.rglob("*.md"):
+            stem = md.stem
+            rel = str(md.relative_to(codex))
+            # Map by stem (exact filename match)
+            index[stem.lower()] = rel
+            # Map by slugified stem
+            index[slugify(stem)] = rel
+
+    # Index entity files
+    entity_root = codex / ENTITY_DIR
+    if entity_root.exists():
+        for md in entity_root.rglob("*.md"):
+            stem = md.stem
+            rel = str(md.relative_to(codex))
+            index[stem.lower()] = rel
+            index[slugify(stem)] = rel
+
+    # Index maps/ directory (MOC-style overview docs)
+    maps_root = codex / "maps"
+    if maps_root.exists():
+        for md in maps_root.rglob("*.md"):
+            stem = md.stem
+            rel = str(md.relative_to(codex))
+            index[stem.lower()] = rel
+            index[slugify(stem)] = rel
+
+    # Index top-level docs that might be link targets
+    for special in ["overview.md", "livingip-overview.md"]:
+        p = codex / special
+        if p.exists():
+            index[p.stem.lower()] = str(p.relative_to(codex))
+
+    # Index agents/ beliefs and positions (sometimes linked)
+    agents_dir = codex / "agents"
+    if agents_dir.exists():
+        for md in agents_dir.rglob("*.md"):
+            stem = md.stem
+            rel = str(md.relative_to(codex))
+            index[stem.lower()] = rel
+
+    return index
+
+
+def resolve_link(link_text: str, index: dict, source_dir: str) -> str | None:
+    """Try to resolve a wiki-link target. Returns file path or None."""
+    text = link_text.strip()
+
+    # Special case: [[_map]] resolves to _map.md in the same domain directory
+    if text == "_map":
+        parts = source_dir.split("/")
+        if len(parts) >= 2:
+            candidate = f"{parts[0]}/{parts[1]}/_map.md"
+            if (CODEX_ROOT / candidate).exists():
+                return candidate
+        return None
+
+    # Path-style references like [[domains/health/_map]]
+    if "/" in text:
+        candidate = text.rstrip("/")
+        if not candidate.endswith(".md"):
+            candidate += ".md"
+        if (CODEX_ROOT / candidate).exists():
+            return candidate
+        return None
+
+    # Try exact stem match (lowercased)
+    key = text.lower()
+    if key in index:
+        return index[key]
+
+    # Try slugified version
+    slug = slugify(text)
+    if slug in index:
+        return index[slug]
+
+    # Try with common variations
+    for variant in [
+        slug.replace("metadaos", "metadao"),
+        slug.replace("ais", "ai"),
+    ]:
+        if variant in index:
+            return index[variant]
+
+    return None
+
+
+def audit(codex: Path) -> dict:
+    """Run the full wiki-link audit."""
+    index = build_index(codex)
+
+    dead_links = []       # {file, link, line_number}
+    link_counts = {}      # file -> {outbound: N, targets: []}
+    all_targets = set()   # files that are linked TO
+    all_files = set()     # all claim/foundation files
+
+    # Scan all markdown files in claim directories
+    for claim_dir in CLAIM_DIRS:
+        d = codex / claim_dir
+        if not d.exists():
+            continue
+        for md in d.rglob("*.md"):
+            rel = str(md.relative_to(codex))
+            all_files.add(rel)
+            source_dir = str(md.parent.relative_to(codex))
+
+            try:
+                content = md.read_text(encoding="utf-8")
+            except Exception:
+                continue
+
+            links_in_file = []
+            for i, line in enumerate(content.split("\n"), 1):
+                for match in WIKI_LINK_RE.finditer(line):
+                    link_text = match.group(1)
+                    # Skip links with | (display text aliases) - take the target part
+                    if "|" in link_text:
+                        link_text = link_text.split("|")[0].strip()
+
+                    resolved = resolve_link(link_text, index, source_dir)
+                    if resolved:
+                        all_targets.add(resolved)
+                        links_in_file.append(resolved)
+                    else:
+                        dead_links.append({
+                            "file": rel,
+                            "link": link_text,
+                            "line": i,
+                        })
+
+            link_counts[rel] = {
+                "outbound": len(links_in_file),
+                "targets": links_in_file,
+            }
+
+    # Find orphaned claims (no inbound links AND no outbound links)
+    files_with_outbound = {f for f, c in link_counts.items() if c["outbound"] > 0}
+    orphaned = sorted(
+        f for f in all_files
+        if f not in all_targets
+        and f not in files_with_outbound
+        and not f.endswith("_map.md")  # MOC files are structural, not orphans
+    )
+
+    # Compute inbound link counts
+    inbound_counts = {}
+    for f, c in link_counts.items():
+        for target in c["targets"]:
+            inbound_counts[target] = inbound_counts.get(target, 0) + 1
+
+    # Claims with high outbound (good connectivity)
+    high_connectivity = sorted(
+        [(f, c["outbound"]) for f, c in link_counts.items() if c["outbound"] >= 3],
+        key=lambda x: -x[1],
+    )
+
+    # Summary stats
+    total_links = sum(c["outbound"] for c in link_counts.values())
+    files_with_links = sum(1 for c in link_counts.values() if c["outbound"] > 0)
+
+    # Domain breakdown of dead links
+    dead_by_domain = {}
+    for dl in dead_links:
+        parts = dl["file"].split("/")
+        domain = parts[1] if len(parts) >= 3 else parts[0]
+        dead_by_domain[domain] = dead_by_domain.get(domain, 0) + 1
+
+    # Domain breakdown of orphans
+    orphan_by_domain = {}
+    for o in orphaned:
+        parts = o.split("/")
+        domain = parts[1] if len(parts) >= 3 else parts[0]
+        orphan_by_domain[domain] = orphan_by_domain.get(domain, 0) + 1
+
+    return {
+        "summary": {
+            "total_files": len(all_files),
+            "total_links": total_links,
+            "files_with_links": files_with_links,
+            "files_without_links": len(all_files) - files_with_links,
+            "dead_link_count": len(dead_links),
+            "orphan_count": len(orphaned),
+            "avg_links_per_file": round(total_links / max(len(all_files), 1), 2),
+            "high_connectivity_count": len(high_connectivity),
+        },
+        "dead_links": dead_links,
+        "dead_by_domain": dict(sorted(dead_by_domain.items(), key=lambda x: -x[1])),
+        "orphaned": orphaned,
+        "orphan_by_domain": dict(sorted(orphan_by_domain.items(), key=lambda x: -x[1])),
+        "high_connectivity": [{"file": f, "outbound_links": n} for f, n in high_connectivity[:20]],
+        "inbound_top20": sorted(
+            [{"file": f, "inbound_links": n} for f, n in inbound_counts.items()],
+            key=lambda x: -x["inbound_links"],
+        )[:20],
+    }
+
+
+if __name__ == "__main__":
+    codex = Path(sys.argv[1]) if len(sys.argv) > 1 else CODEX_ROOT
+    result = audit(codex)
+    json.dump(result, sys.stdout, indent=2)
+    print()
+
+    # Print human-readable summary to stderr
+    s = result["summary"]
+    print(f"\n=== Wiki-Link Audit ===", file=sys.stderr)
+    print(f"Files scanned: {s['total_files']}", file=sys.stderr)
+    print(f"Total links: {s['total_links']}", file=sys.stderr)
+    print(f"Files with links: {s['files_with_links']} ({100*s['files_with_links']//max(s['total_files'],1)}%)", file=sys.stderr)
+    print(f"Dead links: {s['dead_link_count']}", file=sys.stderr)
+    print(f"Orphaned claims: {s['orphan_count']}", file=sys.stderr)
+    print(f"Avg links/file: {s['avg_links_per_file']}", file=sys.stderr)
+    print(f"High connectivity (≥3 links): {s['high_connectivity_count']}", file=sys.stderr)

scripts/contributor-graph.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Generate cumulative contributor + claims PNG for Twitter embedding."""
+
+import json
+import subprocess
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import matplotlib.dates as mdates
+from matplotlib.ticker import MaxNLocator
+
+ACCENT = "#00d4aa"
+PURPLE = "#7c3aed"
+BG = "#0a0a0a"
+TEXT = "#e0e0e0"
+SUBTLE = "#555555"
+OUTPUT = Path("/opt/teleo-eval/static/contributor-graph.png")
+
+
+def get_data():
+    """Fetch from local API."""
+    import urllib.request
+    with urllib.request.urlopen("http://localhost:8081/api/contributor-growth") as r:
+        return json.loads(r.read())
+
+
+def build_continuous_series(milestones, start_date, end_date):
+    """Expand milestone-only contributor data into daily series."""
+    dates = []
+    values = []
+    current = 0
+    milestone_map = {}
+    for m in milestones:
+        d = datetime.strptime(m["date"], "%Y-%m-%d").date()
+        milestone_map[d] = m["cumulative"]
+
+    d = start_date
+    while d <= end_date:
+        if d in milestone_map:
+            current = milestone_map[d]
+        dates.append(d)
+        values.append(current)
+        d += timedelta(days=1)
+    return dates, values
+
+
+def render(data, output_path):
+    fig, ax1 = plt.subplots(figsize=(12, 6.3), dpi=100)
+    fig.patch.set_facecolor(BG)
+    ax1.set_facecolor(BG)
+
+    claims = data["cumulative_claims"]
+    contribs = data["cumulative_contributors"]
+
+    claim_dates = [datetime.strptime(c["date"], "%Y-%m-%d").date() for c in claims]
+    claim_values = [c["cumulative"] for c in claims]
+
+    start = min(claim_dates)
+    end = max(claim_dates)
+
+    contrib_dates, contrib_values = build_continuous_series(contribs, start, end)
+
+    # Claims line (left y-axis)
+    ax1.fill_between(claim_dates, claim_values, alpha=0.15, color=ACCENT)
+    ax1.plot(claim_dates, claim_values, color=ACCENT, linewidth=2.5, label="Claims")
+    ax1.set_ylabel("Claims", color=ACCENT, fontsize=12, fontweight="bold")
+    ax1.tick_params(axis="y", colors=ACCENT, labelsize=10)
+    ax1.set_ylim(bottom=0)
+
+    # Contributors line (right y-axis)
+    ax2 = ax1.twinx()
+    ax2.set_facecolor("none")
+    ax2.fill_between(contrib_dates, contrib_values, alpha=0.1, color=PURPLE, step="post")
+    ax2.step(contrib_dates, contrib_values, color=PURPLE, linewidth=2.5,
+             where="post", label="Contributors")
+    ax2.set_ylabel("Contributors", color=PURPLE, fontsize=12, fontweight="bold")
+    ax2.tick_params(axis="y", colors=PURPLE, labelsize=10)
+    ax2.yaxis.set_major_locator(MaxNLocator(integer=True))
+    ax2.set_ylim(bottom=0, top=max(contrib_values) * 1.8)
+
+    # Annotate contributor milestones with staggered offsets to avoid overlap
+    offsets = {}
+    for i, m in enumerate(contribs):
+        d = datetime.strptime(m["date"], "%Y-%m-%d").date()
+        val = m["cumulative"]
+        names = [n["name"] for n in m["new"]]
+        if len(names) <= 2:
+            label = ", ".join(names)
+        else:
+            label = f"+{len(names)}"
+        y_off = 8 + (i % 2) * 14
+        ax2.annotate(label, (d, val),
+                     textcoords="offset points", xytext=(5, y_off),
+                     fontsize=7, color=PURPLE, alpha=0.8)
+
+    # Hero stats
+    total_claims = data["summary"]["total_claims"]
+    total_contribs = data["summary"]["total_contributors"]
+    days = data["summary"]["days_active"]
+    fig.text(0.14, 0.88, f"{total_claims:,} claims", fontsize=22,
+             color=ACCENT, fontweight="bold", ha="left")
+    fig.text(0.14, 0.82, f"{total_contribs} contributors · {days} days",
+             fontsize=13, color=TEXT, ha="left", alpha=0.7)
+
+    # X-axis
+    ax1.xaxis.set_major_formatter(mdates.DateFormatter("%b %d"))
+    ax1.xaxis.set_major_locator(mdates.WeekdayLocator(interval=2))
+    ax1.tick_params(axis="x", colors=SUBTLE, labelsize=9, rotation=0)
+
+    # Remove spines
+    for ax in [ax1, ax2]:
+        for spine in ax.spines.values():
+            spine.set_visible(False)
+
+    # Subtle grid on claims axis only
+    ax1.grid(axis="y", color=SUBTLE, alpha=0.2, linewidth=0.5)
+    ax1.set_axisbelow(True)
+
+    # Branding
+    fig.text(0.98, 0.02, "livingip.xyz", fontsize=9, color=SUBTLE,
+             ha="right", style="italic")
+
+    plt.tight_layout(rect=[0, 0.03, 1, 0.78])
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    fig.savefig(output_path, facecolor=BG, bbox_inches="tight", pad_inches=0.3)
+    plt.close(fig)
+    print(f"Saved to {output_path} ({output_path.stat().st_size:,} bytes)")
+
+
+if __name__ == "__main__":
+    out = Path(sys.argv[1]) if len(sys.argv) > 1 else OUTPUT
+    data = get_data()
+    render(data, out)

scripts/cumulative-growth.py

@@ -0,0 +1,223 @@
+#!/usr/bin/env python3
+"""Generate cumulative growth time-series data for public dashboard.
+
+Produces JSON with three series:
+  - cumulative_contributors: unique git authors over time
+  - cumulative_claims: domain claim files added over time
+  - github_stars: star count snapshots (requires GitHub API)
+
+Data sources: git log (codex repo), GitHub API.
+Output: JSON to stdout or file, suitable for Chart.js line charts.
+
+Usage:
+  python3 cumulative-growth.py --codex-path /path/to/teleo-codex [--output /path/to/output.json]
+  python3 cumulative-growth.py --codex-path /path/to/teleo-codex --format csv
+"""
+
+import argparse
+import json
+import subprocess
+import sys
+from collections import defaultdict
+from datetime import datetime, timedelta
+
+# Map bot/service accounts to their human principal or exclude them.
+# "Teleo Agents" and "Teleo Pipeline" are bot accounts — attribute to system.
+CONTRIBUTOR_ALIASES = {
+    "Teleo Agents": None,   # system automation, not a contributor
+    "Teleo Pipeline": None, # pipeline bot
+}
+
+# Founding contributors get a badge — anyone who contributed before this date.
+FOUNDING_CUTOFF = "2026-03-15"
+
+
+def git_log_contributors(codex_path: str) -> list[dict]:
+    """Extract per-commit author and date from git log."""
+    result = subprocess.run(
+        ["git", "log", "--format=%ad|%an", "--date=format:%Y-%m-%d", "--all"],
+        capture_output=True, text=True, cwd=codex_path
+    )
+    if result.returncode != 0:
+        print(f"git log failed: {result.stderr}", file=sys.stderr)
+        sys.exit(1)
+
+    entries = []
+    for line in result.stdout.strip().split("\n"):
+        if "|" not in line:
+            continue
+        date, author = line.split("|", 1)
+        canonical = CONTRIBUTOR_ALIASES.get(author, author)
+        if canonical is None:
+            continue
+        entries.append({"date": date, "author": canonical})
+    return entries
+
+
+def git_log_claims(codex_path: str) -> list[dict]:
+    """Extract claim file additions over time from git log."""
+    result = subprocess.run(
+        ["git", "log", "--format=%ad", "--date=format:%Y-%m-%d",
+         "--all", "--diff-filter=A", "--", "domains/*.md"],
+        capture_output=True, text=True, cwd=codex_path
+    )
+    if result.returncode != 0:
+        print(f"git log failed: {result.stderr}", file=sys.stderr)
+        sys.exit(1)
+
+    counts = defaultdict(int)
+    for line in result.stdout.strip().split("\n"):
+        line = line.strip()
+        if line:
+            counts[line] += 1
+    return [{"date": d, "count": c} for d, c in sorted(counts.items())]
+
+
+def github_stars(repo: str = "living-ip/teleo-codex") -> int | None:
+    """Fetch current star count from GitHub API. Returns None on failure."""
+    try:
+        result = subprocess.run(
+            ["gh", "api", f"repos/{repo}", "--jq", ".stargazers_count"],
+            capture_output=True, text=True, timeout=10
+        )
+        if result.returncode == 0:
+            return int(result.stdout.strip())
+    except (subprocess.TimeoutExpired, ValueError):
+        pass
+    return None
+
+
+def build_cumulative_contributors(entries: list[dict]) -> list[dict]:
+    """Build cumulative unique contributor count by date."""
+    first_seen = {}
+    for e in entries:
+        author, date = e["author"], e["date"]
+        if author not in first_seen or date < first_seen[author]:
+            first_seen[author] = date
+
+    by_date = defaultdict(list)
+    for author, date in first_seen.items():
+        by_date[date].append(author)
+
+    timeline = []
+    seen = set()
+    for date in sorted(by_date.keys()):
+        new_authors = by_date[date]
+        seen.update(new_authors)
+        is_founding = date <= FOUNDING_CUTOFF
+        timeline.append({
+            "date": date,
+            "cumulative": len(seen),
+            "new": [
+                {"name": a, "founding": is_founding}
+                for a in sorted(new_authors)
+            ],
+        })
+    return timeline
+
+
+def build_cumulative_claims(claim_entries: list[dict]) -> list[dict]:
+    """Build cumulative claim count by date."""
+    timeline = []
+    cumulative = 0
+    for entry in claim_entries:
+        cumulative += entry["count"]
+        timeline.append({
+            "date": entry["date"],
+            "cumulative": cumulative,
+            "added": entry["count"],
+        })
+    return timeline
+
+
+def build_daily_commits(entries: list[dict]) -> list[dict]:
+    """Build daily commit volume by contributor."""
+    daily = defaultdict(lambda: defaultdict(int))
+    for e in entries:
+        daily[e["date"]][e["author"]] += 1
+
+    timeline = []
+    for date in sorted(daily.keys()):
+        authors = daily[date]
+        timeline.append({
+            "date": date,
+            "total": sum(authors.values()),
+            "by_contributor": dict(sorted(authors.items())),
+        })
+    return timeline
+
+
+def generate_report(codex_path: str) -> dict:
+    entries = git_log_contributors(codex_path)
+    claim_entries = git_log_claims(codex_path)
+    stars = github_stars()
+
+    contributors_timeline = build_cumulative_contributors(entries)
+    claims_timeline = build_cumulative_claims(claim_entries)
+    commits_timeline = build_daily_commits(entries)
+
+    all_contributors = set(e["author"] for e in entries)
+    founding = [
+        a for a in all_contributors
+        if any(
+            e["date"] <= FOUNDING_CUTOFF and e["author"] == a
+            for e in entries
+        )
+    ]
+
+    return {
+        "generated_at": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "summary": {
+            "total_contributors": len(all_contributors),
+            "founding_contributors": sorted(founding),
+            "total_claims": claims_timeline[-1]["cumulative"] if claims_timeline else 0,
+            "github_stars": stars,
+            "codex_start_date": "2026-03-05",
+            "days_active": (datetime.utcnow() - datetime(2026, 3, 5)).days,
+        },
+        "cumulative_contributors": contributors_timeline,
+        "cumulative_claims": claims_timeline,
+        "daily_activity": commits_timeline,
+    }
+
+
+def format_csv(report: dict) -> str:
+    lines = ["date,cumulative_contributors,cumulative_claims"]
+    contrib_map = {e["date"]: e["cumulative"] for e in report["cumulative_contributors"]}
+    claims_map = {e["date"]: e["cumulative"] for e in report["cumulative_claims"]}
+
+    all_dates = sorted(set(list(contrib_map.keys()) + list(claims_map.keys())))
+
+    last_contrib = 0
+    last_claims = 0
+    for d in all_dates:
+        last_contrib = contrib_map.get(d, last_contrib)
+        last_claims = claims_map.get(d, last_claims)
+        lines.append(f"{d},{last_contrib},{last_claims}")
+    return "\n".join(lines)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Generate cumulative growth data")
+    parser.add_argument("--codex-path", required=True, help="Path to teleo-codex repo")
+    parser.add_argument("--output", help="Output file path (default: stdout)")
+    parser.add_argument("--format", choices=["json", "csv"], default="json")
+    args = parser.parse_args()
+
+    report = generate_report(args.codex_path)
+
+    if args.format == "csv":
+        output = format_csv(report)
+    else:
+        output = json.dumps(report, indent=2)
+
+    if args.output:
+        with open(args.output, "w") as f:
+            f.write(output)
+        print(f"Written to {args.output}", file=sys.stderr)
+    else:
+        print(output)
+
+
+if __name__ == "__main__":
+    main()

scripts/scoring_digest.py

@@ -0,0 +1,561 @@
+#!/usr/bin/env python3
+"""Daily scoring digest — classify, score, and broadcast KB contributions.
+
+Runs daily at 8:07 AM London via cron.
+Queries pipeline.db for merged PRs in last 24h, classifies each as
+CREATE/ENRICH/CHALLENGE, scores with importance multiplier and connectivity
+bonus, updates contributors table, posts summary to Telegram.
+
+Spec: Pentagon/sprints/contribution-scoring-algorithm.md
+"""
+
+import json
+import logging
+import os
+import re
+import sqlite3
+import subprocess
+import sys
+import urllib.request
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+log = logging.getLogger("scoring_digest")
+
+# --- Configuration ---
+BASE_DIR = Path(os.environ.get("PIPELINE_BASE", "/opt/teleo-eval"))
+DB_PATH = BASE_DIR / "pipeline" / "pipeline.db"
+CODEX_DIR = BASE_DIR / "workspaces" / "main"
+TELEGRAM_TOKEN_FILE = BASE_DIR / "secrets" / "telegram-bot-token"
+TELEGRAM_CHAT_ID = 2091295364
+DIGEST_JSON_PATH = BASE_DIR / "logs" / "scoring-digest-latest.json"
+LONDON_TZ = ZoneInfo("Europe/London")
+
+# --- Action weights (Leo spec Apr 20) ---
+ACTION_WEIGHTS = {
+    "challenge": 0.40,
+    "create": 0.35,
+    "enrich": 0.25,
+}
+
+# --- Confidence → base importance mapping ---
+CONFIDENCE_BASE = {
+    "proven": 2.0,
+    "likely": 1.5,
+    "experimental": 1.0,
+    "speculative": 1.0,
+    "possible": 1.0,
+    "plausible": 1.0,
+    "medium": 1.5,
+}
+
+DOMAIN_CLAIM_COUNTS: dict[str, int] = {}
+ENTITY_SLUGS: set[str] = set()
+CLAIM_SLUGS: set[str] = set()
+MAP_FILES: set[str] = set()
+
+
+def _slugify(title: str) -> str:
+    s = title.lower().strip()
+    s = re.sub(r"[^\w\s-]", "", s)
+    s = re.sub(r"[\s_]+", "-", s)
+    return s.strip("-")
+
+
+def _init_link_index():
+    """Build indexes for wiki-link resolution."""
+    global ENTITY_SLUGS, CLAIM_SLUGS, MAP_FILES
+
+    entities_dir = CODEX_DIR / "entities"
+    if entities_dir.exists():
+        for f in entities_dir.glob("*.md"):
+            ENTITY_SLUGS.add(f.stem.lower())
+
+    for domain_dir in (CODEX_DIR / "domains").iterdir():
+        if not domain_dir.is_dir():
+            continue
+        for f in domain_dir.glob("*.md"):
+            CLAIM_SLUGS.add(f.stem.lower())
+        map_file = domain_dir / "_map.md"
+        if map_file.exists():
+            MAP_FILES.add("_map")
+            MAP_FILES.add(f"domains/{domain_dir.name}/_map")
+
+    for f in (CODEX_DIR / "foundations").glob("*.md") if (CODEX_DIR / "foundations").exists() else []:
+        CLAIM_SLUGS.add(f.stem.lower())
+    for f in (CODEX_DIR / "core").glob("*.md") if (CODEX_DIR / "core").exists() else []:
+        CLAIM_SLUGS.add(f.stem.lower())
+    for f in (CODEX_DIR / "decisions").glob("*.md") if (CODEX_DIR / "decisions").exists() else []:
+        CLAIM_SLUGS.add(f.stem.lower())
+
+
+def _resolve_link(link_text: str) -> bool:
+    """Check if a [[wiki-link]] resolves to a known entity, claim, or map."""
+    slug = _slugify(link_text)
+    return (
+        slug in ENTITY_SLUGS
+        or slug in CLAIM_SLUGS
+        or slug in MAP_FILES
+        or link_text.lower() in MAP_FILES
+    )
+
+
+def _count_resolved_wiki_links(file_path: Path) -> int:
+    """Count wiki-links in a claim file that resolve to real targets."""
+    if not file_path.exists():
+        return 0
+    try:
+        text = file_path.read_text(encoding="utf-8")
+    except Exception:
+        return 0
+
+    links = re.findall(r"\[\[([^\]]+)\]\]", text)
+    return sum(1 for link in links if _resolve_link(link))
+
+
+def _get_confidence(file_path: Path) -> str:
+    """Extract confidence field from claim frontmatter."""
+    if not file_path.exists():
+        return "experimental"
+    try:
+        text = file_path.read_text(encoding="utf-8")
+    except Exception:
+        return "experimental"
+
+    m = re.search(r"^confidence:\s*(\S+)", text, re.MULTILINE)
+    return m.group(1).strip() if m else "experimental"
+
+
+def _has_cross_domain_ref(file_path: Path) -> bool:
+    """Check if claim references another domain via secondary_domains or cross-domain links."""
+    if not file_path.exists():
+        return False
+    try:
+        text = file_path.read_text(encoding="utf-8")
+    except Exception:
+        return False
+
+    if re.search(r"^secondary_domains:\s*\[.+\]", text, re.MULTILINE):
+        return True
+    if re.search(r"^depends_on:", text, re.MULTILINE):
+        return True
+    return False
+
+
+def _has_challenged_by(file_path: Path) -> bool:
+    """Check if claim has challenged_by field."""
+    if not file_path.exists():
+        return False
+    try:
+        text = file_path.read_text(encoding="utf-8")
+    except Exception:
+        return False
+    return bool(re.search(r"^challenged_by:", text, re.MULTILINE))
+
+
+def _get_domain_weight(domain: str) -> float:
+    """Domain maturity weight: sparse domains get bonus, mature domains get discount."""
+    count = DOMAIN_CLAIM_COUNTS.get(domain, 0)
+    if count < 20:
+        return 1.5
+    elif count > 50:
+        return 0.8
+    return 1.0
+
+
+def _init_domain_counts():
+    """Count claims per domain."""
+    global DOMAIN_CLAIM_COUNTS
+    domains_dir = CODEX_DIR / "domains"
+    if not domains_dir.exists():
+        return
+    for domain_dir in domains_dir.iterdir():
+        if domain_dir.is_dir():
+            count = sum(1 for f in domain_dir.glob("*.md") if f.name != "_map.md")
+            DOMAIN_CLAIM_COUNTS[domain_dir.name] = count
+
+
+def _normalize_contributor(submitted_by: str | None, agent: str | None, branch: str | None = None) -> str:
+    """Normalize contributor handle — strip @, map agent self-directed to agent name.
+
+    For fork PRs (contrib/NAME/...), extract contributor from branch name.
+    """
+    if branch and branch.startswith("contrib/"):
+        parts = branch.split("/")
+        if len(parts) >= 2 and parts[1]:
+            return parts[1].lower()
+
+    raw = submitted_by or agent or "unknown"
+    raw = raw.strip()
+    if raw.startswith("@"):
+        raw = raw[1:]
+    if " (self-directed)" in raw:
+        raw = raw.replace(" (self-directed)", "")
+    if raw in ("pipeline", ""):
+        return agent.strip() if agent and agent.strip() not in ("pipeline", "") else "pipeline"
+    return raw
+
+
+def classify_pr(pr: dict) -> str | None:
+    """Classify a merged PR as create/enrich/challenge or None (skip).
+
+    Uses branch name pattern + commit_type as primary signal.
+    Falls back to file-level analysis for ambiguous cases.
+    """
+    branch = pr.get("branch", "")
+    commit_type = pr.get("commit_type", "")
+
+    if commit_type in ("pipeline", "entity"):
+        return None
+
+    if "challenge" in branch.lower():
+        return "challenge"
+
+    if branch.startswith("extract/") or branch.startswith("research-"):
+        return "create"
+
+    if "reweave" in branch.lower() or "enrich" in branch.lower():
+        return "enrich"
+
+    if commit_type == "research":
+        return "create"
+
+    if commit_type == "reweave":
+        return "enrich"
+
+    if commit_type == "fix":
+        return "enrich"
+
+    if commit_type == "knowledge":
+        return "create"
+
+    return "create"
+
+
+def _find_claim_file(pr: dict) -> Path | None:
+    """Find the claim file for a merged PR."""
+    domain = pr.get("domain")
+    branch = pr.get("branch", "")
+
+    if not domain:
+        return None
+
+    domain_dir = CODEX_DIR / "domains" / domain
+    if not domain_dir.exists():
+        return None
+
+    slug_part = branch.split("/")[-1] if "/" in branch else branch
+    slug_part = re.sub(r"-[a-f0-9]{4}$", "", slug_part)
+
+    for claim_file in domain_dir.glob("*.md"):
+        if claim_file.name == "_map.md":
+            continue
+        claim_slug = _slugify(claim_file.stem)
+        if slug_part and slug_part in claim_slug:
+            return claim_file
+
+    return None
+
+
+def score_contribution(action_type: str, claim_file: Path | None, domain: str) -> tuple[float, dict]:
+    """Compute CI points for a single contribution.
+
+    Returns (score, breakdown_dict) for transparency.
+    """
+    weight = ACTION_WEIGHTS[action_type]
+
+    confidence = _get_confidence(claim_file) if claim_file else "experimental"
+    base = CONFIDENCE_BASE.get(confidence, 1.0)
+
+    if action_type == "challenge" and claim_file and _has_challenged_by(claim_file):
+        base = 3.0 if confidence in ("proven",) else 2.5
+
+    domain_weight = _get_domain_weight(domain)
+
+    connectivity = 0.0
+    if claim_file and _has_cross_domain_ref(claim_file):
+        connectivity += 0.2
+
+    create_multiplier = 1.0
+    resolved_links = 0
+    if action_type == "create" and claim_file:
+        resolved_links = _count_resolved_wiki_links(claim_file)
+        if resolved_links >= 3:
+            create_multiplier = 1.5
+
+    importance = base * domain_weight + connectivity
+    score = weight * importance * create_multiplier
+
+    return score, {
+        "action": action_type,
+        "weight": weight,
+        "confidence": confidence,
+        "base": base,
+        "domain_weight": domain_weight,
+        "connectivity_bonus": connectivity,
+        "create_multiplier": create_multiplier,
+        "resolved_links": resolved_links,
+        "importance": importance,
+        "score": round(score, 4),
+    }
+
+
+def collect_and_score(hours: int = 24) -> dict:
+    """Main scoring pipeline: collect merged PRs, classify, score."""
+    _init_domain_counts()
+    _init_link_index()
+
+    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
+
+    conn = sqlite3.connect(str(DB_PATH))
+    conn.row_factory = sqlite3.Row
+    try:
+        rows = conn.execute(
+            """SELECT number, branch, domain, agent, commit_type, merged_at,
+                      submitted_by, description
+               FROM prs
+               WHERE status = 'merged' AND merged_at >= ?
+               ORDER BY merged_at DESC""",
+            (cutoff,),
+        ).fetchall()
+    finally:
+        conn.close()
+
+    contributions = []
+    contributor_deltas: dict[str, float] = {}
+    domain_activity: dict[str, int] = {}
+    action_counts = {"create": 0, "enrich": 0, "challenge": 0}
+
+    for row in rows:
+        pr = dict(row)
+        action_type = classify_pr(pr)
+        if action_type is None:
+            continue
+
+        claim_file = _find_claim_file(pr)
+        domain = pr.get("domain", "unknown")
+        score, breakdown = score_contribution(action_type, claim_file, domain)
+
+        contributor = _normalize_contributor(
+            pr.get("submitted_by"), pr.get("agent"), pr.get("branch")
+        )
+        contributor_deltas[contributor] = contributor_deltas.get(contributor, 0) + score
+        domain_activity[domain] = domain_activity.get(domain, 0) + 1
+        action_counts[action_type] = action_counts.get(action_type, 0) + 1
+
+        contributions.append({
+            "pr_number": pr["number"],
+            "contributor": contributor,
+            "agent": pr.get("agent", ""),
+            "domain": domain,
+            "action": action_type,
+            "score": round(score, 4),
+            "breakdown": breakdown,
+            "description": pr.get("description", ""),
+            "merged_at": pr.get("merged_at", ""),
+        })
+
+    total_claims = sum(DOMAIN_CLAIM_COUNTS.values())
+
+    return {
+        "period_hours": hours,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "date": datetime.now(LONDON_TZ).strftime("%B %d, %Y"),
+        "contributions": contributions,
+        "contributor_deltas": {k: round(v, 4) for k, v in sorted(
+            contributor_deltas.items(), key=lambda x: -x[1]
+        )},
+        "domain_activity": dict(sorted(domain_activity.items(), key=lambda x: -x[1])),
+        "action_counts": action_counts,
+        "total_contributions": len(contributions),
+        "total_ci_awarded": round(sum(c["score"] for c in contributions), 4),
+        "kb_state": {
+            "total_claims": total_claims,
+            "domains": len(DOMAIN_CLAIM_COUNTS),
+            "domain_breakdown": dict(DOMAIN_CLAIM_COUNTS),
+        },
+    }
+
+
+def update_contributors(digest: dict):
+    """Write CI deltas to contributors table."""
+    if not digest["contributor_deltas"]:
+        return
+
+    conn = sqlite3.connect(str(DB_PATH))
+    try:
+        for handle, delta in digest["contributor_deltas"].items():
+            conn.execute(
+                """INSERT INTO contributors (handle, claims_merged, created_at, updated_at)
+                   VALUES (?, 0, datetime('now'), datetime('now'))
+                   ON CONFLICT(handle) DO UPDATE SET updated_at = datetime('now')""",
+                (handle,),
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+    log.info("Updated %d contributor records", len(digest["contributor_deltas"]))
+
+
+def save_scores_to_db(digest: dict):
+    """Write individual contribution scores to contribution_scores table."""
+    conn = sqlite3.connect(str(DB_PATH))
+    try:
+        conn.execute("""CREATE TABLE IF NOT EXISTS contribution_scores (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            pr_number INTEGER UNIQUE,
+            contributor TEXT NOT NULL,
+            event_type TEXT CHECK(event_type IN ('create','enrich','challenge')),
+            ci_earned REAL,
+            claim_slug TEXT,
+            domain TEXT,
+            scored_at TEXT NOT NULL
+        )""")
+        for c in digest["contributions"]:
+            slug = (c.get("description") or "")[:200] or c.get("breakdown", {}).get("action", "")
+            conn.execute(
+                """INSERT INTO contribution_scores (pr_number, contributor, event_type, ci_earned, claim_slug, domain, scored_at)
+                   VALUES (?, ?, ?, ?, ?, ?, ?)
+                   ON CONFLICT(pr_number) DO UPDATE SET
+                     contributor = excluded.contributor,
+                     ci_earned = excluded.ci_earned,
+                     event_type = excluded.event_type,
+                     scored_at = excluded.scored_at""",
+                (c["pr_number"], c["contributor"], c["action"], c["score"], slug, c["domain"], c["merged_at"]),
+            )
+        conn.commit()
+        log.info("Wrote %d contribution scores to DB", len(digest["contributions"]))
+    finally:
+        conn.close()
+
+
+def save_digest_json(digest: dict):
+    """Save latest digest as JSON for API consumption."""
+    DIGEST_JSON_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with open(DIGEST_JSON_PATH, "w") as f:
+        json.dump(digest, f, indent=2, default=str)
+    log.info("Saved digest to %s", DIGEST_JSON_PATH)
+
+
+def send_telegram(digest: dict):
+    """Post digest summary to Telegram."""
+    token_file = TELEGRAM_TOKEN_FILE
+    if not token_file.exists():
+        log.warning("Telegram token not found at %s", token_file)
+        return
+
+    token = token_file.read_text().strip()
+
+    lines = [f"📊 *Daily KB Digest — {digest['date']}*", ""]
+
+    if digest["contributions"]:
+        lines.append(f"*NEW CONTRIBUTIONS* (last {digest['period_hours']}h):")
+        action_emoji = {"challenge": "⚔️", "create": "🆕", "enrich": "📚"}
+
+        by_contributor: dict[str, list] = {}
+        for c in digest["contributions"]:
+            name = c["contributor"]
+            by_contributor.setdefault(name, []).append(c)
+
+        for name, contribs in sorted(by_contributor.items(), key=lambda x: -sum(c["score"] for c in x[1])):
+            total_score = sum(c["score"] for c in contribs)
+            actions = {}
+            for c in contribs:
+                actions[c["action"]] = actions.get(c["action"], 0) + 1
+
+            action_summary = ", ".join(
+                f"{action_emoji.get(a, '•')} {n} {a}" for a, n in sorted(actions.items(), key=lambda x: -x[1])
+            )
+            lines.append(f"  {name}: {action_summary} → +{total_score:.2f} CI")
+
+        lines.append("")
+
+    lines.append("*KB STATE:*")
+    kb = digest["kb_state"]
+    ac = digest["action_counts"]
+    lines.append(
+        f"Claims: {kb['total_claims']} (+{digest['total_contributions']}) | "
+        f"Domains: {kb['domains']}"
+    )
+    lines.append(
+        f"Creates: {ac.get('create', 0)} | "
+        f"Enrichments: {ac.get('enrich', 0)} | "
+        f"Challenges: {ac.get('challenge', 0)}"
+    )
+
+    if digest["domain_activity"]:
+        top_domain = max(digest["domain_activity"], key=digest["domain_activity"].get)
+        lines.append(f"Most active: {top_domain} ({digest['domain_activity'][top_domain]} events)")
+
+    if digest["contributor_deltas"]:
+        lines.append("")
+        lines.append("*LEADERBOARD CHANGE:*")
+        for i, (name, delta) in enumerate(digest["contributor_deltas"].items(), 1):
+            if i > 5:
+                break
+            lines.append(f"  #{i} {name} +{delta:.2f} CI")
+
+    text = "\n".join(lines)
+
+    url = f"https://api.telegram.org/bot{token}/sendMessage"
+    payload = json.dumps({
+        "chat_id": TELEGRAM_CHAT_ID,
+        "text": text,
+        "parse_mode": "Markdown",
+    }).encode("utf-8")
+
+    req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            result = json.loads(resp.read())
+            if result.get("ok"):
+                log.info("Telegram digest sent successfully")
+            else:
+                log.error("Telegram API error: %s", result)
+    except Exception as e:
+        log.error("Failed to send Telegram message: %s", e)
+
+
+def main():
+    hours = int(sys.argv[1]) if len(sys.argv) > 1 else 24
+    dry_run = "--dry-run" in sys.argv
+    no_telegram = "--no-telegram" in sys.argv
+
+    log.info("Running scoring digest for last %dh (dry_run=%s)", hours, dry_run)
+
+    digest = collect_and_score(hours)
+
+    log.info(
+        "Scored %d contributions: %d create, %d enrich, %d challenge → %.2f total CI",
+        digest["total_contributions"],
+        digest["action_counts"]["create"],
+        digest["action_counts"]["enrich"],
+        digest["action_counts"]["challenge"],
+        digest["total_ci_awarded"],
+    )
+
+    for name, delta in digest["contributor_deltas"].items():
+        log.info("  %s: +%.4f CI", name, delta)
+
+    if dry_run:
+        print(json.dumps(digest, indent=2, default=str))
+        return
+
+    save_digest_json(digest)
+    save_scores_to_db(digest)
+    update_contributors(digest)
+
+    if not no_telegram:
+        send_telegram(digest)
+
+    log.info("Digest complete")
+
+
+if __name__ == "__main__":
+    main()