Recall Conversations

#!/usr/bin/env python3
"""
List/resolve projects in the memory database.

Maps each project NAME to its canonical path + encoded key (the transcript-dir
name), with session counts and date span. Exists because project names are NOT
unique — the same basename (e.g. two 'EzyCopy') maps to different paths, so
`recent_chats.py --project NAME` can silently pull the wrong project or merge
both. Resolve here first, then target by the disambiguated name (and verify the
path) before mining.

Doubles as the sibling-directory candidate generator: --match TOKEN surfaces
every project whose name OR path contains the token, so a project scattered
across CWDs (e.g. a skill that began inside another repo) shows all its homes.

Returns markdown by default, JSON with --json or --format json.
"""

from __future__ import annotations

import argparse
import json
import sqlite3
import sys
from pathlib import Path

from memory_lib.cli_common import emit_error, open_db_or_exit, resolve_format, PLUGIN_VERSION
from memory_lib.db import DEFAULT_DB_PATH


EXAMPLES = """
EXAMPLES
  All projects, most-recently-active first:
    list_projects.py

  Resolve an ambiguous name to its path(s) + key(s):
    list_projects.py --match EzyCopy

  Find every home of a scattered project (sibling-dir discovery):
    list_projects.py --match wiki --json | jq '.projects[] | {name, path, sessions}'
"""


def get_projects(conn: sqlite3.Connection, match: str | None) -> list[dict]:
    """Return one row per project: name, path, key, session_count, date span."""
    cursor = conn.cursor()
    sql = """
        SELECT p.name, p.path, p.key,
               COUNT(DISTINCT s.id) AS sessions,
               MIN(b.started_at) AS first_seen,
               MAX(b.ended_at) AS last_seen
        FROM projects p
        LEFT JOIN sessions s ON s.project_id = p.id
        LEFT JOIN branches b ON b.session_id = s.id AND b.is_active = 1
    """
    params: list = []
    if match:
        sql += " WHERE p.name LIKE ? ESCAPE '\\' OR p.path LIKE ? ESCAPE '\\'"
        escaped = match.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        params.extend([f"%{escaped}%", f"%{escaped}%"])
    sql += " GROUP BY p.id ORDER BY last_seen DESC"

    cursor.execute(sql, params)
    return [
        {
            "name": name,
            "path": path,
            "key": key,
            "sessions": sessions,
            "first_seen": first_seen,
            "last_seen": last_seen,
        }
        for name, path, key, sessions, first_seen, last_seen in cursor.fetchall()
    ]


def format_markdown(rows: list[dict], match: str | None) -> str:
    if not rows:
        return f"No projects found{f' matching {match!r}' if match else ''}."
    header = f"# Projects ({len(rows)}{f', matching {match!r}' if match else ''})\n"
    lines = [header]
    for r in rows:
        span = f"{(r['first_seen'] or '?')[:10]}..{(r['last_seen'] or '?')[:10]}"
        lines.append(f"- {r['name']}  [{r['sessions']} sessions, {span}]")
        lines.append(f"    path: {r['path']}")
        lines.append(f"    key:  {r['key']}")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="List/resolve projects in the memory database (name -> path + key).",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=EXAMPLES,
    )
    parser.add_argument("--match", type=str, default=None,
                        help="Substring filter over project name OR path (case-insensitive). "
                             "Use to resolve ambiguous names or discover a project's sibling dirs.")
    fmt_group = parser.add_mutually_exclusive_group()
    fmt_group.add_argument("--format", choices=["markdown", "json"], default="markdown",
                           help="Output format (default: markdown).")
    fmt_group.add_argument("--json", action="store_true", help="Alias for --format json.")
    parser.add_argument("--db", type=Path, default=DEFAULT_DB_PATH,
                        help=f"Database path (default: {DEFAULT_DB_PATH}).")
    parser.add_argument("--version", action="version",
                        version=f"recall-conversations {PLUGIN_VERSION}")

    args = parser.parse_args()
    fmt = resolve_format(args)

    conn = open_db_or_exit(args.db, fmt)
    try:
        rows = get_projects(conn, args.match)
    except Exception as e:
        emit_error("query_failed", str(e), None, fmt)
        sys.exit(1)
    finally:
        conn.close()

    if fmt == "json":
        print(json.dumps({"projects": rows, "count": len(rows), "match": args.match}, indent=2))
    else:
        print(format_markdown(rows, args.match))


if __name__ == "__main__":
    main()

"""
memory_lib — shared utilities for the claude-memory plugin.

Submodules:
  db         — Database connection, schema, settings, logging
  content    — Message content extraction and tool detection
  parsing    — JSONL parsing, branch detection, metadata extraction
  formatting — Session formatting, time/path utilities
"""

from __future__ import annotations

"""
Shared CLI plumbing for recall-conversations scripts.

Provides:
- Common argument parsing (--project, --all-projects, --limit, --json, etc.)
- Project auto-detection from CWD against the projects table
- Structured stderr error/warning emission for agent callers
"""

from __future__ import annotations

import argparse
import json
import os
import sqlite3
import sys
from pathlib import Path
from typing import NamedTuple, Optional

from .db import DEFAULT_DB_PATH


class ScopeFilter(NamedTuple):
    """A resolved project filter: which projects column to match, and the values.

    column is always a literal chosen by our code ("name" or "path"), never user
    input — safe to interpolate into SQL. values are bound as parameters. Empty
    values means "no projects matched" and callers skip the filter.
    """

    column: str
    values: list[str]


def _get_plugin_version() -> str:
    """Read version from plugin.json. Returns 'unknown' on any failure."""
    try:
        plugin_root = Path(__file__).resolve().parents[4]
        plugin_json = plugin_root / ".claude-plugin" / "plugin.json"
        with open(plugin_json) as f:
            return json.load(f).get("version", "unknown")
    except Exception:
        return "unknown"


PLUGIN_VERSION = _get_plugin_version()
LIMIT_MIN = 1
LIMIT_MAX = 50
# Retrieved output above this many characters is "large" — the orchestrator should
# prefer --summary, and if already summarized, fan out by-project subagents.
FANOUT_SUGGEST_CHARS = 50000


def resolve_project(cwd: str, conn: sqlite3.Connection) -> Optional[ScopeFilter]:
    """
    Resolve current project from CWD by walking up the path tree against
    projects.path in the DB. Returns a path-keyed ScopeFilter or None if no match.

    Strategy:
    1. Walk up from CWD; for each ancestor, look up projects.path = ancestor.
       An exact path match is unambiguous — filter by path so projects that merely
       share a basename (duplicate names are expected — see list_projects.py) are
       never merged into one arc.
    2. If no path match, fall back to projects.name = basename(cwd); resolve that
       to the path(s) of every same-named project so the filter stays path-precise.
    3. If both fail, return None — caller decides how to handle (warn + widen).
    """
    cur = os.path.abspath(cwd)
    cursor = conn.cursor()
    while True:
        cursor.execute("SELECT path FROM projects WHERE path = ?", (cur,))
        row = cursor.fetchone()
        if row and row[0]:
            return ScopeFilter("path", [row[0]])
        parent = os.path.dirname(cur)
        if parent == cur:
            break
        cur = parent

    derived = os.path.basename(os.path.abspath(cwd))
    if derived:
        cursor.execute("SELECT path FROM projects WHERE name = ?", (derived,))
        paths = [r[0] for r in cursor.fetchall() if r[0]]
        if paths:
            return ScopeFilter("path", paths)
    return None


def add_common_args(parser: argparse.ArgumentParser, default_limit: int = 5) -> None:
    """Add the shared CLI flags to a parser."""
    parser.add_argument(
        "--limit", "-n", type=int, default=default_limit,
        help=f"Number of sessions ({LIMIT_MIN}-{LIMIT_MAX}, default: {default_limit})"
    )
    parser.add_argument("--n", type=int, dest="limit", help=argparse.SUPPRESS)
    parser.add_argument("--max-results", type=int, dest="limit", help=argparse.SUPPRESS)

    scope = parser.add_mutually_exclusive_group()
    scope.add_argument(
        "--project", type=str,
        help="Filter by project name(s), comma-separated. "
             "Default: auto-detect from CWD."
    )
    scope.add_argument(
        "--all-projects", action="store_true",
        help="Search across all projects (overrides auto-detect)."
    )

    fmt_group = parser.add_mutually_exclusive_group()
    fmt_group.add_argument(
        "--format", choices=["markdown", "json"], default="markdown",
        help="Output format (default: markdown)."
    )
    fmt_group.add_argument(
        "--json", action="store_true",
        help="Alias for --format json."
    )

    parser.add_argument("--verbose", "-v", action="store_true",
                        help="Include files_modified, commits, tool_counts.")
    parser.add_argument("--summary", action="store_true",
                        help="Emit precomputed per-session summaries instead of full message "
                             "content — token-efficient for broad/retro/multi-session queries.")
    parser.add_argument("--include-notifications", action="store_true",
                        help="Include task notification messages (hidden by default).")
    parser.add_argument("--db", type=Path, default=DEFAULT_DB_PATH,
                        help=f"Database path (default: {DEFAULT_DB_PATH}).")
    parser.add_argument("--cwd", type=str, default=None,
                        help="Override CWD for project auto-detect (default: current working directory).")
    parser.add_argument("--version", action="version",
                        version=f"recall-conversations {PLUGIN_VERSION}")


