AIForge - Claude Code 무인 자동화 워크플로우 시스템

┌─ FastAPI Web Server (:8010)
│  ├─ Routes (Dashboard, Projects, Schedules, Executions, Settings, Logs)
│  ├─ Jinja2 Templates (SSR 다크 테마 UI)
│  └─ Static Assets (CSS, Icons)
│
├─ APScheduler (AsyncIO)
│  ├─ System Jobs (토큰 체크, 로그 모니터링)
│  └─ Project Schedules (크론 기반 실행)
│
├─ Services Layer
│  ├─ Executor (async/sync 모드 Claude CLI 실행)
│  ├─ Log Checker (PID 모니터링, 로그 파싱)
│  └─ Scheduler (Job 등록 및 생명주기 관리)
│
└─ SQLite Database (WAL mode)
   ├─ projects
   ├─ schedules
   ├─ executions
   └─ settings

CREATE TABLE IF NOT EXISTS projects (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    type TEXT NOT NULL CHECK (type IN ('jira', 'schedule')),
    description TEXT,
    jira_search_conditions TEXT,  -- JIRA 타입일 때 검색 조건
    enabled INTEGER DEFAULT 1,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
 
CREATE TABLE IF NOT EXISTS schedules (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id INTEGER NOT NULL,
    name TEXT NOT NULL,
    prompt_template TEXT NOT NULL,  -- Claude에게 보낼 프롬프트
    work_dir TEXT NOT NULL,         -- Claude CLI 작업 디렉토리
    cron_expr TEXT NOT NULL,        -- 5-field 크론 표현식
    status TEXT DEFAULT 'idle'
        CHECK (status IN ('idle', 'running', 'paused', 'error')),
    FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE
);
 
CREATE TABLE IF NOT EXISTS executions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    schedule_id INTEGER NOT NULL,
    pid INTEGER,               -- OS 프로세스 ID
    status TEXT DEFAULT 'running'
        CHECK (status IN ('running', 'success', 'error', 'timeout', 'killed')),
    command TEXT,               -- 실행된 CLI 명령어
    issue_key TEXT,             -- JIRA 이슈 키 (있는 경우)
    log_path TEXT,              -- Claude 로컬 로그 경로
    duration_seconds INTEGER,
    FOREIGN KEY (schedule_id) REFERENCES schedules(id) ON DELETE CASCADE
);

