session_memory_common.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
import tomllib
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

SESSION_MEMORY_HOME = Path(
    os.environ.get("SESSION_MEMORY_HOME", Path.home() / ".session-memory")
).expanduser()
LEGACY_GLOBAL_SESSION_MEMORY_DIRS = (
    Path(os.environ.get("CODEX_HOME", Path.home() / ".codex")).expanduser()
    / "session-memory",
    Path(os.environ.get("CLAUDE_HOME", Path.home() / ".claude")).expanduser()
    / "session-memory",
)
GLOBAL_SESSION_MEMORY_DIR = SESSION_MEMORY_HOME
REGISTRY_PATH = GLOBAL_SESSION_MEMORY_DIR / "registry.json"
STATE_PATH = GLOBAL_SESSION_MEMORY_DIR / "state.json"
SLEEP_LOCK_PATH = GLOBAL_SESSION_MEMORY_DIR / "sleep.lock"
DREAMS_DIR = GLOBAL_SESSION_MEMORY_DIR / "dreams"
LOGS_DIR = GLOBAL_SESSION_MEMORY_DIR / "logs"
SESSION_MEMORY_FILES = ("current.md", "history.md", "research.md", "dream-notes.md")
CONFIG_TABLE_NAMES = ("spaces", "routes")
PROJECT_SESSION_MEMORY_DIR = ".session-memory"
LEGACY_PROJECT_SESSION_MEMORY_DIRS = (
    Path(".codex") / "session-memory",
    Path(".claude") / "session-memory",
)
CURRENT_LAST_UPDATED_FIELD = "Last Updated"
DREAM_LAST_UPDATED_FIELDS = ("Last Updated", "Generated At")
DEFAULT_AGING_DAYS = 7
DEFAULT_STALE_DAYS = 30


@dataclass(frozen=True)
class SpaceRoute:
    name: str
    relative_path: str
    root: Path


@dataclass(frozen=True)
class FreshnessInfo:
    status: str
    source: str
    raw_value: str
    age_minutes: int | None
    age_days: int | None


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def write_if_missing(path: Path, content: str, force: bool) -> None:
    if path.exists() and not force:
        return
    path.write_text(content, encoding="utf-8")


def find_git_root(start: Path) -> Path | None:
    current = start.resolve()
    for candidate in (current, *current.parents):
        if (candidate / ".git").exists():
            return candidate
    return None


def session_memory_dir(root: Path) -> Path:
    return root / PROJECT_SESSION_MEMORY_DIR


def legacy_session_memory_dirs(root: Path) -> tuple[Path, ...]:
    return tuple(root / legacy_dir for legacy_dir in LEGACY_PROJECT_SESSION_MEMORY_DIRS)


def unique_child_path(parent: Path, name: str) -> Path:
    candidate = parent / name
    if not candidate.exists():
        return candidate

    stem = candidate.stem
    suffix = candidate.suffix
    index = 1
    while True:
        next_candidate = parent / f"{stem}-{index}{suffix}"
        if not next_candidate.exists():
            return next_candidate
        index += 1


def legacy_import_dir(target_dir: Path, source_dir: Path) -> Path:
    source_label = "-".join(part for part in source_dir.parts[-3:] if part)
    timestamp = iso_now().replace(":", "").replace("+", "Z")
    return target_dir / "archive" / "legacy-imports" / f"{source_label}-{timestamp}"