def resolve_format(args: argparse.Namespace) -> str:
    """Resolve --json alias to canonical format value."""
    return "json" if getattr(args, "json", False) else args.format


def validate_limit(args: argparse.Namespace, fmt: str) -> int:
    """Validate --limit is in range; emit structured error and exit on failure."""
    if not (LIMIT_MIN <= args.limit <= LIMIT_MAX):
        emit_error(
            "invalid_limit",
            f"--limit must be in [{LIMIT_MIN},{LIMIT_MAX}], got {args.limit}",
            f"--limit 20  (any value in {LIMIT_MIN}-{LIMIT_MAX})",
            fmt,
        )
        sys.exit(2)
    return args.limit


def resolve_scope(
    args: argparse.Namespace,
    conn: sqlite3.Connection,
    fmt: str,
) -> tuple[Optional[ScopeFilter], bool]:
    """
    Resolve project scope from args. Returns (scope_or_none, auto_detected).

    - If --all-projects: returns (None, False) — no filter.
    - If --project NAME[,NAME]: returns (ScopeFilter("name", [names...]), False) —
      explicit names filter by name, preserving the exact requested set.
    - Otherwise: auto-detect from CWD. On success returns (path-keyed ScopeFilter,
      True) so same-named projects are never merged. On failure, warns and returns
      (None, False) — falls through to all-projects so the user gets *some* result.
    """
    if args.all_projects:
        return None, False
    if args.project:
        names = [p.strip() for p in args.project.split(",") if p.strip()]
        return ScopeFilter("name", names), False

    cwd = args.cwd or os.getcwd()
    scope = resolve_project(cwd, conn)
    if scope is not None:
        return scope, True

    emit_warning(
        f"Could not auto-detect project for CWD {cwd}; searching all projects. "
        "Pass --project NAME or --all-projects to make scope explicit.",
        fmt,
    )
    return None, False


def emit_error(code: str, message: str, hint: Optional[str], fmt: str) -> None:
    """Emit a structured error to stderr."""
    if fmt == "json":
        payload = {"error": code, "message": message, "hint": hint}
        sys.stderr.write(json.dumps(payload) + "\n")
    else:
        sys.stderr.write(f"Error: {message}\n")
        if hint:
            sys.stderr.write(f"Hint: {hint}\n")


def emit_warning(message: str, fmt: str) -> None:
    """Emit a warning to stderr."""
    if fmt == "json":
        sys.stderr.write(json.dumps({"warning": message}) + "\n")
    else:
        sys.stderr.write(f"WARN: {message}\n")


def volume_signal(output_chars: int, summary_mode: bool) -> tuple[bool, str]:
    """Classify retrieved output volume and return (is_large, escalation_hint).

    Drives the fan-out decision mechanically rather than by guessing from session
    count: small answers (the common continuation/lookup case) never escalate.
    """
    if output_chars <= FANOUT_SUGGEST_CHARS:
        return False, ""
    if summary_mode:
        hint = ("summarized output still exceeds the volume budget — fan out general-purpose "
                "subagents sharded by project, then reduce")
    else:
        hint = ("re-run with --summary for precomputed summaries; if still large, fan out "
                "general-purpose subagents sharded by project")
    return True, hint


def volume_flags(content_chars: int, summary_mode: bool) -> dict:
    """Two-tier escalation flags derived from retrieved volume — the single source
    of truth for both recall scripts' JSON meta.

    summary_suggested: full-content pull is large → switch to --summary (tier 1).
    fanout_suggested: even --summary output is large → fan out by project (tier 2).
    The two are mutually exclusive by construction.
    """
    is_large = volume_signal(content_chars, summary_mode)[0]
    return {
        "content_chars": content_chars,
        "summary_suggested": is_large and not summary_mode,
        "fanout_suggested": is_large and summary_mode,
    }


def emit_volume_signal(output_chars: int, session_count: int, summary_mode: bool, fmt: str) -> None:
    """Write a stderr nudge when retrieved output is large (markdown mode only)."""
    if fmt == "json":
        return
    is_large, hint = volume_signal(output_chars, summary_mode)
    if is_large:
        sys.stderr.write(
            f"INFO: retrieved {output_chars} chars across {session_count} sessions; {hint}.\n"
        )


def open_db_or_exit(db_path: Path, fmt: str) -> sqlite3.Connection:
    """Open the DB. If it doesn't exist or fails to open, emit structured error and exit."""
    if not db_path.exists():
        emit_error(
            "db_not_found",
            f"Database not found at {db_path}",
            "ls ~/.claude-memory/",
            fmt,
        )
        sys.exit(1)
    try:
        conn = sqlite3.connect(db_path)
        conn.execute("PRAGMA journal_mode = WAL")
        conn.execute("PRAGMA busy_timeout = 5000")
        return conn
    except sqlite3.OperationalError as e:
        emit_error("db_open_failed", str(e), None, fmt)
        sys.exit(1)

#!/usr/bin/env python3
"""
Message content extraction and tool detection utilities.
"""

from __future__ import annotations

import json
import re


def sanitize_fts_term(term: str) -> str:
    """Remove FTS special characters from search term.

    Strips characters that are FTS operators or special syntax:
    quotes, parentheses, asterisks, and FTS keywords.
    Keeps alphanumeric, spaces, and basic punctuation.
    """
    # Remove quotes, parentheses, asterisks, and word boundaries
    sanitized = re.sub(r'["\(\)*\-^]', '', term)
    # Remove FTS keywords: NEAR, AND, OR, NOT (case-insensitive)
    sanitized = re.sub(r'\b(NEAR|AND|OR|NOT)\b', '', sanitized, flags=re.IGNORECASE)
    # Strip whitespace
    sanitized = sanitized.strip()
    return sanitized


def extract_text_content(content) -> tuple[str, bool, bool, str | None]:
    """
    Extract text from message content.
    Returns: (text, has_tool_use, has_thinking, tool_summary_json)

    tool_summary_json is a JSON string like '{"Bash":3,"Read":2}' or None.
    Tool use markers are NOT materialized into text.
    """
    has_tool_use = False
    has_thinking = False
    tool_counts: dict[str, int] = {}

    if isinstance(content, str):
        # Clean up command artifacts
        text = re.sub(r'<command-name>.*?</command-name>', '', content, flags=re.DOTALL)
        text = re.sub(r'<command-message>.*?</command-message>', '', text, flags=re.DOTALL)
        text = re.sub(r'<command-args>.*?</command-args>', '', text, flags=re.DOTALL)
        text = re.sub(r'<local-command-stdout>.*?</local-command-stdout>', '', text, flags=re.DOTALL)
        text = re.sub(r'<channel\b[^>]*>\n?([\s\S]*?)\n?</channel>', r'\1', text, flags=re.DOTALL)
        return text.strip(), False, False, None

    if isinstance(content, list):
        texts = []
        for item in content:
            if isinstance(item, dict):
                item_type = item.get("type", "")
                if item_type == "text":
                    texts.append(item.get("text", ""))
                elif item_type == "tool_use":
                    has_tool_use = True
                    tool_name = item.get("name", "")
                    if tool_name:
                        tool_counts[tool_name] = tool_counts.get(tool_name, 0) + 1
                elif item_type == "thinking":
                    has_thinking = True
        tool_summary = json.dumps(tool_counts) if tool_counts else None
        return "\n".join(texts).strip(), has_tool_use, has_thinking, tool_summary

    return "", False, False, None


def parse_origin(entry: dict) -> str | None:
    """Extract clean platform name from origin.server (e.g. 'telegram' from 'plugin:telegram:telegram')."""
    origin = entry.get("origin")
    if not origin or not isinstance(origin, dict):
        return None
    server = origin.get("server") or ""
    if not server:
        return None
    # Pattern: "plugin:telegram:telegram" -> "telegram"
    parts = server.split(":")
    if len(parts) >= 2 and parts[1]:
        return parts[1]
    return None


def is_task_notification(content) -> bool:
    """Check if content is a task-notification message (subagent result)."""
    if isinstance(content, list):
        texts = [item.get("text", "") for item in content
                 if isinstance(item, dict) and item.get("type") == "text"]
        text = "\n".join(texts).strip()
    elif isinstance(content, str):
        text = content.strip()
    else:
        return False
    return text.startswith("<task-notification>")


