"""위저드 상태 — `WizardState`·`Prompt`·`Option`·`Step` 데이터 타입, v1 provider 선택의 v2 변환, 상태 술어.""" from __future__ import annotations from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Callable, Optional from okstra_ctl.registry.provider_registry import default_provider_registry from okstra_ctl.model_pool import ModelPool from okstra_ctl.role_requirements import ( RoleProfile, RoleProfileError, RoleRequirement, load_role_profile, ) from okstra_ctl.ids import slugify_task_segment from .ids import ( PICK_TYPE_CUSTOM, S_BASE_REF_PICK, S_HANDOFF_STAGE_PICK, _ABORT_OPTION, _SLUG_OK, _STAGE_SCOPED_TASK_TYPES, ) # ---- Data types ---------------------------------------------------------- @dataclass class WizardState: # execution identity v2: role-first canonical model selections execution_identity_version: int = 1 role_counts: dict[str, int] = field(default_factory=dict) role_models: dict[str, list[str]] = field(default_factory=dict) role_selection_order: list[str] = field(default_factory=list) # bootstrap workspace_root: str = "" project_root: str = "" project_id: str = "" preferred_task_key: str = "" host_runtime: str = "claude-code" host_entry_mode: str = "current-session" available_functions: list[str] = field(default_factory=list) picker_offsets: dict[str, int] = field(default_factory=dict) # task identity is_new_task: Optional[bool] = None task_group: str = "" task_id: str = "" existing_brief_path: str = "" # brief-derived suggestions (new-task flow only; set when brief is # accepted, cleared if the user picks "type custom" so the next # `_build_*` falls back to plain text input) task_group_suggestion: str = "" task_id_suggestion: str = "" task_group_pending_text: bool = False task_id_pending_text: bool = False # task-type + dependents task_type: str = "" profile_workers: list[str] = field(default_factory=list) profile_optional_workers: list[str] = field(default_factory=list) # brief keep_existing_brief: Optional[bool] = None brief_path: str = "" brief_path_pending_text: bool = False # analysis evidence and target project_evidence_path: str = "" project_evidence_pending_text: bool = False feature_evidence_path: str = "" feature_evidence_pending_text: bool = False analysis_target: str = "" analysis_target_pending_text: bool = False # worktree reuse_worktree: Optional[bool] = None base_ref: str = "" base_ref_pending_text: bool = False # impl extras approved_plan_path: str = "" approved_plan_pending_text: bool = False # A plan that is approvable (gate ok, no blockers) but not yet `approved`. # Set when the user selects such a plan; the approve-confirm step reads it. approve_plan_candidate: str = "" # HTML Plan Approval 위젯이 남긴 sidecar 감지 결과. 선택된 plan 과 # source-report·seq 가 일치하는 APPROVAL 블록이 있을 때만 채워지고, # approve-confirm 단계가 3-옵션(yes_apply/yes/no)으로 확장된다. html_approval_sidecar: str = "" html_approval_option: str = "" design_prep_queue: list[str] = field(default_factory=list) design_prep_current: str = "" design_prep_decision: str = "" design_prep_overrides_json: str = "" design_prep_notes: str = "" selected_stage: str = "auto" selected_stages: str = "" # implementation 다중선택 위상정렬 CSV executor: str = "" critic: str = "" critic_pending_text: bool = False # release-handoff: PR 로 내보낼 범위. mode 는 stages 선택의 파생값이다 — # whole-task = 빈 stages(whole-task 검증 기반 단일 PR), stage-group = csv. handoff_mode: str = "" # "" | "whole-task" | "stage-group" handoff_stages: str = "" # csv ("2,3"), whole-task 면 "" # resume: 직전 run-inputs 재사용 여부 (None=미응답, True=재사용, False=재입력) reuse_previous: Optional[bool] = None # customize use_defaults: Optional[bool] = None workers_override: str = "" workers_custom_pending: bool = False lead_provider: str = "" lead_model: str = "" claude_model: str = "" codex_model: str = "" antigravity_model: str = "" grok_model: str = "" kimi_model: str = "" report_writer_provider: str = "" report_writer_model: str = "" directive: str = "" directive_pending_text: bool = False last_directive_cached: str = "" related_tasks_raw: str = "" related_tasks_pending_text: bool = False last_siblings_cached: str = "" clarification_response_path: str = "" clarification_pending_text: bool = False last_final_report_cached: str = "" selected_direction_path: str = "" # "" | "auto" | "full" | "" — 사용자가 고른 이번 재실행의 재검증 # 범위. implementation-planning 재실행에서 좁힐 여지가 있을 때만 채워진다. reverify_scope: str = "" reverify_scope_pending_text: bool = False pr_template_path: str = "" pr_template_pending_text: bool = False pr_template_scope: str = "" # "once" | "project" | "global" last_pr_template_cached: str = "" # confirm / edit # "" | "yes" | "no" — done(release-handoff) task 재진입의 fix-cycle 기록 여부 fix_cycle: str = "" confirmed: Optional[bool] = None confirmation_stages: str = "" confirmation_prompt: str = "" confirmation_scope: dict[str, Any] = field(default_factory=dict) user_authorization: dict[str, Any] = field(default_factory=dict) edit_target: str = "" # terminal: user picked 중단 — no further prompt ever applies aborted: bool = False # bookkeeping answered: list[str] = field(default_factory=list) def to_json(self) -> dict[str, Any]: payload = asdict(self) payload.pop("execution_identity_version") payload["executionIdentityVersion"] = 2 payload["roleCounts"] = payload.pop("role_counts") payload["roleModels"] = payload.pop("role_models") payload["roleSelectionOrder"] = payload.pop("role_selection_order") for field_name in _V1_PROVIDER_STATE_FIELDS: payload.pop(field_name, None) return payload @classmethod def from_json(cls, d: dict[str, Any]) -> "WizardState": return _wizard_state_from_json(d) @dataclass class Option: value: str label: str description: str = "" # 이 run 이 계산한 추천. 표시는 이 한 곳에서만 나온다 — 종전에는 추천이 # 순서로만 암시됐고(첫 항목), 리드는 그와 별개로 산문에서 자기 추천을 # 적으라는 지시를 받았다. 둘이 갈리면 사용자는 서로 다른 두 추천을 동시에 # 본다(실측: 계획 run 의 clarification 단계 — 리드는 `직접 입력` 을 권했고 # 화면 첫 줄은 위저드의 캐시된 리포트였다). recommended: bool = False @dataclass class Prompt: step: str kind: str # "pick" | "text" | "pick_group" | "done" | "aborted" label: str = "" options: list[Option] = field(default_factory=list) help: str = "" echo_template: str = "" # e.g. "task-group: {value}" multi: bool = False # only meaningful when kind == "pick" # only meaningful when kind == "pick_group": one entry per AskUserQuestion tab questions: list["Prompt"] = field(default_factory=list) def __post_init__(self) -> None: """자유 입력 선택지는 언제나 마지막이다. 목록이 자유 입력으로 열리면 질문을 통째로 사용자에게 되돌려준 것이다 — 앞선 선택지들이 이 run 이 이미 알아낸 것이고, 자유 입력은 그것들이 전부 맞지 않을 때의 탈출구다. 실측: 재검증 범위 질문의 `unlinked` 분기가 `직접 입력` 을 첫 선택지로 내놓았고, 리드가 preamble 에 해당 stage 번호를 이미 적어 놓은 화면에서도 사용자는 그 번호를 다시 타이핑해야 했다. 중단(abort)만 그 뒤에 올 수 있다. 중단은 탈출구가 아니라 취소이고, `brief_carry` 가 실제로 그 모양이다 — 전환(추천) / 직접 입력 / 중단. 배치 지점이 12곳이라 관례로는 지켜지지 않는다. 값을 만드는 자리에서 막는다. """ values = [option.value for option in self.options] if PICK_TYPE_CUSTOM in values: after = values[values.index(PICK_TYPE_CUSTOM) + 1:] offenders = [value for value in after if value != _ABORT_OPTION] if offenders: raise WizardError( f"wizard step {self.step!r}: the free-input option " f"({PICK_TYPE_CUSTOM!r}) must come after every real choice — " f"only {_ABORT_OPTION!r} may follow it, but {offenders} do" ) # 추천 불변식은 탈출구가 없는 목록에도 적용된다. self._check_recommendations() def _check_recommendations(self) -> None: """단일 추천은 하나다. 모델 선택은 제공자 순서, 그 외에는 추천이 앞이다. 실측(2026-09-09, task 선택 화면): 남은 task 세 줄이 전부 `(추천)` 을 달고 나왔고, 리드는 산문에서 2번을 권했다. 추천이 여럿이면 라벨은 아무것도 고르지 않은 것이고, 추천이 1번이 아니면 사용자는 목록을 끝까지 읽어야 추천을 찾는다. 모델 선택은 제공자별 묶음을 우선한다. 그 외 체크박스(`multi`)는 추천이 여럿일 수 있되 앞머리에 모인다. 그리고 `직접 입력` / `중단` 은 앞의 선택지가 전부 맞지 않을 때의 탈출구이므로 추천 대상이 될 수 없다. """ flags = [option.recommended for option in self.options] grouped_models = self.step.startswith(("role-models:", "role-model:")) if any(flags): if self.multi: if not grouped_models and any(flags[index] for index in range(1, len(flags)) if not flags[index - 1]): raise WizardError( f"wizard step {self.step!r}: recommended options must " "be the leading run of the list" ) elif sum(flags) != 1 or (not grouped_models and not flags[0]): raise WizardError( f"wizard step {self.step!r}: a single-select step carries " "exactly one recommendation and it is the first option" ) escapes = {PICK_TYPE_CUSTOM, _ABORT_OPTION} marked = [ option.value for option in self.options if option.recommended and option.value in escapes ] if marked: raise WizardError( f"wizard step {self.step!r}: {marked} is an escape hatch, " "not a recommendation" ) @property def id(self) -> str: """Stable public identifier shared by static and role-instance prompts.""" return self.step def to_json(self) -> dict[str, Any]: out = { "id": self.step, "step": self.step, "kind": self.kind, "label": self.label, "options": [asdict(o) for o in self.options], "help": self.help, "echoTemplate": self.echo_template, "multi": self.multi, } if self.kind == "pick_group": out["questions"] = [ {"step": q.step, "label": q.label, "options": [asdict(o) for o in q.options], "multi": q.multi} for q in self.questions ] return out class WizardError(Exception): """validation failure surfaced to user.""" def _slug_or_die(value: str, field_name: str) -> str: slug = slugify_task_segment(value or "") if not slug or not _SLUG_OK.search(slug): raise WizardError( f"{field_name} must contain at least one alphanumeric character " f"(got: {value!r})" ) return slug # ---- Roster / profile helpers ------------------------------------------- def _profile_path(workspace_root: Path, task_type: str) -> Path: return workspace_root / "prompts" / "profiles" / f"{task_type}.md" _V2_STATE_FIELDS = ( "roleCounts", "roleModels", "roleSelectionOrder", ) _V1_PROVIDER_STATE_FIELDS = ( "workers_override", "lead_provider", "lead_model", "claude_model", "codex_model", "antigravity_model", "grok_model", "kimi_model", "report_writer_provider", "report_writer_model", "executor", "critic", ) _V1_PROVIDER_STATE_JSON_FIELDS = frozenset({ *_V1_PROVIDER_STATE_FIELDS, "workers", "leadProvider", "leadModel", "claudeModel", "codexModel", "antigravityModel", "grokModel", "kimiModel", "reportWriterProvider", "reportWriterModel", }) def _load_role_profile_for_state(state: WizardState) -> RoleProfile: try: return load_role_profile( _profile_path(Path(state.workspace_root), state.task_type) ) except RoleProfileError as exc: raise WizardError(str(exc)) from exc def _role_selection_enabled(state: WizardState) -> bool: if ( state.execution_identity_version != 2 or not state.workspace_root or not state.task_type ): return False try: load_role_profile(_profile_path(Path(state.workspace_root), state.task_type)) except (OSError, RoleProfileError): return False return True def _legacy_value(payload: dict[str, Any], snake: str, camel: str) -> Any: return payload.get(snake, payload.get(camel, "")) def _legacy_model_ref( pool: ModelPool, provider: str, role: str, raw_model: Any, ) -> str | None: provider_id = provider.strip().lower() if not provider_id: return None model_value = raw_model.strip() if isinstance(raw_model, str) else "" try: model = ( pool.resolve_alias(provider_id, model_value) if model_value else pool.default_candidate(provider_id, role) ) except (ValueError, RuntimeError): return None if model is None: return None availability = pool.availability(str(model.model_ref), role) return str(model.model_ref) if availability.available else None def _legacy_worker_providers(payload: dict[str, Any]) -> list[str]: raw = _legacy_value(payload, "workers_override", "workers") if isinstance(raw, list): values = raw elif isinstance(raw, str): values = raw.split(",") else: values = [] return [ value.strip().lower() for value in values if isinstance(value, str) and value.strip() and value.strip().lower() != "report-writer" ] def _initial_cross_verification_requirement( profile: RoleProfile, ) -> RoleRequirement | None: return next(( requirement for requirement in profile.roles if not requirement.dynamic and requirement.role in { "analyser", "designer", "planner", "verifier", } ), None) def _convert_v1_provider_selections( state: WizardState, payload: dict[str, Any], ) -> None: if not _role_selection_enabled(state): return profile = _load_role_profile_for_state(state) pool = ModelPool.from_registry(default_provider_registry()) cross_requirement = _initial_cross_verification_requirement(profile) providers = _legacy_worker_providers(payload) if cross_requirement is not None: selected_count = cross_requirement.recommended_count if providers and cross_requirement.min_count < cross_requirement.max_count: selected_count = min( cross_requirement.max_count, max(cross_requirement.min_count, len(providers)), ) if cross_requirement.min_count < cross_requirement.max_count: state.role_counts[cross_requirement.role] = selected_count if not providers: providers = [ model.provider_id for model in pool.default_candidates(cross_requirement.role) ] for ordinal, provider in enumerate(providers[:selected_count], start=1): raw_model = _legacy_value( payload, f"{provider}_model", f"{provider}Model", ) model_ref = _legacy_model_ref( pool, provider, cross_requirement.role, raw_model, ) if model_ref is None: break state.role_models.setdefault(cross_requirement.role, []).append(model_ref) if cross_requirement.role in state.role_models: # 여러 인스턴스 역할의 답은 체크박스 한 장이다. state.role_selection_order.append(f"role-models:{cross_requirement.role}") provider_rows = ( ( "leader", _legacy_value(payload, "lead_provider", "leadProvider") or "claude", _legacy_value(payload, "lead_model", "leadModel"), ), ( "report-writer", _legacy_value( payload, "report_writer_provider", "reportWriterProvider", ) or "claude", _legacy_value( payload, "report_writer_model", "reportWriterModel", ), ), ( "implementer", _legacy_value(payload, "executor", "executor") or "claude", _legacy_value( payload, f"{_legacy_value(payload, 'executor', 'executor') or 'claude'}_model", f"{_legacy_value(payload, 'executor', 'executor') or 'claude'}Model", ), ), ( "critic", _legacy_value(payload, "critic", "critic"), _legacy_value( payload, f"{_legacy_value(payload, 'critic', 'critic')}_model", f"{_legacy_value(payload, 'critic', 'critic')}Model", ), ), ) static_roles = { requirement.role for requirement in profile.roles if not requirement.dynamic and requirement.min_count > 0 } for role, provider, raw_model in provider_rows: if role not in static_roles or not isinstance(provider, str) or not provider: continue model_ref = _legacy_model_ref(pool, provider, role, raw_model) if model_ref is None: continue state.role_models[role] = [model_ref] state.role_selection_order.append(f"role-model:{role}:1") def _wizard_state_from_json(payload: dict[str, Any]) -> WizardState: if not isinstance(payload, dict): raise WizardError("wizard state must be a JSON object") data = dict(payload) version = data.pop("executionIdentityVersion", None) has_v2_fields = any(field_name in data for field_name in _V2_STATE_FIELDS) has_v1_fields = any( field_name in data for field_name in _V1_PROVIDER_STATE_JSON_FIELDS ) if version == 2: if has_v1_fields or not all(field_name in data for field_name in _V2_STATE_FIELDS): raise WizardError("mixed wizard state versions are not allowed") data["role_counts"] = data.pop("roleCounts") data["role_models"] = data.pop("roleModels") data["role_selection_order"] = data.pop("roleSelectionOrder") data["execution_identity_version"] = 2 _validate_v2_state_fields(data) elif version in (None, 1): if has_v2_fields: raise WizardError("mixed wizard state versions are not allowed") data.pop("execution_identity_version", None) else: raise WizardError(f"unsupported wizard state version: {version!r}") allowed = WizardState.__dataclass_fields__.keys() state = WizardState(**{key: value for key, value in data.items() if key in allowed}) if version != 2: state.execution_identity_version = 2 _convert_v1_provider_selections(state, payload) _discard_implicit_leader_selection(state) return state def _validate_v2_state_fields(data: dict[str, Any]) -> None: role_counts = data.get("role_counts") role_models = data.get("role_models") selection_order = data.get("role_selection_order") if not isinstance(role_counts, dict) or not all( isinstance(role, str) and isinstance(count, int) and not isinstance(count, bool) for role, count in role_counts.items() ): raise WizardError("wizard state roleCounts must map roles to integers") if not isinstance(role_models, dict) or not all( isinstance(role, str) and isinstance(models, list) and all(isinstance(model_ref, str) for model_ref in models) for role, models in role_models.items() ): raise WizardError("wizard state roleModels must map roles to modelRef arrays") if not isinstance(selection_order, list) or not all( isinstance(step_id, str) for step_id in selection_order ): raise WizardError("wizard state roleSelectionOrder must be a string array") def _is_role_selection_step(step_id: str) -> bool: return step_id.startswith(("role-model:", "role-models:")) def _discard_implicit_leader_selection(state: WizardState) -> None: """호스트에 고정된 리더의 구형 선택값을 상태에서 제거한다.""" state.role_models.pop("leader", None) leader_steps = { step_id for step_id in state.role_selection_order if step_id.startswith("role-model:leader:") } state.role_selection_order = [ step_id for step_id in state.role_selection_order if step_id not in leader_steps ] state.answered = [ step_id for step_id in state.answered if step_id not in leader_steps ] def _base_ref_required(state: WizardState) -> bool: return state.task_type != "final-verification" and state.reuse_worktree is False def _base_ref_ready(state: WizardState) -> bool: return not _base_ref_required(state) or S_BASE_REF_PICK in state.answered # ---- Step descriptors --------------------------------------------------- @dataclass class Step: id: str applies: Callable[[WizardState], bool] build: Callable[[WizardState], Prompt] submit: Callable[[WizardState, str], Optional[str]] # Field names this step owns; resetting the step clears them. owns: tuple[str, ...] = () repeatable: bool = False def _identity_ready(s: WizardState) -> bool: """All identity questions (task pick → executor for impl) answered.""" if not s.task_type: return False # release-handoff 는 brief 가 없다 — prepare 가 검증 보고서 인용 input 을 생성한다. if not s.brief_path and s.task_type != "release-handoff": return False if _base_ref_required(s) and S_BASE_REF_PICK not in s.answered: return False if s.base_ref_pending_text: return False if s.task_type in _STAGE_SCOPED_TASK_TYPES: if not s.approved_plan_path: return False if s.task_type == "implementation" and not _role_selection_enabled(s): if not s.executor: return False if (s.task_type == "release-handoff" and S_HANDOFF_STAGE_PICK not in s.answered): return False return True _FIELD_DEFAULTS: dict[str, Any] = { "is_new_task": None, "task_group": "", "task_id": "", "existing_brief_path": "", "task_type": "", "task_group_suggestion": "", "task_id_suggestion": "", "task_group_pending_text": False, "task_id_pending_text": False, "profile_workers": [], "profile_optional_workers": [], "keep_existing_brief": None, "brief_path": "", "brief_path_pending_text": False, "project_evidence_path": "", "project_evidence_pending_text": False, "feature_evidence_path": "", "feature_evidence_pending_text": False, "analysis_target": "", "analysis_target_pending_text": False, "reuse_worktree": None, "base_ref": "", "base_ref_pending_text": False, "approved_plan_path": "", "approved_plan_pending_text": False, "approve_plan_candidate": "", "html_approval_sidecar": "", "html_approval_option": "", "design_prep_queue": [], "design_prep_current": "", "design_prep_decision": "", "design_prep_overrides_json": "", "design_prep_notes": "", "selected_stage": "auto", "selected_stages": "", "handoff_mode": "", "handoff_stages": "", "executor": "", "critic": "", "critic_pending_text": False, "execution_identity_version": 1, "role_counts": {}, "role_models": {}, "role_selection_order": [], "reuse_previous": None, "use_defaults": None, "workers_override": "", "workers_custom_pending": False, "lead_provider": "", "lead_model": "", "claude_model": "", "codex_model": "", "antigravity_model": "", "grok_model": "", "kimi_model": "", "report_writer_provider": "", "report_writer_model": "", "directive": "", "directive_pending_text": False, "related_tasks_raw": "", "related_tasks_pending_text": False, "clarification_response_path": "", "clarification_pending_text": False, "selected_direction_path": "", "reverify_scope": "", "reverify_scope_pending_text": False, "pr_template_path": "", "pr_template_pending_text": False, "pr_template_scope": "", "fix_cycle": "", "confirmed": None, "edit_target": "", "confirmation_prompt": "", "confirmation_stages": "", "confirmation_scope": {}, "user_authorization": {}, } def _reset_field(state: WizardState, fname: str) -> None: if fname in _FIELD_DEFAULTS: default = _FIELD_DEFAULTS[fname] # copy mutable defaults if isinstance(default, list): setattr(state, fname, []) elif isinstance(default, dict): setattr(state, fname, {}) else: setattr(state, fname, default)