session_watcher.py

#!/usr/bin/env python3.12
"""Session Watcher - extracts learnings from completed OpenClaw sessions."""

import argparse
import json
import logging
from datetime import datetime, timezone
from pathlib import Path

import requests
import yaml

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [session_watcher] %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)

BASE_DIR = Path(__file__).resolve().parent
PROCESSED_PATH = BASE_DIR / "processed_sessions.json"
SESSION_DIRS = [
    Path("~/.openclaw/agents/main/sessions").expanduser(),
    Path("~/.openclaw/cron/runs").expanduser(),
]
MIN_MESSAGES = 5
MAX_MESSAGES = 50
MAX_CHARS = 8000


def load_config():
    with open(BASE_DIR / "config.yaml") as f:
        return yaml.safe_load(f)


def ask_ollama(cfg, prompt, temperature=0.3):
    url = cfg["ollama"]["url"]
    for model in [cfg["ollama"]["model"], cfg["ollama"]["fallback_model"]]:
        try:
            resp = requests.post(url, json={
                "model": model,
                "prompt": prompt,
                "stream": False,
                "options": {"temperature": temperature},
            }, timeout=120)
            resp.raise_for_status()
            return resp.json().get("response", "")
        except Exception as e:
            log.warning("Ollama call failed (model=%s): %s", model, e)
    return None


def load_processed():
    if PROCESSED_PATH.exists():
        try:
            return set(json.loads(PROCESSED_PATH.read_text()))
        except (json.JSONDecodeError, TypeError):
            return set()
    return set()


def save_processed(processed):
    PROCESSED_PATH.write_text(json.dumps(sorted(processed), indent=2))


def find_new_sessions(processed):
    """Find unprocessed .jsonl session files."""
    new = []
    for d in SESSION_DIRS:
        if not d.exists():
            continue
        for f in d.glob("*.jsonl"):
            if f.name not in processed:
                new.append(f)
    return sorted(new, key=lambda p: p.stat().st_mtime)


def extract_conversation(session_path):
    """Read session JSONL and return user/assistant text messages."""
    messages = []
    try:
        for line in session_path.read_text().splitlines():
            if not line.strip():
                continue
            entry = json.loads(line)
            if entry.get("type") != "message":
                continue
            msg = entry.get("message", {})
            role = msg.get("role")
            if role not in ("user", "assistant"):
                continue
            for block in msg.get("content", []):
                if block.get("type") == "text" and block.get("text"):
                    messages.append({"role": role, "text": block["text"]})
    except Exception as e:
        log.warning("Failed to read %s: %s", session_path, e)
    return messages


def truncate_conversation(messages):
    """Keep last MAX_MESSAGES, truncate to MAX_CHARS total."""
    recent = messages[-MAX_MESSAGES:]
    result = []
    total = 0
    for m in reversed(recent):
        text = m["text"]
        if total + len(text) > MAX_CHARS:
            remaining = MAX_CHARS - total
            if remaining > 100:
                result.append({"role": m["role"], "text": text[:remaining] + "..."})
            break
        result.append(m)
        total += len(text)
    return list(reversed(result))


def format_for_prompt(messages):
    return "\n".join(f"[{m['role']}]: {m['text']}" for m in messages)


def analyze_session(cfg, messages):
    """Ask Ollama to extract learnings from a conversation."""
    convo = format_for_prompt(truncate_conversation(messages))
    prompt = f"""Analyze this AI assistant conversation and extract learnings.

{convo}

Return ONLY valid JSON with these keys:
- "title": short title for what was done (string)
- "summary": 1-2 sentence summary (string)
- "outcome": "success", "failure", or "partial" (string)
- "errors": list of errors/problems encountered (list of strings, empty if none)
- "lessons": list of reusable lessons or patterns learned (list of strings, empty if none)

JSON:"""
    raw = ask_ollama(cfg, prompt)
    if not raw:
        return None
    try:
        start = raw.index("{")
        end = raw.rindex("}") + 1
        return json.loads(raw[start:end])
    except (ValueError, json.JSONDecodeError) as e:
        log.warning("Failed to parse analysis JSON: %s", e)
        return None


