Home feed v2: 15-dimension evidence engine + trigger-detection + interleaved feed

Replaces the static 9-signal/5-card-type Home feed with a real evidence-gated
system: 15 tracked dimensions (was 9, plus 3 fold-in sub-signals), a new
categorical evidence path for signals whose mean sits near zero (pacing/energy
arc), and a trigger-detection engine that diffs each session's evidence
against persisted state to fire one of 5 event types (first-time-steady,
drift, recurring, context-shift, single-session anomaly) with a shared
2-session cooldown. Home now interleaves an unconditional per-session recap
card with these frozen, dated dimension events instead of recomputing static
cards from live evidence on every read.

Also reformulates 5 signal computations (curiosity as a rate, room-wide
interruptions, a 4-input reweighted drive score, a building-on-others
denominator fix, and fixed-window vocabulary richness to remove a
session-length confound), and adds two new Supabase tables
(signal_evidence_state, dimension_events) to support the trigger engine.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/main.py

@@ -37,8 +37,11 @@ from pipeline.dimension_scorer import DimensionScorer
 from pipeline.voiceprint import VoiceprintMatcher
 from pipeline.context_detector import ContextDetector
 from pipeline.portrait_synthesizer import PortraitSynthesizer
-from pipeline.evidence_gate import SIGNAL_EVIDENCE_CONFIG, compute_signal_evidence, extract_value
+from pipeline.evidence_gate import (
+    SIGNAL_EVIDENCE_CONFIG, SUB_SIGNAL_EVIDENCE_CONFIG, compute_evidence, extract_value
+)
 from pipeline import home_feed
+from pipeline.trigger_detector import run_trigger_detection
 from pipeline.llm_utils import extract_text
 from anthropic import Anthropic
 from db.database import supabase_admin
@@ -203,15 +206,20 @@ def _compute_profile_evidence(parsed: list) -> dict:
         values_by_signal = _signal_values(sessions)
         result = {}
         for signal_key, values in values_by_signal.items():
-            ev = compute_signal_evidence(signal_key, values)  # filters None internally
+            ev = compute_evidence(signal_key, values)  # filters None internally; dispatches continuous/categorical
+            is_categorical = SIGNAL_EVIDENCE_CONFIG[signal_key]["kind"] == "categorical"
             non_none = [v for v in values if v is not None]  # chronological, oldest→newest
