Bmad Agent Dream Weaver

module,skill,display-name,menu-code,description,action,args,phase,after,before,required,output-location,outputs
Dream Weaver,bmad-agent-dream-weaver,Dream Capture,DL,"Capture and log a dream through guided conversation.",dream-log,,anytime,,,false,,journal entry
Dream Weaver,bmad-agent-dream-weaver,Dream Interpretation,DI,"Analyze a dream for symbolism, meaning, and personal connections.",dream-interpret,,anytime,dw:dream-log,,false,,interpretation
Dream Weaver,bmad-agent-dream-weaver,Recall Training,RT,"Dream recall exercises and progress tracking.",recall-training,,anytime,,,false,,coaching updates
Dream Weaver,bmad-agent-dream-weaver,Lucid Coaching,LC,"Progressive lucid dreaming training and milestone tracking.",lucid-coach,,anytime,dw:recall-training,,false,,coaching updates
Dream Weaver,bmad-agent-dream-weaver,Dream Seeding,DS,"Pre-sleep dream incubation — plant themes and intentions.",dream-seed,,anytime,,,false,,seed log entry
Dream Weaver,bmad-agent-dream-weaver,Pattern Discovery,PD,"Surface recurring themes, symbols, and emotional patterns across dreams.",pattern-discovery,,anytime,dw:dream-log,,false,,pattern analysis
Dream Weaver,bmad-agent-dream-weaver,Dream Query,DQ,"Search dream history by symbol, emotion, date, or keyword.",dream-query,,anytime,dw:dream-log,,false,,search results
Dream Weaver,bmad-agent-dream-weaver,Save Memory,SM,"Save current session context to memory.",save-memory,,anytime,,,false,,memory checkpoint

code: dw
name: "Dream Weaver"
description: "Dream journal, interpretation, and lucid dreaming coach"
module_version: 1.0.0
default_selected: false
module_greeting: >
  Your dream space is ready. I'm Oneira — your guide through the landscapes of sleep.
  Tell me a dream, and we'll begin.

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# dependencies = ["pyyaml"]
# ///
"""Merge module configuration into shared _bmad/config.yaml and config.user.yaml.

Reads a module.yaml definition and a JSON answers file, then writes or updates
the shared config.yaml (core values at root + module section) and config.user.yaml
(user_name, communication_language, plus any module variable with user_setting: true).
Uses an anti-zombie pattern for the module section in config.yaml.

Legacy migration: when --legacy-dir is provided, reads old per-module config files
from {legacy-dir}/{module-code}/config.yaml and {legacy-dir}/core/config.yaml.
Matching values serve as fallback defaults (answers override them). After a
successful merge, the legacy config.yaml files are deleted. Only the current
module and core directories are touched — other module directories are left alone.

Exit codes: 0=success, 1=validation error, 2=runtime error
"""

import argparse
import json
import sys
from pathlib import Path

try:
    import yaml
except ImportError:
    print("Error: pyyaml is required (PEP 723 dependency)", file=sys.stderr)
    sys.exit(2)


def parse_args():
    parser = argparse.ArgumentParser(
        description="Merge module config into shared _bmad/config.yaml with anti-zombie pattern."
    )
    parser.add_argument(
        "--config-path",
        required=True,
        help="Path to the target _bmad/config.yaml file",
    )
    parser.add_argument(
        "--module-yaml",
        required=True,
        help="Path to the module.yaml definition file",
    )
    parser.add_argument(
        "--answers",
        required=True,
        help="Path to JSON file with collected answers",
    )
    parser.add_argument(
        "--user-config-path",
        required=True,
        help="Path to the target _bmad/config.user.yaml file",
    )
    parser.add_argument(
        "--legacy-dir",
        help="Path to _bmad/ directory to check for legacy per-module config files. "
        "Matching values are used as fallback defaults, then legacy files are deleted.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Print detailed progress to stderr",
    )
    return parser.parse_args()


def load_yaml_file(path: str) -> dict:
    """Load a YAML file, returning empty dict if file doesn't exist."""
    file_path = Path(path)
    if not file_path.exists():
        return {}
    with open(file_path, "r", encoding="utf-8") as f:
        content = yaml.safe_load(f)
    return content if content else {}


def load_json_file(path: str) -> dict:
    """Load a JSON file."""
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


# Keys that live at config root (shared across all modules)
_CORE_KEYS = frozenset(
    {"user_name", "communication_language", "document_output_language", "output_folder"}
)


def load_legacy_values(
    legacy_dir: str, module_code: str, module_yaml: dict, verbose: bool = False
) -> tuple[dict, dict, list]:
    """Read legacy per-module config files and return core/module value dicts.

    Reads {legacy_dir}/core/config.yaml and {legacy_dir}/{module_code}/config.yaml.
    Only returns values whose keys match the current schema (core keys or module.yaml
    variable definitions). Other modules' directories are not touched.

    Returns:
        (legacy_core, legacy_module, files_found) where files_found lists paths read.
    """
    legacy_core: dict = {}
    legacy_module: dict = {}
    files_found: list = []

    # Read core legacy config
    core_path = Path(legacy_dir) / "core" / "config.yaml"
    if core_path.exists():
        core_data = load_yaml_file(str(core_path))
        files_found.append(str(core_path))
        for k, v in core_data.items():
            if k in _CORE_KEYS:
                legacy_core[k] = v
        if verbose:
            print(f"Legacy core config: {list(legacy_core.keys())}", file=sys.stderr)

    # Read module legacy config
    mod_path = Path(legacy_dir) / module_code / "config.yaml"
    if mod_path.exists():
        mod_data = load_yaml_file(str(mod_path))
        files_found.append(str(mod_path))
        for k, v in mod_data.items():
            if k in _CORE_KEYS:
                # Core keys duplicated in module config — only use if not already set
                if k not in legacy_core:
                    legacy_core[k] = v
            elif k in module_yaml and isinstance(module_yaml[k], dict):
                # Module-specific key that matches a current variable definition
                legacy_module[k] = v
        if verbose:
            print(
                f"Legacy module config: {list(legacy_module.keys())}", file=sys.stderr
            )

    return legacy_core, legacy_module, files_found


