feat: emergent narrative from agent interactions (closes #1607)

narrative_engine.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+"""
+narrative_engine.py — Emergent narrative from agent interactions.
+
+Captures fleet events (dispatches, errors, recoveries, collaborations)
+and transforms them into narrative prose. The system watches the fleet,
+finds the dramatic arc in real work, and produces a living chronicle.
+
+Usage:
+    python3 narrative_engine.py --watch          # Watch and generate in real-time
+    python3 narrative_engine.py --generate        # Generate from recent events
+    python3 narrative_engine.py --output chronicle.md  # Write to file
+"""
+
+import argparse
+import json
+import os
+import subprocess
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+CHRONICLE_PATH = SCRIPT_DIR / "docs" / "chronicle.md"
+EVENTS_PATH = SCRIPT_DIR / "narrative-events.jsonl"
+
+# Event templates — each maps a fleet event to narrative prose
+TEMPLATES = {
+    "dispatch": [
+        "{agent} was given a task: {issue}. A problem to solve, a wound to close in the code.",
+        "The call went out to {agent}. Issue #{issue} — {title}. The work begins.",
+        "{agent} accepted the charge. {title}. Not for glory, but because the work needed doing.",
+    ],
+    "commit": [
+        "{agent} committed. {message}. The code remembers what the agent learned.",
+        "Lines changed. {agent} shaped something new from something broken.",
+        "{agent} pushed to {branch}. The work is done. The next task waits.",
+    ],
+    "pr_created": [
+        "A pull request emerged: {title}. The work is ready for review. Another step forward.",
+        "{agent} opened PR #{number}. The code speaks for itself now.",
+        "PR #{number}: {title}. The work stands on its own, waiting for eyes.",
+    ],
+    "pr_merged": [
+        "PR #{number} merged. The work is part of the world now.",
+        "It's in. {title}. Merged. The codebase grows, one fix at a time.",
+        "PR #{number} closed. The fix lives in main. The fleet moves on.",
+    ],
+    "error": [
+        "{agent} hit an error: {message}. Not every path is clear.",
+        "The build failed for {agent}. {message}. Errors are teachers, not judges.",
+        "Something broke in {agent}'s work. {message}. The repair will come.",
+    ],
+    "recovery": [
+        "{agent} recovered. After the failure, the fix. This is how systems learn.",
+        "The error passed. {agent} is working again. Resilience is not the absence of failure.",
+        "{agent} back online after {duration}. The dark interval is over.",
+    ],
+    "idle": [
+        "The fleet is quiet. No dispatches. The agents rest, or wait.",
+        "Silence in the burn lanes. All issues claimed. All panes dark or finished.",
+        "The work is done for now. The fleet waits for the next call.",
+    ],
+    "collaboration": [
+        "{agent1} and {agent2} touched the same code: {file}. Collision or collaboration?",
+        "Two agents, one file. {agent1} and {agent2} both worked on {file}.",
+        "{file} was changed by multiple agents. The code is a conversation.",
+    ],
+}
+
+
+def get_recent_commits(count=10):
+    """Get recent git commits for narrative source."""
+    try:
+        result = subprocess.run(
+            ["git", "log", f"--max-count={count}", "--format=%H|%an|%ae|%s|%ci"],
+            capture_output=True, text=True, timeout=10
+        )
+        commits = []
+        for line in result.stdout.strip().split("\n"):
+            if not line:
+                continue
+            parts = line.split("|", 4)
+            if len(parts) == 5:
+                commits.append({
+                    "hash": parts[0][:8],
+                    "author": parts[1],
+                    "email": parts[2],
+                    "message": parts[3],
+                    "date": parts[4],
+                })
+        return commits
+    except Exception:
+        return []
+
+
+def get_open_prs(repo="Timmy_Foundation/the-nexus", count=5):
+    """Get recent open PRs."""
+    try:
+        import urllib.request
+        token_path = Path.home() / ".config" / "gitea" / "token"
+        token = token_path.read_text().strip() if token_path.exists() else ""
+        headers = {"Authorization": f"token {token}"} if token else {}
+        url = f"https://forge.alexanderwhitestone.com/api/v1/repos/{repo}/pulls?state=open&limit={count}"
+        req = urllib.request.Request(url, headers=headers)
+        resp = urllib.request.urlopen(req, timeout=10)
+        return json.loads(resp.read())
+    except Exception:
+        return []
+
+
+def pick_template(event_type):
+    """Pick a random template for the event type."""
+    import random
+    templates = TEMPLATES.get(event_type, TEMPLATES["idle"])
+    return random.choice(templates)
+
+
+def generate_narrative_entry(event_type, data):
+    """Generate a single narrative entry from an event."""
+    template = pick_template(event_type)
+    try:
+        return template.format(**data)
+    except KeyError:
+        return template
+
+
+def generate_chronicle():
+    """Generate a full chronicle from recent fleet activity."""
+    now = datetime.now(timezone.utc)
+    lines = []
+    lines.append(f"# Fleet Chronicle")
+    lines.append(f"\n_Generated: {now.strftime('%Y-%m-%d %H:%M UTC')}_")
+    lines.append("")
+    lines.append("The story of the fleet, told from the data.")
+    lines.append("")
+
+    # Recent commits
+    commits = get_recent_commits(15)
+    if commits:
+        lines.append("## Recent Work")
+        lines.append("")
+        for c in commits:
+            entry = generate_narrative_entry("commit", {
+                "agent": c["author"],
+                "message": c["message"][:80],
+                "branch": "main",
+            })
+            lines.append(f"- {entry}")
+        lines.append("")
+
+    # Open PRs
+    prs = get_open_prs(count=5)
+    if prs:
+        lines.append("## Open Pull Requests")
+        lines.append("")
+        for pr in prs:
+            entry = generate_narrative_entry("pr_created", {
+                "agent": pr.get("user", {}).get("login", "unknown"),
+                "number": pr["number"],
+                "title": pr["title"][:60],
+            })
+            lines.append(f"- {entry}")
+        lines.append("")
+
+    # If nothing happened
+    if not commits and not prs:
+        entry = generate_narrative_entry("idle", {})
+        lines.append(f"> {entry}")
+        lines.append("")
+
+    lines.append("---")
+    lines.append("\n_The fleet writes its own story. We just read it._")
+
+    return "\n".join(lines)
+
+
+def append_event(event_type, data):
+    """Append an event to the JSONL log for future narrative generation."""
+    event = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "type": event_type,
+        "data": data,
+    }
+    with open(EVENTS_PATH, "a") as f:
+        f.write(json.dumps(event) + "\n")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Emergent narrative from agent interactions")
+    parser.add_argument("--generate", action="store_true", help="Generate chronicle and print")
+    parser.add_argument("--output", default=None, help="Write chronicle to file")
+    parser.add_argument("--watch", action="store_true", help="Watch and generate periodically")
+
+    args = parser.parse_args()
+
+    if args.generate or args.output:
+        chronicle = generate_chronicle()
+        if args.output:
+            Path(args.output).parent.mkdir(parents=True, exist_ok=True)
+            Path(args.output).write_text(chronicle)
+            print(f"Chronicle written to {args.output}")
+        else:
+            print(chronicle)
+    elif args.watch:
+        print("Watching for fleet events... (Ctrl+C to stop)")
+        while True:
+            time.sleep(60)
+            chronicle = generate_chronicle()
+            CHRONICLE_PATH.parent.mkdir(parents=True, exist_ok=True)
+            CHRONICLE_PATH.write_text(chronicle)
+            print(f"[{datetime.now().strftime('%H:%M')}] Chronicle updated")
+    else:
+        print("Use --generate, --output <file>, or --watch")
+
+
+if __name__ == "__main__":
+    main()