""" Tests for nexus.chronicle — emergent narrative from agent interactions. """ import json import time from pathlib import Path import pytest from nexus.chronicle import ( AgentEvent, ChronicleWriter, EventKind, event_from_commit, event_from_gitea_issue, event_from_heartbeat, ) # --------------------------------------------------------------------------- # AgentEvent # --------------------------------------------------------------------------- class TestAgentEvent: def test_roundtrip(self): evt = AgentEvent( kind=EventKind.DISPATCH, agent="claude", detail="took issue #42", ) assert AgentEvent.from_dict(evt.to_dict()).kind == EventKind.DISPATCH assert AgentEvent.from_dict(evt.to_dict()).agent == "claude" assert AgentEvent.from_dict(evt.to_dict()).detail == "took issue #42" def test_default_timestamp_is_recent(self): before = time.time() evt = AgentEvent(kind=EventKind.IDLE, agent="mimo") after = time.time() assert before <= evt.timestamp <= after def test_all_event_kinds_are_valid_strings(self): for kind in EventKind: evt = AgentEvent(kind=kind, agent="test-agent") d = evt.to_dict() assert d["kind"] == kind.value restored = AgentEvent.from_dict(d) assert restored.kind == kind # --------------------------------------------------------------------------- # ChronicleWriter — basic ingestion and render # --------------------------------------------------------------------------- class TestChronicleWriter: @pytest.fixture def writer(self, tmp_path): return ChronicleWriter(log_path=tmp_path / "chronicle.jsonl") def test_empty_render(self, writer): text = writer.render() assert "empty" in text.lower() def test_single_event_render(self, writer): writer.ingest(AgentEvent(kind=EventKind.DISPATCH, agent="claude", detail="issue #1")) text = writer.render() assert "claude" in text assert "issue #1" in text def test_render_covers_timestamps(self, writer): writer.ingest(AgentEvent(kind=EventKind.DISPATCH, agent="a", detail="start")) writer.ingest(AgentEvent(kind=EventKind.COMMIT, agent="a", detail="done")) text = writer.render() assert "chronicle covers" in text.lower() def test_events_persisted_to_disk(self, writer, tmp_path): writer.ingest(AgentEvent(kind=EventKind.COMMIT, agent="claude", detail="feat: x")) lines = (tmp_path / "chronicle.jsonl").read_text().strip().splitlines() assert len(lines) == 1 data = json.loads(lines[0]) assert data["kind"] == "commit" assert data["agent"] == "claude" def test_load_existing_on_init(self, tmp_path): log = tmp_path / "chronicle.jsonl" evt = AgentEvent(kind=EventKind.PUSH, agent="mimo", detail="pushed branch") log.write_text(json.dumps(evt.to_dict()) + "\n") writer2 = ChronicleWriter(log_path=log) assert len(writer2._events) == 1 assert writer2._events[0].kind == EventKind.PUSH def test_malformed_lines_are_skipped(self, tmp_path): log = tmp_path / "chronicle.jsonl" log.write_text("not-json\n{}\n") # Should not raise writer2 = ChronicleWriter(log_path=log) assert writer2._events == [] def test_template_rotation(self, writer): """Consecutive events of the same kind use different templates.""" sentences = set() for _ in range(3): writer.ingest(AgentEvent(kind=EventKind.HEARTBEAT, agent="claude")) text = writer.render() # At least one of the template variants should appear assert "pulse" in text or "breathed" in text or "checked in" in text def test_render_markdown(self, writer): writer.ingest(AgentEvent(kind=EventKind.PR_OPEN, agent="claude", detail="PR #99")) md = writer.render_markdown() assert md.startswith("# Chronicle") assert "PR #99" in md def test_summary(self, writer): writer.ingest(AgentEvent(kind=EventKind.DISPATCH, agent="claude", detail="x")) writer.ingest(AgentEvent(kind=EventKind.COMMIT, agent="claude", detail="y")) s = writer.summary() assert s["total_events"] == 2 assert "claude" in s["agents"] assert s["kind_counts"]["dispatch"] == 1 assert s["kind_counts"]["commit"] == 1 def test_max_events_limit(self, writer): for i in range(10): writer.ingest(AgentEvent(kind=EventKind.IDLE, agent="a", detail=str(i))) text = writer.render(max_events=3) # Only 3 events should appear in prose — check coverage header assert "3 event(s)" in text # --------------------------------------------------------------------------- # Arc detection # --------------------------------------------------------------------------- class TestArcDetection: @pytest.fixture def writer(self, tmp_path): return ChronicleWriter(log_path=tmp_path / "chronicle.jsonl") def _ingest(self, writer, *kinds, agent="claude"): for k in kinds: writer.ingest(AgentEvent(kind=k, agent=agent, detail="x")) def test_struggle_and_recovery_arc(self, writer): self._ingest(writer, EventKind.DISPATCH, EventKind.ERROR, EventKind.RECOVERY) text = writer.render() assert "struggle" in text.lower() or "trouble" in text.lower() def test_no_arc_when_no_pattern(self, writer): self._ingest(writer, EventKind.IDLE) text = writer.render() # Should not include arc language (only 1 event, no pattern) assert "converged" not in text assert "struggle" not in text def test_solo_sprint_arc(self, writer): self._ingest( writer, EventKind.DISPATCH, EventKind.COMMIT, EventKind.PR_OPEN, EventKind.PR_MERGE, ) text = writer.render() assert "solo" in text.lower() or "alone" in text.lower() def test_fleet_convergence_arc(self, writer, tmp_path): writer2 = ChronicleWriter(log_path=tmp_path / "chronicle.jsonl") writer2.ingest(AgentEvent(kind=EventKind.DISPATCH, agent="claude", detail="x")) writer2.ingest(AgentEvent(kind=EventKind.COLLABORATION, agent="mimo", detail="x")) writer2.ingest(AgentEvent(kind=EventKind.COMMIT, agent="claude", detail="x")) text = writer2.render() assert "converged" in text.lower() or "fleet" in text.lower() def test_silent_grind_arc(self, writer): self._ingest(writer, EventKind.COMMIT, EventKind.COMMIT, EventKind.COMMIT) text = writer.render() assert "steady" in text.lower() or "quiet" in text.lower() or "grind" in text.lower() def test_abandon_then_retry_arc(self, writer): self._ingest(writer, EventKind.DISPATCH, EventKind.ABANDON, EventKind.DISPATCH) text = writer.render() assert "let go" in text.lower() or "abandon" in text.lower() or "called again" in text.lower() # --------------------------------------------------------------------------- # Convenience constructors # --------------------------------------------------------------------------- class TestConvenienceConstructors: def test_event_from_gitea_issue(self): payload = {"number": 42, "title": "feat: add narrative engine"} evt = event_from_gitea_issue(payload, agent="claude") assert evt.kind == EventKind.DISPATCH assert "42" in evt.detail assert evt.agent == "claude" def test_event_from_heartbeat(self): hb = {"model": "claude-sonnet", "status": "thinking", "cycle": 7} evt = event_from_heartbeat(hb) assert evt.kind == EventKind.HEARTBEAT assert evt.agent == "claude-sonnet" assert "7" in evt.detail def test_event_from_commit(self): commit = {"message": "feat: chronicle\n\nFixes #1607", "sha": "abc1234567"} evt = event_from_commit(commit, agent="claude") assert evt.kind == EventKind.COMMIT assert evt.detail == "feat: chronicle" # subject line only assert evt.metadata["sha"] == "abc12345"