4장: 장기 실행 워크플로우 관리

from temporalio import workflow, activity
from datetime import timedelta
 
@workflow.defn
class OnboardingWorkflow:
    @workflow.run
    async def run(self, application: Application) -> OnboardingResult:
        # 단계 1: 서류 검증 (자동)
        verification = await workflow.execute_activity(
            verify_documents,
            application.documents,
            start_to_close_timeout=timedelta(minutes=5),
        )
 
        if not verification.passed:
            # 추가 서류 요청 후 대기 (최대 72시간)
            await workflow.execute_activity(
                request_additional_docs,
                application.customer_id,
                start_to_close_timeout=timedelta(minutes=1),
            )
 
            # 고객의 서류 제출을 대기 (시그널)
            additional_docs = await workflow.wait_condition(
                lambda: self.additional_docs_received,
                timeout=timedelta(hours=72),
            )
 
            if not additional_docs:
                return OnboardingResult(status="timeout", reason="서류 미제출")
 
        # 단계 2: 심사 승인 대기 (사람)
        approval_result = await workflow.execute_activity(
            request_approval,
            ApprovalRequest(
                customer_id=application.customer_id,
                verification=verification,
            ),
            start_to_close_timeout=timedelta(hours=24),
        )
 
        if not approval_result.approved:
            return OnboardingResult(status="rejected", reason=approval_result.reason)
 
        # 단계 3: 계정 생성 및 알림 (자동)
        account = await workflow.execute_activity(
            create_account,
            application,
            start_to_close_timeout=timedelta(minutes=2),
        )
 
        await workflow.execute_activity(
            send_welcome_email,
            account,
            start_to_close_timeout=timedelta(seconds=30),
        )
 
        return OnboardingResult(status="completed", account_id=account.id)
 
    # 시그널 핸들러: 외부에서 서류 제출 이벤트 수신
    @workflow.signal
    async def on_docs_submitted(self, docs: list[Document]):
        self.additional_docs_received = True
        self.submitted_docs = docs

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver
 
# PostgreSQL 체크포인터 설정
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/workflows"
)
 
# 그래프 컴파일 시 체크포인터 연결
graph = StateGraph(WorkflowState)
# ... 노드와 엣지 추가 ...
workflow = graph.compile(checkpointer=checkpointer)
 
# 실행 시 thread_id로 워크플로우 인스턴스를 식별
config = {"configurable": {"thread_id": "onboarding-12345"}}
 
# 첫 실행
result = await workflow.ainvoke(initial_state, config)
 
# 중단된 워크플로우 재개 (동일한 thread_id 사용)
result = await workflow.ainvoke(
    {"approval_status": "approved"},  # 새로운 입력
    config,
)

from pydantic import BaseModel
from datetime import datetime
import json
 
class WorkflowSnapshot(BaseModel):
    """워크플로우 상태 스냅샷"""
    workflow_id: str
    version: int
    current_node: str
    state_data: dict  # 직렬화 가능한 데이터만
    created_at: datetime
    metadata: dict
 
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat(),
        }
 
class StateManager:
    def __init__(self, store: StateStore):
        self.store = store
 
    async def save_checkpoint(
        self,
        workflow_id: str,
        node: str,
        state: dict,
    ) -> None:
        """상태 체크포인트 저장"""
        # 대용량 데이터는 참조로 저장
        processed_state = self._externalize_large_values(state)
 
        snapshot = WorkflowSnapshot(
            workflow_id=workflow_id,
            version=await self.store.next_version(workflow_id),
            current_node=node,
            state_data=processed_state,
            created_at=datetime.utcnow(),
            metadata={"serializer": "json", "compressed": False},
        )
 
        await self.store.save(snapshot)
 
    def _externalize_large_values(self, state: dict) -> dict:
        """100KB 이상의 값은 외부 저장소로 이동"""
        result = {}
        for key, value in state.items():
            serialized = json.dumps(value, default=str)
            if len(serialized) > 100_000:
                ref = self.store.save_blob(serialized)
                result[key] = {"__ref__": ref}
            else:
                result[key] = value
        return result

class WorkflowController:
    def __init__(self, runtime: WorkflowRuntime):
        self.runtime = runtime
 
    async def pause(self, workflow_id: str, reason: str) -> None:
        """워크플로우 수동 일시 정지"""
        await self.runtime.signal(
            workflow_id,
            signal_name="pause",
            payload={"reason": reason, "paused_by": get_current_user()},
        )
 
    async def resume(self, workflow_id: str, input_data: dict = None) -> None:
        """워크플로우 재개"""
        await self.runtime.signal(
            workflow_id,
            signal_name="resume",
            payload=input_data or {},
        )
 
    async def get_status(self, workflow_id: str) -> WorkflowStatus:
        """워크플로우 현재 상태 조회"""
        state = await self.runtime.get_state(workflow_id)
        return WorkflowStatus(
            id=workflow_id,
            current_node=state.current_node,
            is_paused=state.is_paused,
            paused_since=state.paused_at,
            waiting_for=state.pending_signals,
            progress=self._calculate_progress(state),
        )

from datetime import timedelta
 
@dataclass
class TimeoutConfig:
    """다층 타임아웃 설정"""
    workflow_timeout: timedelta = timedelta(days=7)
    stage_timeouts: dict[str, timedelta] = None
    activity_timeout: timedelta = timedelta(minutes=5)
    llm_call_timeout: timedelta = timedelta(seconds=60)
    api_call_timeout: timedelta = timedelta(seconds=30)
 
    def __post_init__(self):
        if self.stage_timeouts is None:
            self.stage_timeouts = {}
 
    def get_stage_timeout(self, stage: str) -> timedelta:
        return self.stage_timeouts.get(stage, timedelta(hours=4))
 
# 온보딩 워크플로우 타임아웃 설정
onboarding_timeouts = TimeoutConfig(
    workflow_timeout=timedelta(days=14),
    stage_timeouts={
        "document_verification": timedelta(hours=1),
        "human_approval": timedelta(hours=24),
        "customer_document_wait": timedelta(days=5),
        "account_creation": timedelta(minutes=10),
    },
    activity_timeout=timedelta(minutes=5),
)

@dataclass
class ProgressUpdate:
    workflow_id: str
    current_stage: str
    total_stages: int
    completed_stages: int
    estimated_completion: datetime | None
    status_message: str
    details: dict
 
class ProgressTracker:
    def __init__(self, notifier: Notifier):
        self.notifier = notifier
 
    async def update(self, progress: ProgressUpdate) -> None:
        """진행 상태 업데이트 및 알림"""
        await self.store.save_progress(progress)
 
        # 대시보드 실시간 업데이트
        await self.notifier.push_update(
            channel=f"workflow:{progress.workflow_id}",
            data=progress,
        )
 
        # 주요 단계 전환 시 이해관계자 알림
        if self._is_milestone(progress):
            await self.notifier.notify_stakeholders(
                workflow_id=progress.workflow_id,
                message=f"[{progress.current_stage}] {progress.status_message}",
            )