"""cycle_advancer PoC core (Analyzer + Proposer). task-2488 Phase B PoC. 회장-ChatGPT closed loop를 시스템화하기 위한 격리 PoC 코어. evidence → analyzer → root cause + 다음 task 제안 매핑 경로를 mock 환경에서 검증한다. 5단계 처리 로직 (task 명세 그대로 구현): 1. gap analysis (목표 vs 실제) 2. root cause 후보 N개 생성 (mock = 3개 stub) 3. 마아트(facts) + 외부 AI(strategy) 비평 시뮬레이션 (mock) 4. 합의/충돌 판정 5. 다음 task 제안서 draft 생성 YAML frontmatter 스키마 cycle_advancer/v1 ----------------------------------------- :: --- schema: cycle_advancer/v1 source_task_id: task-2485 proposed_task_id: task-2486 classification: MERGE_PENDING_DEPENDENCY proposal_only: true ready_for_dispatch: false chairman_required: false conflict_summary: null generated_at: 2026-05-08T00:00:00Z generator: cycle_advancer/v1-mock deterministic_seed: cycle_advancer-v1-mock --- 합격 조건: 동일 입력 + 동일 fixed-timestamp → 동일 출력 (파일 hash 일치). Forbidden (PoC 범위 외): - 실제 ``.done`` / ``.escalate`` / ``.fail`` 파일 생성 금지 - 실제 dispatch 호출 금지 - 외부 AI 호출 금지 (``mock_ai_adapter``만 사용) Packaging note: 이 모듈은 ``tools.poc.cycle_advancer`` 패키지의 코어이며, 패키지 ``__init__``에서 public symbol을 re-export 한다. """ from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path from typing import Any from .mock_ai_adapter import MockProposal, lookup_proposal from .output_writer import DraftPayload # ---------------------------------------------------------------------- # 분석 단계 결과 dataclass # ---------------------------------------------------------------------- @dataclass(frozen=True) class GapAnalysis: """단계 1: gap analysis 결과 (목표 vs 실제). Attributes: intended_outcome: 본래 의도한 결과 (예: ``MERGED_DONE``). actual_outcome: 실제 도달 상태 (예: ``MERGE_PENDING``). blocker_type: blocker 분류. blocker_description: blocker 설명. """ intended_outcome: str actual_outcome: str blocker_type: str blocker_description: str @dataclass(frozen=True) class CritiqueResult: """단계 3-4: 마아트(facts) + 외부 AI(strategy) 비평 결과. Attributes: facts_view: 마아트 관점 한 줄 요약. strategy_view: 외부 AI 관점 한 줄 요약. agreement: 합의 여부. conflict_summary: 충돌 시 사유 (합의 시 ``None``). """ facts_view: str strategy_view: str agreement: bool conflict_summary: str | None @dataclass(frozen=True) class AnalysisResult: """단계 5: 다음 task 제안서 draft에 필요한 모든 분석 결과. Attributes: gap: gap analysis. root_cause_candidates: root cause 후보 N개. critique: 마아트 + 외부 AI 비평 결과. proposal: mock LLM이 반환한 :class:`MockProposal`. """ gap: GapAnalysis root_cause_candidates: tuple[str, ...] critique: CritiqueResult proposal: MockProposal # ---------------------------------------------------------------------- # 코어 클래스 # ---------------------------------------------------------------------- class CycleAdvancer: """Analyzer + Proposer 코어. 이 클래스는 fixture로부터 evidence를 받아 5단계 분석을 수행하고 :class:`AnalysisResult`를 반환한다. ``DraftPayload`` 변환은 :meth:`build_draft_payload`에서 별도로 수행하여 timestamp 주입 책임을 호출자(CLI)에 위임한다. """ def analyze(self, evidence: dict[str, Any]) -> AnalysisResult: """5단계 분석 파이프라인을 실행한다. Args: evidence: fixture에서 로드한 source task evidence dict. Returns: :class:`AnalysisResult`. Raises: KeyError: evidence의 ``task_id``가 mock 매핑에 없는 경우. """ task_id = evidence["task_id"] # 1. gap analysis gap = self._gap_analysis(evidence) # 2. root cause 후보 N개 (mock = 3개 stub) — proposal에서 가져온다. proposal = lookup_proposal(task_id) root_cause_candidates = proposal.candidate_root_causes # 3. 마아트(facts) + 외부 AI(strategy) 비평 시뮬레이션 (mock) # 4. 합의/충돌 판정 critique = self._simulate_critique(proposal) # 5. 다음 task 제안서 draft 생성 — payload는 build_draft_payload에서. return AnalysisResult( gap=gap, root_cause_candidates=root_cause_candidates, critique=critique, proposal=proposal, ) def build_draft_payload( self, evidence: dict[str, Any], analysis: AnalysisResult, generated_at: str, ) -> DraftPayload: """분석 결과 + timestamp를 결합하여 :class:`DraftPayload`를 만든다. Args: evidence: fixture에서 로드한 evidence dict. analysis: :meth:`analyze` 결과. generated_at: deterministic timestamp (ISO8601 + ``Z``). Returns: :class:`DraftPayload`. """ proposal = analysis.proposal gap = analysis.gap merge_status = evidence.get("merge_status", {}) or {} pr_number = merge_status.get("pr_number") pr_state = merge_status.get("state") pr_url = merge_status.get("pr_url") evidence_summary: list[str] = [ f"source classification: {evidence.get('classification', 'UNKNOWN')}", f"PR: #{pr_number} ({pr_state}) {pr_url}", f"blocker: {gap.blocker_type} — {gap.blocker_description}", ] chain = evidence.get("post_resolution_chain") if chain: evidence_summary.append(f"post-resolution chain: {chain}") chairman_required = ( analysis.critique.agreement is False ) or proposal.chairman_required next_steps: list[str] = [ f"제안된 다음 task: {proposal.proposed_task_id} ({proposal.scope})", "본 draft는 proposal_only=true / ready_for_dispatch=false — 실제 dispatch 금지", ( f"체인 의존: {chain}" if chain else "체인 의존: (evidence에 정의 없음)" ), ] return DraftPayload( source_task_id=evidence["task_id"], proposed_task_id=proposal.proposed_task_id, classification=proposal.classification, chairman_required=chairman_required, conflict_summary=analysis.critique.conflict_summary, generated_at=generated_at, consensus_root_cause=proposal.consensus_root_cause, candidate_root_causes=analysis.root_cause_candidates, evidence_summary=tuple(evidence_summary), scope=proposal.scope, affected_files=proposal.affected_files, allowed_resources=proposal.allowed_resources, next_steps=tuple(next_steps), ) # ------------------------------------------------------------------ # 내부 단계 함수 # ------------------------------------------------------------------ @staticmethod def _gap_analysis(evidence: dict[str, Any]) -> GapAnalysis: """단계 1: 목표 vs 실제의 gap을 evidence에서 추출한다.""" merge_status = evidence.get("merge_status", {}) or {} merge_blocker = evidence.get("merge_blocker", {}) or {} merged_at = merge_status.get("merged_at") actual_outcome = "MERGED" if merged_at else "MERGE_PENDING" return GapAnalysis( intended_outcome="MERGED_DONE", actual_outcome=actual_outcome, blocker_type=merge_blocker.get("type", "unknown"), blocker_description=merge_blocker.get("description", ""), ) @staticmethod def _simulate_critique(proposal: MockProposal) -> CritiqueResult: """단계 3-4: mock 비평 시뮬레이션. deterministic 매핑이므로 ``proposal.chairman_required`` 값과 그에 대응하는 ``proposal.conflict_summary``를 그대로 critique 결과로 승격한다. 외부 AI 호출은 절대 수행하지 않는다. """ agreement = proposal.chairman_required is False return CritiqueResult( facts_view=( "마아트(facts): evidence의 blocker는 essence-fail 아닌 " "infrastructure/lifecycle 의존성" ), strategy_view=( "외부 AI(strategy): chain 의존을 해소하는 후속 task 분리가 " "최소 변경 + 최단 경로" ), agreement=agreement, conflict_summary=proposal.conflict_summary, ) # ---------------------------------------------------------------------- # fixture 로더 # ---------------------------------------------------------------------- def load_fixture(fixture_dir: Path, task_id: str) -> dict[str, Any]: """fixture 디렉토리에서 task evidence JSON을 로드한다. Args: fixture_dir: fixture 디렉토리 경로. task_id: 로드할 task_id (파일명은 ``{task_id}.json``). Returns: 파싱된 evidence dict. Raises: FileNotFoundError: fixture 파일이 없는 경우. """ target = fixture_dir / f"{task_id}.json" if not target.is_file(): raise FileNotFoundError( f"fixture not found: {target} " f"(PoC는 격리 fixture만 사용하며 실제 memory/events를 직접 읽지 않습니다)" ) with target.open("r", encoding="utf-8") as fh: return json.load(fh) __all__ = [ "CycleAdvancer", "AnalysisResult", "GapAnalysis", "CritiqueResult", "MockProposal", "DraftPayload", "lookup_proposal", "load_fixture", ]