diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e0e7c0f..23d20a8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -67,11 +67,6 @@ jobs: push: false tags: ${{ env.IMAGE }} - - name: Verify MCP OAuth runtime - run: >- - docker run --rm "$IMAGE" python -c - 'from mcp.client.auth.oauth2 import OAuthClientProvider' - - name: Smoke-test image shell: bash run: | @@ -125,11 +120,6 @@ jobs: push: false tags: ${{ env.IMAGE }} - - name: Verify MCP OAuth runtime - run: >- - docker run --rm "$IMAGE" python -c - 'from mcp.client.auth.oauth2 import OAuthClientProvider' - - name: Smoke-test image shell: bash run: | diff --git a/Dockerfile b/Dockerfile index 2ea1325..3761a76 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,8 +3,6 @@ FROM python:3.11.15-slim@sha256:db3ff2e1800a8581e2c48a27c3995339d47bdf046da21c76 ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ HOME=/home/cowork \ - HERMES_HOME=/data/hermes \ - HERMES_PYTHON=/usr/local/bin/python \ COWORK_DATABASE_PATH=/data/cowork.db WORKDIR /app @@ -17,7 +15,7 @@ COPY pyproject.toml README.md ./ COPY app ./app RUN python -m pip install --no-cache-dir --disable-pip-version-check \ - "hermes-agent[mcp]==0.19.0" . + . USER 1000:1000 diff --git a/README.md b/README.md index 5c89b55..bf89f30 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # 할 일 올리기 -팀원이 자연어로 할 일만 적으면 서버 agent가 Jira MCP를 통해 `S15P11A705` 프로젝트에 Task를 생성하는 단일 목적 프로토타입입니다. +팀원이 자연어로 할 일을 적으면 SSAFY GMS AI API로 Task 초안을 만들고 Jira REST API를 통해 `S15P11A705` 프로젝트에 Task를 생성하는 서비스입니다. ## 범위 @@ -17,8 +17,9 @@ - FastAPI + vanilla HTML/CSS/JS - private SQLite: 사용자, 세션, 생성 요청 원문, 생성 결과 -- Hermes planner: 자연어를 평평한 Task 목록으로 정리 -- direct Atlassian MCP bridge: 허용된 `createJiraIssue` 하나만 호출 +- SSAFY GMS Anthropic Messages API: 자연어를 평평한 Task 목록으로 정리 +- Jira REST API: 활성 Sprint 조회와 확인된 Task 생성 +- 제한된 AI/Jira worker pool: 여러 팀원의 요청을 병렬 처리하고 초과 요청은 대기 - Mattermost Incoming Webhook: Jira/agent 실패 원문과 입력 원문을 리드에게 전달 화면은 서버의 과거 생성 내역을 조회하지 않습니다. SQLite 기록은 입력 유실 방지와 장애 조사만을 위한 내부 데이터입니다. @@ -32,11 +33,11 @@ cp .env.example .env chmod 600 .env ``` -`.env`에서 실제 `MATTERMOST_WEBHOOK_URL`을 설정합니다. production에서는 Mattermost webhook URL과 secure cookie가 없으면 앱이 시작되지 않습니다. Jira 인증은 `cowork` 전용 OS 계정의 `atlassian` Hermes MCP OAuth를 사용합니다. root의 Hermes profile을 서비스에 노출하지 않습니다. +`.env`에서 `GMS_KEY`, `JIRA_EMAIL`, `JIRA_API_TOKEN`, `MATTERMOST_WEBHOOK_URL`을 설정합니다. production에서는 이 값들과 secure cookie가 없으면 앱이 시작되지 않습니다. API 키와 Token은 저장소나 이미지에 포함하지 않고 Kubernetes Secret으로 주입합니다. ## 사용자 등록 -비밀번호와 Jira accountId는 repository나 shell argument에 넣지 않습니다. Jira accountId는 표시 이름으로 MCP에서 정확히 조회하고, 비밀번호는 TTY에서 두 번 입력합니다. +비밀번호와 Jira accountId는 repository나 shell argument에 넣지 않습니다. Jira accountId는 Jira에서 정확히 확인하고, 비밀번호는 TTY에서 두 번 입력합니다. ```bash set -a; . ./.env; set +a @@ -58,8 +59,8 @@ TLS reverse proxy 뒤에서 운영합니다. `COWORK_COOKIE_SECURE=true`를 유 ## 컨테이너와 k3s 배포 -`Dockerfile`은 Python과 Hermes Agent 버전을 고정하고 UID/GID 1000의 non-root -사용자로 실행합니다. SQLite와 Hermes profile/OAuth 상태는 `/data` 아래에 두며, +`Dockerfile`은 Python 베이스 이미지를 digest로 고정하고 UID/GID 1000의 non-root +사용자로 실행합니다. SQLite 상태는 `/data` 아래에 두며, 운영에서는 하나의 영속 PVC를 마운트합니다. image tag는 Git commit SHA만 사용하고 `main` CI가 검증한 image만 GHCR에 게시합니다. diff --git a/SPEC.md b/SPEC.md index 2b22c8f..138f9e7 100644 --- a/SPEC.md +++ b/SPEC.md @@ -2,7 +2,7 @@ ## 목적 -Jira에 익숙하지 않은 팀원이 자연어로 할 일만 적으면 서버 agent가 Jira MCP로 티켓을 생성한다. 이 서비스는 Jira를 대체하지 않고 티켓 생성만 전담한다. +Jira에 익숙하지 않은 팀원이 자연어로 할 일만 적으면 SSAFY GMS AI API가 초안을 만들고 서버가 Jira REST API로 티켓을 생성한다. 이 서비스는 Jira를 대체하지 않고 티켓 생성만 전담한다. ## 확정 결정 @@ -12,7 +12,7 @@ Jira에 익숙하지 않은 팀원이 자연어로 할 일만 적으면 서버 a - Jira 프로젝트: `S15P11A705` - 타입: Task - 담당자: 로그인 사용자의 필수 `jira_account_id` -- 생성 주체: 서버의 현재 Atlassian MCP 인증 +- 생성 주체: Cowork 전용 Jira 이메일/API Token 인증 - 스프린트·parent·story point: 설정하지 않음 ## 사용자 흐름 diff --git a/app/agent.py b/app/agent.py index 3a39b74..2e77443 100644 --- a/app/agent.py +++ b/app/agent.py @@ -2,14 +2,9 @@ import json import re -import subprocess from dataclasses import dataclass from typing import Any -from urllib.parse import urlparse - -from .config import Settings -from .roles import ticket_summary, validate_role_tag class AgentError(RuntimeError): @@ -28,28 +23,6 @@ class AgentReconciliationRequired(AgentError): pass -HERMES_QUERY_BOOTSTRAP = """ -import sys -from hermes_cli.main import main - -turns = sys.argv[1] -prompt = sys.stdin.read() -sys.argv = [ - "hermes", - "chat", - "-q", - prompt, - "--safe-mode", - "-Q", - "--source", - "tool", - "--ignore-rules", - "--max-turns", - turns, -] -raise SystemExit(main()) -""" - def _looks_like_network_error(text: str) -> bool: lowered = text.lower() @@ -158,37 +131,9 @@ def _extract_json(output: str) -> dict[str, Any]: raise AgentError("agent returned no JSON object") return max(candidates, key=lambda item: item[0])[1] - -class HermesJiraAgent: - def __init__(self, settings: Settings): - self.settings = settings - +class PlanningAgent: def _run(self, prompt: str, *, max_turns: int) -> dict[str, Any]: - command = [ - self.settings.hermes_python, - "-c", - HERMES_QUERY_BOOTSTRAP, - str(max_turns), - ] - try: - completed = subprocess.run( - command, - input=" ".join(prompt.splitlines()) + "\n", - capture_output=True, - text=True, - timeout=self.settings.agent_timeout_seconds, - check=False, - ) - except subprocess.TimeoutExpired as exc: - raise AgentTimeout("agent subprocess timed out") from exc - except OSError as exc: - raise AgentError("agent subprocess unavailable") from exc - if completed.returncode != 0: - detail = _bounded_error_detail(completed.stderr) - if _looks_like_network_error(detail): - raise AgentNetworkError(f"planner network failure: {detail}") - raise AgentError(f"agent subprocess failed: {detail}") - return _extract_json(completed.stdout) + raise NotImplementedError def plan(self, raw_input: str) -> Plan: source_lines = [line.strip() for line in raw_input.splitlines() if line.strip()] @@ -235,105 +180,3 @@ def plan(self, raw_input: str) -> Plan: ) corrected_payload = self._run(correction_prompt, max_turns=2) return _parse_plan_payload(corrected_payload, source_lines) - - def list_active_sprints(self) -> list[ActiveSprint]: - request = { - "cloudId": self.settings.jira_base_url, - "projectKey": self.settings.project_key, - } - try: - completed = subprocess.run( - [self.settings.hermes_python, "-m", "app.mcp_sprint_bridge"], - input=json.dumps(request, ensure_ascii=False), - capture_output=True, - text=True, - timeout=self.settings.agent_timeout_seconds, - check=False, - ) - except subprocess.TimeoutExpired as exc: - raise AgentTimeout("active sprint lookup timed out") from exc - except OSError as exc: - raise AgentError("Jira MCP sprint bridge unavailable") from exc - if completed.returncode != 0: - detail = _bounded_error_detail(completed.stdout or completed.stderr) - if _looks_like_network_error(detail): - raise AgentNetworkError(f"active sprint lookup failed: {detail}") - raise AgentError(f"active sprint lookup failed: {detail}") - try: - response = json.loads(completed.stdout) - raw_sprints = response["sprints"] - if set(response) != {"sprints"} or not isinstance(raw_sprints, list): - raise ValueError - sprints = [ActiveSprint(id=item["id"], name=item["name"]) for item in raw_sprints] - except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc: - raise AgentError("invalid active sprint response") from exc - if any( - not isinstance(sprint.id, int) - or sprint.id <= 0 - or not isinstance(sprint.name, str) - or not sprint.name.strip() - for sprint in sprints - ): - raise AgentError("invalid active sprint response") - return sprints - - def create_task( - self, - task: PlannedTask, - jira_account_id: str, - sprint_id: int, - role_tag: str, - ) -> CreatedTicket: - role_tag = validate_role_tag(role_tag) - request = { - "cloudId": self.settings.jira_base_url, - "projectKey": self.settings.project_key, - "issueTypeName": "Task", - "summary": ticket_summary(task.summary, role_tag), - "assignee_account_id": jira_account_id, - "contentFormat": "markdown", - "additional_fields": { - "customfield_10020": sprint_id, - "labels": [role_tag], - }, - } - if task.description is not None: - request["description"] = task.description - try: - completed = subprocess.run( - [self.settings.hermes_python, "-m", "app.mcp_create_bridge"], - input=json.dumps(request, ensure_ascii=False), - capture_output=True, - text=True, - timeout=self.settings.agent_timeout_seconds, - check=False, - ) - except subprocess.TimeoutExpired as exc: - raise AgentReconciliationRequired("Jira create timed out after dispatch") from exc - except OSError as exc: - raise AgentError("Jira MCP bridge unavailable") from exc - if completed.returncode == 42: - raise AgentReconciliationRequired("Jira create result requires reconciliation") - if completed.returncode != 0: - detail = _bounded_error_detail(completed.stdout or completed.stderr) - if _looks_like_network_error(detail): - raise AgentReconciliationRequired( - f"Jira create network result requires reconciliation: {detail}" - ) - raise AgentError(f"Jira MCP create failed: {detail}") - try: - response = json.loads(completed.stdout) - except json.JSONDecodeError as exc: - raise AgentReconciliationRequired("Jira create result requires reconciliation") from exc - if not isinstance(response, dict) or set(response) != {"issue_key"}: - raise AgentReconciliationRequired("Jira create result requires reconciliation") - issue_key = response["issue_key"] - if not isinstance(issue_key, str) or not re.fullmatch( - rf"{re.escape(self.settings.project_key)}-\d+", issue_key - ): - raise AgentReconciliationRequired("Jira create result requires reconciliation") - base = urlparse(self.settings.jira_base_url) - if base.scheme != "https" or not base.netloc: - raise AgentError("invalid configured Jira URL") - url = f"{self.settings.jira_base_url}/browse/{issue_key}" - return CreatedTicket(issue_key=issue_key, summary=task.summary, url=url) diff --git a/app/api_agent.py b/app/api_agent.py new file mode 100644 index 0000000..292f3fc --- /dev/null +++ b/app/api_agent.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import re +from typing import Any + +import httpx + +from .agent import ( + ActiveSprint, + AgentError, + AgentNetworkError, + AgentReconciliationRequired, + AgentTimeout, + CreatedTicket, + PlanningAgent, + PlannedTask, + _extract_json, +) +from .config import Settings +from .roles import ticket_summary, validate_role_tag + + +class ApiTaskAgent(PlanningAgent): + """SSAFY GMS planner and direct Jira REST adapter.""" + + def __init__(self, settings: Settings, *, http_client: httpx.Client | None = None): + self.settings = settings + self.http_client = http_client or httpx.Client(timeout=settings.agent_timeout_seconds) + + def _run(self, prompt: str, *, max_turns: int) -> dict[str, Any]: + del max_turns + if not self.settings.gms_api_key: + raise AgentError("GMS API key is not configured") + try: + response = self.http_client.post( + self.settings.gms_api_url, + headers={ + "Content-Type": "application/json", + "x-api-key": self.settings.gms_api_key, + "anthropic-version": "2023-06-01", + }, + json={ + "model": self.settings.gms_model, + "max_tokens": 2048, + "messages": [{"role": "user", "content": prompt}], + }, + ) + except httpx.TimeoutException as exc: + raise AgentTimeout("GMS request timed out") from exc + except httpx.HTTPError as exc: + raise AgentNetworkError("GMS network failure") from exc + if response.status_code == 429 or response.status_code >= 500: + raise AgentNetworkError(f"GMS temporary failure: HTTP {response.status_code}") + if response.status_code != 200: + raise AgentError(f"GMS request rejected: HTTP {response.status_code}") + try: + content = response.json()["content"] + text = "\n".join( + item["text"] + for item in content + if isinstance(item, dict) and item.get("type") == "text" + ) + except (ValueError, KeyError, TypeError) as exc: + raise AgentError("invalid GMS response") from exc + return _extract_json(text) + + def _jira_auth(self) -> httpx.BasicAuth: + if not self.settings.jira_email or not self.settings.jira_api_token: + raise AgentError("Jira API credentials are not configured") + return httpx.BasicAuth(self.settings.jira_email, self.settings.jira_api_token) + + def list_active_sprints(self) -> list[ActiveSprint]: + url = f"{self.settings.jira_base_url}/rest/agile/1.0/board/{self.settings.jira_board_id}/sprint" + try: + response = self.http_client.get( + url, params={"state": "active", "maxResults": 50}, auth=self._jira_auth() + ) + except httpx.TimeoutException as exc: + raise AgentTimeout("active sprint lookup timed out") from exc + except httpx.HTTPError as exc: + raise AgentNetworkError("active sprint lookup network failure") from exc + if response.status_code == 429 or response.status_code >= 500: + raise AgentNetworkError(f"active sprint lookup temporary failure: HTTP {response.status_code}") + if response.status_code != 200: + raise AgentError(f"active sprint lookup rejected: HTTP {response.status_code}") + try: + values = response.json()["values"] + sprints = [ActiveSprint(id=item["id"], name=item["name"]) for item in values] + except (ValueError, KeyError, TypeError) as exc: + raise AgentError("invalid active sprint response") from exc + if any(type(s.id) is not int or s.id <= 0 or not isinstance(s.name, str) or not s.name.strip() for s in sprints): + raise AgentError("invalid active sprint response") + return sprints + + def create_task(self, task: PlannedTask, jira_account_id: str, sprint_id: int, role_tag: str) -> CreatedTicket: + role_tag = validate_role_tag(role_tag) + fields: dict[str, Any] = { + "project": {"key": self.settings.project_key}, + "issuetype": {"name": "Task"}, + "summary": ticket_summary(task.summary, role_tag), + "assignee": {"accountId": jira_account_id}, + "customfield_10020": sprint_id, + "labels": [role_tag], + } + if task.description is not None: + fields["description"] = task.description + try: + response = self.http_client.post( + f"{self.settings.jira_base_url}/rest/api/2/issue", + auth=self._jira_auth(), json={"fields": fields}, + ) + except (httpx.TimeoutException, httpx.HTTPError) as exc: + raise AgentReconciliationRequired("Jira create result requires reconciliation") from exc + if response.status_code in {408, 429} or response.status_code >= 500: + raise AgentReconciliationRequired(f"Jira create result requires reconciliation: HTTP {response.status_code}") + if response.status_code not in {200, 201}: + raise AgentError(f"Jira create rejected: HTTP {response.status_code}") + try: + issue_key = response.json()["key"] + except (ValueError, KeyError, TypeError) as exc: + raise AgentReconciliationRequired("Jira create result requires reconciliation") from exc + if not isinstance(issue_key, str) or not re.fullmatch(rf"{re.escape(self.settings.project_key)}-\d+", issue_key): + raise AgentReconciliationRequired("Jira create result requires reconciliation") + return CreatedTicket(issue_key, task.summary, f"{self.settings.jira_base_url}/browse/{issue_key}") diff --git a/app/api_smoke.py b/app/api_smoke.py new file mode 100644 index 0000000..4546ba9 --- /dev/null +++ b/app/api_smoke.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import os + +import httpx + +from .agent import _extract_json, _parse_plan_payload + +GMS_URL = "https://gms.ssafy.io/gmsapi/api.anthropic.com/v1/messages" +JIRA_BASE_URL = "https://ssafy.atlassian.net" +PROJECT_KEY = "S15P11A705" +BOARD_ID = "13941" + + +def run_read_only_smoke( + *, + gms_key: str, + jira_email: str, + jira_api_token: str, + client: httpx.Client | None = None, +) -> int: + """Exercise GMS planning and Jira sprint GET without mutating Jira.""" + if not all((gms_key, jira_email, jira_api_token)): + raise RuntimeError("required API credentials are missing") + http = client or httpx.Client(timeout=30.0) + source = "CI 읽기 전용 smoke test" + gms = http.post( + GMS_URL, + headers={ + "x-api-key": gms_key, + "anthropic-version": "2023-06-01", + "content-type": "application/json", + }, + json={ + "model": "claude-opus-4-8", + "max_tokens": 1024, + "messages": [ + { + "role": "user", + "content": ( + "Return JSON only with exactly these keys: tasks and excluded. " + "Put the input into one task using source_indices [0], a concise " + "summary, and description null. excluded must be empty.\nINPUT:\n" + + source + ), + } + ], + }, + ) + if gms.status_code != 200: + raise RuntimeError(f"GMS request failed with HTTP {gms.status_code}") + try: + body = gms.json() + text = next( + item["text"] + for item in body["content"] + if isinstance(item, dict) and item.get("type") == "text" + ) + _parse_plan_payload(_extract_json(text), [source]) + except (KeyError, TypeError, ValueError, StopIteration) as exc: + raise RuntimeError("GMS returned an invalid structured response") from exc + + jira = http.get( + f"{JIRA_BASE_URL}/rest/agile/1.0/board/{BOARD_ID}/sprint", + params={"state": "active", "maxResults": "50"}, + auth=httpx.BasicAuth(jira_email, jira_api_token), + headers={"accept": "application/json"}, + ) + if jira.status_code != 200: + raise RuntimeError(f"Jira read-only request failed with HTTP {jira.status_code}") + try: + values = jira.json()["values"] + if not isinstance(values, list): + raise TypeError + except (KeyError, TypeError, ValueError) as exc: + raise RuntimeError("Jira returned an invalid sprint response") from exc + return len(values) + + +def main() -> None: + try: + count = run_read_only_smoke( + gms_key=os.environ["GMS_KEY"], + jira_email=os.environ["JIRA_EMAIL"], + jira_api_token=os.environ["JIRA_API_TOKEN"], + ) + except Exception as exc: + # Deliberately suppress external response bodies and credential-bearing request data. + raise SystemExit(f"api_smoke=failed type={type(exc).__name__}") from None + print(f"gms=ok jira_read=ok active_sprints={count}") + + +if __name__ == "__main__": + main() diff --git a/app/config.py b/app/config.py index 84478cc..8c1a2db 100644 --- a/app/config.py +++ b/app/config.py @@ -13,8 +13,14 @@ class Settings: cookie_secure: bool agent_timeout_seconds: int mattermost_webhook_url: str | None - hermes_command: str - hermes_python: str = "/usr/local/lib/hermes-agent/venv/bin/python" + gms_api_url: str = "https://gms.ssafy.io/gmsapi/api.anthropic.com/v1/messages" + gms_api_key: str | None = None + gms_model: str = "claude-opus-4-8" + jira_email: str | None = None + jira_api_token: str | None = None + jira_board_id: int = 13941 + ai_max_workers: int = 3 + jira_max_workers: int = 2 @classmethod def from_env(cls) -> "Settings": @@ -33,6 +39,13 @@ def from_env(cls) -> "Settings": raise RuntimeError("MATTERMOST_WEBHOOK_URL is required in production") if environment == "production" and not cookie_secure: raise RuntimeError("secure cookies are required in production") + gms_api_key = os.getenv("GMS_KEY") or None + jira_email = os.getenv("JIRA_EMAIL") or None + jira_api_token = os.getenv("JIRA_API_TOKEN") or None + if environment == "production" and not all( + (gms_api_key, jira_email, jira_api_token) + ): + raise RuntimeError("GMS_KEY, JIRA_EMAIL, and JIRA_API_TOKEN are required in production") return cls( project_key=project_key, jira_base_url=os.getenv("JIRA_BASE_URL", "https://ssafy.atlassian.net").rstrip("/"), @@ -40,8 +53,15 @@ def from_env(cls) -> "Settings": cookie_secure=cookie_secure, agent_timeout_seconds=timeout, mattermost_webhook_url=mattermost_webhook_url, - hermes_command=os.getenv("HERMES_COMMAND", "hermes"), - hermes_python=os.getenv( - "HERMES_PYTHON", "/usr/local/lib/hermes-agent/venv/bin/python" + gms_api_url=os.getenv( + "GMS_API_URL", + "https://gms.ssafy.io/gmsapi/api.anthropic.com/v1/messages", ), + gms_api_key=gms_api_key, + gms_model=os.getenv("GMS_MODEL", "claude-opus-4-8"), + jira_email=jira_email, + jira_api_token=jira_api_token, + jira_board_id=int(os.getenv("JIRA_BOARD_ID", "13941")), + ai_max_workers=int(os.getenv("COWORK_AI_MAX_WORKERS", "3")), + jira_max_workers=int(os.getenv("COWORK_JIRA_MAX_WORKERS", "2")), ) diff --git a/app/jobs.py b/app/jobs.py new file mode 100644 index 0000000..7df1407 --- /dev/null +++ b/app/jobs.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import logging +import threading +import time +from concurrent.futures import CancelledError, Future, ThreadPoolExecutor +from typing import Protocol + +logger = logging.getLogger(__name__) + + +class JobWorker(Protocol): + def prepare(self, submission_id: str) -> None: ... + + def create(self, submission_id: str) -> None: ... + + +class ParallelJobRunner: + def __init__( + self, + worker: JobWorker, + *, + ai_max_workers: int, + jira_max_workers: int, + ai_queue_capacity: int = 20, + jira_queue_capacity: int = 20, + ): + if ai_max_workers < 1 or jira_max_workers < 1: + raise ValueError("worker counts must be positive") + if ai_queue_capacity < 0 or jira_queue_capacity < 0: + raise ValueError("queue capacities must not be negative") + self.worker = worker + self._ai = ThreadPoolExecutor(max_workers=ai_max_workers, thread_name_prefix="cowork-ai") + self._jira = ThreadPoolExecutor( + max_workers=jira_max_workers, thread_name_prefix="cowork-jira" + ) + self._lock = threading.RLock() + self._active: dict[tuple[str, str], Future[None]] = {} + self._capacity = { + "prepare": threading.BoundedSemaphore(ai_max_workers + ai_queue_capacity), + "create": threading.BoundedSemaphore(jira_max_workers + jira_queue_capacity), + } + self._closed = False + + def _submit(self, kind: str, submission_id: str) -> bool: + key = (kind, submission_id) + with self._lock: + if self._closed or key in self._active: + return False + permit = self._capacity[kind] + if not permit.acquire(blocking=False): + return False + executor = self._ai if kind == "prepare" else self._jira + method = self.worker.prepare if kind == "prepare" else self.worker.create + try: + future = executor.submit(method, submission_id) + except RuntimeError: + permit.release() + return False + self._active[key] = future + future.add_done_callback(lambda done, job_key=key: self._finished(job_key, done)) + return True + + def _finished(self, key: tuple[str, str], future: Future[None]) -> None: + try: + future.result() + except CancelledError: + pass + except Exception: + logger.exception("Background job failed", extra={"job_kind": key[0]}) + finally: + with self._lock: + self._active.pop(key, None) + self._capacity[key[0]].release() + + def submit_prepare(self, submission_id: str) -> bool: + return self._submit("prepare", submission_id) + + def submit_create(self, submission_id: str) -> bool: + return self._submit("create", submission_id) + + def wait_for_idle(self, timeout: float) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + with self._lock: + if not self._active: + return True + time.sleep(0.01) + with self._lock: + return not self._active + + def shutdown(self) -> None: + with self._lock: + self._closed = True + self._ai.shutdown(wait=False, cancel_futures=True) + self._jira.shutdown(wait=False, cancel_futures=True) + + +class InlineJobRunner: + """Deterministic runner for injected workers and unit tests.""" + + def __init__(self, worker: JobWorker): + self.worker = worker + + def submit_prepare(self, submission_id: str) -> bool: + self.worker.prepare(submission_id) + return True + + def submit_create(self, submission_id: str) -> bool: + self.worker.create(submission_id) + return True + + def shutdown(self) -> None: + return None diff --git a/app/main.py b/app/main.py index 4cbf17c..18777af 100644 --- a/app/main.py +++ b/app/main.py @@ -9,15 +9,16 @@ from threading import Lock from typing import Any -from fastapi import BackgroundTasks, Cookie, FastAPI, Header, HTTPException, Request, Response +from fastapi import Cookie, FastAPI, Header, HTTPException, Request, Response from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field -from .agent import HermesJiraAgent +from .api_agent import ApiTaskAgent from .alerts import MattermostAlerter from .config import Settings from .database import Database, IdempotencyConflict +from .jobs import InlineJobRunner, ParallelJobRunner from .roles import ticket_description, ticket_summary, validate_role_tag from .security import hash_password, new_token, token_digest, verify_password from .worker import SubmissionWorker @@ -25,6 +26,7 @@ COOKIE_NAME = "cowork_session" SESSION_SECONDS = 30 * 24 * 60 * 60 DUMMY_PASSWORD_HASH = hash_password("cowork-dummy-password-never-used") +QUEUE_BUSY_MESSAGE = "요청이 많아 잠시 처리할 수 없어요. 잠시 후 다시 시도해주세요" class LoginLimiter: @@ -94,14 +96,24 @@ def create_app( *, database: Database | None = None, worker: SubmissionWorker | None = None, + job_runner=None, ) -> FastAPI: settings = settings or Settings.from_env() database = database or Database(settings.database_path) database.initialize() + injected_worker = worker is not None worker = worker or SubmissionWorker( - database, - HermesJiraAgent(settings), - MattermostAlerter(settings.mattermost_webhook_url), + database, ApiTaskAgent(settings), MattermostAlerter(settings.mattermost_webhook_url) + ) + owns_job_runner = job_runner is None and not injected_worker + job_runner = job_runner or ( + InlineJobRunner(worker) + if injected_worker + else ParallelJobRunner( + worker, + ai_max_workers=settings.ai_max_workers, + jira_max_workers=settings.jira_max_workers, + ) ) login_limiter = LoginLimiter() @@ -109,8 +121,12 @@ def create_app( async def lifespan(_: FastAPI): recover = getattr(worker, "recover_inflight", None) if callable(recover): - recover() - yield + recover(job_runner.submit_prepare) + try: + yield + finally: + if owns_job_runner: + job_runner.shutdown() app = FastAPI( title="할 일 올리기", @@ -122,6 +138,7 @@ async def lifespan(_: FastAPI): app.state.settings = settings app.state.database = database app.state.worker = worker + app.state.job_runner = job_runner static_dir = Path(__file__).parent / "static" app.mount("/static", StaticFiles(directory=static_dir), name="static") @@ -248,7 +265,6 @@ def logout( @app.post("/api/submissions", status_code=202) def submit( payload: SubmissionRequest, - background_tasks: BackgroundTasks, cowork_session: str | None = Cookie(default=None), x_csrf_token: str | None = Header(default=None), ) -> dict[str, str]: @@ -292,13 +308,17 @@ def submit( except IdempotencyConflict as exc: raise HTTPException(status_code=409, detail="요청을 다시 확인해주세요") from exc if created: - background_tasks.add_task(worker.prepare, submission_id) + if not job_runner.submit_prepare(submission_id): + database.update_submission( + submission_id, "failed", public_message=QUEUE_BUSY_MESSAGE, + internal_error="AI worker queue is full or shutting down", + ) + raise HTTPException(status_code=503, detail=QUEUE_BUSY_MESSAGE) return {"submission_id": submission_id} @app.post("/api/submissions/{submission_id}/confirm", status_code=202) def confirm_submission( submission_id: str, - background_tasks: BackgroundTasks, cowork_session: str | None = Cookie(default=None), x_csrf_token: str | None = Header(default=None), ) -> dict[str, str]: @@ -328,7 +348,12 @@ def confirm_submission( ) claimed = database.claim_confirmation(submission_id, user["id"]) if claimed: - background_tasks.add_task(worker.create, submission_id) + if not job_runner.submit_create(submission_id): + database.update_submission( + submission_id, "failed", public_message=QUEUE_BUSY_MESSAGE, + internal_error="Jira worker queue is full or shutting down", + ) + raise HTTPException(status_code=503, detail=QUEUE_BUSY_MESSAGE) elif submission["state"] not in {"creating", "completed", "partial", "reconcile"}: raise HTTPException(status_code=409, detail="티켓 정보를 다시 확인해주세요") return {"submission_id": submission_id} diff --git a/app/mcp_create_bridge.py b/app/mcp_create_bridge.py deleted file mode 100644 index e859760..0000000 --- a/app/mcp_create_bridge.py +++ /dev/null @@ -1,106 +0,0 @@ -from __future__ import annotations - -import json -import re -import sys -from typing import Any, NoReturn - -from tools.mcp_tool import discover_mcp_tools -from tools.registry import registry - -from .roles import VALID_ROLE_TAGS - -TOOL_NAME = "mcp__atlassian__createJiraIssue" -ALLOWED = { - "cloudId", - "projectKey", - "issueTypeName", - "summary", - "description", - "assignee_account_id", - "contentFormat", - "additional_fields", -} - - -def _decode(value: Any) -> Any: - if isinstance(value, str): - return json.loads(value) - return value - - -def _ambiguous() -> NoReturn: - json.dump({"error": "post_create_ambiguous"}, sys.stdout) - raise SystemExit(42) - - -def main() -> None: - try: - payload: Any = json.load(sys.stdin) - except (json.JSONDecodeError, UnicodeDecodeError): - raise SystemExit("invalid request") - if not isinstance(payload, dict) or set(payload) - ALLOWED: - raise SystemExit("invalid request") - required = { - "cloudId", - "projectKey", - "issueTypeName", - "summary", - "description", - "assignee_account_id", - } - if not required.issubset(payload): - raise SystemExit("invalid request") - if payload["issueTypeName"] != "Task": - raise SystemExit("invalid request") - if not all(isinstance(payload[key], str) and payload[key] for key in required): - raise SystemExit("invalid request") - if not payload["description"].strip(): - raise SystemExit("invalid request") - additional_fields = payload.get("additional_fields") - if ( - not isinstance(additional_fields, dict) - or set(additional_fields) != {"customfield_10020", "labels"} - or not isinstance(additional_fields["customfield_10020"], int) - or isinstance(additional_fields["customfield_10020"], bool) - or additional_fields["customfield_10020"] <= 0 - or not isinstance(additional_fields["labels"], list) - or len(additional_fields["labels"]) != 1 - or additional_fields["labels"][0] not in VALID_ROLE_TAGS - ): - raise SystemExit("invalid request") - - discover_mcp_tools() - try: - raw = registry.dispatch(TOOL_NAME, payload) - except Exception: - _ambiguous() - try: - outer = _decode(raw) - if not isinstance(outer, dict) or set(outer) != {"result"}: - _ambiguous() - body = _decode(outer["result"]) - if not isinstance(body, dict): - _ambiguous() - if set(body) == {"statusCode", "data"}: - if body["statusCode"] not in {200, 201}: - raise SystemExit("create failed") - data = body["data"] - elif set(body) == {"id", "key", "self"}: - data = body - else: - _ambiguous() - if not isinstance(data, dict) or not set(data).issubset({"id", "key", "self"}): - _ambiguous() - issue_key = data.get("key") - if not isinstance(issue_key, str) or not re.fullmatch( - rf"{re.escape(payload['projectKey'])}-\d+", issue_key - ): - _ambiguous() - except (KeyError, TypeError, json.JSONDecodeError): - _ambiguous() - json.dump({"issue_key": issue_key}, sys.stdout) - - -if __name__ == "__main__": - main() diff --git a/app/mcp_lookup_bridge.py b/app/mcp_lookup_bridge.py deleted file mode 100644 index 85642c4..0000000 --- a/app/mcp_lookup_bridge.py +++ /dev/null @@ -1,55 +0,0 @@ -from __future__ import annotations - -import json -import sys -from typing import Any - -from tools.mcp_tool import discover_mcp_tools -from tools.registry import registry - -TOOL_NAME = "mcp__atlassian__lookupJiraAccountId" - - -def _decode(value: Any) -> Any: - if isinstance(value, str): - return json.loads(value) - return value - - -def main() -> None: - try: - request = json.load(sys.stdin) - except (json.JSONDecodeError, UnicodeDecodeError): - raise SystemExit("invalid request") - if not isinstance(request, dict) or set(request) != {"cloudId", "displayName"}: - raise SystemExit("invalid request") - if not all(isinstance(value, str) and value for value in request.values()): - raise SystemExit("invalid request") - - discover_mcp_tools() - raw = registry.dispatch( - TOOL_NAME, - {"cloudId": request["cloudId"], "searchString": request["displayName"]}, - ) - try: - outer = _decode(raw) - body = _decode(outer["result"]) - users_block = body["data"]["users"] - users = users_block["users"] - except (KeyError, TypeError, json.JSONDecodeError): - raise SystemExit("lookup failed") - exact = [ - user - for user in users - if isinstance(user, dict) and user.get("displayName") == request["displayName"] - ] - if users_block.get("total") != 1 or len(exact) != 1: - raise SystemExit("lookup was not unique") - account_id = exact[0].get("accountId") - if not isinstance(account_id, str) or not account_id: - raise SystemExit("lookup returned no account identifier") - json.dump({"account_id": account_id}, sys.stdout, ensure_ascii=False) - - -if __name__ == "__main__": - main() diff --git a/app/mcp_sprint_bridge.py b/app/mcp_sprint_bridge.py deleted file mode 100644 index 5d02615..0000000 --- a/app/mcp_sprint_bridge.py +++ /dev/null @@ -1,82 +0,0 @@ -from __future__ import annotations - -import json -import re -import sys -from typing import Any - -from tools.mcp_tool import discover_mcp_tools -from tools.registry import registry - -TOOL_NAME = "mcp__atlassian__searchJiraIssuesUsingJql" -SPRINT_FIELD = "customfield_10020" - - -def _decode(value: Any) -> Any: - if isinstance(value, str): - return json.loads(value) - return value - - -def main() -> None: - try: - payload: Any = json.load(sys.stdin) - except (json.JSONDecodeError, UnicodeDecodeError): - raise SystemExit("invalid request") - if not isinstance(payload, dict) or set(payload) != {"cloudId", "projectKey"}: - raise SystemExit("invalid request") - cloud_id = payload["cloudId"] - project_key = payload["projectKey"] - if not isinstance(cloud_id, str) or not cloud_id.startswith("https://"): - raise SystemExit("invalid request") - if not isinstance(project_key, str) or not re.fullmatch(r"[A-Z][A-Z0-9_]+", project_key): - raise SystemExit("invalid request") - - discover_mcp_tools() - raw = registry.dispatch( - TOOL_NAME, - { - "cloudId": cloud_id, - "jql": f"project = {project_key} AND sprint in openSprints() ORDER BY updated DESC", - "maxResults": 100, - "fields": [SPRINT_FIELD], - "searchResultMode": "issues", - }, - ) - try: - outer = _decode(raw) - body = _decode(outer["result"]) - issues = body["issues"] - if not isinstance(issues, list): - raise ValueError - sprints: dict[int, str] = {} - for issue in issues: - values = issue.get("fields", {}).get(SPRINT_FIELD) or [] - if not isinstance(values, list): - raise ValueError - for sprint in values: - if not isinstance(sprint, dict) or sprint.get("state") != "active": - continue - sprint_id = sprint.get("id") - name = sprint.get("name") - if ( - not isinstance(sprint_id, int) - or isinstance(sprint_id, bool) - or sprint_id <= 0 - or not isinstance(name, str) - or not name.strip() - ): - raise ValueError - sprints[sprint_id] = name.strip() - except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: - raise SystemExit("invalid Jira sprint response") from exc - - json.dump( - {"sprints": [{"id": sprint_id, "name": name} for sprint_id, name in sorted(sprints.items())]}, - sys.stdout, - ensure_ascii=False, - ) - - -if __name__ == "__main__": - main() diff --git a/app/seed_user.py b/app/seed_user.py index 7d9aa35..29caaac 100644 --- a/app/seed_user.py +++ b/app/seed_user.py @@ -2,10 +2,9 @@ import argparse import getpass -import json import re -import subprocess +import httpx from .config import Settings from .database import Database @@ -13,32 +12,30 @@ from .security import hash_password -def lookup_account_id(settings: Settings, display_name: str) -> str: - command = [ - settings.hermes_python, - "-m", - "app.mcp_lookup_bridge", - ] - result = subprocess.run( - command, - input=json.dumps( - {"cloudId": settings.jira_base_url, "displayName": display_name}, - ensure_ascii=False, - ), - capture_output=True, - text=True, - timeout=settings.agent_timeout_seconds, - check=False, - ) - if result.returncode != 0: +def lookup_account_id( + settings: Settings, display_name: str, *, http_client: httpx.Client | None = None +) -> str: + if not settings.jira_email or not settings.jira_api_token: + raise RuntimeError("Jira API credentials are not configured") + client = http_client or httpx.Client(timeout=settings.agent_timeout_seconds) + try: + response = client.get( + f"{settings.jira_base_url}/rest/api/3/user/assignable/search", + params={"project": settings.project_key, "query": display_name}, + auth=httpx.BasicAuth(settings.jira_email, settings.jira_api_token), + ) + except httpx.HTTPError as exc: + raise RuntimeError("Jira account lookup failed") from exc + if response.status_code != 200: raise RuntimeError("Jira account lookup failed") try: - payload = json.loads(result.stdout) - except json.JSONDecodeError as exc: + users = response.json() + exact = [user for user in users if user.get("displayName") == display_name] + except (ValueError, TypeError, AttributeError) as exc: raise RuntimeError("Jira account lookup returned an invalid response") from exc - if set(payload) != {"account_id"}: + if len(exact) != 1: raise RuntimeError("Jira account lookup was not unique") - account_id = payload["account_id"] + account_id = exact[0].get("accountId") if not isinstance(account_id, str) or not re.fullmatch(r"[A-Za-z0-9:_-]{8,200}", account_id): raise RuntimeError("Jira account lookup returned an invalid identifier") return account_id diff --git a/app/worker.py b/app/worker.py index 14e9944..e37e7ae 100644 --- a/app/worker.py +++ b/app/worker.py @@ -6,7 +6,7 @@ AgentNetworkError, AgentReconciliationRequired, AgentTimeout, - HermesJiraAgent, + Plan, PlannedTask, ) @@ -22,7 +22,7 @@ class SubmissionWorker: - def __init__(self, database: Database, agent: HermesJiraAgent, alerter: MattermostAlerter): + def __init__(self, database: Database, agent, alerter: MattermostAlerter): self.database = database self.agent = agent self.alerter = alerter @@ -247,9 +247,12 @@ def process(self, submission_id: str) -> None: ) self.create(submission_id) - def recover_inflight(self) -> None: + def recover_inflight(self, enqueue_prepare=None) -> None: self.flush_pending_alerts() for submission in self.database.list_inflight_submissions(): + if submission["state"] in {"received", "organizing"} and enqueue_prepare: + if enqueue_prepare(submission["id"]): + continue detail = self._alert( error="server restarted before submission completed", raw_input=submission["raw_input"], diff --git a/pyproject.toml b/pyproject.toml index bdac575..2d0a091 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ dependencies = [ "starlette==1.0.1", "uvicorn[standard]==0.35.0", "bcrypt==4.3.0", + "httpx==0.28.1", ] [project.optional-dependencies] diff --git a/tests/test_agent.py b/tests/test_agent.py index a34d0a5..d53e644 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -1,138 +1,35 @@ from __future__ import annotations -import json -import os -import subprocess - import pytest -from app.agent import ( - AgentError, - AgentReconciliationRequired, - HermesJiraAgent, - PlannedTask, - _parse_plan_payload, -) -from app.config import Settings - - -def settings(tmp_path): - return Settings( - project_key="S15P11A705", - jira_base_url="https://ssafy.atlassian.net", - database_path=tmp_path / "db", - cookie_secure=False, - agent_timeout_seconds=60, - mattermost_webhook_url=None, - hermes_command="hermes", - ) - - -def completed(payload): - return subprocess.CompletedProcess( - args=[], returncode=0, stdout="session_id: synthetic\n" + json.dumps(payload), stderr="" - ) - +from app.agent import AgentError, _parse_plan_payload -def bridge_completed(payload, returncode=0): - return subprocess.CompletedProcess( - args=[], returncode=returncode, stdout=json.dumps(payload), stderr="" - ) - -def test_planner_accepts_only_closed_json(monkeypatch, tmp_path): - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: completed( +def test_planner_schema_covers_every_input_line(): + payload = { + "tasks": [ { - "tasks": [ - { - "source_indices": [0, 1], - "summary": "카카오페이 결제 연동 완료", - "description": None, - }, - { - "source_indices": [0], - "summary": "결제 오류 로그 확인", - "description": "실패 케이스를 확인한다", - }, - ], - "excluded": [{"source_index": 2, "text": "2시 회의"}], + "source_indices": [0], + "summary": "로그인 오류 수정", + "description": None, } - ), - ) - plan = HermesJiraAgent(settings(tmp_path)).plan( - "작업\n실패 케이스를 확인한다\n2시 회의" - ) - assert [task.summary for task in plan.tasks] == [ - "카카오페이 결제 연동 완료", - "결제 오류 로그 확인", - ] - assert plan.tasks[0].description == "작업\n실패 케이스를 확인한다" - assert plan.tasks[1].description == "실패 케이스를 확인한다" - assert plan.excluded == ["2시 회의"] + ], + "excluded": [{"source_index": 1, "text": "오후 회의"}], + } + plan = _parse_plan_payload(payload, ["로그인 오류 수정", "오후 회의"]) + assert plan.tasks[0].description == "로그인 오류 수정" + assert plan.excluded == ["오후 회의"] - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: completed({"tasks": [], "excluded": [], "private": "x"}), - ) - with pytest.raises(AgentError): - HermesJiraAgent(settings(tmp_path)).plan("작업") - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: completed( - { - "tasks": [ - {"source_indices": [0], "summary": "첫 작업", "description": None} - ], - "excluded": [], - } - ), - ) +def test_planner_rejects_dropped_input_line(): + payload = { + "tasks": [ + {"source_indices": [0], "summary": "첫 작업", "description": None} + ], + "excluded": [], + } with pytest.raises(AgentError, match="dropped input lines"): - HermesJiraAgent(settings(tmp_path)).plan("첫 작업\n둘째 작업") - - -def test_planner_retries_once_after_schema_mismatch(monkeypatch, tmp_path): - responses = iter( - [ - completed( - { - "source_indices": [0], - "summary": "로그인 오류 재현", - "description": None, - } - ), - completed( - { - "tasks": [ - { - "source_indices": [0], - "summary": "로그인 오류 재현", - "description": None, - } - ], - "excluded": [], - } - ), - ] - ) - calls = [] - - def fake_run(*args, **kwargs): - calls.append(kwargs["input"]) - return next(responses) - - monkeypatch.setattr(subprocess, "run", fake_run) - plan = HermesJiraAgent(settings(tmp_path)).plan("로그인 오류 재현하기") - - assert [task.summary for task in plan.tasks] == ["로그인 오류 재현"] - assert len(calls) == 2 - assert "schema correction retry" in calls[1] + _parse_plan_payload(payload, ["첫 작업", "둘째 작업"]) @pytest.mark.parametrize("source_indices", [[[0]], [True], [0, 0]]) @@ -149,110 +46,3 @@ def test_planner_rejects_malformed_source_indices(source_indices): } with pytest.raises(AgentError, match="invalid planned source indices"): _parse_plan_payload(payload, ["로그인 오류 재현"]) - - -def test_planner_accepts_structured_input_over_twenty_lines(monkeypatch, tmp_path): - source_lines = [f"상세 내용 {index}" for index in range(21)] - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: completed( - { - "tasks": [ - { - "source_indices": list(range(21)), - "summary": "구조화 작업 처리", - "description": None, - } - ], - "excluded": [], - } - ), - ) - - plan = HermesJiraAgent(settings(tmp_path)).plan("\n".join(source_lines)) - - assert len(plan.tasks) == 1 - assert plan.tasks[0].description == "\n".join(source_lines) - - -def test_creator_calls_only_direct_mcp_bridge_with_fixed_fields(monkeypatch, tmp_path): - agent = HermesJiraAgent(settings(tmp_path)) - task = PlannedTask("결제 연동 완료", None) - observed = {} - - def success(command, **kwargs): - observed["command"] = command - observed["request"] = json.loads(kwargs["input"]) - return bridge_completed({"issue_key": "S15P11A705-10"}) - - monkeypatch.setattr(subprocess, "run", success) - ticket = agent.create_task(task, "account-id", 50563, "Infra") - assert ticket.issue_key == "S15P11A705-10" - assert observed["command"][0] == agent.settings.hermes_python - assert observed["command"][1:] == ["-m", "app.mcp_create_bridge"] - assert observed["request"] == { - "cloudId": "https://ssafy.atlassian.net", - "projectKey": "S15P11A705", - "issueTypeName": "Task", - "summary": "[Infra] 결제 연동 완료", - "assignee_account_id": "account-id", - "contentFormat": "markdown", - "additional_fields": { - "customfield_10020": 50563, - "labels": ["Infra"], - }, - } - - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: bridge_completed({"issue_key": "OTHER-10"}), - ) - with pytest.raises(AgentError): - agent.create_task(task, "account-id", 50563, "Infra") - - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: bridge_completed({"error": "synthetic"}, returncode=1), - ) - with pytest.raises(AgentError, match="synthetic"): - agent.create_task(task, "account-id", 50563, "Infra") - - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: bridge_completed( - {"error": "post_create_ambiguous"}, returncode=42 - ), - ) - with pytest.raises(AgentReconciliationRequired): - agent.create_task(task, "account-id", 50563, "Infra") - - -def test_agent_passes_prompts_as_single_argument_without_shell(monkeypatch, tmp_path): - observed = {} - - def fake_run(command, **kwargs): - observed["command"] = command - observed["kwargs"] = kwargs - return completed( - { - "tasks": [], - "excluded": [ - { - "source_index": 0, - "text": "$(touch /tmp/should-not-exist)", - } - ], - } - ) - - monkeypatch.setattr(subprocess, "run", fake_run) - HermesJiraAgent(settings(tmp_path)).plan("$(touch /tmp/should-not-exist)") - assert observed["kwargs"].get("shell") is None - assert isinstance(observed["command"], list) - assert "$(touch /tmp/should-not-exist)" not in " ".join(observed["command"]) - assert "$(touch /tmp/should-not-exist)" in observed["kwargs"]["input"] - assert not os.path.exists("/tmp/should-not-exist") diff --git a/tests/test_api.py b/tests/test_api.py index fc80af9..664ccaf 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -61,7 +61,6 @@ def client_and_worker(tmp_path): cookie_secure=False, agent_timeout_seconds=60, mattermost_webhook_url=None, - hermes_command="hermes", ) database = Database(settings.database_path) database.initialize() diff --git a/tests/test_api_clients.py b/tests/test_api_clients.py new file mode 100644 index 0000000..6b84836 --- /dev/null +++ b/tests/test_api_clients.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import json + +import httpx +import pytest + +from app.agent import AgentNetworkError, AgentReconciliationRequired, PlannedTask +from app.api_agent import ApiTaskAgent +from app.config import Settings + + +def settings(tmp_path): + return Settings( + project_key="S15P11A705", + jira_base_url="https://ssafy.atlassian.net", + database_path=tmp_path / "db", + cookie_secure=False, + agent_timeout_seconds=60, + mattermost_webhook_url=None, + gms_api_url="https://gms.ssafy.io/gmsapi/api.anthropic.com/v1/messages", + gms_api_key="secret-key", + gms_model="claude-opus-4-8", + jira_email="cowork@example.com", + jira_api_token="jira-token", + jira_board_id=13941, + ) + + +def test_gms_planner_uses_anthropic_messages_contract_and_closed_schema(tmp_path): + observed = {} + + def handler(request: httpx.Request): + observed["request"] = request + observed["body"] = json.loads(request.content) + payload = { + "tasks": [ + {"source_indices": [0], "summary": "로그인 오류 수정", "description": None} + ], + "excluded": [{"source_index": 1, "text": "오후 회의"}], + } + return httpx.Response( + 200, + json={"content": [{"type": "text", "text": json.dumps(payload, ensure_ascii=False)}]}, + ) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + plan = ApiTaskAgent(settings(tmp_path), http_client=client).plan("로그인 오류 수정\n오후 회의") + + assert [task.summary for task in plan.tasks] == ["로그인 오류 수정"] + assert plan.excluded == ["오후 회의"] + assert observed["request"].headers["x-api-key"] == "secret-key" + assert observed["request"].headers["anthropic-version"] == "2023-06-01" + assert observed["body"]["model"] == "claude-opus-4-8" + assert observed["body"]["max_tokens"] == 2048 + + +def test_gms_429_is_a_retryable_network_failure_without_secret_leak(tmp_path): + client = httpx.Client( + transport=httpx.MockTransport( + lambda request: httpx.Response(429, headers={"retry-after": "2"}, text="limited") + ) + ) + with pytest.raises(AgentNetworkError) as error: + ApiTaskAgent(settings(tmp_path), http_client=client).plan("작업") + assert "secret-key" not in str(error.value) + + +def test_jira_rest_lists_active_sprints_and_creates_fixed_task(tmp_path): + observed = [] + + def handler(request: httpx.Request): + observed.append(request) + if request.method == "GET": + return httpx.Response( + 200, + json={"values": [{"id": 50563, "name": "S15P11A7 스프린트", "state": "active"}]}, + ) + body = json.loads(request.content) + assert body == { + "fields": { + "project": {"key": "S15P11A705"}, + "issuetype": {"name": "Task"}, + "summary": "[Infra] 배포 자동화", + "description": "배포를 자동화한다", + "assignee": {"accountId": "account-id"}, + "customfield_10020": 50563, + "labels": ["Infra"], + } + } + return httpx.Response(201, json={"id": "100", "key": "S15P11A705-100"}) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + agent = ApiTaskAgent(settings(tmp_path), http_client=client) + sprints = agent.list_active_sprints() + ticket = agent.create_task( + PlannedTask("배포 자동화", "배포를 자동화한다"), "account-id", 50563, "Infra" + ) + + assert [(s.id, s.name) for s in sprints] == [(50563, "S15P11A7 스프린트")] + assert ticket.issue_key == "S15P11A705-100" + assert observed[0].url.path == "/rest/agile/1.0/board/13941/sprint" + assert observed[1].url.path == "/rest/api/2/issue" + assert observed[0].headers["authorization"].startswith("Basic ") + + +def test_jira_create_http_408_requires_reconciliation(tmp_path): + client = httpx.Client( + transport=httpx.MockTransport(lambda request: httpx.Response(408)) + ) + agent = ApiTaskAgent(settings(tmp_path), http_client=client) + with pytest.raises(AgentReconciliationRequired): + agent.create_task(PlannedTask("작업", "설명"), "account-id", 17, "BE") diff --git a/tests/test_api_smoke.py b/tests/test_api_smoke.py new file mode 100644 index 0000000..c8ec675 --- /dev/null +++ b/tests/test_api_smoke.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import json + +import httpx +import pytest + +from app.api_smoke import run_read_only_smoke + + +def test_smoke_calls_gms_post_and_jira_get_only(): + observed = [] + + def handler(request: httpx.Request) -> httpx.Response: + observed.append(request) + if request.url.host == "gms.ssafy.io": + assert request.method == "POST" + assert request.headers["x-api-key"] == "gms-test-key" + return httpx.Response( + 200, + json={ + "content": [ + { + "type": "text", + "text": json.dumps( + { + "tasks": [ + { + "source_indices": [0], + "summary": "CI smoke test", + "description": None, + } + ], + "excluded": [], + } + ), + } + ] + }, + ) + assert request.method == "GET" + assert request.url.path == "/rest/agile/1.0/board/13941/sprint" + assert request.url.params["state"] == "active" + assert request.headers["authorization"].startswith("Basic ") + return httpx.Response(200, json={"values": [{"id": 17, "name": "Sprint 17"}]}) + + count = run_read_only_smoke( + gms_key="gms-test-key", + jira_email="cowork@example.com", + jira_api_token="jira-test-token", + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + assert count == 1 + assert [request.method for request in observed] == ["POST", "GET"] + assert all("/issue" not in request.url.path for request in observed) + + +def test_smoke_error_does_not_include_external_response_body(): + client = httpx.Client( + transport=httpx.MockTransport( + lambda request: httpx.Response(401, text="sensitive upstream diagnostic") + ) + ) + with pytest.raises(RuntimeError) as captured: + run_read_only_smoke( + gms_key="gms-test-key", + jira_email="cowork@example.com", + jira_api_token="jira-test-token", + client=client, + ) + assert "sensitive upstream diagnostic" not in str(captured.value) diff --git a/tests/test_ci_contract.py b/tests/test_ci_contract.py index 735da79..cfd5285 100644 --- a/tests/test_ci_contract.py +++ b/tests/test_ci_contract.py @@ -33,7 +33,7 @@ def test_required_ci_workflow_is_pinned_and_runs_all_gates(): assert "docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a" in workflow assert "ghcr.io/team-pinlog/cowork:${{ github.sha }}" in workflow assert "docker run --detach" in workflow - assert workflow.count("from mcp.client.auth.oauth2 import OAuthClientProvider") == 2 + assert "mcp.client" not in workflow assert "curl --fail --silent --show-error http://127.0.0.1:18080/health" in workflow assert "docker push" in workflow assert ":latest" not in workflow diff --git a/tests/test_container_contract.py b/tests/test_container_contract.py index 7259bc7..7aaf828 100644 --- a/tests/test_container_contract.py +++ b/tests/test_container_contract.py @@ -12,22 +12,18 @@ def test_container_is_pinned_non_root_and_uses_persistent_paths(): dockerfile = Path("Dockerfile").read_text(encoding="utf-8") assert f"FROM {BASE_IMAGE}" in dockerfile - assert "hermes-agent[mcp]==0.19.0" in dockerfile + assert "hermes-agent" not in dockerfile assert "USER 1000:1000" in dockerfile - assert "HERMES_HOME=/data/hermes" in dockerfile assert "COWORK_DATABASE_PATH=/data/cowork.db" in dockerfile - assert "HERMES_PYTHON=/usr/local/bin/python" in dockerfile assert 'CMD ["uvicorn", "app.main:create_app", "--factory"' in dockerfile assert ":latest" not in dockerfile -def test_runtime_dependencies_are_compatible_with_hermes_mcp(): +def test_runtime_dependencies_include_http_api_client(): project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"] assert "starlette==1.0.1" in project["dependencies"] - test_dependencies = project["optional-dependencies"]["test"] - assert "httpx==0.28.1" in test_dependencies - assert all(not dependency.startswith("httpx2") for dependency in test_dependencies) + assert "httpx==0.28.1" in project["dependencies"] def test_container_context_excludes_credentials_and_runtime_data(): diff --git a/tests/test_jobs.py b/tests/test_jobs.py new file mode 100644 index 0000000..805181f --- /dev/null +++ b/tests/test_jobs.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import threading +import time + +from app.jobs import ParallelJobRunner + + +class BlockingWorker: + def __init__(self): + self.barrier = threading.Barrier(2) + self.lock = threading.Lock() + self.active = 0 + self.max_active = 0 + self.completed = [] + + def prepare(self, submission_id: str): + with self.lock: + self.active += 1 + self.max_active = max(self.max_active, self.active) + self.barrier.wait(timeout=2) + time.sleep(0.03) + with self.lock: + self.active -= 1 + self.completed.append(submission_id) + + def create(self, submission_id: str): + self.prepare(submission_id) + + +def test_prepare_jobs_from_different_users_run_in_parallel(): + worker = BlockingWorker() + runner = ParallelJobRunner(worker, ai_max_workers=2, jira_max_workers=1) + try: + runner.submit_prepare("submission-a") + runner.submit_prepare("submission-b") + assert runner.wait_for_idle(timeout=3) + finally: + runner.shutdown() + + assert worker.max_active == 2 + assert set(worker.completed) == {"submission-a", "submission-b"} + + +def test_immediately_completed_job_does_not_deadlock_submitter(): + class Worker: + def prepare(self, submission_id): + return None + + def create(self, submission_id): + return None + + runner = ParallelJobRunner(Worker(), ai_max_workers=1, jira_max_workers=1) + result = [] + thread = threading.Thread(target=lambda: result.append(runner.submit_prepare("fast"))) + thread.start() + thread.join(timeout=1) + try: + assert not thread.is_alive() + assert result == [True] + assert runner.wait_for_idle(timeout=1) + finally: + runner.shutdown() + + +def test_duplicate_job_key_is_not_dispatched_twice(): + started = threading.Event() + release = threading.Event() + + class Worker: + calls = 0 + + def prepare(self, submission_id): + self.calls += 1 + started.set() + release.wait(timeout=2) + + def create(self, submission_id): + raise AssertionError + + worker = Worker() + runner = ParallelJobRunner(worker, ai_max_workers=1, jira_max_workers=1) + try: + assert runner.submit_prepare("same") is True + assert started.wait(timeout=1) + assert runner.submit_prepare("same") is False + release.set() + assert runner.wait_for_idle(timeout=2) + finally: + runner.shutdown() + assert worker.calls == 1 + + +def test_queue_capacity_rejects_unbounded_backlog(): + started = threading.Event() + release = threading.Event() + + class Worker: + def prepare(self, submission_id): + started.set() + release.wait(timeout=2) + + def create(self, submission_id): + return None + + runner = ParallelJobRunner( + Worker(), ai_max_workers=1, jira_max_workers=1, ai_queue_capacity=0 + ) + try: + assert runner.submit_prepare("running") + assert started.wait(timeout=1) + assert not runner.submit_prepare("must-not-queue") + finally: + release.set() + runner.shutdown() + + +def test_shutdown_does_not_wait_for_running_or_queued_jobs(): + started = threading.Event() + release = threading.Event() + + class Worker: + def prepare(self, submission_id): + started.set() + release.wait(timeout=2) + + def create(self, submission_id): + return None + + runner = ParallelJobRunner( + Worker(), ai_max_workers=1, jira_max_workers=1, ai_queue_capacity=1 + ) + assert runner.submit_prepare("running") + assert started.wait(timeout=1) + assert runner.submit_prepare("queued") + before = time.monotonic() + try: + runner.shutdown() + assert time.monotonic() - before < 0.25 + finally: + release.set() diff --git a/tests/test_mcp_create_bridge.py b/tests/test_mcp_create_bridge.py deleted file mode 100644 index 2afc081..0000000 --- a/tests/test_mcp_create_bridge.py +++ /dev/null @@ -1,116 +0,0 @@ -import importlib -import io -import json -import sys -import types -from types import SimpleNamespace - -import pytest - - -def test_dispatch_exception_exits_as_post_create_ambiguous(monkeypatch, capsys): - tools_package = types.ModuleType("tools") - mcp_module = types.ModuleType("tools.mcp_tool") - registry_module = types.ModuleType("tools.registry") - - mcp_module.discover_mcp_tools = lambda: None - - def raise_timeout(*args, **kwargs): - raise TimeoutError("synthetic MCP timeout") - - registry_module.registry = SimpleNamespace(dispatch=raise_timeout) - monkeypatch.setitem(sys.modules, "tools", tools_package) - monkeypatch.setitem(sys.modules, "tools.mcp_tool", mcp_module) - monkeypatch.setitem(sys.modules, "tools.registry", registry_module) - sys.modules.pop("app.mcp_create_bridge", None) - bridge = importlib.import_module("app.mcp_create_bridge") - - request = { - "cloudId": "https://ssafy.atlassian.net", - "projectKey": "S15P11A705", - "issueTypeName": "Task", - "summary": "합성 작업", - "description": "합성 작업 상세", - "assignee_account_id": "synthetic-account", - "contentFormat": "markdown", - "additional_fields": {"customfield_10020": 50563, "labels": ["Infra"]}, - } - monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(request))) - - with pytest.raises(SystemExit) as exc_info: - bridge.main() - - assert exc_info.value.code == 42 - assert json.loads(capsys.readouterr().out) == {"error": "post_create_ambiguous"} - - -def test_direct_atlassian_create_response_returns_issue_key(monkeypatch, capsys): - tools_package = types.ModuleType("tools") - mcp_module = types.ModuleType("tools.mcp_tool") - registry_module = types.ModuleType("tools.registry") - - mcp_module.discover_mcp_tools = lambda: None - registry_module.registry = SimpleNamespace( - dispatch=lambda *_args, **_kwargs: { - "result": json.dumps( - { - "id": "1561640", - "key": "S15P11A705-17", - "self": "https://api.atlassian.com/ex/jira/site/rest/api/3/issue/1561640", - } - ) - } - ) - monkeypatch.setitem(sys.modules, "tools", tools_package) - monkeypatch.setitem(sys.modules, "tools.mcp_tool", mcp_module) - monkeypatch.setitem(sys.modules, "tools.registry", registry_module) - sys.modules.pop("app.mcp_create_bridge", None) - bridge = importlib.import_module("app.mcp_create_bridge") - - request = { - "cloudId": "https://ssafy.atlassian.net", - "projectKey": "S15P11A705", - "issueTypeName": "Task", - "summary": "합성 작업", - "description": "합성 작업 상세", - "assignee_account_id": "synthetic-account", - "contentFormat": "markdown", - "additional_fields": {"customfield_10020": 50563, "labels": ["Infra"]}, - } - monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(request))) - - bridge.main() - - assert json.loads(capsys.readouterr().out) == {"issue_key": "S15P11A705-17"} - - -def test_empty_description_is_rejected_before_mcp_dispatch(monkeypatch): - tools_package = types.ModuleType("tools") - mcp_module = types.ModuleType("tools.mcp_tool") - registry_module = types.ModuleType("tools.registry") - - def unexpected_call(*_args, **_kwargs): - raise AssertionError("MCP must not be called") - - setattr(mcp_module, "discover_mcp_tools", unexpected_call) - setattr(registry_module, "registry", SimpleNamespace(dispatch=unexpected_call)) - monkeypatch.setitem(sys.modules, "tools", tools_package) - monkeypatch.setitem(sys.modules, "tools.mcp_tool", mcp_module) - monkeypatch.setitem(sys.modules, "tools.registry", registry_module) - sys.modules.pop("app.mcp_create_bridge", None) - bridge = importlib.import_module("app.mcp_create_bridge") - - request = { - "cloudId": "https://ssafy.atlassian.net", - "projectKey": "S15P11A705", - "issueTypeName": "Task", - "summary": "합성 작업", - "description": " ", - "assignee_account_id": "synthetic-account", - "contentFormat": "markdown", - "additional_fields": {"customfield_10020": 50563, "labels": ["Infra"]}, - } - monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(request))) - - with pytest.raises(SystemExit, match="invalid request"): - bridge.main() diff --git a/tests/test_seed_user.py b/tests/test_seed_user.py index edd0dc9..7a2c575 100644 --- a/tests/test_seed_user.py +++ b/tests/test_seed_user.py @@ -1,6 +1,4 @@ -import json -import subprocess - +import httpx import pytest from app.config import Settings @@ -15,40 +13,44 @@ def settings(tmp_path): cookie_secure=False, agent_timeout_seconds=60, mattermost_webhook_url=None, - hermes_command="hermes", + jira_email="cowork@example.com", + jira_api_token="token", ) -def test_lookup_uses_direct_read_only_bridge(monkeypatch, tmp_path): +def test_lookup_uses_direct_read_only_jira_rest_api(tmp_path): observed = {} - def fake_run(command, **kwargs): - observed["command"] = command - observed["request"] = json.loads(kwargs["input"]) - return subprocess.CompletedProcess( - args=command, - returncode=0, - stdout=json.dumps({"account_id": "synthetic-account-id"}), - stderr="", + def handler(request): + observed["request"] = request + return httpx.Response( + 200, + json=[{"displayName": "김팀원", "accountId": "synthetic-account-id"}], ) - monkeypatch.setattr(subprocess, "run", fake_run) - value = lookup_account_id(settings(tmp_path), "김팀원") + value = lookup_account_id( + settings(tmp_path), + "김팀원", + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) assert value == "synthetic-account-id" - assert observed["command"][1:] == ["-m", "app.mcp_lookup_bridge"] - assert observed["request"] == { - "cloudId": "https://ssafy.atlassian.net", - "displayName": "김팀원", - } - - -def test_lookup_rejects_invalid_response(monkeypatch, tmp_path): - monkeypatch.setattr( - subprocess, - "run", - lambda *args, **kwargs: subprocess.CompletedProcess( - args=[], returncode=0, stdout=json.dumps({"error": "not found"}), stderr="" - ), + assert observed["request"].url.path == "/rest/api/3/user/assignable/search" + assert observed["request"].url.params["project"] == "S15P11A705" + assert observed["request"].url.params["query"] == "김팀원" + assert observed["request"].headers["authorization"].startswith("Basic ") + + +def test_lookup_rejects_non_unique_response(tmp_path): + client = httpx.Client( + transport=httpx.MockTransport( + lambda request: httpx.Response( + 200, + json=[ + {"displayName": "김팀원", "accountId": "first-account"}, + {"displayName": "김팀원", "accountId": "second-account"}, + ], + ) + ) ) - with pytest.raises(RuntimeError): - lookup_account_id(settings(tmp_path), "없는 사용자") + with pytest.raises(RuntimeError, match="not unique"): + lookup_account_id(settings(tmp_path), "김팀원", http_client=client)