task_server.py

#!/usr/bin/env python3
"""Thin MCP server exposing only task management tools.

Shares the same SQLite database as the main sqlite-kb server and keeps the task
surface independently deployable; ``unified_server.py`` also mounts it.
"""

from __future__ import annotations

import json
import sqlite3
import threading
import uuid
from typing import Any

from fastmcp_compat import FastMCP

from db_utils import (
    get_conn as _get_conn,
    get_conn_immediate as _get_conn_immediate,
    fts_query as _fts_query,
    TaskDAO,
    TASK_ACTIVE_EXCLUSIONS as _TASK_ACTIVE_EXCLUSIONS,
    TASK_PRIORITIES as _TASK_PRIORITIES,
    validate_task_fields as _validate_task_fields,
    build_priority_order_sql,
    now_iso as _now,
    setup_logger,
    apply_task_mutation as _apply_task_mutation,
    create_task_with_ledger as _create_task_with_ledger,
    DB_PATH as _DB_PATH,
)
from retrieval_contract import (
    RETRIEVAL_CONTRACT_VERSION,
    classify_lookup_confidence,
    is_visible_lookup_match,
    order_surface_hits,
    normalize_lookup_text,
    score_lookup_surface,
)
from premium_runtime import maybe_mount_premium_extensions
from task_search import TaskSearchEngine
from smart_retrieval import (
    READY_CONTEXT_CONTRACT_VERSION as _READY_CONTEXT_CONTRACT_VERSION,
    prime_context as _prime_context,
    ready_context as _ready_context,
    suggested_ready as _suggested_ready,
)

# Explicit injection seam for authoritative task mutations. Production writes
# reserve the SQLite writer slot before their read/modify/write sequence; tests
# that replace the connection factory must replace this seam as well so an
# isolated fixture can never fall through to the live database.
_get_write_conn = _get_conn_immediate

# Pre-built SQL for active-task exclusion
_EXCL_PH = ",".join("?" for _ in _TASK_ACTIVE_EXCLUSIONS)

# ── Logging (file-only, NEVER stdout — breaks MCP stdio) ────────────────

logger = setup_logger("sqlite-tasks", "task_server.log")

_search_engine: TaskSearchEngine | None = None


def _get_search_engine() -> TaskSearchEngine:
    global _search_engine
    if _search_engine is None:
        _search_engine = TaskSearchEngine()
    return _search_engine