def apply_legacy_defaults(answers: dict, legacy_core: dict, legacy_module: dict) -> dict:
    """Apply legacy values as fallback defaults under the answers.

    Legacy values fill in any key not already present in answers.
    Explicit answers always win.
    """
    merged = dict(answers)

    if legacy_core:
        core = merged.get("core", {})
        filled_core = dict(legacy_core)  # legacy as base
        filled_core.update(core)  # answers override
        merged["core"] = filled_core

    if legacy_module:
        mod = merged.get("module", {})
        filled_mod = dict(legacy_module)  # legacy as base
        filled_mod.update(mod)  # answers override
        merged["module"] = filled_mod

    return merged


def cleanup_legacy_configs(
    legacy_dir: str, module_code: str, verbose: bool = False
) -> list:
    """Delete legacy config.yaml files for this module and core only.

    Returns list of deleted file paths.
    """
    deleted = []
    for subdir in (module_code, "core"):
        legacy_path = Path(legacy_dir) / subdir / "config.yaml"
        if legacy_path.exists():
            if verbose:
                print(f"Deleting legacy config: {legacy_path}", file=sys.stderr)
            legacy_path.unlink()
            deleted.append(str(legacy_path))
    return deleted


def extract_module_metadata(module_yaml: dict) -> dict:
    """Extract non-variable metadata fields from module.yaml."""
    meta = {}
    for k in ("name", "description"):
        if k in module_yaml:
            meta[k] = module_yaml[k]
    meta["version"] = module_yaml.get("module_version")  # null if absent
    if "default_selected" in module_yaml:
        meta["default_selected"] = module_yaml["default_selected"]
    return meta


def apply_result_templates(
    module_yaml: dict, module_answers: dict, verbose: bool = False
) -> dict:
    """Apply result templates from module.yaml to transform raw answer values.

    For each answer, if the corresponding variable definition in module.yaml has
    a 'result' field, replaces {value} in that template with the answer. Skips
    the template if the answer already contains '{project-root}' to prevent
    double-prefixing.
    """
    transformed = {}
    for key, value in module_answers.items():
        var_def = module_yaml.get(key)
        if (
            isinstance(var_def, dict)
            and "result" in var_def
            and "{project-root}" not in str(value)
        ):
            template = var_def["result"]
            transformed[key] = template.replace("{value}", str(value))
            if verbose:
                print(
                    f"Applied result template for '{key}': {value} → {transformed[key]}",
                    file=sys.stderr,
                )
        else:
            transformed[key] = value
    return transformed


def merge_config(
    existing_config: dict,
    module_yaml: dict,
    answers: dict,
    verbose: bool = False,
) -> dict:
    """Merge answers into config, applying anti-zombie pattern.

    Args:
        existing_config: Current config.yaml contents (may be empty)
        module_yaml: The module definition
        answers: JSON with 'core' and/or 'module' keys
        verbose: Print progress to stderr

    Returns:
        Updated config dict ready to write
    """
    config = dict(existing_config)
    module_code = module_yaml.get("code")

    if not module_code:
        print("Error: module.yaml must have a 'code' field", file=sys.stderr)
        sys.exit(1)

    # Migrate legacy core: section to root
    if "core" in config and isinstance(config["core"], dict):
        if verbose:
            print("Migrating legacy 'core' section to root", file=sys.stderr)
        config.update(config.pop("core"))

    # Strip user-only keys from config — they belong exclusively in config.user.yaml
    for key in _CORE_USER_KEYS:
        if key in config:
            if verbose:
                print(f"Removing user-only key '{key}' from config (belongs in config.user.yaml)", file=sys.stderr)
            del config[key]

    # Write core values at root (global properties, not nested under "core")
    # Exclude user-only keys — those belong exclusively in config.user.yaml
    core_answers = answers.get("core")
    if core_answers:
        shared_core = {k: v for k, v in core_answers.items() if k not in _CORE_USER_KEYS}
        if shared_core:
            if verbose:
                print(f"Writing core config at root: {list(shared_core.keys())}", file=sys.stderr)
            config.update(shared_core)

    # Anti-zombie: remove existing module section
    if module_code in config:
        if verbose:
            print(
                f"Removing existing '{module_code}' section (anti-zombie)",
                file=sys.stderr,
            )
        del config[module_code]

    # Build module section: metadata + variable values
    module_section = extract_module_metadata(module_yaml)
    module_answers = apply_result_templates(
        module_yaml, answers.get("module", {}), verbose
    )
    module_section.update(module_answers)

    if verbose:
        print(
            f"Writing '{module_code}' section with keys: {list(module_section.keys())}",
            file=sys.stderr,
        )

    config[module_code] = module_section

    return config


# Core keys that are always written to config.user.yaml
_CORE_USER_KEYS = ("user_name", "communication_language")


def extract_user_settings(module_yaml: dict, answers: dict) -> dict:
    """Collect settings that belong in config.user.yaml.

    Includes user_name and communication_language from core answers, plus any
    module variable whose definition contains user_setting: true.
    """
    user_settings = {}

    core_answers = answers.get("core", {})
    for key in _CORE_USER_KEYS:
        if key in core_answers:
            user_settings[key] = core_answers[key]

    module_answers = answers.get("module", {})
    for var_name, var_def in module_yaml.items():
        if isinstance(var_def, dict) and var_def.get("user_setting") is True:
            if var_name in module_answers:
                user_settings[var_name] = module_answers[var_name]

    return user_settings