def is_teammate_message(content) -> bool:
    """Detect teammate coordination messages (team reports, idle notifications, shutdown)."""
    if isinstance(content, list):
        texts = [item.get("text", "") for item in content
                 if isinstance(item, dict) and item.get("type") == "text"]
        text = "\n".join(texts).strip()
    elif isinstance(content, str):
        text = content.strip()
    else:
        return False
    return text.startswith("<teammate-message")


def is_tool_result(content) -> bool:
    """Check if content is a tool result (not a real user message)."""
    if isinstance(content, list) and content:
        first = content[0]
        if isinstance(first, dict) and first.get("type") == "tool_result":
            return True
    return False


def extract_files_modified(content) -> list[str]:
    """Extract file paths from Edit/Write/MultiEdit tool uses."""
    files = []
    if isinstance(content, list):
        for item in content:
            if isinstance(item, dict) and item.get("type") == "tool_use":
                name = item.get("name", "")
                inp = item.get("input", {})
                if name in ("Edit", "Write", "MultiEdit") and "file_path" in inp:
                    files.append(inp["file_path"])
    return files


def extract_commits(content) -> list[str]:
    """Extract git commit messages from Bash tool uses."""
    commits = []
    if isinstance(content, list):
        for item in content:
            if isinstance(item, dict) and item.get("type") == "tool_use":
                if item.get("name") == "Bash":
                    cmd = item.get("input", {}).get("command", "")
                    if "git commit" in cmd:
                        m = re.search(r'-m\s+["\']([^"\']+)["\']', cmd)
                        if m:
                            commits.append(m.group(1)[:100])
    return commits

#!/usr/bin/env python3
"""
Database connection, schema management, settings, and logging.
"""

from __future__ import annotations

import json
import logging
import sqlite3
import time
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Optional

# Default paths
DEFAULT_DB_PATH = Path.home() / ".claude-memory" / "conversations.db"
DEFAULT_PROJECTS_DIR = Path.home() / ".claude" / "projects"
DEFAULT_LOG_PATH = Path.home() / ".claude-memory" / "memory.log"
CONFIG_PATH = Path.home() / ".claude-memory" / "config.json"

# Codex Desktop integration paths (consolidated here per single-home rule).
DEFAULT_CODEX_SESSIONS_DIR = Path.home() / ".codex" / "sessions"
CODEX_IMPORT_SENTINEL = Path.home() / ".claude-memory" / ".last-codex-import"
# Sentinel project path for Codex sessions where session_meta.cwd is missing.
# Without this, missing-cwd sessions create a project per Codex date directory
# (e.g. ".../2026/05/03" → project named "03").
CODEX_UNKNOWN_PROJECT_PATH = "/(unknown-codex)"

# Bulk-import coordination
IMPORT_LOCK_PATH = Path.home() / ".claude-memory" / "import.lock"
BACKUP_RETENTION = 10

# Default settings
DEFAULT_SETTINGS = {
    "db_path": str(DEFAULT_DB_PATH),
    "auto_inject_context": True,
    "max_context_sessions": 2,
    "exclude_projects": [],
    "logging_enabled": False,
    "sync_on_stop": True,
    "consolidation_reminder_enabled": True,
    "consolidation_min_hours": 24,
    "consolidation_min_sessions": 5,
}

# Keys in config.json that override DEFAULT_SETTINGS
_CONFIG_KEYS = {
    "auto_inject_context",
    "consolidation_reminder_enabled",
    "consolidation_min_hours",
    "consolidation_min_sessions",
    "max_context_sessions",
}

# Database schema — v3: messages stored once, branches as separate index
# Split into core (tables/indexes) and FTS variants for compatibility
SCHEMA_CORE = """
-- Projects table (derived from directory structure)
CREATE TABLE IF NOT EXISTS projects (
  id INTEGER PRIMARY KEY,
  path TEXT UNIQUE NOT NULL,
  key TEXT UNIQUE NOT NULL,
  name TEXT,
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_projects_key ON projects(key);

-- Sessions table (ONE row per session UUID)
CREATE TABLE IF NOT EXISTS sessions (
  id INTEGER PRIMARY KEY,
  uuid TEXT UNIQUE NOT NULL,
  project_id INTEGER REFERENCES projects(id),
  parent_session_id INTEGER REFERENCES sessions(id),
  git_branch TEXT,
  cwd TEXT,
  imported_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_sessions_project ON sessions(project_id);

-- Branches table (one row per branch per session)
CREATE TABLE IF NOT EXISTS branches (
  id INTEGER PRIMARY KEY,
  session_id INTEGER NOT NULL REFERENCES sessions(id),
  leaf_uuid TEXT NOT NULL,
  fork_point_uuid TEXT,
  is_active INTEGER DEFAULT 1,
  started_at DATETIME,
  ended_at DATETIME,
  exchange_count INTEGER DEFAULT 0,
  files_modified TEXT,
  commits TEXT,
  tool_counts TEXT,
  aggregated_content TEXT,
  context_summary TEXT,
  context_summary_json TEXT,
  summary_version INTEGER DEFAULT 0,
  UNIQUE(session_id, leaf_uuid)
);
CREATE INDEX IF NOT EXISTS idx_branches_session ON branches(session_id);
CREATE INDEX IF NOT EXISTS idx_branches_active ON branches(is_active);
CREATE INDEX IF NOT EXISTS idx_branches_summary_version ON branches(summary_version);

-- Messages table (ALL messages stored ONCE per session)
CREATE TABLE IF NOT EXISTS messages (
  id INTEGER PRIMARY KEY,
  session_id INTEGER NOT NULL REFERENCES sessions(id),
  uuid TEXT,
  parent_uuid TEXT,
  timestamp DATETIME,
  role TEXT CHECK(role IN ('user', 'assistant')),
  content TEXT NOT NULL,
  tool_summary TEXT,
  has_tool_use INTEGER DEFAULT 0,
  has_thinking INTEGER DEFAULT 0,
  is_notification INTEGER DEFAULT 0,
  origin TEXT,
  UNIQUE(session_id, uuid)
);
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id);
CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON messages(timestamp);
CREATE INDEX IF NOT EXISTS idx_messages_session_uuid ON messages(session_id, uuid);

-- Branch-messages mapping (many-to-many)
CREATE TABLE IF NOT EXISTS branch_messages (
  branch_id INTEGER NOT NULL REFERENCES branches(id),
  message_id INTEGER NOT NULL REFERENCES messages(id),
  PRIMARY KEY (branch_id, message_id)
);
CREATE INDEX IF NOT EXISTS idx_branch_messages_message ON branch_messages(message_id);

-- Import tracking
CREATE TABLE IF NOT EXISTS import_log (
  id INTEGER PRIMARY KEY,
  file_path TEXT UNIQUE NOT NULL,
  file_hash TEXT,
  imported_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  messages_imported INTEGER DEFAULT 0
);

"""

# FTS5 schema (best: porter stemming + unicode61, BM25 ranking)
SCHEMA_FTS5 = """
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
  content,
  content=messages,
  content_rowid=id,
  tokenize='porter unicode61'
);

CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
  INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
  INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
  INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
  INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;

CREATE VIRTUAL TABLE IF NOT EXISTS branches_fts USING fts5(
  aggregated_content,
  content=branches,
  content_rowid=id,
  tokenize='porter unicode61'
);

CREATE TRIGGER IF NOT EXISTS branches_ai AFTER INSERT ON branches BEGIN
  INSERT INTO branches_fts(rowid, aggregated_content) VALUES (new.id, new.aggregated_content);
END;
CREATE TRIGGER IF NOT EXISTS branches_ad AFTER DELETE ON branches BEGIN
  INSERT INTO branches_fts(branches_fts, rowid, aggregated_content) VALUES('delete', old.id, old.aggregated_content);
END;
CREATE TRIGGER IF NOT EXISTS branches_au AFTER UPDATE ON branches BEGIN
  INSERT INTO branches_fts(branches_fts, rowid, aggregated_content) VALUES('delete', old.id, old.aggregated_content);
  INSERT INTO branches_fts(rowid, aggregated_content) VALUES (new.id, new.aggregated_content);
END;
"""

# FTS4 schema (fallback: porter stemming, no BM25 but supports MATCH + snippet)
SCHEMA_FTS4 = """
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts4(
  content,
  content=messages,
  tokenize=porter
);

CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
  INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
  INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
  INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
  INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;

CREATE VIRTUAL TABLE IF NOT EXISTS branches_fts USING fts4(
  aggregated_content,
  content=branches,
  tokenize=porter
);

CREATE TRIGGER IF NOT EXISTS branches_ai AFTER INSERT ON branches BEGIN
  INSERT INTO branches_fts(rowid, aggregated_content) VALUES (new.id, new.aggregated_content);
END;
CREATE TRIGGER IF NOT EXISTS branches_ad AFTER DELETE ON branches BEGIN
  INSERT INTO branches_fts(branches_fts, rowid, aggregated_content) VALUES('delete', old.id, old.aggregated_content);
END;
CREATE TRIGGER IF NOT EXISTS branches_au AFTER UPDATE ON branches BEGIN
  INSERT INTO branches_fts(branches_fts, rowid, aggregated_content) VALUES('delete', old.id, old.aggregated_content);
  INSERT INTO branches_fts(rowid, aggregated_content) VALUES (new.id, new.aggregated_content);
END;
"""