-            # recent-vs-established only computed here for a SINGLE context (see
-            # by_context below) — computing it on the cross-context pool is
-            # misleading: a run of e.g. social sessions can drag the "recent"
-            # average down for reasons that have nothing to do with a genuine
-            # behavioral shift, just a different conversation type happening
-            # recently. Self-relative framing has to stay within one context.
-            if compute_shift and ev["is_steady"] and len(non_none) >= 3:
+            # recent-vs-established shift is only meaningful for a continuous
+            # signal (needs a numeric mean) — categorical dims (pacing_arc,
+            # energy_arc) express "progress" via drift/context-shift events
+            # instead, not a shift_pct card. Also only computed here for a
+            # SINGLE context (see by_context below) — computing it on the
+            # cross-context pool is misleading: a run of e.g. social sessions
+            # can drag the "recent" average down for reasons that have nothing
+            # to do with a genuine behavioral shift, just a different
+            # conversation type happening recently. Self-relative framing has
+            # to stay within one context.
+            if not is_categorical and compute_shift and ev["is_steady"] and len(non_none) >= 3:
                 recent = non_none[-3:]
                 ev["recent_mean"] = round(float(np.mean(recent)), 3)
                 ev["shift_pct"] = (
@@ -234,8 +242,28 @@ def _compute_profile_evidence(parsed: list) -> dict:
     return {"overall": overall, "by_context": by_context}
 
 
+def _compute_sub_signal_evidence(parsed: list) -> dict:
+    """Evidence for the 3 fold-in sub-signals (question_pickup, gets_interrupted,
+    long_turn_rate) — pooled only, no by_context split (subs don't get their own
+    context-shift treatment, see evidence_gate.SUB_SIGNAL_EVIDENCE_CONFIG). Kept
+    separate from _compute_profile_evidence's main 15-dimension dicts so the
+    You-page steady/still-forming lists and profile_strength_pct stay scoped to
+    the real dimensions — this is only consumed for sub-note gating (e.g.
+    portrait_synthesizer's "how it may land" section)."""
+    result = {}
+    for sub_key in SUB_SIGNAL_EVIDENCE_CONFIG:
+        values = []
+        for p in parsed:
+            try:
+                values.append(extract_value(sub_key, p["sig"]))
+            except (KeyError, TypeError):
+                pass
+        result[sub_key] = compute_evidence(sub_key, values, SUB_SIGNAL_EVIDENCE_CONFIG)
+    return result
+
+
 def _get_or_synthesize_portrait(user_id: str, session_count: int, evidence: dict,
-                                blind_spots: list) -> dict:
+                                blind_spots: list, parsed: list) -> dict:
     """Evidence-based replacement for the old dimension-scoring personality
     synthesis (retired). Reuses the user_profiles.personality_json /
     session_count_at_synthesis columns (repurposed, different shape — no schema
@@ -261,7 +289,8 @@ def _get_or_synthesize_portrait(user_id: str, session_count: int, evidence: dict
     except Exception:
         pass
 
-    portrait = portrait_synth.synthesize(evidence, blind_spots, session_count)
+    sub_evidence = _compute_sub_signal_evidence(parsed)
+    portrait = portrait_synth.synthesize(evidence, blind_spots, session_count, sub_evidence)
     _portrait_cache[cache_key] = portrait
 
     try:
@@ -1044,6 +1073,19 @@ def _run_finalize_job(session_id, confirmed_speaker):
         )
         logger.info("[%s]    ✓ Session saved to Supabase", _sid(session_id))
 
+        # Home feed dimension-maturation events — diffs this session's evidence
+        # against persisted state and writes any newly fired event. Wrapped
+        # defensively (same pattern as the consolidation trigger below) so a
+        # bug here can never break finalize itself.
+        try:
+            fired_events = run_trigger_detection(user_id, session_id, primary_context)
+            if fired_events:
+                logger.info("[%s]    ✓ %d dimension event(s) fired: %s", _sid(session_id),
+                            len(fired_events), [e["trigger_type"] for e in fired_events])
+        except Exception:
+            logger.error("[%s]    ✕ Trigger detection failed (non-fatal):\n%s",
+                          _sid(session_id), _tb.format_exc())
+
         # Trigger background consolidation if user has enough sessions
         try:
             count_res = supabase_admin.table("sessions").select(
@@ -1350,23 +1392,30 @@ def get_profile(user_id: str = Depends(get_current_user)):
     # numeric dimension-scoring personality synthesis entirely — no invented
     # dimensions, no 0-100 scores, no trait labels.
     evidence = _compute_profile_evidence(parsed)
-    portrait_llm = _get_or_synthesize_portrait(user_id, n, evidence, blind_spots)
+    portrait_llm = _get_or_synthesize_portrait(user_id, n, evidence, blind_spots, parsed)
     llm_notes_by_signal = {s["signal_key"]: s for s in portrait_llm.get("signals", [])}
     llm_context_notes = {s["signal_key"]: s["note"] for s in portrait_llm.get("context_shifts", [])}
 
     steady_signals = []
     still_forming = []
     for signal_key, ev in evidence["overall"].items():
-        label = SIGNAL_EVIDENCE_CONFIG[signal_key]["label"]
+        cfg = SIGNAL_EVIDENCE_CONFIG[signal_key]
+        label = cfg["label"]
         if ev["is_steady"]:
             llm = llm_notes_by_signal.get(signal_key, {})
             steady_signals.append({
                 "signal_key": signal_key,
                 "label": label,
-                "mean": ev["mean"],
+                "kind": cfg["kind"],
+                # continuous dims populate mean/recent_mean/shift_pct; categorical
+                # dims (pacing_arc, energy_arc) populate mode_label/agreement_ratio
+                # instead — never both, since categorical has no numeric mean.
+                "mean": ev.get("mean"),
+                "mode_label": ev.get("mode_label"),
+                "agreement_ratio": ev.get("agreement_ratio"),
                 "sample_count": ev["sample_count"],
-                "recent_mean": ev["recent_mean"],
-                "shift_pct": ev["shift_pct"],
+                "recent_mean": ev.get("recent_mean"),
+                "shift_pct": ev.get("shift_pct"),
                 "framing": llm.get("framing", "observation"),
                 "note": llm.get("note", ""),
             })
@@ -1383,8 +1432,10 @@ def get_profile(user_id: str = Depends(get_current_user)):
 
     how_you_shift_by_context = []
     for signal_key, note in llm_context_notes.items():
+        is_categorical = SIGNAL_EVIDENCE_CONFIG[signal_key]["kind"] == "categorical"
+        value_field = "mode_label" if is_categorical else "mean"
         by_ctx = {
-            ctx: data[signal_key]["mean"]
+            ctx: data[signal_key][value_field]
             for ctx, data in evidence["by_context"].items()
             if data.get(signal_key, {}).get("is_steady")
         }
@@ -1428,31 +1479,30 @@ def _get_dismissed_card_keys(user_id: str) -> set:
 
 @app.get("/api/home")
 def get_home(user_id: str = Depends(get_current_user)):
+    """v2 Home feed: a single interleaved timeline of session recap cards
+    (unconditional, one per session) and dimension-maturation event cards
+    (fired by pipeline/trigger_detector.py at finalize time), sorted newest
+    first. No longer needs evidence/portrait computation at all — that LLM
+    call only runs when the You page itself is viewed, not on every Home load.
+    The "7 visible + 8th faded + Show more" behavior is a frontend rendering
+    contract over this complete, uncapped list — backend returns everything.
+    """
     parsed = _fetch_and_parse_sessions(user_id)
     if len(parsed) < 1:
         return {"insufficient_data": True, "session_count": 0, "cards": []}
 
-    n = len(parsed)
-    recorded_contexts = set(p["context"] for p in parsed)
-    blind_spots = _compute_blind_spots(recorded_contexts)
-
-    # Same evidence + portrait computation /api/profile uses — same cache key,
-    # so viewing Home and You back-to-back costs zero extra LLM calls either way.
-    evidence = _compute_profile_evidence(parsed)
-    portrait_llm = _get_or_synthesize_portrait(user_id, n, evidence, blind_spots)
-    llm_notes_by_signal = {s["signal_key"]: s for s in portrait_llm.get("signals", [])}
-
     dismissed = _get_dismissed_card_keys(user_id)
 
+    events_res = supabase_admin.table("dimension_events").select("*").eq(
+        "user_id", user_id
+    ).order("created_at", desc=True).execute()
+
     cards = []
-    cards += home_feed.build_strength_cards(evidence, llm_notes_by_signal, dismissed)
-    cards += home_feed.build_observation_cards(evidence, llm_notes_by_signal, dismissed)
-    cards += home_feed.build_how_it_may_land_cards(portrait_llm, dismissed)
-    cards += home_feed.build_progress_cards(evidence, dismissed)
-    cards += home_feed.build_still_forming_cards(evidence, dismissed)
-    cards += home_feed.build_session_observation_card(parsed, dismissed)
+    cards += home_feed.build_session_recap_cards(parsed, dismissed)
+    cards += home_feed.build_dimension_event_cards(events_res.data, dismissed)
+    cards.sort(key=lambda c: c["date"], reverse=True)
 
-    return {"insufficient_data": False, "session_count": n, "cards": cards}
+    return {"insufficient_data": False, "session_count": len(parsed), "cards": cards}
 
 
 @app.post("/api/home/dismiss")
@@ -1718,7 +1768,7 @@ def _get_context_evidence(user_id: str, context: str) -> dict:
             values = [extract_value(signal_key, sig) for sig in all_signals]
         except (KeyError, TypeError):
             values = []
-        evidence[signal_key] = compute_signal_evidence(signal_key, values)
+        evidence[signal_key] = compute_evidence(signal_key, values)
 
     return {"context": context, "session_count": len(context_sessions), "signals": evidence}
 

backend/pipeline/evidence_gate.py

@@ -1,62 +1,159 @@
 import numpy as np
+from collections import Counter
 
-# Per-signal (min_samples, cv_threshold) — see roadmap/product_decisions memory for
-# the reasoning behind each threshold. These are starting defaults with no real
+# Per-signal (min_samples, threshold) — see roadmap/product_decisions memory for
+# the reasoning behind each value. These are starting defaults with no real
 # user data yet; expect to tune once sessions accumulate.
+#
+# "kind" dispatches which evidence function applies:
+#   "continuous"  — a numeric rate/ratio, steadiness = coefficient of variation
+#                    (std/mean) at or below cv_threshold.
+#   "categorical" — a string label (e.g. "accelerating"/"stable"/"decelerating"),
+#                    steadiness = the most common label's share of the rolling
+#                    window at or above agreement_threshold. Used for the two
+#                    "arc" signals whose natural magnitude sits near zero for most
+#                    people, which would make cv unstable (division by ~0).
 SIGNAL_EVIDENCE_CONFIG = {
     "talk_ratio": {
-        "min_samples": 5,
+        "kind": "continuous",
+        "min_samples": 3,
         "cv_threshold": 0.20,
         "extract": lambda sig: sig["talk_ratio"]["user_ratio"],
         "label": "talk-share",
     },
-    "questions": {
+    "curiosity": {
+        "kind": "continuous",
         "min_samples": 5,
         "cv_threshold": 0.35,
-        "extract": lambda sig: sig["questions"]["user_questions_asked"],
+        "extract": lambda sig: sig["curiosity"]["question_turn_rate_per_100_words"],
         "label": "curiosity",
     },
-    "speech_rate": {
-        "min_samples": 5,
-        "cv_threshold": 0.15,
-        "extract": lambda sig: sig["speech_rate"]["overall_wpm"],
-        "label": "pace",
+    "turn_taking_assertiveness": {
+        "kind": "continuous",
+        "min_samples": 4,
+        "cv_threshold": 0.30,
+        "extract": lambda sig: sig["interruptions"]["user_interrupt_rate_per_10_transitions"],
+        "label": "turn-taking assertiveness",
     },
-    "response_latency": {
-        "min_samples": 5,
-        "cv_threshold": 0.40,
-        "extract": lambda sig: sig["pauses"]["response_latency"]["mean_s"],
-        "label": "pauses",
+    "conversational_drive": {
+        "kind": "continuous",
+        "min_samples": 6,
+        "cv_threshold": 0.35,
+        "extract": lambda sig: sig["drive_vs_follow"]["drive_score"],
+        "label": "conversational drive",
     },
     "hedging": {
+        "kind": "continuous",
         "min_samples": 5,
         "cv_threshold": 0.30,
         "extract": lambda sig: sig["hedging"]["rate_per_100_words"],
         "label": "hedging",
     },
     "directness": {
+        "kind": "continuous",
         "min_samples": 5,
         "cv_threshold": 0.30,
         "extract": lambda sig: sig["directness"]["rate_per_100_words"],
         "label": "directness",
     },
-    "question_impact": {
+    "building_on_others": {
+        "kind": "continuous",
+        "min_samples": 6,
+        "cv_threshold": 0.40,
+        "extract": lambda sig: sig["building_on_others"]["building_on_rate"],
+        "label": "building on others",
+    },
+    "pace": {
+        "kind": "continuous",
+        "min_samples": 3,
+        "cv_threshold": 0.15,
+        "extract": lambda sig: sig["speech_rate"]["overall_wpm"],
+        "label": "pace",
+    },
+    "pacing_arc": {
+        "kind": "categorical",
+        "min_samples": 5,
+        "agreement_threshold": 0.60,
+        "extract": lambda sig: sig["speech_acceleration"]["trend"],
+        "label": "pacing arc",
+    },
+    "vocal_expressiveness": {
+        "kind": "continuous",
+        "min_samples": 3,
+        "cv_threshold": 0.25,
+        "extract": lambda sig: sig["pitch_features"]["std_hz"],
+        "label": "vocal expressiveness",
+    },
+    "energy_arc": {
+        "kind": "categorical",
+        "min_samples": 5,
+        "agreement_threshold": 0.60,
+        "extract": lambda sig: sig["vocal_energy"]["trend"],
+        "label": "energy arc",
+    },
+    "turn_length": {
+        "kind": "continuous",
+        "min_samples": 4,
+        "cv_threshold": 0.25,
+        "extract": lambda sig: sig["monologue"]["avg_turn_length_s"],
+        "label": "turn length",
+    },
+    "vocabulary_richness": {
+        "kind": "continuous",
+        "min_samples": 5,
+        "cv_threshold": 0.20,
+        "extract": lambda sig: sig["vocabulary_richness"]["type_token_ratio"],
+        "label": "vocabulary richness",
+    },
+    "fillers": {
+        "kind": "continuous",
+        "min_samples": 5,
+        "cv_threshold": 0.35,
+        "extract": lambda sig: sig["filler_words"]["rate_per_100_words"],
+        "label": "fillers",
+    },
+    "response_latency": {
+        "kind": "continuous",
+        "min_samples": 5,
+        "cv_threshold": 0.40,
+        "extract": lambda sig: sig["pauses"]["response_latency"]["mean_s"],
+        "label": "pauses",
+    },
+}
+
+# Sub-signals: fold into a parent dimension's card as a bonus note rather than
+# getting their own dimension slot. Own evidence tracking (looser thresholds —
+# each depends more on the other person's behavior than the user's own), but
+# only ever fire "first_time_steady" — no drift/recurring/context_shift/anomaly
+# of their own. Kept out of SIGNAL_EVIDENCE_CONFIG so profile_strength_pct and
+# the You-page steady/still-forming lists stay scoped to the 15 real dimensions.
+SUB_SIGNAL_EVIDENCE_CONFIG = {
+    "question_pickup": {
+        "kind": "continuous",
         "min_samples": 6,
         "cv_threshold": 0.45,
         "extract": lambda sig: sig["question_impact"]["pickup_rate"],
         "label": "question follow-through",
+        "parent": "curiosity",
+        "allowed_triggers": ["first_time_steady"],
     },
-    "drive_vs_follow": {
+    "gets_interrupted": {
+        "kind": "continuous",
         "min_samples": 6,
-        "cv_threshold": 0.35,
-        "extract": lambda sig: sig["drive_vs_follow"]["drive_score"],
-        "label": "conversational drive",
+        "cv_threshold": 0.40,
+        "extract": lambda sig: sig["interruptions"]["user_was_interrupted_rate_per_10_transitions"],
+        "label": "gets interrupted",
+        "parent": "turn_taking_assertiveness",
+        "allowed_triggers": ["first_time_steady"],
     },
-    "building_on_others": {
+    "long_turn_rate": {
+        "kind": "continuous",
         "min_samples": 6,
         "cv_threshold": 0.40,
-        "extract": lambda sig: sig["building_on_others"]["building_on_rate"],
-        "label": "building on others",
+        "extract": lambda sig: sig["monologue"]["long_turn_rate"],
+        "label": "long speaking stretches",
+        "parent": "turn_length",
+        "allowed_triggers": ["first_time_steady"],
     },
 }
 
@@ -65,15 +162,23 @@ SIGNAL_EVIDENCE_CONFIG = {
 ROLLING_WINDOW = 10
 
 
-def compute_signal_evidence(signal_key: str, historical_values: list) -> dict:
-    """Given all past values of one signal (oldest→newest) for a user+context,
-    return an evidence summary: sample_count, mean, cv, is_steady.
+def _cfg_for(signal_key: str) -> dict:
+    """Look a key up across both the main dimension config and the sub-signal
+    config, so callers don't need to know which dict a given key lives in."""
+    if signal_key in SIGNAL_EVIDENCE_CONFIG:
+        return SIGNAL_EVIDENCE_CONFIG[signal_key]
+    return SUB_SIGNAL_EVIDENCE_CONFIG[signal_key]
+
+
+def compute_signal_evidence(signal_key: str, historical_values: list, config: dict = None) -> dict:
+    """Given all past values of one CONTINUOUS signal (oldest→newest) for a
+    user+context, return an evidence summary: sample_count, mean, cv, is_steady.
 
     Pure function over already-extracted values — doesn't fetch data itself,
     so it's independently testable regardless of where the values came from.
     """