def merge_registry_file(target_path: Path, source_path: Path) -> bool:
    if target_path.name != "registry.json" or source_path.name != "registry.json":
        return False

    try:
        target_registry = json.loads(target_path.read_text(encoding="utf-8"))
        source_registry = json.loads(source_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return False

    if not isinstance(target_registry, dict) or not isinstance(source_registry, dict):
        return False

    target_registry.setdefault("version", 1)
    target_registry.setdefault("projects", {})
    source_registry.setdefault("projects", {})
    migrate_registry_payload(target_registry)
    migrate_registry_payload(source_registry)

    for key, entry in source_registry.get("projects", {}).items():
        if not isinstance(entry, dict):
            continue
        existing = target_registry["projects"].get(key)
        if isinstance(existing, dict):
            target_registry["projects"][key] = merge_registry_entries(existing, entry)
        else:
            target_registry["projects"][key] = entry

    _write_json(target_path, target_registry)
    source_path.unlink()
    return True


def remove_empty_legacy_parent(source_dir: Path) -> None:
    parent = source_dir.parent
    if parent.name not in {".codex", ".claude"}:
        return
    try:
        parent.rmdir()
    except OSError:
        pass


def move_directory_contents(
    source_dir: Path,
    target_dir: Path,
    *,
    remove_empty_parent: bool = False,
) -> None:
    target_dir.mkdir(parents=True, exist_ok=True)
    conflict_dir: Path | None = None

    for child in sorted(source_dir.iterdir()):
        destination = target_dir / child.name
        if not destination.exists():
            shutil.move(str(child), str(destination))
            continue

        if merge_registry_file(destination, child):
            continue

        if conflict_dir is None:
            conflict_dir = legacy_import_dir(target_dir, source_dir)
            conflict_dir.mkdir(parents=True, exist_ok=True)
        shutil.move(str(child), str(unique_child_path(conflict_dir, child.name)))

    source_dir.rmdir()
    if remove_empty_parent:
        remove_empty_legacy_parent(source_dir)


def migrate_session_memory_dirs(
    target_dir: Path,
    legacy_dirs: tuple[Path, ...],
    *,
    remove_empty_parent: bool = False,
) -> list[str]:
    migrated: list[str] = []
    target_resolved = target_dir.resolve(strict=False)

    for source_dir in legacy_dirs:
        if not source_dir.exists() or not source_dir.is_dir():
            continue
        if source_dir.resolve(strict=False) == target_resolved:
            continue
        move_directory_contents(
            source_dir,
            target_dir,
            remove_empty_parent=remove_empty_parent,
        )
        migrated.append(str(source_dir))

    return migrated


def migrate_project_session_memory(root: Path) -> list[str]:
    return migrate_session_memory_dirs(
        session_memory_dir(root),
        legacy_session_memory_dirs(root),
        remove_empty_parent=True,
    )


def migrate_global_session_memory() -> list[str]:
    migrated = migrate_session_memory_dirs(
        GLOBAL_SESSION_MEMORY_DIR,
        tuple(
            legacy_dir
            for legacy_dir in LEGACY_GLOBAL_SESSION_MEMORY_DIRS
            if legacy_dir != GLOBAL_SESSION_MEMORY_DIR
        ),
    )
    migrated.extend(normalize_registry_file())
    return migrated


def project_root_from_session_memory_path(path: Path) -> Path | None:
    resolved = path.expanduser().resolve(strict=False)
    parts = resolved.parts

    new_marker = (PROJECT_SESSION_MEMORY_DIR,)
    for index in range(0, len(parts) - len(new_marker) + 1):
        if parts[index : index + len(new_marker)] == new_marker:
            return Path(*parts[:index])

    for legacy_dir in LEGACY_PROJECT_SESSION_MEMORY_DIRS:
        marker = legacy_dir.parts
        marker_size = len(marker)
        for index in range(0, len(parts) - marker_size + 1):
            if parts[index : index + marker_size] == marker:
                return Path(*parts[:index])

    return None


def remap_legacy_session_memory_path(path: Path) -> Path:
    expanded = path.expanduser()
    resolved = expanded.resolve(strict=False)

    for legacy_dir in LEGACY_GLOBAL_SESSION_MEMORY_DIRS:
        legacy_resolved = legacy_dir.resolve(strict=False)
        if resolved == legacy_resolved or resolved.is_relative_to(legacy_resolved):
            return GLOBAL_SESSION_MEMORY_DIR.resolve(strict=False) / resolved.relative_to(
                legacy_resolved
            )

    parts = resolved.parts
    for legacy_dir in LEGACY_PROJECT_SESSION_MEMORY_DIRS:
        marker = legacy_dir.parts
        marker_size = len(marker)
        for index in range(0, len(parts) - marker_size + 1):
            if parts[index : index + marker_size] != marker:
                continue
            project_root = Path(*parts[:index])
            remainder = parts[index + marker_size :]
            target = project_root / PROJECT_SESSION_MEMORY_DIR
            if remainder:
                target = target.joinpath(*remainder)
            return target

    return expanded


def iter_registry_project_roots(registry: dict[str, Any]) -> set[Path]:
    projects = registry.get("projects")
    if not isinstance(projects, dict):
        return set()

    path_fields = (
        "target_dir",
        "current_path",
        "history_path",
        "research_path",
        "dream_notes_path",
        "last_dream_notes_path",
    )
    roots: set[Path] = set()
    for key, entry in projects.items():
        key_root = project_root_from_session_memory_path(Path(str(key)))
        if key_root is not None:
            roots.add(key_root)
        if not isinstance(entry, dict):
            continue
        for field in path_fields:
            value = entry.get(field)
            if not isinstance(value, str) or not value:
                continue
            value_root = project_root_from_session_memory_path(Path(value))
            if value_root is not None:
                roots.add(value_root)
    return roots


def migrate_registry_project_memories(registry: dict[str, Any]) -> list[str]:
    migrated: list[str] = []
    for project_root in sorted(iter_registry_project_roots(registry)):
        migrated.extend(migrate_project_session_memory(project_root))
    return migrated


def merge_registry_entries(
    existing: dict[str, Any],
    incoming: dict[str, Any],
) -> dict[str, Any]:
    merged = dict(incoming)
    merged.update({key: value for key, value in existing.items() if value not in (None, "")})

    timestamp_fields = (
        "last_seen_at",
        "last_active_at",
        "last_dream_at",
        "last_dream_consumed_at",
        "last_history_archive_at",
        "last_research_archive_at",
    )
    for field in timestamp_fields:
        existing_value = existing.get(field)
        incoming_value = incoming.get(field)
        existing_dt = parse_iso(existing_value)
        incoming_dt = parse_iso(incoming_value)
        if existing_dt is None:
            chosen = incoming_value
        elif incoming_dt is None:
            chosen = existing_value
        else:
            chosen = existing_value if existing_dt >= incoming_dt else incoming_value
        if chosen:
            merged[field] = chosen

    return merged


def migrate_registry_payload(registry: dict[str, Any]) -> bool:
    projects = registry.get("projects")
    if not isinstance(projects, dict):
        registry["projects"] = {}
        return True

    migrated_projects: dict[str, dict[str, Any]] = {}
    changed = False
    path_fields = (
        "target_dir",
        "current_path",
        "history_path",
        "research_path",
        "dream_notes_path",
        "last_dream_notes_path",
    )

    for key, entry in projects.items():
        if not isinstance(entry, dict):
            changed = True
            continue

        migrated_entry = dict(entry)
        for field in path_fields:
            value = migrated_entry.get(field)
            if not isinstance(value, str) or not value:
                continue
            remapped = remap_legacy_session_memory_path(Path(value))
            if str(remapped) != value:
                migrated_entry[field] = str(remapped)
                changed = True

        remapped_key = remap_legacy_session_memory_path(Path(str(key)))
        final_key = str(migrated_entry.get("target_dir") or remapped_key)
        if final_key != str(key):
            changed = True

        existing = migrated_projects.get(final_key)
        if existing is None:
            migrated_projects[final_key] = migrated_entry
        else:
            migrated_projects[final_key] = merge_registry_entries(existing, migrated_entry)
            changed = True

    if changed:
        registry["projects"] = migrated_projects
    return changed


def normalize_registry_file() -> list[str]:
    if not REGISTRY_PATH.exists():
        return []

    try:
        registry = json.loads(REGISTRY_PATH.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return []

    if not isinstance(registry, dict):
        return []

    migrated = migrate_registry_project_memories(registry)
    changed = migrate_registry_payload(registry)
    if changed:
        _write_json(REGISTRY_PATH, registry)
    return migrated


def session_memory_config_path(root: Path) -> Path:
    return session_memory_dir(root) / "config.toml"


def legacy_session_memory_config_paths(root: Path) -> tuple[Path, ...]:
    return tuple(legacy_dir / "config.toml" for legacy_dir in legacy_session_memory_dirs(root))


def has_session_memory_state(root: Path) -> bool:
    candidate_dirs = (session_memory_dir(root), *legacy_session_memory_dirs(root))
    return any(
        candidate_dir.exists()
        and any((candidate_dir / name).exists() for name in SESSION_MEMORY_FILES)
        for candidate_dir in candidate_dirs
    )


def find_nearest_config_root(start: Path) -> Path | None:
    current = start.resolve()
    for candidate in (current, *current.parents):
        config_paths = (
            session_memory_config_path(candidate),
            *legacy_session_memory_config_paths(candidate),
        )
        if any(config_path.exists() for config_path in config_paths):
            return candidate
    return None


def find_nearest_memory_root(start: Path) -> Path | None:
    current = start.resolve()
    for candidate in (current, *current.parents):
        if has_session_memory_state(candidate):
            return candidate
    return None


def normalize_route_path(project_root: Path, raw_path: str) -> tuple[str, Path]:
    raw_text = raw_path.strip()
    if not raw_text:
        raise SystemExit("Invalid session-memory config: route path cannot be empty.")

    candidate = Path(raw_text)
    if candidate.is_absolute():
        raise SystemExit(
            f"Invalid session-memory config: absolute paths are not allowed: {raw_text}"
        )

    clean_parts = [part for part in candidate.parts if part not in ("", ".")]
    if not clean_parts:
        raise SystemExit(
            f"Invalid session-memory config: route path must point to a subdirectory: {raw_text}"
        )
    if any(part == ".." for part in clean_parts):
        raise SystemExit(
            f"Invalid session-memory config: route path cannot escape project root: {raw_path}"
        )

    relative_path = Path(*clean_parts)
    resolved_root = (project_root / relative_path).resolve()
    if not resolved_root.is_relative_to(project_root):
        raise SystemExit(
            f"Invalid session-memory config: route path must stay inside project root: {raw_path}"
        )

    return relative_path.as_posix(), resolved_root


def load_space_routes(project_root: Path) -> tuple[Path | None, list[SpaceRoute]]:
    config_path = session_memory_config_path(project_root)
    if not config_path.exists():
        return None, []

    try:
        payload = tomllib.loads(config_path.read_text(encoding="utf-8"))
    except tomllib.TOMLDecodeError as exc:
        raise SystemExit(f"Invalid session-memory config {config_path}: {exc}") from exc

    route_entries: list[tuple[str, str]] = []
    for table_name in CONFIG_TABLE_NAMES:
        table = payload.get(table_name)
        if table is None:
            continue
        if not isinstance(table, dict):
            raise SystemExit(
                f"Invalid session-memory config {config_path}: [{table_name}] must be a table."
            )
        for name, route_path in table.items():
            if not isinstance(route_path, str):
                raise SystemExit(
                    f"Invalid session-memory config {config_path}: route {name!r} must map to a string path."
                )
            route_entries.append((str(name).strip(), route_path))

    if not route_entries:
        for name, route_path in payload.items():
            if name in CONFIG_TABLE_NAMES or not isinstance(route_path, str):
                continue
            route_entries.append((str(name).strip(), route_path))

    routes: list[SpaceRoute] = []
    seen_names: set[str] = set()
    seen_paths: set[str] = set()
    for name, route_path in route_entries:
        if not name:
            raise SystemExit(
                f"Invalid session-memory config {config_path}: route name cannot be empty."
            )
        if name in seen_names:
            raise SystemExit(
                f"Invalid session-memory config {config_path}: duplicated route name {name!r}."
            )
        seen_names.add(name)
        relative_path, route_root = normalize_route_path(project_root, route_path)
        if relative_path in seen_paths:
            raise SystemExit(
                f"Invalid session-memory config {config_path}: duplicated route path {relative_path!r}."
            )
        seen_paths.add(relative_path)
        routes.append(
            SpaceRoute(name=name, relative_path=relative_path, root=route_root)
        )

    routes.sort(key=lambda item: len(item.root.parts), reverse=True)
    return config_path, routes


def match_space_route(base: Path, routes: list[SpaceRoute]) -> SpaceRoute | None:
    current = base.resolve()
    for route in routes:
        if current == route.root or current.is_relative_to(route.root):
            return route
    return None


def resolve_project_root(base: Path) -> tuple[Path, str]:
    git_root = find_git_root(base)
    if git_root is not None:
        return git_root, "git-root"

    config_root = find_nearest_config_root(base)
    if config_root is not None:
        return config_root, "project-root"

    memory_root = find_nearest_memory_root(base)
    if memory_root is not None:
        return memory_root, "project-root"

    return base.resolve(), "workspace"


def resolve_target_dir(base: Path, scope: str) -> tuple[Path, str]:
    paths = resolve_memory_paths(base, scope)
    return Path(paths["target_dir"]), str(paths["scope"])


def resolve_memory_paths(base: Path, scope: str) -> dict[str, Any]:
    workspace = base.resolve()
    migrated_from: list[str] = []
    if scope == "global":
        migrated_from.extend(migrate_global_session_memory())
        target_dir = GLOBAL_SESSION_MEMORY_DIR / "global"
        return {
            "workspace": workspace,
            "project_root": target_dir,
            "context_root": target_dir,
            "scope": "global",
            "root_scope": "global",
            "target_dir": target_dir,
            "current": target_dir / "current.md",
            "history": target_dir / "history.md",
            "research": target_dir / "research.md",
            "dream_notes": target_dir / "dream-notes.md",
            "config_path": "",
            "space_name": "",
            "space_path": "",
            "migrated_from": migrated_from,
        }

    if scope == "workspace":
        project_root = workspace
        context_root = workspace
        migrated_from.extend(migrate_project_session_memory(workspace))
        target_dir = session_memory_dir(workspace)
        resolved_scope = "workspace"
        root_scope = "workspace"
        config_path = ""
        space_name = ""
        space_path = ""
    else:
        project_root, root_scope = resolve_project_root(workspace)
        migrated_from.extend(migrate_project_session_memory(project_root))
        config_candidate, routes = load_space_routes(project_root)
        matched_route = match_space_route(workspace, routes) if routes else None
        context_root = matched_route.root if matched_route else project_root
        if context_root != project_root:
            migrated_from.extend(migrate_project_session_memory(context_root))
        target_dir = session_memory_dir(context_root)
        resolved_scope = "configured-space" if matched_route else root_scope
        config_path = str(config_candidate) if config_candidate else ""
        space_name = matched_route.name if matched_route else ""
        space_path = matched_route.relative_path if matched_route else ""

    return {
        "workspace": workspace,
        "project_root": project_root,
        "context_root": context_root,
        "scope": resolved_scope,
        "root_scope": root_scope,
        "target_dir": target_dir,
        "current": target_dir / "current.md",
        "history": target_dir / "history.md",
        "research": target_dir / "research.md",
        "dream_notes": target_dir / "dream-notes.md",
        "config_path": config_path,
        "space_name": space_name,
        "space_path": space_path,
        "migrated_from": migrated_from,
    }


def iter_resolution_summary(paths: dict[str, Any]) -> list[str]:
    lines = [
        f"workspace={paths['workspace']}",
        f"project_root={paths['project_root']}",
        f"context_root={paths['context_root']}",
        f"scope={paths['scope']}",
        f"target_dir={paths['target_dir']}",
    ]
    if paths.get("config_path"):
        lines.append(f"config_path={paths['config_path']}")
    if paths.get("space_name"):
        lines.append(f"space_name={paths['space_name']}")
    if paths.get("space_path"):
        lines.append(f"space_path={paths['space_path']}")
    for migrated_from in paths.get("migrated_from") or []:
        lines.append(f"migrated_from={migrated_from}")
    return lines


def ensure_global_layout() -> None:
    migrate_global_session_memory()
    GLOBAL_SESSION_MEMORY_DIR.mkdir(parents=True, exist_ok=True)
    DREAMS_DIR.mkdir(parents=True, exist_ok=True)
    LOGS_DIR.mkdir(parents=True, exist_ok=True)


def iso_now() -> str:
    return datetime.now(UTC).replace(microsecond=0).isoformat()


def parse_iso(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def find_metadata_value(text: str, field_names: tuple[str, ...]) -> tuple[str, str] | None:
    prefixes = tuple(f"- {field_name}:" for field_name in field_names)
    for line in text.splitlines():
        stripped = line.strip()
        for field_name, prefix in zip(field_names, prefixes):
            if stripped.startswith(prefix):
                return field_name, stripped[len(prefix) :].strip()
    return None


def upsert_metadata_value(text: str, field_name: str, value: str) -> str:
    line_value = f"- {field_name}: {value}"
    lines = text.splitlines()
    target_prefix = f"- {field_name}:"

    for index, line in enumerate(lines):
        if line.strip().startswith(target_prefix):
            lines[index] = line_value
            return "\n".join(lines).rstrip() + "\n"

    insert_at = 1 if lines and lines[0].startswith("# ") else 0
    lines.insert(insert_at, line_value)
    return "\n".join(lines).rstrip() + "\n"


def update_file_timestamp(
    path: Path,
    field_name: str = CURRENT_LAST_UPDATED_FIELD,
    timestamp: str | None = None,
) -> str:
    if not path.exists():
        raise FileNotFoundError(path)
    text = path.read_text(encoding="utf-8")
    value = timestamp or iso_now()
    path.write_text(upsert_metadata_value(text, field_name, value), encoding="utf-8")
    return value


def freshness_from_timestamp(
    timestamp: datetime,
    source: str,
    raw_value: str,
    aging_days: int,
    stale_days: int,
) -> FreshnessInfo:
    age = datetime.now(UTC) - timestamp.astimezone(UTC)
    age_minutes = int(age.total_seconds() // 60)
    age_days = age.days
    if age_days >= stale_days:
        status = "stale"
    elif age_days >= aging_days:
        status = "aging"
    else:
        status = "fresh"
    return FreshnessInfo(
        status=status,
        source=source,
        raw_value=raw_value,
        age_minutes=age_minutes,
        age_days=age_days,
    )


def missing_freshness() -> FreshnessInfo:
    return FreshnessInfo(
        status="missing",
        source="missing",
        raw_value="",
        age_minutes=None,
        age_days=None,
    )


def get_path_freshness(
    path: Path,
    field_names: tuple[str, ...],
    aging_days: int = DEFAULT_AGING_DAYS,
    stale_days: int = DEFAULT_STALE_DAYS,
    fallback_to_mtime: bool = True,
) -> FreshnessInfo:
    if not path.exists():
        return missing_freshness()

    text = path.read_text(encoding="utf-8")
    metadata = find_metadata_value(text, field_names)
    if metadata is not None:
        field_name, raw_value = metadata
        timestamp = parse_iso(raw_value)
        if timestamp is not None:
            return freshness_from_timestamp(
                timestamp=timestamp,
                source=f"metadata:{field_name}",
                raw_value=raw_value,
                aging_days=aging_days,
                stale_days=stale_days,
            )

    if fallback_to_mtime:
        modified = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC)
        return freshness_from_timestamp(
            timestamp=modified,
            source="mtime",
            raw_value=modified.astimezone(UTC).replace(microsecond=0).isoformat(),
            aging_days=aging_days,
            stale_days=stale_days,
        )

    return missing_freshness()


def _read_json(path: Path, default: dict[str, Any]) -> dict[str, Any]:
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return default


def _write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        "w", encoding="utf-8", delete=False, dir=path.parent
    ) as handle:
        json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
        handle.write("\n")
        temp_path = Path(handle.name)
    temp_path.replace(path)


def load_registry() -> dict[str, Any]:
    ensure_global_layout()
    registry = _read_json(REGISTRY_PATH, {"version": 1, "projects": {}})
    registry.setdefault("version", 1)
    registry.setdefault("projects", {})
    return registry


def save_registry(registry: dict[str, Any]) -> None:
    ensure_global_layout()
    _write_json(REGISTRY_PATH, registry)


def load_state() -> dict[str, Any]:
    ensure_global_layout()
    state = _read_json(STATE_PATH, {"version": 1})
    state.setdefault("version", 1)
    return state


def save_state(state: dict[str, Any]) -> None:
    ensure_global_layout()
    _write_json(STATE_PATH, state)


def project_key(paths: dict[str, Any]) -> str:
    return str(paths["target_dir"])


def upsert_project_registry(
    paths: dict[str, Any],
    action: str,
    active_at: str | None = None,
    mark_active: bool = False,
) -> dict[str, Any]:
    registry = load_registry()
    key = project_key(paths)
    now = active_at or iso_now()
    projects = registry["projects"]
    entry = projects.get(key, {})
    entry.update(
        {
            "workspace": str(paths["workspace"]),
            "project_root": str(paths["project_root"]),
            "context_root": str(paths["context_root"]),
            "scope": str(paths["scope"]),
            "root_scope": str(paths.get("root_scope") or paths["scope"]),
            "target_dir": str(paths["target_dir"]),
            "current_path": str(paths["current"]),
            "history_path": str(paths["history"]),
            "research_path": str(paths["research"]),
            "dream_notes_path": str(paths["dream_notes"]),
            "config_path": str(paths.get("config_path") or ""),
            "space_name": str(paths.get("space_name") or ""),
            "space_path": str(paths.get("space_path") or ""),
            "last_action": action,
            "last_seen_at": now,
        }
    )
    if mark_active or not entry.get("last_active_at"):
        entry["last_active_at"] = now
    projects[key] = entry
    save_registry(registry)
    return entry


def update_project_registry(
    target_dir: str,
    **updates: Any,
) -> dict[str, Any] | None:
    registry = load_registry()
    entry = registry.get("projects", {}).get(target_dir)
    if entry is None:
        return None
    entry.update(updates)
    registry["projects"][target_dir] = entry
    save_registry(registry)
    return entry


def get_project_entry(paths: dict[str, Any]) -> dict[str, Any] | None:
    registry = load_registry()
    return registry.get("projects", {}).get(project_key(paths))


def record_dream_consumed(paths: dict[str, Any], consumed_at: str) -> None:
    update_project_registry(project_key(paths), last_dream_consumed_at=consumed_at)


def due_for_dream(entry: dict[str, Any]) -> bool:
    last_active = parse_iso(entry.get("last_active_at"))
    last_dream = parse_iso(entry.get("last_dream_at"))
    if last_active is None:
        return False
    if last_dream is None:
        return True
    return last_active > last_dream


def iter_due_projects() -> list[dict[str, Any]]:
    registry = load_registry()
    due_entries: list[dict[str, Any]] = []
    for entry in registry.get("projects", {}).values():
        current_path = Path(entry["current_path"])
        history_path = Path(entry["history_path"])
        research_path = Path(
            entry.get("research_path") or Path(entry["target_dir"]) / "research.md"
        )
        if not current_path.exists() and not history_path.exists() and not research_path.exists():
            continue
        if due_for_dream(entry):
            due_entries.append(entry)
    return due_entries


def _process_running(pid: int) -> bool:
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False


def read_sleep_lock() -> dict[str, Any] | None:
    if not SLEEP_LOCK_PATH.exists():
        return None
    return _read_json(SLEEP_LOCK_PATH, {})


def release_sleep_lock() -> None:
    if SLEEP_LOCK_PATH.exists():
        SLEEP_LOCK_PATH.unlink()


def lock_is_active(grace_minutes: int = 10) -> bool:
    lock = read_sleep_lock()
    if not lock:
        return False

    pid = int(lock.get("pid") or 0)
    if pid > 0 and _process_running(pid):
        return True

    started_at = parse_iso(lock.get("started_at"))
    if pid == 0 and started_at is not None:
        if datetime.now(UTC) - started_at < timedelta(minutes=grace_minutes):
            return True

    release_sleep_lock()
    return False


def acquire_sleep_lock(trigger_action: str, due_count: int) -> bool:
    ensure_global_layout()
    if lock_is_active():
        return False

    payload = {
        "started_at": iso_now(),
        "pid": 0,
        "trigger_action": trigger_action,
        "due_count": due_count,
    }

    try:
        fd = os.open(SLEEP_LOCK_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    except FileExistsError:
        return False

    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
            handle.write("\n")
    except Exception:
        try:
            SLEEP_LOCK_PATH.unlink()
        except FileNotFoundError:
            pass
        raise

    return True


def update_sleep_lock(**updates: Any) -> None:
    lock = read_sleep_lock() or {}
    lock.update(updates)
    _write_json(SLEEP_LOCK_PATH, lock)


def spawn_background_dream(trigger_action: str) -> int:
    ensure_global_layout()
    script_path = Path(__file__).resolve().parent / "dream_session_memory.py"
    log_path = LOGS_DIR / "dream.log"
    log_handle = open(log_path, "a", encoding="utf-8")
    process = subprocess.Popen(
        [sys.executable, str(script_path), "--trigger-action", trigger_action],
        stdout=log_handle,
        stderr=subprocess.STDOUT,
        stdin=subprocess.DEVNULL,
        start_new_session=True,
    )
    update_sleep_lock(pid=process.pid, started_at=iso_now())
    return process.pid


def run_preflight(
    paths: dict[str, Any],
    action: str,
    sleep_threshold_hours: float = 5.0,
    mark_active: bool = False,
) -> dict[str, Any]:
    now = iso_now()
    upsert_project_registry(paths, action=action, active_at=now, mark_active=mark_active)

    state = load_state()
    last_check = parse_iso(state.get("last_sleep_check_at"))
    threshold = timedelta(hours=sleep_threshold_hours)
    result: dict[str, Any] = {
        "registered": True,
        "project": project_key(paths),
        "sleep_check": {
            "status": "not-run",
            "threshold_hours": sleep_threshold_hours,
        },
    }

    if last_check is not None and datetime.now(UTC) - last_check < threshold:
        result["sleep_check"]["status"] = "threshold-not-met"
        result["sleep_check"]["last_sleep_check_at"] = state.get("last_sleep_check_at")
        return result

    state["last_sleep_check_at"] = now
    save_state(state)

    due_projects = iter_due_projects()
    result["sleep_check"]["due_projects"] = len(due_projects)
    result["sleep_check"]["last_sleep_check_at"] = now

    if not due_projects:
        result["sleep_check"]["status"] = "no-dream-needed"
        return result

    if not acquire_sleep_lock(trigger_action=action, due_count=len(due_projects)):
        result["sleep_check"]["status"] = "already-running"
        return result

    try:
        pid = spawn_background_dream(trigger_action=action)
    except Exception:
        release_sleep_lock()
        raise

    result["sleep_check"]["status"] = "started"
    result["sleep_check"]["pid"] = pid
    return result