import json from typing import Any from sqlalchemy.orm import Session from app.memory.service import MemoryService from app.pomodoro.service import PomodoroService from app.projects.service import ProjectService TOOL_DEFINITIONS: list[dict[str, Any]] = [ { "type": "function", "function": { "name": "get_pomodoro_status", "description": "ОБЯЗАТЕЛЬНО вызывай перед любым ответом о таймере. Статус, фаза и прогресс цикла.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "start_pomodoro", "description": "Запустить фазу работы в цикле помидоро (25 мин по умолчанию).", "parameters": { "type": "object", "properties": { "duration_min": {"type": "integer", "description": "Минуты работы"}, "task_note": {"type": "string", "description": "Над чем работаем"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "start_short_break", "description": "Запустить короткий перерыв между работами.", "parameters": { "type": "object", "properties": { "duration_min": {"type": "integer", "description": "Минуты перерыва"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "start_long_break", "description": "Запустить длинный перерыв после завершения цикла работ.", "parameters": { "type": "object", "properties": { "duration_min": {"type": "integer", "description": "Минуты перерыва"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "stop_pomodoro", "description": "Остановить текущую фазу таймера.", "parameters": { "type": "object", "properties": { "result": {"type": "string", "description": "Отчёт о сделанном"}, "completed": { "type": "boolean", "description": "True если фаза полностью завершена", }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "skip_pomodoro_phase", "description": "Досрочно завершить текущую фазу и перейти к следующей в цикле.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "reset_pomodoro_cycle", "description": "Сбросить цикл помидоро: обнулить счётчик работ и остановить таймер.", "parameters": { "type": "object", "properties": { "clear_task": { "type": "boolean", "description": "Также очистить текущую задачу", }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "get_pomodoro_history", "description": "История помидоро-сессий (таймер), не Taiga-задачи.", "parameters": { "type": "object", "properties": { "limit": {"type": "integer", "description": "Сколько сессий вернуть"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "sync_taiga_projects", "description": "Синхронизировать список проектов из Taiga API. Вызывай если проекты неизвестны.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "list_taiga_projects", "description": "Список проектов Taiga с привязкой Gitea.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "list_taiga_tasks", "description": ( "ОБЯЗАТЕЛЬНО при вопросах «какие задачи», «покажи задачи проекта», «что открыто в Taiga». " "Живые user stories и tasks из Taiga API. НЕ путать с list_work_items." ), "parameters": { "type": "object", "properties": { "project_slug": { "type": "string", "description": "Slug проекта, например home_assistant. Пусто = все проекты.", }, "limit": {"type": "integer", "description": "Макс. на проект, по умолчанию 20"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "create_work_item", "description": ( "Создать фичу/баг из вольного текста: структурировать через LLM, " "создать Taiga story + Gitea issue. Вызывай при «заведи баг», «оформи фичу», «добавь в таигу»." ), "parameters": { "type": "object", "properties": { "text": {"type": "string", "description": "Полное описание от пользователя"}, "project_slug": { "type": "string", "description": "Slug проекта Taiga, если известен", }, }, "required": ["text"], }, }, }, { "type": "function", "function": { "name": "remember_fact", "description": ( "Сохранить факт в долгосрочную память. " "Когда пользователь просит «запомни», или сообщает устойчивое предпочтение/факт." ), "parameters": { "type": "object", "properties": { "content": {"type": "string", "description": "Что запомнить"}, "category": { "type": "string", "description": "preference, person, habit, project, fact", }, "importance": {"type": "integer", "description": "1-5, по умолчанию 3"}, }, "required": ["content"], }, }, }, { "type": "function", "function": { "name": "recall_memories", "description": ( "Поиск в долгосрочной памяти. " "Когда спрашивают «что ты помнишь», «что я говорил про X»." ), "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Подстрока для поиска"}, "category": {"type": "string"}, "limit": {"type": "integer"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "forget_memory", "description": "Удалить (деактивировать) факт по id из recall_memories или снимка памяти.", "parameters": { "type": "object", "properties": { "memory_id": {"type": "integer"}, }, "required": ["memory_id"], }, }, }, { "type": "function", "function": { "name": "update_profile", "description": ( "Обновить профиль пользователя: name, timezone, language, notes. " "Передавай только изменившиеся поля." ), "parameters": { "type": "object", "properties": { "name": {"type": "string"}, "timezone": {"type": "string"}, "language": {"type": "string"}, "notes": {"type": "string"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "update_session_summary", "description": ( "Сохранить краткую сводку темы текущего чата " "(когда диалог длинный или пользователь просит «сожми контекст»)." ), "parameters": { "type": "object", "properties": { "summary": {"type": "string", "description": "2-5 предложений о теме чата"}, "session_id": {"type": "integer"}, }, "required": ["summary", "session_id"], }, }, }, { "type": "function", "function": { "name": "list_work_items", "description": ( "Только задачи, созданные ЭТИМ ассистентом через create_work_item (локальная БД). " "НЕ использовать для общего вопроса «какие задачи в Taiga» — для того list_taiga_tasks." ), "parameters": { "type": "object", "properties": { "status": {"type": "string", "description": "open или closed"}, "limit": {"type": "integer"}, }, "required": [], }, }, }, ] async def execute_tool( db: Session, name: str, arguments: dict[str, Any], *, session_id: int | None = None, ) -> str: pomodoro = PomodoroService(db) projects = ProjectService(db) memory = MemoryService(db) try: if name == "get_pomodoro_status": result = pomodoro.get_status() elif name == "start_pomodoro": result = pomodoro.start_work( duration_min=arguments.get("duration_min"), task_note=arguments.get("task_note", ""), ) elif name == "start_short_break": result = pomodoro.start_short_break(duration_min=arguments.get("duration_min")) elif name == "start_long_break": result = pomodoro.start_long_break(duration_min=arguments.get("duration_min")) elif name == "stop_pomodoro": result = pomodoro.stop( result=arguments.get("result", ""), completed=arguments.get("completed", False), ) elif name == "skip_pomodoro_phase": result = pomodoro.skip_phase() elif name == "reset_pomodoro_cycle": result = pomodoro.reset_cycle(clear_task=arguments.get("clear_task", False)) elif name == "get_pomodoro_history": result = pomodoro.history(limit=arguments.get("limit", 10)) elif name == "sync_taiga_projects": result = projects.sync_taiga_projects() elif name == "list_taiga_projects": result = projects.list_projects() elif name == "list_taiga_tasks": result = projects.list_taiga_open_tasks( project_slug=arguments.get("project_slug"), limit=arguments.get("limit", 20), ) elif name == "create_work_item": result = await projects.create_work_item( arguments.get("text", ""), project_slug=arguments.get("project_slug"), ) elif name == "list_work_items": result = projects.list_work_items( limit=arguments.get("limit", 20), status=arguments.get("status"), ) elif name == "remember_fact": result = memory.remember_fact( arguments.get("content", ""), category=arguments.get("category", "fact"), importance=arguments.get("importance", 3), session_id=session_id, source="tool", ) elif name == "recall_memories": result = memory.recall_memories( query=arguments.get("query"), category=arguments.get("category"), limit=arguments.get("limit", 20), ) elif name == "forget_memory": result = memory.forget_memory(int(arguments["memory_id"])) elif name == "update_profile": updates = { k: arguments[k] for k in ("name", "timezone", "language", "notes") if k in arguments and arguments[k] is not None } result = memory.update_profile(updates) elif name == "update_session_summary": result = memory.update_session_summary( int(arguments["session_id"]), arguments.get("summary", ""), ) else: return json.dumps({"error": f"Unknown tool: {name}"}, ensure_ascii=False) return json.dumps(result, ensure_ascii=False) except ValueError as exc: return json.dumps({"error": str(exc)}, ensure_ascii=False) except Exception as exc: return json.dumps({"error": str(exc)}, ensure_ascii=False)