-    cfg = SIGNAL_EVIDENCE_CONFIG[signal_key]
-    # Some signals (e.g. question_impact) legitimately have no value for a session
+    cfg = (config or SIGNAL_EVIDENCE_CONFIG).get(signal_key) or SUB_SIGNAL_EVIDENCE_CONFIG.get(signal_key)
+    # Some signals (e.g. question_pickup) legitimately have no value for a session
     # (zero questions asked) — None, not a fake 0, so drop it rather than let it
     # pollute the mean/cv or crash np.mean.
     historical_values = [v for v in historical_values if v is not None]
@@ -100,7 +205,48 @@ def compute_signal_evidence(signal_key: str, historical_values: list) -> dict:
     return result
 
 
+def compute_categorical_evidence(signal_key: str, historical_labels: list, config: dict = None) -> dict:
+    """Given all past CATEGORICAL labels (oldest→newest) for a user+context,
+    return an evidence summary based on majority agreement within the rolling
+    window, not variance — see the "kind" docstring above `SIGNAL_EVIDENCE_CONFIG`
+    for why (a magnitude-based cv is unstable for signals whose natural mean
+    sits near zero, e.g. pacing/energy arcs for people with no strong drift).
+    """
+    cfg = (config or SIGNAL_EVIDENCE_CONFIG).get(signal_key) or SUB_SIGNAL_EVIDENCE_CONFIG.get(signal_key)
+    labels = [v for v in historical_labels if v and v != "insufficient_data"]
+    n = len(labels)
+    result = {
+        "signal": signal_key,
+        "sample_count": n,
+        "min_samples_required": cfg["min_samples"],
+        "is_steady": False,
+        "mode_label": None,
+        "agreement_ratio": None,
+    }
+    if n < cfg["min_samples"]:
+        return result
+
+    window = labels[-ROLLING_WINDOW:]
+    mode_label, mode_count = Counter(window).most_common(1)[0]
+    agreement = mode_count / len(window)
+    result["mode_label"] = mode_label
+    result["agreement_ratio"] = round(agreement, 3)
+    result["is_steady"] = agreement >= cfg["agreement_threshold"]
+    return result
+
+
+def compute_evidence(signal_key: str, historical_values: list, config: dict = None) -> dict:
+    """Dispatches to the right evidence function based on the signal's "kind".
+    Every caller should go through this rather than calling either function
+    directly, so categorical dimensions plug in transparently everywhere."""
+    cfg = _cfg_for(signal_key) if config is None else config.get(signal_key, _cfg_for(signal_key))
+    if cfg["kind"] == "categorical":
+        return compute_categorical_evidence(signal_key, historical_values, config)
+    return compute_signal_evidence(signal_key, historical_values, config)
+
+
 def extract_value(signal_key: str, signals: dict):
-    """Pull this signal's tracked scalar out of a full `signals` dict (as stored
-    in sessions.signals_json)."""
-    return SIGNAL_EVIDENCE_CONFIG[signal_key]["extract"](signals)
+    """Pull this signal's tracked scalar/label out of a full `signals` dict (as
+    stored in sessions.signals_json). Checks both the main dimension config and
+    the sub-signal config."""
+    return _cfg_for(signal_key)["extract"](signals)

backend/pipeline/home_feed.py

@@ -1,136 +1,76 @@
-"""Home feed card builders — replaces the old mirror_feed_synthesizer.py LLM
-free-form feed. Every function here is a pure function over already-computed
-data (evidence, portrait LLM notes, parsed sessions, dismissed keys) — zero new
-LLM calls except the how_it_may_land note, which reuses portrait_synthesizer's
-existing single LLM round-trip (see PortraitSynthesizer.synthesize).
+"""Home feed card builders — v2: a single interleaved timeline of two card
+sources, replacing the earlier v1 (5 static per-signal card types built fresh
+from evidence on every read, no event history, no triggers).
 
