"""AI Pulse watcher — polls AI Pulse for new items in the "AI Adoption focus" tab and creates a stub launch in the AI Tool Rollout Agent for each one. Data contract (what AI Pulse must expose): GET https://ai-pulse.khalids-mac.com/ai-adoption-focus.json → JSON array of items, each with at minimum: { "id": "stable-unique-id", "title": "Tool or feature name", "source": "where it came from (news, partner, X.com, etc.)", "published": "YYYY-MM-DD", "url": "link back to the source" } On detection of a new id (not in watcher/seen.json): 1. Create launches//state.json with Stage 1 in_progress. 2. Append the id to seen.json. 3. Send a macOS desktop notification. 4. Write an entry to watcher/activity.log. Run manually: python3 watcher/ai_pulse_watcher.py [--once] Run as launchd service: see watcher/com.dewa.aipulse.watcher.plist and watcher/install.sh """ from __future__ import annotations import argparse import json import re import subprocess import sys import time import urllib.error import urllib.request from datetime import datetime from pathlib import Path FEED_URL = "https://ai-pulse.khalids-mac.com/ai-adoption-focus.json" POLL_INTERVAL_SECONDS = 300 # 5 minutes ROOT = Path(__file__).resolve().parent.parent LAUNCHES_DIR = ROOT / "launches" SEEN_FILE = ROOT / "watcher" / "seen.json" LOG_FILE = ROOT / "watcher" / "activity.log" def slugify(name: str) -> str: s = name.lower().strip() s = re.sub(r"[^a-z0-9]+", "-", s) s = re.sub(r"-+", "-", s).strip("-") return s or "untitled" def load_seen() -> set[str]: if SEEN_FILE.exists(): try: return set(json.loads(SEEN_FILE.read_text(encoding="utf-8"))) except json.JSONDecodeError: return set() return set() def save_seen(seen: set[str]) -> None: SEEN_FILE.parent.mkdir(exist_ok=True) SEEN_FILE.write_text(json.dumps(sorted(seen), indent=2), encoding="utf-8") def log(msg: str) -> None: LOG_FILE.parent.mkdir(exist_ok=True) stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with LOG_FILE.open("a", encoding="utf-8") as f: f.write(f"[{stamp}] {msg}\n") print(f"[{stamp}] {msg}") def notify(title: str, body: str) -> None: """macOS desktop notification via osascript. Silent failure on other OSes.""" if sys.platform != "darwin": return safe_title = title.replace('"', '\\"') safe_body = body.replace('"', '\\"') script = f'display notification "{safe_body}" with title "{safe_title}"' try: subprocess.run(["osascript", "-e", script], check=False, timeout=5) except Exception: pass def fetch_items() -> list[dict] | None: req = urllib.request.Request( FEED_URL, headers={ "Accept": "application/json", "User-Agent": "DEWA-AI-Pulse-Watcher/1.0", }, ) try: with urllib.request.urlopen(req, timeout=15) as resp: data = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: log(f"HTTP error fetching feed: {e.code} {e.reason}") return None except urllib.error.URLError as e: log(f"Network error fetching feed: {e.reason}") return None except json.JSONDecodeError as e: log(f"Feed did not return valid JSON: {e}") return None if not isinstance(data, list): log("Feed returned non-list payload; expected a JSON array.") return None return data def build_state(item: dict, slug: str) -> dict: today = datetime.now().strftime("%Y-%m-%d") return { "tool_name": item.get("title", slug), "slug": slug, "created": today, "trigger_source": f"AI Pulse — {item.get('source', 'AI Adoption focus')}", "urgency": "High — auto-triggered from AI Pulse AI Adoption focus feed", "current_stage": 1, "overall_status": "active", "schema_version": "9-stage-v1", "auto_created_by": "ai_pulse_watcher", "source_metadata": { "ai_pulse_id": item.get("id"), "source_url": item.get("url"), "published": item.get("published"), }, "stages": { "1_feature_evaluation": {"owner": "ai-acceleration", "status": "in_progress", "decision": None, "artifact": None, "completed_date": None}, "2_isrp_submission": {"owner": "ai-acceleration", "status": "pending", "artifact": None, "completed_date": None}, "3_security_review": {"owner": "it-security", "status": "pending", "decision": None, "artifact": None, "completed_date": None}, "4_test_enablement": {"owner": "product-owner", "status": "pending", "artifact": None, "completed_date": None}, "5_feature_testing": {"owner": "ai-acceleration", "status": "pending", "decision": None, "artifact": None, "completed_date": None, "troubleshooting_rounds": 0}, "6_enterprise_enablement": {"owner": "product-owner", "status": "pending", "artifact": None, "completed_date": None}, "7_adoption_plan": {"owner": "ai-acceleration", "status": "pending", "artifact": None, "completed_date": None}, "8_training_comms": {"owner": "ai-adoption-office", "status": "pending", "sub_stages": {"training_materials": "pending", "internal_comms": "pending", "enablement_sessions": "pending"}, "artifacts": [], "completed_date": None}, "9_kpi_tracking": {"owner": "aicc", "status": "pending", "decision": None, "artifact": None, "completed_date": None, "loop_rounds": 0}, }, "decisions": [], "backlog_reason": None, "notes": [ f"Auto-created by AI Pulse watcher on {today}.", f"Source: {item.get('source', 'AI Adoption focus')}", f"Source URL: {item.get('url', 'not provided')}", "Open Claude Code in this project and run /resume " + slug + " to continue Stage 1 with ai-acceleration.", ], } def process_new_item(item: dict) -> str | None: title = item.get("title") if not title: log(f"Skipping item with no title: {item}") return None slug = slugify(title) launch_dir = LAUNCHES_DIR / slug if launch_dir.exists(): log(f"Launch folder already exists for {slug}; skipping creation.") return None launch_dir.mkdir(parents=True, exist_ok=True) state = build_state(item, slug) (launch_dir / "state.json").write_text( json.dumps(state, indent=2), encoding="utf-8" ) log(f"Created new launch: {slug} (from AI Pulse item '{title}')") notify( "AI Pulse → New Rollout", f"{title}\nOpen Claude Code and run /resume {slug}", ) return slug def tick() -> None: items = fetch_items() if items is None: return seen = load_seen() new_ids: list[str] = [] for item in items: item_id = item.get("id") or item.get("title") if not item_id: continue if item_id in seen: continue slug = process_new_item(item) if slug: seen.add(item_id) new_ids.append(item_id) if new_ids: save_seen(seen) log(f"Tick complete — {len(new_ids)} new item(s) processed.") else: log(f"Tick complete — {len(items)} item(s), nothing new.") def main() -> None: parser = argparse.ArgumentParser(description="AI Pulse watcher") parser.add_argument("--once", action="store_true", help="Run one poll and exit") parser.add_argument("--seed", action="store_true", help="Populate seen.json with current feed without creating launches (avoids back-fill on first run)") args = parser.parse_args() if args.seed: items = fetch_items() or [] seen = {i.get("id") or i.get("title") for i in items if (i.get("id") or i.get("title"))} save_seen(seen) log(f"Seeded seen.json with {len(seen)} existing item(s).") return if args.once: tick() return log(f"AI Pulse watcher started. Polling every {POLL_INTERVAL_SECONDS}s.") while True: try: tick() except Exception as e: log(f"Unexpected error: {e}") time.sleep(POLL_INTERVAL_SECONDS) if __name__ == "__main__": main()