"""roles 단계 — reuse previous, executor, critic, defaults/custom, workers, 역할별 모델 픽의 build/submit.""" from __future__ import annotations from typing import Optional from okstra_ctl.application.resolve_assignment import resolve_lead_provider from okstra_ctl.domain.host import ( HostCapabilityMismatch, HostNotRegistered, ProviderUnavailable, ) from okstra_ctl.registry.host_registry import default_host_registry from okstra_ctl.registry.provider_registry import default_provider_registry from okstra_ctl.ids import slugify_task_segment from okstra_ctl.analysis_inputs import AnalysisInputError, resolve_analysis_target from okstra_ctl.models import ( UnknownModelError, UnknownProviderError, provider_ids, provider_supports_role, resolve_model_metadata, ) from okstra_ctl.run import recommended_role_models from okstra_ctl.workers import ( WorkersError, normalize_workers, validate_workers_against_profile, ) from .ids import ( ANTIGRAVITY_MODEL_OPTIONS, CLAUDE_MODEL_OPTIONS, CODEX_MODEL_OPTIONS, EXECUTORS, GROK_MODEL_OPTIONS, KIMI_MODEL_OPTIONS, PICK_TYPE_CUSTOM, ROLE_BLURBS, S_ANTIGRAVITY_MODEL, S_CLAUDE_MODEL, S_CODEX_MODEL, S_CRITIC_PICK, S_CRITIC_TEXT, S_DEFAULTS_OR_CUSTOM, S_EXECUTOR, S_EXECUTOR_MODEL, S_GROK_MODEL, S_KIMI_MODEL, S_LEAD_MODEL, S_REPORT_WRITER_MODEL, S_REUSE_PREVIOUS, S_WORKERS_CUSTOM, S_WORKERS_OVERRIDE, _DEFAULT_ROSTER_TOKEN, _REUSE_FILLED_STEPS, ) from .state import ( Option, Prompt, WizardError, WizardState, _convert_v1_provider_selections, _role_selection_enabled, ) from .prompts import _opt, _p from .sources import _latest_run_inputs, _resolved_roster, _suggest_latest_final_report from .steps_analysis import _restore_analysis_evidence_inputs def _validate_model(provider: str, raw: str) -> str: """Empty / 'default' → '' (use phase default). Known alias → canonical option (passed verbatim to render-bundle). Unknown → WizardError. """ raw = (raw or "").strip() if raw == "" or raw.lower() == "default": return "" try: resolve_model_metadata( provider=provider, raw_value=raw, default_display="x", default_execution="x", ) except UnknownModelError as exc: raise WizardError(str(exc)) return raw def _executor_model_field(executor: str) -> str: return {"claude": "claude_model", "codex": "codex_model", "antigravity": "antigravity_model"}[executor] def _executor_model_options(executor: str) -> list[str]: return {"claude": CLAUDE_MODEL_OPTIONS, "codex": CODEX_MODEL_OPTIONS, "antigravity": ANTIGRAVITY_MODEL_OPTIONS}[executor] def _critic_provider_choices() -> list[str]: return list(EXECUTORS) def _critic_choices() -> list[str]: return list(_critic_provider_choices()) def _critic_provider_label(provider: str, t: dict) -> str: labels = t.get("labels", {}) if provider == "claude": template = labels.get("provider_recommended", "{provider} critic (recommended)") else: template = labels.get("provider", "{provider} critic") return template.format(provider=provider) def _build_critic_pick(state: WizardState) -> Prompt: t = _p(state.workspace_root, "critic_pick") options = [ _opt(provider, _critic_provider_label(provider, t)) for provider in _critic_provider_choices() ] return Prompt( step=S_CRITIC_PICK, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_critic_pick(state: WizardState, value: str) -> Optional[str]: choice = (value or "").strip().lower() choices = _critic_choices() if choice not in choices: raise WizardError(f"critic must be one of {choices}, got: {value!r}") state.critic = choice state.critic_pending_text = False return f"critic: {choice}" def _build_critic_text(state: WizardState) -> Prompt: t = _p(state.workspace_root, "critic_text") options = [ _opt(provider, _critic_provider_label(provider, t)) for provider in _critic_provider_choices() ] return Prompt( step=S_CRITIC_TEXT, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_critic_text(state: WizardState, value: str) -> Optional[str]: choice = (value or "").strip().lower() providers = _critic_provider_choices() if choice not in providers: raise WizardError(f"critic must be one of {providers}, got: {value!r}") state.critic = choice state.critic_pending_text = False return f"critic: {choice}" def _build_executor(state: WizardState) -> Prompt: t = _p(state.workspace_root, "executor") default_suffix = t["options"].get("_DEFAULT_SUFFIX", "") options = [_opt(e, e + (default_suffix if e == "claude" else "")) for e in EXECUTORS] return Prompt( step=S_EXECUTOR, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_executor(state: WizardState, value: str) -> Optional[str]: if value not in EXECUTORS: raise WizardError(f"executor must be one of {EXECUTORS}, got: {value!r}") state.executor = value return f"executor: {value}" def _safe_model(provider: str, raw: object) -> str: """run-inputs 의 모델 display 값을 wizard 필드로 환원. 알 수 없는 값은 빈 문자열(phase 기본값)로 안전하게 떨어뜨린다.""" if not isinstance(raw, str): return "" try: return _validate_model(provider, raw) except WizardError: return "" def _safe_provider(raw: object, role: str) -> str: provider = raw.strip().lower() if isinstance(raw, str) else "" if not provider: return "" return provider if provider_supports_role(provider, role) else "" def _assignment_model(inputs: dict, worker_id: str, provider: str) -> str: assignments = inputs.get("workerAssignments") if not isinstance(assignments, list): return "" for assignment in assignments: if not isinstance(assignment, dict): continue if assignment.get("workerId") != worker_id: continue return _safe_model(provider, assignment.get("model")) return "" def _build_reuse_previous(state: WizardState) -> Prompt: t = _p(state.workspace_root, "reuse_previous") return Prompt( step=S_REUSE_PREVIOUS, kind="pick", label=t["label"], options=[_opt(k, v) for k, v in t["options"].items()], echo_template=t["echo_template"], ) def _restore_reused_lead(state: WizardState, inputs: dict) -> None: """리드는 지금 host 에서 파생한다. 직전 run 의 기록을 물려받지 않는다. run-inputs 의 ``leadProvider`` 는 그 run 이 어느 host 에서 돌았는지의 기록이지 사용자가 고른 값이 아니다. wizard 에는 리드 provider 를 묻는 단계 자체가 없고 ``resolve_lead_provider`` 가 host 의 네이티브 provider 로 파생한다. 그 기록을 그대로 요청으로 되먹이면 Codex 에서 돌린 run 을 Claude Code 에서 이어받을 때 host 검사에 걸려 brief/workers/모델 복원까지 통째로 막힌다. 네이티브 리드가 없는 host(``external``) 에서만 기록된 값이 유일한 단서이므로 그때만 참고한다. """ try: host = default_host_registry().resolve(state.host_runtime).descriptor requested_provider = ( "" if host.native_provider_id else _safe_provider(inputs.get("leadProvider"), "lead") ) assignment = resolve_lead_provider( host_id=state.host_runtime, requested_provider=requested_provider, host_registry=default_host_registry(), provider_registry=default_provider_registry(), ) except ( HostCapabilityMismatch, HostNotRegistered, ProviderUnavailable, UnknownProviderError, ) as exc: raise WizardError( f"previous run cannot be reused: {exc}; choose step-by-step re-entry" ) from exc state.lead_provider = assignment.provider state.lead_model = _safe_model(assignment.provider, inputs.get("leadModel")) def _submit_reuse_previous(state: WizardState, value: str) -> Optional[str]: if value not in ("yes", "no"): raise WizardError(f"expected 'yes' or 'no', got: {value!r}") if value == "no": state.reuse_previous = False return "reuse-previous: no (step-by-step re-entry)" state.reuse_previous = True seg = slugify_task_segment(state.task_type) inputs = _latest_run_inputs(state, phase_segment=seg) state.use_defaults = False workers = inputs.get("workers") state.workers_override = ( ",".join(w for w in workers if isinstance(w, str)) if isinstance(workers, list) else "") _restore_reused_lead(state, inputs) state.claude_model = _safe_model("claude", inputs.get("claudeModel")) state.codex_model = _safe_model("codex", inputs.get("codexModel")) state.antigravity_model = _safe_model("antigravity", inputs.get("antigravityModel")) state.grok_model = _assignment_model(inputs, "grok", "grok") state.kimi_model = _assignment_model(inputs, "kimi", "kimi") state.report_writer_provider = _safe_provider( inputs.get("reportWriterProvider"), "report-writer", ) report_writer_provider = state.report_writer_provider or "claude" state.report_writer_model = _safe_model( report_writer_provider, inputs.get("reportWriterModel")) if _role_selection_enabled(state): conversion_payload = dict(inputs) conversion_payload.setdefault("grokModel", state.grok_model) conversion_payload.setdefault("kimiModel", state.kimi_model) state.role_counts.clear() state.role_models.clear() state.role_selection_order.clear() _convert_v1_provider_selections(state, conversion_payload) directive = inputs.get("directive") state.directive = directive if isinstance(directive, str) else "" related = inputs.get("relatedTasks") state.related_tasks_raw = related if isinstance(related, str) else "" evidence_inputs, candidates = _restore_analysis_evidence_inputs( state, inputs.get("evidenceInputs") ) target = inputs.get("analysisTarget") if state.task_type == "feature-analysis": try: resolved_target = resolve_analysis_target( target if isinstance(target, str) else "", evidence_inputs, candidates, ) except AnalysisInputError as exc: raise WizardError(str(exc)) from exc requested_value = resolved_target.get("requestedValue") if not isinstance(requested_value, str): raise WizardError("analysis target resolver returned an invalid requestedValue") state.analysis_target = requested_value else: state.analysis_target = target if isinstance(target, str) else "" # clarification 은 직전 run 의 입력이 아니라 "직전 run 의 산출물(가장 최근 # final-report)" 을 재실행 입력으로 자동 선택한다 — resume-clarification 의 본질. state.clarification_response_path = _suggest_latest_final_report(state) for sid in _REUSE_FILLED_STEPS: if sid not in state.answered: state.answered.append(sid) return (f"reuse-previous: yes " f"(workers={state.workers_override or 'profile-default'}, " f"lead-model={state.lead_model or 'default'})") def _role_model_lines(state: WizardState) -> str: """이번 run 에서 실제로 모델을 고르게 되는 역할만, 추천 모델과 함께 나열한다. 뒤따르는 *_model 단계의 등장 조건과 1:1 로 맞춰 안내와 실제 화면이 어긋나지 않게 한다 (그래서 분석에 참여하지 않는 antigravity 는 executor 일 때만 나온다).""" # 이 화면의 lead 줄은 뒤따르는 lead-model picker 와 같은 provider 를 # 봐야 한다 — picker 는 호스트의 native provider 만 제시한다. rec = recommended_role_models( lead_provider=default_host_registry() .resolve(state.host_runtime) .descriptor.native_provider_id or "", report_writer_provider=state.report_writer_provider or "claude", ) roster = _resolved_roster(state) impl = state.task_type == "implementation" def line(role_label: str, blurb: str, model: str) -> str: return f" · {role_label} ({blurb}) — 추천 {model}" lines = [line("lead", ROLE_BLURBS["lead"], rec["lead"])] if impl: ex = state.executor or "claude" lines.append(line(f"executor={ex}", "코드 구현 실행자", rec.get(ex, "auto"))) else: for w in provider_ids("analyser"): if w in roster: lines.append(line(w, ROLE_BLURBS[w], rec[w])) if impl or "report-writer" in roster: lines.append(line("report-writer", ROLE_BLURBS["report-writer"], rec["report-writer"])) return "\n".join(lines) def _build_defaults_or_custom(state: WizardState) -> Prompt: t = _p(state.workspace_root, "defaults_or_custom", role_models=_role_model_lines(state)) return Prompt( step=S_DEFAULTS_OR_CUSTOM, kind="pick", label=t["label"], options=[_opt(k, v) for k, v in t["options"].items()], echo_template=t["echo_template"], ) def _submit_defaults_or_custom(state: WizardState, value: str) -> Optional[str]: if value not in ("defaults", "customize"): raise WizardError(f"expected 'defaults' or 'customize', got: {value!r}") state.use_defaults = value == "defaults" mode = ("defaults (recommended models as-is)" if state.use_defaults else "customize (manual model pick)") return f"model-mode: {mode}" def _analyser_choices(state: WizardState) -> list[str]: """프로필이 분석 워커로 허용하는 전체 후보 (report-writer 제외).""" return [ w for w in (state.profile_workers + state.profile_optional_workers) if w != "report-writer" ] def _default_analysers(state: WizardState) -> list[str]: """옵션 워커를 하나도 고르지 않았을 때 쓰는 기본 분석 로스터.""" return [w for w in state.profile_workers if w != "report-writer"] def _analyser_option(state: WizardState, worker: str, suffix: str) -> Option: label = f"{worker}{suffix}" if worker in state.profile_optional_workers else worker return _opt(value=worker, label=label) def _finalize_workers(state: WizardState, workers: list[str]) -> str: """정규화 → 프로필 allowlist 검증 → report-writer 강제 포함까지의 확정 경로. 두 워커 단계가 공유한다.""" try: chosen = normalize_workers(",".join(workers)) validate_workers_against_profile( chosen, state.profile_workers, state.profile_optional_workers, ) except WorkersError as exc: raise WizardError(str(exc)) # report-writer 는 프로필이 Required 로 선언했을 때만 강제 포함. if ("report-writer" in state.profile_workers and "report-writer" not in chosen): chosen.append("report-writer") state.workers_override = ",".join(chosen) return f"workers: {state.workers_override}" def _build_workers_override(state: WizardState) -> Prompt: """분석 워커 멀티픽. 기본 로스터(예: claude·codex)는 매 run 사실상 고정이라 옵션에서 빼고 결과에 항상 포함시킨다 — 화면에는 '기본 그대로' + 옵션 워커 + '직접 선택'만 남는다. report-writer 도 같은 이유로 빠진다. 기본 로스터에서 워커를 빼는 축소는 '직접 선택'(`workers_custom`)에서만 가능하다.""" t = _p(state.workspace_root, "workers_override") labels = t["labels"] options = [_opt( _DEFAULT_ROSTER_TOKEN, labels["default_roster"].format( workers=" + ".join(_default_analysers(state))), )] for w in state.profile_optional_workers: options.append(_opt(w, labels["add_optional"].format(worker=w))) options.append(_opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM])) return Prompt( step=S_WORKERS_OVERRIDE, kind="pick", multi=True, label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_workers_override(state: WizardState, value: str) -> Optional[str]: t = _p(state.workspace_root, "workers_override") picked = [v.strip() for v in (value or "").split(",") if v.strip()] if not picked: raise WizardError(t["errors"]["min_one_required"]) if PICK_TYPE_CUSTOM in picked: if len(picked) > 1: raise WizardError(t["errors"]["custom_must_be_alone"]) state.workers_custom_pending = True return None allowed = {_DEFAULT_ROSTER_TOKEN, *state.profile_optional_workers} unknown = [w for w in picked if w not in allowed] if unknown: raise WizardError( t["errors"]["unknown_option"].format(values=",".join(unknown))) state.workers_custom_pending = False added = [w for w in picked if w != _DEFAULT_ROSTER_TOKEN] return _finalize_workers(state, _default_analysers(state) + added) def _build_workers_custom(state: WizardState) -> Prompt: """'직접 선택' 화면 — 기본 로스터까지 포함한 전체 분석 워커 후보. 기본 워커를 빼는 축소는 이 화면에서만 가능하다.""" t = _p(state.workspace_root, "workers_custom") suffix = t["options"].get("_OPTIONAL_SUFFIX", "") return Prompt( step=S_WORKERS_CUSTOM, kind="pick", multi=True, label=t["label"], options=[_analyser_option(state, w, suffix) for w in _analyser_choices(state)], echo_template=t["echo_template"], ) def _submit_workers_custom(state: WizardState, value: str) -> Optional[str]: picked = [v.strip() for v in (value or "").split(",") if v.strip()] if not picked: t = _p(state.workspace_root, "workers_custom") raise WizardError(t["errors"]["min_one_required"]) state.workers_custom_pending = False return _finalize_workers(state, picked) def _model_pick(step: str, label: str, options: list[str], echo: str) -> Prompt: # "default" picks the role's recommended model — leaving it here yields # the SAME result as the 'Use defaults' branch. Spell that out on the # label so default ↔ customize never reads as "no difference". opts = [ _opt(o, "default (recommended model)" if o == "default" else o) for o in options ] return Prompt(step=step, kind="pick", label=label, options=opts, echo_template=echo) def _qualified_role_model_options(role: str) -> list[str]: registry = default_provider_registry() options = ["default", *_provider_picker_options("claude", registry)] for provider in registry.ids(role): if provider == "claude": continue options.extend( f"{provider}:{model}" for model in _provider_picker_options(provider, registry) ) return options def _provider_picker_options(provider: str, registry=None) -> list[str]: provider_registry = registry or default_provider_registry() return [ alias for alias, model in provider_registry.resolve(provider).models.items() if model.in_picker ] def _submit_qualified_role_model( state: WizardState, value: str, role: str, model_field: str, provider_field: str, ) -> str: raw = (value or "").strip() if ":" not in raw: setattr(state, provider_field, "") setattr(state, model_field, _validate_model("claude", raw)) return getattr(state, model_field) or "default" provider, model = (part.strip().lower() for part in raw.split(":", 1)) if not provider_supports_role(provider, role): raise WizardError(f"provider {provider!r} does not support the {role} role") setattr(state, provider_field, provider) setattr(state, model_field, _validate_model(provider, model)) return raw def _build_lead_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "lead_model") provider = default_host_registry().resolve( state.host_runtime ).descriptor.native_provider_id options = ( ["default", *_provider_picker_options(provider)] if provider else _qualified_role_model_options("lead") ) return _model_pick(S_LEAD_MODEL, t["label"], options, t["echo_template"]) def _submit_lead_model(state: WizardState, value: str) -> Optional[str]: native_provider = default_host_registry().resolve( state.host_runtime ).descriptor.native_provider_id if native_provider: raw = (value or "").strip() requested_provider = "" if ":" in raw: requested_provider, raw = ( part.strip().lower() for part in raw.split(":", 1) ) try: assignment = resolve_lead_provider( host_id=state.host_runtime, requested_provider=requested_provider, host_registry=default_host_registry(), provider_registry=default_provider_registry(), ) except ( HostCapabilityMismatch, HostNotRegistered, ProviderUnavailable, UnknownProviderError, ) as exc: raise WizardError(str(exc)) from exc state.lead_provider = assignment.provider state.lead_model = _validate_model(assignment.provider, raw) return f"lead-model: {assignment.provider}:{state.lead_model or 'default'}" selected = _submit_qualified_role_model( state, value, "lead", "lead_model", "lead_provider", ) return f"lead-model: {selected}" def _build_executor_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "executor_model", executor=state.executor) return _model_pick( S_EXECUTOR_MODEL, t["label"], _executor_model_options(state.executor), t["echo_template"].replace("{executor}", state.executor), ) def _submit_executor_model(state: WizardState, value: str) -> Optional[str]: resolved = _validate_model(state.executor, value) setattr(state, _executor_model_field(state.executor), resolved) return f"{state.executor}-model: {resolved or 'default'}" def _build_claude_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "claude_model") return _model_pick(S_CLAUDE_MODEL, t["label"], CLAUDE_MODEL_OPTIONS, t["echo_template"]) def _submit_claude_model(state: WizardState, value: str) -> Optional[str]: state.claude_model = _validate_model("claude", value) return f"claude-model: {state.claude_model or 'default'}" def _build_codex_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "codex_model") return _model_pick(S_CODEX_MODEL, t["label"], CODEX_MODEL_OPTIONS, t["echo_template"]) def _submit_codex_model(state: WizardState, value: str) -> Optional[str]: state.codex_model = _validate_model("codex", value) return f"codex-model: {state.codex_model or 'default'}" def _build_antigravity_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "antigravity_model") return _model_pick(S_ANTIGRAVITY_MODEL, t["label"], ANTIGRAVITY_MODEL_OPTIONS, t["echo_template"]) def _submit_antigravity_model(state: WizardState, value: str) -> Optional[str]: state.antigravity_model = _validate_model("antigravity", value) return f"antigravity-model: {state.antigravity_model or 'default'}" def _build_grok_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "grok_model") return _model_pick( S_GROK_MODEL, t["label"], GROK_MODEL_OPTIONS, t["echo_template"], ) def _submit_grok_model(state: WizardState, value: str) -> Optional[str]: state.grok_model = _validate_model("grok", value) return f"grok-model: {state.grok_model or 'default'}" def _build_kimi_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "kimi_model") return _model_pick( S_KIMI_MODEL, t["label"], KIMI_MODEL_OPTIONS, t["echo_template"], ) def _submit_kimi_model(state: WizardState, value: str) -> Optional[str]: state.kimi_model = _validate_model("kimi", value) return f"kimi-model: {state.kimi_model or 'default'}" def _build_report_writer_model(state: WizardState) -> Prompt: t = _p(state.workspace_root, "report_writer_model") return _model_pick(S_REPORT_WRITER_MODEL, t["label"], _qualified_role_model_options("report-writer"), t["echo_template"]) def _submit_report_writer_model(state: WizardState, value: str) -> Optional[str]: selected = _submit_qualified_role_model( state, value, "report-writer", "report_writer_model", "report_writer_provider", ) return f"report-writer-model: {selected}"