# Combined schema (core + FTS5) for test fixtures and simple single-shot setup
SCHEMA = SCHEMA_CORE + SCHEMA_FTS5


def detect_fts_support(conn: sqlite3.Connection) -> str | None:
    """Detect the best available FTS extension."""
    try:
        opts = {row[0] for row in conn.execute("PRAGMA compile_options").fetchall()}
    except Exception:
        return None
    if "ENABLE_FTS5" in opts:
        return "fts5"
    if "ENABLE_FTS4" in opts or "ENABLE_FTS3" in opts:
        return "fts4"
    return None


def migrate_db(conn: sqlite3.Connection) -> bool:
    """
    Migrate database to v3 schema (messages-once + branch index).
    Detects old schema by checking if 'branches' table exists.
    If not, deletes the DB file so a fresh import is triggered.
    Returns True if migration was performed (DB was deleted and recreated).
    """
    cursor = conn.cursor()

    # Check if branches table exists (v3 indicator)
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='branches'")
    if cursor.fetchone():
        return False  # Already on v3

    # Check if sessions table exists at all (could be a fresh DB)
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='sessions'")
    if not cursor.fetchone():
        return False  # Fresh DB, no migration needed

    # Old schema detected — backup then nuke and recreate
    db_path = None
    # Get the database file path from connection
    cursor.execute("PRAGMA database_list")
    for row in cursor.fetchall():
        if row[1] == "main" and row[2]:
            db_path = Path(row[2])
            break

    if db_path and db_path.exists():
        # JSONL source files expire after 30 days — data older than that
        # exists only in this DB. Back up before destroying.
        backed_up = _backup_db_before_migration(db_path, "pre-v3-nuke")
        if not backed_up:
            # Backup failed (disk full, permissions, etc.) — refuse to destroy
            # the only copy. Return False so caller uses the old schema as-is.
            return False
        conn.close()
        db_path.unlink()
        # Clean up WAL/SHM files — orphaned WAL replayed into an empty DB
        # causes "database disk image is malformed" errors.
        for suffix in ("-wal", "-shm"):
            wal_path = db_path.with_name(db_path.name + suffix)
            if wal_path.exists():
                wal_path.unlink()

    # Reconnect and create fresh schema
    new_conn = sqlite3.connect(str(db_path) if db_path else ":memory:")
    new_conn.execute("PRAGMA journal_mode = WAL")
    new_conn.execute("PRAGMA busy_timeout = 5000")
    fts = detect_fts_support(new_conn)
    new_conn.executescript(SCHEMA_CORE)
    if fts == "fts5":
        new_conn.executescript(SCHEMA_FTS5)
    elif fts == "fts4":
        new_conn.executescript(SCHEMA_FTS4)
    new_conn.commit()

    # We can't return the new connection through the old reference,
    # so we signal that migration happened and caller should reconnect
    new_conn.close()
    return True


CURRENT_ONBOARDING_VERSION = 1


def load_config() -> dict:
    """Read ~/.claude-memory/config.json. Returns empty dict on missing/error."""
    try:
        if CONFIG_PATH.exists():
            result = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
            return result if isinstance(result, dict) else {}
    except Exception:
        pass
    return {}


def load_settings() -> dict:
    """Return settings with config.json overrides merged on top of defaults."""
    settings = DEFAULT_SETTINGS.copy()
    config = load_config()
    for key in _CONFIG_KEYS:
        if key in config:
            settings[key] = config[key]
    return settings


def get_db_path(settings: Optional[dict] = None) -> Path:
    """Get database path from settings or default."""
    if settings and "db_path" in settings:
        return Path(settings["db_path"]).expanduser()
    return DEFAULT_DB_PATH


def _reaggregate_notification_branches(cursor: sqlite3.Cursor) -> None:
    """Re-aggregate branches that contain notification messages.

    Updates aggregated_content and exchange_count to exclude notifications.
    Called after backfilling is_notification on existing messages.
    """
    cursor.execute("""
        SELECT DISTINCT bm.branch_id
        FROM branch_messages bm
        JOIN messages m ON bm.message_id = m.id
        WHERE m.is_notification = 1
    """)
    affected_branches = [row[0] for row in cursor.fetchall()]
    for bid in affected_branches:
        cursor.execute("""
            SELECT m.content FROM branch_messages bm
            JOIN messages m ON bm.message_id = m.id
            WHERE bm.branch_id = ? AND COALESCE(m.is_notification, 0) = 0
            ORDER BY m.timestamp ASC
        """, (bid,))
        agg = "\n".join(row[0] for row in cursor.fetchall())
        cursor.execute("UPDATE branches SET aggregated_content = ? WHERE id = ?", (agg, bid))
        cursor.execute("""
            SELECT COUNT(*) FROM branch_messages bm
            JOIN messages m ON bm.message_id = m.id
            WHERE bm.branch_id = ? AND m.role = 'user' AND COALESCE(m.is_notification, 0) = 0
        """, (bid,))
        human_user_count = cursor.fetchone()[0]
        cursor.execute("UPDATE branches SET exchange_count = ? WHERE id = ?",
                       (human_user_count, bid))


def _migrate_columns(conn: sqlite3.Connection) -> None:
    """Add missing columns (DDL, idempotent) and run versioned data migrations (DML)."""
    cursor = conn.cursor()
    cursor.execute("PRAGMA table_info(messages)")
    existing = {row[1] for row in cursor.fetchall()}

    # --- DDL migrations (column-existence gated, idempotent) ---
    if "tool_summary" not in existing:
        cursor.execute("ALTER TABLE messages ADD COLUMN tool_summary TEXT")
        conn.commit()
    if "is_notification" not in existing:
        cursor.execute("ALTER TABLE messages ADD COLUMN is_notification INTEGER DEFAULT 0")
        conn.commit()
    if "origin" not in existing:
        cursor.execute("ALTER TABLE messages ADD COLUMN origin TEXT")
        conn.commit()

    # branches DDL migration
    cursor.execute("PRAGMA table_info(branches)")
    branch_cols = {row[1] for row in cursor.fetchall()}
    if "tool_counts" not in branch_cols:
        cursor.execute("ALTER TABLE branches ADD COLUMN tool_counts TEXT")
        conn.commit()
    if "context_summary" not in branch_cols:
        cursor.execute("ALTER TABLE branches ADD COLUMN context_summary TEXT")
    if "context_summary_json" not in branch_cols:
        cursor.execute("ALTER TABLE branches ADD COLUMN context_summary_json TEXT")
    if "summary_version" not in branch_cols:
        cursor.execute("ALTER TABLE branches ADD COLUMN summary_version INTEGER DEFAULT 0")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_branches_summary_version ON branches(summary_version)")
    conn.commit()

    # token_snapshots table (new table, not a column add)
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='token_snapshots'")
    if not cursor.fetchone():
        cursor.executescript("""
CREATE TABLE IF NOT EXISTS token_snapshots (
  id INTEGER PRIMARY KEY,
  session_uuid TEXT UNIQUE NOT NULL,
  project_path TEXT,
  start_time DATETIME,
  duration_minutes INTEGER,
  user_message_count INTEGER,
  assistant_message_count INTEGER,
  input_tokens INTEGER DEFAULT 0,
  output_tokens INTEGER DEFAULT 0,
  cache_read_tokens INTEGER DEFAULT 0,
  cache_creation_tokens INTEGER DEFAULT 0,
  tool_counts TEXT,
  tool_errors INTEGER DEFAULT 0,
  uses_task_agent INTEGER DEFAULT 0,
  uses_web_search INTEGER DEFAULT 0,
  uses_web_fetch INTEGER DEFAULT 0,
  user_response_times TEXT,
  lines_added INTEGER DEFAULT 0,
  lines_removed INTEGER DEFAULT 0,
  goal_categories TEXT,
  outcome TEXT,
  session_type TEXT,
  friction_counts TEXT,
  brief_summary TEXT,
  imported_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_token_snapshots_session ON token_snapshots(session_uuid);
CREATE INDEX IF NOT EXISTS idx_token_snapshots_start ON token_snapshots(start_time);
""")
        conn.commit()

    # Ensure data_source column exists (added by get-token-insights ingest script)
    try:
        conn.execute("ALTER TABLE token_snapshots ADD COLUMN data_source TEXT")
        conn.commit()
    except sqlite3.OperationalError:
        pass

    # --- DML migrations (version-gated via PRAGMA user_version, run once) ---
    version = conn.execute("PRAGMA user_version").fetchone()[0]

    # Resolve db_path for backup operations (PRAGMA database_list returns (seq, name, file))
    db_path = Path(conn.execute("PRAGMA database_list").fetchone()[2])

    if version < 1:
        # v0.5.0: Backfill task-notification messages
        cursor.execute("""
            UPDATE messages SET is_notification = 1
            WHERE role = 'user' AND content LIKE '<task-notification>%' AND is_notification = 0
        """)
        _reaggregate_notification_branches(cursor)
        conn.execute("PRAGMA user_version = 1")
        conn.commit()

    if version < 2:
        # v0.7.1: Backfill teammate messages as notifications
        cursor.execute("""
            UPDATE messages SET is_notification = 1
            WHERE role = 'user' AND content LIKE '<teammate-message%' AND is_notification = 0
        """)
        _reaggregate_notification_branches(cursor)
        conn.execute("PRAGMA user_version = 2")
        conn.commit()

    if version < 3:
        # v0.8.0: Selective backfill of origin column from JSONL files.
        # Phase 1: UPDATE existing messages with origin data.
        # Phase 2: Nullify file_hash for sessions with channel messages
        #          (isMeta+origin entries previously filtered) so the next
        #          normal import re-processes just those sessions.
        _backfill_origin(conn, cursor)
        conn.execute("PRAGMA user_version = 3")
        conn.commit()

    if version < 4:
        # v0.8.70: Clear stale task-notification values from origin column.
        # parse_origin had a kind-fallback bug that leaked task-notification
        # into origin (reserved for channel sources: telegram, discord, slack).
        _backup_db_before_migration(db_path, "v4")
        cursor.execute(
            "UPDATE messages SET origin = NULL WHERE origin = 'task-notification'"
        )
        conn.execute("PRAGMA user_version = 4")
        conn.commit()


