feat(D3c): chaptered narrative (story arcs instead of FIFO trim)

docs/guide/persona.md

@@ -99,6 +99,15 @@ Each daemon cycle, `persona.apply_suffering_effects()` reads the agent's sufferi
 
 These changes are **not prompt text** -- they're actual parameter values that affect goal generation, tool selection, and decision-making.
 
+## Narrative & chapters
+
+Alongside the persona, each agent keeps an `AgentIdentity` with an evolving **narrative** --
+a dated log of goal outcomes. When the open narrative grows past its size limit it is
+**sealed into a chapter** (a compact summary with a date span and entry count) rather than
+dropping the oldest lines, so long-run history survives as a story arc. During goal pursuit
+the agent's prompt includes a "Story so far" section (recent chapter summaries) plus the
+current "Recent history", so the agent's own past informs its work.
+
 ## Checkpointing
 
 Persona state is included in checkpoints:

src/hive/agents/identity.py

@@ -64,6 +64,35 @@
 MAX_NARRATIVE = 800
 MAX_OPINIONS = 20
 MAX_QUESTIONS = 12
+MAX_CHAPTERS = 20
+
+
+def _entry_date(line: str) -> str:
+    """Extract the bracketed date prefix from a ``[date] ...`` narrative line."""
+    if line.startswith("[") and "]" in line:
+        return line[1 : line.index("]")]
+    return ""
+
+
+def _entry_goal(line: str) -> str:
+    """Extract the goal text from a ``[date] goal: outcome`` narrative line."""
+    body = line.split("] ", 1)[-1]
+    return body.rsplit(": ", 1)[0].strip()[:48]
+
+
+class Chapter(BaseModel):
+    """A sealed span of an agent's narrative.
+
+    When the open narrative grows past ``MAX_NARRATIVE`` it is sealed into a
+    Chapter (a compact summary) rather than FIFO-dropping its oldest lines, so
+    long-run history is preserved as a story arc instead of being lost.
+    """
+
+    index: int
+    summary: str
+    entry_count: int
+    started: str = ""
+    ended: str = ""
 
 
 class AgentIdentity(BaseModel):
@@ -72,11 +101,25 @@ class AgentIdentity(BaseModel):
     traits: list[str] = []
     domains: list[str] = []
     narrative: str = ""
+    chapters: list[Chapter] = Field(default_factory=list)
     worldview: str = ""
     opinions: list[dict[str, Any]] = Field(default_factory=list)
     open_questions: list[str] = Field(default_factory=list)
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
+    def full_narrative(self) -> str:
+        """The whole story: sealed chapter summaries + the current open narrative.
+
+        ``narrative`` alone holds only the current (unsealed) chapter, so callers
+        that want the agent's complete history (e.g. life summaries) must use this.
+        """
+        parts: list[str] = []
+        if self.chapters:
+            parts.append("\n".join(f"- {c.summary}" for c in self.chapters))
+        if self.narrative:
+            parts.append(self.narrative)
+        return "\n\n".join(parts)
+
 
 class IdentityManager:
     """Creates, loads, saves, and builds LLM preambles from agent identities."""
@@ -142,19 +185,66 @@ def save(self, identity: AgentIdentity) -> None:
         tmp.rename(path)
 
     def update_narrative(self, agent_id: str, goal_text: str, outcome: str) -> None:
-        """Append goal outcome to the agent's narrative."""
+        """Append a goal outcome to the agent's narrative.
+
+        When the open narrative would overflow ``MAX_NARRATIVE``, it is sealed
+        into a Chapter first (preserving the history as a summary) and the new
+        entry starts a fresh chapter -- rather than FIFO-dropping old lines.
+        """
         identity = self.load(agent_id)
         if not identity:
             return
