This commit was merged in pull request #791.
This commit is contained in:
@@ -228,6 +228,27 @@
|
||||
"max_items": 5
|
||||
},
|
||||
"outputs": []
|
||||
},
|
||||
{
|
||||
"id": "weekly_narrative",
|
||||
"name": "Weekly Narrative Summary",
|
||||
"description": "Generates a human-readable weekly summary of work themes, agent contributions, and token economy shifts",
|
||||
"script": "timmy_automations/daily_run/weekly_narrative.py",
|
||||
"category": "daily_run",
|
||||
"enabled": true,
|
||||
"trigger": "scheduled",
|
||||
"schedule": "weekly",
|
||||
"executable": "python3",
|
||||
"config": {
|
||||
"lookback_days": 7,
|
||||
"output_file": ".loop/weekly_narrative.json",
|
||||
"gitea_api": "http://localhost:3000/api/v1",
|
||||
"repo_slug": "rockachopa/Timmy-time-dashboard"
|
||||
},
|
||||
"outputs": [
|
||||
".loop/weekly_narrative.json",
|
||||
".loop/weekly_narrative.md"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -17,6 +17,10 @@
|
||||
"manual": {
|
||||
"description": "Run on-demand only",
|
||||
"automations": ["agent_workspace", "kimi_bootstrap", "kimi_resume", "backfill_retro"]
|
||||
},
|
||||
"weekly": {
|
||||
"description": "Run once per week (Sundays)",
|
||||
"automations": ["weekly_narrative"]
|
||||
}
|
||||
},
|
||||
"triggers": {
|
||||
|
||||
745
timmy_automations/daily_run/weekly_narrative.py
Normal file
745
timmy_automations/daily_run/weekly_narrative.py
Normal file
@@ -0,0 +1,745 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Weekly narrative summary generator — human-readable loop analysis.
|
||||
|
||||
Analyzes the past week's activity across the development loop to produce
|
||||
a narrative summary of:
|
||||
- What changed (themes, areas of focus)
|
||||
- How agents and Timmy contributed
|
||||
- Any shifts in tests, triage, or token economy
|
||||
|
||||
The output is designed to be skimmable — a quick read that gives context
|
||||
on the week's progress without drowning in metrics.
|
||||
|
||||
Run: python3 timmy_automations/daily_run/weekly_narrative.py [--json]
|
||||
Env: See timmy_automations/config/automations.json for configuration
|
||||
|
||||
Refs: #719
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from collections import Counter
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
# ── Configuration ─────────────────────────────────────────────────────────
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
|
||||
CONFIG_PATH = Path(__file__).parent.parent / "config" / "automations.json"
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
"gitea_api": "http://localhost:3000/api/v1",
|
||||
"repo_slug": "rockachopa/Timmy-time-dashboard",
|
||||
"token_file": "~/.hermes/gitea_token",
|
||||
"lookback_days": 7,
|
||||
"output_file": ".loop/weekly_narrative.json",
|
||||
"enabled": True,
|
||||
}
|
||||
|
||||
|
||||
# ── Data Loading ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def load_automation_config() -> dict:
|
||||
"""Load configuration for weekly_narrative from automations manifest."""
|
||||
config = DEFAULT_CONFIG.copy()
|
||||
if CONFIG_PATH.exists():
|
||||
try:
|
||||
manifest = json.loads(CONFIG_PATH.read_text())
|
||||
for auto in manifest.get("automations", []):
|
||||
if auto.get("id") == "weekly_narrative":
|
||||
config.update(auto.get("config", {}))
|
||||
config["enabled"] = auto.get("enabled", True)
|
||||
break
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
print(f"[weekly_narrative] Warning: Could not load config: {exc}", file=sys.stderr)
|
||||
|
||||
# Environment variable overrides
|
||||
if os.environ.get("TIMMY_GITEA_API"):
|
||||
config["gitea_api"] = os.environ.get("TIMMY_GITEA_API")
|
||||
if os.environ.get("TIMMY_REPO_SLUG"):
|
||||
config["repo_slug"] = os.environ.get("TIMMY_REPO_SLUG")
|
||||
if os.environ.get("TIMMY_GITEA_TOKEN"):
|
||||
config["token"] = os.environ.get("TIMMY_GITEA_TOKEN")
|
||||
if os.environ.get("TIMMY_WEEKLY_NARRATIVE_ENABLED"):
|
||||
config["enabled"] = os.environ.get("TIMMY_WEEKLY_NARRATIVE_ENABLED", "true").lower() == "true"
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def get_token(config: dict) -> str | None:
|
||||
"""Get Gitea token from environment or file."""
|
||||
if "token" in config:
|
||||
return config["token"]
|
||||
|
||||
token_file = Path(config["token_file"]).expanduser()
|
||||
if token_file.exists():
|
||||
return token_file.read_text().strip()
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def load_jsonl(path: Path) -> list[dict]:
|
||||
"""Load a JSONL file, skipping bad lines."""
|
||||
if not path.exists():
|
||||
return []
|
||||
entries = []
|
||||
for line in path.read_text().strip().splitlines():
|
||||
try:
|
||||
entries.append(json.loads(line))
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
continue
|
||||
return entries
|
||||
|
||||
|
||||
def parse_ts(ts_str: str) -> datetime | None:
|
||||
"""Parse an ISO timestamp, tolerating missing tz."""
|
||||
if not ts_str:
|
||||
return None
|
||||
try:
|
||||
dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=UTC)
|
||||
return dt
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
# ── Gitea API Client ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
"""Simple Gitea API client with graceful degradation."""
|
||||
|
||||
def __init__(self, config: dict, token: str | None):
|
||||
self.api_base = config["gitea_api"].rstrip("/")
|
||||
self.repo_slug = config["repo_slug"]
|
||||
self.token = token
|
||||
self._available: bool | None = None
|
||||
|
||||
def _headers(self) -> dict:
|
||||
headers = {"Accept": "application/json"}
|
||||
if self.token:
|
||||
headers["Authorization"] = f"token {self.token}"
|
||||
return headers
|
||||
|
||||
def _api_url(self, path: str) -> str:
|
||||
return f"{self.api_base}/repos/{self.repo_slug}/{path}"
|
||||
|
||||
def is_available(self) -> bool:
|
||||
"""Check if Gitea API is reachable."""
|
||||
if self._available is not None:
|
||||
return self._available
|
||||
|
||||
try:
|
||||
req = Request(
|
||||
f"{self.api_base}/version",
|
||||
headers=self._headers(),
|
||||
method="GET",
|
||||
)
|
||||
with urlopen(req, timeout=5) as resp:
|
||||
self._available = resp.status == 200
|
||||
return self._available
|
||||
except (HTTPError, URLError, TimeoutError):
|
||||
self._available = False
|
||||
return False
|
||||
|
||||
def get_paginated(self, path: str, params: dict | None = None) -> list:
|
||||
"""Fetch all pages of a paginated endpoint."""
|
||||
all_items = []
|
||||
page = 1
|
||||
limit = 50
|
||||
|
||||
while True:
|
||||
url = self._api_url(path)
|
||||
query_parts = [f"limit={limit}", f"page={page}"]
|
||||
if params:
|
||||
for key, val in params.items():
|
||||
query_parts.append(f"{key}={val}")
|
||||
url = f"{url}?{'&'.join(query_parts)}"
|
||||
|
||||
req = Request(url, headers=self._headers(), method="GET")
|
||||
with urlopen(req, timeout=15) as resp:
|
||||
batch = json.loads(resp.read())
|
||||
|
||||
if not batch:
|
||||
break
|
||||
|
||||
all_items.extend(batch)
|
||||
if len(batch) < limit:
|
||||
break
|
||||
page += 1
|
||||
|
||||
return all_items
|
||||
|
||||
|
||||
# ── Data Collection ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def collect_cycles_data(since: datetime) -> dict:
|
||||
"""Load cycle retrospective data from the lookback period."""
|
||||
cycles_file = REPO_ROOT / ".loop" / "retro" / "cycles.jsonl"
|
||||
if not cycles_file.exists():
|
||||
return {"cycles": [], "total": 0, "successes": 0, "failures": 0}
|
||||
|
||||
entries = load_jsonl(cycles_file)
|
||||
recent = []
|
||||
for e in entries:
|
||||
ts = parse_ts(e.get("timestamp", ""))
|
||||
if ts and ts >= since:
|
||||
recent.append(e)
|
||||
|
||||
successes = [e for e in recent if e.get("success")]
|
||||
failures = [e for e in recent if not e.get("success")]
|
||||
|
||||
return {
|
||||
"cycles": recent,
|
||||
"total": len(recent),
|
||||
"successes": len(successes),
|
||||
"failures": len(failures),
|
||||
"success_rate": round(len(successes) / len(recent), 2) if recent else 0,
|
||||
}
|
||||
|
||||
|
||||
def collect_issues_data(client: GiteaClient, since: datetime) -> dict:
|
||||
"""Collect issue activity from Gitea."""
|
||||
if not client.is_available():
|
||||
return {"error": "Gitea unavailable", "issues": [], "closed": [], "opened": []}
|
||||
|
||||
try:
|
||||
issues = client.get_paginated("issues", {"state": "all", "sort": "updated", "limit": 100})
|
||||
except (HTTPError, URLError) as exc:
|
||||
return {"error": str(exc), "issues": [], "closed": [], "opened": []}
|
||||
|
||||
touched = []
|
||||
closed = []
|
||||
opened = []
|
||||
|
||||
for issue in issues:
|
||||
updated_at = issue.get("updated_at", "")
|
||||
created_at = issue.get("created_at", "")
|
||||
|
||||
updated = parse_ts(updated_at)
|
||||
created = parse_ts(created_at)
|
||||
|
||||
if updated and updated >= since:
|
||||
touched.append(issue)
|
||||
|
||||
if issue.get("state") == "closed":
|
||||
closed_at = issue.get("closed_at", "")
|
||||
closed_dt = parse_ts(closed_at)
|
||||
if closed_dt and closed_dt >= since:
|
||||
closed.append(issue)
|
||||
elif created and created >= since:
|
||||
opened.append(issue)
|
||||
|
||||
return {
|
||||
"issues": touched,
|
||||
"closed": closed,
|
||||
"opened": opened,
|
||||
"touched_count": len(touched),
|
||||
"closed_count": len(closed),
|
||||
"opened_count": len(opened),
|
||||
}
|
||||
|
||||
|
||||
def collect_prs_data(client: GiteaClient, since: datetime) -> dict:
|
||||
"""Collect PR activity from Gitea."""
|
||||
if not client.is_available():
|
||||
return {"error": "Gitea unavailable", "prs": [], "merged": [], "opened": []}
|
||||
|
||||
try:
|
||||
prs = client.get_paginated("pulls", {"state": "all", "sort": "updated", "limit": 100})
|
||||
except (HTTPError, URLError) as exc:
|
||||
return {"error": str(exc), "prs": [], "merged": [], "opened": []}
|
||||
|
||||
touched = []
|
||||
merged = []
|
||||
opened = []
|
||||
|
||||
for pr in prs:
|
||||
updated_at = pr.get("updated_at", "")
|
||||
created_at = pr.get("created_at", "")
|
||||
merged_at = pr.get("merged_at", "")
|
||||
|
||||
updated = parse_ts(updated_at)
|
||||
created = parse_ts(created_at)
|
||||
merged_dt = parse_ts(merged_at) if merged_at else None
|
||||
|
||||
if updated and updated >= since:
|
||||
touched.append(pr)
|
||||
|
||||
if pr.get("merged") and merged_dt and merged_dt >= since:
|
||||
merged.append(pr)
|
||||
elif created and created >= since:
|
||||
opened.append(pr)
|
||||
|
||||
return {
|
||||
"prs": touched,
|
||||
"merged": merged,
|
||||
"opened": opened,
|
||||
"touched_count": len(touched),
|
||||
"merged_count": len(merged),
|
||||
"opened_count": len(opened),
|
||||
}
|
||||
|
||||
|
||||
def collect_triage_data(since: datetime) -> dict:
|
||||
"""Load triage and introspection data."""
|
||||
triage_file = REPO_ROOT / ".loop" / "retro" / "triage.jsonl"
|
||||
insights_file = REPO_ROOT / ".loop" / "retro" / "insights.json"
|
||||
|
||||
triage_entries = load_jsonl(triage_file)
|
||||
recent_triage = [
|
||||
e for e in triage_entries
|
||||
if parse_ts(e.get("timestamp", "")) and parse_ts(e.get("timestamp", "")) >= since
|
||||
]
|
||||
|
||||
insights = {}
|
||||
if insights_file.exists():
|
||||
try:
|
||||
insights = json.loads(insights_file.read_text())
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
return {
|
||||
"triage_runs": len(recent_triage),
|
||||
"triage_entries": recent_triage,
|
||||
"latest_insights": insights,
|
||||
}
|
||||
|
||||
|
||||
def collect_token_data(since: datetime) -> dict:
|
||||
"""Load token economy data from the lightning ledger."""
|
||||
# The ledger is in-memory but we can look for any persisted data
|
||||
# For now, return placeholder that will be filled by the ledger module
|
||||
return {
|
||||
"note": "Token economy data is ephemeral — check dashboard for live metrics",
|
||||
"balance_sats": 0, # Placeholder
|
||||
"transactions_week": 0,
|
||||
}
|
||||
|
||||
|
||||
# ── Analysis Functions ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def extract_themes(issues: list[dict]) -> list[dict]:
|
||||
"""Extract themes from issue labels."""
|
||||
label_counts = Counter()
|
||||
layer_counts = Counter()
|
||||
type_counts = Counter()
|
||||
|
||||
for issue in issues:
|
||||
for label in issue.get("labels", []):
|
||||
name = label.get("name", "")
|
||||
label_counts[name] += 1
|
||||
|
||||
if name.startswith("layer:"):
|
||||
layer_counts[name.replace("layer:", "")] += 1
|
||||
if name in ("bug", "feature", "refactor", "docs", "test", "chore"):
|
||||
type_counts[name] += 1
|
||||
|
||||
# Top themes (labels excluding layer prefixes)
|
||||
themes = [
|
||||
{"name": name, "count": count}
|
||||
for name, count in label_counts.most_common(10)
|
||||
if not name.startswith(("layer:", "size:"))
|
||||
]
|
||||
|
||||
# Layers
|
||||
layers = [
|
||||
{"name": name, "count": count}
|
||||
for name, count in layer_counts.most_common()
|
||||
]
|
||||
|
||||
# Types
|
||||
types = [
|
||||
{"name": name, "count": count}
|
||||
for name, count in type_counts.most_common()
|
||||
]
|
||||
|
||||
return {
|
||||
"top_labels": themes,
|
||||
"layers": layers,
|
||||
"types": types,
|
||||
}
|
||||
|
||||
|
||||
def extract_agent_contributions(issues: list[dict], prs: list[dict], cycles: list[dict]) -> dict:
|
||||
"""Extract agent contribution patterns."""
|
||||
# Count by assignee
|
||||
assignee_counts = Counter()
|
||||
for issue in issues:
|
||||
assignee = issue.get("assignee")
|
||||
if assignee and isinstance(assignee, dict):
|
||||
assignee_counts[assignee.get("login", "unknown")] += 1
|
||||
|
||||
# Count PR authors
|
||||
pr_authors = Counter()
|
||||
for pr in prs:
|
||||
user = pr.get("user")
|
||||
if user and isinstance(user, dict):
|
||||
pr_authors[user.get("login", "unknown")] += 1
|
||||
|
||||
# Check for Kimi mentions in cycle notes
|
||||
kimi_mentions = sum(
|
||||
1 for c in cycles
|
||||
if "kimi" in c.get("notes", "").lower() or "kimi" in c.get("reason", "").lower()
|
||||
)
|
||||
|
||||
return {
|
||||
"active_assignees": [
|
||||
{"login": login, "issues_count": count}
|
||||
for login, count in assignee_counts.most_common()
|
||||
],
|
||||
"pr_authors": [
|
||||
{"login": login, "prs_count": count}
|
||||
for login, count in pr_authors.most_common()
|
||||
],
|
||||
"kimi_mentioned_cycles": kimi_mentions,
|
||||
}
|
||||
|
||||
|
||||
def analyze_test_shifts(cycles: list[dict]) -> dict:
|
||||
"""Analyze shifts in test patterns."""
|
||||
if not cycles:
|
||||
return {"note": "No cycle data available"}
|
||||
|
||||
total_tests_passed = sum(c.get("tests_passed", 0) for c in cycles)
|
||||
total_tests_added = sum(c.get("tests_added", 0) for c in cycles)
|
||||
avg_tests_per_cycle = round(total_tests_passed / len(cycles), 1) if cycles else 0
|
||||
|
||||
# Look for test-related issues
|
||||
test_focused = [
|
||||
c for c in cycles
|
||||
if c.get("type") == "test" or "test" in c.get("notes", "").lower()
|
||||
]
|
||||
|
||||
return {
|
||||
"total_tests_passed": total_tests_passed,
|
||||
"total_tests_added": total_tests_added,
|
||||
"avg_tests_per_cycle": avg_tests_per_cycle,
|
||||
"test_focused_cycles": len(test_focused),
|
||||
}
|
||||
|
||||
|
||||
def analyze_triage_shifts(triage_data: dict) -> dict:
|
||||
"""Analyze shifts in triage patterns."""
|
||||
insights = triage_data.get("latest_insights", {})
|
||||
recommendations = insights.get("recommendations", [])
|
||||
|
||||
high_priority_recs = [
|
||||
r for r in recommendations
|
||||
if r.get("severity") == "high"
|
||||
]
|
||||
|
||||
return {
|
||||
"triage_runs": triage_data.get("triage_runs", 0),
|
||||
"insights_generated": insights.get("generated_at") is not None,
|
||||
"high_priority_recommendations": len(high_priority_recs),
|
||||
"recent_recommendations": recommendations[:3] if recommendations else [],
|
||||
}
|
||||
|
||||
|
||||
def generate_vibe_summary(
|
||||
cycles_data: dict,
|
||||
issues_data: dict,
|
||||
prs_data: dict,
|
||||
themes: dict,
|
||||
agent_contrib: dict,
|
||||
test_shifts: dict,
|
||||
triage_shifts: dict,
|
||||
) -> dict:
|
||||
"""Generate the human-readable 'vibe' summary."""
|
||||
# Determine overall vibe
|
||||
success_rate = cycles_data.get("success_rate", 0)
|
||||
failures = cycles_data.get("failures", 0)
|
||||
closed_count = issues_data.get("closed_count", 0)
|
||||
merged_count = prs_data.get("merged_count", 0)
|
||||
|
||||
if success_rate >= 0.9 and closed_count > 0:
|
||||
vibe = "productive"
|
||||
vibe_description = "A strong week with solid delivery and healthy success rates."
|
||||
elif success_rate >= 0.7:
|
||||
vibe = "steady"
|
||||
vibe_description = "Steady progress with some bumps. Things are moving forward."
|
||||
elif failures > cycles_data.get("successes", 0):
|
||||
vibe = "struggling"
|
||||
vibe_description = "A challenging week with more failures than successes. Time to regroup."
|
||||
else:
|
||||
vibe = "quiet"
|
||||
vibe_description = "A lighter week with limited activity."
|
||||
|
||||
# Focus areas from themes
|
||||
focus_areas = []
|
||||
for layer in themes.get("layers", [])[:3]:
|
||||
focus_areas.append(f"{layer['name']} ({layer['count']} items)")
|
||||
|
||||
# Agent activity summary
|
||||
agent_summary = ""
|
||||
active_assignees = agent_contrib.get("active_assignees", [])
|
||||
if active_assignees:
|
||||
top_agent = active_assignees[0]
|
||||
agent_summary = f"{top_agent['login']} led with {top_agent['issues_count']} assigned issues."
|
||||
|
||||
# Notable events
|
||||
notable = []
|
||||
if merged_count > 5:
|
||||
notable.append(f"{merged_count} PRs merged — high integration velocity")
|
||||
if triage_shifts.get("high_priority_recommendations", 0) > 0:
|
||||
notable.append("High-priority recommendations from loop introspection")
|
||||
if test_shifts.get("test_focused_cycles", 0) > 3:
|
||||
notable.append("Strong test coverage focus")
|
||||
if not notable:
|
||||
notable.append("Regular development flow")
|
||||
|
||||
return {
|
||||
"overall": vibe,
|
||||
"description": vibe_description,
|
||||
"focus_areas": focus_areas,
|
||||
"agent_summary": agent_summary,
|
||||
"notable_events": notable,
|
||||
}
|
||||
|
||||
|
||||
# ── Narrative Generation ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def generate_narrative(
|
||||
cycles_data: dict,
|
||||
issues_data: dict,
|
||||
prs_data: dict,
|
||||
triage_data: dict,
|
||||
themes: dict,
|
||||
agent_contrib: dict,
|
||||
test_shifts: dict,
|
||||
triage_shifts: dict,
|
||||
token_data: dict,
|
||||
since: datetime,
|
||||
until: datetime,
|
||||
) -> dict:
|
||||
"""Generate the complete weekly narrative."""
|
||||
vibe = generate_vibe_summary(
|
||||
cycles_data, issues_data, prs_data, themes, agent_contrib, test_shifts, triage_shifts
|
||||
)
|
||||
|
||||
return {
|
||||
"generated_at": datetime.now(UTC).isoformat(),
|
||||
"period": {
|
||||
"start": since.isoformat(),
|
||||
"end": until.isoformat(),
|
||||
"days": 7,
|
||||
},
|
||||
"vibe": vibe,
|
||||
"activity": {
|
||||
"cycles": {
|
||||
"total": cycles_data.get("total", 0),
|
||||
"successes": cycles_data.get("successes", 0),
|
||||
"failures": cycles_data.get("failures", 0),
|
||||
"success_rate": cycles_data.get("success_rate", 0),
|
||||
},
|
||||
"issues": {
|
||||
"touched": issues_data.get("touched_count", 0),
|
||||
"closed": issues_data.get("closed_count", 0),
|
||||
"opened": issues_data.get("opened_count", 0),
|
||||
},
|
||||
"pull_requests": {
|
||||
"touched": prs_data.get("touched_count", 0),
|
||||
"merged": prs_data.get("merged_count", 0),
|
||||
"opened": prs_data.get("opened_count", 0),
|
||||
},
|
||||
},
|
||||
"themes": themes,
|
||||
"agents": agent_contrib,
|
||||
"test_health": test_shifts,
|
||||
"triage_health": triage_shifts,
|
||||
"token_economy": token_data,
|
||||
}
|
||||
|
||||
|
||||
def generate_markdown_summary(narrative: dict) -> str:
|
||||
"""Generate a human-readable markdown summary."""
|
||||
vibe = narrative.get("vibe", {})
|
||||
activity = narrative.get("activity", {})
|
||||
cycles = activity.get("cycles", {})
|
||||
issues = activity.get("issues", {})
|
||||
prs = activity.get("pull_requests", {})
|
||||
|
||||
lines = [
|
||||
"# Weekly Narrative Summary",
|
||||
"",
|
||||
f"**Period:** {narrative['period']['start'][:10]} to {narrative['period']['end'][:10]}",
|
||||
f"**Vibe:** {vibe.get('overall', 'unknown').title()}",
|
||||
"",
|
||||
f"{vibe.get('description', '')}",
|
||||
"",
|
||||
"## Activity Highlights",
|
||||
"",
|
||||
f"- **Development Cycles:** {cycles.get('total', 0)} total ({cycles.get('successes', 0)} success, {cycles.get('failures', 0)} failure)",
|
||||
f"- **Issues:** {issues.get('closed', 0)} closed, {issues.get('opened', 0)} opened",
|
||||
f"- **Pull Requests:** {prs.get('merged', 0)} merged, {prs.get('opened', 0)} opened",
|
||||
"",
|
||||
]
|
||||
|
||||
# Focus areas
|
||||
focus = vibe.get("focus_areas", [])
|
||||
if focus:
|
||||
lines.append("## Focus Areas")
|
||||
lines.append("")
|
||||
for area in focus:
|
||||
lines.append(f"- {area}")
|
||||
lines.append("")
|
||||
|
||||
# Agent contributions
|
||||
agent_summary = vibe.get("agent_summary", "")
|
||||
if agent_summary:
|
||||
lines.append("## Agent Activity")
|
||||
lines.append("")
|
||||
lines.append(agent_summary)
|
||||
lines.append("")
|
||||
|
||||
# Notable events
|
||||
notable = vibe.get("notable_events", [])
|
||||
if notable:
|
||||
lines.append("## Notable Events")
|
||||
lines.append("")
|
||||
for event in notable:
|
||||
lines.append(f"- {event}")
|
||||
lines.append("")
|
||||
|
||||
# Triage health
|
||||
triage = narrative.get("triage_health", {})
|
||||
if triage.get("high_priority_recommendations", 0) > 0:
|
||||
lines.append("## Triage Notes")
|
||||
lines.append("")
|
||||
lines.append(f"⚠️ {triage['high_priority_recommendations']} high-priority recommendation(s) from loop introspection.")
|
||||
lines.append("")
|
||||
for rec in triage.get("recent_recommendations", [])[:2]:
|
||||
lines.append(f"- **{rec.get('category', 'general')}:** {rec.get('finding', '')}")
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ── Main ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
p = argparse.ArgumentParser(
|
||||
description="Generate weekly narrative summary of work and vibes",
|
||||
)
|
||||
p.add_argument(
|
||||
"--json", "-j",
|
||||
action="store_true",
|
||||
help="Output as JSON instead of markdown",
|
||||
)
|
||||
p.add_argument(
|
||||
"--output", "-o",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Output file path (default from config)",
|
||||
)
|
||||
p.add_argument(
|
||||
"--days",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Override lookback days (default 7)",
|
||||
)
|
||||
p.add_argument(
|
||||
"--force",
|
||||
action="store_true",
|
||||
help="Run even if disabled in config",
|
||||
)
|
||||
return p.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
config = load_automation_config()
|
||||
|
||||
# Check if enabled
|
||||
if not config.get("enabled", True) and not args.force:
|
||||
print("[weekly_narrative] Skipped — weekly narrative is disabled in config")
|
||||
print("[weekly_narrative] Use --force to run anyway")
|
||||
return 0
|
||||
|
||||
# Determine lookback period
|
||||
days = args.days if args.days is not None else config.get("lookback_days", 7)
|
||||
until = datetime.now(UTC)
|
||||
since = until - timedelta(days=days)
|
||||
|
||||
print(f"[weekly_narrative] Generating narrative for the past {days} days...")
|
||||
|
||||
# Setup Gitea client
|
||||
token = get_token(config)
|
||||
client = GiteaClient(config, token)
|
||||
|
||||
if not client.is_available():
|
||||
print("[weekly_narrative] Warning: Gitea API unavailable — will use local data only")
|
||||
|
||||
# Collect data
|
||||
cycles_data = collect_cycles_data(since)
|
||||
issues_data = collect_issues_data(client, since)
|
||||
prs_data = collect_prs_data(client, since)
|
||||
triage_data = collect_triage_data(since)
|
||||
token_data = collect_token_data(since)
|
||||
|
||||
# Analyze
|
||||
themes = extract_themes(issues_data.get("issues", []))
|
||||
agent_contrib = extract_agent_contributions(
|
||||
issues_data.get("issues", []),
|
||||
prs_data.get("prs", []),
|
||||
cycles_data.get("cycles", []),
|
||||
)
|
||||
test_shifts = analyze_test_shifts(cycles_data.get("cycles", []))
|
||||
triage_shifts = analyze_triage_shifts(triage_data)
|
||||
|
||||
# Generate narrative
|
||||
narrative = generate_narrative(
|
||||
cycles_data,
|
||||
issues_data,
|
||||
prs_data,
|
||||
triage_data,
|
||||
themes,
|
||||
agent_contrib,
|
||||
test_shifts,
|
||||
triage_shifts,
|
||||
token_data,
|
||||
since,
|
||||
until,
|
||||
)
|
||||
|
||||
# Determine output path
|
||||
output_path = args.output or config.get("output_file", ".loop/weekly_narrative.json")
|
||||
output_file = REPO_ROOT / output_path
|
||||
output_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Write JSON output
|
||||
output_file.write_text(json.dumps(narrative, indent=2) + "\n")
|
||||
|
||||
# Write markdown summary alongside JSON
|
||||
md_output_file = output_file.with_suffix(".md")
|
||||
md_output_file.write_text(generate_markdown_summary(narrative))
|
||||
|
||||
# Print output
|
||||
if args.json:
|
||||
print(json.dumps(narrative, indent=2))
|
||||
else:
|
||||
print()
|
||||
print(generate_markdown_summary(narrative))
|
||||
|
||||
print(f"\n[weekly_narrative] Written to: {output_file}")
|
||||
print(f"[weekly_narrative] Markdown summary: {md_output_file}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user