def write_config(config: dict, config_path: str, verbose: bool = False) -> None:
    """Write config dict to YAML file, creating parent dirs as needed."""
    path = Path(config_path)
    path.parent.mkdir(parents=True, exist_ok=True)

    if verbose:
        print(f"Writing config to {path}", file=sys.stderr)

    with open(path, "w", encoding="utf-8") as f:
        yaml.dump(
            config,
            f,
            default_flow_style=False,
            allow_unicode=True,
            sort_keys=False,
        )


def main():
    args = parse_args()

    # Load inputs
    module_yaml = load_yaml_file(args.module_yaml)
    if not module_yaml:
        print(f"Error: Could not load module.yaml from {args.module_yaml}", file=sys.stderr)
        sys.exit(1)

    answers = load_json_file(args.answers)
    existing_config = load_yaml_file(args.config_path)

    if args.verbose:
        exists = Path(args.config_path).exists()
        print(f"Config file exists: {exists}", file=sys.stderr)
        if exists:
            print(f"Existing sections: {list(existing_config.keys())}", file=sys.stderr)

    # Legacy migration: read old per-module configs as fallback defaults
    legacy_files_found = []
    if args.legacy_dir:
        module_code = module_yaml.get("code", "")
        legacy_core, legacy_module, legacy_files_found = load_legacy_values(
            args.legacy_dir, module_code, module_yaml, args.verbose
        )
        if legacy_core or legacy_module:
            answers = apply_legacy_defaults(answers, legacy_core, legacy_module)
            if args.verbose:
                print("Applied legacy values as fallback defaults", file=sys.stderr)

    # Merge and write config.yaml
    updated_config = merge_config(existing_config, module_yaml, answers, args.verbose)
    write_config(updated_config, args.config_path, args.verbose)

    # Merge and write config.user.yaml
    user_settings = extract_user_settings(module_yaml, answers)
    existing_user_config = load_yaml_file(args.user_config_path)
    updated_user_config = dict(existing_user_config)
    updated_user_config.update(user_settings)
    if user_settings:
        write_config(updated_user_config, args.user_config_path, args.verbose)

    # Legacy cleanup: delete old per-module config files
    legacy_deleted = []
    if args.legacy_dir:
        legacy_deleted = cleanup_legacy_configs(
            args.legacy_dir, module_yaml["code"], args.verbose
        )

    # Output result summary as JSON
    module_code = module_yaml["code"]
    result = {
        "status": "success",
        "config_path": str(Path(args.config_path).resolve()),
        "user_config_path": str(Path(args.user_config_path).resolve()),
        "module_code": module_code,
        "core_updated": bool(answers.get("core")),
        "module_keys": list(updated_config.get(module_code, {}).keys()),
        "user_keys": list(user_settings.keys()),
        "legacy_configs_found": legacy_files_found,
        "legacy_configs_deleted": legacy_deleted,
    }
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# dependencies = []
# ///
"""Merge module help entries into shared _bmad/module-help.csv.

Reads a source CSV with module help entries and merges them into a target CSV.
Uses an anti-zombie pattern: all existing rows matching the source module code
are removed before appending fresh rows.

Legacy cleanup: when --legacy-dir and --module-code are provided, deletes old
per-module module-help.csv files from {legacy-dir}/{module-code}/ and
{legacy-dir}/core/. Only the current module and core are touched.

Exit codes: 0=success, 1=validation error, 2=runtime error
"""

import argparse
import csv
import json
import sys
from io import StringIO
from pathlib import Path

# CSV header for module-help.csv
HEADER = [
    "module",
    "skill",
    "display-name",
    "menu-code",
    "description",
    "action",
    "args",
    "phase",
    "after",
    "before",
    "required",
    "output-location",
    "outputs",
]


def parse_args():
    parser = argparse.ArgumentParser(
        description="Merge module help entries into shared _bmad/module-help.csv with anti-zombie pattern."
    )
    parser.add_argument(
        "--target",
        required=True,
        help="Path to the target _bmad/module-help.csv file",
    )
    parser.add_argument(
        "--source",
        required=True,
        help="Path to the source module-help.csv with entries to merge",
    )
    parser.add_argument(
        "--legacy-dir",
        help="Path to _bmad/ directory to check for legacy per-module CSV files.",
    )
    parser.add_argument(
        "--module-code",
        help="Module code (required with --legacy-dir for scoping cleanup).",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Print detailed progress to stderr",
    )
    return parser.parse_args()


def read_csv_rows(path: str) -> tuple[list[str], list[list[str]]]:
    """Read CSV file returning (header, data_rows).

    Returns empty header and rows if file doesn't exist.
    """
    file_path = Path(path)
    if not file_path.exists():
        return [], []

    with open(file_path, "r", encoding="utf-8", newline="") as f:
        content = f.read()

    reader = csv.reader(StringIO(content))
    rows = list(reader)

    if not rows:
        return [], []

    return rows[0], rows[1:]


def extract_module_codes(rows: list[list[str]]) -> set[str]:
    """Extract unique module codes from data rows."""
    codes = set()
    for row in rows:
        if row and row[0].strip():
            codes.add(row[0].strip())
    return codes


def filter_rows(rows: list[list[str]], module_code: str) -> list[list[str]]:
    """Remove all rows matching the given module code."""
    return [row for row in rows if not row or row[0].strip() != module_code]