-Card types, per CLAUDE.md's Home = "stream of insight cards" description:
-strength, observation (growth_area + observation framings unified), progress,
-still_forming, and a zero-LLM session-level observation card. Every card is
-self-relative, carries its evidence, and has a stable `card_key` for dismissal.
+  - Session recap cards: one unconditional card per session, dated to that
+    session, built straight from insight_generator's existing output — zero
+    new LLM cost.
+  - Dimension event cards: one card per row in `dimension_events` (written by
+    pipeline/trigger_detector.py at finalize time) — frozen content, dated to
+    the session that triggered them, so they interleave genuinely by date
+    alongside recap cards rather than always sorting as "now".
+
+Both card types share the existing `dismissed_cards` dismissal mechanism.
+`build_how_it_may_land_cards` is kept (not currently called from /api/home)
+pending its future home on the You page's "how it may land" section.
 """
 
 from pipeline.evidence_gate import SIGNAL_EVIDENCE_CONFIG
-from pipeline.portrait_synthesizer import _format_mean
-
-# Shift magnitude below which a change isn't worth surfacing as a progress
-# card — same threshold the old /api/trends widget used (main.py's _trend()).
-_PROGRESS_SHIFT_THRESHOLD_PCT = 15
-# How close to evidence-steady a signal needs to be to show up as "still forming"
-# rather than being silently omitted — avoids listing every not-yet-steady signal.
-_STILL_FORMING_PROXIMITY = 3
-_STILL_FORMING_CAP = 2
 
 
-def build_strength_cards(evidence: dict, llm_notes_by_signal: dict, dismissed: set) -> list:
+def build_session_recap_cards(parsed: list, dismissed: set) -> list:
+    """One card per session, unconditional. Uses the FULL insight_generator
+    output (conversation_summary, observations, coaching_suggestions,
+    notable_pattern) rather than just the single top observation the old v1
+    session_observation card showed."""
     cards = []
-    for signal_key, ev in evidence.get("overall", {}).items():
-        if not ev.get("is_steady"):
-            continue
-        llm = llm_notes_by_signal.get(signal_key, {})
-        if llm.get("framing") != "strength":
-            continue
-        card_key = f"strength:{signal_key}"
+    for p in parsed:
+        card_key = f"session_recap:{p['id']}"
         if card_key in dismissed:
             continue
+        ins = p.get("ins") or {}
         cards.append({
-            "type": "strength",
+            "type": "session_recap",
             "card_key": card_key,
-            "signal_key": signal_key,
-            "label": SIGNAL_EVIDENCE_CONFIG[signal_key]["label"],
-            "note": llm.get("note", ""),
-            "mean": ev["mean"],
-            "sample_count": ev["sample_count"],
+            "session_id": p["id"],
+            "context": p["context"],
+            "date": p["date"],
+            "conversation_summary": ins.get("conversation_summary", ""),
+            "observations": ins.get("observations", []),
+            "coaching_suggestions": ins.get("coaching_suggestions", []),
+            "notable_pattern": ins.get("notable_pattern"),
         })
     return cards
 
 
-def build_observation_cards(evidence: dict, llm_notes_by_signal: dict, dismissed: set) -> list:
-    """growth_area and observation framings share one card type and one
-    dismissal key — dismissing commentary on a signal should hold even if the
-    portrait LLM later flips which of the two framings it uses for it."""
+def build_dimension_event_cards(events: list, dismissed: set) -> list:
+    """`events` = rows fetched from the `dimension_events` table by main.py.
+    Content is frozen at fire-time (card_copy_json) — never regenerated here,
+    even if the underlying evidence has since shifted again (a new event
+    fires separately instead)."""
+    import json
+
     cards = []
-    for signal_key, ev in evidence.get("overall", {}).items():
-        if not ev.get("is_steady"):
-            continue
-        llm = llm_notes_by_signal.get(signal_key, {})
-        if llm.get("framing") not in ("growth_area", "observation"):
-            continue
-        card_key = f"observation:{signal_key}"
+    for e in events:
+        card_key = f"dimension_event:{e['id']}"
         if card_key in dismissed:
             continue
+        try:
+            copy = json.loads(e["card_copy_json"])
+        except (KeyError, TypeError, ValueError):
+            copy = {}
         cards.append({
-            "type": "observation",
-            "card_key": card_key,
-            "signal_key": signal_key,
-            "label": SIGNAL_EVIDENCE_CONFIG[signal_key]["label"],
-            "framing": llm.get("framing"),
-            "note": llm.get("note", ""),
-            "mean": ev["mean"],
-            "sample_count": ev["sample_count"],
-        })
-    return cards
-
-
-def build_still_forming_cards(evidence: dict, dismissed: set) -> list:
-    candidates = []
-    for signal_key, ev in evidence.get("overall", {}).items():
-        if ev.get("is_steady"):
-            continue
-        remaining = ev["min_samples_required"] - ev["sample_count"]
-        # remaining <= 0 means the sample floor is already met but the signal is
-        # still too variable (high CV) to call steady — that's not "forming",
-        # more sessions alone won't fix it. Say nothing rather than mislabel it
-        # as approaching-steady forever (CLAUDE.md rule #7: silence is allowed).
-        if remaining <= 0 or remaining > _STILL_FORMING_PROXIMITY:
-            continue
-        card_key = f"still_forming:{signal_key}"
-        if card_key in dismissed:
-            continue
-        candidates.append({
-            "type": "still_forming",
+            "type": "dimension_event",
             "card_key": card_key,
-            "signal_key": signal_key,
-            "label": SIGNAL_EVIDENCE_CONFIG[signal_key]["label"],
-            "sample_count": ev["sample_count"],
-            "min_needed": ev["min_samples_required"],
-            "_remaining": remaining,
+            "dimension_key": e["dimension_key"],
+            "scope": e["scope"],
+            "trigger_type": e["trigger_type"],
+            "direction": e.get("direction"),
+            "session_id": e["session_id"],
+            "date": e["created_at"],
+            "label": copy.get("label", SIGNAL_EVIDENCE_CONFIG.get(e["dimension_key"], {}).get("label", e["dimension_key"])),
+            "note": copy.get("note", ""),
         })
-    candidates.sort(key=lambda c: c["_remaining"])
-    for c in candidates:
-        del c["_remaining"]
-    return candidates[:_STILL_FORMING_CAP]
-
-
-def build_progress_cards(evidence: dict, dismissed: set) -> list:
-    """Scoped to by_context evidence only — recent-vs-established shift is
-    only meaningful within a single context (see _compute_profile_evidence's
-    compute_shift docstring for the cross-context contamination this avoids)."""
-    cards = []
-    for context, signals in evidence.get("by_context", {}).items():
-        for signal_key, ev in signals.items():
-            if not ev.get("is_steady") or ev.get("shift_pct") is None:
-                continue
-            if abs(ev["shift_pct"]) < _PROGRESS_SHIFT_THRESHOLD_PCT:
-                continue
-            direction = "up" if ev["shift_pct"] > 0 else "down"
-            card_key = f"progress:{signal_key}:{direction}"
-            if card_key in dismissed:
-                continue
-            label = SIGNAL_EVIDENCE_CONFIG[signal_key]["label"]
-            cards.append({
-                "type": "progress",
-                "card_key": card_key,
-                "signal_key": signal_key,
-                "label": label,
-                "context": context,
-                "direction": direction,
-                "note": (
-                    f"Your {label} has shifted from {_format_mean(signal_key, ev['mean'])} "
-                    f"to {_format_mean(signal_key, ev['recent_mean'])} over your last few "
-                    f"{context.replace('_', ' ')} sessions."
-                ),
-            })
     return cards
 
 
@@ -151,30 +91,3 @@ def build_how_it_may_land_cards(portrait_llm: dict, dismissed: set) -> list:
             "note": entry.get("note", ""),
         })
     return cards
-
-
-def build_session_observation_card(parsed: list, dismissed: set) -> list:
-    """Zero-LLM — surfaces the most recent session's top observation, already
-    generated by insight_generator.py. Self-expires: a new session produces a
-    new session_id, so this card never needs its own dismissal cleanup."""
-    if not parsed:
-        return []
-    latest = parsed[-1]
-    observations = latest["ins"].get("observations", [])
-    if not observations:
-        return []
-    obs = observations[0]
-    signal = obs.get("signal", "")
-    card_key = f"session_observation:{latest['id']}:{signal}"
-    if card_key in dismissed:
-        return []
-    return [{
-        "type": "session_observation",
-        "card_key": card_key,
-        "session_id": latest["id"],
-        "context": latest["context"],
-        "date": latest["date"],
-        "signal": signal,
-        "note": obs.get("observation", ""),
-        "resonance_prompt": obs.get("resonance_prompt", ""),
-    }]

backend/pipeline/insight_generator.py

@@ -1,7 +1,7 @@
 import json
 from anthropic import Anthropic
 
-from pipeline.evidence_gate import SIGNAL_EVIDENCE_CONFIG
+from pipeline.evidence_gate import SIGNAL_EVIDENCE_CONFIG, extract_value
 from pipeline.llm_utils import extract_text
 
 
@@ -98,22 +98,28 @@ class InsightGenerator:
             # in which case it's listed as not-yet-steady so the prompt can explicitly
             # instruct against inventing a pattern for it. Self-relative only (rule #4) —
             # no population comparison of any kind.
-            signal_current_value = {
-                "talk_ratio": prepared["talk_ratio_user"],
-                "questions": prepared["user_questions_asked"],
-                "speech_rate": prepared["speech_rate_wpm"],
-                "response_latency": prepared["avg_response_latency_s"],
-                "hedging": prepared["hedging_rate"],
-                "directness": prepared["directness_rate"],
-                "question_impact": prepared["question_pickup_rate"],
-                "drive_vs_follow": prepared["drive_score"],
-                "building_on_others": prepared["building_on_rate"],
-            }
+            #
+            # Uses extract_value (the same extraction logic the evidence system itself
+            # uses) rather than a hand-maintained key mapping — a parallel mapping here
+            # would silently go stale every time a dimension is renamed or added, which
+            # is exactly what happened to the old 9-key version of this block.
             steady = {}
             not_yet_steady = []
             for sig_key, sig_evidence in evidence.get("signals", {}).items():
-                current = signal_current_value.get(sig_key)
-                if sig_evidence["is_steady"] and current is not None:
+                try:
+                    current = extract_value(sig_key, signals)
+                except (KeyError, TypeError):
+                    current = None
+                is_categorical = SIGNAL_EVIDENCE_CONFIG.get(sig_key, {}).get("kind") == "categorical"
+
+                if sig_evidence["is_steady"] and current is not None and is_categorical:
+                    steady[sig_key] = {
+                        "current_label": current,
+                        "your_usual_label": sig_evidence["mode_label"],
+                        "agreement_ratio": sig_evidence["agreement_ratio"],
+                        "sample_count": sig_evidence["sample_count"],
+                    }
+                elif sig_evidence["is_steady"] and current is not None:
                     delta = current - sig_evidence["mean"]
                     delta_pct = (delta / sig_evidence["mean"] * 100) if sig_evidence["mean"] else None
                     steady[sig_key] = {
@@ -181,6 +187,16 @@ CONVERSATION TRANSCRIPT:
             lines = []
             for sig_key, c in ev["steady"].items():
                 label = SIGNAL_EVIDENCE_CONFIG[sig_key]["label"]
+                if "current_label" in c:
+                    # Categorical dim (pacing_arc, energy_arc) — no numeric delta,
+                    # just whether this session's label matches the established mode.
+                    same = c["current_label"] == c["your_usual_label"]
+                    lines.append(
+                        f"  {label}: {c['current_label']} this session "
+                        f"({'matches' if same else 'differs from'} your usual {c['your_usual_label']}, "
+                        f"based on {c['sample_count']} past {ctx_label} sessions)"
+                    )
+                    continue
                 if c["delta"] > 0:
                     arrow = "more than"
                 elif c["delta"] < 0:

backend/pipeline/portrait_synthesizer.py

@@ -5,23 +5,33 @@ from pipeline.evidence_gate import SIGNAL_EVIDENCE_CONFIG
 from pipeline.llm_utils import extract_text
 
 # How to format each signal's raw mean into human-readable text for the prompt.
-# (value * scale) formatted with `fmt`, followed by `unit`.
+# (value * scale) formatted with `fmt`, followed by `unit`. Categorical signals
+# (pacing_arc, energy_arc) aren't listed here — they have no numeric mean to
+# scale, see the branch in _format_mean below.
 _SIGNAL_FORMAT = {
-    "talk_ratio":         (100, "{:.0f}", "% of speaking time"),
-    "questions":          (1,   "{:.1f}", " questions asked per session"),
-    "speech_rate":        (1,   "{:.0f}", " words per minute"),
-    "response_latency":   (1,   "{:.1f}", "s before responding"),
-    "hedging":            (1,   "{:.1f}", " hedging phrases per 100 words"),
-    "directness":         (1,   "{:.1f}", " direct/assertive phrases per 100 words"),
-    "question_impact":    (100, "{:.0f}", "% of your questions picked up by the room"),
-    "drive_vs_follow":    (100, "{:.0f}", "% drive score (higher = initiates more, lower = follows more)"),
-    "building_on_others": (100, "{:.0f}", "% of your turns build on someone else's point"),
+    "talk_ratio":               (100, "{:.0f}", "% of speaking time"),
+    "curiosity":                (1,   "{:.2f}", " question-turns per 100 words"),
+    "turn_taking_assertiveness": (1,  "{:.1f}", " interruptions per 10 speaker changes"),
+    "conversational_drive":     (100, "{:.0f}", "% drive score (higher = initiates more, lower = follows more)"),
+    "hedging":                  (1,   "{:.1f}", " hedging phrases per 100 words"),
+    "directness":               (1,   "{:.1f}", " direct/assertive phrases per 100 words"),
+    "building_on_others":       (100, "{:.0f}", "% of your turns build on someone else's point"),
+    "pace":                     (1,   "{:.0f}", " words per minute"),
+    "vocal_expressiveness":     (1,   "{:.1f}", " Hz of pitch variation"),
+    "turn_length":              (1,   "{:.1f}", "s per turn on average"),
+    "vocabulary_richness":      (100, "{:.0f}", "% unique words in a typical stretch of speech"),
+    "fillers":                  (1,   "{:.2f}", " filler words per 100 words"),
+    "response_latency":         (1,   "{:.1f}", "s before responding"),
 }
 
 
-def _format_mean(signal_key: str, mean: float) -> str:
+def _format_mean(signal_key: str, value) -> str:
+    """Formats a continuous signal's numeric mean, or a categorical signal's
+    string mode label, into human-readable text for the prompt."""
+    if SIGNAL_EVIDENCE_CONFIG.get(signal_key, {}).get("kind") == "categorical":
+        return f"consistently {value}"
     scale, fmt, unit = _SIGNAL_FORMAT[signal_key]
-    return fmt.format(mean * scale) + unit
+    return fmt.format(value * scale) + unit
 
 
 class PortraitSynthesizer:
@@ -35,7 +45,8 @@ class PortraitSynthesizer:
     def __init__(self, api_key: str):
         self.client = Anthropic(api_key=api_key)
 
-    def synthesize(self, evidence: dict, blind_spots: list, session_count: int) -> dict:
+    def synthesize(self, evidence: dict, blind_spots: list, session_count: int,
+                    sub_evidence: dict = None) -> dict:
         overall = evidence.get("overall", {})
         by_context = evidence.get("by_context", {})
 
@@ -48,8 +59,10 @@ class PortraitSynthesizer:
         # their own contexts, never against other people.
         context_shift_candidates = {}
         for signal_key in SIGNAL_EVIDENCE_CONFIG:
+            is_categorical = SIGNAL_EVIDENCE_CONFIG[signal_key]["kind"] == "categorical"
+            value_field = "mode_label" if is_categorical else "mean"
             steady_contexts = {
-                ctx: data[signal_key]["mean"]
+                ctx: data[signal_key][value_field]
                 for ctx, data in by_context.items()
                 if data.get(signal_key, {}).get("is_steady")
             }
@@ -59,7 +72,8 @@ class PortraitSynthesizer:
         if not steady_overall and not context_shift_candidates:
             return {"signals": [], "context_shifts": [], "how_it_may_land": []}
 
-        prompt = self._build_prompt(steady_overall, context_shift_candidates, session_count)
+        prompt = self._build_prompt(steady_overall, context_shift_candidates, session_count,
+                                     sub_evidence or {})
         response = self.client.messages.create(
             model="claude-sonnet-5",
             max_tokens=1800,
@@ -69,27 +83,32 @@ class PortraitSynthesizer:
         return self._parse(extract_text(response))
 
     def _build_prompt(self, steady_overall: dict, context_shift_candidates: dict,
-                      session_count: int) -> str:
+                      session_count: int, sub_evidence: dict) -> str:
         signal_lines = []
         for signal_key, ev in steady_overall.items():
             label = SIGNAL_EVIDENCE_CONFIG[signal_key]["label"]
-            desc = f"  {label} ({signal_key}): established at {_format_mean(signal_key, ev['mean'])}, " \
+            is_categorical = SIGNAL_EVIDENCE_CONFIG[signal_key]["kind"] == "categorical"
+            value = ev["mode_label"] if is_categorical else ev["mean"]
+            desc = f"  {label} ({signal_key}): established at {_format_mean(signal_key, value)}, " \
                    f"based on {ev['sample_count']} sessions"
             signal_lines.append(desc)
 
-        wants_how_it_may_land = "question_impact" in steady_overall
+        # question_pickup is a sub-signal (folded into "curiosity" on Home, see
+        # evidence_gate.SUB_SIGNAL_EVIDENCE_CONFIG) — not part of steady_overall,
+        # so its own evidence dict is passed in separately.
+        wants_how_it_may_land = sub_evidence.get("question_pickup", {}).get("is_steady", False)
 
         context_lines = []
         for signal_key, by_ctx in context_shift_candidates.items():
             label = SIGNAL_EVIDENCE_CONFIG[signal_key]["label"]
             parts = ", ".join(
-                f"{ctx.replace('_', ' ')}: {_format_mean(signal_key, mean)}"
-                for ctx, mean in by_ctx.items()
+                f"{ctx.replace('_', ' ')}: {_format_mean(signal_key, value)}"
+                for ctx, value in by_ctx.items()
             )
             context_lines.append(f"  {label} ({signal_key}) — {parts}")
 
         how_it_may_land_task = (
-            "\nALSO: \"question follow-through\" (question_impact) is established. Write ONE "
+            "\nALSO: \"question follow-through\" is established. Write ONE "
             "additional sentence describing how this person's questions tend to LAND in the room — "
             "effect-on-others phrasing, e.g. \"your questions tend to get picked up and built on by "
             "the room\" — still self-relative (this person's own tendency), never a claim about what "
@@ -98,7 +117,7 @@ class PortraitSynthesizer:
         )
         how_it_may_land_schema = (
             ',\n  "how_it_may_land": [\n'
-            '    {"signal_key": "question_impact", "note": "one sentence"}\n  ]'
+            '    {"signal_key": "curiosity", "note": "one sentence"}\n  ]'
             if wants_how_it_may_land else ""
         )
 

backend/pipeline/signal_extractor.py

@@ -106,13 +106,18 @@ class SignalExtractor:
 
         turn_dynamics = self._compute_turn_dynamics(user_segs, other_segs)
         questions = self._compute_questions(user_segs, other_segs)
+        # Room-wide (all_other_segs), not dyadic — otherwise this collapses a
+        # multi-party meeting to one arbitrary "other" (CLAUDE.md forbids this).
+        # Computed before drive_vs_follow so its interruption_asymmetry input
+        # can reuse this result instead of computing interruptions twice.
+        interruptions = self._compute_interruptions(user_segs, all_other_segs)
 
         signals = {
             "session_duration_s": self._get_session_duration(),
             "talk_ratio": self._compute_talk_ratio(user_segs, all_other_segs),
             "speech_rate": self._compute_speech_rate(user_segs),
             "pauses": self._compute_pauses(user_segs, other_segs),
-            "interruptions": self._compute_interruptions(user_segs, other_segs),
+            "interruptions": interruptions,
             "filler_words": self._compute_filler_words(user_segs),
             "turn_dynamics": turn_dynamics,
             "pitch_features": self._compute_pitch_features(user_segs),
@@ -120,6 +125,7 @@ class SignalExtractor:
             "vocal_energy": self._compute_vocal_energy(user_segs),
             "speech_acceleration": self._compute_speech_acceleration(user_segs),
             "questions": questions,
+            "curiosity": self._compute_curiosity(user_segs, all_other_segs),
             "monologue": self._compute_monologue(user_segs),
             "vocabulary_richness": self._compute_vocabulary_richness(user_segs),
             "silence_ratio": self._compute_silence_ratio(user_segs, all_other_segs),
@@ -129,7 +135,7 @@ class SignalExtractor:
             "directness": self._compute_directness(user_segs),
             "question_impact": self._compute_question_impact(user_segs, all_other_segs),
             "drive_vs_follow": self._compute_drive_vs_follow(
-                user_segs, other_segs, all_other_segs, turn_dynamics, questions
+                user_segs, other_segs, all_other_segs, turn_dynamics, questions, interruptions
             ),
             "building_on_others": self._compute_building_on_others(user_segs, all_other_segs),
         }
@@ -229,10 +235,14 @@ class SignalExtractor:
         return sorted(turns, key=lambda x: x["start"])
 
     def _compute_interruptions(self, user_segs, other_segs) -> dict:
+        """`other_segs` should be the room-wide `all_other_segs` (not just the
+        primary dyadic partner) — otherwise this silently collapses a
+        multi-party meeting to one arbitrary "other", which CLAUDE.md forbids."""
         all_turns = self._build_all_turns(user_segs, other_segs)
 
         user_interrupts_other = 0
         other_interrupts_user = 0
+        transitions = 0
 
         for i in range(1, len(all_turns)):
             prev = all_turns[i - 1]
@@ -240,6 +250,7 @@ class SignalExtractor:
 
             if prev["speaker"] == curr["speaker"]:
                 continue
+            transitions += 1
 
             gap = curr["start"] - prev["end"]
             prev_duration = prev["end"] - prev["start"]
@@ -252,7 +263,10 @@ class SignalExtractor:
 
         return {
             "user_interrupted_other": user_interrupts_other,
-            "user_was_interrupted": other_interrupts_user
+            "user_was_interrupted": other_interrupts_user,
+            "total_transitions": transitions,
+            "user_interrupt_rate_per_10_transitions": round(user_interrupts_other / transitions * 10, 2) if transitions else None,
+            "user_was_interrupted_rate_per_10_transitions": round(other_interrupts_user / transitions * 10, 2) if transitions else None,
         }
 
     def _compute_hedging(self, user_segs) -> dict:
@@ -291,6 +305,42 @@ class SignalExtractor:
             "breakdown": marker_counts
         }
 
+    @staticmethod
+    def _is_question_turn(text: str) -> bool:
+        """A turn counts as a question if it contains '?' or a question-word
+        (English or Hindi). Shared by _compute_question_impact and
+        _compute_curiosity so both count "a question" the same way."""
+        text = text.lower()
+        return "?" in text or any(
+            re.search(r'\b' + w + r'\b', text) for w in SignalExtractor.QUESTION_WORDS
+        )
+
+    def _compute_curiosity(self, user_segs, all_other_segs) -> dict:
+        """Rate of the user's turns that are questions, per 100 words spoken —
+        a rate rather than _compute_questions' raw "?"-count, so it's comparable
+        across sessions of different length. Uses the same per-turn question
+        test as _compute_question_impact for consistency between the two."""
+        all_turns = self._build_all_turns(user_segs, all_other_segs)
+
+        user_turns = 0
+        question_turns = 0
+        for turn in all_turns:
+            if turn["speaker"] != self.user_speaker:
+                continue
+            user_turns += 1
+            if self._is_question_turn(turn["text"]):
+                question_turns += 1
+
+        total_words = sum(len(s.get("words", [])) for s in user_segs)
+        rate = round(question_turns / total_words * 100, 2) if total_words > 0 else None
+
+        return {
+            "question_turn_count": question_turns,
+            "user_turns": user_turns,
+            "total_words": total_words,
+            "question_turn_rate_per_100_words": rate,
+        }
+
     def _compute_question_impact(self, user_segs, all_other_segs) -> dict:
         """Did the room pick up the user's questions? Uses the room-wide turn
         sequence (any participant, not just the primary dyadic partner)."""
@@ -303,11 +353,7 @@ class SignalExtractor:
         for i, turn in enumerate(all_turns):
             if turn["speaker"] != self.user_speaker:
                 continue
-            text = turn["text"].lower()
-            is_question = "?" in text or any(
-                re.search(r'\b' + w + r'\b', text) for w in self.QUESTION_WORDS
-            )
-            if not is_question:
+            if not self._is_question_turn(turn["text"]):
                 continue
             total_questions += 1
 
@@ -334,11 +380,15 @@ class SignalExtractor:
             "avg_pickup_latency_s": round(float(np.mean(pickup_latencies)), 3) if pickup_latencies else None,
         }
 
-    def _compute_drive_vs_follow(self, user_segs, other_segs, all_other_segs, turn_dynamics, questions) -> dict:
-        """Composite of three already-computed signals — a single proxy (e.g. just
-        "turns not preceded by a question") is too thin on its own; this reuses the
-        same inputs dimension_scorer.py's _analyze_driver already treats as a
-        reasonable way to operationalize "who drove" in this codebase."""
+    def _compute_drive_vs_follow(self, user_segs, other_segs, all_other_segs, turn_dynamics,
+                                  questions, interruptions) -> dict:
+        """Composite of four already-computed signals — a single proxy (e.g. just
+        "turns not preceded by a question") is too thin on its own; three of these
+        inputs match dimension_scorer.py's _analyze_driver, and interruption_asymmetry
+        is added as a 4th, more direct signal of assertiveness (pure timing, not an
+        inferred proxy). Weighted equally (0.25 each) — the original 0.4/0.3/0.3
+        weights were never empirically validated, so there's no principled reason to
+        keep favoring one input now that a 4th is added."""
         all_turns = self._build_all_turns(user_segs, all_other_segs)
 
         user_turns_total = 0
@@ -366,8 +416,14 @@ class SignalExtractor:
         total_q = user_q + other_q
         question_asymmetry = round(user_q / total_q, 3) if total_q > 0 else 0.5
 
+        ui = interruptions.get("user_interrupted_other", 0)
+        uwi = interruptions.get("user_was_interrupted", 0)
+        total_i = ui + uwi
+        interruption_asymmetry = round(ui / total_i, 3) if total_i > 0 else 0.5
+
         drive_score = round(
-            0.4 * initiation_fraction + 0.3 * turn_length_asymmetry + 0.3 * question_asymmetry, 3
+            0.25 * initiation_fraction + 0.25 * turn_length_asymmetry +
+            0.25 * question_asymmetry + 0.25 * interruption_asymmetry, 3
         )
 
         return {
@@ -375,6 +431,7 @@ class SignalExtractor:
             "initiation_fraction": initiation_fraction,
             "turn_length_asymmetry": turn_length_asymmetry,
             "question_asymmetry": question_asymmetry,
+            "interruption_asymmetry": interruption_asymmetry,
         }
 
     def _compute_building_on_others(self, user_segs, all_other_segs) -> dict:
@@ -384,6 +441,8 @@ class SignalExtractor:
         all_turns = self._build_all_turns(user_segs, all_other_segs)
 
         total_user_turns = 0
+        eligible_turns = 0  # user turns immediately following an other-speaker turn —
+                            # the only turns where the lexical-overlap check can fire
         marker_matches = 0
         overlap_matches = 0
         building_on_count = 0
@@ -393,14 +452,17 @@ class SignalExtractor:
                 continue
             total_user_turns += 1
             text = turn["text"].lower()
+            prev = all_turns[i - 1] if i > 0 else None
+            is_eligible = bool(prev and prev["speaker"] != self.user_speaker)
+            if is_eligible:
+                eligible_turns += 1
 
             has_marker = any(text.startswith(m) for m in self.BUILDING_ON_MARKERS)
             if has_marker:
                 marker_matches += 1
 
             has_overlap = False
-            prev = all_turns[i - 1] if i > 0 else None
-            if prev and prev["speaker"] != self.user_speaker:
+            if is_eligible:
                 prev_words = {
                     w for w in re.findall(r"[a-z']+", prev["text"].lower())
                     if len(w) > 3 and w not in self.STOPWORDS
@@ -417,10 +479,15 @@ class SignalExtractor:
                 building_on_count += 1
 
         return {
-            "building_on_rate": round(building_on_count / total_user_turns, 3) if total_user_turns > 0 else 0,
+            # Denominator is eligible_turns (turns that could structurally show
+            # overlap), not total_user_turns — turns following your own prior
+            # turn can only ever match via has_marker, so counting them in the
+            # denominator quietly diluted the rate.
+            "building_on_rate": round(building_on_count / eligible_turns, 3) if eligible_turns > 0 else 0,
             "marker_matches": marker_matches,
             "lexical_overlap_matches": overlap_matches,
             "total_user_turns": total_user_turns,
+            "eligible_turns": eligible_turns,
             "interpretation_note": "approximate proxy — marker phrases + lexical overlap, not semantic reasoning",
         }
 
@@ -597,29 +664,38 @@ class SignalExtractor:
         """Detect long uninterrupted speaking stretches."""
         long_turns = [s for s in user_segs if (s["end"] - s["start"]) > 30]
         all_lengths = [s["end"] - s["start"] for s in user_segs]
+        total_turns = len(user_segs)
 
         return {
             "long_turn_count": len(long_turns),  # turns > 30 seconds
             "longest_turn_s": round(max(all_lengths), 1) if all_lengths else 0,
-            "avg_turn_length_s": round(float(np.mean(all_lengths)), 1) if all_lengths else 0
+            "avg_turn_length_s": round(float(np.mean(all_lengths)), 1) if all_lengths else 0,
+            "long_turn_rate": round(len(long_turns) / total_turns, 3) if total_turns > 0 else None,
         }
 
+    # First N words only \u2014 type-token ratio mechanically decreases as text gets
+    # longer (common words get reused more), which would confound a whole-session
+    # TTR with session length rather than actual vocabulary behavior. A fixed
+    # window makes every session's TTR comparable regardless of how long it ran.
+    VOCAB_WINDOW_WORDS = 150
+
     def _compute_vocabulary_richness(self, user_segs) -> dict:
-        """Type-token ratio: unique words / total words. Higher = richer vocabulary."""
+        """Type-token ratio over a fixed-size window: unique words / total words
+        in the first VOCAB_WINDOW_WORDS words spoken. Higher = richer vocabulary."""
         all_text = " ".join(s.get("text", "").lower() for s in user_segs)
         words = re.findall(r'\b[a-zA-Z\u0900-\u097F]+\b', all_text)
 
-        if len(words) < 10:
+        if len(words) < self.VOCAB_WINDOW_WORDS:
             return {"type_token_ratio": None, "unique_words": 0, "total_words": len(words)}
 
-        unique = len(set(words))
-        total = len(words)
-        ttr = round(unique / total, 3)
+        window = words[:self.VOCAB_WINDOW_WORDS]
+        unique = len(set(window))
+        ttr = round(unique / len(window), 3)
 
         return {
             "type_token_ratio": ttr,
             "unique_words": unique,
-            "total_words": total
+            "total_words": len(words),  # full-session count, kept for display \u2014 TTR itself is windowed
         }
 
     def _compute_silence_ratio(self, user_segs, other_segs) -> dict:

backend/pipeline/trigger_detector.py

@@ -0,0 +1,309 @@
+"""Trigger-detection engine for the Home feed's dimension-maturation cards.
+
+Diffs a user's per-dimension evidence — computed fresh after a session just
+finalized — against the last known persisted state, and writes any newly
+fired event into `dimension_events` (the frozen, append-only log the Home
+feed reads). Also upserts `signal_evidence_state` (the "last known state"
+row) every run, whether or not anything fires.
+
+Five trigger types, per dimension+scope:
+  - first_time_steady: not steady -> steady, for the first time ever (overall scope)
+  - context_shift:      same as above, but for a specific conversation-type scope
+  - recurring:          was steady, lost steadiness, regained it (direction
+                        distinguishes "back to usual" from a fresh drift found
+                        via the regain)
+  - drift:              already steady, established mean/mode has moved beyond
+                        the dimension's own noise band
+  - anomaly:            a single session's raw value contradicts an already-
+                        established baseline — independent of the above, can
+                        co-fire alongside any of them
+
+Deliberately does NOT run from `reanalyze_session` (main.py) — retroactively
+correcting which speaker is the user after evidence has already accumulated
+on the wrong speaker's data is a reconciliation problem (would need to
+un-fire/re-fire historical events) explicitly out of scope for this pass.
+"""
+import json
+
+from db.database import supabase_admin
+from pipeline.evidence_gate import (
+    SIGNAL_EVIDENCE_CONFIG, SUB_SIGNAL_EVIDENCE_CONFIG,
+    compute_evidence, extract_value,
+)
+
+COOLDOWN_SESSIONS = 2       # minimum sessions between any two fired events for the same dimension+scope
+ANOMALY_BAND_MULTIPLIER = 2.5  # anomaly band = this many x the dimension's own cv_threshold
+STRONG_MODE_AGREEMENT = 0.80   # categorical anomaly requires an established mode at least this strong
+
+
+def run_trigger_detection(user_id: str, session_id: str, context: str) -> list:
+    """Entry point — call right after _save_session() succeeds in main.py's
+    finalize flow. Re-fetches all parsed sessions (now including the
+    just-saved one) rather than taking `signals` as a param, so it reasons
+    over the exact same shape _compute_profile_evidence does. Returns the
+    list of newly fired events (for logging) — never raises; callers should
+    still wrap this in try/except so a bug here can never break finalize."""
+    from main import _fetch_and_parse_sessions  # local import — avoids a circular import at module load
+
+    parsed = _fetch_and_parse_sessions(user_id)
+    if not parsed:
+        return []
+
+    fired = []
+
+    # Pooled ("overall") + this session's own context, for the 15 main dimensions.
+    for dimension_key, cfg in SIGNAL_EVIDENCE_CONFIG.items():
+        for scope in ("overall", context):
+            fired += _check_dimension(user_id, session_id, dimension_key, scope, parsed,
+                                       cfg, SIGNAL_EVIDENCE_CONFIG)
+
+    # Sub-signals: overall scope only, first_time_steady only (see their
+    # "allowed_triggers" in evidence_gate.SUB_SIGNAL_EVIDENCE_CONFIG).
+    for sub_key, cfg in SUB_SIGNAL_EVIDENCE_CONFIG.items():
+        fired += _check_dimension(user_id, session_id, sub_key, "overall", parsed,
+                                   cfg, SUB_SIGNAL_EVIDENCE_CONFIG,
+                                   allowed_triggers=cfg.get("allowed_triggers"))
+
+    return fired
+
+
+def _scoped_values(parsed: list, signal_key: str, scope: str, config: dict) -> list:
+    """Oldest->newest values for this signal, filtered to `scope` ('overall'
+    or a specific context string). Config-agnostic — works for both the main
+    15 dimensions and the 3 sub-signals via extract_value's dual-dict lookup."""
+    values = []
+    for p in parsed:
+        if scope != "overall" and p["context"] != scope:
+            continue
+        try:
+            values.append(extract_value(signal_key, p["sig"]))
+        except (KeyError, TypeError):
+            pass
+    return values
+
+
+def _fetch_state(user_id: str, dimension_key: str, scope: str) -> dict:
+    res = supabase_admin.table("signal_evidence_state").select("*").eq(
+        "user_id", user_id
+    ).eq("dimension_key", dimension_key).eq("scope", scope).execute()
+    return res.data[0] if res.data else {}
+
+
+def _upsert_state(user_id, dimension_key, scope, kind, is_steady, has_ever_been_steady,
+                   last_steady_mean, last_steady_mode_label, last_steady_agreement_ratio,
+                   sample_count, last_fired_trigger_type, last_fired_session_id,
+                   last_fired_sample_count):
+    supabase_admin.table("signal_evidence_state").upsert({
+        "user_id": user_id, "dimension_key": dimension_key, "scope": scope, "kind": kind,
+        "is_steady": is_steady, "has_ever_been_steady": has_ever_been_steady,
+        "last_steady_mean": last_steady_mean,
+        "last_steady_mode_label": last_steady_mode_label,
+        "last_steady_agreement_ratio": last_steady_agreement_ratio,
+        "sample_count": sample_count,
+        "last_fired_trigger_type": last_fired_trigger_type,
+        "last_fired_session_id": last_fired_session_id,
+        "last_fired_sample_count": last_fired_sample_count,
+    }, on_conflict="user_id,dimension_key,scope").execute()
+
+
+def _insert_event(user_id, session_id, dimension_key, scope, trigger_type, direction,
+                   value_at_trigger, previous_value, label_at_trigger, previous_label,
+                   sample_count, card_copy: dict) -> dict:
+    row = {
+        "user_id": user_id, "session_id": session_id, "dimension_key": dimension_key,
+        "scope": scope, "trigger_type": trigger_type, "direction": direction,
+        "value_at_trigger": value_at_trigger, "previous_value": previous_value,
+        "label_at_trigger": label_at_trigger, "previous_label": previous_label,
+        "sample_count": sample_count, "card_copy_json": json.dumps(card_copy),
+    }
+    res = supabase_admin.table("dimension_events").insert(row).execute()
+    return res.data[0] if res.data else row
+
+
+def _has_drifted(is_categorical: bool, prior: dict, current: dict, cfg: dict) -> bool:
+    if is_categorical:
+        prev_label = prior.get("last_steady_mode_label")
+        return prev_label is not None and current.get("mode_label") != prev_label
+    prev_mean = prior.get("last_steady_mean")
+    if prev_mean is None or abs(prev_mean) < 1e-9:
+        return False
+    relative_change = abs(current["mean"] - prev_mean) / abs(prev_mean)
+    return relative_change > cfg["cv_threshold"]
+
+
+def _recurring_direction(is_categorical: bool, prior: dict, current: dict, cfg: dict) -> str:
+    """Distinguishes 'back_to_usual' (regained the SAME value) from a fresh
+    drift discovered via the regain — same underlying content as `drift`,
+    just framed as acknowledging the instability first."""
+    return "drift" if _has_drifted(is_categorical, prior, current, cfg) else "back_to_usual"
+
+
+def _check_anomaly(is_categorical: bool, prior: dict, values: list, cfg: dict):
+    """This session's raw value vs. the established baseline. Returns
+    (direction, value_or_label) or None. `values` is oldest->newest for this
+    dimension+scope — the last entry is this session's own raw value."""
+    if not values or values[-1] is None:
+        return None
+    this_value = values[-1]
+
+    if is_categorical:
+        established_label = prior.get("last_steady_mode_label")
+        established_agreement = prior.get("last_steady_agreement_ratio") or 0
+        if not established_label or established_agreement < STRONG_MODE_AGREEMENT:
+            return None
+        if this_value == established_label:
+            return None
+        return ("contradicts_established", this_value)
+
+    established_mean = prior.get("last_steady_mean")
+    if established_mean is None or abs(established_mean) < 1e-9:
+        return None
+    band = ANOMALY_BAND_MULTIPLIER * cfg["cv_threshold"]
+    relative_change = (this_value - established_mean) / abs(established_mean)
+    if relative_change > band:
+        return ("up", this_value)
+    if relative_change < -band:
+        return ("down", this_value)
+    return None
+
+
+def _value_str(dimension_key: str, value) -> str:
+    if value is None:
+        return "n/a"
+    try:
+        from pipeline.portrait_synthesizer import _format_mean
+        return _format_mean(dimension_key, value)
+    except Exception:
+        return f"{value}"
+
+
+def _phrase_event(dimension_key, label, trigger_type, direction, prior, current, cfg, scope) -> dict:
+    """Deterministic string templates, no LLM call — matches home_feed.py's
+    existing f-string convention (e.g. the old build_progress_cards)."""
+    is_categorical = cfg["kind"] == "categorical"
+    n = current["sample_count"]
+    scope_phrase = "" if scope == "overall" else f" in your {scope.replace('_', ' ')} conversations"
+    curr_str = current["mode_label"] if is_categorical else _value_str(dimension_key, current["mean"])
+
+    if trigger_type == "first_time_steady":
+        note = (f"Over your last {n} sessions, your {label} has settled into a steady "
+                 f"pattern — {curr_str}.")
+    elif trigger_type == "context_shift":
+        note = (f"Your {label} has also settled into a steady pattern specifically"
+                 f"{scope_phrase} — {curr_str}.")
+    elif trigger_type == "recurring" and direction == "back_to_usual":
+        note = (f"Your {label} had been varying more than usual for a bit, but it's "
+                 f"settled back to your usual {curr_str}.")
+    elif trigger_type == "recurring":  # direction == "drift"
+        note = (f"After a stretch of inconsistency, your {label} has settled into a new "
+                 f"pattern — {curr_str}.")
+    elif trigger_type == "drift":
+        prev_str = (prior.get("last_steady_mode_label") if is_categorical
+                    else _value_str(dimension_key, prior.get("last_steady_mean")))
+        note = f"Your {label} has shifted — from {prev_str} to {curr_str} over your last {n} sessions."
+    else:
+        note = f"Your {label} showed a new pattern this session."
+
+    return {"label": label, "note": note}
+
+
+def _phrase_anomaly(dimension_key, label, direction, prior, this_value, this_label, cfg) -> dict:
+    is_categorical = cfg["kind"] == "categorical"
+    if is_categorical:
+        established = prior.get("last_steady_mode_label", "n/a")
+        note = (f"Unlike your usual pattern of {established} {label}, this session was "
+                 f"{this_label} — worth noting, not necessarily a new pattern yet.")
+    else:
+        established = _value_str(dimension_key, prior.get("last_steady_mean"))
+        this_str = _value_str(dimension_key, this_value)
+        word = "higher" if direction == "up" else "lower"
+        note = (f"This session your {label} was notably {word} than usual — {this_str} vs "
+                 f"your typical {established}. Worth noting, not necessarily a new pattern yet.")
+    return {"label": label, "note": note}
+
+
+def _check_dimension(user_id, session_id, dimension_key, scope, parsed, cfg, config_dict,
+                      allowed_triggers=None) -> list:
+    """Returns the list of newly fired events for this (dimension, scope) this run."""
+    is_categorical = cfg["kind"] == "categorical"
+    label = cfg["label"]
+    values = _scoped_values(parsed, dimension_key, scope, config_dict)
+    current = compute_evidence(dimension_key, values, config_dict)
+
+    prior = _fetch_state(user_id, dimension_key, scope)
+    prior_is_steady = bool(prior.get("is_steady"))
+    has_ever_been_steady = bool(prior.get("has_ever_been_steady"))
+    last_fired_sample_count = prior.get("last_fired_sample_count")
+
+    def allowed(trigger_type):
+        return allowed_triggers is None or trigger_type in allowed_triggers
+
+    cooldown_clear = (
+        last_fired_sample_count is None or
+        current["sample_count"] - last_fired_sample_count >= COOLDOWN_SESSIONS
+    )
+
+    candidates = []  # list of (trigger_type, direction)
+    if current["is_steady"] and not prior_is_steady:
+        if not has_ever_been_steady:
+            candidates.append(("first_time_steady" if scope == "overall" else "context_shift", None))
+        else:
+            candidates.append(("recurring", _recurring_direction(is_categorical, prior, current, cfg)))
+    elif current["is_steady"] and prior_is_steady:
+        if _has_drifted(is_categorical, prior, current, cfg):
+            candidates.append(("drift", None))
+    # steady -> not-steady, not-steady -> not-steady: no candidate from this branch.
+
+    anomaly = None
+    if scope == "overall" and prior_is_steady:
+        anomaly = _check_anomaly(is_categorical, prior, values, cfg)
+        if anomaly:
+            candidates.append(("anomaly", anomaly[0]))
+
+    fired = []
+    for trigger_type, direction in candidates:
+        if not allowed(trigger_type) or not cooldown_clear:
+            continue
+        if trigger_type == "anomaly":
+            _, anomaly_value_or_label = anomaly
+            this_value = None if is_categorical else anomaly_value_or_label
+            this_label = anomaly_value_or_label if is_categorical else None
+            copy = _phrase_anomaly(dimension_key, label, direction, prior, this_value, this_label, cfg)
+            fired.append(_insert_event(
+                user_id, session_id, dimension_key, scope, "anomaly", direction,
+                this_value, prior.get("last_steady_mean"), this_label, prior.get("last_steady_mode_label"),
+                current["sample_count"], copy,
+            ))
+        else:
+            copy = _phrase_event(dimension_key, label, trigger_type, direction, prior, current, cfg, scope)
+            fired.append(_insert_event(
+                user_id, session_id, dimension_key, scope, trigger_type, direction,
+                current.get("mean"), prior.get("last_steady_mean"),
+                current.get("mode_label"), prior.get("last_steady_mode_label"),
+                current["sample_count"], copy,
+            ))
+
+    # Always update state — freeze last_steady_* while not steady, so a later
+    # drift/anomaly check compares against the last REAL steady baseline, not
+    # noise from an unsteady stretch.
+    if current["is_steady"]:
+        last_steady_mean = current.get("mean")
+        last_steady_mode_label = current.get("mode_label")
+        last_steady_agreement_ratio = current.get("agreement_ratio")
+    else:
+        last_steady_mean = prior.get("last_steady_mean")
+        last_steady_mode_label = prior.get("last_steady_mode_label")
+        last_steady_agreement_ratio = prior.get("last_steady_agreement_ratio")
+
+    any_fired = len(fired) > 0
+    _upsert_state(
+        user_id, dimension_key, scope, cfg["kind"],
+        current["is_steady"], has_ever_been_steady or current["is_steady"],
+        last_steady_mean, last_steady_mode_label, last_steady_agreement_ratio,
+        current["sample_count"],
+        last_fired_trigger_type=(fired[-1]["trigger_type"] if any_fired else prior.get("last_fired_trigger_type")),
+        last_fired_session_id=(session_id if any_fired else prior.get("last_fired_session_id")),
+        last_fired_sample_count=(current["sample_count"] if any_fired else last_fired_sample_count),
+    )
+
+    return fired

