@@ -0,0 +1,773 @@
from fastapi import FastAPI , HTTPException
from pydantic import BaseModel
import json
import os
import re
import subprocess
import requests
import yaml
app = FastAPI ( title = " Project Agent " )
CONFIG_PATH = " /app/config.yml "
class SyncRequest ( BaseModel ) :
project : str | None = None
class TaskFromCodeRequest ( BaseModel ) :
project : str
text : str
def load_config ( ) :
with open ( CONFIG_PATH , " r " , encoding = " utf-8 " ) as f :
return yaml . safe_load ( f ) or { }
def run ( cmd : list [ str ] , cwd : str | None = None , timeout : int = 120 ) :
result = subprocess . run (
cmd ,
cwd = cwd ,
text = True ,
capture_output = True ,
timeout = timeout ,
)
return {
" cmd " : cmd ,
" returncode " : result . returncode ,
" stdout " : result . stdout ,
" stderr " : result . stderr ,
}
def strip_markdown_json ( text : str ) - > str :
text = text . strip ( )
fenced = re . search ( r " ```(?:json)? \ s*(.*?) \ s*``` " , text , re . DOTALL | re . IGNORECASE )
if fenced :
return fenced . group ( 1 ) . strip ( )
return text
def sync_one_project ( project : dict ) :
repo_url = project . get ( " repo_url " )
repo_path = project . get ( " repo_path " )
if not repo_url :
return { " skipped " : True , " reason " : " repo_url is empty " }
if not repo_path :
return { " skipped " : True , " reason " : " repo_path is empty " }
if os . path . exists ( repo_path ) :
fetch = run ( [ " git " , " fetch " , " --all " , " --prune " ] , cwd = repo_path )
pull = run ( [ " git " , " pull " , " --ff-only " ] , cwd = repo_path )
return {
" action " : " pull " ,
" fetch " : fetch ,
" pull " : pull ,
}
os . makedirs ( os . path . dirname ( repo_path ) , exist_ok = True )
clone = run ( [ " git " , " clone " , repo_url , repo_path ] )
return {
" action " : " clone " ,
" clone " : clone ,
}
def read_file_safe ( path : str , max_chars : int = 12000 ) - > str :
try :
if not os . path . isfile ( path ) :
return " "
if os . path . getsize ( path ) > 512 * 1024 :
return " "
with open ( path , " r " , encoding = " utf-8 " , errors = " ignore " ) as f :
return f . read ( max_chars )
except Exception :
return " "
def collect_repo_context ( repo_path : str , task_text : str ) - > str :
parts = [ ]
tree = run (
[
" bash " ,
" -lc " ,
" find . -maxdepth 4 "
" -not -path ' ./.git/* ' "
" -not -path ' ./node_modules/* ' "
" -not -path ' ./vendor/* ' "
" -not -path ' ./dist/* ' "
" -not -path ' ./build/* ' "
" -not -path ' ./target/* ' "
" | sort | head -300 " ,
] ,
cwd = repo_path ,
timeout = 30 ,
)
parts . append ( " ## File tree \n " + tree [ " stdout " ] )
git_log = run (
[ " git " , " log " , " --oneline " , " -15 " ] ,
cwd = repo_path ,
timeout = 30 ,
)
parts . append ( " ## Recent commits \n " + git_log [ " stdout " ] )
candidate_files = [
" README.md " ,
" readme.md " ,
" package.json " ,
" pyproject.toml " ,
" requirements.txt " ,
" Dockerfile " ,
" docker-compose.yml " ,
" compose.yml " ,
" pom.xml " ,
" build.gradle " ,
" settings.gradle " ,
" go.mod " ,
" Cargo.toml " ,
" .gitignore " ,
]
for rel in candidate_files :
content = read_file_safe ( os . path . join ( repo_path , rel ) , max_chars = 10000 )
if content :
parts . append ( f " ## { rel } \n { content } " )
words = [ ]
for word in re . findall ( r " [A-Za-zА { 4,} " , task_text ) :
word = word . strip ( ) . lower ( )
if word not in words :
words . append ( word )
words = words [ : 8 ]
if words :
pattern = " | " . join ( re . escape ( w ) for w in words )
grep = run (
[
" bash " ,
" -lc " ,
f " rg -n -i --glob ' !node_modules ' --glob ' !vendor ' --glob ' !dist ' --glob ' !build ' --glob ' !target ' \" { pattern } \" . | head -80 " ,
] ,
cwd = repo_path ,
timeout = 30 ,
)
if grep [ " stdout " ] :
parts . append ( " ## Relevant grep matches \n " + grep [ " stdout " ] )
context = " \n \n " . join ( parts )
return context [ : 50000 ]
def aimtr_make_task ( raw_text : str , repo_context : str ) - > dict :
base_url = os . getenv ( " AIMTR_BASE_URL " , " " ) . rstrip ( " / " )
api_key = os . getenv ( " AIMTR_API_KEY " )
model = os . getenv ( " AIMTR_MODEL " , " claude-haiku-4.5 " )
if not base_url or not api_key :
raise HTTPException ( status_code = 500 , detail = " AIMTR_BASE_URL or AIMTR_API_KEY is missing " )
system_prompt = """
Ты технический ассистент и тимлид.
Тебе дают описание задачи и краткий контекст репозитория.
Сформируй задачу для Taiga с учетом структуры кода.
Отвечай только валидным JSON без markdown.
Схема:
{
" title " : " короткое название " ,
" description " : " описание с учетом контекста кода " ,
" type " : " Story " ,
" priority " : " low|normal|high " ,
" tags " : [ " tag1 " , " tag2 " ],
" acceptance_criteria " : [
" проверяемый критерий "
],
" children " : [
{
" title " : " техническая подзадача " ,
" description " : " что сделать и где примерно смотреть в коде " ,
" type " : " Task " ,
" priority " : " low|normal|high "
}
],
" questions " : [
" уточняющий вопрос, если данных не хватает "
],
" code_notes " : [
" заметка по найденному контексту кода "
]
}
Правила:
- Пиши на русском.
- Не выдумывай файлы, если их нет в контексте.
- Если предполагаешь файл или модуль, явно пиши ' вероятно ' или ' проверить ' .
- Acceptance criteria должны быть проверяемыми.
- Подзадачи должны быть полезны разработчику.
""" . strip ( )
user_prompt = f """
Описание задачи:
{ raw_text }
Контекст репозитория:
{ repo_context }
""" . strip ( )
response = requests . post (
f " { base_url } /chat/completions " ,
headers = {
" Authorization " : f " Bearer { api_key } " ,
" Content-Type " : " application/json " ,
} ,
json = {
" model " : model ,
" messages " : [
{ " role " : " system " , " content " : system_prompt } ,
{ " role " : " user " , " content " : user_prompt } ,
] ,
" temperature " : 0.2 ,
} ,
timeout = 90 ,
)
response . raise_for_status ( )
content = response . json ( ) [ " choices " ] [ 0 ] [ " message " ] [ " content " ]
clean = strip_markdown_json ( content )
try :
return json . loads ( clean )
except json . JSONDecodeError as exc :
raise HTTPException ( status_code = 500 , detail = f " LLM returned invalid JSON: { content [ : 1000 ] } " ) from exc
def taiga_auth ( ) - > str :
base_url = os . getenv ( " TAIGA_BASE_URL " , " " ) . rstrip ( " / " )
username = os . getenv ( " TAIGA_USERNAME " )
password = os . getenv ( " TAIGA_PASSWORD " )
if not base_url or not username or not password :
raise HTTPException ( status_code = 500 , detail = " Taiga env vars are missing " )
response = requests . post (
f " { base_url } /api/v1/auth " ,
json = {
" type " : " normal " ,
" username " : username ,
" password " : password ,
} ,
timeout = 20 ,
)
response . raise_for_status ( )
return response . json ( ) [ " auth_token " ]
def taiga_headers ( token : str ) :
return {
" Authorization " : f " Bearer { token } " ,
" Content-Type " : " application/json " ,
}
def format_story_description ( task : dict , raw_text : str ) - > str :
lines = [ ]
if task . get ( " description " ) :
lines . append ( task [ " description " ] )
else :
lines . append ( raw_text )
if task . get ( " code_notes " ) :
lines . append ( " " )
lines . append ( " ## Заметки по коду " )
for item in task [ " code_notes " ] :
lines . append ( f " - { item } " )
if task . get ( " acceptance_criteria " ) :
lines . append ( " " )
lines . append ( " ## Acceptance criteria " )
for item in task [ " acceptance_criteria " ] :
lines . append ( f " - { item } " )
if task . get ( " questions " ) :
lines . append ( " " )
lines . append ( " ## Вопросы / уточнения " )
for item in task [ " questions " ] :
lines . append ( f " - { item } " )
if task . get ( " tags " ) :
lines . append ( " " )
lines . append ( " ## Теги " )
lines . append ( " , " . join ( task [ " tags " ] ) )
lines . append ( " " )
lines . append ( " ## Исходное описание " )
lines . append ( raw_text )
return " \n " . join ( lines ) . strip ( )
def create_userstory ( token : str , project_id : int , task : dict , raw_text : str ) - > dict :
base_url = os . getenv ( " TAIGA_BASE_URL " , " " ) . rstrip ( " / " )
response = requests . post (
f " { base_url } /api/v1/userstories " ,
headers = taiga_headers ( token ) ,
json = {
" project " : project_id ,
" subject " : ( task . get ( " title " ) or raw_text ) [ : 500 ] ,
" description " : format_story_description ( task , raw_text ) ,
} ,
timeout = 30 ,
)
response . raise_for_status ( )
return response . json ( )
def create_subtask ( token : str , project_id : int , userstory_id : int , child : dict ) - > dict :
base_url = os . getenv ( " TAIGA_BASE_URL " , " " ) . rstrip ( " / " )
response = requests . post (
f " { base_url } /api/v1/tasks " ,
headers = taiga_headers ( token ) ,
json = {
" project " : project_id ,
" user_story " : userstory_id ,
" subject " : ( child . get ( " title " ) or " Подзадача " ) [ : 500 ] ,
" description " : child . get ( " description " ) or " " ,
} ,
timeout = 30 ,
)
response . raise_for_status ( )
return response . json ( )
@app.get ( " /health " )
def health ( ) :
return {
" ok " : True ,
" repos_dir " : os . getenv ( " AGENT_REPOS_DIR " , " /repos " ) ,
" state_dir " : os . getenv ( " AGENT_STATE_DIR " , " /state " ) ,
}
@app.get ( " /projects " )
def projects ( ) :
config = load_config ( )
return config . get ( " projects " , { } )
@app.post ( " /repos/sync " )
def sync_repos ( req : SyncRequest ) :
config = load_config ( )
projects = config . get ( " projects " , { } )
if req . project :
if req . project not in projects :
raise HTTPException ( status_code = 404 , detail = f " Unknown project: { req . project } " )
selected = { req . project : projects [ req . project ] }
else :
selected = projects
results = { }
for name , project in selected . items ( ) :
results [ name ] = sync_one_project ( project )
return results
@app.post ( " /tasks/from-code " )
def task_from_code ( req : TaskFromCodeRequest ) :
config = load_config ( )
projects = config . get ( " projects " , { } )
if req . project not in projects :
raise HTTPException ( status_code = 404 , detail = f " Unknown project: { req . project } " )
project = projects [ req . project ]
taiga_project_id = project . get ( " taiga_project_id " )
repo_path = project . get ( " repo_path " )
if not taiga_project_id :
raise HTTPException ( status_code = 400 , detail = " taiga_project_id is missing " )
sync_result = sync_one_project ( project )
if not repo_path or not os . path . exists ( repo_path ) :
raise HTTPException ( status_code = 400 , detail = " Repo path does not exist after sync " )
repo_context = collect_repo_context ( repo_path , req . text )
structured = aimtr_make_task ( req . text , repo_context )
token = taiga_auth ( )
story = create_userstory ( token , int ( taiga_project_id ) , structured , req . text )
subtasks = [ ]
for child in structured . get ( " children " , [ ] ) :
if isinstance ( child , dict ) :
subtasks . append ( create_subtask ( token , int ( taiga_project_id ) , story [ " id " ] , child ) )
return {
" project " : req . project ,
" sync " : sync_result ,
" story " : {
" id " : story [ " id " ] ,
" ref " : story [ " ref " ] ,
" subject " : story [ " subject " ] ,
} ,
" subtasks " : [
{
" id " : task . get ( " id " ) ,
" ref " : task . get ( " ref " ) ,
" subject " : task . get ( " subject " ) ,
}
for task in subtasks
] ,
" structured " : structured ,
}
class NextActionRequest ( BaseModel ) :
project : str
minutes : int = 25
energy : str = " normal "
notes : str | None = None
def compact_userstory ( us : dict ) - > dict :
return {
" id " : us . get ( " id " ) ,
" ref " : us . get ( " ref " ) ,
" subject " : us . get ( " subject " ) ,
" status " : ( us . get ( " status_extra_info " ) or { } ) . get ( " name " ) ,
" is_closed " : us . get ( " is_closed " ) ,
" assigned_to " : ( us . get ( " assigned_to_extra_info " ) or { } ) . get ( " username " ) ,
" created_date " : us . get ( " created_date " ) ,
" modified_date " : us . get ( " modified_date " ) ,
" description " : ( us . get ( " description " ) or " " ) [ : 1500 ] ,
}
def compact_task ( task : dict ) - > dict :
return {
" id " : task . get ( " id " ) ,
" ref " : task . get ( " ref " ) ,
" subject " : task . get ( " subject " ) ,
" status " : ( task . get ( " status_extra_info " ) or { } ) . get ( " name " ) ,
" is_closed " : task . get ( " is_closed " ) ,
" assigned_to " : ( task . get ( " assigned_to_extra_info " ) or { } ) . get ( " username " ) ,
" user_story " : task . get ( " user_story " ) ,
" created_date " : task . get ( " created_date " ) ,
" modified_date " : task . get ( " modified_date " ) ,
" description " : ( task . get ( " description " ) or " " ) [ : 1200 ] ,
}
def taiga_get_backlog ( token : str , project_id : int ) - > dict :
base_url = os . getenv ( " TAIGA_BASE_URL " , " " ) . rstrip ( " / " )
stories_resp = requests . get (
f " { base_url } /api/v1/userstories " ,
headers = taiga_headers ( token ) ,
params = {
" project " : project_id ,
" order_by " : " -modified_date " ,
} ,
timeout = 30 ,
)
stories_resp . raise_for_status ( )
tasks_resp = requests . get (
f " { base_url } /api/v1/tasks " ,
headers = taiga_headers ( token ) ,
params = {
" project " : project_id ,
" order_by " : " -modified_date " ,
} ,
timeout = 30 ,
)
tasks_resp . raise_for_status ( )
stories = stories_resp . json ( )
tasks = tasks_resp . json ( )
open_stories = [ compact_userstory ( x ) for x in stories if not x . get ( " is_closed " ) ]
open_tasks = [ compact_task ( x ) for x in tasks if not x . get ( " is_closed " ) ]
return {
" userstories " : open_stories [ : 50 ] ,
" tasks " : open_tasks [ : 80 ] ,
}
def aimtr_next_action ( project_name : str , minutes : int , energy : str , notes : str | None , backlog : dict ) - > dict :
base_url = os . getenv ( " AIMTR_BASE_URL " , " " ) . rstrip ( " / " )
api_key = os . getenv ( " AIMTR_API_KEY " )
model = os . getenv ( " AIMTR_MODEL " , " claude-haiku-4.5 " )
if not base_url or not api_key :
raise HTTPException ( status_code = 500 , detail = " AIMTR_BASE_URL or AIMTR_API_KEY is missing " )
system_prompt = """
Ты техлид и персональный фокус-ассистент.
Твоя задача — выбрать ОДНО лучшее действие на ближайший pomodoro.
Отвечай только валидным JSON без markdown.
Схема:
{
" recommended " : {
" kind " : " task|story|meta " ,
" ref " : 123,
" title " : " что делать " ,
" why_now " : " почему именно это сейчас " ,
" expected_result " : " что должно быть готово к концу pomodoro " ,
" risk " : " главный риск или блокер "
},
" pomodoro_plan " : [
" шаг 1 " ,
" шаг 2 " ,
" шаг 3 "
],
" definition_of_done " : [
" критерий готовности 1 " ,
" критерий готовности 2 "
],
" skip_reasoning " : [
" почему не выбраны другие важные задачи "
],
" questions " : [
" короткий вопрос, если без него нельзя начать "
]
}
Правила:
- Выбирай задачу, которую реально сдвинуть за заданное количество минут.
- Предпочитай маленький понятный next step, а не огромную важную задачу.
- Если есть недавно созданные подзадачи без прогресса — они хорошие кандидаты.
- Не выбирай закрытые задачи.
- Если данных мало, всё равно предложи лучший следующий шаг.
- Пиши на русском.
""" . strip ( )
payload = {
" project " : project_name ,
" minutes " : minutes ,
" energy " : energy ,
" notes " : notes ,
" backlog " : backlog ,
}
response = requests . post (
f " { base_url } /chat/completions " ,
headers = {
" Authorization " : f " Bearer { api_key } " ,
" Content-Type " : " application/json " ,
} ,
json = {
" model " : model ,
" messages " : [
{ " role " : " system " , " content " : system_prompt } ,
{
" role " : " user " ,
" content " : json . dumps ( payload , ensure_ascii = False ) ,
} ,
] ,
" temperature " : 0.2 ,
} ,
timeout = 90 ,
)
response . raise_for_status ( )
content = response . json ( ) [ " choices " ] [ 0 ] [ " message " ] [ " content " ]
clean = strip_markdown_json ( content )
try :
return json . loads ( clean )
except json . JSONDecodeError as exc :
raise HTTPException ( status_code = 500 , detail = f " LLM returned invalid JSON: { content [ : 1000 ] } " ) from exc
@app.post ( " /next-action " )
def next_action ( req : NextActionRequest ) :
config = load_config ( )
projects = config . get ( " projects " , { } )
if req . project not in projects :
raise HTTPException ( status_code = 404 , detail = f " Unknown project: { req . project } " )
project = projects [ req . project ]
taiga_project_id = project . get ( " taiga_project_id " )
if not taiga_project_id :
raise HTTPException ( status_code = 400 , detail = " taiga_project_id is missing " )
token = taiga_auth ( )
backlog = taiga_get_backlog ( token , int ( taiga_project_id ) )
recommendation = aimtr_next_action (
project_name = req . project ,
minutes = req . minutes ,
energy = req . energy ,
notes = req . notes ,
backlog = backlog ,
)
return {
" project " : req . project ,
" minutes " : req . minutes ,
" energy " : req . energy ,
" backlog_counts " : {
" userstories " : len ( backlog [ " userstories " ] ) ,
" tasks " : len ( backlog [ " tasks " ] ) ,
} ,
" recommendation " : recommendation ,
}
class PomodoroFinishRequest ( BaseModel ) :
project : str
task_ref : int
minutes : int = 25
result : str
done : bool = False
notes : str | None = None
def taiga_find_task_by_ref ( token : str , project_id : int , task_ref : int ) - > dict :
base_url = os . getenv ( " TAIGA_BASE_URL " , " " ) . rstrip ( " / " )
response = requests . get (
f " { base_url } /api/v1/tasks/by_ref " ,
headers = taiga_headers ( token ) ,
params = {
" project " : project_id ,
" ref " : task_ref ,
} ,
timeout = 30 ,
)
response . raise_for_status ( )
return response . json ( )
def taiga_get_closed_task_status ( token : str , project_id : int ) - > int | None :
base_url = os . getenv ( " TAIGA_BASE_URL " , " " ) . rstrip ( " / " )
response = requests . get (
f " { base_url } /api/v1/task-statuses " ,
headers = taiga_headers ( token ) ,
params = { " project " : project_id } ,
timeout = 30 ,
)
response . raise_for_status ( )
statuses = response . json ( )
for status in statuses :
if status . get ( " is_closed " ) :
return status . get ( " id " )
return None
def format_pomodoro_comment ( req : PomodoroFinishRequest ) - > str :
lines = [
f " 🍅 Pomodoro report: { req . minutes } min " ,
" " ,
" Result: " ,
req . result . strip ( ) ,
]
if req . notes :
lines . extend ( [ " " , " Notes: " , req . notes . strip ( ) ] )
lines . extend ( [
" " ,
f " Marked done: { ' yes ' if req . done else ' no ' } " ,
] )
return " \n " . join ( lines )
def taiga_update_task_comment (
token : str ,
task : dict ,
comment : str ,
close : bool ,
closed_status_id : int | None ,
) - > dict :
base_url = os . getenv ( " TAIGA_BASE_URL " , " " ) . rstrip ( " / " )
payload = {
" version " : task . get ( " version " ) ,
" comment " : comment ,
}
if close and closed_status_id :
payload [ " status " ] = closed_status_id
response = requests . patch (
f " { base_url } /api/v1/tasks/ { task [ ' id ' ] } " ,
headers = taiga_headers ( token ) ,
json = payload ,
timeout = 30 ,
)
response . raise_for_status ( )
return response . json ( )
@app.post ( " /pomodoro/finish " )
def pomodoro_finish ( req : PomodoroFinishRequest ) :
config = load_config ( )
projects = config . get ( " projects " , { } )
if req . project not in projects :
raise HTTPException ( status_code = 404 , detail = f " Unknown project: { req . project } " )
project = projects [ req . project ]
taiga_project_id = project . get ( " taiga_project_id " )
if not taiga_project_id :
raise HTTPException ( status_code = 400 , detail = " taiga_project_id is missing " )
token = taiga_auth ( )
task = taiga_find_task_by_ref ( token , int ( taiga_project_id ) , req . task_ref )
comment = format_pomodoro_comment ( req )
closed_status_id = None
if req . done :
closed_status_id = taiga_get_closed_task_status ( token , int ( taiga_project_id ) )
updated_task = taiga_update_task_comment (
token = token ,
task = task ,
comment = comment ,
close = req . done ,
closed_status_id = closed_status_id ,
)
backlog = taiga_get_backlog ( token , int ( taiga_project_id ) )
recommendation = aimtr_next_action (
project_name = req . project ,
minutes = req . minutes ,
energy = " normal " ,
notes = f " Just finished task # { req . task_ref } . Result: { req . result } " ,
backlog = backlog ,
)
return {
" project " : req . project ,
" task " : {
" id " : updated_task . get ( " id " ) ,
" ref " : updated_task . get ( " ref " ) ,
" subject " : updated_task . get ( " subject " ) ,
" status " : ( updated_task . get ( " status_extra_info " ) or { } ) . get ( " name " ) ,
" is_closed " : updated_task . get ( " is_closed " ) ,
} ,
" comment_added " : True ,
" closed " : bool ( req . done and closed_status_id ) ,
" next_recommendation " : recommendation ,
}