def _backup_db_before_migration(db_path: Path, label: str) -> bool:
    """Create a timestamped WAL-safe backup using sqlite3.Connection.backup().

    Returns True if backup succeeded and was verified, False otherwise.
    sqlite3.Connection.backup() does a page-level copy that respects WAL journaling.
    """
    if not db_path.name or not db_path.exists():
        return False
    ts = time.strftime("%Y%m%d-%H%M%S")
    backup_path = db_path.with_suffix(f".pre-{label}-{ts}.db")
    src = None
    dst = None
    try:
        src = sqlite3.connect(str(db_path))
        dst = sqlite3.connect(str(backup_path))
        src.backup(dst)
        # Verify backup is non-empty
        if not backup_path.exists() or backup_path.stat().st_size == 0:
            return False
        return True
    except Exception:
        return False
    finally:
        if dst:
            dst.close()
        if src:
            src.close()


def _backfill_origin(conn: sqlite3.Connection, cursor: sqlite3.Cursor) -> None:
    """Selectively backfill origin column from JSONL files without full reimport.

    For each file in import_log, scan for entries with origin fields
    and UPDATE existing message rows by session_id + uuid.

    Note: Previously had a Phase 2 that nullified file_hash to force reimport
    of sessions with channel messages. Removed because the reimport path is
    destructive (delete-all-then-insert) and JSONL files expire after 30 days —
    triggering a reimport risks irrecoverable data loss.
    """
    import json
    from memory_lib.content import parse_origin

    # Guard: sessions table may not exist in minimal test DBs
    tables = {r[0] for r in cursor.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    if "sessions" not in tables or "import_log" not in tables:
        return

    # Build file_path -> session mapping from import_log + sessions
    # The file stem (minus .jsonl and optional agent- prefix) is the session uuid
    file_session_map = {}
    all_import_rows = cursor.execute("SELECT file_path FROM import_log").fetchall()
    for (file_path,) in all_import_rows:
        p = Path(file_path)
        stem = p.stem
        if stem.startswith("agent-"):
            stem = stem[6:]
        row = cursor.execute(
            "SELECT id FROM sessions WHERE uuid = ?", (stem,)
        ).fetchone()
        if row:
            file_session_map[file_path] = row[0]

    for file_path, session_id in file_session_map.items():
        p = Path(file_path)
        if not p.exists():
            continue

        try:
            with open(p, "r", encoding="utf-8", errors="replace") as f:
                for line in f:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        obj = json.loads(line)
                    except json.JSONDecodeError:
                        continue

                    origin = obj.get("origin")
                    if not origin:
                        continue

                    # UPDATE origin for existing messages
                    uuid = obj.get("uuid")
                    if uuid and obj.get("type") in ("user", "assistant"):
                        origin_value = parse_origin(obj)
                        if origin_value:
                            cursor.execute(
                                "UPDATE messages SET origin = ? WHERE session_id = ? AND uuid = ?",
                                (origin_value, session_id, uuid)
                            )
        except OSError:
            continue

    conn.commit()


def _migrate_project_paths(conn: sqlite3.Connection) -> None:
    """Fix project paths that were incorrectly derived from hyphenated directory keys.

    When projects were first imported, path/name were derived from the Claude project
    directory key (e.g. '-Users-foo-repos-meta-ads-cli') using a lossy replace('-', '/')
    heuristic. For directories with hyphens in their name (e.g. 'meta-ads-cli'), this
    produces wrong paths ('/Users/foo/repos/meta/ads/cli' instead of the real path).

    The sessions.cwd column stores the REAL filesystem path recorded at runtime. This
    migration uses the most-common cwd across each project's sessions to correct the
    project path and name. It also merges duplicate projects that resolve to the same
    real path.

    This is idempotent: after fixing, the project path matches session cwd, so subsequent
    runs find no mismatch and do nothing.
    """
    cursor = conn.cursor()

    # Find all projects that have at least one session with a non-null cwd
    cursor.execute("""
        SELECT p.id, p.path, p.name,
               s.cwd,
               COUNT(*) AS cwd_count
        FROM projects p
        JOIN sessions s ON s.project_id = p.id
        WHERE s.cwd IS NOT NULL AND s.cwd != ''
        GROUP BY p.id, s.cwd
        ORDER BY p.id, cwd_count DESC
    """)
    rows = cursor.fetchall()
    if not rows:
        return

    # For each project, pick the most common cwd as the authoritative real path
    best_cwd: dict[int, str] = {}
    for proj_id, _path, _name, cwd, _count in rows:
        if proj_id not in best_cwd:
            best_cwd[proj_id] = cwd  # rows ordered by cwd_count DESC, first wins

    # Now check which projects need updating
    cursor.execute("SELECT id, path, name FROM projects")
    projects = cursor.fetchall()

    for proj_id, stored_path, _stored_name in projects:
        real_cwd = best_cwd.get(proj_id)
        if not real_cwd or real_cwd == stored_path:
            continue  # No cwd data or already correct

        real_name = Path(real_cwd).name

        # Check if another project already has real_cwd as its path (merge conflict)
        cursor.execute("SELECT id FROM projects WHERE path = ? AND id != ?", (real_cwd, proj_id))
        existing = cursor.fetchone()

        if existing:
            # Merge: reassign all sessions from this (wrong) project to the existing one
            keeper_id = existing[0]
            cursor.execute(
                "UPDATE sessions SET project_id = ? WHERE project_id = ?",
                (keeper_id, proj_id)
            )
            cursor.execute("DELETE FROM projects WHERE id = ?", (proj_id,))
        else:
            # Simple fix: update path and name
            cursor.execute(
                "UPDATE projects SET path = ?, name = ? WHERE id = ?",
                (real_cwd, real_name, proj_id)
            )

    conn.commit()


def get_db_connection(settings: Optional[dict] = None) -> sqlite3.Connection:
    """
    Get database connection, initializing schema and running migrations if needed.
    Uses settings-based path if provided.
    Sets WAL mode and busy_timeout for concurrent access safety.
    """
    db_path = get_db_path(settings)
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(db_path)

    # WAL mode: readers never block writers, writers never block readers
    conn.execute("PRAGMA journal_mode = WAL")
    # busy_timeout: wait up to 5s on writer-writer collisions instead of failing
    conn.execute("PRAGMA busy_timeout = 5000")
    # Enforce foreign key constraints to prevent orphaned data
    conn.execute("PRAGMA foreign_keys = ON")

    # Check if migration needed (old schema -> v3)
    migrated = migrate_db(conn)
    if migrated:
        # Connection was closed during migration, reconnect
        conn = sqlite3.connect(db_path)
        conn.execute("PRAGMA journal_mode = WAL")
        conn.execute("PRAGMA busy_timeout = 5000")
        conn.execute("PRAGMA foreign_keys = ON")

    if not migrated:
        # Apply schema (handles fresh databases, idempotent)
        fts = detect_fts_support(conn)
        conn.executescript(SCHEMA_CORE)
        if fts == "fts5":
            conn.executescript(SCHEMA_FTS5)
        elif fts == "fts4":
            conn.executescript(SCHEMA_FTS4)
        conn.commit()

    # Add any missing columns (e.g. tool_summary)
    _migrate_columns(conn)

    # Fix project paths that were incorrectly derived from hyphenated directory keys
    try:
        _migrate_project_paths(conn)
    except Exception:
        pass  # Never block DB connection on data migration errors

    return conn



def setup_logging(settings: Optional[dict] = None) -> logging.Logger:
    """
    Set up logging with rotation.
    Returns a null logger if logging is disabled.
    """
    logger = logging.getLogger("claude-memory")
    logger.handlers.clear()

    if not settings or not settings.get("logging_enabled", False):
        logger.addHandler(logging.NullHandler())
        return logger

    log_path = DEFAULT_LOG_PATH
    log_path.parent.mkdir(parents=True, exist_ok=True)

    handler = RotatingFileHandler(
        log_path,
        maxBytes=1_000_000,  # 1MB
        backupCount=2
    )
    formatter = logging.Formatter(
        "%(asctime)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

    return logger

#!/usr/bin/env python3
"""
Session formatting, time utilities, and project path helpers.
"""

from __future__ import annotations

import json
from datetime import datetime
from pathlib import Path
from typing import Optional


def format_time(ts_str: Optional[str], fmt: str = "%H:%M") -> str:
    """
    Format ISO timestamp to specified format.
    Default: HH:MM
    """
    if not ts_str:
        return "??:??"
    try:
        dt = datetime.fromisoformat(ts_str.replace('Z', '+00:00'))
        return dt.astimezone().strftime(fmt)
    except Exception:
        return ts_str[:16] if ts_str else "??:??"


def format_time_full(ts_str: Optional[str]) -> str:
    """Format ISO timestamp to YYYY-MM-DD HH:MM."""
    return format_time(ts_str, "%Y-%m-%d %H:%M")


_WORKTREE_MARKER = "/.claude/worktrees/"
_WORKTREE_KEY_MARKER = "--claude-worktrees-"


def normalize_cwd(cwd: str) -> str:
    """Strip .claude/worktrees/<name> suffix from a raw path, returning the base repo path.

    Normalizes backslashes to forward slashes first so Windows paths
    (C:\\Users\\...) match the forward-slash worktree marker.
    """
    cwd = cwd.replace("\\", "/")
    idx = cwd.rfind(_WORKTREE_MARKER)
    return cwd[:idx] if idx != -1 else cwd


def get_project_key(cwd: str) -> str:
    """Convert working directory to project key format.

    Resolves .claude/worktrees/<name> paths to the base repo path
    so worktree sessions share project context with the main repo.
    """
    return normalize_cwd(cwd).replace("/", "-").replace(":", "-").replace(".", "-")


def normalize_project_key(key: str) -> str:
    """Strip worktree suffix from an already-encoded project key.

    Encoded worktree keys contain '--claude-worktrees-' (from /.claude/worktrees/).
    """
    idx = key.rfind(_WORKTREE_KEY_MARKER)
    return key[:idx] if idx != -1 else key


def parse_project_key(key: str) -> str:
    """Convert directory key back to original path (lossy — hyphens in dir names are lost).
    Prefer using session cwd metadata when available.

    Detects Windows-style keys (starting with a drive letter like 'C-') and
    reconstructs with the correct prefix. Unix keys start with '-' (from '/').
    """
    # Detect Windows drive letter: key starts with "<letter>--" (colon+slash both → hyphen)
    if len(key) >= 3 and key[0].isalpha() and key[1:3] == "--":
        parts = key[3:].replace("-", "/")
        return key[0].upper() + ":/" + parts.lstrip("/")
    parts = key.lstrip("-").replace("-", "/")
    return "/" + parts.lstrip("/")


def extract_project_name(path: str) -> str:
    """Extract short project name from path."""
    return Path(path).name


def format_markdown_session(session: dict, verbose: bool = False) -> str:
    """Format a single session as markdown."""
    lines = []

    started = format_time_full(session.get("started_at"))
    project = session.get("project", "Unknown")
    lines.append(f"## {project} | {started}")
    lines.append(f"Session: {session.get('uuid', 'unknown')[:8]}")

    if session.get("git_branch"):
        lines.append(f"Branch: {session['git_branch']}")

    if verbose:
        files = session.get("files_modified", [])
        if files:
            lines.append("\n### Files Modified")
            for f in files[-10:]:
                lines.append(f"- `{f}`")
            if len(files) > 10:
                lines.append(f"- ...and {len(files) - 10} more")

        commits = session.get("commits", [])
        if commits:
            lines.append("\n### Commits")
            for c in commits:
                lines.append(f"- {c}")

        tool_counts = session.get("tool_counts", {})
        if tool_counts:
            sorted_tools = sorted(tool_counts.items(), key=lambda x: x[1], reverse=True)
            tools_str = ", ".join(f"{name}: {count}" for name, count in sorted_tools)
            lines.append("\n### Tools Used")
            lines.append(tools_str)

    if session.get("summary") is not None:
        lines.append("\n### Summary\n")
        summ = (session.get("summary") or "").strip()
        lines.append(summ if summ else "_(summary unavailable — re-run this session without --summary)_")
        lines.append("\n---\n")
        return "\n".join(lines)

    lines.append("\n### Conversation\n")

    for msg in session.get("messages", []):
        if msg.get("is_notification"):
            role = "Subagent Result"
        else:
            role = "User" if msg["role"] == "user" else "Assistant"
        lines.append(f"**{role}:** {msg['content']}\n")

    lines.append("---\n")
    return "\n".join(lines)


def format_json_sessions(sessions: list[dict], extra: Optional[dict] = None) -> str:
    """Format sessions as JSON with metadata."""
    output = {
        "sessions": sessions,
        "total_sessions": len(sessions),
    }
    # Summary-mode sessions carry a "summary" string instead of a "messages" list, so
    # total_messages would always be 0 and mislead JSON consumers. Report the matching
    # counter instead.
    if any("summary" in s for s in sessions):
        output["total_summaries"] = sum(1 for s in sessions if "summary" in s)
    else:
        output["total_messages"] = sum(len(s.get("messages", [])) for s in sessions)
    if extra:
        output.update(extra)
    return json.dumps(output, indent=2)

#!/usr/bin/env python3
"""
JSONL parsing, branch detection, and metadata extraction.
"""

from __future__ import annotations

import json
import sqlite3
from pathlib import Path
from typing import Generator

from memory_lib.content import (
    extract_commits,
    extract_files_modified,
    is_task_notification,
    is_teammate_message,
    is_tool_result,
)


def parse_jsonl_file(filepath: Path) -> Generator[dict, None, None]:
    """Parse JSONL file, yielding user/assistant entries for import."""
    with open(filepath, "r", encoding="utf-8", errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
                if obj.get("isMeta") and not obj.get("origin"):
                    continue
                if obj.get("type") in ("user", "assistant"):
                    yield obj
            except json.JSONDecodeError:
                pass


def parse_all_with_uuids(filepath: Path) -> Generator[dict, None, None]:
    """
    Parse JSONL file yielding ALL entries with UUIDs.
    Used for building the parentUuid chain to find branches.
    """
    with open(filepath, "r", encoding="utf-8", errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
                if obj.get("uuid"):
                    yield obj
            except json.JSONDecodeError:
                pass


def extract_session_metadata(entries: list[dict]) -> dict:
    """Extract session metadata from entries."""
    metadata = {
        "started_at": None,
        "ended_at": None,
        "git_branch": None,
        "cwd": None,
    }

    for entry in entries:
        ts = entry.get("timestamp")
        if ts:
            if metadata["started_at"] is None or ts < metadata["started_at"]:
                metadata["started_at"] = ts
            if metadata["ended_at"] is None or ts > metadata["ended_at"]:
                metadata["ended_at"] = ts

        if not metadata["git_branch"]:
            metadata["git_branch"] = entry.get("gitBranch")
        if not metadata["cwd"]:
            metadata["cwd"] = entry.get("cwd")

    return metadata


def find_all_branches(all_entries: list[dict]) -> list[dict]:
    """
    Find all conversation branches (from rewinds).

    Returns list of branches, each with:
      - leaf_uuid: UUID of the last message in this branch
      - uuids: set of all UUIDs on this branch path
      - is_active: True if this is the current active branch
      - fork_point_uuid: UUID where this branch diverged (None for active)

    Algorithm:
    1. Find active branch (trace from latest message back to root)
    2. Find fork points on the active path where non-active children
       lead to subtrees with user messages (actual rewinds, not tree noise)
    3. For each rewind fork, collect the abandoned subtree + common prefix
    """
    uuid_to_entry: dict[str, dict] = {}
    uuid_to_parent: dict[str, str | None] = {}
    children: dict[str, list[str]] = {}

    for entry in all_entries:
        uuid = entry.get("uuid")
        if not uuid:
            continue
        uuid_to_entry[uuid] = entry
        parent = entry.get("parentUuid")
        uuid_to_parent[uuid] = parent
        if parent:
            children.setdefault(parent, []).append(uuid)

    if not uuid_to_entry:
        return []

    # Step 1: Find active branch (latest -> root)
    latest = max(uuid_to_entry.values(), key=lambda e: e.get("timestamp") or "")
    active_uuids: set[str] = set()
    current: str | None = latest["uuid"]
    while current:
        active_uuids.add(current)
        current = uuid_to_parent.get(current)

    branches: list[dict] = [
        {"leaf_uuid": latest["uuid"], "uuids": active_uuids, "is_active": True, "fork_point_uuid": None}
    ]

    # Step 2: Find rewind forks on the active path
    def has_user_descendant(uuid: str, depth: int = 0) -> bool:
        if depth > 100:
            return False
        entry = uuid_to_entry.get(uuid)
        if entry and entry.get("type") == "user":
            return True
        for kid in children.get(uuid, []):
            if has_user_descendant(kid, depth + 1):
                return True
        return False

    def collect_subtree(uuid: str) -> set[str]:
        result: set[str] = set()
        stack = [uuid]
        while stack:
            node = stack.pop()
            result.add(node)
            stack.extend(children.get(node, []))
        return result

    for uuid in active_uuids:
        kids = children.get(uuid, [])
        if len(kids) <= 1:
            continue

        for kid in kids:
            if kid in active_uuids:
                continue
            if not has_user_descendant(kid):
                continue

            # Real rewind fork — build the abandoned branch
            # Common prefix: fork point back to root
            prefix: set[str] = set()
            cur: str | None = uuid
            while cur:
                prefix.add(cur)
                cur = uuid_to_parent.get(cur)

            subtree = collect_subtree(kid)
            branch_uuids = prefix | subtree

            subtree_entries = [uuid_to_entry[u] for u in subtree if u in uuid_to_entry]
            if not subtree_entries:
                continue
            leaf = max(subtree_entries, key=lambda e: e.get("timestamp") or "")

            branches.append({
                "leaf_uuid": leaf["uuid"],
                "uuids": branch_uuids,
                "is_active": False,
                "fork_point_uuid": uuid,
            })

    return branches


def compute_branch_metadata(entries: list[dict]) -> tuple[int, list[str], list[str], dict[str, int]]:
    """
    Compute metadata for a branch's entries in one pass.
    Returns: (exchange_count, files_modified, commits, tool_counts)
    """
    exchange_count = 0
    all_files = []
    all_commits = []
    tool_counts: dict[str, int] = {}
    has_user = False

    for entry in entries:
        entry_type = entry.get("type")
        if entry_type not in ("user", "assistant"):
            continue

        message = entry.get("message", {})
        content = message.get("content", "")

        if entry_type == "user" and is_tool_result(content):
            continue

        if entry_type == "user" and (is_task_notification(content) or is_teammate_message(content)):
            continue

        if entry_type == "user":
            if has_user:
                exchange_count += 1
            has_user = True

        if entry_type == "assistant":
            all_files.extend(extract_files_modified(content))
            all_commits.extend(extract_commits(content))
            # Count tool usage from all assistant entries (including tool-only ones)
            if isinstance(content, list):
                for item in content:
                    if isinstance(item, dict) and item.get("type") == "tool_use":
                        tool_name = item.get("name", "")
                        if tool_name:
                            tool_counts[tool_name] = tool_counts.get(tool_name, 0) + 1

    if has_user:
        exchange_count += 1

    # Deduplicate files preserving order
    seen = {}
    unique_files = []
    for f in all_files:
        if f not in seen:
            seen[f] = True
            unique_files.append(f)

    return exchange_count, unique_files, all_commits, tool_counts


def aggregate_branch_content(cursor: sqlite3.Cursor, branch_db_id: int) -> str:
    """Concatenate all message content for a branch in timestamp order, excluding notifications."""
    cursor.execute("""
        SELECT m.content FROM branch_messages bm
        JOIN messages m ON bm.message_id = m.id
        WHERE bm.branch_id = ? AND COALESCE(m.is_notification, 0) = 0
        ORDER BY m.timestamp ASC
    """, (branch_db_id,))
    return "\n".join(row[0] for row in cursor.fetchall())

#!/usr/bin/env python3
"""
Precompute structured context summaries for session injection.

Runs at Stop time (sync_current.py) and import time. Produces both a JSON
source-of-truth and a pre-rendered markdown template stored on the branches table.
All extraction is deterministic Python — no LLM calls.
"""

from __future__ import annotations

import json
import re
import sqlite3

from memory_lib.formatting import format_time, format_time_full

# Truncation limits
_FRONT_CHARS = 300
_BACK_CHARS = 600

# Session disposition patterns
_COMPLETION_RE = re.compile(
    r'(?:done|pushed|merged|all (?:tests? )?pass|completed|finished|shipped|deployed|'
    r'PR #?\d+|commit(?:ted)?|changes? (?:are )?live)',
    re.IGNORECASE
)
_SHORT_CONFIRM_RE = re.compile(
    r'^(?:y(?:a|ep|es)?|thanks?|(?:looks? )?good|nice|perfect|great|ok|lgtm|k)\s*[.!]?$',
    re.IGNORECASE
)
_NEW_INSTRUCTION_RE = re.compile(
    r'^(?:now |next |also |can you |let\'?s |please |I (?:want|need) )',
    re.IGNORECASE
)


def truncate_mid(text: str, front: int = _FRONT_CHARS, back: int = _BACK_CHARS) -> str:
    """Mid-truncate text, keeping front and back portions."""
    if not text or len(text) <= front + back + 20:
        return text
    return text[:front] + "\n[... truncated ...]\n" + text[-back:]


def detect_disposition(exchanges: list[dict]) -> str:
    """Classify session ending as COMPLETED, IN_PROGRESS, or INTERRUPTED.

    Heuristics based on the final exchange pair:
    - COMPLETED: assistant uses completion language and user confirms briefly
    - IN_PROGRESS: user gives a new instruction as their last message
    - INTERRUPTED: default / session ends mid-flow
    """
    if not exchanges:
        return "INTERRUPTED"

    last = exchanges[-1]
    last_user = last.get("user", "").strip()
    last_asst = last.get("assistant", "").strip()

    # If user's last message is a new instruction, work is in progress
    if _NEW_INSTRUCTION_RE.search(last_user):
        return "IN_PROGRESS"

    # If assistant used completion language and user confirmed briefly
    if _COMPLETION_RE.search(last_asst) and _SHORT_CONFIRM_RE.match(last_user):
        return "COMPLETED"

    # If assistant used completion language (even without user confirm — session may have ended)
    if _COMPLETION_RE.search(last_asst) and len(last_user) < 30:
        return "COMPLETED"

    # If user confirmed briefly (likely accepting the work)
    if _SHORT_CONFIRM_RE.match(last_user):
        return "COMPLETED"

    return "IN_PROGRESS"


def build_exchange_pairs(messages: list[dict]) -> list[dict]:
    """
    Build exchange pairs from sequential messages.

    Each message is {"role": str, "content": str, "timestamp": str}.
    Returns list of {"user": str, "assistant": str, "timestamp": str, "index": int}.
    """
    exchanges = []
    current_user = None
    current_user_ts = None
    current_asst_parts = []  # type: list[str]

    for m in messages:
        if m["role"] == "user":
            if current_user is not None:
                exchanges.append({
                    "user": current_user,
                    "assistant": "\n\n".join(current_asst_parts),
                    "timestamp": current_user_ts,
                    "index": len(exchanges),
                })
            current_user = m["content"]
            current_user_ts = m.get("timestamp")
            current_asst_parts = []
        elif m["role"] == "assistant" and current_user is not None:
            cleaned = re.sub(r'\[Tool: \w+\]', '', m["content"]).strip()
            if cleaned:
                current_asst_parts.append(cleaned)

    if current_user is not None:
        exchanges.append({
            "user": current_user,
            "assistant": "\n\n".join(current_asst_parts),
            "timestamp": current_user_ts,
            "index": len(exchanges),
        })

    return exchanges


def build_context_summary_json(branch_row: dict, messages: list[dict]) -> dict:
    """
    Assemble the structured JSON summary from branch metadata and messages.

    branch_row keys: started_at, ended_at, exchange_count, files_modified,
                     commits, tool_counts, git_branch.
    messages: list of {"role", "content", "timestamp"} dicts, ordered by time.
    """
    exchanges = build_exchange_pairs(messages)
    if not exchanges:
        return {"version": 2, "topic": "", "first_exchanges": [],
                "last_exchanges": [], "metadata": {}}

    # Topic from first user message
    topic = exchanges[0]["user"]
    if len(topic) > 120:
        topic = topic[:120] + "..."

    disposition = detect_disposition(exchanges)

    # First exchanges (up to 2)
    first_exchanges = [
        {"user": ex["user"], "assistant": ex["assistant"], "timestamp": ex["timestamp"]}
        for ex in exchanges[:2]
    ]

    # Last exchanges (up to 6)
    if len(exchanges) <= 8:
        # Short/medium session: all exchanges go into last_exchanges
        last_exchanges = [
            {"user": ex["user"], "assistant": ex["assistant"], "timestamp": ex["timestamp"]}
            for ex in exchanges
        ]
    else:
        # Take last 6 exchanges
        last_exchanges = [
            {"user": ex["user"], "assistant": ex["assistant"], "timestamp": ex["timestamp"]}
            for ex in exchanges[-6:]
        ]

    # Parse JSON fields from branch_row
    files = branch_row.get("files_modified") or "[]"
    if isinstance(files, str):
        try:
            files = json.loads(files)
        except (json.JSONDecodeError, TypeError):
            files = []

    commits = branch_row.get("commits") or "[]"
    if isinstance(commits, str):
        try:
            commits = json.loads(commits)
        except (json.JSONDecodeError, TypeError):
            commits = []

    tool_counts = branch_row.get("tool_counts") or "{}"
    if isinstance(tool_counts, str):
        try:
            tool_counts = json.loads(tool_counts)
        except (json.JSONDecodeError, TypeError):
            tool_counts = {}

    return {
        "version": 2,
        "topic": topic,
        "disposition": disposition,
        "first_exchanges": first_exchanges,
        "last_exchanges": last_exchanges,
        "metadata": {
            "exchange_count": branch_row.get("exchange_count", len(exchanges)),
            "files_modified": files,
            "commits": commits,
            "tool_counts": tool_counts,
            "started_at": branch_row.get("started_at"),
            "ended_at": branch_row.get("ended_at"),
            "git_branch": branch_row.get("git_branch"),
        },
    }


def _build_gap_summary(summary_json: dict) -> str:
    """Build a one-line summary of what happened in the omitted middle exchanges."""
    files = summary_json.get("metadata", {}).get("files_modified", [])
    if files:
        short = [f.rsplit("/", 1)[-1] for f in files[:3]]
        return ", ".join(short)
    return ""


def render_context_summary(summary_json: dict) -> str:
    """
    Render the JSON summary to injection-ready markdown.

    Short sessions (<=8 exchanges) render all exchanges once, no first/last split.
    Longer sessions show first 2 exchanges + gap + last 6 exchanges.
    """
    if not summary_json or not summary_json.get("first_exchanges"):
        return ""

    meta = summary_json.get("metadata", {})
    lines = []

    # Header
    start = format_time_full(meta.get("started_at"))
    end = format_time_full(meta.get("ended_at"))
    header = f"### Session: {start} -> {end}"
    branch = meta.get("git_branch")
    if branch:
        header += f" (branch: {branch})"
    lines.append(header + "\n")

    # Topic and disposition
    topic = summary_json.get("topic", "")
    disposition = summary_json.get("disposition", "")
    if topic or disposition:
        parts = []
        if topic:
            parts.append(f"**Topic:** {topic}")
        if disposition:
            parts.append(f"**Status:** {disposition}")
        lines.append(" | ".join(parts))
        lines.append("")

    # Metadata: files, commits, tools
    files = meta.get("files_modified", [])
    if files:
        file_strs = [f"`{f}`" for f in files[:6]]
        line = "Modified: " + ", ".join(file_strs)
        if len(files) > 6:
            line += f" +{len(files) - 6} more"
        lines.append(line)

    commits = meta.get("commits", [])
    if commits:
        commit_strs = commits[:3]
        lines.append("Commits: " + "; ".join(commit_strs))

    tool_counts = meta.get("tool_counts", {})
    if tool_counts:
        sorted_tools = sorted(tool_counts.items(), key=lambda x: x[1], reverse=True)[:8]
        tools_str = ", ".join(f"{name}({count})" for name, count in sorted_tools)
        lines.append("Tools: " + tools_str)

    lines.append("")

    # Key Signals section (omitted if no markers)
    exchange_count = meta.get("exchange_count", 0)
    first_exs = summary_json.get("first_exchanges", [])
    last_exs = summary_json.get("last_exchanges", [])

    if exchange_count <= 8:
        # Short/medium session: render all exchanges once
        lines.append("### Conversation\n")
        for ex in last_exs:
            t = format_time(ex.get("timestamp"))
            lines.append(f"**[{t}] User:**")
            lines.append(ex["user"])
            lines.append("")
            if ex["assistant"]:
                lines.append(f"**[{t}] Assistant:**")
                lines.append(truncate_mid(ex["assistant"]))
                lines.append("")
    else:
        # Where We Left Off first — most recent context at top, where attention is
        # highest and where inline-preview truncation (if any) clips from below.
        lines.append("### Where We Left Off\n")
        for ex in last_exs:
            t = format_time(ex.get("timestamp"))
            lines.append(f"**[{t}] User:**")
            lines.append(ex["user"])
            lines.append("")
            if ex["assistant"]:
                lines.append(f"**[{t}] Assistant:**")
                lines.append(truncate_mid(ex["assistant"]))
                lines.append("")

        # Gap indicator with summary of middle exchanges
        gap = exchange_count - len(first_exs) - len(last_exs)
        if gap > 0:
            gap_detail = _build_gap_summary(summary_json)
            if gap_detail:
                lines.append(f"[... {gap} earlier exchanges covering: {gap_detail} ...]\n")
            else:
                lines.append(f"[... {gap} earlier exchanges ...]\n")

        # Earlier in This Session — first 2 exchanges kept for origin context,
        # placed last so they're the first thing clipped under truncation.
        lines.append("### Earlier in This Session\n")
        for ex in first_exs:
            t = format_time(ex.get("timestamp"))
            lines.append(f"**[{t}] User:**")
            lines.append(ex["user"])
            lines.append("")
            if ex["assistant"]:
                lines.append(f"**[{t}] Assistant:**")
                lines.append(truncate_mid(ex["assistant"]))
                lines.append("")

    # Contextual recall priming footer
    topic = summary_json.get("topic", "")
    files = meta.get("files_modified", [])
    footer_parts = [f"{exchange_count} exchanges"]
    if topic:
        short_topic = topic[:80] + "..." if len(topic) > 80 else topic
        footer_parts.append(f'about "{short_topic}"')
    if files:
        short_files = [f.rsplit("/", 1)[-1] for f in files[:3]]
        footer_parts.append(f"({', '.join(short_files)})")
    footer = " ".join(footer_parts)
    lines.append(
        f"[{footer} — proactively use /recall-conversations "
        "to retrieve relevant context from past conversations when the user references "
        "prior work, asks about decisions made earlier, or when you sense useful context "
        "from previous sessions would improve your response.]"
    )

    return "\n".join(lines)


def compute_context_summary(cursor: sqlite3.Cursor, branch_db_id: int) -> tuple[str, str]:
    """
    Orchestrator: fetch branch + messages from DB, return (markdown, json_string).

    Raises on DB errors; caller should wrap in try/except.
    """
    # Fetch branch row
    cursor.execute("""
        SELECT b.started_at, b.ended_at, b.exchange_count, b.files_modified,
               b.commits, b.tool_counts, s.git_branch
        FROM branches b
        JOIN sessions s ON b.session_id = s.id
        WHERE b.id = ?
    """, (branch_db_id,))
    row = cursor.fetchone()
    if not row:
        return "", ""

    branch_row = {
        "started_at": row[0],
        "ended_at": row[1],
        "exchange_count": row[2],
        "files_modified": row[3],
        "commits": row[4],
        "tool_counts": row[5],
        "git_branch": row[6],
    }

    # Fetch messages for this branch
    cursor.execute("""
        SELECT m.role, m.content, m.timestamp
        FROM branch_messages bm
        JOIN messages m ON bm.message_id = m.id
        WHERE bm.branch_id = ?
          AND COALESCE(m.is_notification, 0) = 0
        ORDER BY m.timestamp ASC
    """, (branch_db_id,))

    messages = [
        {"role": r, "content": c, "timestamp": t}
        for r, c, t in cursor.fetchall()
    ]

    if not messages:
        return "", ""

    summary_json = build_context_summary_json(branch_row, messages)
    summary_md = render_context_summary(summary_json)
    json_str = json.dumps(summary_json, ensure_ascii=False)

    return summary_md, json_str