-        entry = f"[{datetime.now(UTC).strftime('%m-%d')}] {goal_text}: {outcome}"
+        # Normalize newlines: a multi-line goal/outcome (e.g. LLM text) would
+        # otherwise split one entry across lines and break chapter sealing
+        # (entry_count / date / goal extraction all operate per-line).
+        goal_text = goal_text.replace("\n", " ").replace("\r", " ")
+        outcome = outcome.replace("\n", " ").replace("\r", " ")
+        # Full date (%Y-%m-%d) so chapter spans are unambiguous across year boundaries.
+        entry = f"[{datetime.now(UTC).strftime('%Y-%m-%d')}] {goal_text}: {outcome}"
+        # Cap a single pathological entry so the open narrative can never exceed
+        # MAX_NARRATIVE (a lone over-long entry would otherwise bypass sealing).
+        if len(entry) > MAX_NARRATIVE:
+            entry = entry[: MAX_NARRATIVE - 1] + "…"
+        if identity.narrative and len(identity.narrative) + len(entry) + 1 > MAX_NARRATIVE:
+            self._seal_chapter(identity)
         identity.narrative = (identity.narrative + "\n" + entry).strip()
-        if len(identity.narrative) > MAX_NARRATIVE:
-            lines = identity.narrative.splitlines()
-            while len(identity.narrative) > MAX_NARRATIVE and len(lines) > 1:
-                lines.pop(0)
-                identity.narrative = "\n".join(lines)
         self.save(identity)
 
+    @staticmethod
+    def _seal_chapter(identity: AgentIdentity) -> None:
+        """Roll the open narrative into a sealed Chapter and clear it."""
+        lines = [ln for ln in identity.narrative.splitlines() if ln.strip()]
+        if not lines:
+            identity.narrative = ""  # defensive: clear a whitespace-only narrative
+            return
+        started = _entry_date(lines[0])
+        ended = _entry_date(lines[-1])
+        index = identity.chapters[-1].index + 1 if identity.chapters else 1
+        if started and ended and started != ended:
+            span = f" ({started}–{ended})"
+        elif started:
+            span = f" ({started})"
+        else:
+            span = ""
+        # Carry goal text so the summary is semantically useful, not just a count:
+        # the first goal (theme) and, if different, the last (arc).
+        first_goal = _entry_goal(lines[0])
+        last_goal = _entry_goal(lines[-1])
+        theme = first_goal if first_goal == last_goal else f"{first_goal} → {last_goal}"
+        suffix = f" — {theme}" if theme else ""
+        identity.chapters.append(
+            Chapter(
+                index=index,
+                summary=f"Ch{index}{span}: {len(lines)} entries{suffix}",
+                entry_count=len(lines),
+                started=started,
+                ended=ended,
+            )
+        )
+        if len(identity.chapters) > MAX_CHAPTERS:
+            identity.chapters = identity.chapters[-MAX_CHAPTERS:]
+        identity.narrative = ""
+
     def add_opinion(self, agent_id: str, domain: str, opinion: str) -> None:
         """Record an opinion the agent has formed."""
         identity = self.load(agent_id)
@@ -203,6 +293,11 @@ def render_preamble(identity: AgentIdentity) -> str:
         if identity.domains:
             parts.append(f"Expertise: {', '.join(identity.domains)}")
 
+        if identity.chapters:
+            recent_chapters = identity.chapters[-5:]
+            chapter_lines = "\n".join(f"  - {c.summary}" for c in recent_chapters)
+            parts.append(f"\nStory so far:\n{chapter_lines}")
+
         if identity.narrative:
             recent = identity.narrative[-400:]
             parts.append(f"\nRecent history:\n{recent}")

src/hive/world/life_summary.py

@@ -114,7 +114,7 @@ def generate(
             peak_happiness_cycle=stats.cycles_alive,
             lowest_happiness=stats.happiness,
             lowest_happiness_cycle=0,
-            narrative=identity.narrative if identity else "",
+            narrative=identity.full_narrative() if identity else "",
         )
 
         return summary

tests/agents/test_identity.py

