diff --git a/.gitea/workflows/run-agent.yml b/.gitea/workflows/run-agent.yml new file mode 100644 index 00000000..d14b98c2 --- /dev/null +++ b/.gitea/workflows/run-agent.yml @@ -0,0 +1,108 @@ +# 지정한 Agent YAML 을 AgentBackend API 로 실행하는 Gitea Actions 워크플로우. +# +# 사용법: Gitea 저장소 → Actions 탭 → "Run Agent YAML via API" → Run workflow +# yaml_path 에 repo 상대경로 입력, 예: +# Case_02_Comparison_Research/YAML_Prompts/1. Stage_1/v.5/Stage_1_Part_4.yml +# +# 요구사항: +# - 저장소 설정에서 Actions 활성화 + act_runner 등록 (runs-on: ubuntu-latest 라벨) +# - runner 에서 legalpoc.eroomai.com 접근 가능해야 함 +# +# 실행 흐름 (SKILL.md §0.6 경로 A): +# upload-agent 등록 → SSE start → stage_complete 자동 confirm → execution_complete +# stage_error 발생 시 세션을 cancel 하고 실패 처리한다. +# 전체 이벤트 로그 / final_output / summary 는 아티팩트로 저장된다. +# +# 실행 시간 상한 (주의): +# Gitea 의 [actions] ENDLESS_TASK_TIMEOUT 과 act_runner 의 runner.timeout 기본값이 +# 각각 3h 라서, timeout-minutes 를 아무리 크게 줘도 3h 에서 강제 종료된다. +# 강제 종료되면 if: always() 아티팩트 업로드도 건너뛰므로, +# 스크립트의 --max-runtime(기본 9000s=150m) 이 먼저 세션을 정리하고 종료하도록 +# max_runtime < timeout-minutes(175m) < 3h 순서를 유지한다. +# 3h 이상 돌려야 하면: app.ini 의 ENDLESS_TASK_TIMEOUT, act_runner config 의 +# runner.timeout, 아래 timeout-minutes, max_runtime 입력을 모두 함께 올릴 것. + +name: Run Agent YAML via API + +on: + workflow_dispatch: + inputs: + yaml_path: + description: '실행할 Agent YAML 경로 (repo 상대경로)' + required: true + type: string + user_id: + description: 'AgentBackend user_id' + required: false + default: 'jsahn' + type: string + workspace_id: + description: 'AgentBackend workspace_id' + required: false + default: 'e1d0baed-cec9-44af-9989-b7aab09bcfd1' + type: string + user_input: + description: 'Agent 에 전달할 user_input' + required: false + default: '' + type: string + start_stage_index: + description: '시작 stage 인덱스 (0-based)' + required: false + default: '0' + type: string + api_base: + description: 'AgentBackend API base URL' + required: false + default: 'https://legalpoc.eroomai.com/api' + type: string + max_runtime_seconds: + description: '총 실행 시간 상한(초). Gitea/act_runner 3h 상한보다 짧게' + required: false + default: '9000' + type: string + +jobs: + run-agent: + runs-on: ubuntu-latest + timeout-minutes: 175 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Install dependencies + run: python -m pip install --quiet httpx httpx-sse + + - name: Run agent via API + env: + YAML_PATH: ${{ inputs.yaml_path }} + API_BASE: ${{ inputs.api_base }} + AGENT_USER_ID: ${{ inputs.user_id }} + AGENT_WORKSPACE_ID: ${{ inputs.workspace_id }} + AGENT_USER_INPUT: ${{ inputs.user_input }} + START_STAGE_INDEX: ${{ inputs.start_stage_index }} + MAX_RUNTIME_SECONDS: ${{ inputs.max_runtime_seconds }} + # '=' 형식 필수: user_input 이 '-' 로 시작하는 자유 텍스트여도 argparse 가 값으로 인식 + run: | + python scripts/run_agent_api.py \ + --yaml-path="$YAML_PATH" \ + --api-base="$API_BASE" \ + --user-id="$AGENT_USER_ID" \ + --workspace-id="$AGENT_WORKSPACE_ID" \ + --user-input="$AGENT_USER_INPUT" \ + --start-stage-index="$START_STAGE_INDEX" \ + --max-runtime="$MAX_RUNTIME_SECONDS" \ + --output-dir=agent_run_output + + - name: Upload run artifacts + if: always() + uses: actions/upload-artifact@v4 + with: + name: agent-run-${{ github.run_number }} + path: agent_run_output/ + if-no-files-found: warn diff --git a/scripts/run_agent_api.py b/scripts/run_agent_api.py new file mode 100644 index 00000000..670542ed --- /dev/null +++ b/scripts/run_agent_api.py @@ -0,0 +1,403 @@ +#!/usr/bin/env python3 +""" +Upload an Agent YAML to AgentBackend and execute it via SSE. + +Flow (SKILL.md §0.6 — 경로 A): + 1. POST {api_base}/upload-agent?user_id=... (multipart YAML 등록) + 2. POST {api_base}/sse/agent/{name}/start (SSE 실행) + 3. stage_complete 이벤트 수신 시 자동 confirm (CI 무인 실행, 재시도 포함) + 4. 연결 끊김 시 /sse/agent/{name}/reconnect/{sid} (재연결) + 5. stage_error / --max-runtime 초과 / SIGTERM 시 세션 cancel 후 종료 + +Exit code: 0 = execution_complete / 1 = 그 외 (stage_error, stopped, timeout, ...) + +--output-dir 에 저장되는 파일: + events.jsonl — 수신한 모든 SSE 이벤트 (한 줄당 1개) + final_output.txt — execution_complete 이벤트의 final_outputs (stage 별 출력 dict) + summary.md — 실행 요약 (GITHUB_STEP_SUMMARY 에도 기록) +""" + +from __future__ import annotations + +import argparse +import json +import os +import signal +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import quote + +import httpx +from httpx_sse import SSEError, connect_sse + +PRINT_TRUNCATE = 500 + + +class RunTimeout(Exception): + """--max-runtime 초과.""" + + +def _now() -> str: + return datetime.now(timezone.utc).strftime("%H:%M:%S") + + +def log(msg: str) -> None: + print(f"[{_now()}] {msg}", flush=True) + + +def truncate(value: object, limit: int = PRINT_TRUNCATE) -> str: + text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) + return text if len(text) <= limit else text[:limit] + f"... (+{len(text) - limit} chars)" + + +class AgentRunner: + def __init__(self, args: argparse.Namespace) -> None: + self.api_base = args.api_base.rstrip("/") + self.yaml_path = Path(args.yaml_path) + self.user_id = args.user_id + self.workspace_id = args.workspace_id + self.user_input = args.user_input + self.start_stage_index = args.start_stage_index + self.read_timeout = args.read_timeout + self.max_reconnects = args.max_reconnects + self.max_runtime = args.max_runtime + self.output_dir = Path(args.output_dir) + self.output_dir.mkdir(parents=True, exist_ok=True) + + self.agent_name: str = "" + self.session_id: str = "" + self.final_status: str = "unknown" + self.final_error: str = "" + self.event_counts: dict[str, int] = {} + self.stages_confirmed = 0 + self.started_at = time.monotonic() + self.deadline = self.started_at + self.max_runtime + + self.sse_client = httpx.Client(timeout=httpx.Timeout(30.0)) + # action/cancel 등 단발 요청은 SSE 스트림과 분리된 클라이언트로 보낸다 + self.api_client = httpx.Client(timeout=60.0) + + self.events_file = (self.output_dir / "events.jsonl").open("a", encoding="utf-8") + + def _remaining(self) -> float: + return self.deadline - time.monotonic() + + # ------------------------------------------------------------------ upload + + def upload(self) -> None: + if not self.yaml_path.is_file(): + raise SystemExit(f"ERROR: YAML not found: {self.yaml_path}") + + log(f"Uploading agent YAML: {self.yaml_path}") + with self.yaml_path.open("rb") as f: + resp = self.api_client.post( + f"{self.api_base}/upload-agent", + params={"user_id": self.user_id}, + files={"file": (self.yaml_path.name, f, "application/x-yaml")}, + ) + if resp.status_code != 200: + raise SystemExit(f"ERROR: upload-agent failed HTTP {resp.status_code}: {resp.text[:1000]}") + + data = resp.json() + if not data.get("success") or not data.get("agent_name"): + raise SystemExit(f"ERROR: upload-agent rejected: {json.dumps(data, ensure_ascii=False)[:1000]}") + + self.agent_name = data["agent_name"] + log(f"Agent registered: {self.agent_name} (stages={data.get('stages')})") + + # ------------------------------------------------------------------ actions + + def cancel_session(self) -> None: + if not (self.agent_name and self.session_id): + return + try: + self.api_client.post( + f"{self.api_base}/sse/agent/{quote(self.agent_name, safe='')}" + f"/cancel/{self.session_id}" + ) + log(f"Cancel requested for session {self.session_id}") + except httpx.HTTPError as exc: + log(f"WARNING: cancel request failed: {exc}") + + def _send_action(self, action: str, attempts: int = 5) -> bool: + """Send a stage action; retry transient failures inside the server's 600s wait window.""" + url = ( + f"{self.api_base}/sse/agent/{quote(self.agent_name, safe='')}" + f"/action/{self.session_id}" + ) + for i in range(1, attempts + 1): + try: + resp = self.api_client.post(url, json={"action": action}) + if resp.status_code == 200: + log(f" -> action '{action}' sent") + return True + if resp.status_code in (400, 404): + # 세션 소멸/거부 — 재시도 무의미 + log(f" -> action '{action}' failed HTTP {resp.status_code}: {resp.text[:300]}") + return False + log(f" -> action '{action}' attempt {i}/{attempts} HTTP {resp.status_code}: {resp.text[:300]}") + except httpx.HTTPError as exc: + log(f" -> action '{action}' attempt {i}/{attempts} request error: {exc}") + if i < attempts: + time.sleep(min(2 ** i, 60)) + return False + + # ------------------------------------------------------------------ events + + def _record(self, event: dict) -> None: + event["_received_at"] = datetime.now(timezone.utc).isoformat() + self.events_file.write(json.dumps(event, ensure_ascii=False) + "\n") + self.events_file.flush() + etype = event.get("type", "unknown") + self.event_counts[etype] = self.event_counts.get(etype, 0) + 1 + + def _handle(self, event: dict) -> bool: + """Returns True when execution reached a terminal state.""" + self._record(event) + etype = event.get("type", "unknown") + + if etype == "session_started": + self.session_id = event.get("session_id", self.session_id) + log( + f"Session started: {self.session_id} " + f"(total_stages={event.get('total_stages')}, start_index={event.get('start_stage_index')})" + ) + elif etype == "stage_start": + log(f"=== Stage start: index={event.get('stage_index')} {event.get('stage_name', '')}") + elif etype == "task_start": + log(f" task start: {event.get('task_name', '?')}") + elif etype == "task_iteration": + log(f" task iter : {event.get('task_name', '?')} #{event.get('iteration', '?')}") + elif etype == "task_complete": + log(f" task done : {event.get('task_name', '?')}") + elif etype == "stage_complete": + log(f"=== Stage complete: index={event.get('stage_index')} — auto-confirming") + if self._send_action("confirm"): + self.stages_confirmed += 1 + else: + # confirm 미전달 상태로 방치하면 서버 600s action-timeout 까지 세션이 잠긴다 + self.final_status = "error" + self.final_error = "confirm action could not be delivered — cancelling session" + log(f"ERROR: {self.final_error}") + self.cancel_session() + return True + elif etype == "stage_error": + # 서버는 stage_error 후 클라이언트 개입 없이는 같은 stage 를 재실행하거나 + # 실패 stage 를 건너뛰므로, CI 에서는 즉시 세션을 취소하고 실패 처리한다. + self.final_status = "stage_error" + self.final_error = str(event.get("error", "")) + log(f"=== Stage ERROR: index={event.get('stage_index')} — {truncate(self.final_error)}") + self.cancel_session() + return True + elif etype == "execution_complete": + self.final_status = "completed" + # 현행 백엔드는 final_outputs(stage 별 dict), 구버전 문서는 final_output + final_output = event.get("final_outputs") or event.get("final_output", "") + if not final_output: + log("WARNING: execution_complete has neither final_outputs nor final_output") + out_path = self.output_dir / "final_output.txt" + out_path.write_text( + final_output if isinstance(final_output, str) + else json.dumps(final_output, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + log(f"Execution complete. final_output saved to {out_path}") + return True + elif etype == "execution_stopped": + self.final_status = "stopped" + self.final_error = str(event.get("reason") or event.get("message") or "") + log(f"Execution STOPPED: {self.final_error}") + return True + elif etype == "error": + self.final_status = "error" + self.final_error = str(event.get("error", "")) + log(f"Execution ERROR: {truncate(self.final_error)}") + return True + else: + log(f" event [{etype}]: {truncate({k: v for k, v in event.items() if k != 'type'})}") + + return False + + # ------------------------------------------------------------------ stream + + def _stream(self, mode: str) -> bool: + """Open one SSE connection and consume events. Returns True on terminal event.""" + remaining = self._remaining() + if remaining <= 0: + raise RunTimeout() + + if mode == "start": + url = f"{self.api_base}/sse/agent/{quote(self.agent_name, safe='')}/start" + body = { + "user_id": self.user_id, + "workspace_id": self.workspace_id, + "user_input": self.user_input, + "start_stage_index": self.start_stage_index, + } + else: + url = ( + f"{self.api_base}/sse/agent/{quote(self.agent_name, safe='')}" + f"/reconnect/{self.session_id}" + ) + body = {} + + # read timeout: 이벤트 간 무응답 상한이며 남은 실행 예산을 넘지 않게 잡는다 + timeout = httpx.Timeout( + connect=30.0, + read=min(self.read_timeout, max(remaining, 30.0)), + write=30.0, + pool=30.0, + ) + + log(f"SSE {mode}: {url}") + with connect_sse(self.sse_client, "POST", url, json=body, timeout=timeout) as event_source: + for sse in event_source.iter_sse(): + if self._remaining() <= 0: + raise RunTimeout() + if not sse.data: + continue + try: + event = json.loads(sse.data) + except json.JSONDecodeError: + log(f" (unparseable SSE data) {truncate(sse.data)}") + continue + if self._handle(event): + return True + return False + + def run(self) -> int: + try: + self.upload() + + mode = "start" + reconnects = 0 + while True: + try: + if self._stream(mode): + break + # 스트림이 종료 이벤트 없이 닫힘 → 재연결 시도 + raise ConnectionError("SSE stream ended without a terminal event") + except RunTimeout: + self.final_status = "timeout" + self.final_error = ( + f"--max-runtime {int(self.max_runtime)}s exceeded — cancelling session" + ) + log(f"ERROR: {self.final_error}") + self.cancel_session() + break + except SSEError as exc: + # 서버가 SSE 가 아닌 응답을 반환 (404/500 등) + self.final_status = "error" + self.final_error = f"SSE handshake failed ({mode}): {exc}" + log(f"ERROR: {self.final_error}") + break + except (httpx.HTTPError, ConnectionError) as exc: + if self.final_status != "unknown": + break + reconnects += 1 + if not self.session_id or reconnects > self.max_reconnects: + self.final_status = "error" + self.final_error = ( + f"connection lost ({exc}); reconnect attempts exhausted " + f"({reconnects - 1}/{self.max_reconnects}). " + f"과거 세션 조회: GET {self.api_base}/history/sessions" + f"?user_id={self.user_id}&agent_name={self.agent_name}" + ) + log(f"ERROR: {self.final_error}") + self.cancel_session() + break + wait = min(10 * reconnects, 60) + log(f"Connection lost ({exc}); reconnect {reconnects}/{self.max_reconnects} in {wait}s") + time.sleep(wait) + mode = "reconnect" + except KeyboardInterrupt: + # SIGINT / SIGTERM(핸들러가 KeyboardInterrupt 로 변환) — 세션 정리 후 종료 + self.final_status = "interrupted" + self.final_error = "interrupted by SIGINT/SIGTERM — session cancelled" + log(self.final_error) + self.cancel_session() + except SystemExit as exc: + # upload() 의 ERROR 경로 — summary 에 남기고 실패 처리 + self.final_status = "error" + self.final_error = str(exc) + log(self.final_error) + except Exception as exc: # noqa: BLE001 — 어떤 실패든 summary 를 남긴다 + self.final_status = "error" + self.final_error = f"unexpected {type(exc).__name__}: {exc}" + log(f"ERROR: {self.final_error}") + finally: + try: + self._write_summary() + except Exception as exc: # noqa: BLE001 + log(f"WARNING: summary write failed: {exc}") + self.events_file.close() + + return 0 if self.final_status == "completed" else 1 + + # ------------------------------------------------------------------ report + + def _write_summary(self) -> None: + elapsed = int(time.monotonic() - self.started_at) + status_icon = "✅" if self.final_status == "completed" else "❌" + lines = [ + "# Agent Run Summary", + "", + "| 항목 | 값 |", + "|------|-----|", + f"| Agent | `{self.agent_name or '-'}` |", + f"| YAML | `{self.yaml_path}` |", + f"| Session | `{self.session_id or '-'}` |", + f"| user_id / workspace_id | `{self.user_id}` / `{self.workspace_id}` |", + f"| 최종 상태 | {status_icon} `{self.final_status}` |", + f"| 소요 시간 | {elapsed // 60}m {elapsed % 60}s (상한 {int(self.max_runtime) // 60}m) |", + f"| 확인(confirm)한 stage 수 | {self.stages_confirmed} |", + ] + if self.final_error: + lines += ["", f"**오류/중단 사유:** {self.final_error}"] + lines += ["", "## 이벤트 수신 통계", "", "| type | count |", "|------|-------|"] + for etype, count in sorted(self.event_counts.items()): + lines.append(f"| {etype} | {count} |") + summary = "\n".join(lines) + "\n" + + (self.output_dir / "summary.md").write_text(summary, encoding="utf-8") + step_summary = os.environ.get("GITHUB_STEP_SUMMARY") + if step_summary: + with open(step_summary, "a", encoding="utf-8") as f: + f.write(summary) + log(f"Summary written ({self.final_status}, {elapsed}s)") + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--yaml-path", required=True, help="실행할 Agent YAML 경로 (repo 상대/절대)") + p.add_argument("--api-base", default="https://legalpoc.eroomai.com/api") + p.add_argument("--user-id", default="jsahn") + p.add_argument("--workspace-id", default="e1d0baed-cec9-44af-9989-b7aab09bcfd1") + p.add_argument("--user-input", default="") + p.add_argument("--start-stage-index", type=int, default=0) + p.add_argument("--output-dir", default="agent_run_output") + p.add_argument("--read-timeout", type=float, default=1800.0, + help="SSE 이벤트 간 최대 대기 초 (초과 시 재연결)") + p.add_argument("--max-reconnects", type=int, default=5) + p.add_argument("--max-runtime", type=float, default=9000.0, + help="총 실행 시간 상한 초. 초과 시 세션 cancel 후 exit 1. " + "Gitea/act_runner 의 3h 태스크 상한 및 job timeout-minutes 보다 짧게 잡을 것") + return p.parse_args() + + +def _raise_interrupt(signum, frame): # noqa: ARG001 + raise KeyboardInterrupt + + +def main() -> None: + args = parse_args() + # job timeout/취소 시 SIGTERM 이 오므로 KeyboardInterrupt 경로로 합류시켜 세션을 정리한다 + signal.signal(signal.SIGTERM, _raise_interrupt) + runner = AgentRunner(args) + sys.exit(runner.run()) + + +if __name__ == "__main__": + main()