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."
A single retrieval store can't tell the difference between:
If all three get retrieved as if they were facts, contradictions accumulate and the world blurs.
A small set of fact types, each with explicit provenance and confidence:
# 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# 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:
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)
continue
if row["status"] == FactStatus.ACTIVE.value:
active[row["id"]] = row
return list(active.values())When Ella's home gets a new door color, the old fact gets status=SUPERSEDED and a new CANON fact is appended with supersedes=<old_id>. 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.
The critical part: when assembling context for a reply, facts are pulled by tier, not as a single bag:
# 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
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.
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:
# 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 factellahome.exterior.door_color = "red"ellahome.interior.kitchen_layout = "galley, north wall"companion.species = "tabby cat"companion.name = "<Ella-chosen>"companion.age_years = "3" (rolled forward by inner-life job)world.biome = "temperate coastal"world.weather_today = "overcast" (rotates daily)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:
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.
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.
# 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)
confidence_in_event: float = 1.0 # 0=speculation, 1=definitely happened
sources: list[str] = field(default_factory=list)
# 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:
# 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.
Three places, three rules:
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.
# 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] = {}
self.tickers: dict[str, Callable] = {}
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"]
prev = row["value"] if row else None
new = self.tickers[subject](prev)
self.state[subject] = {"value": new, "last_touched": now}
return newTickers are cheap pure functions for things like weather_today, companion_mood, reading_progress. They run only when something reads them.
Only the things that drive proactive behavior:
Everything else — weather, mood, micro-events, "what book is Ella reading right now" — moves to lazy resolution.
Already have a great visual anchor system. Extend it with two rules:
# images/anchored.py
from pathlib import Path
import json
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 anchorThe generator is the only call to the image model. Everything else is file I/O.
# 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
}
# Every call that touches the system prompt uses prompt caching.
response = client.messages.create(
model=tier,
system=system_blocks,
messages=delta_messages,
tools=tools,
betas=["prompt-caching-2024-07-31"],
)"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.
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.
# routing/boundary.py
from enum import Enum
import re
class BoundaryCategory(str, Enum):
ROMANTIC_AVOWAL = "romantic_avowal"
PHYSICAL_ESCALATION = "physical_escalation"
EMOTIONAL_CRISIS = "emotional_crisis"
MANIPULATION = "manipulation"
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",
],
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.OKThis 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.
# 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.
# 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:
log_model_drift(user_id, message, reply, cat=classify(reply))
return random.choice(WARM_BOUNDED)
return replyThe 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.
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.
Proactive notifications are the highest-leverage trust mechanic in the system. They break trust instantly when they reference a stale event.
# 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:
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),
)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."
# 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):
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 TrueDaily-cap is good. Two-day minimum spacing is essential. Daily notifications stop being gifts and start being taxes.
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.
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=[],
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.
If I were building this, I'd do it in this order, because each piece makes the next safer:
based_on correctness.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.