hkuds--deeptutor/deeptutor/tools/list_notebook.py

"""List the user's notebooks or the records inside a single notebook.

Two modes, picked by whether ``notebook_id`` is supplied:

* **Index mode** (``notebook_id`` empty): list every notebook the
  active user owns — id, name, description, record count, last-updated.
  Equivalent to the sidebar view a human sees.
* **Drill-down mode** (``notebook_id`` supplied): list every record
  inside one notebook — record id, title, type, summary, created_at.
  Equivalent to opening a notebook and skimming entries.

The tool itself is stateless and synchronous; the chat pipeline auto-
mounts it only when the active user actually has notebooks, so empty
runs are impossible by construction. Output is markdown so the LLM can
quote the rendered list back to the user without reformatting.
"""

from __future__ import annotations

from dataclasses import dataclass
import datetime as _dt
import logging
from typing import Any

logger = logging.getLogger(__name__)

# Defensive ceilings so a freak case (user with thousands of notebooks
# or a notebook with thousands of records) can't blow up the LLM's
# context window. The cap is recorded in the rendered output so the
# model knows truncation happened.
MAX_NOTEBOOKS_RENDERED = 50
MAX_RECORDS_RENDERED = 80
MAX_TITLE_PREVIEW = 100
MAX_SUMMARY_PREVIEW = 240


@dataclass(frozen=True)
class ListOutcome:
    """Result of one ``list_notebook`` invocation."""

    ok: bool
    text: str = ""
    error: str = ""
    # Structured echo for the frontend / downstream tools. Kept tiny:
    # callers needing the full dump should call the manager directly.
    summary: dict[str, Any] | None = None


def list_notebooks_or_records(
    *,
    notebook_id: str = "",
    notebook_manager: Any = None,
) -> ListOutcome:
    """Run the list operation. Empty ``notebook_id`` → index mode.

    Errors out cleanly as ``ListOutcome(ok=False, error=...)`` instead
    of raising — the chat loop converts errors back into LLM-visible
    tool results, not exceptions.
    """
    manager = notebook_manager
    if manager is None:
        from deeptutor.services.notebook import get_notebook_manager

        manager = get_notebook_manager()

    nid = (notebook_id or "").strip()
    if not nid:
        return _render_index(manager)
    return _render_records(manager, nid)


def _render_index(manager: Any) -> ListOutcome:
    notebooks = manager.list_notebooks() or []
    if not isinstance(notebooks, list) or not notebooks:
        return ListOutcome(
            ok=True,
            text="The user has no notebooks yet.",
            summary={"mode": "index", "count": 0},
        )
    total = len(notebooks)
    sliced = notebooks[:MAX_NOTEBOOKS_RENDERED]
    lines: list[str] = ["**User notebooks**"]
    for nb in sliced:
        nb_id = str(nb.get("id") or "").strip()
        name = str(nb.get("name") or nb.get("title") or "").strip() or nb_id
        description = _clip(str(nb.get("description") or "").strip(), 200)
        count = nb.get("record_count")
        if not isinstance(count, int):
            try:
                count = int(count or 0)
            except (TypeError, ValueError):
                count = 0
        updated = _format_timestamp(nb.get("updated_at"))
        bullet = f"- `{nb_id}` — **{name}** ({count} records"
        if updated:
            bullet += f", updated {updated}"
        bullet += ")"
        if description:
            bullet += f"\n  {description}"
        lines.append(bullet)
    if total > len(sliced):
        lines.append(
            f"\n_(showing {len(sliced)} of {total} notebooks — call with "
            "a specific notebook_id to drill in.)_"
        )
    return ListOutcome(
        ok=True,
        text="\n".join(lines),
        summary={"mode": "index", "count": total},
    )


def _render_records(manager: Any, notebook_id: str) -> ListOutcome:
    # Validate the id against the user's actual notebooks first so the
    # error message can tell the model the right ids to choose from.
    notebooks = manager.list_notebooks() or []
    matched = next(
        (nb for nb in notebooks if str(nb.get("id") or "").strip() == notebook_id),
        None,
    )
    if matched is None:
        valid_ids = ", ".join(f"`{nb.get('id')}`" for nb in notebooks if nb.get("id"))
        return ListOutcome(
            ok=False,
            error=(f"Unknown notebook_id {notebook_id!r}. Valid ids: {valid_ids or '(none)'}."),
        )

    try:
        records = manager.get_records(notebook_id) or []
    except Exception as exc:  # pragma: no cover — defensive
        logger.warning("list_notebook: get_records failed", exc_info=True)
        return ListOutcome(ok=False, error=f"Failed to load records: {exc}")

    notebook_name = str(matched.get("name") or matched.get("title") or notebook_id)
    if not records:
        return ListOutcome(
            ok=True,
            text=f"Notebook **{notebook_name}** (`{notebook_id}`) has no records yet.",
            summary={"mode": "records", "notebook_id": notebook_id, "count": 0},
        )

    total = len(records)
    # Newest first — the most-recent record is typically the one the
    # user wants to find or edit.
    sorted_records = sorted(
        records,
        key=lambda r: r.get("created_at") or 0,
        reverse=True,
    )
    sliced = sorted_records[:MAX_RECORDS_RENDERED]
    lines: list[str] = [
        f"**Records in notebook `{notebook_id}` — {notebook_name}**",
    ]
    for rec in sliced:
        rid = str(rec.get("id") or "").strip()
        title = _clip(str(rec.get("title") or "").strip(), MAX_TITLE_PREVIEW)
        rtype = str(rec.get("type") or "").strip()
        summary = _clip(
            str(rec.get("summary") or "").strip(),
            MAX_SUMMARY_PREVIEW,
        )
        created = _format_timestamp(rec.get("created_at"))
        head = f"- `{rid}` — **{title or '(untitled)'}**"
        if rtype:
            head += f" [{rtype}]"
        if created:
            head += f" — {created}"
        lines.append(head)
        if summary:
            lines.append(f"  {summary}")
    if total > len(sliced):
        lines.append(f"\n_(showing {len(sliced)} most-recent of {total} records.)_")
    return ListOutcome(
        ok=True,
        text="\n".join(lines),
        summary={"mode": "records", "notebook_id": notebook_id, "count": total},
    )


def _clip(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    return text[:limit].rstrip() + "…"


def _format_timestamp(raw: Any) -> str:
    """Render an epoch-seconds value as ``YYYY-MM-DD`` for at-a-glance scan.

    Falls back to ``""`` on missing / unparseable inputs so the caller
    can omit the field cleanly.
    """
    if raw is None:
        return ""
    try:
        ts = float(raw)
    except (TypeError, ValueError):
        return ""
    if ts <= 0:
        return ""
    try:
        return _dt.datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
    except (OverflowError, OSError, ValueError):
        return ""


__all__ = [
    "MAX_NOTEBOOKS_RENDERED",
    "MAX_RECORDS_RENDERED",
    "ListOutcome",
    "list_notebooks_or_records",
]