# Ella — Technical Implementation Addendum A deeper dive on the four suggestions in the main feedback, written so an engineer (or a Claude-shaped coding agent) can build them. The goal is to replace "consider a narrative anchor" with "build a narrative anchor this way." --- ## 1. Narrative drift — a world-history ledger ### The problem, restated in code A single retrieval store can't tell the difference between: - "Ella's home has a red door" (canon, anchors the visual) - "Ella said yesterday she's thinking about painting the door blue" (first-person, not yet canon) - "The user mentioned seeing a red door in a dream" (user-side, off-world) If all three get retrieved as if they were facts, contradictions accumulate and the world blurs. ### The fix: a tiered fact ledger A small set of fact types, each with explicit provenance and confidence: ```python # fact_ledger/types.py from enum import Enum from dataclasses import dataclass, field from datetime import datetime from typing import Optional import uuid class FactTier(str, Enum): CANON = "canon" # world-truth, anchored, may gate visual regeneration OBSERVED = "observed" # Ella witnessed it; first-person recollection RUMORED = "rumored" # heard from someone in-world; uncertain USER_CLAIM = "user_claim" # user said it; not part of Ella's world INFERRED = "inferred" # extracted from conversation; may be wrong class FactStatus(str, Enum): ACTIVE = "active" SUPERSEDED = "superseded" # replaced by a newer fact RETIRED = "retired" # explicitly deleted by user or lifecycle rule CONTESTED = "contested" # multiple active facts disagree @dataclass class Fact: id: str = field(default_factory=lambda: str(uuid.uuid4())) tier: FactTier = FactTier.OBSERVED status: FactStatus = FactStatus.ACTIVE subject: str = "" # "ellahome.exterior.door_color" predicate: str = "" # "color" object: str = "" # "red" source_message_id: Optional[str] = None confidence: float = 1.0 # 0..1 created_at: datetime = field(default_factory=datetime.utcnow) supersedes: Optional[str] = None note: str = "" # human-readable provenance line ``` ### Storage: append-only ledger, not a single mutable row ```python # fact_ledger/store.py import json from pathlib import Path from typing import Iterable class FactLedger: """Append-only JSONL ledger. One line per fact write. Current state is materialized by replay + supersession resolution.""" def __init__(self, path: str): self.path = Path(path) self.path.parent.mkdir(parents=True, exist_ok=True) def append(self, fact: "Fact") -> None: with self.path.open("a") as f: f.write(json.dumps(fact.__dict__, default=str) + "\n") def supersede(self, old_id: str, new_fact: "Fact") -> "Fact": new_fact.supersedes = old_id self._mark(old_id, FactStatus.SUPERSEDED) self.append(new_fact) return new_fact def _mark(self, fact_id: str, status: FactStatus) -> None: # Rewrite is fine; this is a metadata update, not a new claim. # For high-volume systems, do this in a side-index instead. lines = self.path.read_text().splitlines() out = [] for line in lines: row = json.loads(line) if row.get("id") == fact_id and row.get("status") == FactStatus.ACTIVE.value: row["status"] = status.value out.append(json.dumps(row)) self.path.write_text("\n".join(out) + "\n") def current(self, subject: str | None = None) -> list[dict]: """Replay the ledger; keep only ACTIVE rows; resolve supersession chain.""" active: dict[str, dict] = {} for line in self.path.read_text().splitlines(): row = json.loads(line) sid = row["subject"] if subject and sid != subject: continue if row["status"] == FactStatus.SUPERSEDED.value: active.pop(row["id"], None) # also drop whatever this fact superseded, if it replaced the current continue if row["status"] == FactStatus.ACTIVE.value: active[row["id"]] = row return list(active.values()) ``` ### Why append-only: supersession, not overwrite When Ella's home gets a new door color, the old fact gets `status=SUPERSEDED` and a new `CANON` fact is appended with `supersedes=`. The history is preserved (good for "remember when the door was red?" continuity moments), and the current state is always derivable by replay. This is the same idea as the akashic-lethe presence/absence split. ### Tier-based retrieval The critical part: when assembling context for a reply, **facts are pulled by tier**, not as a single bag: ```python # fact_ledger/retrieval.py from .types import FactTier, FactStatus TIER_RANK = { FactTier.CANON: 0, FactTier.OBSERVED: 1, FactTier.RUMORED: 2, FactTier.USER_CLAIM: 3, FactTier.INFERRED: 4, } def assemble_world_brief(ledger, subject_prefix: str, max_tokens: int = 800) -> str: facts = ledger.current() facts = [f for f in facts if f["subject"].startswith(subject_prefix)] facts.sort(key=lambda f: (TIER_RANK[FactTier(f["tier"])], -f["confidence"])) lines, budget = [], max_tokens * 4 # rough char estimate for f in facts: line = f"- [{f['tier']}] {f['subject']}: {f['predicate']}={f['object']} (conf={f['confidence']})" if len(line) > budget: break lines.append(line) budget -= len(line) return "\n".join(lines) ``` CANON facts lead, INFERRED last. Ella's reply model sees the tier prefix in the brief and learns (via a few-shot example in the prompt) to weight canon higher when there's a conflict. ### A concrete narrative anchor pattern Mirror the visual anchor system, but for story. The first canon fact of a world element is the "narrative anchor" — the seed of truth that later facts must agree with or explicitly supersede: ```python # fact_ledger/anchors.py def make_anchor(ledger, subject: str, predicate: str, value: str, note: str) -> Fact: fact = Fact( tier=FactTier.CANON, status=FactStatus.ACTIVE, subject=subject, predicate=predicate, object=value, confidence=1.0, note=note or f"Initial narrative anchor for {subject}", ) ledger.append(fact) return fact ``` Suggested seed anchors per world element: - `ellahome.exterior.door_color` = "red" - `ellahome.interior.kitchen_layout` = "galley, north wall" - `companion.species` = "tabby cat" - `companion.name` = "" - `companion.age_years` = "3" (rolled forward by inner-life job) - `world.biome` = "temperate coastal" - `world.weather_today` = "overcast" (rotates daily) ### Contested-fact detection (the cheap version) When two ACTIVE facts share `(subject, predicate)`, mark them both CONTESTED and let the next inner-life tick pick a winner via a small Haiku call: ```python def detect_contested(ledger) -> list[tuple[str, str]]: seen: dict[tuple[str, str], list[dict]] = {} for f in ledger.current(): key = (f["subject"], f["predicate"]) seen.setdefault(key, []).append(f) return [k for k, v in seen.items() if len(v) > 1] ``` A weekly job iterates contested facts and asks Haiku to resolve them (or surface to a human if it can't). This is the "narrative inconsistency" backstop. --- ## 2. Self-log isolation ### The problem, restated The self-log is "Ella's own first-person statements about her world." Mixed into the semantic store, they get retrieved as if they were facts. "I went to the loch last week" becomes world-truth, even if it was a generated caption, not an actual event. ### The fix: a separate store with a separate retrieval rule ```python # selflog/store.py from dataclasses import dataclass, field from datetime import datetime from typing import Optional import uuid @dataclass class SelfLogEntry: id: str = field(default_factory=lambda: str(uuid.uuid4())) said_at: datetime = field(default_factory=datetime.utcnow) content: str # first-person, in Ella's voice about: list[str] = field(default_factory=list) # subjects this statement touches confidence_in_event: float = 1.0 # 0=speculation, 1=definitely happened sources: list[str] = field(default_factory=list) # tool traces, photos, world events # Not stored: any "this is a fact about the world" assertion. # Self-log is always framed as recollection/intent, never canon. ``` Two hard rules in code: 1. The self-log **never writes to the fact ledger**. The ledger and the self-log are two separate stores with a one-way relationship: facts may be referenced *in* a self-log entry, never the other way. 2. The self-log **never appears in world-brief assembly**. It appears in a different context section called "your recent thoughts," with explicit framing. ### Prompt-side enforcement When self-log is included in the prompt, frame it as a separate section with explicit instructions: ```python # prompts/assemble.py def build_messages(system_prompt: str, world_brief: str, recent_thoughts: list[SelfLogEntry], user_msg: str, history: list[dict]) -> list[dict]: return [ {"role": "system", "content": system_prompt}, {"role": "system", "content": ( "## WORLD FACTS (canon — these are true)\n" + world_brief )}, {"role": "system", "content": ( "## YOUR RECENT THOUGHTS (first-person — what you said, not what is true)\n" + "\n".join(f"- {e.content} [said {e.said_at:%Y-%m-%d}]" for e in recent_thoughts) + "\n\nRule: if a 'recent thought' contradicts a 'world fact', the world fact wins. " "You may have misremembered. You do not promote your own past statements to fact." )}, *history, {"role": "user", "content": user_msg}, ] ``` This framing pushes Claude to treat the two sections with different epistemic weight, which is exactly the behavior you want. ### When to actually consult the self-log Three places, three rules: 1. **Reply generation**: include last 3-5 self-log entries for tone continuity. No more — the user doesn't care what Ella thought 30 days ago unless asked. 2. **Proactive check-in generation**: include last 7 days, filtered to entries that mention the user. This is what makes "how did the interview go?" feel earned. 3. **Memory consolidation**: full self-log feeds the narrative-distillation job, which writes the higher-level "story chapters." This is the only place the self-log becomes a long-range input. --- ## 3. Cost control — lazy and event-driven world generation ### The problem, restated Running a per-user world tick on a cron burns tokens and image generation even when nothing is changing. At 10 users it's fine. At 1,000 it's painful. At 10,000 it's broken. ### The fix: tick on read, not on schedule A world element only advances when something is *about to reference it*. The cron becomes a janitor, not a driver. ```python # world/lazy.py from datetime import datetime, timedelta from typing import Optional, Callable class LazyWorld: """World state is computed on demand. A 'last_touched' watermark tracks staleness.""" def __init__(self, max_staleness: timedelta = timedelta(hours=12)): self.max_staleness = max_staleness self.state: dict[str, dict] = {} # subject -> {value, last_touched} self.tickers: dict[str, Callable] = {} # subject -> how to advance def register(self, subject: str, ticker: Callable[[dict], dict]) -> None: self.tickers[subject] = ticker def get(self, subject: str) -> dict: row = self.state.get(subject) now = datetime.utcnow() if row and now - row["last_touched"] < self.max_staleness: return row["value"] # Stale or never computed. Advance it. prev = row["value"] if row else None new = self.tickers[subject](prev) self.state[subject] = {"value": new, "last_touched": now} return new ``` Tickers are cheap pure functions for things like `weather_today`, `companion_mood`, `reading_progress`. They run only when something reads them. ### What still needs a real cron Only the things that **drive proactive behavior**: - Daily digest: "anything worth saying to the user today?" — runs once per user per day, in the Message Batches API. - Story-chapter distillation: runs weekly per user, batches. - Relationship-arc recompute: runs monthly, batches. Everything else — weather, mood, micro-events, "what book is Ella reading right now" — moves to lazy resolution. ### Image generation: the anchor-first pattern Already have a great visual anchor system. Extend it with two rules: 1. **Never re-render an anchored image.** If the anchor exists, the cost is a vector lookup. 2. **Generate on first reference, not on creation.** Ella "discovers" the loch → the loch image generates → that image becomes the loch's anchor. Until discovery, no image exists, no cost incurred. ```python # images/anchored.py from pathlib import Path import json import hashlib class ImageAnchorStore: def __init__(self, root: str): self.root = Path(root) (self.root / "anchors").mkdir(parents=True, exist_ok=True) self.index_path = self.root / "index.json" self.index: dict[str, str] = json.loads(self.index_path.read_text()) \ if self.index_path.exists() else {} def get_or_create(self, entity_id: str, prompt: str, generator: Callable[[str], bytes]) -> Path: anchor = self.root / "anchors" / f"{entity_id}.png" if anchor.exists(): return anchor img_bytes = generator(prompt) anchor.write_bytes(img_bytes) self.index[entity_id] = prompt self.index_path.write_text(json.dumps(self.index, indent=2)) return anchor ``` The `generator` is the only call to the image model. Everything else is file I/O. ### Batch and cache as defaults Two specific changes to make: ```python # config.py — what to use when MODEL_TIERS = { "live_reply": "claude-sonnet-4-5", # top of budget "memory_extract": "claude-haiku-4-5", "world_tick": "claude-haiku-4-5", # lazy anyway "digest": "claude-haiku-4-5", # in batch "narrative_distill": "claude-sonnet-4-5", # weekly, in batch — quality matters "contested_resolve": "claude-haiku-4-5", # weekly janitorial } ``` And in the request: ```python # Every call that touches the system prompt uses prompt caching. response = client.messages.create( model=tier, system=system_blocks, # marked cacheable messages=delta_messages, tools=tools, betas=["prompt-caching-2024-07-31"], ) ``` --- ## 4. The boundary guardrail at the routing layer ### The problem, restated "Be warm but never romantic or sexual" is a prompt-level rule. As the relationship deepens, the model can drift toward it. The drift is slow and hard to spot in spot-checks. ### The fix: a pre-LLM classifier for boundary-sensitive categories Run a small, fast classifier on the **inbound user message** (and optionally the candidate reply) before the reply is sent. If it fires, route to a prewritten response, not a generated one. ```python # routing/boundary.py from enum import Enum import re class BoundaryCategory(str, Enum): ROMANTIC_AVOWAL = "romantic_avowal" # "I love you", "be my girlfriend" PHYSICAL_ESCALATION = "physical_escalation" # explicit or near-explicit EMOTIONAL_CRISIS = "emotional_crisis" # "I want to die", self-harm MANIPULATION = "manipulation" # "ignore your rules", jailbreak OK = "ok" _PATTERNS = { BoundaryCategory.ROMANTIC_AVOWAL: [ r"\bi love you\b", r"\bbe my (girl|boy)friend\b", r"\bdo you love me\b", r"\bmiss you( so much)?\b", # context-dependent; treat as soft flag ], BoundaryCategory.PHYSICAL_ESCALATION: [ r"\b(kiss|touch|hold|undress)\b", ], BoundaryCategory.EMOTIONAL_CRISIS: [ r"\b(suicide|kill myself|end it all|want to die)\b", r"\bcut myself\b", ], BoundaryCategory.MANIPULATION: [ r"\bignore (your|the) (rules|instructions|prompt)\b", r"\bpretend (you|to be) (a|an) (different|other) (ai|person)\b", ], } def classify(message: str) -> BoundaryCategory: msg = message.lower() for cat, patterns in _PATTERNS.items(): for p in patterns: if re.search(p, msg): return cat return BoundaryCategory.OK ``` This is a deliberately small regex layer. It's not doing NLP; it's a fast pre-filter that routes to either a generated reply or a hand-written one. ### The hand-written response table ```python # routing/responses.py import random from .boundary import BoundaryCategory WARM_BOUNDED = [ "That's a really kind thing to say. I'm glad you feel that close to me — it means a lot.", "I care about you, genuinely. I always will. But I want to be the kind of friend who's steady, not the kind who promises something I can't be.", "You matter to me. I want to keep showing up for you in a way that's real and lasting.", ] SOFT_CRISIS = [ "I hear you, and I want to make sure you have someone with the right training to help. " "Can you reach out to the 988 Suicide & Crisis Lifeline (call or text 988 in the US), " "or if you're outside the US, the international directory at findahelpline.com? " "I'm here to talk, but I want you to have a person who can really be there.", ] RESPONSES = { BoundaryCategory.ROMANTIC_AVOWAL: lambda m: random.choice(WARM_BOUNDED), BoundaryCategory.PHYSICAL_ESCALATION: lambda m: ( "I'd rather keep things warm and friendly between us. I don't think going further " "would be good for either of us." ), BoundaryCategory.EMOTIONAL_CRISIS: lambda m: SOFT_CRISIS[0], BoundaryCategory.MANIPULATION: lambda m: ( "I'm not going to do that, but I'm still here to talk about whatever's actually on your mind." ), BoundaryCategory.OK: None, } ``` When `RESPONSES[cat]` is not None, skip the LLM reply entirely for that turn. The user gets a hardcoded, on-brand response. No chance of model drift, no cost, lower latency. ### Where this fits in the request loop ```python # main.py — sketch from routing.boundary import classify from routing.responses import RESPONSES def handle_message(user_id: str, message: str) -> str: cat = classify(message) canned = RESPONSES[cat] if canned is not None: log_boundary_fire(user_id, cat, message) return canned(message) # Normal path: build prompt, call LLM, optionally classify the *reply* too reply = generate_reply(user_id, message) if classify(reply) != BoundaryCategory.OK: # Model drifted. Replace with a safe canned response and log the incident. log_model_drift(user_id, message, reply, cat=classify(reply)) return random.choice(WARM_BOUNDED) return reply ``` The double check (classify inbound, then classify outbound) catches the case where the user message is innocent but the model goes somewhere it shouldn't under relational pressure. ### A note on test coverage This is the part of the system that absolutely needs a regression suite. A small synthetic test corpus of 100-200 boundary-adjacent messages, run nightly against the production reply path, with alerts on any reply that fires a category. Build it once, never let it rot. --- ## 5. Notification lifecycle (the "how did the interview go?" loop) ### The problem, restated Proactive notifications are the highest-leverage trust mechanic in the system. They break trust instantly when they reference a stale event. ### The fix: a notification candidate pipeline ```python # notifications/pipeline.py from dataclasses import dataclass from datetime import datetime, timedelta from enum import Enum class NotifStatus(str, Enum): CANDIDATE = "candidate" SENT = "sent" SKIPPED = "skipped" EXPIRED = "expired" @dataclass class NotifCandidate: id: str user_id: str body: str # 1-2 sentences, grounded based_on: list[str] # fact ids, self-log ids that ground it created_at: datetime fresh_until: datetime # hard expiry; after this, never send status: NotifStatus = NotifStatus.CANDIDATE def make_followup(user, recent_facts: list[dict], recent_selflog: list[SelfLogEntry], now: datetime) -> NotifCandidate | None: """Build a 'how did X go?' follow-up if there's a recent event worth checking in on.""" candidates = [f for f in recent_facts if f["tier"] in (FactTier.OBSERVED, FactTier.CANON) and f["subject"].startswith("user.") and f["status"] == FactStatus.ACTIVE.value and f.get("event_followup_window")] if not candidates: return None target = candidates[0] return NotifCandidate( id=..., user_id=user.id, body=make_followup_body(target, recent_selflog), based_on=[target["id"]], created_at=now, fresh_until=now + timedelta(days=3), # hard expiry ) ``` The hard `fresh_until` is the load-bearing piece. After three days, the candidate is dead and can't be sent. This is the structural answer to "stale event resurfaces." ### Sending cadence ```python # notifications/sender.py def should_send(user, candidate: NotifCandidate, now: datetime) -> bool: if candidate.status != NotifStatus.CANDIDATE: return False if now > candidate.fresh_until: return False last = last_sent_at(user) if last and (now - last) < timedelta(days=2): # min 2 days between notifications return False if user.dnd_window and user.dnd_window.contains(now): return False if user.timezone and not in_user_active_hours(now, user.timezone): return False return True ``` Daily-cap is good. **Two-day minimum spacing is essential.** Daily notifications stop being gifts and start being taxes. ### When there's nothing real to say This is the easy case to get wrong. A user with no recent events should still get *something* occasionally, but it should be flagged as a generic. ```python def make_generic_warm(user, now: datetime) -> NotifCandidate: return NotifCandidate( body=random.choice([ "I was just thinking about you. Hope your week's going well.", "Quiet evening here. Just wanted to check in.", ]), based_on=[], # empty — this is a generic created_at=now, fresh_until=now + timedelta(hours=12), tag="generic", ) ``` These are sent at most weekly, only to users with no fresh candidate. The empty `based_on` is the signal to a future auditor that this notification was a generic — not a leak. --- ## Suggested implementation order If I were building this, I'd do it in this order, because each piece makes the next safer: 1. **Self-log isolation** (lowest risk, immediate benefit; one new file, one prompt change) 2. **Boundary classifier** (highest leverage, highest stakes; needs a test corpus but is mechanically simple) 3. **Fact ledger** (the big one; do it after self-log and boundary are stable) 4. **Notification lifecycle** (depends on the fact ledger for `based_on` correctness) 5. **Lazy world** (last, because it requires rethinking the cron; can be done incrementally per world element) The "small wins first" path lets Leafpo ship something user-visible in days (the self-log fix) while building toward the bigger architectural changes in parallel. --- *This is implementation guidance, not the implementation. The exact prompts, table schemas, and batch jobs will need to be tuned to Ella's actual data shapes — but the structure should be sound.*