def write_csv(path: str, header: list[str], rows: list[list[str]], verbose: bool = False) -> None:
    """Write header + rows to CSV file, creating parent dirs as needed."""
    file_path = Path(path)
    file_path.parent.mkdir(parents=True, exist_ok=True)

    if verbose:
        print(f"Writing {len(rows)} data rows to {path}", file=sys.stderr)

    with open(file_path, "w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(header)
        for row in rows:
            writer.writerow(row)


def cleanup_legacy_csvs(
    legacy_dir: str, module_code: str, verbose: bool = False
) -> list:
    """Delete legacy per-module module-help.csv files for this module and core only.

    Returns list of deleted file paths.
    """
    deleted = []
    for subdir in (module_code, "core"):
        legacy_path = Path(legacy_dir) / subdir / "module-help.csv"
        if legacy_path.exists():
            if verbose:
                print(f"Deleting legacy CSV: {legacy_path}", file=sys.stderr)
            legacy_path.unlink()
            deleted.append(str(legacy_path))
    return deleted


def main():
    args = parse_args()

    # Read source entries
    source_header, source_rows = read_csv_rows(args.source)
    if not source_rows:
        print(f"Error: No data rows found in source {args.source}", file=sys.stderr)
        sys.exit(1)

    # Determine module codes being merged
    source_codes = extract_module_codes(source_rows)
    if not source_codes:
        print("Error: Could not determine module code from source rows", file=sys.stderr)
        sys.exit(1)

    if args.verbose:
        print(f"Source module codes: {source_codes}", file=sys.stderr)
        print(f"Source rows: {len(source_rows)}", file=sys.stderr)

    # Read existing target (may not exist)
    target_header, target_rows = read_csv_rows(args.target)
    target_existed = Path(args.target).exists()

    if args.verbose:
        print(f"Target exists: {target_existed}", file=sys.stderr)
        if target_existed:
            print(f"Existing target rows: {len(target_rows)}", file=sys.stderr)

    # Use source header if target doesn't exist or has no header
    header = target_header if target_header else (source_header if source_header else HEADER)

    # Anti-zombie: remove all rows for each source module code
    filtered_rows = target_rows
    removed_count = 0
    for code in source_codes:
        before_count = len(filtered_rows)
        filtered_rows = filter_rows(filtered_rows, code)
        removed_count += before_count - len(filtered_rows)

    if args.verbose and removed_count > 0:
        print(f"Removed {removed_count} existing rows (anti-zombie)", file=sys.stderr)

    # Append source rows
    merged_rows = filtered_rows + source_rows

    # Write result
    write_csv(args.target, header, merged_rows, args.verbose)

    # Legacy cleanup: delete old per-module CSV files
    legacy_deleted = []
    if args.legacy_dir:
        if not args.module_code:
            print(
                "Error: --module-code is required when --legacy-dir is provided",
                file=sys.stderr,
            )
            sys.exit(1)
        legacy_deleted = cleanup_legacy_csvs(
            args.legacy_dir, args.module_code, args.verbose
        )

    # Output result summary as JSON
    result = {
        "status": "success",
        "target_path": str(Path(args.target).resolve()),
        "target_existed": target_existed,
        "module_codes": sorted(source_codes),
        "rows_removed": removed_count,
        "rows_added": len(source_rows),
        "total_rows": len(merged_rows),
        "legacy_csvs_deleted": legacy_deleted,
    }
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml"]
# ///
"""
Dream recall metrics calculator for Dream Weaver.

Analyzes journal entries to calculate recall rates, streaks,
vividness trends, and quality distributions.

Usage:
    uv run scripts/recall_metrics.py --journal-path PATH [--verbose]
"""

import argparse
import json
import sys
from collections import Counter, defaultdict
from datetime import datetime, timedelta
from pathlib import Path

import yaml


def parse_frontmatter(file_path: Path) -> dict | None:
    """Extract YAML frontmatter from a markdown file."""
    try:
        content = file_path.read_text(encoding="utf-8")
        if not content.startswith("---"):
            return None
        end = content.index("---", 3)
        return yaml.safe_load(content[3:end])
    except (ValueError, yaml.YAMLError):
        return None


def scan_journal(journal_path: Path) -> list[dict]:
    """Scan all journal entries and extract metadata."""
    entries = []
    for file in sorted(journal_path.glob("*.md")):
        fm = parse_frontmatter(file)
        if not fm:
            continue

        entry_date = fm.get("date")
        if isinstance(entry_date, str):
            try:
                entry_date = datetime.strptime(entry_date, "%Y-%m-%d").date()
            except ValueError:
                continue

        entries.append({
            "file": file.name,
            "date": entry_date,
            "vividness": fm.get("vividness"),
            "recall_quality": fm.get("recall_quality", "medium"),
            "lucid": fm.get("lucid", False),
        })

    return entries


def calculate_metrics(entries: list[dict]) -> dict:
    """Calculate recall metrics from journal entries."""
    if not entries:
        return {
            "total_dreams": 0,
            "dreams_per_week": 0,
            "current_streak": 0,
            "longest_streak": 0,
            "avg_vividness": 0,
            "vividness_trend": "insufficient_data",
            "quality_distribution": {},
            "lucid_count": 0,
            "weekly_counts": [],
        }

    # Group by date
    dreams_by_date = defaultdict(list)
    for entry in entries:
        if entry["date"]:
            dreams_by_date[entry["date"]].append(entry)

    sorted_dates = sorted(dreams_by_date.keys())

    if not sorted_dates:
        return {"total_dreams": len(entries), "error": "no_dated_entries"}

    # Date range
    first_date = sorted_dates[0]
    last_date = sorted_dates[-1]
    total_days = max((last_date - first_date).days, 1)
    total_weeks = max(total_days / 7, 1)

    # Dreams per week
    dreams_per_week = round(len(entries) / total_weeks, 1)

    # Streak calculation
    today = datetime.now().date()
    current_streak = 0
    check_date = today
    while check_date in dreams_by_date or check_date == today:
        if check_date in dreams_by_date:
            current_streak += 1
        elif check_date != today:
            break
        check_date -= timedelta(days=1)

    longest_streak = 0
    streak = 0
    for i, date in enumerate(sorted_dates):
        if i == 0 or (date - sorted_dates[i - 1]).days == 1:
            streak += 1
        else:
            longest_streak = max(longest_streak, streak)
            streak = 1
    longest_streak = max(longest_streak, streak)

    # Vividness
    vividness_scores = [
        e["vividness"] for e in entries if e["vividness"] is not None
    ]
    avg_vividness = (
        round(sum(vividness_scores) / len(vividness_scores), 1)
        if vividness_scores
        else 0
    )

    # Vividness trend (compare first half to second half)
    vividness_trend = "insufficient_data"
    if len(vividness_scores) >= 4:
        mid = len(vividness_scores) // 2
        first_half = sum(vividness_scores[:mid]) / mid
        second_half = sum(vividness_scores[mid:]) / (len(vividness_scores) - mid)
        diff = second_half - first_half
        if diff > 0.5:
            vividness_trend = "improving"
        elif diff < -0.5:
            vividness_trend = "declining"
        else:
            vividness_trend = "stable"

    # Quality distribution
    quality_counts = Counter(e["recall_quality"] for e in entries)

    # Lucid count
    lucid_count = sum(1 for e in entries if e.get("lucid"))

    # Weekly counts (last 8 weeks)
    weekly_counts = []
    for weeks_ago in range(7, -1, -1):
        week_start = today - timedelta(weeks=weeks_ago, days=today.weekday())
        week_end = week_start + timedelta(days=6)
        count = sum(
            len(dreams)
            for date, dreams in dreams_by_date.items()
            if week_start <= date <= week_end
        )
        weekly_counts.append({
            "week_start": str(week_start),
            "count": count,
        })

    return {
        "total_dreams": len(entries),
        "dreams_per_week": dreams_per_week,
        "current_streak": current_streak,
        "longest_streak": longest_streak,
        "avg_vividness": avg_vividness,
        "vividness_trend": vividness_trend,
        "quality_distribution": dict(quality_counts),
        "lucid_count": lucid_count,
        "weekly_counts": weekly_counts,
        "date_range": {
            "first": str(first_date),
            "last": str(last_date),
            "total_days": total_days,
        },
    }


def main():
    parser = argparse.ArgumentParser(
        description="Calculate dream recall metrics"
    )
    parser.add_argument(
        "--journal-path", required=True, help="Path to journal folder"
    )
    parser.add_argument(
        "--verbose", action="store_true", help="Print diagnostics to stderr"
    )
    args = parser.parse_args()

    journal_path = Path(args.journal_path)
    if not journal_path.is_dir():
        print(
            json.dumps({
                "script": "recall_metrics",
                "status": "error",
                "error": f"Journal path not found: {journal_path}",
            }),
            file=sys.stdout,
        )
        sys.exit(2)

    entries = scan_journal(journal_path)

    if args.verbose:
        print(f"Found {len(entries)} journal entries", file=sys.stderr)

    metrics = calculate_metrics(entries)

    output = {
        "script": "recall_metrics",
        "version": "1.0.0",
        "journal_path": str(journal_path),
        "timestamp": datetime.now().isoformat(),
        "status": "pass",
        "metrics": metrics,
    }

    print(json.dumps(output, indent=2))
    sys.exit(0)


if __name__ == "__main__":
    main()

# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml"]
# ///
"""
Dream seed tracking and success rate analysis for Dream Weaver.

Reads seed-log.yaml and calculates incubation success rates,
technique effectiveness, and correlation stats.

Usage:
    uv run scripts/seed_tracker.py --seed-log PATH [--verbose]
"""

import argparse
import json
import sys
from collections import Counter
from datetime import datetime
from pathlib import Path

import yaml


def load_seed_log(seed_log_path: Path) -> list[dict]:
    """Load and parse seed-log.yaml."""
    try:
        content = seed_log_path.read_text(encoding="utf-8")
        data = yaml.safe_load(content)
        if not data or "seeds" not in data:
            return []
        return data["seeds"]
    except (yaml.YAMLError, FileNotFoundError):
        return []


def analyze_seeds(seeds: list[dict]) -> dict:
    """Analyze seed success rates and technique effectiveness."""
    if not seeds:
        return {
            "total_seeds": 0,
            "success_rate": 0,
            "technique_stats": {},
            "result_distribution": {},
        }

    # Result distribution
    result_counts = Counter()
    technique_results = {}

    for seed in seeds:
        result = seed.get("result", "pending")
        technique = seed.get("technique", "unknown")

        result_counts[result] += 1

        if technique not in technique_results:
            technique_results[technique] = Counter()
        technique_results[technique][result] += 1

    # Overall success rate (partial + full / total non-pending)
    resolved = sum(
        count for result, count in result_counts.items() if result != "pending"
    )
    successes = result_counts.get("partial", 0) + result_counts.get("full", 0)
    success_rate = round(successes / resolved, 2) if resolved > 0 else 0

    # Technique effectiveness
    technique_stats = {}
    for technique, results in technique_results.items():
        tech_resolved = sum(
            count for result, count in results.items() if result != "pending"
        )
        tech_successes = results.get("partial", 0) + results.get("full", 0)
        technique_stats[technique] = {
            "total": sum(results.values()),
            "resolved": tech_resolved,
            "successes": tech_successes,
            "success_rate": (
                round(tech_successes / tech_resolved, 2) if tech_resolved > 0 else 0
            ),
            "results": dict(results),
        }

    # Recent trend (last 5 resolved seeds)
    resolved_seeds = [s for s in seeds if s.get("result") not in ("pending", None)]
    recent = resolved_seeds[-5:] if len(resolved_seeds) >= 5 else resolved_seeds
    recent_successes = sum(
        1 for s in recent if s.get("result") in ("partial", "full")
    )
    recent_rate = (
        round(recent_successes / len(recent), 2) if recent else 0
    )

    return {
        "total_seeds": len(seeds),
        "pending": result_counts.get("pending", 0),
        "resolved": resolved,
        "success_rate": success_rate,
        "recent_trend_rate": recent_rate,
        "result_distribution": dict(result_counts),
        "technique_stats": technique_stats,
        "best_technique": (
            max(technique_stats, key=lambda t: technique_stats[t]["success_rate"])
            if technique_stats
            else None
        ),
    }


def main():
    parser = argparse.ArgumentParser(
        description="Analyze dream seed success rates"
    )
    parser.add_argument(
        "--seed-log", required=True, help="Path to seed-log.yaml"
    )
    parser.add_argument(
        "--verbose", action="store_true", help="Print diagnostics to stderr"
    )
    args = parser.parse_args()

    seed_log_path = Path(args.seed_log)
    if not seed_log_path.is_file():
        print(
            json.dumps({
                "script": "seed_tracker",
                "status": "error",
                "error": f"Seed log not found: {seed_log_path}",
            }),
            file=sys.stdout,
        )
        sys.exit(2)

    seeds = load_seed_log(seed_log_path)

    if args.verbose:
        print(f"Found {len(seeds)} seed entries", file=sys.stderr)

    analysis = analyze_seeds(seeds)

    output = {
        "script": "seed_tracker",
        "version": "1.0.0",
        "seed_log_path": str(seed_log_path),
        "timestamp": datetime.now().isoformat(),
        "status": "pass",
        "analysis": analysis,
    }

    print(json.dumps(output, indent=2))
    sys.exit(0)


if __name__ == "__main__":
    main()

# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml"]
# ///
"""
Symbol frequency analysis for Dream Weaver journal entries.

Scans journal folder for dream entries with YAML frontmatter,
extracts symbols, and outputs frequency statistics as JSON.

Usage:
    uv run scripts/symbol_stats.py --journal-path PATH [--days N] [--verbose]
"""

import argparse
import json
import sys
from collections import Counter, defaultdict
from datetime import datetime, timedelta
from pathlib import Path

import yaml


def parse_frontmatter(file_path: Path) -> dict | None:
    """Extract YAML frontmatter from a markdown file."""
    try:
        content = file_path.read_text(encoding="utf-8")
        if not content.startswith("---"):
            return None
        end = content.index("---", 3)
        return yaml.safe_load(content[3:end])
    except (ValueError, yaml.YAMLError):
        return None


def scan_journal(journal_path: Path, days: int | None = None) -> list[dict]:
    """Scan journal entries and extract frontmatter data."""
    entries = []
    cutoff = None
    if days:
        cutoff = datetime.now().date() - timedelta(days=days)

    for file in sorted(journal_path.glob("*.md")):
        fm = parse_frontmatter(file)
        if not fm or "symbols" not in fm:
            continue

        entry_date = fm.get("date")
        if isinstance(entry_date, str):
            try:
                entry_date = datetime.strptime(entry_date, "%Y-%m-%d").date()
            except ValueError:
                continue

        if cutoff and entry_date and entry_date < cutoff:
            continue

        entries.append({
            "file": file.name,
            "date": str(entry_date) if entry_date else None,
            "symbols": fm.get("symbols", []),
            "emotions": fm.get("emotions", []),
            "vividness": fm.get("vividness"),
            "lucid": fm.get("lucid", False),
        })

    return entries


def analyze_symbols(entries: list[dict]) -> dict:
    """Analyze symbol frequency and emotion correlations."""
    symbol_count = Counter()
    symbol_emotions = defaultdict(Counter)
    symbol_dates = defaultdict(list)
    symbol_contexts = defaultdict(set)

    for entry in entries:
        symbols = entry.get("symbols", [])
        emotions = entry.get("emotions", [])
        date = entry.get("date")

        for symbol in symbols:
            symbol = symbol.lower().strip()
            symbol_count[symbol] += 1
            if date:
                symbol_dates[symbol].append(date)
            for emotion in emotions:
                symbol_emotions[symbol][emotion.lower().strip()] += 1

    results = {}
    for symbol, count in symbol_count.most_common():
        dates = sorted(symbol_dates[symbol])
        results[symbol] = {
            "count": count,
            "first_seen": dates[0] if dates else None,
            "last_seen": dates[-1] if dates else None,
            "emotion_correlation": dict(symbol_emotions[symbol]),
        }

    return results


def main():
    parser = argparse.ArgumentParser(
        description="Analyze dream journal symbol frequency"
    )
    parser.add_argument(
        "--journal-path", required=True, help="Path to journal folder"
    )
    parser.add_argument(
        "--days", type=int, default=None, help="Limit to last N days"
    )
    parser.add_argument(
        "--verbose", action="store_true", help="Print diagnostics to stderr"
    )
    args = parser.parse_args()

    journal_path = Path(args.journal_path)
    if not journal_path.is_dir():
        print(
            json.dumps({
                "script": "symbol_stats",
                "status": "error",
                "error": f"Journal path not found: {journal_path}",
            }),
            file=sys.stdout,
        )
        sys.exit(2)

    entries = scan_journal(journal_path, args.days)

    if args.verbose:
        print(f"Found {len(entries)} journal entries", file=sys.stderr)

    symbols = analyze_symbols(entries)

    output = {
        "script": "symbol_stats",
        "version": "1.0.0",
        "journal_path": str(journal_path),
        "timestamp": datetime.now().isoformat(),
        "status": "pass",
        "entries_scanned": len(entries),
        "unique_symbols": len(symbols),
        "symbols": symbols,
        "summary": {
            "total_symbols": sum(s["count"] for s in symbols.values()),
            "unique_symbols": len(symbols),
            "top_5": [
                {"symbol": s, "count": symbols[s]["count"]}
                for s in list(symbols.keys())[:5]
            ],
        },
    }

    print(json.dumps(output, indent=2))
    sys.exit(0)


if __name__ == "__main__":
    main()

# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml", "pytest"]
# ///
"""Tests for recall_metrics.py."""

import json
import sys
from datetime import datetime, timedelta
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parent.parent))
from recall_metrics import calculate_metrics, parse_frontmatter, scan_journal


