athena-oracle/theme_scan.py

#!/usr/bin/env python3
"""Phase 6 trend-tracking: theme-based arrival counter.

The question this answers: is the 2026-07-08 practitioner cluster a real TREND
or a one-day COINCIDENCE?

Design (per review):
  - Tag by THEME, not by entry ID.
  - Count NEW theme-tagged ARRIVALS per cron cycle (only classify rows that
    have no theme_tags yet -> fresh entries each run).
  - If new theme arrivals stay ~0 all week => coincidence.
  - If 2-4+ new relevant items/cycle across sources => trend.
    Only then does the ponytail-generalization idea graduate from a one-day
    read to something worth further investment.

Themes:
  tool-call   : tool-call gating on confidence / reliability of tool use
  context     : context / token compression before window ceiling
  compute     : routing to smallest sufficient model / inference cost
  trust       : trust boundaries on what a model may learn / vetted adapters

Run: python3 theme_scan.py            # classify + report new arrivals
     python3 theme_scan.py --history  # also print per-cycle history
"""

import argparse
import os
import re
import sqlite3
import sys
from collections import Counter

DB = os.path.join(os.path.dirname(__file__), "oracle.db")

# Theme -> regex over title+summary+extracted text (case-insensitive).
# Deliberately keyword-anchored, not semantic — cheap, auditable, reproducible.
THEME_PATTERNS = {
    "tool-call": re.compile(
        r"\b(tool[- ]?call|tool[- ]?use|competence gate|confidence gate|"
        r"gate[d]? tool|action gate|tool reliability|function call gate)\b",
        re.I),
    "context": re.compile(
        r"\b(context (compress|window|ceiling|summar)|semantic compress|"
        r"token (compress|budget)|compress (context|session)|context (limit|overflow))\b",
        re.I),
    "compute": re.compile(
        r"\b(small(er|est)? model|route to|inference cost|cpu (tts|infer)|"
        r"cheap(er)? model|model routing|tiny model|on[- ]device (llm|model))\b",
        re.I),
    "trust": re.compile(
        r"\b(trust(ed)? (adapter|lora)|vetted adapter|learn (only|what).*adapter|"
        r"trust boundary|what a model (can|may) learn|auditable (adapter|skill))\b",
        re.I),
}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--history", action="store_true",
                    help="Print per-cycle new-arrival history after the run")
    args = ap.parse_args()

    if not os.path.exists(DB):
        print("No oracle.db — nothing to scan")
        return

    conn = sqlite3.connect(DB)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    # Ensure theme_tags table exists (defensive; schema.sql creates it).
    cur.execute("""CREATE TABLE IF NOT EXISTS theme_tags (
        entry_id INTEGER NOT NULL,
        theme TEXT NOT NULL,
        first_seen_cycle TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')),
        PRIMARY KEY (entry_id, theme))""")

    # Only rows not yet classified -> fresh arrivals this cycle.
    cur.execute("""
        SELECT e.id, e.source, e.title,
               COALESCE(e.summary,'') AS summary,
               COALESCE(e.extracted_text,'') AS extracted
        FROM entries e
        WHERE e.id NOT IN (SELECT entry_id FROM theme_tags)
    """)
    fresh = cur.fetchall()

    new_counts = Counter()
    for row in fresh:
        blob = f"{row['title']} {row['summary']} {row['extracted']}"
        for theme, pat in THEME_PATTERNS.items():
            if pat.search(blob):
                cur.execute(
                    "INSERT OR IGNORE INTO theme_tags (entry_id, theme) VALUES (?, ?)",
                    (row["id"], theme))
                new_counts[theme] += 1

    conn.commit()

    print("=== Theme trend scan ===")
    print(f"  Fresh (unclassified) entries this cycle: {len(fresh)}")
    if new_counts:
        print("  NEW theme arrivals this cycle:")
        for theme in ("tool-call", "context", "compute", "trust"):
            if new_counts.get(theme):
                print(f"    {theme}: +{new_counts[theme]}")
    else:
        print("  NEW theme arrivals this cycle: 0")

    # Cumulative context for the trend question.
    cur.execute("SELECT theme, COUNT(*) AS c FROM theme_tags GROUP BY theme")
    cum = {r["theme"]: r["c"] for r in cur.fetchall()}
    print(f"  Cumulative theme_tags totals: {cum}")

    if args.history:
        print("\n  Per-cycle new arrivals (by first_seen_cycle):")
        cur.execute("""
            SELECT substr(first_seen_cycle,1,10) AS day, theme, COUNT(*) AS c
            FROM theme_tags GROUP BY day, theme ORDER BY day, theme
        """)
        for r in cur.fetchall():
            print(f"    {r['day']}  {r['theme']}: {r['c']}")

    conn.close()


if __name__ == "__main__":
    main()