fix: resolve issues found by post-merge review (PR #40) (#40)

* fix(events): split JSONL on newline only, not str.splitlines()

str.splitlines() also breaks on \r, \v, \f, NEL (U+0085) and Unicode line
separators U+2028/U+2029 -- all legal *unescaped* inside JSON strings. A
single event whose text (e.g. ASSISTANT_MESSAGE/TOOL_RESULT) carried one
was one physical line with no \n, but splitlines() shredded it, so replay()
raised a spurious 'corruption' error (whole session failed) and stream()
crashed. Split on \n only and rstrip \r for CRLF tolerance.

Test: a record containing NEL round-trips through replay without shredding.

* fix(events): fsync parent dir on new session file (durability)

With fsync=True, content was fsync'd but a newly-created session file's
directory entry wasn't durably linked, so a crash could lose a just-created
(content-fsync'd) file. fsync the parent dir once on file creation; guarded
for filesystems that don't support directory fsync.

* fix(config): drop dead HIVE_MAX_TURNS / HIVE_SESSION_TIMEOUT env mappings

These mapped to model.max_turns / model.session_timeout, which ModelConfig
does not define -- pydantic silently ignored them, so setting either env
var was a no-op. Removed the dead rows (the remaining mappings all target
real fields).

* fix(daemon): fresh identity in checkpoint + drop duplicate crisis line

- update_narrative reloads/saves its own identity copy, so the in-memory
  identity passed to checkpoint.save was stale (missing the just-appended
  entry/chapter). Reload before snapshotting.
- Skip the derived mood line when in crisis: the suffering fragment already
  emits the crisis directive, so 'overwhelmed' duplicated it in the prompt.

* fix(store): mark only the nudges just read as delivered (by id)

get_pending_nudges did SELECT (delivered=0) then UPDATE delivered=1 for the
whole agent, so a nudge inserted between the two statements (e.g. by another
process) was marked delivered without being returned -- silently lost. Now
it updates only the nudge_ids actually read. Test: a nudge added after a
read is still delivered exactly once on the next read.

* docs(changelog): cover mood, chaptered narrative, narrative-in-prompt + F1 in 0.5.4

The 0.5.4 entry listed only the durability work, but #34/#35/#38/#39 all
merged into this unreleased version. Add Added entries (mood model,
chaptered narrative, narrative-in-prompt, F1 coverage) and a Fixed section
for the review follow-ups. Mirrored in docs/changelog.md.

* test(store): exercise the actual nudge SELECT-vs-UPDATE race (Greptile)

The prior test only validated sequential delivery -- the old
WHERE delivered=0 sweep passed it too. New test commits a competing nudge
from a separate connection in the window between the SELECT and the UPDATE
(via a sync execute wrapper that preserves aiosqlite's await/async-with
duality). It fails against the old sweep (n2 marked delivered and lost) and
passes with the by-id UPDATE.

Author: chiruu12

CHANGELOG.md

@@ -2,17 +2,43 @@
 
 ## [0.5.4] — 2026-06-01
 
-Durability & dependency-hardening release. All changes are additive and
-backward compatible; existing databases upgrade automatically on first open.
+Durability, simulation, and hardening release. All changes are additive and
+backward compatible; existing databases and identities upgrade automatically.
 
 ### Added
+- **Derived mood model** — a `CircumplexMood` maps an agent's `happiness` and
+  `suffering` onto a valence/arousal circumplex and names the mood (content,
+  motivated, steady, restless, discouraged, anxious, overwhelmed). Pure/derived
+  (no persisted state), swappable via `MoodRegistry`, and surfaced in the
+  goal-pursuit prompt.
+- **Chaptered narrative** — instead of FIFO-dropping its oldest lines, an
+  agent's narrative seals into compact `Chapter` summaries (date span + entry
+  count + goal theme) on overflow, and the preamble shows a "Story so far"
+  section. `AgentIdentity.full_narrative()` exposes the complete history.
+- **Identity narrative in the runtime prompt** — the agent *pursuing* a goal
+  now sees its persistent self (name + accumulated narrative), not just the
+  goal-generation path.
 - **Event-log fsync durability (C5)** — `EventLog(fsync=True)` flushes and
   `os.fsync()`s every append so a power/OS crash cannot lose an acknowledged
   event. Gated by the new `event_log_fsync` config option (env
   `HIVE_EVENT_LOG_FSYNC`), default off to protect the hot heartbeat write path;
   the daemon honors it. Reads now tolerate a torn/partial last line.
 - **`HiveStore.delete_agent()`** — deletes an agent and, via the new cascade,
   all of its child rows in one call.
+- **Test coverage (F1)** — first tests for the CLI, the MCP server protocol,
+  and the structured logging writer/reader.
+
+### Fixed
+- **Event log no longer shreds records with Unicode line separators** —
+  `replay()`/`stream()` split on `\n` only (not `str.splitlines()`, which also
+  breaks on `\r`/`\f`/NEL/`U+2028`/`U+2029` — all legal unescaped in JSON), so
+  an event whose text carried one is no longer mis-parsed as corruption.
+- **Nudge delivery race** — `get_pending_nudges` marks only the nudges it
+  actually read as delivered, so one inserted concurrently isn't lost.
+- Dropped dead `HIVE_MAX_TURNS`/`HIVE_SESSION_TIMEOUT` env mappings (no backing
+  fields); checkpoints now snapshot the freshly-saved identity; the crisis
+  directive is no longer duplicated in the pursuit prompt; event-log fsync also
+  fsyncs the parent directory on file creation.
 
 ### Changed
 - **FK cascades (C3)** — every child table's foreign key to `agents`

docs/changelog.md

@@ -2,11 +2,20 @@
 
 ## [0.5.4] -- 2026-06-01
 
-Durability & dependency-hardening release. All changes are additive and backward compatible; existing databases upgrade automatically on first open.
+Durability, simulation, and hardening release. All changes are additive and backward compatible; existing databases and identities upgrade automatically.
 
 ### Added
+- **Derived mood model**: a `CircumplexMood` maps `happiness` + `suffering` onto a valence/arousal circumplex and names the mood (content/motivated/steady/restless/discouraged/anxious/overwhelmed). Pure/derived, swappable via `MoodRegistry`, surfaced in the goal-pursuit prompt.
+- **Chaptered narrative**: the narrative seals into compact `Chapter` summaries (date span + entry count + goal theme) on overflow instead of FIFO-dropping; the preamble shows a "Story so far" section; `AgentIdentity.full_narrative()` exposes the full history.
+- **Identity narrative in the runtime prompt**: the goal-*pursuing* agent now sees its persistent self (name + accumulated narrative), not just goal generation.
 - **Event-log fsync durability (C5)**: `EventLog(fsync=True)` flushes and `os.fsync()`s every append so a power/OS crash can't lose an acknowledged event. Gated by the `event_log_fsync` config option (env `HIVE_EVENT_LOG_FSYNC`), default off to protect the hot heartbeat write path; the daemon honors it. Reads tolerate a torn/partial last line.
 - **`HiveStore.delete_agent()`**: deletes an agent and, via cascade, all of its child rows in one call.
+- **Test coverage (F1)**: first tests for the CLI, the MCP server protocol, and the structured logging writer/reader.
+
+### Fixed
+- **Event log no longer shreds records with Unicode line separators**: `replay()`/`stream()` split on `\n` only (not `str.splitlines()`, which also breaks on `\r`/`\f`/NEL/`U+2028`/`U+2029`, all legal unescaped in JSON).
+- **Nudge delivery race**: `get_pending_nudges` marks only the nudges it actually read as delivered, so one inserted concurrently isn't lost.
+- Dropped dead `HIVE_MAX_TURNS`/`HIVE_SESSION_TIMEOUT` env mappings; checkpoints snapshot the freshly-saved identity; the crisis directive isn't duplicated in the pursuit prompt; event-log fsync also fsyncs the parent directory on file creation.
 
 ### Changed
 - **FK cascades (C3)**: every child table's foreign key to `agents` (`sessions`, `goals`, `nudges`, `schedules`, `sub_agents`, `tasks`, `alarms`) now declares `ON DELETE CASCADE`; a `user_version` 1->2 migration rebuilds existing tables to add it (data preserved). FK enforcement is opt-in per operation, so writing child rows for not-yet-persisted agents keeps working.

src/hive/config.py

@@ -188,8 +188,6 @@ def load(cls, hive_dir: Path | None = None) -> "HiveConfig":
             "HIVE_HEARTBEAT": ("daemon", "heartbeat", int),
             "HIVE_MAX_RETRIES": ("daemon", "max_retries", int),
             "HIVE_DEFAULT_MODEL": ("model", "default_model", str),
-            "HIVE_MAX_TURNS": ("model", "max_turns", int),
-            "HIVE_SESSION_TIMEOUT": ("model", "session_timeout", int),
             "HIVE_STARTING_BALANCE": ("economy", "starting_balance", float),
             "HIVE_PROFILES_DIR": ("profiles_dir", None, str),
             "HIVE_LOGS_DIR": ("logs_dir", None, str),

src/hive/daemon/loop.py

@@ -450,8 +450,10 @@ async def _run_agent_cycle_inner(self, agent: AgentState, suffering: SufferingSt
             # narrative/opinions -- which the persona/profile system prompt alone
             # doesn't carry. Same context channel the suffering fragment uses.
             # A derived mood (from happiness + suffering) colours the framing.
+            # Skip it in a crisis: the suffering fragment already states the
+            # crisis directive, so the "overwhelmed" mood line would duplicate it.
             mood_line = ""
-            if persona is not None:
+            if persona is not None and not suffering.in_crisis:
                 mood = MoodRegistry.default().derive(
                     persona.happiness, suffering.cumulative_load, suffering.in_crisis
                 )
@@ -510,6 +512,9 @@ async def _run_agent_cycle_inner(self, agent: AgentState, suffering: SufferingSt
                     metadata={"type": "goal_completed", "goal_id": active_goal["goal_id"]},
                 )
                 goals_snap = await self._store.list_agent_goals(agent.agent_id, limit=10)
+                # Reload so the snapshot reflects the narrative entry/chapter that
+                # update_narrative just wrote (it persists its own reloaded copy).
+                identity = self._identity.load(agent.agent_id) or identity
                 self._checkpoint.save(
                     agent.agent_id,
                     "goal_completed",

src/hive/memory/events.py

@@ -73,11 +73,23 @@ async def append(self, event: HiveEvent) -> None:
         await asyncio.to_thread(self._write_line, path, line)
 
     def _write_line(self, path: Path, line: str) -> None:
+        is_new = not path.exists()
         with open(path, "a") as f:
             f.write(line)
             if self._fsync:
                 f.flush()
                 os.fsync(f.fileno())
+        if self._fsync and is_new:
+            # fsync the parent dir so a freshly-created file's directory entry
+            # is durably linked (content fsync alone doesn't guarantee this).
+            try:
+                dir_fd = os.open(path.parent, os.O_RDONLY)
+                try:
+                    os.fsync(dir_fd)
+                finally:
+                    os.close(dir_fd)
+            except OSError:
+                pass  # some platforms/filesystems don't support directory fsync
 
     async def replay(self, agent_id: str, session_id: str) -> list[HiveEvent]:
         path = self._session_path(agent_id, session_id)
@@ -91,9 +103,15 @@ async def replay(self, agent_id: str, session_id: str) -> list[HiveEvent]:
         # final line on a newline-terminated file) must parse -- a failure there
         # is real corruption and is surfaced, not silently dropped.
         ends_clean = text.endswith("\n")
-        lines = text.splitlines()
+        # Split ONLY on "\n" -- str.splitlines() also breaks on \r, \v, \f, NEL
+        # and Unicode 
/
, which are legal (unescaped) inside JSON
+        # strings, so a single record carrying one in its text would be shredded.
+        lines = text.split("\n")
+        if ends_clean and lines and lines[-1] == "":
+            lines = lines[:-1]  # drop the empty element after a trailing newline
         events = []
-        for idx, line in enumerate(lines):
+        for idx, raw in enumerate(lines):
+            line = raw.rstrip("\r")  # tolerate CRLF
             if not line.strip():
                 continue
             try:
@@ -126,7 +144,10 @@ async def stream(self, agent_id: str) -> AsyncIterator[HiveEvent]:
             nl = chunk.rfind("\n")
             if nl != -1:
                 complete = chunk[: nl + 1]
-                for line in complete.splitlines():
+                # Split only on "\n" (see replay): splitlines() would shred a
+                # record containing a Unicode line separator in its text.
+                for raw in complete.split("\n"):
+                    line = raw.rstrip("\r")
                     if not line.strip():
                         continue
                     # These are complete (newline-terminated) lines, so a parse

src/hive/memory/store.py

@@ -593,16 +593,22 @@ async def get_pending_nudges(self, agent_id: str) -> list[str]:
         async with self._connect() as db:
             db.row_factory = aiosqlite.Row
             async with db.execute(
-                "SELECT message FROM nudges WHERE agent_id = ? AND delivered = 0",
+                "SELECT nudge_id, message FROM nudges WHERE agent_id = ? AND delivered = 0",
                 (agent_id,),
             ) as cursor:
-                messages = [row["message"] async for row in cursor]
-            await db.execute(
-                "UPDATE nudges SET delivered = 1 WHERE agent_id = ? AND delivered = 0",
-                (agent_id,),
-            )
-            await db.commit()
-            return messages
+                rows = [(row["nudge_id"], row["message"]) async for row in cursor]
+            if rows:
+                # Mark delivered only the rows just read, by id -- so a nudge
+                # inserted (e.g. by another process) between the SELECT and the
+                # UPDATE isn't marked delivered without ever being returned.
+                ids = [r[0] for r in rows]
+                placeholders = ",".join("?" * len(ids))
+                await db.execute(
+                    f"UPDATE nudges SET delivered = 1 WHERE nudge_id IN ({placeholders})",
+                    ids,
+                )
+                await db.commit()
+            return [r[1] for r in rows]
 
     async def save_schedule(
         self,

tests/memory/test_store.py

@@ -227,6 +227,78 @@ async def writer(i: int) -> None:
         # No exception == no "database is locked". Spot-check a couple landed.
         assert await store.get_pending_nudges("agent-0")
 
+    @pytest.mark.asyncio
+    async def test_get_pending_nudges_only_marks_returned(self, tmp_path: Path) -> None:
+        """A nudge added after a read isn't marked delivered, so it isn't lost."""
+        store = HiveStore(tmp_path / "state.db")
+        await store.initialize()
+        await store.save_nudge("n1", "a1", "first")
+        await store.save_nudge("n2", "a1", "second")
+
+        first = await store.get_pending_nudges("a1")
+        assert sorted(first) == ["first", "second"]
+
+        # A nudge that arrives later must still be delivered exactly once.
+        await store.save_nudge("n3", "a1", "third")
+        assert await store.get_pending_nudges("a1") == ["third"]
+        assert await store.get_pending_nudges("a1") == []
+
+    @pytest.mark.asyncio
+    async def test_get_pending_nudges_survives_concurrent_insert(self, tmp_path: Path) -> None:
+        """A nudge committed by another connection *between* the SELECT and the
+        UPDATE must not be marked delivered. This is the actual race the by-id
+        UPDATE fixes -- the old `WHERE delivered = 0` sweep would have lost it."""
+        from contextlib import asynccontextmanager
+
+        db_path = tmp_path / "state.db"
+        store = HiveStore(db_path)
+        await store.initialize()
+        await store.save_nudge("n1", "a1", "first")
+
+        real_connect = store._connect
+        state = {"injected": False}
+
+        @asynccontextmanager
+        async def connect_with_injection(foreign_keys: bool = False):  # type: ignore[no-untyped-def]
+            async with real_connect(foreign_keys=foreign_keys) as db:
+                real_execute = db.execute
+
+                # Sync wrapper preserves aiosqlite's `await`/`async with` duality.
+                # Just before the UPDATE runs (rows already SELECTed), a *separate*
+                # connection commits a new nudge -- the exact race window where the
+                # old `WHERE delivered = 0` sweep would have marked it delivered.
+                def execute(*args, **kwargs):  # type: ignore[no-untyped-def]
+                    sql = args[0] if args else kwargs.get("sql", "")
+                    if (
+                        not state["injected"]
+                        and sql.lstrip().upper().startswith("UPDATE")
+                        and "nudges" in sql
+                    ):
+                        state["injected"] = True
+
+                        async def _inject_then_update():  # type: ignore[no-untyped-def]
+                            async with aiosqlite.connect(db_path) as other:
+                                await other.execute(
+                                    "INSERT INTO nudges (nudge_id, agent_id, message, "
+                                    "delivered, created_at) VALUES ('n2','a1','second',0,'t')"
+                                )
+                                await other.commit()
+                            return await real_execute(*args, **kwargs)
+
+                        return _inject_then_update()
+                    return real_execute(*args, **kwargs)
+
+                db.execute = execute  # type: ignore[method-assign]
+                yield db
+
+        store._connect = connect_with_injection  # type: ignore[assignment,method-assign]
+        returned = await store.get_pending_nudges("a1")
+        store._connect = real_connect  # type: ignore[method-assign]
+
+        assert returned == ["first"]
+        # n2 raced in mid-call; it must survive as still-pending, not be lost.
+        assert await store.get_pending_nudges("a1") == ["second"]
+
 
 class TestCascades:
     @pytest.mark.asyncio

tests/test_events.py

@@ -123,6 +123,40 @@ async def _run():
     asyncio.run(_run())
 
 
+def test_replay_handles_unicode_line_separators(tmp_dir):
+    """A record whose text contains U+2028/U+2029/NEL must not be shredded.
+
+    These are legal (unescaped) inside JSON strings and have no '\\n', so the
+    record is one physical line -- but str.splitlines() would break it apart.
+    """
+    log = EventLog(tmp_dir)
+
+    async def _run():
+        tricky = "before
middle
after\x85end"
+        await log.append(
+            HiveEvent(
+                event_type=EventType.ASSISTANT_MESSAGE,
+                agent_id="agent-1",
+                session_id="sess-1",
+                data={"text": tricky},
+            )
+        )
+        await log.append(
+            HiveEvent(
+                event_type=EventType.TOOL_USED,
+                agent_id="agent-1",
+                session_id="sess-1",
+                data={"tool": "x"},
+            )
+        )
+        events = await log.replay("agent-1", "sess-1")
+        assert len(events) == 2  # not shredded into extra/broken lines
+        assert events[0].data["text"] == tricky
+        assert events[1].data["tool"] == "x"
+
+    asyncio.run(_run())
+
+
 def test_replay_raises_on_mid_log_corruption(tmp_dir):
     """A malformed line that is NOT the torn final line is real corruption -- surface it."""
     import pytest