def store_experience(cfg, session_id, analysis, dry_run=False):
    """Store session experience in AOMS episodic tier."""
    body = {
        "type": "experience",
        "payload": {
            "title": analysis.get("title", "Untitled session"),
            "summary": analysis.get("summary", ""),
            "outcome": analysis.get("outcome", "partial"),
            "session_id": session_id,
        },
        "tags": ["session"],
    }
    if dry_run:
        log.info("[DRY RUN] Would store episodic: %s", json.dumps(body, indent=2))
        return True
    try:
        resp = requests.post(cfg["aoms"]["url"] + "/memory/episodic", json=body, timeout=10)
        resp.raise_for_status()
        return True
    except Exception as e:
        log.warning("AOMS episodic store failed: %s", e)
        return False


def store_lessons(cfg, analysis, dry_run=False):
    """Store lessons as procedural skills in AOMS."""
    stored = 0
    for lesson in analysis.get("lessons", []):
        if not lesson or len(lesson) < 10:
            continue
        body = {
            "type": "skill",
            "payload": {
                "skill_name": lesson[:60],
                "description": lesson,
                "when_to_use": "When encountering similar situations in future sessions",
                "procedure": lesson,
                "category": "session_learning",
            },
            "tags": ["session_learning"],
        }
        if dry_run:
            log.info("[DRY RUN] Would store procedural: %s", body["payload"]["skill_name"])
            stored += 1
            continue
        try:
            resp = requests.post(cfg["aoms"]["url"] + "/memory/procedural", json=body, timeout=10)
            resp.raise_for_status()
            stored += 1
        except Exception as e:
            log.warning("AOMS procedural store failed: %s", e)
    return stored


def log_result(session_id, analysis):
    log_dir = BASE_DIR / "logs"
    log_dir.mkdir(exist_ok=True)
    log_file = log_dir / "session_watcher.log"
    entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "session_id": session_id,
        "title": analysis.get("title", ""),
        "outcome": analysis.get("outcome", ""),
        "lessons_count": len(analysis.get("lessons", [])),
    }
    with open(log_file, "a") as f:
        f.write(json.dumps(entry) + "\n")


def main():
    parser = argparse.ArgumentParser(description="Session Watcher - learn from sessions")
    parser.add_argument("--dry-run", action="store_true", help="Log actions without writing to AOMS")
    args = parser.parse_args()

    cfg = load_config()
    processed = load_processed()

    new_sessions = find_new_sessions(processed)
    if not new_sessions:
        log.info("No new sessions to process.")
        return

    log.info("Found %d new session(s)", len(new_sessions))

    for session_path in new_sessions:
        sid = session_path.name
        messages = extract_conversation(session_path)
        if len(messages) < MIN_MESSAGES:
            log.info("Skipping %s (%d messages, need %d)", sid, len(messages), MIN_MESSAGES)
            processed.add(sid)
            continue

        log.info("Analyzing %s (%d messages)", sid, len(messages))
        analysis = analyze_session(cfg, messages)
        if not analysis:
            log.warning("Analysis failed for %s, will retry next run", sid)
            continue

        store_experience(cfg, sid, analysis, dry_run=args.dry_run)
        lessons_stored = store_lessons(cfg, analysis, dry_run=args.dry_run)
        log.info("Session %s: %s (outcome=%s, lessons=%d)",
                 sid, analysis.get("title", "?"), analysis.get("outcome", "?"), lessons_stored)

        log_result(sid, analysis)
        if not args.dry_run:
            processed.add(sid)

    if not args.dry_run:
        save_processed(processed)

    # Also run auto-research scanner
    try:
        from auto_research import main as auto_research_main
        log.info("Running auto-research scanner...")
        auto_research_main()
    except Exception as e:
        log.warning("Auto-research scanner failed: %s", e)

    log.info("Session watcher complete.")


if __name__ == "__main__":
    main()