Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions docs/guide/persona.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
103 changes: 96 additions & 7 deletions src/hive/agents/identity.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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."""
Expand Down Expand Up @@ -142,19 +185,60 @@ 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}"
# 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}"
Comment thread
chiruu12 marked this conversation as resolved.
# 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:
return
Comment thread
greptile-apps[bot] marked this conversation as resolved.
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)
Expand Down Expand Up @@ -203,6 +287,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}")
Expand Down
2 changes: 1 addition & 1 deletion src/hive/world/life_summary.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
152 changes: 152 additions & 0 deletions tests/agents/test_identity.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,152 @@
"""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")
Comment thread
greptile-apps[bot] marked this conversation as resolved.
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


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
Loading