claudetools/.claude/hooks/periodic_save_check.py

#!/usr/bin/env python3
"""
Periodic Context Save - Windows Task Scheduler Version

This script is designed to be called every minute by Windows Task Scheduler.
It tracks active time and saves context every 5 minutes of activity.

Usage:
    Schedule this to run every minute via Task Scheduler:
    python .claude/hooks/periodic_save_check.py
"""

import os
import sys
import json
import subprocess
from datetime import datetime, timezone
from pathlib import Path

import requests

# Configuration
SCRIPT_DIR = Path(__file__).parent
CLAUDE_DIR = SCRIPT_DIR.parent
PROJECT_ROOT = CLAUDE_DIR.parent
STATE_FILE = CLAUDE_DIR / ".periodic-save-state.json"
LOG_FILE = CLAUDE_DIR / "periodic-save.log"
CONFIG_FILE = CLAUDE_DIR / "context-recall-config.env"
LOCK_FILE = CLAUDE_DIR / ".periodic-save.lock"  # Mutex lock to prevent overlaps

SAVE_INTERVAL_SECONDS = 300  # 5 minutes


def log(message):
    """Write log message"""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_message = f"[{timestamp}] {message}\n"

    try:
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            f.write(log_message)
    except:
        pass  # Silent fail if can't write log


def load_config():
    """Load configuration from context-recall-config.env"""
    config = {
        "api_url": "http://172.16.3.30:8001",
        "jwt_token": None,
    }

    if CONFIG_FILE.exists():
        with open(CONFIG_FILE) as f:
            for line in f:
                line = line.strip()
                if line.startswith("CLAUDE_API_URL="):
                    config["api_url"] = line.split("=", 1)[1]
                elif line.startswith("JWT_TOKEN="):
                    config["jwt_token"] = line.split("=", 1)[1]

    return config


def detect_project_id():
    """Detect project ID from git config"""
    try:
        os.chdir(PROJECT_ROOT)

        # Try git config first
        result = subprocess.run(
            ["git", "config", "--local", "claude.projectid"],
            capture_output=True,
            text=True,
            check=False,
            cwd=PROJECT_ROOT,
            timeout=5,  # Prevent hung processes
        )
        if result.returncode == 0 and result.stdout.strip():
            return result.stdout.strip()

        # Try to derive from git remote URL
        result = subprocess.run(
            ["git", "config", "--get", "remote.origin.url"],
            capture_output=True,
            text=True,
            check=False,
            cwd=PROJECT_ROOT,
            timeout=5,  # Prevent hung processes
        )
        if result.returncode == 0 and result.stdout.strip():
            import hashlib
            return hashlib.md5(result.stdout.strip().encode()).hexdigest()

    except Exception:
        pass

    return "unknown"


def is_claude_active():
    """Check if Claude Code is actively running"""
    try:
        # Check for Claude Code process
        result = subprocess.run(
            ["tasklist.exe"],
            capture_output=True,
            text=True,
            check=False,
            timeout=5,  # Prevent hung processes
        )

        # Look for claude, node, or other indicators
        output_lower = result.stdout.lower()
        if any(proc in output_lower for proc in ["claude", "node.exe", "code.exe"]):
            # Also check for recent file modifications
            import time
            two_minutes_ago = time.time() - 120

            # Check a few common directories for recent activity
            for check_dir in [PROJECT_ROOT, PROJECT_ROOT / "api", PROJECT_ROOT / ".claude"]:
                if check_dir.exists():
                    for file in check_dir.rglob("*"):
                        if file.is_file():
                            try:
                                if file.stat().st_mtime > two_minutes_ago:
                                    return True
                            except:
                                continue

    except Exception as e:
        log(f"Error checking activity: {e}")

    return False


def acquire_lock():
    """Acquire execution lock to prevent overlapping runs"""
    try:
        # Check if lock file exists and is recent (< 60 seconds old)
        if LOCK_FILE.exists():
            lock_age = datetime.now().timestamp() - LOCK_FILE.stat().st_mtime
            if lock_age < 60:  # Lock is fresh, another instance is running
                log("[INFO] Another instance is running, skipping")
                return False

        # Create/update lock file
        LOCK_FILE.touch()
        return True
    except Exception as e:
        log(f"[WARNING] Lock acquisition failed: {e}")
        return True  # Proceed anyway if lock fails


def release_lock():
    """Release execution lock"""
    try:
        if LOCK_FILE.exists():
            LOCK_FILE.unlink()
    except Exception:
        pass  # Ignore errors on cleanup


def load_state():
    """Load state from state file"""
    if STATE_FILE.exists():
        try:
            with open(STATE_FILE) as f:
                return json.load(f)
        except Exception:
            pass

    return {
        "active_seconds": 0,
        "last_check": None,
        "last_save": None,
    }


def save_state(state):
    """Save state to state file"""
    state["last_check"] = datetime.now(timezone.utc).isoformat()
    try:
        with open(STATE_FILE, "w") as f:
            json.dump(state, f, indent=2)
    except:
        pass  # Silent fail


def save_periodic_context(config, project_id):
    """Save context to database via API"""
    if not config["jwt_token"]:
        log("No JWT token - cannot save context")
        return False

    title = f"Periodic Save - {datetime.now().strftime('%Y-%m-%d %H:%M')}"
    summary = f"Auto-saved context after {SAVE_INTERVAL_SECONDS // 60} minutes of active work. Session in progress on project: {project_id}"

    payload = {
        "context_type": "session_summary",
        "title": title,
        "dense_summary": summary,
        "relevance_score": 5.0,
        "tags": json.dumps(["auto-save", "periodic", "active-session", project_id]),
    }

    try:
        url = f"{config['api_url']}/api/conversation-contexts"
        headers = {
            "Authorization": f"Bearer {config['jwt_token']}",
            "Content-Type": "application/json",
        }

        response = requests.post(url, json=payload, headers=headers, timeout=10)

        if response.status_code in [200, 201]:
            context_id = response.json().get('id', 'unknown')
            log(f"[OK] Context saved (ID: {context_id}, Active time: {SAVE_INTERVAL_SECONDS}s)")
            return True
        else:
            log(f"[ERROR] Failed to save: HTTP {response.status_code}")
            return False

    except Exception as e:
        log(f"[ERROR] Error saving context: {e}")
        return False


def main():
    """Main entry point - called every minute by Task Scheduler"""
    # Acquire lock to prevent overlapping executions
    if not acquire_lock():
        return 0  # Another instance is running, exit gracefully

    try:
        config = load_config()
        state = load_state()

        # Check if Claude is active
        if is_claude_active():
            # Increment active time (60 seconds per check)
            state["active_seconds"] += 60

            # Check if we've reached the save interval
            if state["active_seconds"] >= SAVE_INTERVAL_SECONDS:
                log(f"{SAVE_INTERVAL_SECONDS}s active time reached - saving context")

                project_id = detect_project_id()
                if save_periodic_context(config, project_id):
                    state["last_save"] = datetime.now(timezone.utc).isoformat()

                # Reset timer
                state["active_seconds"] = 0

            save_state(state)
        else:
            # Not active - don't increment timer but save state
            save_state(state)

        return 0
    finally:
        # Always release lock, even if error occurs
        release_lock()


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as e:
        log(f"Fatal error: {e}")
        sys.exit(1)