@pytest.fixture
def journal_dir(tmp_path):
    """Create a journal directory with entries spanning multiple days."""
    journal = tmp_path / "journal"
    journal.mkdir()

    today = datetime.now().date()
    for i in range(7):
        day = today - timedelta(days=i)
        entry = journal / f"{day}-1.md"
        vividness = 5 + (i % 4)
        lucid = "true" if i == 2 else "false"
        quality = "high" if i < 3 else "medium"
        entry.write_text(
            f"---\ndate: {day}\nsequence: 1\nvividness: {vividness}\n"
            f"lucid: {lucid}\nrecall_quality: {quality}\n"
            f"emotions: [curiosity]\nsymbols: [water]\nseeded: false\n---\n\nDream.\n"
        )

    return journal


@pytest.fixture
def empty_journal(tmp_path):
    journal = tmp_path / "journal"
    journal.mkdir()
    return journal


class TestParseFrontmatter:
    def test_valid(self, tmp_path):
        f = tmp_path / "test.md"
        f.write_text("---\ndate: 2026-03-10\nvividness: 7\n---\nContent")
        result = parse_frontmatter(f)
        assert result["vividness"] == 7

    def test_missing(self, tmp_path):
        f = tmp_path / "test.md"
        f.write_text("No frontmatter")
        assert parse_frontmatter(f) is None


