#!/usr/bin/env python3 """ Deep Think Memory Upgrader ========================== Two-phase pipeline: Phase 1: Gemini Flash generates a specific, actionable Deep Think prompt targeting one memory upgrade area. Phase 2: Gemini Pro (thinking enabled) executes the prompt via api_mode(). Phase 3: Results saved, KG-ingested, code proposals extracted. CLI flags: --loop Continuous 30-minute cycles (round-robin topics) --topic TOPIC Single topic run (exact slug, see --list-topics) --list-topics Show all topics and their coverage status --status Show tracker state summary --dry-run Generate prompt only, skip Deep Think execution Usage: python3 scripts/deep_think_memory_upgrader.py --list-topics python3 scripts/deep_think_memory_upgrader.py --topic hybrid_weights python3 scripts/deep_think_memory_upgrader.py --loop python3 scripts/deep_think_memory_upgrader.py --dry-run --topic auto_capture """ import argparse import asyncio import json import logging import os import re import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Optional def utcnow_iso() -> str: """Return current UTC time as ISO string (timezone-aware, no deprecation).""" return datetime.now(timezone.utc).isoformat() # --------------------------------------------------------------------------- # Paths (all E: drive) # --------------------------------------------------------------------------- BASE_DIR = Path("/mnt/e/genesis-system") SCRIPTS_DIR = BASE_DIR / "scripts" DT_BRIDGE = SCRIPTS_DIR / "deep_think_bridge.py" DT_RESULTS_DIR = BASE_DIR / "deep_think_results" DT_RESULTS_DIR.mkdir(parents=True, exist_ok=True) MEMORY_SRC_INDEX = SCRIPTS_DIR / "aiva_memory" / "enhanced_index.ts" MEMORY_SRC_CONFIG = SCRIPTS_DIR / "aiva_memory" / "enhanced_config.ts" PROPOSED_DIR = SCRIPTS_DIR / "aiva_memory" / "proposed_upgrades" PROPOSED_DIR.mkdir(parents=True, exist_ok=True) TRACKER_JSON = BASE_DIR / "data" / "deep_think_memory_tracker.json" TRACKER_JSON.parent.mkdir(parents=True, exist_ok=True) LOG_DIR = BASE_DIR / "data" / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) LOG_FILE = LOG_DIR / "dt_memory_upgrader.log" KG_INGEST_SCRIPT = SCRIPTS_DIR / "deep_think_kg_ingest.py" # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler(sys.stdout), ], ) logger = logging.getLogger("dt_memory_upgrader") # --------------------------------------------------------------------------- # Import bridge functions (api_mode + get_api_key) # --------------------------------------------------------------------------- # Add scripts dir to path so we can import deep_think_bridge directly sys.path.insert(0, str(SCRIPTS_DIR)) sys.path.insert(0, str(BASE_DIR)) try: from deep_think_bridge import api_mode, get_api_key logger.info("Imported api_mode + get_api_key from deep_think_bridge.py") except ImportError as e: logger.error(f"Failed to import deep_think_bridge: {e}") logger.error("Ensure deep_think_bridge.py exists at scripts/deep_think_bridge.py") sys.exit(1) # --------------------------------------------------------------------------- # Topic Registry # --------------------------------------------------------------------------- TOPICS = [ { "slug": "hybrid_weights", "name": "Hybrid Retrieval Weight Optimization", "description": ( "Current weights: vector=0.4, bm25=0.3, recency=0.2, importance=0.1. " "Evaluate whether these are optimal for a long-running personal AI assistant. " "Consider: conversation type distribution, memory age profiles, importance score " "distributions in real usage. Should weights be dynamic (learned from retrieval " "feedback)? Should they be per-category? What would A/B testing look like?" ), }, { "slug": "auto_capture", "name": "Auto-Capture Trigger Pattern Expansion", "description": ( "Current system has ~30 regex trigger patterns covering preferences, decisions, " "contacts, dates, locations, skills, goals, projects, URLs, and numbers. " "What important personal information patterns are MISSING? Consider: relationship " "dynamics, health/medical facts, financial state, recurring frustrations, " "user values and beliefs, professional history, cultural preferences, language " "code-switching signals. How can we capture implicit rather than explicit statements?" ), }, { "slug": "consolidation_algo", "name": "Memory Consolidation Algorithm Improvements", "description": ( "Current consolidation: semantic dedup at 0.80 similarity (merge), staleness " "pruning at 90 days + importance < 0.5. Problems: O(n²) merge pass is slow at " "scale, merge strategy (keep-longer) loses nuance, staleness threshold is " "arbitrary. Design a better consolidation algorithm: incremental (not full-scan), " "smarter merge strategy (combine both texts into a synthesis), adaptive staleness " "based on category (events decay faster, facts decay slower), importance decay " "acceleration for contradicted memories." ), }, { "slug": "cross_session_transfer", "name": "Cross-Session Memory Transfer Protocols", "description": ( "LanceDB persists memory vectors across sessions but there is no protocol for: " "session boundary detection, end-of-session consolidation triggers, priority " "boosting for session-critical memories, session context injection at startup, " "or detecting 'new conversation topic' vs 'same running thread'. Design a complete " "cross-session memory lifecycle: pre-session load, mid-session updates, " "post-session flush, and long-term archival strategy." ), }, { "slug": "episodic_semantic_separation", "name": "Episodic vs Semantic Memory Separation", "description": ( "Current system uses a single unified memory store. Cognitive science distinguishes: " "episodic memory (specific events with time/place) vs semantic memory (general " "knowledge/facts). How should these be separated in the vector store architecture? " "Should episodic memories decay faster? Should semantic memories be organized " "differently (concept graphs vs flat vectors)? Design a dual-store architecture " "with cross-store retrieval fusion and automatic routing on store." ), }, { "slug": "importance_decay_curves", "name": "Memory Importance Decay Curve Engineering", "description": ( "Current decay: linear at 0.01/day with 0.1 floor, half-life recency at 30 days, " "freshness boost of 1.3x for 24h window. Real human memory doesn't decay linearly " "— it follows power law forgetting curves (Ebbinghaus) with interference effects. " "Design category-specific decay curves: contact info (slow decay), events (rapid " "post-date decay), preferences (periodic refresh), skills (very slow decay). " "Include spaced repetition boosting for recalled memories." ), }, { "slug": "context_aware_injection", "name": "Context-Aware Memory Injection", "description": ( "Current auto-recall injects the top-N memories by hybrid score. No awareness of: " "current conversation topic, user's current emotional state, task context " "(coding vs personal vs business), time of day, or whether the memory is " "actually relevant to what's being discussed. Design a context-aware injection " "layer: topic modeling on conversation, memory relevance re-ranking based on " "detected intent, suppression of irrelevant memories even if high-scoring, " "and injection placement optimization (beginning vs end of context)." ), }, { "slug": "memory_compression", "name": "Memory Compression for Long-Term Storage", "description": ( "At scale (1000+ memories), vector storage and retrieval costs grow. Current " "system has no compression strategy. Design: progressive summarization (recent " "memories = full detail, older = compressed summaries), hierarchical memory " "clusters (group related memories into meta-memories), vector quantization " "for embedding storage efficiency, and a cold-storage tier for rarely-accessed " "old memories that can be re-hydrated on demand." ), }, { "slug": "multi_agent_shared_memory", "name": "Multi-Agent Shared Memory Protocols", "description": ( "Genesis has Queen AIVA and SubAIVA instances. Current memory is fully isolated " "per-agent. Design a shared memory protocol: what patterns should SubAIVAs " "propagate upward to Queen (aggregate learnings, not personal data), how should " "Queen push capability updates down to SubAIVAs, privacy partitioning (customer " "data stays in SubAIVA, behavioral patterns flow up), conflict resolution when " "SubAIVAs have contradictory preferences, and federated memory training loops." ), }, { "slug": "surprise_learning_integration", "name": "Surprise-Based Learning Integration (Titan Memory Connection)", "description": ( "Genesis has Titan Memory (surprise-based learning) but it is disconnected from " "the AIVA LanceDB memory system. Design integration: surprise score computation " "for new memories (how unexpected is this?), high-surprise memories get boosted " "importance, surprise patterns trigger automatic memory category reclassification, " "connecting Genesis KG axiom generation to AIVA personal memory, and a " "feedback loop where recalled memories that proved useful increase their " "importance score (reinforcement)." ), }, ] TOPIC_SLUGS = {t["slug"]: t for t in TOPICS} # --------------------------------------------------------------------------- # Tracker Management # --------------------------------------------------------------------------- def load_tracker() -> dict: """Load tracker JSON, creating empty structure if missing.""" if TRACKER_JSON.exists(): try: return json.loads(TRACKER_JSON.read_text()) except json.JSONDecodeError: logger.warning("Tracker JSON corrupt — resetting") return { "version": 1, "created_at": utcnow_iso(), "pass": 1, "topics": { t["slug"]: { "runs": 0, "last_run": None, "last_file": None, "status": "pending", } for t in TOPICS }, "total_cycles": 0, } def save_tracker(tracker: dict) -> None: """Persist tracker JSON.""" TRACKER_JSON.write_text(json.dumps(tracker, indent=2)) def next_topic(tracker: dict) -> dict: """ Pick the next topic to run: - Prefer topics with 0 runs in current pass - If all topics have been run in pass N, start pass N+1 - Within pending topics, prefer ones not run most recently """ current_pass = tracker.get("pass", 1) topic_states = tracker["topics"] # Pending = run count < current_pass pending = [ slug for slug, state in topic_states.items() if state["runs"] < current_pass ] if not pending: # All topics covered — start next pass tracker["pass"] = current_pass + 1 logger.info(f"All topics covered in pass {current_pass}. Starting pass {tracker['pass']}.") pending = list(topic_states.keys()) # Sort by last_run ascending (None first = never run) def sort_key(slug: str): lr = topic_states[slug].get("last_run") return lr or "0000-00-00T00:00:00" pending.sort(key=sort_key) chosen_slug = pending[0] return TOPIC_SLUGS[chosen_slug] # --------------------------------------------------------------------------- # Memory Source Reader # --------------------------------------------------------------------------- def read_memory_sources() -> tuple[str, str]: """Read enhanced_index.ts and enhanced_config.ts. Return (index_src, config_src).""" index_src = MEMORY_SRC_INDEX.read_text(encoding="utf-8") if MEMORY_SRC_INDEX.exists() else "" config_src = MEMORY_SRC_CONFIG.read_text(encoding="utf-8") if MEMORY_SRC_CONFIG.exists() else "" return index_src, config_src def read_recent_dt_results(max_files: int = 5) -> str: """Read the most recent DT results for context.""" pattern = "DT_MEMORY_UPGRADE_*.md" files = sorted(DT_RESULTS_DIR.glob(pattern), key=lambda f: f.stat().st_mtime, reverse=True) if not files: # Fall back to any recent DT result files = sorted(DT_RESULTS_DIR.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True) summaries = [] for f in files[:max_files]: try: content = f.read_text(encoding="utf-8", errors="ignore") # Take first 1500 chars as summary summaries.append(f"=== {f.name} ===\n{content[:1500]}\n") except Exception: pass return "\n".join(summaries) if summaries else "(No prior Deep Think results found)" # --------------------------------------------------------------------------- # Phase 1: Flash Prompt Generation # --------------------------------------------------------------------------- FLASH_SYSTEM_CONTEXT = """You are a world-class AI memory systems architect with deep expertise in: - Vector database design and hybrid retrieval systems - Cognitive science and computational memory models - Production AI system optimization - TypeScript/Node.js backend systems - LanceDB, BM25, MMR, and embedding-based retrieval You are analyzing AIVA's production memory system (OpenClaw LanceDB plugin) to generate a highly specific, technically deep, and immediately actionable Deep Think prompt. Your output will be fed directly to Gemini Pro with 16384 thinking tokens — the most powerful reasoning mode available. Your prompt must be worthy of that compute budget. Do NOT generate a vague or generic prompt. Generate a prompt that: 1. References specific code constants/functions from the source (by name) 2. Asks for concrete algorithmic proposals with pseudocode or TypeScript sketches 3. Requests measurable improvement targets (e.g., "improve P90 retrieval precision by X%") 4. Frames the problem with quantitative context (current values, thresholds, limits) 5. Asks for production-ready implementation guidance, not theoretical discussion """ def build_flash_prompt(topic: dict, index_src: str, config_src: str, recent_results: str, tracker: dict) -> str: """Construct the Flash prompt that generates the Deep Think prompt.""" topic_state = tracker["topics"].get(topic["slug"], {}) runs_so_far = topic_state.get("runs", 0) current_pass = tracker.get("pass", 1) pass_context = "" if current_pass > 1: pass_context = f""" NOTE: This is Pass {current_pass} for this topic. Prior analysis was done in earlier passes. The Deep Think prompt should build on what was already explored and go DEEPER. Prior pass findings are included in the recent results section below. """ return f"""{FLASH_SYSTEM_CONTEXT} --- ## YOUR TASK: Generate a Deep Think Prompt You must generate ONE highly specific Deep Think prompt targeting this memory upgrade area: **Topic**: {topic['name']} **Target**: {topic['description']} --- ## CURRENT CODEBASE CONTEXT ### enhanced_config.ts (173 lines — full source): ```typescript {config_src} ``` ### enhanced_index.ts (1,811 lines — full source): ```typescript {index_src} ``` --- ## RECENT DEEP THINK RESULTS (for continuity — avoid repeating what's been explored): {recent_results} --- ## TRACKER STATE: - Topic: {topic['slug']} — Run #{runs_so_far + 1} (Pass {current_pass}) - Total cycles completed: {tracker.get('total_cycles', 0)} {pass_context} --- ## OUTPUT FORMAT Generate ONLY the Deep Think prompt text. No preamble, no meta-commentary. Just the prompt. The prompt must: 1. Open with the current system state (cite actual code constants by name) 2. Identify 3-5 specific weaknesses or optimization opportunities with quantitative framing 3. Ask for concrete proposals in TypeScript (matching the existing codebase style) 4. Request A/B test designs or evaluation metrics for each proposal 5. End with: "Provide a complete implementation roadmap with TypeScript code snippets suitable for direct integration into the enhanced_index.ts / enhanced_config.ts files." Length: 600-900 words. Dense, technical, specific. """ def generate_flash_prompt(topic: dict, index_src: str, config_src: str, recent_results: str, tracker: dict, api_key: str) -> str: """Use Gemini Flash (NO thinking) to generate the Deep Think prompt.""" import google.genai as genai client = genai.Client(api_key=api_key) flash_prompt = build_flash_prompt(topic, index_src, config_src, recent_results, tracker) logger.info(f"[Phase 1] Generating Deep Think prompt via Gemini Flash for topic: {topic['slug']}") start = time.time() response = client.models.generate_content( model="gemini-2.5-flash", contents=flash_prompt, config=genai.types.GenerateContentConfig( # NO thinking for Flash — fast and cheap thinking_config=genai.types.ThinkingConfig( thinking_budget=0, ), temperature=0.8, max_output_tokens=2048, ), ) elapsed = time.time() - start generated_prompt = response.text.strip() logger.info(f"[Phase 1] Flash generated prompt in {elapsed:.2f}s ({len(generated_prompt)} chars)") return generated_prompt # --------------------------------------------------------------------------- # Phase 2: Deep Think Execution # --------------------------------------------------------------------------- def run_deep_think(prompt: str, topic: dict, dry_run: bool = False) -> Optional[dict]: """Execute the Deep Think prompt via Gemini Pro with full thinking budget.""" if dry_run: logger.info("[Phase 2 DRY-RUN] Skipping Deep Think execution") logger.info(f"[Phase 2 DRY-RUN] Prompt preview ({len(prompt)} chars):\n{prompt[:500]}...") return None logger.info(f"[Phase 2] Firing Deep Think — model=gemini-2.5-pro, budget=16384") result = api_mode( prompt=prompt, model="gemini-2.5-pro", thinking_budget=16384, ) logger.info( f"[Phase 2] Deep Think complete — {result.get('elapsed_seconds', 0):.1f}s, " f"response={len(result.get('response', ''))} chars, " f"thinking={len(result.get('thinking', ''))} chars" ) return result # --------------------------------------------------------------------------- # Phase 3: Result Processing # --------------------------------------------------------------------------- def next_result_index(topic_slug: str) -> int: """Count existing files for this topic to determine next N.""" pattern = f"DT_MEMORY_UPGRADE_*_{topic_slug}.md" existing = list(DT_RESULTS_DIR.glob(pattern)) return len(existing) + 1 def save_result(result: dict, topic: dict, generated_prompt: str) -> tuple[Path, Path]: """Save Deep Think result to both .md and .json files.""" n = next_result_index(topic["slug"]) stem = f"DT_MEMORY_UPGRADE_{n}_{topic['slug']}" md_path = DT_RESULTS_DIR / f"{stem}.md" json_path = DT_RESULTS_DIR / f"{stem}.json" timestamp = utcnow_iso() # Markdown md_content = f"""# Deep Think: Memory Upgrade — {topic['name']} **Topic**: `{topic['slug']}` **Cycle**: {n} **Model**: {result.get('model', 'gemini-2.5-pro')} **Generated**: {timestamp} **Elapsed**: {result.get('elapsed_seconds', 0):.1f}s **Thinking tokens**: {result.get('thinking_budget', 16384)} --- ## Generated Prompt (Phase 1 — Gemini Flash) {generated_prompt} --- ## Thinking Process (Phase 2 — Gemini Pro) {result.get('thinking', '(no thinking trace captured)')} --- ## Deep Think Response {result.get('response', '')} """ md_path.write_text(md_content, encoding="utf-8") # JSON json_data = { "topic_slug": topic["slug"], "topic_name": topic["name"], "cycle": n, "model": result.get("model"), "thinking_budget": result.get("thinking_budget"), "elapsed_seconds": result.get("elapsed_seconds"), "generated_at": timestamp, "generated_prompt": generated_prompt, "thinking": result.get("thinking", ""), "response": result.get("response", ""), } json_path.write_text(json.dumps(json_data, indent=2), encoding="utf-8") logger.info(f"[Phase 3] Saved: {md_path.name} + {json_path.name}") return md_path, json_path def extract_code_proposals(result: dict, topic: dict, cycle: int) -> None: """ Extract TypeScript code blocks from the Deep Think response and save them as proposed upgrade files. """ response = result.get("response", "") if not response: return # Find all TypeScript code blocks ts_blocks = re.findall(r"```typescript\n(.*?)```", response, re.DOTALL) ts_blocks += re.findall(r"```ts\n(.*?)```", response, re.DOTALL) if not ts_blocks: logger.info("[Phase 3] No TypeScript code blocks found in response") return timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") output_path = PROPOSED_DIR / f"proposal_{topic['slug']}_cycle{cycle}_{timestamp}.ts" header = f"""// PROPOSED MEMORY UPGRADE // Topic: {topic['name']} ({topic['slug']}) // Cycle: {cycle} // Generated: {utcnow_iso()} // Source: Deep Think Memory Upgrader (gemini-2.5-pro, thinking_budget=16384) // Status: PROPOSED — review before applying to enhanced_index.ts // // To apply: review each block below and integrate into: // /mnt/e/genesis-system/scripts/aiva_memory/enhanced_index.ts // /mnt/e/genesis-system/scripts/aiva_memory/enhanced_config.ts """ combined = header + "\n\n// ---\n\n".join(ts_blocks) output_path.write_text(combined, encoding="utf-8") logger.info(f"[Phase 3] Extracted {len(ts_blocks)} code block(s) -> {output_path.name}") def ingest_to_kg(md_path: Path) -> None: """Run the KG ingest pipeline on the new result file.""" if not KG_INGEST_SCRIPT.exists(): logger.warning(f"[Phase 3] KG ingest script not found: {KG_INGEST_SCRIPT}") return import subprocess try: filter_name = md_path.stem[:30] # match by filename stem prefix cmd = [ sys.executable, str(KG_INGEST_SCRIPT), "--file", filter_name, ] logger.info(f"[Phase 3] Ingesting to KG: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode == 0: logger.info(f"[Phase 3] KG ingest complete:\n{result.stdout[-500:]}") else: logger.warning(f"[Phase 3] KG ingest stderr: {result.stderr[-300:]}") except subprocess.TimeoutExpired: logger.warning("[Phase 3] KG ingest timed out after 120s") except Exception as e: logger.warning(f"[Phase 3] KG ingest error: {e}") # --------------------------------------------------------------------------- # Single Cycle Execution # --------------------------------------------------------------------------- def run_cycle(topic: dict, tracker: dict, api_key: str, dry_run: bool = False) -> None: """Execute one full upgrade cycle for the given topic.""" logger.info(f"=== CYCLE START: {topic['slug']} ===") # Read source files index_src, config_src = read_memory_sources() recent_results = read_recent_dt_results(max_files=3) # Phase 1: Generate prompt via Flash try: generated_prompt = generate_flash_prompt( topic, index_src, config_src, recent_results, tracker, api_key ) except Exception as e: logger.error(f"[Phase 1] Flash prompt generation failed: {e}") return if dry_run: print("\n" + "=" * 80) print(f"DRY-RUN: Generated prompt for topic '{topic['slug']}'") print("=" * 80) print(generated_prompt) print("=" * 80 + "\n") logger.info("[DRY-RUN] Cycle complete (no Deep Think fired)") return # Phase 2: Deep Think try: result = run_deep_think(generated_prompt, topic, dry_run=False) if result is None: return except Exception as e: logger.error(f"[Phase 2] Deep Think execution failed: {e}") return # Phase 3: Save + extract + ingest try: n = next_result_index(topic["slug"]) md_path, json_path = save_result(result, topic, generated_prompt) extract_code_proposals(result, topic, n) ingest_to_kg(md_path) except Exception as e: logger.error(f"[Phase 3] Result processing failed: {e}") return # Update tracker topic_state = tracker["topics"].setdefault(topic["slug"], { "runs": 0, "last_run": None, "last_file": None, "status": "pending" }) topic_state["runs"] = topic_state.get("runs", 0) + 1 topic_state["last_run"] = utcnow_iso() topic_state["last_file"] = str(md_path.name) topic_state["status"] = "completed" tracker["total_cycles"] = tracker.get("total_cycles", 0) + 1 save_tracker(tracker) logger.info(f"=== CYCLE COMPLETE: {topic['slug']} (cycle #{topic_state['runs']}) ===") # --------------------------------------------------------------------------- # CLI Commands # --------------------------------------------------------------------------- def cmd_list_topics(tracker: dict) -> None: """Print all topics and their tracker status.""" current_pass = tracker.get("pass", 1) total_cycles = tracker.get("total_cycles", 0) print(f"\nDeep Think Memory Upgrader — Topic Status") print(f"Pass: {current_pass} | Total cycles: {total_cycles}") print("=" * 70) for topic in TOPICS: slug = topic["slug"] state = tracker["topics"].get(slug, {"runs": 0, "last_run": None, "status": "pending"}) runs = state.get("runs", 0) last_run_raw = state.get("last_run") if last_run_raw: last_run = last_run_raw[:19].replace("T", " ") else: last_run = "never" status_icon = "+" if state.get("status") == "completed" else "-" print(f" [{status_icon}] {slug:<35} runs={runs:<3} last={last_run}") print(f" {topic['name']}") print("=" * 70) print(f"Tracker: {TRACKER_JSON}") print(f"Results: {DT_RESULTS_DIR}") print(f"Proposals: {PROPOSED_DIR}\n") def cmd_status(tracker: dict) -> None: """Print compact tracker summary.""" current_pass = tracker.get("pass", 1) total = tracker.get("total_cycles", 0) completed = sum( 1 for s in tracker["topics"].values() if s.get("status") == "completed" ) pending = len(TOPICS) - completed print(f"\nPass {current_pass} | Total cycles: {total} | Completed: {completed} | Pending: {pending}") next_t = next_topic(tracker) print(f"Next topic: {next_t['slug']} — {next_t['name']}\n") def cmd_loop(tracker: dict, api_key: str, interval_seconds: int = 1800) -> None: """Continuous loop: one cycle every interval_seconds (default 30 min).""" logger.info(f"Starting continuous loop — interval={interval_seconds}s ({interval_seconds//60} min)") while True: topic = next_topic(tracker) logger.info(f"Loop selected topic: {topic['slug']}") try: run_cycle(topic, tracker, api_key, dry_run=False) except KeyboardInterrupt: logger.info("Loop interrupted by user (KeyboardInterrupt)") break except Exception as e: logger.error(f"Cycle error (continuing): {e}") logger.info(f"Sleeping {interval_seconds}s before next cycle...") try: time.sleep(interval_seconds) except KeyboardInterrupt: logger.info("Sleep interrupted — exiting loop") break # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser( description="Deep Think Memory Upgrader — Flash generates prompts, Pro executes them", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 scripts/deep_think_memory_upgrader.py --list-topics python3 scripts/deep_think_memory_upgrader.py --status python3 scripts/deep_think_memory_upgrader.py --dry-run --topic hybrid_weights python3 scripts/deep_think_memory_upgrader.py --topic auto_capture python3 scripts/deep_think_memory_upgrader.py --loop """, ) parser.add_argument("--loop", action="store_true", help="Continuous 30-minute cycles (round-robin topics)") parser.add_argument("--topic", type=str, metavar="SLUG", help="Run a single topic by slug (see --list-topics)") parser.add_argument("--list-topics", action="store_true", help="Show all topics and their tracker status") parser.add_argument("--status", action="store_true", help="Show compact tracker summary") parser.add_argument("--dry-run", action="store_true", help="Generate prompt only, skip Deep Think API call") parser.add_argument("--interval", type=int, default=1800, help="Loop interval in seconds (default: 1800 = 30 min)") args = parser.parse_args() # Load tracker tracker = load_tracker() # Info-only commands if args.list_topics: cmd_list_topics(tracker) return 0 if args.status: cmd_status(tracker) return 0 # Validate API key (needed for all execution commands) api_key = get_api_key() if not api_key: logger.error("No Gemini API key found. Set GEMINI_API_KEY or ensure Credentials/gemini_api_key.txt exists.") return 1 # Dry-run info commands if args.dry_run: if args.topic: if args.topic not in TOPIC_SLUGS: logger.error(f"Unknown topic slug: {args.topic}. Use --list-topics to see valid slugs.") return 1 topic = TOPIC_SLUGS[args.topic] else: topic = next_topic(tracker) run_cycle(topic, tracker, api_key, dry_run=True) return 0 # Single topic run if args.topic: if args.topic not in TOPIC_SLUGS: logger.error(f"Unknown topic slug: {args.topic}. Use --list-topics to see valid slugs.") return 1 topic = TOPIC_SLUGS[args.topic] run_cycle(topic, tracker, api_key, dry_run=False) return 0 # Loop mode if args.loop: cmd_loop(tracker, api_key, interval_seconds=args.interval) return 0 # Default: run next topic once topic = next_topic(tracker) logger.info(f"No mode specified — running next topic: {topic['slug']}") run_cycle(topic, tracker, api_key, dry_run=False) return 0 if __name__ == "__main__": sys.exit(main())