playchat/backend/app/core/notes.py

"""ユーザーとAIが共同で編集する共有ノート。

システムプロンプトにはページの「タイトル一覧」だけを注入し、本文はモデルが
read_note ツールで必要なときだけ読み出す。書き込みは write_note ツールで行う。
"""
import logging
import re

from sqlalchemy.orm import Session as DBSession

from app.core.config import NOTES_TOOL_CONTENT_LIMIT
from app.db.models import NotePage

logger = logging.getLogger(__name__)


READ_NOTE_TOOL = {
    "type": "function",
    "function": {
        "name": "read_note",
        "description": "共有ノートの特定ページの本文を読み出す。システムプロンプトに提示されたタイトル一覧の中から、"
        "今の会話に関係しそうなページがあるときに使う。",
        "parameters": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "読み出すページのタイトル（タイトル一覧に載っているもの）"},
            },
            "required": ["title"],
        },
    },
}

WRITE_NOTE_TOOL = {
    "type": "function",
    "function": {
        "name": "write_note",
        "description": "共有ノートのページに書き込む。ユーザーからノートへの記録・追記・整理を頼まれたときに使う。"
        "存在しないタイトルを指定した場合は新しいページを作成する。"
        "既存ページを replace で上書きすると元の内容は失われるので、追記でよい場合は必ず append を使うこと。",
        "parameters": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "書き込む対象のページタイトル"},
                "content": {"type": "string", "description": "書き込む本文（Markdown可）"},
                "mode": {
                    "type": "string",
                    "enum": ["append", "replace"],
                    "description": "append=末尾に追記（既定） / replace=ページ全体を置き換え",
                },
            },
            "required": ["title", "content"],
        },
    },
}

NOTE_TOOLS = [READ_NOTE_TOOL, WRITE_NOTE_TOOL]


def _find_page(db: DBSession, title: str) -> NotePage | None:
    """タイトルでページを探す。完全一致 → 大文字小文字・前後空白を無視した一致 の順に試す。"""
    pages = db.query(NotePage).order_by(NotePage.position, NotePage.id).all()
    for page in pages:
        if page.title == title:
            return page
    normalized = title.strip().casefold()
    for page in pages:
        if page.title.strip().casefold() == normalized:
            return page
    return None


def next_position(db: DBSession) -> int:
    """新規ページを末尾に置くための position を返す。"""
    last = db.query(NotePage).order_by(NotePage.position.desc()).first()
    return (last.position + 1) if last else 0


def get_notes_context(db: DBSession) -> str | None:
    """共有ノートのタイトル一覧をシステムプロンプト用のテキストとして返す。"""
    pages = db.query(NotePage).order_by(NotePage.position, NotePage.id).all()

    lines = [
        "【共有ノート】",
        "ユーザーとあなたが共同で編集できるノートです。以下のページがあります。",
        "",
    ]
    if pages:
        for page in pages:
            preview = " ".join(page.content.split())[:40]
            suffix = f" — {preview}…" if preview else " — (空)"
            lines.append(f"- {page.title}{suffix}")
    else:
        lines.append("- (まだページがありません)")
    lines += [
        "",
        "本文はここには含まれていません。内容が必要なときは read_note ツールで読み出してください。",
        "ノートへの記録・追記を頼まれたときは write_note ツールで書き込んでください"
        "（新しいタイトルを指定すればページが新規作成されます）。",
        "ユーザーが明示的に頼んでいないのに勝手に書き込まないこと。",
    ]
    return "\n".join(lines)


def read_note(db: DBSession, title: str) -> str:
    """ページ本文をツール結果用のテキストとして返す。"""
    page = _find_page(db, title)
    if not page:
        available = ", ".join(p.title for p in db.query(NotePage).order_by(NotePage.position, NotePage.id).all())
        return f"「{title}」というページはありません。現在のページ: {available or '(なし)'}"
    if not page.content.strip():
        return f"ページ「{page.title}」は空です。"
    content = page.content
    if len(content) > NOTES_TOOL_CONTENT_LIMIT:
        content = content[:NOTES_TOOL_CONTENT_LIMIT] + "\n…(以下省略)"
    return f"ページ「{page.title}」の本文:\n\n{content}"


_LIST_ITEM = re.compile(r"^\s*([-*+]|\d+[.)])\s")


def _append(existing: str, addition: str) -> str:
    """既存本文に追記する。箇条書き同士をつなぐ場合だけ空行を挟まない。"""
    if not existing.strip():
        return addition
    body = existing.rstrip()
    last_line = body.rsplit("\n", 1)[-1]
    first_line = addition.lstrip("\n").split("\n", 1)[0]
    separator = "\n" if _LIST_ITEM.match(last_line) and _LIST_ITEM.match(first_line) else "\n\n"
    return f"{body}{separator}{addition}"


def write_note(db: DBSession, title: str, content: str, mode: str = "append") -> str:
    """ページに書き込む。存在しなければ新規作成する。"""
    if not title.strip():
        return "タイトルが空のため書き込めませんでした。"

    page = _find_page(db, title)
    if page is None:
        page = NotePage(title=title.strip(), content=content, position=next_position(db))
        db.add(page)
        db.commit()
        return f"新しいページ「{page.title}」を作成し、本文を書き込みました。"

    if mode == "replace":
        page.content = content
        result = f"ページ「{page.title}」の内容を置き換えました。"
    else:
        page.content = _append(page.content, content)
        result = f"ページ「{page.title}」に追記しました。"
    db.commit()
    return result


def execute_note_tool(db: DBSession, name: str, args: dict) -> tuple[str, bool]:
    """ノート系ツールを実行する。戻り値: (ツール結果テキスト, ノートを変更したか)。"""
    try:
        if name == "read_note":
            return read_note(db, args.get("title", "")), False
        if name == "write_note":
            result = write_note(
                db,
                args.get("title", ""),
                args.get("content", ""),
                args.get("mode", "append"),
            )
            return result, True
    except Exception as e:
        db.rollback()
        logger.exception("Note tool %s failed", name)
        return f"ノート操作に失敗しました: {e}", False
    return f"未知のツールです: {name}", False