async def execute_async(
    prompt: str,
    work_dir: str,
    issue_key: str | None = None
) -> dict:
    """Fire-and-forget 실행. PID 기록 후 즉시 반환."""
    env = os.environ.copy()
    env.pop("CLAUDECODE", None)  # 재귀 실행 방지
 
    cmd = [
        "claude", "--chrome",
        "--dangerously-skip-permissions",
        "-p", prompt
    ]
 
    process = await asyncio.create_subprocess_exec(
        *cmd,
        cwd=work_dir,
        env=env,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
 
    return {"pid": process.pid, "status": "running"}

async def execute_sync(
    prompt: str,
    work_dir: str,
    timeout: int = 180
) -> dict:
    """동기 실행. 결과를 JSON으로 파싱하여 반환."""
    # ... subprocess 실행 ...
 
    stdout, stderr = await asyncio.wait_for(
        process.communicate(), timeout=timeout
    )
 
    # 3단계 JSON 파싱 폴백
    result = stdout.decode()
    try:
        return json.loads(result)           # 1. 순수 JSON
    except json.JSONDecodeError:
        # 2. ```json ... ``` 마크다운 블록에서 추출
        match = re.search(r'```json\s*(.*?)\s*```', result, re.DOTALL)
        if match:
            return json.loads(match.group(1))
        # 3. { ... } 객체 리터럴에서 추출
        match = re.search(r'\{.*\}', result, re.DOTALL)
        if match:
            return json.loads(match.group(0))

1. 크론 트리거 → 스케줄 실행
2. 토큰 한도 체크
3. Claude 동기 실행: "다음 조건에 해당하는 JIRA 검색 해줘 {conditions}"
4. JSON 파싱 → 이슈 키 목록 추출
5. 이슈별로:
   - 이미 실행 중인지 확인
   - prompt_template에서 {issue_key} 치환
   - Claude CLI 비동기 실행
   - PID 기록
6. 백그라운드 로그 체커가 완료 모니터링

poll_prompt = (
    f"다음 조건에 해당하는 JIRA 검색 해줘 {conditions} "
    "조회해서 json 으로 알려줘 출력에 다른값은 넣지 않고 json 만 넣어줘"
)

[{"key": "FAS-130"}, {"key": "FAS-131"}]

1. 시스템 잡 (기본 60분 간격)
2. Claude 동기 실행: "claude.ai/settings/usage 접속해서 사용량 확인"
3. 세션 사용률 / 주간 사용률 파싱
4. settings 테이블에 상태 저장
5. 한도 초과 시 → 모든 프로젝트 자동 일시정지

token_check_config = {
    "enabled": True,
    "interval_minutes": 60,
    "session_limit_percent": 80,   # 세션 사용량 80% 이상이면 정지
    "weekly_limit_percent": 90     # 주간 사용량 90% 이상이면 정지
}

async def check_running_executions():
    """실행 중인 모든 작업의 상태를 확인한다."""
    running = await get_running_executions()
 
    for execution in running:
        pid = execution["pid"]
 
        # PID가 아직 살아있는지 확인
        if not psutil.pid_exists(pid):
            # 프로세스가 종료됨 → 로그 파일 분석
            log_content = await read_log_file(execution["log_path"])
 
            if has_error_pattern(log_content):
                await update_execution_status(execution["id"], "error")
            else:
                await update_execution_status(execution["id"], "success")
 
            # 실행 시간 계산 및 스케줄 상태 업데이트
            await update_schedule_status(execution["schedule_id"], "idle")

# Config 키 - 사용자가 변경
CONFIG_KEYS = ["token_check_config", "log_monitor_config", "global"]
 
# Status 키 - 시스템이 업데이트
STATUS_KEYS = ["token_check_status", "log_monitor_status"]

run.py (uvicorn 엔트리)
  ↓
app.main:lifespan (FastAPI lifespan)
  ├─ init_db()
  │  ├─ schema.sql 읽기
  │  ├─ 테이블 생성 (없으면)
  │  └─ 기본 설정값 삽입
  │
  └─ init_scheduler()
     ├─ _register_system_jobs()
     │  ├─ 로그 모니터 (interval)
     │  └─ 토큰 체크 (interval)
     │
     └─ _register_project_schedules()
        └─ DB에서 활성 스케줄 로드
           └─ APScheduler 크론 잡으로 등록

aiforge/
├── run.py                    # uvicorn 엔트리포인트
├── schema.sql                # DB 스키마
├── requirements.txt          # Python 의존성
│
├── app/
│   ├── main.py               # FastAPI 앱, lifespan, 라우트 등록
│   ├── database.py           # SQLite 헬퍼, 설정 관리
│   │
│   ├── services/
│   │   ├── scheduler.py      # APScheduler, 실행 로직 (352 LOC)
│   │   ├── executor.py       # Claude CLI 서브프로세스 관리 (200 LOC)
│   │   └── log_checker.py    # PID 모니터링, 로그 파싱 (124 LOC)
│   │
│   ├── routes/
│   │   ├── dashboard.py      # 대시보드 통계
│   │   ├── projects.py       # 프로젝트 CRUD
│   │   ├── schedules.py      # 스케줄 CRUD + 수동 실행
│   │   ├── executions.py     # 실행 이력
│   │   ├── settings.py       # 설정 관리
│   │   ├── logs.py           # 로그 리포트
│   │   └── templates.py      # 워크스페이스 템플릿
│   │
│   └── templates/            # Jinja2 HTML 템플릿 (11개)
│
└── docs/
    ├── ARCHITECTURE.md
    └── PROJECT_ANALYSIS.md

# 의존성 설치
pip install -r requirements.txt
 
# 실행
python run.py
# 또는
uvicorn app.main:app --host 0.0.0.0 --port 8010 --reload