Function save_chat_message

skills/storage-helper/scripts/db_ops.py:329–410

Persist a GUI chat message and update its parent session summary.

(
    user_id: str,
    session_id: str,
    role: str,
    content: str,
    metadata: Optional[Dict[str, Any]] = None,
)

def save_chat_message(
    user_id: str,
    session_id: str,
    role: str,
    content: str,
    metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """Persist a GUI chat message and update its parent session summary."""
    normalized_user_id = _normalize_identifier(user_id)
    normalized_session_id = _normalize_identifier(session_id)
    normalized_role = str(role or "").strip().lower()
    if not normalized_user_id or not normalized_session_id:
        raise ValueError("user_id and session_id are required")
    if normalized_role not in {"user", "assistant"}:
        raise ValueError("role must be user or assistant")

    ensure_chat_tables()
    create_chat_session(normalized_user_id, str(content or ""), normalized_session_id)

    now = _chat_now()
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """
        INSERT INTO chat_messages (session_id, user_id, role, content, metadata, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        (
            normalized_session_id,
            normalized_user_id,
            normalized_role,
            str(content or ""),
            json.dumps(metadata or {}, ensure_ascii=False),
            now,
        ),
    )
    message_id = cursor.lastrowid
    if normalized_role == "user":
        cursor.execute(
            """
            UPDATE chat_sessions
            SET title = CASE WHEN title IN ('', '新对话', 'New chat') THEN ? ELSE title END,
                updated_at = ?
            WHERE user_id = ? AND session_id = ?
            """,
            (_chat_title_from_content(content), now, normalized_user_id, normalized_session_id),
        )
    else:
        cursor.execute(
            """
            UPDATE chat_sessions
            SET updated_at = ?
            WHERE user_id = ? AND session_id = ?
            """,
            (now, normalized_user_id, normalized_session_id),
        )
    cursor.execute(
        """