frontend/src/components/HomeView.jsx

@@ -2,31 +2,21 @@ import { useState, useEffect } from "react"
 import api from "../lib/api"
 import Reveal, { RevealItem } from "./Reveal"
 
-const G = "linear-gradient(135deg, #1d4ed8 0%, #0891b2 100%)"
-
-// Mirrors ProfileView's FRAMING_CONFIG — growth_area/observation framings of a
-// steady signal render with this palette; other card types get their own fixed
-// color below. Kept as a separate copy rather than a shared import since the
-// two views are deliberately allowed to diverge visually over time.
-const FRAMING_CONFIG = {
-  strength:    { color: "#34d399", label: "Strength" },
-  growth_area: { color: "#fb923c", label: "Growth area" },
-  observation: { color: "#818cf8", label: "Observation" },
+// Visual identity per trigger type — independent of the (not-yet-built) per-
+// dimension color palette on the You page; this is scoped to what Home's
+// dimension-event cards need: a quick visual read on WHAT KIND of thing
+// happened (a first discovery vs. a shift vs. something to just note), not
+// WHICH of the 15 dimensions it is.
+const TRIGGER_CONFIG = {
+  first_time_steady: { color: "#34d399", label: "New pattern" },
+  context_shift:      { color: "#22d3ee", label: "Context shift" },
+  drift:              { color: "#f59e0b", label: "Shifted" },
+  recurring:          { color: "#a78bfa", label: "Recurring" },
+  anomaly:            { color: "#fb7185", label: "Worth noting" },
 }
 
-const TYPE_CONFIG = {
-  strength:            { color: "#34d399", label: "Strength" },
-  how_it_may_land:     { color: "#22d3ee", label: "How it may land" },
-  progress:            { color: "#5b9cf6", label: "Progress" },
-  still_forming:       { color: "#6b6888", label: "Still forming" },
-  session_observation: { color: "#818cf8", label: "From your last session" },
-}
-
-function cardVisual(card) {
-  if (card.type === "observation") {
-    return FRAMING_CONFIG[card.framing] || FRAMING_CONFIG.observation
-  }
-  return TYPE_CONFIG[card.type] || { color: "#8b89aa", label: card.type }
+function recurringLabel(direction) {
+  return direction === "back_to_usual" ? "Back to usual" : "New pattern"
 }
 
 function DismissButton({ onDismiss }) {
@@ -40,82 +30,120 @@ function DismissButton({ onDismiss }) {
   )
 }
 
-function HomeCard({ card, i, onDismiss }) {
-  const cfg = cardVisual(card)
-  const title = card.type === "session_observation"
-    ? (card.signal || "").replace(/_/g, " ") || cfg.label
-    : (card.label || cfg.label)
-
+function CardShell({ color, title, badge, faded, onDismiss, children }) {
   return (
-    <RevealItem index={i}>
     <div style={{
       padding: "14px 16px", borderRadius: 10,
       background: "#151922", border: "1px solid #1e2438",
-      borderLeft: `3px solid ${cfg.color}`,
+      borderLeft: `3px solid ${color}`,
+      opacity: faded ? 0.45 : 1,
+      transition: "opacity 0.3s ease",
     }}>
       <div style={{ display: "flex", alignItems: "flex-start", justifyContent: "space-between", marginBottom: 6 }}>
         <div style={{ display: "flex", alignItems: "center", gap: 8, flexWrap: "wrap" }}>
           <span style={{ fontSize: 13, fontWeight: 700, color: "#f0eeff", textTransform: "capitalize" }}>
             {title}
           </span>
-          <span style={{ fontSize: 11, color: cfg.color, fontWeight: 600,
-            background: `${cfg.color}15`, border: `1px solid ${cfg.color}30`,
-            borderRadius: 20, padding: "3px 10px", whiteSpace: "nowrap" }}>
-            {cfg.label}{card.type === "progress" && (card.direction === "up" ? " ↑" : " ↓")}
-          </span>
+          {badge && (
+            <span style={{ fontSize: 11, color, fontWeight: 600,
+              background: `${color}15`, border: `1px solid ${color}30`,
+              borderRadius: 20, padding: "3px 10px", whiteSpace: "nowrap" }}>
+              {badge}
+            </span>
+          )}
         </div>
-        <DismissButton onDismiss={() => onDismiss(card.card_key)} />
+        <DismissButton onDismiss={onDismiss} />
       </div>
+      {children}
+    </div>
+  )
+}
 
-      {card.note && (
-        <p style={{ margin: "0 0 8px", fontSize: 13, color: "#c4c2d8", lineHeight: 1.65 }}>
-          {card.note}
+function SessionRecapCard({ card, faded, onDismiss }) {
+  const dateLabel = card.date ? new Date(card.date).toLocaleDateString(undefined, { month: "short", day: "numeric" }) : ""
+  return (
+    <CardShell color="#818cf8" title="Session recap"
+      badge={[card.context?.replace(/_/g, " "), dateLabel].filter(Boolean).join(" · ")}
+      faded={faded} onDismiss={onDismiss}>
+      {card.conversation_summary && (
+        <p style={{ margin: "0 0 10px", fontSize: 13, color: "#c4c2d8", lineHeight: 1.65 }}>
+          {card.conversation_summary}
         </p>
       )}
 
-      {card.type === "still_forming" && (
-        <div style={{ marginTop: 4 }}>
-          <div style={{ height: 3, background: "#1e2438", borderRadius: 2 }}>
-            <div style={{ height: "100%", borderRadius: 2,
-              width: `${Math.min(100, Math.round((card.sample_count / card.min_needed) * 100))}%`,
-              background: "#3a3a52" }} />
-          </div>
-          <p style={{ margin: "8px 0 0", fontSize: 11, color: "#4a4865" }}>
-            {card.sample_count} of {card.min_needed} sessions
-          </p>
+      {card.observations?.length > 0 && (
+        <div style={{ display: "flex", flexDirection: "column", gap: 8, marginBottom: card.coaching_suggestions?.length ? 10 : 0 }}>
+          {card.observations.map((obs, i) => (
+            <div key={i}>
+              <p style={{ margin: 0, fontSize: 13, color: "#c4c2d8", lineHeight: 1.6 }}>
+                {obs.observation}
+              </p>
+              {obs.resonance_prompt && (
+                <p style={{ margin: "4px 0 0", fontSize: 12, color: "#6b6888", fontStyle: "italic", lineHeight: 1.5 }}>
+                  💭 {obs.resonance_prompt}
+                </p>
+              )}
+            </div>
+          ))}
         </div>
       )}
 
-      {(card.type === "strength" || card.type === "observation") && card.sample_count != null && (
-        <p style={{ margin: 0, fontSize: 11, color: "#4a4865" }}>
-          Based on {card.sample_count} sessions
-        </p>
+      {card.coaching_suggestions?.length > 0 && (
+        <div style={{ display: "flex", flexDirection: "column", gap: 6, marginTop: 4 }}>
+          {card.coaching_suggestions.map((s, i) => (
+            <p key={i} style={{ margin: 0, fontSize: 12, color: "#8b89aa", lineHeight: 1.55 }}>
+              <span style={{ color: "#a5a3c2", fontWeight: 600 }}>{s.area}: </span>
+              {s.suggestion}
+            </p>
+          ))}
+        </div>
       )}
 
-      {card.type === "progress" && card.context && (
-        <p style={{ margin: 0, fontSize: 11, color: "#4a4865" }}>
-          In your {card.context.replace(/_/g, " ")} conversations
+      {card.notable_pattern && (
+        <p style={{ margin: "10px 0 0", fontSize: 12, color: "#4a4865", lineHeight: 1.5, borderTop: "1px solid #1e2438", paddingTop: 8 }}>
+          {card.notable_pattern}
         </p>
       )}
+    </CardShell>
+  )
+}
 
-      {card.type === "session_observation" && card.resonance_prompt && (
-        <p style={{ margin: "8px 0 0", fontSize: 12, color: "#6b6888",
-          fontStyle: "italic", lineHeight: 1.5 }}>
-          💭 {card.resonance_prompt}
+function DimensionEventCard({ card, faded, onDismiss }) {
+  const trig = TRIGGER_CONFIG[card.trigger_type] || { color: "#8b89aa", label: card.trigger_type }
+  const badge = card.trigger_type === "recurring" ? recurringLabel(card.direction) : trig.label
+  return (
+    <CardShell color={trig.color} title={card.label} badge={badge} faded={faded} onDismiss={onDismiss}>
+      {card.note && (
+        <p style={{ margin: 0, fontSize: 13, color: "#c4c2d8", lineHeight: 1.65 }}>
+          {card.note}
         </p>
       )}
-    </div>
+    </CardShell>
+  )
+}
+
+function HomeCard({ card, i, faded, onDismiss }) {
+  const dismiss = () => onDismiss(card.card_key)
+  return (
+    <RevealItem index={i}>
+      {card.type === "dimension_event"
+        ? <DimensionEventCard card={card} faded={faded} onDismiss={dismiss} />
+        : <SessionRecapCard card={card} faded={faded} onDismiss={dismiss} />}
     </RevealItem>
   )
 }
 
+const VISIBLE_COUNT = 7
+
 export default function HomeView({ active }) {
   const [data, setData] = useState(null)
   const [loading, setLoading] = useState(true)
+  const [expanded, setExpanded] = useState(false)
 
   useEffect(() => {
     if (!active) return
     setLoading(true)
+    setExpanded(false)
     api.get("/api/home")
       .then(res => setData(res.data))
       .catch(console.error)
@@ -152,6 +180,11 @@ export default function HomeView({ active }) {
   }
 
   const cards = data.cards || []
+  // Backend returns the complete, newest-first, dismissed-filtered list —
+  // pagination is purely a frontend rendering concern: 7 visible + an 8th
+  // shown faded as a teaser, "Show more" reveals the rest uncapped.
+  const hasOverflow = cards.length > VISIBLE_COUNT
+  const visibleCards = expanded ? cards : cards.slice(0, VISIBLE_COUNT + 1)
 
   return (
     <div style={{ display: "flex", flexDirection: "column", gap: 28 }}>
@@ -176,9 +209,19 @@ export default function HomeView({ active }) {
       ) : (
         <Reveal delay={80}>
         <div style={{ display: "flex", flexDirection: "column", gap: 8 }}>
-          {cards.map((card, i) => (
-            <HomeCard key={card.card_key} card={card} i={i} onDismiss={handleDismiss} />
+          {visibleCards.map((card, i) => (
+            <HomeCard key={card.card_key} card={card} i={i}
+              faded={!expanded && hasOverflow && i === VISIBLE_COUNT}
+              onDismiss={handleDismiss} />
           ))}
+          {!expanded && hasOverflow && (
+            <button onClick={() => setExpanded(true)}
+              style={{ alignSelf: "center", marginTop: 4, background: "none",
+                border: "none", cursor: "pointer", color: "#7c8fd6",
+                fontSize: 13, fontWeight: 600, padding: "8px 0" }}>
+              Show more
+            </button>
+          )}
         </div>
         </Reveal>
       )}