class _TaskEmbeddingScheduler:
    """Coalesce optional embedding refreshes onto one daemon worker."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._pending: dict[str, None] = {}
        self._running = False

    def request(self, task_id: str) -> None:
        with self._lock:
            self._pending[task_id] = None
            if self._running:
                return
            self._running = True
        threading.Thread(
            target=self._run,
            name="sqlite-task-embedding",
            daemon=True,
        ).start()

    def _run(self) -> None:
        while True:
            with self._lock:
                if not self._pending:
                    self._running = False
                    return
                task_id = next(iter(self._pending))
                self._pending.pop(task_id, None)
            try:
                from vec_search import vec_sync_task_detached

                vec_sync_task_detached(_DB_PATH, task_id)
            except Exception as exc:
                logger.debug(
                    "detached vec_sync_task(%s) skipped: %s",
                    task_id,
                    exc,
                    exc_info=True,
                )


_task_embedding_scheduler = _TaskEmbeddingScheduler()


def _vec_sync_task_safe(task_id: str) -> None:
    """Schedule derived embedding work after the authoritative commit."""
    try:
        _task_embedding_scheduler.request(task_id)
    except Exception as exc:
        # The authoritative task is already committed.  Optional cache
        # scheduling must never turn that success into a retry/duplicate.
        logger.debug(
            "task embedding scheduling skipped for %s: %s",
            task_id,
            exc,
            exc_info=True,
        )


def _normalize_title_key(value: str | None) -> str:
    return " ".join((value or "").casefold().split())


def _find_note_by_title_project(
    conn, *, title: str, project: str | None
) -> dict[str, Any] | None:
    title_key = _normalize_title_key(title)
    if not title_key:
        return None
    project_predicate = "project IS NULL" if project is None else "project = ?"
    params = () if project is None else (project,)
    rows = conn.execute(
        f"SELECT id, title FROM tasks WHERE type = 'note' AND {project_predicate}",
        params,
    ).fetchall()
    for row in rows:
        if _normalize_title_key(row["title"]) == title_key:
            return dict(row)
    return None


# ── FastMCP app ──────────────────────────────────────────────────────────

mcp = FastMCP(
    "sqlite-tasks",
    instructions=(
        "Task management tools for SQLite-backed persistent memory. "
        "Create, update, query, and digest tasks. "
        "query_tasks returns live work by default: done/archived/cancelled rows are "
        "excluded unless you pass an explicit status or include_completed=True. "
        "Use find_by_title when only a remembered phrase is known: it searches tasks, notes, "
        "and entities across title/name, description, notes, observations, and project "
        "regardless of section, type, or project filters, using retrieval contract "
        f"{RETRIEVAL_CONTRACT_VERSION} with confidence gating; finished rows stay "
        "reachable but always rank below live work. "
        "Use upsert_note_by_title_project for idempotent research/decision notes "
        "when a repeated agent run must update an existing title/project instead of "
        "creating duplicates. "
        "Use description as the default primary body for task/note content; "
        "use notes only for auxiliary/internal metadata. Shares DB with sqlite-kb."
    ),
)


# ═══════════════════════════════════════════════════════════════════════════
# Tool 1: create_task_or_note
# ═══════════════════════════════════════════════════════════════════════════
@mcp.tool()
def create_task_or_note(
    title: str,
    type: str = "task",
    description: str = "",
    section: str = "inbox",
    priority: str = "medium",
    due_date: str = "",
    project: str = "",
    parent_id: str = "",
    notes: str = "",
    recurring: str = "",
    reminder_at: str = "",
) -> str:
    """Create a new task or note. Returns the UUID.

    Put the main long-form task/note text in ``description`` by default.
    Use ``notes`` only for auxiliary, internal, or machine-readable metadata.

    Args:
        title: Task title (required).
        type: task | note.
        description: Primary task/note body and main long-form content.
        section: inbox | today | next | someday | waiting.
        priority: low | medium | high | critical.
        due_date: YYYY-MM-DD format or empty to skip.
        project: Project tag for grouping.
        parent_id: UUID of parent task (for subtasks).
        notes: Secondary/internal notes or machine-readable metadata.
        recurring: JSON config for recurrence (e.g. '{"every":"week","day":"monday"}').
        reminder_at: ISO datetime for reminder (e.g. '2026-03-15T14:00:00').
    """
    # Normalize empty strings to None
    description = description or None
    due_date = due_date or None
    project = project or None
    parent_id = parent_id or None
    notes = notes or None
    recurring = recurring or None
    reminder_at = reminder_at or None

    task_id = str(uuid.uuid4())
    now = _now()

    if err := _validate_task_fields(
        section=section,
        priority=priority,
        type=type,
        due_date=due_date,
        recurring=recurring,
        reminder_at=reminder_at,
    ):
        return json.dumps({"error": err}, ensure_ascii=False)

    with _get_write_conn() as conn:
        if parent_id:
            if not TaskDAO.exists(conn, parent_id):
                return json.dumps(
                    {"error": f"Parent task {parent_id} not found"}, ensure_ascii=False
                )
        _create_task_with_ledger(
            conn,
            task_id,
            title,
            now,
            description=description,
            status="not_started",
            priority=priority,
            section=section,
            due_date=due_date,
            project=project,
            parent_id=parent_id,
            notes=notes,
            recurring=recurring,
            reminder_at=reminder_at,
            type=type,
            tool_name="sqlite-tasks.create_task_or_note",
        )
    _vec_sync_task_safe(task_id)

    logger.info("create_task_or_note: %s (%s)", title, task_id)
    return json.dumps(
        {"task_id": task_id, "title": title, "type": type, "status": "not_started"},
        ensure_ascii=False,
    )


# ═══════════════════════════════════════════════════════════════════════════
# Tool 1b: upsert_note_by_title_project
# ═══════════════════════════════════════════════════════════════════════════
@mcp.tool()
def upsert_note_by_title_project(
    title: str,
    project: str = "",
    description: str = "",
    notes: str = "",
    section: str = "",
    priority: str = "",
    update_if_found: bool = True,
) -> str:
    """Create or update a note by exact normalized title + project.

    This is the idempotent write surface for durable research/decision notes.
    It prevents repeated agent retries from creating near-duplicate note rows.

    Matching rule:
    - exact normalized title + project;
    - no fuzzy overwrite;
    - updates go through the task mutation ledger.

    Args:
        title: Note title (required).
        project: Project tag for matching and grouping.
        description: Primary long-form note body.
        notes: Optional auxiliary/internal metadata.
        section: Section for new notes; updates only when explicitly set.
        priority: Priority for new notes; updates only when explicitly set.
        update_if_found: When false, return the existing row without mutation.
    """
    title = (title or "").strip()
    if not title:
        return json.dumps({"error": "title is required"}, ensure_ascii=False)

    project_value = project.strip() if project else None
    create_section = section or "next"
    create_priority = priority or "medium"
    if err := _validate_task_fields(
        section=create_section,
        priority=create_priority,
        type="note",
    ):
        return json.dumps({"error": err}, ensure_ascii=False)
    update_validation = {
        key: value
        for key, value in {"section": section, "priority": priority}.items()
        if value
    }
    if update_validation and (err := _validate_task_fields(**update_validation)):
        return json.dumps({"error": err}, ensure_ascii=False)

    now = _now()
    embed_task_id: str | None = None
    response: dict[str, Any]
    with _get_write_conn() as conn:
        existing = _find_note_by_title_project(conn, title=title, project=project_value)
        if existing:
            if not update_if_found:
                return json.dumps(
                    {
                        "task_id": existing["id"],
                        "title": existing["title"],
                        "type": "note",
                        "action": "existing",
                        "matched_on": "normalized_title_project",
                    },
                    ensure_ascii=False,
                )

            updates: dict[str, Any] = {}
            if description:
                updates["description"] = description
            if notes:
                updates["notes"] = notes
            if section:
                updates["section"] = section
            if priority:
                updates["priority"] = priority

            if not updates:
                return json.dumps(
                    {
                        "task_id": existing["id"],
                        "title": existing["title"],
                        "type": "note",
                        "action": "existing",
                        "matched_on": "normalized_title_project",
                    },
                    ensure_ascii=False,
                )

            result = _apply_task_mutation(
                conn,
                existing["id"],
                updates,
                timestamp=now,
                tool_name="sqlite-tasks.upsert_note_by_title_project",
            )
            if {"title", "description", "notes"} & set(
                result.get("changed_fields", ())
            ):
                embed_task_id = existing["id"]
            response = {
                "task_id": existing["id"],
                "title": title,
                "type": "note",
                "action": "updated",
                "fields": result.get("changed_fields", []),
                "matched_on": "normalized_title_project",
            }
        else:
            task_id = str(uuid.uuid4())
            _create_task_with_ledger(
                conn,
                task_id,
                title,
                now,
                description=description or None,
                status="not_started",
                priority=create_priority,
                section=create_section,
                project=project_value,
                notes=notes or None,
                type="note",
                tool_name="sqlite-tasks.upsert_note_by_title_project",
            )
            embed_task_id = task_id
            response = {
                "task_id": task_id,
                "title": title,
                "type": "note",
                "status": "not_started",
                "action": "created",
                "matched_on": "normalized_title_project",
            }

    if embed_task_id:
        _vec_sync_task_safe(embed_task_id)
    logger.info(
        "upsert_note_by_title_project: %s (%s)",
        title,
        response["task_id"],
    )
    return json.dumps(response, ensure_ascii=False)


# ═══════════════════════════════════════════════════════════════════════════
# Tool 2: update_task
# ═══════════════════════════════════════════════════════════════════════════
@mcp.tool()
def update_task(
    task_id: str,
    title: str = "",
    description: str = "",
    status: str = "",
    priority: str = "",
    section: str = "",
    due_date: str = "",
    project: str = "",
    parent_id: str = "",
    notes: str = "",
    recurring: str = "",
    reminder_at: str = "",
    type: str = "",
) -> str:
    """Update a task's fields. Only non-empty fields are changed.

    Pass special value "CLEAR" to set a field to NULL.

    Explicitly confirming an archived/cancelled status whose row conflicts with
    visible field/event authority records that confirmation once. Repeating a
    consistent same-status command remains a no-op.

    ``description`` is the primary body field for task/note text.
    ``notes`` is reserved for secondary/internal metadata.

    Args:
        task_id: UUID of the task to update (required).
        title: New title.
        description: New main task/note body.
        status: not_started | in_progress | done | archived | cancelled.
        priority: low | medium | high | critical.
        section: inbox | today | next | someday | waiting.
        due_date: YYYY-MM-DD or "CLEAR" to remove.
        project: Project tag or "CLEAR" to remove.
        parent_id: Parent UUID or "CLEAR" to remove.
        notes: New auxiliary/internal notes or "CLEAR" to remove.
        recurring: JSON config or "CLEAR" to remove.
        reminder_at: ISO datetime or "CLEAR" to remove.
        type: task | note.
    """
    fields = {
        "title": title,
        "description": description,
        "status": status,
        "priority": priority,
        "section": section,
        "due_date": due_date,
        "project": project,
        "parent_id": parent_id,
        "notes": notes,
        "recurring": recurring,
        "reminder_at": reminder_at,
        "type": type,
    }
    updates = {}
    for k, v in fields.items():
        if v == "CLEAR":
            updates[k] = None
        elif v:  # non-empty string = update
            updates[k] = v
    if not updates:
        return json.dumps(
            {"error": "No fields to update. Pass non-empty values."}, ensure_ascii=False
        )

    val_fields = {
        k: v
        for k, v in updates.items()
        if k
        in (
            "status",
            "section",
            "priority",
            "type",
            "due_date",
            "recurring",
            "reminder_at",
        )
        and v is not None
    }
    if err := _validate_task_fields(**val_fields):
        return json.dumps({"error": err}, ensure_ascii=False)

    updates["updated_at"] = _now()

    reembed = False
    with _get_write_conn() as conn:
        result = _apply_task_mutation(
            conn,
            task_id,
            {k: v for k, v in updates.items() if k != "updated_at"},
            timestamp=updates["updated_at"],
            tool_name="sqlite-tasks.update_task",
            explicit_clear_fields=("parent_id",) if parent_id == "CLEAR" else (),
        )
        if result.get("updated", 0) == 0 and result.get("missing"):
            return json.dumps(
                {"error": f"Task {task_id} not found"}, ensure_ascii=False
            )
        # Re-embed if content fields changed
        if {"title", "description", "notes"} & set(result.get("changed_fields", ())):
            reembed = True

    if reembed:
        _vec_sync_task_safe(task_id)
    logger.info("update_task: %s updated %s", task_id, list(updates.keys()))
    return json.dumps(
        {"updated": task_id, "fields": list(updates.keys())}, ensure_ascii=False
    )


# ═══════════════════════════════════════════════════════════════════════════
# Tool 3: query_tasks
# ═══════════════════════════════════════════════════════════════════════════

# Allowlist: maps a caller-supplied sort key to a hardcoded SQL expression.
# Values are NEVER derived from caller input — this is the SQL-injection guard.
# "priority" maps to the enum CASE (critical-first), not the raw TEXT column.
_QUERY_TASKS_SORT_COLUMNS: dict[str, str] = {
    "created_at": "t.created_at",
    "updated_at": "t.updated_at",
    "due_date": "t.due_date",
    "priority": build_priority_order_sql("t."),
    "status": "t.status",
    "title": "t.title",
    "project": "t.project",
    "section": "t.section",
    "type": "t.type",
}

# Per-field default direction when sort_order is omitted. Dates/recency and
# priority lead with the most actionable rows; text fields are alphabetical.
# For "priority" the CASE is critical=0..low=3, so "desc" (highest first)
# means CASE ASC — handled explicitly in the builder below.
_QUERY_TASKS_SORT_DEFAULT_DIR: dict[str, str] = {
    "created_at": "desc",
    "updated_at": "desc",
    "due_date": "asc",
    "priority": "desc",
    "status": "asc",
    "title": "asc",
    "project": "asc",
    "section": "asc",
    "type": "asc",
}

# Default ORDER BY when sort_by is empty — byte-for-byte the historical
# behavior (priority critical-first, due_date ASC NULLS LAST, created_at ASC).
_QUERY_TASKS_DEFAULT_ORDER = (
    f"{build_priority_order_sql('t.')}, t.due_date ASC NULLS LAST, t.created_at ASC"
)


def _build_query_tasks_order(sort_by: str, sort_order: str) -> tuple[str, str | None]:
    """Build a validated ORDER BY clause for query_tasks.

    Returns (order_clause, error). When sort_by is empty, returns the
    historical default and no error. On invalid sort_by/sort_order, returns
    ("", error_message). The SQL column is taken only from the hardcoded
    allowlist — caller input is never interpolated into SQL.
    """
    if not sort_by:
        return _QUERY_TASKS_DEFAULT_ORDER, None

    col = _QUERY_TASKS_SORT_COLUMNS.get(sort_by)
    if col is None:
        allowed = ", ".join(sorted(_QUERY_TASKS_SORT_COLUMNS))
        return "", f"invalid_sort_by: {sort_by}; allowed: [{allowed}]"

    if sort_order:
        direction = sort_order.lower()
        if direction not in ("asc", "desc"):
            return "", f"invalid_sort_order: {sort_order}; allowed: [asc, desc]"
    else:
        direction = _QUERY_TASKS_SORT_DEFAULT_DIR[sort_by]

    if sort_by == "priority":
        # CASE is critical=0..low=3; highest-first ("desc") == CASE ASC.
        sql_dir = "ASC" if direction == "desc" else "DESC"
        primary = f"{col} {sql_dir}"
    elif sort_by == "due_date":
        # NULLS LAST so undated tasks never crowd the top of an ASC sort.
        sql_dir = "ASC" if direction == "asc" else "DESC"
        primary = f"{col} {sql_dir} NULLS LAST"
    else:
        sql_dir = "ASC" if direction == "asc" else "DESC"
        primary = f"{col} {sql_dir}"

    # Deterministic tie-breaker so equal keys never reorder between calls.
    return f"{primary}, t.created_at DESC, t.id ASC", None


@mcp.tool()
def query_tasks(
    section: str = "",
    status: str = "",
    priority: str = "",
    project: str = "",
    parent_id: str = "",
    type: str = "",
    overdue_only: bool = False,
    search: str = "",
    summary_only: bool = False,
    offset: int = 0,
    limit: int = 50,
    sort_by: str = "",
    sort_order: str = "",
    include_completed: bool = False,
) -> str:
    """Query tasks with optional filters. Returns markdown table.

    Filters are combined with AND. Leave empty to skip a filter.
    overdue_only=True shows only tasks past due_date.
    search: full-text search across title, description, notes.
    summary_only=True omits description/notes (faster).

    Liveness default: finished rows (done/archived/cancelled) keep their
        due_date, so an unfiltered query would otherwise be dominated by
        completed work under the priority/due_date default order. Unless the
        caller names a ``status`` explicitly, or sets include_completed=True,
        finished rows are excluded. ``overdue_only`` keeps its historical
        unconditional exclusion — finished work is never "overdue".
    include_completed: opt back in to done/archived/cancelled rows when no
        explicit ``status`` filter is given (e.g. for history/audit views).

    sort_by: optional column to sort by. Empty ("") preserves the default
        ordering (priority critical-first, then due_date ASC NULLS LAST,
        then created_at ASC). Allowed: created_at, updated_at, due_date,
        priority, status, title, project, section, type. Any other value
        returns an error (no silent fallback).
    sort_order: "asc" or "desc" (case-insensitive). Empty ("") uses a
        sensible per-field default: due_date ASC (soonest first);
        created_at/updated_at DESC (most recent first); priority DESC
        (highest first); title/project/section/status/type ASC. Ignored
        when sort_by is empty. Any other value returns an error.

    Note: when ``search`` is set, results are re-ranked by the search engine,
    so ``sort_by`` has little visible effect on that path.
    """
    order_clause, sort_err = _build_query_tasks_order(sort_by, sort_order)
    if sort_err is not None:
        return json.dumps({"error": sort_err}, ensure_ascii=False)

    conditions: list[str] = []
    params: list[Any] = []

    if section:
        conditions.append("t.section = ?")
        params.append(section)
    if status:
        conditions.append("t.status = ?")
        params.append(status)
    if priority:
        conditions.append("t.priority = ?")
        params.append(priority)
    if project:
        conditions.append("t.project = ?")
        params.append(project)
    if parent_id:
        conditions.append("t.parent_id = ?")
        params.append(parent_id)
    if type:
        conditions.append("t.type = ?")
        params.append(type)
    if overdue_only:
        conditions.append("t.due_date < date('now')")
    # Liveness guard. Without it the caller-argument loop above can leave
    # ``WHERE 1=1``, and because completed tasks retain their due_date the
    # default order (priority, then due_date ASC NULLS LAST) buries live work
    # behind finished rows. An explicit ``status`` means the caller already
    # named the lifecycle they want, so it is left untouched.
    if overdue_only or not (status or include_completed):
        conditions.append(f"t.status NOT IN ({_EXCL_PH})")
        params.extend(_TASK_ACTIVE_EXCLUSIONS)

    if summary_only:
        cols = "t.id, t.title, t.status, t.priority, t.section, t.due_date, t.project, t.parent_id, t.notes, t.created_at, t.updated_at"
    else:
        cols = "t.id, t.title, t.description, t.notes, t.status, t.priority, t.section, t.due_date, t.project, t.parent_id, t.created_at, t.updated_at"

    from_clause = "tasks t"

    where = " AND ".join(conditions) if conditions else "1=1"

    with _get_conn() as conn:
        if search:
            # FTS5 pre-filter: narrow rows before search engine re-ranks
            fts_tokens = search.split()
            fts_where = where
            fts_params = list(params)
            if fts_tokens:
                # Prefix match (token*) so pre-filter is broader than
                # the search engine's fuzzy/substring matching
                escaped = []
                for t in fts_tokens:
                    clean = "".join(c for c in t if c.isalnum() or c == "_")
                    if clean:
                        escaped.append(clean + "*")
                fts_match = " OR ".join(escaped) if escaped else None
                if fts_match:
                    try:
                        # Verify FTS5 query is valid before using it
                        conn.execute(
                            "SELECT 1 FROM tasks_fts WHERE tasks_fts MATCH ? LIMIT 1",
                            (fts_match,),
                        )
                        fts_where = (
                            f"{where} AND t.rowid IN "
                            f"(SELECT rowid FROM tasks_fts WHERE tasks_fts MATCH ?)"
                        )
                        fts_params.append(fts_match)
                    except sqlite3.Error as e:
                        logger.debug(
                            "Task FTS5 prefilter failed for %r: %s",
                            search,
                            e,
                            exc_info=True,
                        )

            sql = (
                f"SELECT {cols} FROM {from_clause} WHERE {fts_where} "
                f"ORDER BY {order_clause}"
            )
            all_rows = conn.execute(sql, fts_params).fetchall()
            results = _get_search_engine().search(
                search, [dict(r) for r in all_rows], conn=conn
            )
            total = len(results)
            rows = results[offset : offset + limit]
        else:
            sql = (
                f"SELECT {cols} FROM {from_clause} WHERE {where} "
                f"ORDER BY {order_clause} "
                f"LIMIT ? OFFSET ?"
            )
            rows = conn.execute(sql, params + [limit, offset]).fetchall()
            count_sql = f"SELECT COUNT(*) FROM {from_clause} WHERE {where}"
            total = conn.execute(count_sql, params).fetchone()[0]

    rows = [dict(r) if not isinstance(r, dict) else r for r in rows] if rows else []

    if not rows:
        return json.dumps(
            {"tasks": [], "count": 0, "total": total, "message": "No tasks match"},
            ensure_ascii=False,
        )

    lines = [
        "| # | Title | Status | Priority | Section | Due | Project | Created | Notes |",
        "|---|-------|--------|----------|---------|-----|---------|---------|-------|",
    ]
    for i, r in enumerate(rows, 1):
        due = r["due_date"] or "—"
        proj = r["project"] or "—"
        created = (r.get("created_at") or "—")[:16]
        notes = (r["notes"] or "—")[:80]
        lines.append(
            f"| {i + offset} | {r['title']} | {r['status']} | {r['priority']} "
            f"| {r['section']} | {due} | {proj} | {created} | {notes} |"
        )

    result = {
        "tasks": rows,
        "count": len(rows),
        "total": total,
        "offset": offset,
        "limit": limit,
        "markdown": "\n".join(lines),
    }
    if total > offset + limit:
        result["has_more"] = True
        result["next_offset"] = offset + limit
    return json.dumps(result, ensure_ascii=False)


# ═══════════════════════════════════════════════════════════════════════════
# Tool 3b: find_by_title
# ═══════════════════════════════════════════════════════════════════════════

_LOOKUP_TASK_COLUMNS = (
    "id, title, description, notes, type, status, section, priority, due_date, "
    "project, updated_at, created_at"
)
_LOOKUP_ENTITY_COLUMNS = "id, name, entity_type, project, updated_at, created_at"


def _lookup_prefilter_ids(
    conn: sqlite3.Connection, query: str, candidate_limit: int
) -> tuple[list[int], list[int]] | None:
    """Return broad FTS/project candidates, or None for exact full-scan fallback."""
    try:
        fts_q = _fts_query(query)
        task_ids = {
            row["rowid"]
            for row in conn.execute(
                "SELECT rowid FROM tasks_fts WHERE tasks_fts MATCH ? "
                "ORDER BY rank LIMIT ?",
                (fts_q, candidate_limit),
            )
        }
        entity_ids = {
            row["rowid"]
            for row in conn.execute(
                "SELECT rowid FROM memory_fts WHERE memory_fts MATCH ? "
                "ORDER BY rank LIMIT ?",
                (fts_q, candidate_limit),
            )
        }

        # Project is part of the retrieval contract but is not indexed by either
        # FTS table. Include normalized project phrase hits so the prefilter does
        # not erase project-only recall (covered by the locked eval corpus).
        normalized = normalize_lookup_text(query)
        if normalized and normalized.isascii():
            project_expr = (
                "lower(replace(replace(coalesce(project, ''), '-', ' '), '_', ' '))"
            )
            task_ids.update(
                row["rowid"]
                for row in conn.execute(
                    f"SELECT rowid FROM tasks WHERE instr({project_expr}, ?) > 0 "
                    "LIMIT ?",
                    (normalized, candidate_limit),
                )
            )
            entity_ids.update(
                row["id"]
                for row in conn.execute(
                    f"SELECT id FROM entities WHERE instr({project_expr}, ?) > 0 "
                    "LIMIT ?",
                    (normalized, candidate_limit),
                )
            )
    except sqlite3.Error:
        return None
    if not task_ids and not entity_ids:
        return None
    return sorted(task_ids), sorted(entity_ids)


def _lookup_rows(
    conn: sqlite3.Connection,
    query: str,
    limit: int,
    *,
    force_full_scan: bool = False,
) -> tuple[list[sqlite3.Row], list[sqlite3.Row], dict[int, list[str]], str]:
    """Load only lookup candidates; retain an exact full-scan fallback."""
    candidate_ids = None
    if not force_full_scan:
        # Stay below SQLite's historical 999 host-parameter ceiling because
        # candidate IDs are consumed by one IN (...) query per surface.
        candidate_limit = min(900, max(200, limit * 10))
        candidate_ids = _lookup_prefilter_ids(conn, query, candidate_limit)

    if candidate_ids is None:
        # No status filter here, deliberately. The liveness sort key already
        # guarantees a finished row never outranks a live one, so excluding
        # them buys nothing for ranking. It would only change behaviour in the
        # one case where there is no live answer at all — turning a
        # low-confidence closed match into "no result". This is the last-resort
        # recall path of a system whose job is remembering past work, so
        # dropping finished rows from it costs more than it returns.
        task_rows = conn.execute(f"SELECT {_LOOKUP_TASK_COLUMNS} FROM tasks").fetchall()
        entity_rows = conn.execute(
            f"SELECT {_LOOKUP_ENTITY_COLUMNS} FROM entities"
        ).fetchall()
        strategy = "full_scan_fallback"
    else:
        task_ids, entity_ids = candidate_ids
        if task_ids:
            ph = ",".join("?" * len(task_ids))
            task_rows = conn.execute(
                f"SELECT {_LOOKUP_TASK_COLUMNS} FROM tasks WHERE rowid IN ({ph})",
                task_ids,
            ).fetchall()
        else:
            task_rows = []
        if entity_ids:
            ph = ",".join("?" * len(entity_ids))
            entity_rows = conn.execute(
                f"SELECT {_LOOKUP_ENTITY_COLUMNS} FROM entities WHERE id IN ({ph})",
                entity_ids,
            ).fetchall()
        else:
            entity_rows = []
        strategy = "fts_prefilter"

    obs_by_entity: dict[int, list[str]] = {}
    if entity_rows:
        entity_ids = [row["id"] for row in entity_rows]
        ph = ",".join("?" * len(entity_ids))
        for row in conn.execute(
            f"SELECT entity_id, content FROM observations "
            f"WHERE entity_id IN ({ph}) ORDER BY entity_id, id",
            entity_ids,
        ):
            obs_by_entity.setdefault(row["entity_id"], []).append(row["content"])
    return task_rows, entity_rows, obs_by_entity, strategy


def _score_task_lookup(row: sqlite3.Row, query: str) -> dict[str, Any] | None:
    title = row["title"] or ""
    surface_scores = {
        surface: score
        for surface, score in (
            ("title", score_lookup_surface("title", query, title)),
            (
                "description",
                score_lookup_surface("description", query, row["description"]),
            ),
            ("notes", score_lookup_surface("notes", query, row["notes"])),
            ("project", score_lookup_surface("project", query, row["project"])),
        )
        if score > 0
    }
    if not surface_scores:
        return None
    ordered_hits = order_surface_hits(surface_scores)
    matched_in = [surface for surface, _ in ordered_hits]
    return {
        "kind": "note" if row["type"] == "note" else "task",
        "id": row["id"],
        "title": title,
        "type": row["type"],
        "status": row["status"],
        "section": row["section"],
        "priority": row["priority"],
        "due_date": row["due_date"],
        "project": row["project"],
        "updated_at": row["updated_at"],
        "created_at": row["created_at"],
        "score": float(ordered_hits[0][1]),
        "matched_in": matched_in,
        "surface_scores": surface_scores,
        "primary_surface": matched_in[0],
        "confidence": classify_lookup_confidence(surface_scores),
        "ranking_contract_version": RETRIEVAL_CONTRACT_VERSION,
    }


def _score_entity_lookup(
    row: sqlite3.Row,
    query: str,
    observations: list[str],
) -> dict[str, Any] | None:
    name = row["name"] or ""
    obs_text = "\n".join(observations)
    surface_scores = {
        surface: score
        for surface, score in (
            ("name", score_lookup_surface("name", query, name)),
            (
                "observations",
                score_lookup_surface("observations", query, obs_text),
            ),
            ("project", score_lookup_surface("project", query, row["project"])),
            (
                "entity_type",
                score_lookup_surface("entity_type", query, row["entity_type"]),
            ),
        )
        if score > 0
    }
    if not surface_scores:
        return None
    ordered_hits = order_surface_hits(surface_scores)
    matched_in = [surface for surface, _ in ordered_hits]
    return {
        "kind": "entity",
        "id": row["id"],
        "title": name,
        "entityType": row["entity_type"],
        "project": row["project"],
        "updated_at": row["updated_at"],
        "created_at": row["created_at"],
        "score": float(ordered_hits[0][1]),
        "matched_in": matched_in,
        "surface_scores": surface_scores,
        "primary_surface": matched_in[0],
        "confidence": classify_lookup_confidence(surface_scores),
        "ranking_contract_version": RETRIEVAL_CONTRACT_VERSION,
    }


def _score_lookup_candidates(
    task_rows: list[sqlite3.Row],
    entity_rows: list[sqlite3.Row],
    obs_by_entity: dict[int, list[str]],
    query: str,
) -> list[dict[str, Any]]:
    matches = [
        scored
        for row in task_rows
        if (scored := _score_task_lookup(row, query)) is not None
    ]
    matches.extend(
        scored
        for row in entity_rows
        if (
            scored := _score_entity_lookup(row, query, obs_by_entity.get(row["id"], []))
        )
        is not None
    )
    return matches


def _lookup_liveness(item: dict[str, Any]) -> int:
    """Liveness term for lookup ordering: 0 for finished task-lifecycle rows.

    Deliberately outside ``retrieval_contract.SURFACE_SCORE_RULES``: scoring
    stays a pure text-surface function of the query, and status only breaks the
    ordering of already-scored matches. Entities carry no lifecycle status, so
    they are never demoted as stale.
    """
    return 0 if (item.get("status") or "") in _TASK_ACTIVE_EXCLUSIONS else 1


@mcp.tool()
def find_by_title(title_fragment: str, limit: int = 20) -> str:
    """Find tasks, notes, or entities by partial title or remembered phrase.

    This is the cross-surface lookup tool when the caller only remembers a
    phrase, not the storage surface. It searches across task titles,
    descriptions, notes, projects, entity names, and entity observations.
    It applies no section, type, or project filter. Finished rows
    (done/archived/cancelled) stay reachable on every path, including the
    unbounded full-scan fallback; they are never excluded, only ranked below
    live work.

    Args:
        title_fragment: Any distinctive substring or remembered phrase.
        limit: Maximum number of matches to return.
    """
    query = (title_fragment or "").strip()
    if not query:
        return json.dumps(
            {"matches": [], "count": 0, "message": "Empty title fragment"},
            ensure_ascii=False,
        )

    limit = max(1, min(int(limit), 100))
    matches: list[dict[str, Any]] = []

    with _get_conn() as conn:
        task_rows, entity_rows, obs_by_entity, strategy = _lookup_rows(
            conn, query, limit
        )
        matches.extend(
            _score_lookup_candidates(task_rows, entity_rows, obs_by_entity, query)
        )

        # FTS OR queries can surface only low-signal token collisions. In that
        # case use the exact historical scan so a strong substring match outside
        # the bounded candidate pool cannot disappear silently.
        if strategy == "fts_prefilter" and not any(
            is_visible_lookup_match(item.get("surface_scores") or {})
            for item in matches
        ):
            task_rows, entity_rows, obs_by_entity, strategy = _lookup_rows(
                conn, query, limit, force_full_scan=True
            )
            matches = _score_lookup_candidates(
                task_rows, entity_rows, obs_by_entity, query
            )

    # Liveness leads the sort key: a stale row never outranks a live one, no
    # matter how well it scores. Score then recency decide within a liveness
    # band; Python's stable sort keeps equal keys in deterministic row order.
    matches.sort(
        key=lambda item: (
            _lookup_liveness(item),
            float(item.get("score") or 0.0),
            item.get("updated_at") or "",
        ),
        reverse=True,
    )
    confident_matches = [
        item
        for item in matches
        if is_visible_lookup_match(item.get("surface_scores") or {})
    ]
    hidden_low_confidence = (
        len(matches) - len(confident_matches) if confident_matches else 0
    )
    visible_matches = confident_matches if confident_matches else matches
    matches = visible_matches[:limit]

    lines = [
        "| # | Kind | Title | Status/Type | Section | Project | Match | Confidence |",
        "|---|------|-------|-------------|---------|---------|-------|------------|",
    ]
    for idx, item in enumerate(matches, 1):
        if item["kind"] == "entity":
            status_or_type = item.get("entityType") or "entity"
            section = "—"
        else:
            status_or_type = (
                f"{item.get('status') or '—'} / {item.get('type') or 'task'}"
            )
            section = item.get("section") or "—"
        lines.append(
            f"| {idx} | {item['kind']} | {item['title']} | {status_or_type} | "
            f"{section} | {item.get('project') or '—'} | {', '.join(item.get('matched_in') or []) or '—'} | "
            f"{item.get('confidence') or 'low'} |"
        )

    return json.dumps(
        {
            "matches": matches,
            "count": len(matches),
            "query": query,
            "hidden_low_confidence_count": hidden_low_confidence,
            "ranking_contract_version": RETRIEVAL_CONTRACT_VERSION,
            "lookup_strategy": strategy,
            "markdown": "\n".join(lines) if matches else "",
            "message": None if matches else "No title matches",
        },
        ensure_ascii=False,
    )


# ═══════════════════════════════════════════════════════════════════════════
# Tool 4: task_digest
# ═══════════════════════════════════════════════════════════════════════════
@mcp.tool()
def task_digest(
    include_overdue: bool = True,
    limit: int = 20,
) -> str:
    """Generate a formatted task digest for session start.

    Shows pending/in-progress tasks grouped by section,
    plus overdue tasks highlighted separately.
    """
    target_sections = ["today", "inbox", "next"]

    with _get_conn() as conn:
        active, overdue, counts = TaskDAO.digest_snapshot(
            conn,
            target_sections,
            include_overdue=include_overdue,
            limit=limit,
        )

    lines = ["## Task Digest"]
    if counts:
        stats = {r["status"]: r["cnt"] for r in counts}
        total = sum(stats.values())
        lines.append(
            f"**Total active:** {total} | "
            f"Not started: {stats.get('not_started', 0)} | "
            f"In progress: {stats.get('in_progress', 0)} | "
            f"Done: {stats.get('done', 0)}"
        )
        lines.append("")

    if overdue:
        lines.append(f"### OVERDUE ({len(overdue)})")
        for t in overdue:
            note_hint = f" | {t['notes'][:60]}..." if t["notes"] else ""
            lines.append(
                f"- [{t['priority'].upper()}] {t['title']} (due: {t['due_date']}){note_hint}"
            )
        lines.append("")

    by_section: dict[str, list] = {}
    for t in active:
        by_section.setdefault(t["section"], []).append(t)

    for sec in target_sections:
        tasks = by_section.get(sec, [])
        if tasks:
            lines.append(f"### {sec.upper()} ({len(tasks)})")
            for t in tasks:
                due = f" [due: {t['due_date']}]" if t["due_date"] else ""
                prio = (
                    f"[{t['priority'].upper()}] " if t["priority"] != "medium" else ""
                )
                note_hint = f" | {t['notes'][:60]}..." if t["notes"] else ""
                lines.append(f"- {prio}{t['title']}{due}{note_hint}")
            lines.append("")

    return json.dumps(
        {
            "digest": "\n".join(lines),
            "active_count": len(active),
            "overdue_count": len(overdue),
        },
        ensure_ascii=False,
    )


# ═══════════════════════════════════════════════════════════════════════════
# Tool 5: archive_done_tasks
# ═══════════════════════════════════════════════════════════════════════════
@mcp.tool()
def archive_done_tasks(older_than_days: int = 7) -> str:
    """Archive completed tasks older than N days.

    Moves tasks with status='done' and updated_at older than threshold to 'archived'.
    """
    if older_than_days < 0:
        return json.dumps(
            {"error": "older_than_days must be non-negative"}, ensure_ascii=False
        )

    with _get_write_conn() as conn:
        affected_ids = TaskDAO.archive_done(conn, older_than_days)

    logger.info(
        "archive_done_tasks: %d archived (older than %d days)",
        len(affected_ids),
        older_than_days,
    )
    return json.dumps(
        {"archived": len(affected_ids), "threshold_days": older_than_days},
        ensure_ascii=False,
    )


# ═══════════════════════════════════════════════════════════════════════════
# Tool 6: bump_overdue_priority
# ═══════════════════════════════════════════════════════════════════════════
@mcp.tool()
def bump_overdue_priority(target_priority: str = "high") -> str:
    """Bump priority of overdue tasks not done/archived.

    Only bumps tasks with priority lower than target.
    """
    if target_priority not in _TASK_PRIORITIES:
        return json.dumps(
            {"error": f"Invalid priority: {target_priority}"}, ensure_ascii=False
        )
    if _TASK_PRIORITIES.index(target_priority) == 0:
        return json.dumps(
            {"bumped": 0, "message": "No lower priorities to bump"}, ensure_ascii=False
        )

    with _get_write_conn() as conn:
        bumped = TaskDAO.bump_overdue_priority(
            conn,
            target_priority,
            tool_name="sqlite-tasks.bump_overdue_priority",
        )

    logger.info("bump_overdue_priority: %d bumped to %s", len(bumped), target_priority)
    return json.dumps(
        {"bumped": len(bumped), "target_priority": target_priority}, ensure_ascii=False
    )


def _strip_ready_task_payload(record: dict[str, Any]) -> dict[str, Any]:
    out = dict(record)
    out.pop("task", None)
    return out


# ═══════════════════════════════════════════════════════════════════════════
# Tool 8: ready_context
# ═══════════════════════════════════════════════════════════════════════════
@mcp.tool()
def ready_context(
    mode: str = "ready",
    include_readings: bool = False,
    limit: int = 12,
) -> str:
    """Return deterministic ready/prime context with reasons and provenance.

    This is sqlite-memory-mcp's cross-project/cross-machine answer to Beads
    ``bd ready`` / ``bd prime``.

    Args:
        mode: ready | suggested | prime.
            ready returns structured ready-context records.
            suggested returns the bounded Suggested-tab candidate set.
            prime returns a compact session boot pack from the same records.
        include_readings: include reading notes in work surfaces.
        limit: max items per list.
    """
    safe_limit = max(1, min(int(limit or 12), 100))
    with _get_conn() as conn:
        tasks = TaskDAO.get_active(conn)
        tasks.extend(TaskDAO.get_ready_review_candidates(conn, limit=50))

    normalized_mode = (mode or "ready").strip().lower()
    if normalized_mode == "suggested":
        rows = _suggested_ready(
            tasks,
            include_readings=include_readings,
            limit=safe_limit,
        )
        return json.dumps(
            {
                "contract_version": _READY_CONTEXT_CONTRACT_VERSION,
                "mode": "suggested",
                "count": len(rows),
                "items": rows,
            },
            ensure_ascii=False,
        )

    if normalized_mode == "prime":
        pack = _prime_context(
            tasks,
            include_readings=include_readings,
            limit=min(safe_limit, 50),
        )
        for key in (
            "top_ready_items",
            "blocked_or_waiting",
            "cleanup_candidates",
            "explicit_exclusions",
            "risk_or_escalation_items",
        ):
            pack[key] = [
                _strip_ready_task_payload(record) for record in pack.get(key, [])
            ]
        pack["mode"] = "prime"
        return json.dumps(pack, ensure_ascii=False)

    if normalized_mode != "ready":
        return json.dumps(
            {
                "error": f"Invalid mode: {mode}",
                "valid_modes": ["ready", "suggested", "prime"],
            },
            ensure_ascii=False,
        )

    records = _ready_context(tasks, include_readings=include_readings)
    selected = [_strip_ready_task_payload(record) for record in records[:safe_limit]]
    return json.dumps(
        {
            "contract_version": _READY_CONTEXT_CONTRACT_VERSION,
            "mode": "ready",
            "count": len(selected),
            "truncated": len(records) > safe_limit,
            "items": selected,
        },
        ensure_ascii=False,
    )


# ── Entry point ──────────────────────────────────────────────────────────
def main() -> None:
    maybe_mount_premium_extensions(mcp, server_name="sqlite-tasks")
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()