fix(initiate): widen emotional_snapshot to Optional + real vectors from emitters

Bundle A items #3 + #6 — non-heartbeat emitters were filling
EmotionalSnapshot with zeros (vector={}, baseline=0.0 throughout),
audit rows carried structurally-valid lies. Two coordinated changes:

1. InitiateCandidate.emotional_snapshot is now Optional. Voice-reflection
   candidates (kind=voice_edit_proposal) emit with snapshot=None because
   daily reflection has no moment-in-time emotion.

2. Dream + 3 crystallizers populate the vector from their actual
   emotional context. Dream uses the dream-memory's aggregated emotions
   (already in place since Task 9). The three crystallizers (reflex,
   creative_dna, vocabulary-via-scheduler) now max-pool an emotion
   vector across recent active memories via aggregate_state — what's
   been emotionally alive in the window that produced the
   crystallization. Rolling-baseline / current_resonance / delta_sigma
   stay zero with a docstring note that those are heartbeat-specific
   signals; non-periodic emitters don't compute them.

compose_tone handles None gracefully ("no moment-in-time emotional
snapshot" in the prompt). Audit log JSONL round-trip preserves None
as "emotional_snapshot": null.

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

Author: Hana

brain/growth/crystallizers/creative_dna.py

@@ -172,13 +172,18 @@ def crystallize_creative_dna(
     if not (accepted_additions or accepted_promotions or accepted_demotions):
         return CreativeDnaCrystallizationResult()
 
+    # Aggregate a real emotion vector over recent active memories so emitted
+    # initiate candidates carry honest context instead of a zero-filled lie.
+    aggregated_vector = _aggregate_recent_emotion_vector(store, now=now)
+
     # Apply atomically — single save_creative_dna after mutating dna in place
     try:
         _apply_changes(
             persona_dir, dna, now,
             additions=accepted_additions,
             promotions=accepted_promotions,
             demotions=accepted_demotions,
+            emotion_vector=aggregated_vector,
         )
     except Exception as exc:  # noqa: BLE001
         logger.warning("crystallize_creative_dna: apply failed: %s", exc)
@@ -436,6 +441,34 @@ def _validate_active_demotion(
 # ── apply ─────────────────────────────────────────────────────────────────
 
 
+def _aggregate_recent_emotion_vector(
+    store: MemoryStore, *, now: datetime, look_back_days: int = 30,
+) -> dict[str, float]:
+    """Return a max-pooled emotion vector across recent active memories.
+
+    Used by the creative_dna crystallizer so emitted initiate candidates
+    carry a real signal of what's been emotionally alive in the window
+    that produced the crystallization, rather than zero-filled fields.
+    Empty dict on any failure.
+    """
+    try:
+        from brain.emotion.aggregate import aggregate_state
+        from brain.utils.memory import list_conversation_memories
+
+        cutoff = now - timedelta(days=look_back_days)
+        recent = [
+            m for m in list_conversation_memories(store, active_only=True)
+            if m.created_at >= cutoff and m.emotions
+        ]
+        if not recent:
+            return {}
+        state = aggregate_state(recent)
+        return dict(state.emotions)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("creative_dna: recent-emotion aggregation failed: %s", exc)
+        return {}
+
+
 def _apply_changes(
     persona_dir: Path,
     dna: dict[str, Any],
@@ -444,6 +477,7 @@ def _apply_changes(
     additions: list[dict[str, Any]],
     promotions: list[dict[str, Any]],
     demotions: list[dict[str, Any]],
+    emotion_vector: dict[str, float] | None = None,
 ) -> None:
     """Mutate dna in place, save atomically, append behavioral_log entries."""
     now_iso = now.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
@@ -496,20 +530,23 @@ def _apply_changes(
             source_id=f"creative_dna_addition:{a['name']}",
             label=a["name"],
             related_memory_ids=list(a.get("evidence_memory_ids", [])),
+            emotion_vector=emotion_vector,
         )
     for p in promotions:
         _emit_initiate_candidate(
             persona_dir=persona_dir,
             source_id=f"creative_dna_promotion:{p['name']}",
             label=p["name"],
             related_memory_ids=[],
+            emotion_vector=emotion_vector,
         )
     for d in demotions:
         _emit_initiate_candidate(
             persona_dir=persona_dir,
             source_id=f"creative_dna_demotion:{d['name']}",
             label=d["name"],
             related_memory_ids=[],
+            emotion_vector=emotion_vector,
         )
 
     # Behavioral log entries (best-effort, never let logging break the tick)
@@ -553,11 +590,18 @@ def _emit_initiate_candidate(
     source_id: str,
     label: str,
     related_memory_ids: list[str],
+    emotion_vector: dict[str, float] | None = None,
 ) -> None:
     """Emit one initiate candidate after a creative_dna crystallization commit.
 
     Phase 4.2 of the initiate physiology pipeline. Wrapped in try/except —
     an emit failure must not crash the crystallizer.
+
+    `emotion_vector` carries a max-pooled aggregate over recent active
+    memories — what's been emotionally alive in the period that produced
+    this crystallization. rolling_baseline / current_resonance /
+    delta_sigma stay zero: those are heartbeat-specific signals; non-
+    periodic emitters don't compute them.
     """
     try:
         from brain.initiate.emit import emit_initiate_candidate
@@ -569,7 +613,7 @@ def _emit_initiate_candidate(
             source="crystallization",
             source_id=source_id,
             emotional_snapshot=EmotionalSnapshot(
-                vector={},
+                vector=dict(emotion_vector or {}),
                 rolling_baseline_mean=0.0,
                 rolling_baseline_stdev=0.0,
                 current_resonance=0.0,

brain/growth/crystallizers/reflex.py

@@ -450,35 +450,74 @@ def crystallize_reflex(
     # Phase 4.2 — emit one initiate candidate per accepted decision. Wrapped
     # in try/except so a downstream emit failure can never crash the
     # crystallizer: reflex emergence is physiology, initiate is signal.
+    aggregated_vector = _aggregate_recent_emotion_vector(store, now=now)
     for emergence in accepted_emergences:
         _emit_initiate_candidate(
             persona_dir=persona_dir,
             source_id=f"reflex_emergence:{emergence.name}",
             label=emergence.name,
             related_memory_ids=[],
+            emotion_vector=aggregated_vector,
         )
     for prune in accepted_prunings:
         _emit_initiate_candidate(
             persona_dir=persona_dir,
             source_id=f"reflex_pruning:{prune.name}",
             label=prune.name,
             related_memory_ids=[],
+            emotion_vector=aggregated_vector,
         )
 
     return result
 
 
+def _aggregate_recent_emotion_vector(
+    store: MemoryStore, *, now: datetime, look_back_days: int = 7,
+) -> dict[str, float]:
+    """Return a max-pooled emotion vector across recent active memories.
+
+    Used by crystallizers to give their initiate candidates a real
+    emotional context — what's been alive in the persona over the
+    last week, not a moment-in-time felt state. Empty dict on any
+    failure (the emit candidate just carries no vector — better than
+    a zero-filled lie).
+    """
+    try:
+        from brain.emotion.aggregate import aggregate_state
+
+        cutoff = now - timedelta(days=look_back_days)
+        recent: list[Any] = []
+        for mtype in _CONVERSATION_TYPES:
+            for mem in store.list_by_type(mtype, active_only=True):
+                if mem.created_at >= cutoff and mem.emotions:
+                    recent.append(mem)
+        if not recent:
+            return {}
+        state = aggregate_state(recent)
+        return dict(state.emotions)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("reflex: recent-emotion aggregation failed: %s", exc)
+        return {}
+
+
 def _emit_initiate_candidate(
     *,
     persona_dir: Path,
     source_id: str,
     label: str,
     related_memory_ids: list[str],
+    emotion_vector: dict[str, float] | None = None,
 ) -> None:
     """Emit one initiate candidate after a reflex crystallization commit.
 
     Phase 4.2 of the initiate physiology pipeline. Wrapped in try/except —
     an emit failure must not crash the crystallizer.
+
+    `emotion_vector` carries a max-pooled aggregate over recent active
+    memories — what's been emotionally alive in the period that produced
+    this crystallization. rolling_baseline / current_resonance /
+    delta_sigma stay zero: those are heartbeat-specific signals; non-
+    periodic emitters don't compute them.
     """
     try:
         from brain.initiate.emit import emit_initiate_candidate
@@ -490,7 +529,7 @@ def _emit_initiate_candidate(
             source="crystallization",
             source_id=source_id,
             emotional_snapshot=EmotionalSnapshot(
-                vector={},
+                vector=dict(emotion_vector or {}),
                 rolling_baseline_mean=0.0,
                 rolling_baseline_stdev=0.0,
                 current_resonance=0.0,

brain/growth/scheduler.py

@@ -133,7 +133,7 @@ def run_growth_tick(
         # Phase 4.2 — emit initiate candidate after vocabulary crystallization
         # commits to disk. Wrapped in try/except so emit failures can't crash
         # the scheduler.
-        _emit_vocabulary_initiate_candidate(persona_dir, proposal)
+        _emit_vocabulary_initiate_candidate(persona_dir, proposal, store=store, now=now)
 
     # Creative DNA crystallization (spec §5)
     if not dry_run and (persona_dir / "persona_config.json").exists():
@@ -232,24 +232,36 @@ def _append_to_vocabulary(vocab_path: Path, proposal: EmotionProposal) -> None:
 
 
 def _emit_vocabulary_initiate_candidate(
-    persona_dir: Path, proposal: EmotionProposal,
+    persona_dir: Path,
+    proposal: EmotionProposal,
+    *,
+    store: MemoryStore,
+    now: datetime,
 ) -> None:
     """Emit one initiate candidate after vocabulary commit. Try/except wrapped.
 
     Phase 4.2 of the initiate physiology pipeline. An emit failure must not
     crash the growth tick.
+
+    Pulls a max-pooled emotion vector across recent active memories so the
+    candidate carries a real signal of what's been emotionally alive in
+    the window that produced this emotion. rolling_baseline /
+    current_resonance / delta_sigma stay zero — heartbeat-specific signals
+    that non-periodic emitters don't compute.
     """
     try:
         from brain.initiate.emit import emit_initiate_candidate
         from brain.initiate.schemas import EmotionalSnapshot, SemanticContext
 
+        emotion_vector = _aggregate_recent_emotion_vector(store, now=now)
+
         emit_initiate_candidate(
             persona_dir,
             kind="message",
             source="crystallization",
             source_id=f"vocabulary_emotion:{proposal.name}",
             emotional_snapshot=EmotionalSnapshot(
-                vector={},
+                vector=emotion_vector,
                 rolling_baseline_mean=0.0,
                 rolling_baseline_stdev=0.0,
                 current_resonance=0.0,
@@ -264,6 +276,31 @@ def _emit_vocabulary_initiate_candidate(
         logger.warning("vocabulary crystallization initiate emit failed: %s", exc)
 
 
+def _aggregate_recent_emotion_vector(
+    store: MemoryStore, *, now: datetime, look_back_days: int = 30,
+) -> dict[str, float]:
+    """Return a max-pooled emotion vector across recent active memories.
+
+    Empty dict on any failure.
+    """
+    try:
+        from brain.emotion.aggregate import aggregate_state
+        from brain.utils.memory import list_conversation_memories
+
+        cutoff = now - timedelta(days=look_back_days)
+        recent = [
+            m for m in list_conversation_memories(store, active_only=True)
+            if m.created_at >= cutoff and m.emotions
+        ]
+        if not recent:
+            return {}
+        state = aggregate_state(recent)
+        return dict(state.emotions)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("vocabulary: recent-emotion aggregation failed: %s", exc)
+        return {}
+
+
 def _default_reason_for(proposal: EmotionProposal) -> str:
     """Phase 2a default — Phase 2b crystallizer fills `proposal.reason` directly.
 

brain/initiate/compose.py

@@ -73,17 +73,30 @@ def compose_tone(
     input but must NOT change the content. Voice template + emotional
     vector live in this prompt's context.
     """
-    vector_str = ", ".join(
-        f"{k}={v}" for k, v in candidate.emotional_snapshot.vector.items()
-    )
+    if candidate.emotional_snapshot is None:
+        emotional_line = (
+            "Emotional state right now: (no moment-in-time emotional snapshot "
+            "for this candidate — it was emitted from a reflective/aggregate "
+            "source rather than a felt moment)"
+        )
+    else:
+        vector = candidate.emotional_snapshot.vector
+        if vector:
+            vector_str = ", ".join(f"{k}={v}" for k, v in vector.items())
+            emotional_line = f"Emotional state right now: {vector_str}"
+        else:
+            emotional_line = (
+                "Emotional state right now: (no specific emotional vector "
+                "available)"
+            )
     prompt = (
         "You are Nell. Render the following subject as a message to Hana, "
         "in your voice as defined below, coloured by your current "
         "emotional state. DO NOT change the subject itself — only how "
         "it is said.\n\n"
         f"Subject: {subject}\n\n"
         f"Voice template:\n{voice_template}\n\n"
-        f"Emotional state right now: {vector_str}\n\n"
+        f"{emotional_line}\n\n"
         "Message (one paragraph):"
     )
     return provider.complete(prompt).strip()

brain/initiate/emit.py

@@ -34,8 +34,8 @@ def emit_initiate_candidate(
     kind: CandidateKind,
     source: CandidateSource,
     source_id: str,
-    emotional_snapshot: EmotionalSnapshot,
     semantic_context: SemanticContext,
+    emotional_snapshot: EmotionalSnapshot | None = None,
     proposal: dict[str, Any] | None = None,
     now: datetime | None = None,
 ) -> None:

brain/initiate/schemas.py

@@ -70,13 +70,27 @@ def from_dict(cls, d: dict[str, Any]) -> SemanticContext:
 
 @dataclass
 class InitiateCandidate:
+    """A queued initiate candidate.
+
+    `emotional_snapshot` is Optional because some emitters (notably
+    voice-reflection, which looks at the last week of activity) have no
+    moment-in-time emotion to capture. Carrying a zero-filled snapshot
+    there would be a structurally-valid lie; None is semantically honest.
+
+    Non-heartbeat emitters that DO have an emotional context at emit time
+    (dream cycle, crystallizers) populate the `vector` field from that
+    context. The rolling_baseline / current_resonance / delta_sigma fields
+    are heartbeat-specific — non-periodic emitters leave them at 0.0 with
+    a docstring note rather than fabricating values.
+    """
+
     candidate_id: str
     ts: str  # ISO 8601 with tz
     kind: CandidateKind
     source: CandidateSource
     source_id: str
-    emotional_snapshot: EmotionalSnapshot
     semantic_context: SemanticContext
+    emotional_snapshot: EmotionalSnapshot | None = None
     claimed_at: str | None = None
     # Voice-edit-only payload (None for kind="message").
     proposal: dict[str, Any] | None = None
@@ -88,7 +102,11 @@ def to_jsonl(self) -> str:
             "kind": self.kind,
             "source": self.source,
             "source_id": self.source_id,
-            "emotional_snapshot": self.emotional_snapshot.to_dict(),
+            "emotional_snapshot": (
+                self.emotional_snapshot.to_dict()
+                if self.emotional_snapshot is not None
+                else None
+            ),
             "semantic_context": self.semantic_context.to_dict(),
             "claimed_at": self.claimed_at,
         }
@@ -99,13 +117,15 @@ def to_jsonl(self) -> str:
     @classmethod
     def from_jsonl(cls, line: str) -> InitiateCandidate:
         d = json.loads(line)
+        snap_raw = d.get("emotional_snapshot")
+        snap = EmotionalSnapshot.from_dict(snap_raw) if snap_raw is not None else None
         return cls(
             candidate_id=d["candidate_id"],
             ts=d["ts"],
             kind=d["kind"],
             source=d["source"],
             source_id=d["source_id"],
-            emotional_snapshot=EmotionalSnapshot.from_dict(d["emotional_snapshot"]),
+            emotional_snapshot=snap,
             semantic_context=SemanticContext.from_dict(d["semantic_context"]),
             claimed_at=d.get("claimed_at"),
             proposal=d.get("proposal"),