Grigo/ProjectAgent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
Grigo
2026-05-18 20:35:01 +00:00
commit 5466d14040
17 changed files with 940 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config.yml
+36
View File
@@ -0,0 +1,36 @@
projects:
AISHub:
taiga_project_id: 5
repo_url: ssh://git@host.docker.internal:222/Grigo/AISHub.git
repo_path: /repos/AISHub
default_branch: main
AIsMas-Web-Service:
taiga_project_id: 4
repo_url: ssh://git@host.docker.internal:222/Grigo/WebAisMap.git
repo_path: /repos/AIsMas-Web-Service
default_branch: main
AndroidAisMap:
taiga_project_id: 3
repo_url: ssh://git@host.docker.internal:222/Grigo/AndroidAisMap.git
repo_path: /repos/AndroidAisMap
default_branch: main
PrivateTest:
taiga_project_id: 6
repo_url: ssh://git@host.docker.internal:222/Grigo/PrivateTest.git
repo_path: /repos/PrivateTest
default_branch: main
Testing:
taiga_project_id: 1
repo_url: ssh://git@host.docker.internal:222/Grigo/Testing.git
repo_path: /repos/Testing
default_branch: main
ClawSetUp:
taiga_project_id: 8
repo_url:
repo_path:
default_branch:
app/project_agent/__init__.py
View File
app/project_agent/__pycache__/__init__.cpython-312.pyc
BIN
View File
Binary file not shown.
app/project_agent/__pycache__/server.cpython-312.pyc
BIN
View File
Binary file not shown.
app/project_agent/server.py
+773
View File
@@ -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А-Яа-я0-9_/-]{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,
}