zioinfo-mail/workspace/guardia-docs/10_확장개발_Priority2_코어모듈.md

GUARDiA ITSM — Priority 2: 코어 모듈 구현

문서 버전: 1.0 | 작성일: 2026-05-25

---

1. core/vibe_bridge.py — Claude CLI SDK 연동

목적

subprocess 방식 대신 Python SDK를 직접 연동하여 Claude CLI 세션을 관리한다.

주요 클래스

class VibeBridge:
    """Claude CLI SDK 비동기 브리지"""

    async def start_session(self, sr_id: str, project_path: str) -> str:
        """새 Claude 세션 시작 → session_id 반환"""

    async def send_message(self, session_id: str, message: str) -> str:
        """세션에 메시지 전송 → 응답 반환"""

    async def get_session_status(self, session_id: str) -> dict:
        """세션 상태 조회 {active, last_response, tokens_used}"""

    async def close_session(self, session_id: str) -> bool:
        """세션 종료"""

    async def resume_session(self, session_id: str, message: str) -> str:
        """기존 세션 재개"""

환경 변수

CLAUDE_CLI_PATH=/usr/local/bin/claude
CLAUDE_WORKSPACE_ROOT=/opt/guardia/workspaces
CLAUDE_SESSION_TIMEOUT=3600  # 세션 타임아웃 (초)

사용 예시

bridge = VibeBridge()
session_id = await bridge.start_session(
    sr_id="SR-20260525-000001",
    project_path="/opt/src/myproject"
)
response = await bridge.send_message(session_id, "버그를 수정해주세요")

---

2. core/deploy_pipeline.py — 배포 파이프라인 오케스트레이터

목적

빌드 → 테스트 → 배포 → 헬스체크 → ITSM 콜백의 전체 파이프라인을 관리한다.

파이프라인 단계

단계	함수	설명
pre_check	_pre_check()	배포 환경 사전 점검 (서버 연결, 디스크 용량)
build	_build()	tb_project.build_cmd 실행
test	_test()	tb_project.test_cmd 실행, 실패 시 중단
backup	_backup()	현재 배포본 백업 (rollback 대비)
deploy	_deploy()	SSH로 파일 전송 → deploy_path
restart	_restart()	was_restart_cmd 실행
health_check	_health_check()	HTTP GET → health_check_url (10회 재시도)
notify	_notify()	WorkLog 등록 + 메신저 알림

클래스 구조

class DeployPipeline:
    async def run(self, session_id: int) -> PipelineResult:
        """전체 파이프라인 실행"""

    async def rollback(self, session_id: int) -> bool:
        """이전 버전으로 롤백"""

    async def get_status(self, session_id: int) -> PipelineStatus:
        """현재 진행 단계 조회"""

오류 처리

• 각 단계 실패 시 tb_vibe_session.error_msg 업데이트
• test 단계 실패 → 배포 중단, SR 상태 = FAILED_ROLLBACK
• health_check 10회 실패 → 자동 rollback 실행

---

3. 배치 잡 동적 APScheduler 등록

목적

tb_batch_job 테이블의 cron_expr을 APScheduler에 동적으로 등록/제거한다.

구현 위치

core/scheduler.py 확장

API 연동 흐름

POST /api/batch/jobs/{id}/enable
    → scheduler.add_job(
        run_batch_job,
        CronTrigger.from_crontab(job.cron_expr),
        id=f"batch_{job.id}",
        args=[job.id],
        replace_existing=True
      )

POST /api/batch/jobs/{id}/disable
    → scheduler.remove_job(f"batch_{job.id}")

서버 시작 시 (lifespan):
    → 모든 is_active=True 배치 잡 자동 등록

run_batch_job 함수

async def run_batch_job(job_id: int):
    async with SessionLocal() as db:
        job = await db.get(BatchJob, job_id)
        run = BatchRun(job_id=job_id, status="RUNNING", started_at=datetime.utcnow())
        db.add(run)
        await db.commit()

        try:
            result = await execute_ssh_command(job.server_id, job.command, job.timeout_sec)
            run.status = "SUCCESS" if result.exit_code == 0 else "FAILED"
            run.stdout = result.stdout[-5000:]  # 마지막 5000자
            run.exit_code = result.exit_code
        except asyncio.TimeoutError:
            run.status = "TIMEOUT"
        finally:
            run.completed_at = datetime.utcnow()
            await db.commit()

        # alert_on_fail 처리
        if run.status != "SUCCESS" and job.alert_on_fail:
            await create_sr_for_batch_failure(job, run, db)

---

4. SM 스크립트 실행 이력 UI

목적

SSH 명령 실행 결과를 tb_work_log 기반으로 조회하여 index.html에 표시한다.

API

GET /api/work-logs?sr_id={id}&work_type=SSH_EXEC

index.html 추가 뷰

• 스크립트 관리 뷰에 "실행 이력" 탭 추가
• 테이블: 서버명, 스크립트명, 실행시각, 결과(PASS/FAIL), 소요시간
• 행 클릭 시 stdout/stderr 상세 모달