class TestScanJournal:
    def test_scans_all(self, journal_dir):
        entries = scan_journal(journal_dir)
        assert len(entries) == 7

    def test_empty(self, empty_journal):
        entries = scan_journal(empty_journal)
        assert len(entries) == 0

    def test_extracts_vividness(self, journal_dir):
        entries = scan_journal(journal_dir)
        assert all(e["vividness"] is not None for e in entries)


class TestCalculateMetrics:
    def test_basic_metrics(self, journal_dir):
        entries = scan_journal(journal_dir)
        metrics = calculate_metrics(entries)
        assert metrics["total_dreams"] == 7
        assert metrics["dreams_per_week"] > 0
        assert metrics["current_streak"] > 0
        assert metrics["avg_vividness"] > 0

    def test_empty_entries(self):
        metrics = calculate_metrics([])
        assert metrics["total_dreams"] == 0
        assert metrics["dreams_per_week"] == 0
        assert metrics["current_streak"] == 0

    def test_quality_distribution(self, journal_dir):
        entries = scan_journal(journal_dir)
        metrics = calculate_metrics(entries)
        assert "high" in metrics["quality_distribution"]
        assert "medium" in metrics["quality_distribution"]

    def test_lucid_count(self, journal_dir):
        entries = scan_journal(journal_dir)
        metrics = calculate_metrics(entries)
        assert metrics["lucid_count"] == 1

    def test_weekly_counts(self, journal_dir):
        entries = scan_journal(journal_dir)
        metrics = calculate_metrics(entries)
        assert len(metrics["weekly_counts"]) == 8

    def test_vividness_trend_with_data(self, journal_dir):
        entries = scan_journal(journal_dir)
        metrics = calculate_metrics(entries)
        assert metrics["vividness_trend"] in (
            "improving", "declining", "stable", "insufficient_data"
        )

    def test_streak_calculation(self, journal_dir):
        entries = scan_journal(journal_dir)
        metrics = calculate_metrics(entries)
        assert metrics["longest_streak"] >= metrics["current_streak"]

# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml", "pytest"]
# ///
"""Tests for seed_tracker.py."""

import json
import sys
from pathlib import Path

import pytest
import yaml

sys.path.insert(0, str(Path(__file__).parent.parent))
from seed_tracker import analyze_seeds, load_seed_log


@pytest.fixture
def seed_log(tmp_path):
    """Create a seed-log.yaml with mixed results."""
    log = tmp_path / "seed-log.yaml"
    data = {
        "seeds": [
            {
                "date": "2026-03-01",
                "intention": "Dream about the ocean",
                "technique": "visualization",
                "result": "full",
                "dream_ref": "2026-03-02-1",
                "notes": "Dreamed of swimming in deep water",
            },
            {
                "date": "2026-03-03",
                "intention": "Meet my grandmother",
                "technique": "mantra",
                "result": "none",
                "dream_ref": None,
                "notes": None,
            },
            {
                "date": "2026-03-05",
                "intention": "Fly over mountains",
                "technique": "visualization",
                "result": "partial",
                "dream_ref": "2026-03-06-1",
                "notes": "Floated but didn't fly",
            },
            {
                "date": "2026-03-08",
                "intention": "Open the locked door",
                "technique": "symbol-return",
                "result": "pending",
                "dream_ref": None,
                "notes": None,
            },
            {
                "date": "2026-03-10",
                "intention": "Explore underwater",
                "technique": "question",
                "result": "full",
                "dream_ref": "2026-03-11-1",
                "notes": "Breathed underwater",
            },
        ],
        "success_rate": 0.5,
    }
    log.write_text(yaml.dump(data))
    return log


@pytest.fixture
def empty_seed_log(tmp_path):
    log = tmp_path / "seed-log.yaml"
    log.write_text(yaml.dump({"seeds": []}))
    return log


