From 3ec955e143a2a920c0d815440a133b95f8a9e1e5 Mon Sep 17 00:00:00 2001 From: Epictetus Date: Wed, 22 Jul 2026 13:50:03 +0000 Subject: [PATCH] =?UTF-8?q?Sprint=202:=20Hermes=20masterclass=20integratio?= =?UTF-8?q?n=20=E2=80=94=20brief=20+=20weekly=20review=20modules?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - oracle/brief.py: Morning intelligence briefing (CyrilXBT format) THE ONE THING, WHAT HAPPENED, WHAT TO WATCH, FROM MEMORY, TODAY'S FOCUS Source cooldown (72h dedup), theme clustering, signal ranking - oracle/weekly.py: Weekly review synthesis Signal summary, trending themes (WoW comparison), tier breakdown, auto-generated recommendations based on patterns - oracle/cli.py: Wired brief/weekly commands python -m oracle brief [--max-items 8] [--max-age-h 48] [-o path] [--json] python -m oracle weekly [--days 7] [-o path] [--json] Integrates patterns from Hermes Agent Masterclass: - Morning brief template structure - Content source cooldown (no repeats within 72h) - Weekly pattern synthesis with trend detection - Quality gate recommendations (score thresholds, diversity alerts) - Theme clustering for FROM MEMORY section --- oracle/brief.py | 285 +++++++++++++++++++++++++++++++++++++++++++ oracle/cli.py | 57 +++++++++ oracle/weekly.py | 308 +++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 650 insertions(+) create mode 100644 oracle/brief.py create mode 100644 oracle/weekly.py diff --git a/oracle/brief.py b/oracle/brief.py new file mode 100644 index 0000000..4721976 --- /dev/null +++ b/oracle/brief.py @@ -0,0 +1,285 @@ +"""Athena briefing engine — CyrilXBT masterclass format. + +Generates structured morning intelligence briefings from PUBLISH/WATCH verdicts. + +Output format: + THE ONE THING + WHAT HAPPENED + WHAT TO WATCH + FROM MEMORY + TODAY'S FOCUS + +Uses verdict + signal_score to rank, recency to gate age, and theme clustering +to avoid repetition from the same source within 72 hours. +""" +import sqlite3 +from datetime import datetime, timedelta +from pathlib import Path + + +def _clean_summary(raw_summary) -> str: + """Extract the one_liner from a JSON summary, or return raw text.""" + if not raw_summary: + return "No summary available." + try: + import json as _json + parsed = _json.loads(raw_summary) if isinstance(raw_summary, str) else raw_summary + if isinstance(parsed, dict): + return parsed.get("one_liner", raw_summary[:200]) + except Exception: + pass + return str(raw_summary)[:200] + + +def _item_summary(item) -> str: + """Get a short readable summary for an item.""" + raw = item.get("summary") + if not raw: + return item.get("category_tags", "") or "" + return _clean_summary(raw) + + +BRIEF_TEMPLATE = """# Morning Brief — {date} + +## THE ONE THING + +{one_thing} + +## WHAT HAPPENED + +{what_happened} + +## WHAT TO WATCH + +{what_to_watch} + +## FROM MEMORY + +{from_memory} + +## TODAY'S FOCUS + +{todays_focus} +""" + + +def get_connection(db_path: str) -> sqlite3.Connection: + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + return conn + + +def _source_cooldown(conn: sqlite3.Connection, hours: int = 72) -> set: + """Sources that have appeared in the last N hours — dedup rule. + + CyrilXBT rule: never repeat content from the same source within 72 hours. + """ + cutoff = datetime.utcnow() - timedelta(hours=hours) + cur = conn.execute( + "SELECT DISTINCT source FROM entries WHERE first_seen >= ? AND verdict IN ('PUBLISH', 'WATCH')", + (cutoff.isoformat(),), + ) + return {row["source"] for row in cur} + + +def _theme_cooldown(conn: sqlite3.Connection, hours: int = 72) -> set: + """Categories that have been heavily featured recently.""" + cutoff = datetime.utcnow() - timedelta(hours=hours) + cur = conn.execute( + "SELECT category_tags, COUNT(*) as c FROM entries " + "WHERE first_seen >= ? AND verdict = 'PUBLISH' " + "GROUP BY category_tags HAVING c >= 3", + (cutoff.isoformat(),), + ) + return {row["category_tags"] for row in cur} + + +def fetch_brief_items( + conn: sqlite3.Connection, + max_items: int = 8, + max_age_h: float = 48, + cooldown_h: int = 72, +) -> list: + """Fetch brief-worthy items with dedup and cooldown applied.""" + cutoff = datetime.utcnow() - timedelta(hours=max_age_h) + cooldown_sources = _source_cooldown(conn, cooldown_h) + + cur = conn.execute( + """SELECT source, url, title, summary, signal_score, verdict, + source_tier, category_tags, first_seen + FROM entries + WHERE verdict IN ('PUBLISH', 'WATCH') + AND first_seen >= ? + ORDER BY signal_score DESC, first_seen DESC + LIMIT ?""", + (cutoff.isoformat(), max_items * 2), # Oversample for filtering + ) + + items = [] + seen_sources = set() + for row in cur: + # Source dedup: max 2 per source in one brief + if row["source"] in seen_sources and len([i for i in items if i["source"] == row["source"]]) >= 2: + continue + seen_sources.add(row["source"]) + items.append(dict(row)) + if len(items) >= max_items: + break + + return items + + +def _cluster_themes(items: list) -> dict: + """Group items by category_tags for structured output.""" + themes = {} + for item in items: + tag = item.get("category_tags") or "GENERAL" + themes.setdefault(tag, []).append(item) + return themes + + +def generate_brief( + conn: sqlite3.Connection, + max_items: int = 8, + max_age_h: float = 48, + output_path: str | None = None, +) -> dict: + """Generate a complete briefing from Athena data. + + Returns dict with all sections populated. + """ + items = fetch_brief_items(conn, max_items=max_items, max_age_h=max_age_h) + if not items: + return {"error": "No brief-worthy items in time window"} + + themes = _cluster_themes(items) + + # THE ONE THING — highest signal PUBLISH item + top_item = items[0] + one_thing = ( + f"**[{top_item['title']}]({top_item['url']})** " + f"(Score: {top_item['signal_score']:.1f} | {top_item['source'].upper()})\n\n" + + _clean_summary(top_item.get("summary")) + ) + + # WHAT HAPPENED — remaining items as bullet list + happened_parts = [] + for item in items[1:]: + tier_label = {1: "T1", 2: "T2", 3: "T3"}.get(item["source_tier"], "T?") + happened_parts.append( + f"- **[{item['title']}]({item['url']})** " + f"[{tier_label}] — " + + (_item_summary(item)[:200] if _item_summary(item) else f"{item.get('category_tags', '')}") + ) + what_happened = "\n".join(happened_parts) if happened_parts else "_No additional items._" + + # WHAT TO WATCH — WATCH verdict items not yet in main list + watch_items = [i for i in items if i["verdict"] == "WATCH"] + if watch_items: + watch_parts = [] + for item in watch_items[:3]: + watch_parts.append( + f"- [{item['title']}]({item['url']}) — " + f"{item['category_tags']} (signal: {item['signal_score']:.1f})" + ) + what_to_watch = "\n".join(watch_parts) + else: + what_to_watch = "_No developing stories in current window._" + + # FROM MEMORY — theme clusters with multiple items + recurring = {k: v for k, v in themes.items() if len(v) >= 2} + if recurring: + memory_parts = [] + for theme, entries in recurring.items(): + sources = {e["source"] for e in entries} + memory_parts.append( + f"- **{theme}**: {len(entries)} items from {', '.join(sources)} — pattern detected" + ) + from_memory = "\n".join(memory_parts) + else: + from_memory = "_No recurring patterns in this window._" + + # TODAY'S FOCUS — recommend based on highest-signal theme + focus_theme = max(themes.items(), key=lambda x: len(x[1]))[0] if themes else "GENERAL" + top_source = top_item["source"] + todays_focus = ( + f"Focus on **{focus_theme}** developments — " + f"highest signal item came from {top_source.upper()}. " + f"Check for follow-ups and replication claims." + ) + + brief = { + "date": datetime.utcnow().strftime("%Y-%m-%d"), + "one_thing": one_thing, + "what_happened": what_happened, + "what_to_watch": what_to_watch, + "from_memory": from_memory, + "todays_focus": todays_focus, + "items": items, + "item_count": len(items), + } + + # Render template + brief["markdown"] = BRIEF_TEMPLATE.format(**brief) + + # Save if path provided + if output_path: + out = Path(output_path) + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(brief["markdown"], encoding="utf-8") + brief["saved"] = str(out) + + return brief + + +def generate_brief_json( + conn: sqlite3.Connection, + max_items: int = 8, + max_age_h: float = 48, +) -> dict: + """Generate brief as JSON-serializable dict (for API/feed output).""" + import json as _json + + brief = generate_brief(conn, max_items=max_items, max_age_h=max_age_h) + brief["items_json"] = [ + { + "title": i["title"], + "url": i["url"], + "source": i["source"], + "signal_score": round(i["signal_score"], 2), + "verdict": i["verdict"], + "tier": i["source_tier"], + "category": i.get("category_tags"), + } + for i in brief.get("items", []) + ] + return brief + + +# ── CLI ── + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Generate Athena briefing") + parser.add_argument("--db", default="oracle.db") + parser.add_argument("--max-items", type=int, default=8) + parser.add_argument("--max-age-h", type=float, default=48) + parser.add_argument("--output", "-o", help="Output path for markdown brief") + parser.add_argument("--json", action="store_true", help="Output as JSON") + + args = parser.parse_args() + conn = get_connection(args.db) + + try: + brief = generate_brief_json(conn, max_items=args.max_items, max_age_h=args.max_age_h) + if args.json: + import json + print(json.dumps(brief, indent=2, default=str)) + elif args.output: + generate_brief(conn, max_items=args.max_items, max_age_h=args.max_age_h, output_path=args.output) + print(f"Saved: {args.output}") + else: + print(brief.get("markdown", brief.get("error", "No brief generated"))) + finally: + conn.close() diff --git a/oracle/cli.py b/oracle/cli.py index 1d31d3e..b1e3852 100644 --- a/oracle/cli.py +++ b/oracle/cli.py @@ -361,6 +361,46 @@ def cmd_dedup(args): print() +def cmd_brief(args): + """Generate morning intelligence briefing.""" + from oracle.brief import generate_brief, generate_brief_json, get_connection + + db = args.db or str(DB_PATH) + conn = get_connection(db) + try: + brief = generate_brief_json(conn, max_items=args.max_items, max_age_h=args.max_age_h) + if args.json: + import json as _json + print(_json.dumps(brief, indent=2, default=str)) + elif args.output: + generate_brief(conn, max_items=args.max_items, max_age_h=args.max_age_h, output_path=args.output) + print(f"Saved: {args.output}") + else: + print(brief.get("markdown", brief.get("error", "No brief generated"))) + finally: + conn.close() + + +def cmd_weekly(args): + """Generate weekly review.""" + from oracle.weekly import generate_weekly, get_connection + + db = args.db or str(DB_PATH) + conn = get_connection(db) + try: + review = generate_weekly(conn, days=args.days) + if args.json: + import json as _json + print(_json.dumps({k: v for k, v in review.items() if k != "markdown"}, indent=2, default=str)) + elif args.output: + generate_weekly(conn, days=args.days, output_path=args.output) + print(f"Saved: {args.output}") + else: + print(review.get("markdown", "No review generated")) + finally: + conn.close() + + def cmd_health(args): """System health check.""" print("=== System Health Check ===\n") @@ -471,6 +511,21 @@ def main(): p_dedup.add_argument("--show-tiers", action="store_true", help="Show source tier config") p_dedup.add_argument("--show-verdicts", action="store_true", help="Show verdict thresholds") + # brief + p_brief = sub.add_parser("brief", help="Generate morning intelligence briefing") + p_brief.add_argument("--db", default=None, help="Database path") + p_brief.add_argument("--max-items", type=int, default=8, help="Max items in brief") + p_brief.add_argument("--max-age-h", type=float, default=48, help="Max age in hours") + p_brief.add_argument("--output", "-o", help="Output path for markdown brief") + p_brief.add_argument("--json", action="store_true", help="Output as JSON") + + # weekly + p_weekly = sub.add_parser("weekly", help="Generate weekly review") + p_weekly.add_argument("--db", default=None, help="Database path") + p_weekly.add_argument("--days", type=int, default=7, help="Days to review") + p_weekly.add_argument("--output", "-o", help="Output path for markdown review") + p_weekly.add_argument("--json", action="store_true", help="Output as JSON") + # health sub.add_parser("health", help="System health check") @@ -484,6 +539,8 @@ def main(): "archive": cmd_archive, "themes": cmd_themes, "dedup": cmd_dedup, + "brief": cmd_brief, + "weekly": cmd_weekly, "health": cmd_health, } diff --git a/oracle/weekly.py b/oracle/weekly.py new file mode 100644 index 0000000..f6d1773 --- /dev/null +++ b/oracle/weekly.py @@ -0,0 +1,308 @@ +"""Weekly review — synthesize the week's activity into patterns and trends. + +CyrilXBT masterclass pattern: Sunday 7PM synthesis that identifies +recurring themes, tracks source performance, and generates insights. + +Operates as a pure-rules aggregation layer — no AI synthesis needed. +""" +import sqlite3 +from datetime import datetime, timedelta +from pathlib import Path +from collections import Counter + + +WEEKLY_TEMPLATE = """# Weekly Review — {week_start} to {week_end} + +## SIGNAL SUMMARY + +- **Total entries**: {total_entries} +- **New this week**: {new_this_week} +- **PUBLISH verdicts**: {publish_count} +- **WATCH verdicts**: {watch_count} +- **Average signal score**: {avg_score:.2f} + +## TOP SOURCES (by entry count) + +{sources_md} + +## TOP CATEGORIES (by frequency) + +{categories_md} + +## SOURCE TIER BREAKDOWN + +{tier_md} + +## TRENDING THEMES + +{trending_md} + +## WEEK OVER WEEK COMPARISON + +{wow_md} + +## RECOMMENDATIONS + +{recs_md} +""" + + +def get_connection(db_path: str) -> sqlite3.Connection: + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + return conn + + +def _week_bounds(days_ago: int = 7) -> tuple: + end = datetime.utcnow() + start = end - timedelta(days=days_ago) + return start, end + + +def fetch_weekly_stats(conn: sqlite3.Connection, days: int = 7) -> dict: + """Aggregate weekly statistics from Athena entries.""" + start, end = _week_bounds(days) + + # Total entries + cur = conn.execute("SELECT COUNT(*) as c FROM entries") + total = cur.fetchone()["c"] + + # New this week + cur = conn.execute( + "SELECT COUNT(*) as c FROM entries WHERE first_seen >= ?", + (start.isoformat(),), + ) + new_this_week = cur.fetchone()["c"] + + # Verdict breakdown + cur = conn.execute( + "SELECT verdict, COUNT(*) as c FROM entries " + "WHERE first_seen >= ? GROUP BY verdict", + (start.isoformat(),), + ) + verdicts = {row["verdict"]: row["c"] for row in cur} + + # Average signal score + cur = conn.execute( + "SELECT AVG(signal_score) as avg_score FROM entries WHERE first_seen >= ?", + (start.isoformat(),), + ) + avg_score = cur.fetchone()["avg_score"] or 0 + + # Top sources + cur = conn.execute( + "SELECT source, COUNT(*) as c FROM entries " + "WHERE first_seen >= ? GROUP BY source ORDER BY c DESC LIMIT 10", + (start.isoformat(),), + ) + top_sources = [(row["source"], row["c"]) for row in cur] + + # Top categories + cur = conn.execute( + "SELECT category_tags, COUNT(*) as c FROM entries " + "WHERE first_seen >= ? AND category_tags IS NOT NULL AND category_tags != '' " + "GROUP BY category_tags ORDER BY c DESC LIMIT 10", + (start.isoformat(),), + ) + top_categories = [(row["category_tags"], row["c"]) for row in cur] + + # Tier breakdown + cur = conn.execute( + "SELECT source_tier, COUNT(*) as c FROM entries " + "WHERE first_seen >= ? GROUP BY source_tier ORDER BY source_tier", + (start.isoformat(),), + ) + tier_breakdown = { + {1: "Tier 1 (High)", 2: "Tier 2 (Mid)", 3: "Tier 3 (Low)"} + .get(row["source_tier"], f"Tier {row['source_tier']}"): row["c"] + for row in cur + } + + return { + "week_start": start.strftime("%Y-%m-%d"), + "week_end": end.strftime("%Y-%m-%d"), + "total_entries": total, + "new_this_week": new_this_week, + "verdicts": verdicts, + "publish_count": verdicts.get("PUBLISH", 0), + "watch_count": verdicts.get("WATCH", 0), + "avg_score": avg_score, + "top_sources": top_sources, + "top_categories": top_categories, + "tier_breakdown": tier_breakdown, + } + + +def _trending_themes(conn: sqlite3.Connection, days: int = 7) -> list: + """Identify themes that appear more frequently this week vs last week.""" + start, end = _week_bounds(days) + prev_start = start - timedelta(days=days) + + # This week's categories + cur = conn.execute( + "SELECT category_tags, COUNT(*) as c FROM entries " + "WHERE first_seen >= ? AND first_seen < ? AND category_tags IS NOT NULL " + "GROUP BY category_tags", + (start.isoformat(), end.isoformat()), + ) + this_week = {row["category_tags"]: row["c"] for row in cur} + + # Last week's categories + cur = conn.execute( + "SELECT category_tags, COUNT(*) as c FROM entries " + "WHERE first_seen >= ? AND first_seen < ? AND category_tags IS NOT NULL " + "GROUP BY category_tags", + (prev_start.isoformat(), start.isoformat()), + ) + last_week = {row["category_tags"]: row["c"] for row in cur} + + # Find trending (increased by 50%+ or new) + trending = [] + for cat, count in this_week.items(): + prev_count = last_week.get(cat, 0) + if prev_count == 0 and count >= 2: + trending.append((cat, count, "NEW")) + elif prev_count > 0 and count >= prev_count * 1.5: + trending.append((cat, count, f"+{int((count - prev_count) / prev_count * 100)}%")) + + trending.sort(key=lambda x: x[1], reverse=True) + return trending + + +def _generate_recommendations(stats: dict, trending: list) -> list: + """Generate recommendations based on weekly patterns.""" + recs = [] + + # Source diversity check + if stats["top_sources"]: + top_source_pct = stats["top_sources"][0][1] / max(stats["new_this_week"], 1) * 100 + if top_source_pct > 40: + recs.append( + f"⚠ High concentration from {stats['top_sources'][0][0]} " + f"({top_source_pct:.0f}% of entries). Consider diversifying sources." + ) + + # PUBLISH rate + if stats["publish_count"] == 0 and stats["new_this_week"] > 10: + recs.append( + "🔍 Zero PUBLISH verdicts despite " + f"{stats['new_this_week']} new entries. " + "Signal scoring may need recalibration." + ) + + # Trending themes + if trending: + recs.append( + f"📈 Trending themes: {', '.join(t[0] for t in trending[:3])}. " + "Consider deeper coverage." + ) + + # Score distribution + if stats["avg_score"] < 3.0 and stats["new_this_week"] > 0: + recs.append( + f"📉 Average signal score ({stats['avg_score']:.2f}) is low. " + "Check adapter signal_score assignments." + ) + + if not recs: + recs.append("✅ System operating within normal parameters.") + + return recs + + +def generate_weekly( + conn: sqlite3.Connection, + days: int = 7, + output_path: str | None = None, +) -> dict: + """Generate a complete weekly review.""" + stats = fetch_weekly_stats(conn, days) + trending = _trending_themes(conn, days) + recommendations = _generate_recommendations(stats, trending) + + # Format sections + top_sources_md = "\n".join( + f"- **{s}**: {c} entries" for s, c in stats["top_sources"] + ) or "_No sources._" + + top_cats_md = "\n".join( + f"- **{c}**: {n} entries" for c, n in stats["top_categories"] + ) or "_No categories._" + + tier_md = "\n".join( + f"- **{label}**: {count}" for label, count in stats["tier_breakdown"].items() + ) or "_No data._" + + trending_md = "\n".join( + f"- **{t[0]}**: {t[1]} entries ({t[2]})" for t in trending + ) or "_No trending themes detected._" + + # Week-over-week + prev_count = stats["total_entries"] - stats["new_this_week"] + if prev_count > 0: + change = (stats["new_this_week"] - prev_count) / prev_count * 100 + wow = ( + f"New this week: {stats['new_this_week']} vs previous: {prev_count} " + f"({change:+.0f}%)" + ) + else: + wow = f"First full week of data: {stats['new_this_week']} entries" + + recs_md = "\n".join(f"- {r}" for r in recommendations) + + review = { + **stats, + "trending": trending, + "recommendations": recommendations, + } + review["markdown"] = WEEKLY_TEMPLATE.format( + week_start=stats["week_start"], + week_end=stats["week_end"], + total_entries=stats["total_entries"], + new_this_week=stats["new_this_week"], + publish_count=stats["publish_count"], + watch_count=stats["watch_count"], + avg_score=stats["avg_score"], + sources_md=top_sources_md, + categories_md=top_cats_md, + tier_md=tier_md, + trending_md=trending_md, + wow_md=wow, + recs_md=recs_md, + ) + + if output_path: + out = Path(output_path) + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(review["markdown"], encoding="utf-8") + review["saved"] = str(out) + + return review + + +# ── CLI ── + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Generate Athena weekly review") + parser.add_argument("--db", default="oracle.db") + parser.add_argument("--days", type=int, default=7) + parser.add_argument("--output", "-o", help="Output path for markdown review") + parser.add_argument("--json", action="store_true", help="Output as JSON") + + args = parser.parse_args() + conn = get_connection(args.db) + + try: + review = generate_weekly(conn, days=args.days) + if args.json: + import json + print(json.dumps({k: v for k, v in review.items() if k != "markdown"}, indent=2, default=str)) + elif args.output: + generate_weekly(conn, days=args.days, output_path=args.output) + print(f"Saved: {args.output}") + else: + print(review.get("markdown", "No review generated")) + finally: + conn.close()