@@ -0,0 +1,175 @@
+"""Tests for AgentIdentity chaptered narrative (D3c)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from hive.agents.identity import (
+    MAX_CHAPTERS,
+    MAX_NARRATIVE,
+    AgentIdentity,
+    Chapter,
+    IdentityManager,
+)
+
+
+class TestChapterModel:
+    def test_chapters_default_empty(self) -> None:
+        ident = AgentIdentity(agent_id="a1", display_name="Atlas")
+        assert ident.chapters == []
+
+    def test_legacy_json_without_chapters_loads_empty(self) -> None:
+        """An identity serialized before chapters existed still loads."""
+        legacy = '{"agent_id": "a1", "display_name": "Atlas", "narrative": "old"}'
+        ident = AgentIdentity.model_validate_json(legacy)
+        assert ident.chapters == []
+        assert ident.narrative == "old"
+
+    def test_chapter_round_trip(self) -> None:
+        ident = AgentIdentity(
+            agent_id="a1",
+            display_name="Atlas",
+            chapters=[Chapter(index=1, summary="Ch1: 5 entries", entry_count=5)],
+        )
+        restored = AgentIdentity.model_validate_json(ident.model_dump_json())
+        assert len(restored.chapters) == 1
+        assert restored.chapters[0].index == 1
+        assert restored.chapters[0].entry_count == 5
+
+    def test_manager_persists_chapters(self, tmp_path: Path) -> None:
+        idm = IdentityManager(tmp_path)
+        ident = AgentIdentity(
+            agent_id="a1",
+            display_name="Atlas",
+            chapters=[Chapter(index=1, summary="Ch1", entry_count=3)],
+        )
+        idm.save(ident)
+        loaded = idm.load("a1")
+        assert loaded is not None
+        assert len(loaded.chapters) == 1
+
+
+class TestSealing:
+    def _idm_with_agent(self, tmp_path: Path) -> IdentityManager:
+        idm = IdentityManager(tmp_path)
+        idm.save(AgentIdentity(agent_id="a1", display_name="Atlas"))
+        return idm
+
+    def test_no_chapter_below_threshold(self, tmp_path: Path) -> None:
+        idm = self._idm_with_agent(tmp_path)
+        idm.update_narrative("a1", "small goal", "done")
+        ident = idm.load("a1")
+        assert ident is not None
+        assert ident.chapters == []
+        assert "small goal" in ident.narrative
+
+    def test_overflow_seals_chapter_and_preserves_count(self, tmp_path: Path) -> None:
+        idm = self._idm_with_agent(tmp_path)
+        # Many entries; each ~40 chars, so we cross MAX_NARRATIVE (800) and seal.
+        for i in range(40):
+            idm.update_narrative("a1", f"goal number {i}", "completed ok")
+        ident = idm.load("a1")
+        assert ident is not None
+        assert len(ident.chapters) >= 1, "no chapter sealed despite overflow"
+        # Total entries are conserved across sealed chapters + the open narrative.
+        sealed = sum(c.entry_count for c in ident.chapters)
+        open_lines = len([ln for ln in ident.narrative.splitlines() if ln.strip()])
+        assert sealed + open_lines == 40
+
+    def test_open_narrative_stays_bounded(self, tmp_path: Path) -> None:
+        idm = self._idm_with_agent(tmp_path)
+        for i in range(60):
+            idm.update_narrative("a1", f"goal {i}", "done")
+        ident = idm.load("a1")
+        assert ident is not None
+        # Strict bound: the open narrative never exceeds MAX_NARRATIVE.
+        assert len(ident.narrative) <= MAX_NARRATIVE
+
+    def test_single_oversized_entry_is_bounded(self, tmp_path: Path) -> None:
+        """A lone entry longer than MAX_NARRATIVE must not bypass the bound."""
+        idm = self._idm_with_agent(tmp_path)
+        idm.update_narrative("a1", "x" * (MAX_NARRATIVE * 2), "done")
+        ident = idm.load("a1")
+        assert ident is not None
+        assert len(ident.narrative) <= MAX_NARRATIVE
+
+    def test_chapter_indices_monotonic(self, tmp_path: Path) -> None:
+        idm = self._idm_with_agent(tmp_path)
+        for i in range(80):
+            idm.update_narrative("a1", f"goal {i}", "done")
+        ident = idm.load("a1")
+        assert ident is not None
+        indices = [c.index for c in ident.chapters]
+        assert indices == sorted(indices)
+        assert len(ident.chapters) <= MAX_CHAPTERS
+
+    def test_chapter_summary_carries_goal_text(self, tmp_path: Path) -> None:
+        """Summaries include goal text (theme/arc), not just a count + dates."""
+        idm = self._idm_with_agent(tmp_path)
+        for i in range(40):
+            idm.update_narrative("a1", f"objective-{i}", "done")
+        ident = idm.load("a1")
+        assert ident is not None and ident.chapters
+        # The first sealed chapter began with objective-0.
+        assert "objective-0" in ident.chapters[0].summary
+
+    def test_newlines_in_entry_are_normalized(self, tmp_path: Path) -> None:
+        """A multi-line goal/outcome must stay one narrative line (sealing is per-line)."""
+        idm = self._idm_with_agent(tmp_path)
+        idm.update_narrative("a1", "multi\nline\ngoal", "did\r\nthings")
+        ident = idm.load("a1")
+        assert ident is not None
+        assert len([ln for ln in ident.narrative.splitlines() if ln.strip()]) == 1
+        assert "multi line goal" in ident.narrative
+
+    def test_max_chapters_trim_drops_oldest(self, tmp_path: Path) -> None:
+        """Past MAX_CHAPTERS, the oldest chapter is dropped so history stays bounded."""
+        idm = self._idm_with_agent(tmp_path)
+        # Large outcomes => each entry nearly fills a chapter, so the next entry
+        # seals it; enough iterations to exceed the 20-chapter cap.
+        big = "x" * (MAX_NARRATIVE - 60)
+        for i in range(MAX_CHAPTERS + 6):
+            idm.update_narrative("a1", f"goal-{i}", big)
+        ident = idm.load("a1")
+        assert ident is not None
+        assert len(ident.chapters) == MAX_CHAPTERS  # capped; oldest dropped
+        assert ident.chapters[0].index > 1  # earliest chapters were trimmed
+        assert [c.index for c in ident.chapters] == sorted(c.index for c in ident.chapters)
+
+
+class TestFullNarrative:
+    def test_full_narrative_includes_chapters_and_open(self, tmp_path: Path) -> None:
+        idm = IdentityManager(tmp_path)
+        idm.save(AgentIdentity(agent_id="a1", display_name="Atlas"))
+        for i in range(40):
+            idm.update_narrative("a1", f"goal {i}", "done")
+        ident = idm.load("a1")
+        assert ident is not None and ident.chapters  # at least one sealed chapter
+        full = ident.full_narrative()
+        # Both the sealed chapter summary and a current open line are present.
+        assert ident.chapters[-1].summary in full
+        assert ident.narrative in full
+
+    def test_full_narrative_empty_identity(self) -> None:
+        assert AgentIdentity(agent_id="a1", display_name="Atlas").full_narrative() == ""
+
+
+class TestRenderPreamble:
+    def test_no_story_section_without_chapters(self, tmp_path: Path) -> None:
+        idm = IdentityManager(tmp_path)
+        idm.save(AgentIdentity(agent_id="a1", display_name="Atlas", narrative="[01-01] x: y"))
+        preamble = idm.build_preamble("a1")
+        assert "Story so far" not in preamble
+        assert "Recent history" in preamble
+
+    def test_chapters_render_as_story_so_far(self, tmp_path: Path) -> None:
+        idm = IdentityManager(tmp_path)
+        idm.save(AgentIdentity(agent_id="a1", display_name="Atlas"))
+        for i in range(40):
+            idm.update_narrative("a1", f"goal {i}", "done")
+        preamble = idm.build_preamble("a1")
+        assert "Story so far" in preamble
+        # The most recent sealed chapter's summary appears.
+        ident = idm.load("a1")
+        assert ident is not None and ident.chapters
+        assert ident.chapters[-1].summary in preamble