class TestLoadSeedLog:
    def test_loads_seeds(self, seed_log):
        seeds = load_seed_log(seed_log)
        assert len(seeds) == 5

    def test_empty_log(self, empty_seed_log):
        seeds = load_seed_log(empty_seed_log)
        assert len(seeds) == 0

    def test_missing_file(self, tmp_path):
        seeds = load_seed_log(tmp_path / "nonexistent.yaml")
        assert len(seeds) == 0

    def test_invalid_yaml(self, tmp_path):
        f = tmp_path / "bad.yaml"
        f.write_text(": [invalid yaml {{")
        seeds = load_seed_log(f)
        assert len(seeds) == 0


class TestAnalyzeSeeds:
    def test_basic_analysis(self, seed_log):
        seeds = load_seed_log(seed_log)
        result = analyze_seeds(seeds)
        assert result["total_seeds"] == 5
        assert result["pending"] == 1
        assert result["resolved"] == 4

    def test_success_rate(self, seed_log):
        seeds = load_seed_log(seed_log)
        result = analyze_seeds(seeds)
        # 3 successes (full + partial) out of 4 resolved = 0.75
        assert result["success_rate"] == 0.75

    def test_technique_stats(self, seed_log):
        seeds = load_seed_log(seed_log)
        result = analyze_seeds(seeds)
        assert "visualization" in result["technique_stats"]
        assert result["technique_stats"]["visualization"]["total"] == 2

    def test_best_technique(self, seed_log):
        seeds = load_seed_log(seed_log)
        result = analyze_seeds(seeds)
        assert result["best_technique"] is not None

    def test_empty_seeds(self):
        result = analyze_seeds([])
        assert result["total_seeds"] == 0
        assert result["success_rate"] == 0

    def test_result_distribution(self, seed_log):
        seeds = load_seed_log(seed_log)
        result = analyze_seeds(seeds)
        dist = result["result_distribution"]
        assert dist["full"] == 2
        assert dist["partial"] == 1
        assert dist["none"] == 1
        assert dist["pending"] == 1

    def test_recent_trend(self, seed_log):
        seeds = load_seed_log(seed_log)
        result = analyze_seeds(seeds)
        assert 0 <= result["recent_trend_rate"] <= 1

# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml", "pytest"]
# ///
"""Tests for symbol_stats.py."""

import json
import sys
from datetime import datetime, timedelta
from pathlib import Path

import pytest

# Add parent to path for import
sys.path.insert(0, str(Path(__file__).parent.parent))
from symbol_stats import analyze_symbols, parse_frontmatter, scan_journal


@pytest.fixture
def journal_dir(tmp_path):
    """Create a temporary journal directory with sample entries."""
    journal = tmp_path / "journal"
    journal.mkdir()

    today = datetime.now().date()
    yesterday = today - timedelta(days=1)

    entry1 = journal / f"{today}-1.md"
    entry1.write_text(
        f"---\ndate: {today}\nsequence: 1\nvividness: 7\nlucid: false\n"
        f"emotions: [anxiety, curiosity]\nsymbols: [water, doorway]\n"
        f"recall_quality: high\nseeded: false\n---\n\nDream narrative here.\n"
    )

    entry2 = journal / f"{yesterday}-1.md"
    entry2.write_text(
        f"---\ndate: {yesterday}\nsequence: 1\nvividness: 5\nlucid: true\n"
        f"emotions: [peace, awe]\nsymbols: [water, flying]\n"
        f"recall_quality: medium\nseeded: true\n---\n\nAnother dream.\n"
    )

    return journal


@pytest.fixture
def empty_journal(tmp_path):
    """Create an empty journal directory."""
    journal = tmp_path / "journal"
    journal.mkdir()
    return journal


class TestParseFrontmatter:
    def test_valid_frontmatter(self, tmp_path):
        f = tmp_path / "test.md"
        f.write_text("---\ndate: 2026-03-10\nsymbols: [water]\n---\nContent")
        result = parse_frontmatter(f)
        assert result is not None
        assert result["symbols"] == ["water"]

    def test_no_frontmatter(self, tmp_path):
        f = tmp_path / "test.md"
        f.write_text("No frontmatter here.")
        assert parse_frontmatter(f) is None

    def test_invalid_yaml(self, tmp_path):
        f = tmp_path / "test.md"
        f.write_text("---\n: invalid: yaml: [[\n---\nContent")
        assert parse_frontmatter(f) is None


class TestScanJournal:
    def test_scans_entries(self, journal_dir):
        entries = scan_journal(journal_dir)
        assert len(entries) == 2

    def test_empty_journal(self, empty_journal):
        entries = scan_journal(empty_journal)
        assert len(entries) == 0

    def test_days_filter(self, journal_dir):
        entries = scan_journal(journal_dir, days=0)
        # With days=0, cutoff is today, so yesterday's entry is excluded
        assert len(entries) <= 2

    def test_extracts_symbols(self, journal_dir):
        entries = scan_journal(journal_dir)
        all_symbols = [s for e in entries for s in e["symbols"]]
        assert "water" in all_symbols
        assert "doorway" in all_symbols


class TestAnalyzeSymbols:
    def test_basic_analysis(self, journal_dir):
        entries = scan_journal(journal_dir)
        result = analyze_symbols(entries)
        assert "water" in result
        assert result["water"]["count"] == 2

    def test_emotion_correlation(self, journal_dir):
        entries = scan_journal(journal_dir)
        result = analyze_symbols(entries)
        assert "anxiety" in result["water"]["emotion_correlation"]

    def test_empty_entries(self):
        result = analyze_symbols([])
        assert result == {}

    def test_first_last_seen(self, journal_dir):
        entries = scan_journal(journal_dir)
        result = analyze_symbols(entries)
        assert result["water"]["first_seen"] is not None
        assert result["water"]["last_seen"] is not None