"""okstra-run wizard — state machine for interactive task setup. The okstra-run skill used to encode the full input-collection flow in prose, which drifted into seven internal inconsistencies (executor placement, free-text-via-AskUserQuestion, missing clarification prompt for impl, etc.). This module owns the flow as code so the skill becomes a thin loop. Public surface: - ``WizardState`` - serializable dataclass; one state file per run. - ``Prompt`` - what the skill should ask next. - ``init_state()`` - seed from project-root / project-id / workspace-root. - ``next_prompt()`` - deterministic; pure read on state. - ``submit()`` - validate + advance. - ``render_args()`` - final args for ``okstra render-bundle``. - ``wizard_outcome()`` - final args plus persistence actions and confirmation. The skill calls these via the ``okstra wizard`` CLI subcommand; it never imports this module directly. """ from __future__ import annotations import copy import hashlib import json import math import re import subprocess from dataclasses import asdict, dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Callable, Optional from okstra_ctl.application.advance_wizard import plan_prompt from okstra_ctl.application.resolve_assignment import resolve_lead_provider from okstra_ctl.assignment_environment import load_assignment_context from okstra_ctl.assignment_resolver import ( AssignmentContext, AssignmentResolutionError, RoleInstance, resolve_assignments, resolve_model_assignment, ) from okstra_ctl.domain.host import ( CurrentSessionModelAttestation, HostCapabilityMismatch, HostNotRegistered, HostSessionContext, ProviderUnavailable, ) from okstra_ctl.domain.wizard.interaction import ( InteractionPlan, WizardAnswerError, WizardOption, WizardPrompt, normalize_planned_answer, ) from okstra_ctl.dispatch_state import detect_terminal_backend from okstra_ctl.registry.host_registry import default_host_registry from okstra_ctl.registry.provider_registry import default_provider_registry from okstra_ctl.model_defaults import ModelDefaultScopes, default_candidates from okstra_ctl.model_pool import ModelPool from okstra_ctl import next_phase from okstra_ctl.legacy_model_selection import serialize_host_session_context from okstra_ctl.role_requirements import ( RoleProfile, RoleProfileError, RoleRequirement, load_role_profile, ) from okstra_ctl.ids import slugify_task_segment from okstra_ctl.brief_frontmatter import read_brief_frontmatter from okstra_ctl.analysis_inputs import ( ANALYSIS_TASK_TYPES, AnalysisInputError, AnalysisReportCandidate, _resolved_within, list_evidence_candidates, load_analysis_report_candidate, resolve_analysis_target, resolve_evidence_inputs, ) from okstra_ctl.models import ( PROVIDER_MAPPINGS, UnknownModelError, UnknownProviderError, picker_options, provider_ids, provider_supports_role, resolve_model_metadata, ) from okstra_ctl.clarification_items import ( scan_approval_gate, sidecar_answers, user_response_sidecars, ) from okstra_ctl.convergence_store import write_json_atomic from okstra_ctl.json_boundary import JsonBoundaryError, load_owned_object from okstra_ctl.incremental_scope import ( parse_stage_graph, preview_link_availability_for_report, ) from okstra_ctl.design_prep import ( DesignPrepError, load_design_prep_items, resolve_design_prep, write_design_prep_input, ) from okstra_ctl.implementation_direction import ( DirectionSelectionError, lexical_absolute_path, validate_task_artifact_path, ) from okstra_ctl.final_report_paths import final_report_data_path from okstra_ctl.plan_run_root import list_implementation_planning_reports from okstra_ctl.pr_template import PrTemplateError, resolve_pr_template_path from okstra_ctl.run import ( APPROVED_FRONTMATTER_PATTERN, PrepareError, _apply_cli_implementation_option, _extract_frontmatter_block, _load_final_report_data_if_present, _assignment_resolution_message, _record_approved_flag, _reject_blocking_plan_body_gate, _set_data_json_approved_true_if_present, _model_default_scopes, recommended_role_models, ) from okstra_ctl.stage_map import ( StageMapError, parse_stage_map_file, stage_map_records, ) from okstra_ctl.user_response import ( PlanDecisionRecord, parse_plan_decision, ) from okstra_ctl.workers import ( ALLOWED_WORKERS, WorkersError, normalize_workers, resolve_optional_workers, resolve_profile_workers, validate_workers_against_profile, ) from okstra_ctl.workflow import PHASE_SEQUENCE from okstra_ctl.wizard_stage_intent import ( WHOLE_TASK_STAGE, WizardStageIntent, WizardStageIntentError, resolve_wizard_stage_intent, wizard_stage_confirmation_label, ) from okstra_ctl import fix_cycles, worktree_registry from okstra_ctl.worktree import ( compute_worktree_path, is_git_work_tree, main_worktree_path, preview_worktree_decision, resolve_stage_worktree_decision, ) from okstra_ctl.paths import RunRef, task_dir, task_runs_dir from okstra_ctl.work_categories import resolve_work_category from okstra_ctl.run_context import latest_run_inputs from okstra_project.dirs import project_json_path from okstra_project.state import ( StateError, list_project_tasks, read_latest_task, read_task_manifest, find_task_root, ) # ---- Constants ----------------------------------------------------------- _ANALYSIS_TASK_TYPE_DESCRIPTIONS = ( "Map project structure and bounded codebase scope (read-only)", "Analyse feature behavior and test coverage scope (read-only)", "Assess change impact and preserved behavior (read-only)", ) TASK_TYPES: list[tuple[str, str]] = [ ("requirements-discovery", "Classify request and route to next safe phase"), ("improvement-discovery", "Find improvement candidates within a codebase scope and lens whitelist"), *zip(ANALYSIS_TASK_TYPES, _ANALYSIS_TASK_TYPE_DESCRIPTIONS), ("error-analysis", "Evidence-based root-cause analysis (no code changes)"), ("implementation-option-selection", "Compare implementation options (read-only)"), ("implementation-planning", "Plan options + request user approval"), ("implementation", "Execute approved plan (requires approved final-report)"), ("final-verification", "Acceptance + residual-risk review"), ("release-handoff", "Drive commit/push/PR — reuse the implementation task-key (new keys fail the empty-commits gate)"), ] TASK_TYPE_VALUES = [tt for tt, _ in TASK_TYPES] EXECUTORS = ["claude", "codex", "antigravity"] # 역할 한 줄 설명 — defaults_or_custom 안내에서 각 역할이 무슨 일을 하는지 보여준다. ROLE_BLURBS = { "lead": "Okstra lead · 워커 dispatch·교차검증 수렴·최종 합성", "claude": "독립 분석 워커 (Claude)", "codex": "독립 분석 워커 (Codex)", "antigravity": "독립 분석 워커 (Antigravity)", "grok": "독립 분석 워커 (Grok)", "kimi": "독립 분석 워커 (Kimi)", "report-writer": "최종 리포트 작성 (분석은 하지 않음)", } # Task types that consume an approved plan and need a stage-scope pick: # implementation executes a stage, final-verification verifies a stage # (or the whole task via `auto`). Both gate the approved-plan + stage steps. _STAGE_SCOPED_TASK_TYPES = ("implementation", "final-verification") # brief 는 entry phase 의 입력물(비개발자/타 팀의 아이디어·에러 초안)이다. # downstream phase 는 task manifest 의 taskBriefPath 를 자동 carry-in 하며 # 위저드에서 brief 를 묻지 않는다 (release-handoff 는 prepare 가 검증 보고서 # 인용 input 문서를 자동 생성하므로 brief 자체가 없다). _BRIEF_ENTRY_TASK_TYPES = ( "requirements-discovery", "improvement-discovery", *ANALYSIS_TASK_TYPES, "error-analysis", ) CANONICAL_BASE_REFS = ["main", "dev", "staging", "preprod", "prod"] BASE_REF_FREE_INPUT_TOKEN = "__free_input__" CLAUDE_MODEL_OPTIONS = ["default", *picker_options("claude")] CODEX_MODEL_OPTIONS = ["default", *picker_options("codex")] ANTIGRAVITY_MODEL_OPTIONS = ["default", *picker_options("antigravity")] GROK_MODEL_OPTIONS = ["default", *picker_options("grok")] KIMI_MODEL_OPTIONS = ["default", *picker_options("kimi")] # special pick value: start a brand-new task TASK_PICK_NEW_TOKEN = "__new__" # AskUserQuestion renders at most 4 options per question; the last slot is # always reserved for the direct-input option, so recommendation lists from # dynamic sources (catalog, manifest) are capped at 3. _RECOMMENDATION_CAP = 3 # Pick-vs-free-text tokens shared by suggestion-aware prompts. PICK_USE_SUGGESTED = "__use_suggested__" PICK_TYPE_CUSTOM = "__free_input__" # workers_override 에서 "옵션 워커를 추가하지 않음" 을 뜻하는 sentinel. _DEFAULT_ROSTER_TOKEN = "__default_roster__" _RECENT_PREFIX = "__recent:" _REPORT_PREFIX = "__report:" _BRIEF_PREFIX = "__brief:" def _looks_like_template_placeholder(value: str) -> bool: """Treat ````, ``<...>``, empty strings, and ``self`` as non-suggestions. Anything else (a real slug-like value) is honored.""" v = (value or "").strip() if not v: return True if v.startswith("<") and v.endswith(">"): return True if v.lower() in ("self", "tbd", "n/a", "na", "none"): return True return False def _brief_suggestions(path: Path) -> tuple[str, str]: """Return ``(task_group_suggestion, task_id_suggestion)`` extracted from the brief's frontmatter, or empty strings when no usable value exists. - ``task_group`` ← frontmatter ``task-group``. - ``task_id`` ← frontmatter ``brief-id`` (which matches the filename stem in okstra-brief-gen output and is the strongest single identifier of the task). A brief without frontmatter, or with placeholder values, yields two empty strings — callers fall back to plain-text input. """ fm = read_brief_frontmatter(path) tg_raw = fm.get("task-group", "") bid_raw = fm.get("brief-id", "") tg = "" if _looks_like_template_placeholder(tg_raw) else tg_raw tid = "" if _looks_like_template_placeholder(bid_raw) else bid_raw return tg, tid def _project_relative_path(path: Path, project_root: Path) -> str: try: return str(path.relative_to(project_root)) except ValueError: return str(path) def _project_relative_value(path_value: str, project_root: Path) -> str: p = Path(path_value) if p.is_absolute(): return _project_relative_path(p, project_root) return str(p) def _parse_iso_timestamp(value: Any) -> float: if not isinstance(value, str): return 0.0 text = value.strip() if not text: return 0.0 if text.endswith("Z"): text = f"{text[:-1]}+00:00" try: return datetime.fromisoformat(text).timestamp() except ValueError: return 0.0 def _file_recency(path: Path) -> float: try: stat = path.stat() except OSError: return 0.0 return max(stat.st_mtime, float(getattr(stat, "st_birthtime", 0.0) or 0.0)) def _accept_brief_path(state: WizardState, path: Path) -> None: """Record a validated brief and derive safe identity suggestions. New-task runs now ask for ``task_group`` before the brief so the wizard can offer same-group brief candidates. If the selected brief carries a conflicting frontmatter ``task-group``, fail fast; otherwise keep using the brief's ``brief-id`` as the task-id suggestion. """ tg_suggestion, tid_suggestion = _brief_suggestions(path) if state.task_group and tg_suggestion: suggested_group = _slug_or_die(tg_suggestion, "task_group") if suggested_group != state.task_group: raise WizardError( "brief task-group does not match selected task-group: " f"{tg_suggestion!r} != {state.task_group!r} ({path})" ) state.brief_path = str(path) state.brief_path_pending_text = False if state.is_new_task: if not state.task_group: state.task_group_suggestion = tg_suggestion if not state.task_id: state.task_id_suggestion = tid_suggestion # ---- Step IDs ------------------------------------------------------------ S_TASK_PICK = "task_pick" S_TASK_GROUP = "task_group" S_TASK_GROUP_TEXT = "task_group_text" S_TASK_ID = "task_id" S_TASK_ID_TEXT = "task_id_text" S_TASK_TYPE = "task_type" S_TASK_TYPE_TEXT = "task_type_text" S_BRIEF_KEEP = "brief_keep" S_BRIEF_PATH_PICK = "brief_path_pick" S_BRIEF_PATH = "brief_path" S_BRIEF_CARRY = "brief_carry" S_FEATURE_EVIDENCE_PICK = "feature_evidence_pick" S_FEATURE_EVIDENCE = "feature_evidence" S_PROJECT_EVIDENCE_PICK = "project_evidence_pick" S_PROJECT_EVIDENCE = "project_evidence" S_ANALYSIS_TARGET_PICK = "analysis_target_pick" S_ANALYSIS_TARGET = "analysis_target" S_BASE_REF_PICK = "base_ref_pick" S_BASE_REF_TEXT = "base_ref_text" S_SELECTED_DIRECTION_PICK = "selected_direction_pick" S_APPROVED_PLAN_PICK = "approved_plan_pick" S_APPROVED_PLAN = "approved_plan" S_APPROVE_PLAN_CONFIRM = "approve_plan_confirm" S_DESIGN_PREP_DECISION = "design_prep_decision" S_DESIGN_PREP_OVERRIDES = "design_prep_overrides" S_DESIGN_PREP_CONFIRM = "design_prep_confirm" S_STAGE_PICK = "stage_pick" S_HANDOFF_STAGE_PICK = "handoff_stage_pick" S_EXECUTOR = "executor" S_CRITIC_PICK = "critic_pick" S_CRITIC_TEXT = "critic_text" S_REUSE_PREVIOUS = "reuse_previous" S_DEFAULTS_OR_CUSTOM = "defaults_or_custom" S_WORKERS_OVERRIDE = "workers_override" S_WORKERS_CUSTOM = "workers_custom" S_LEAD_MODEL = "lead_model" S_EXECUTOR_MODEL = "executor_model" S_CLAUDE_MODEL = "claude_model" S_CODEX_MODEL = "codex_model" S_ANTIGRAVITY_MODEL = "antigravity_model" S_GROK_MODEL = "grok_model" S_KIMI_MODEL = "kimi_model" S_REPORT_WRITER_MODEL = "report_writer_model" S_DIRECTIVE_PICK = "directive_pick" S_DIRECTIVE = "directive" S_RELATED_TASKS_PICK = "related_tasks_pick" S_RELATED_TASKS = "related_tasks" S_CLARIFICATION_PICK = "clarification_pick" S_CLARIFICATION = "clarification" S_REVERIFY_SCOPE_PICK = "reverify_scope_pick" S_REVERIFY_SCOPE_STAGES = "reverify_scope_stages" S_PR_TEMPLATE_PICK = "pr_template_pick" S_PR_TEMPLATE = "pr_template" S_PR_TEMPLATE_SCOPE = "pr_template_scope" S_FIX_CYCLE_CONFIRM = "fix_cycle_confirm" S_CONFIRM = "confirm" S_EDIT_TARGET = "edit_target" S_DONE = "done" S_ABORTED = "aborted" # ---- 멀티탭 배치 프롬프트 그룹 (방출 계층 전용) ---- # 그룹 id 는 S_* 가 아니므로 prompts JSON SOT / step-id 동기화 검사 대상이 아니다. GROUP_MODELS = "models" GROUP_OPTIONS = "options" GROUP_MAX_TABS = 4 # AskUserQuestion 의 질문(탭) 수 한도 # 멤버는 모두 서로 의존이 없는 단일선택 픽 step 이어야 한다. # *_TEXT 후속 / workers_override / pr_template_scope 는 의존성 때문에 개별 유지. PROMPT_GROUPS: dict[str, tuple[str, ...]] = { GROUP_MODELS: (S_LEAD_MODEL, S_EXECUTOR_MODEL, S_CLAUDE_MODEL, S_CODEX_MODEL, S_ANTIGRAVITY_MODEL, S_GROK_MODEL, S_KIMI_MODEL, S_REPORT_WRITER_MODEL), GROUP_OPTIONS: (S_DIRECTIVE_PICK, S_RELATED_TASKS_PICK, S_CLARIFICATION_PICK, S_PR_TEMPLATE_PICK), } GROUP_LABELS: dict[str, str] = { GROUP_MODELS: "모델 선택 (탭별로 선택)", GROUP_OPTIONS: "추가 옵션 (탭별로 선택)", } _STEP_TO_GROUP: dict[str, str] = { sid: gid for gid, ids in PROMPT_GROUPS.items() for sid in ids } # ---- 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 = "" host_runtime: str = "claude-code" host_entry_mode: str = "current-session" available_functions: list[str] = field(default_factory=list) # 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 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 = "" @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) @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.""" # ---- Validation helpers -------------------------------------------------- _SLUG_OK = re.compile(r"[a-z0-9]") 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 def _resolve_path(path_str: str, project_root: Path) -> Path: p = Path(path_str).expanduser() return p if p.is_absolute() else (project_root / p).resolve() def _require_file(path_str: str, project_root: Path, label: str) -> Path: if not (path_str or "").strip(): raise WizardError(f"{label}: empty path") p = _resolve_path(path_str, project_root) if not p.is_file(): raise WizardError(f"{label}: file not found: {p}") return p _SELECTION_REPORT_RE = re.compile( r"^final-report-implementation-option-selection-(?P\d{3,})\.md$" ) def _selected_direction_candidates(state: WizardState) -> list[str]: if not state.project_root or not state.task_group or not state.task_id: return [] project_root = Path(state.project_root).resolve() task_root = lexical_absolute_path( task_dir(project_root, state.task_group, state.task_id) ) reports = ( task_runs_dir(project_root, state.task_group, state.task_id) / "implementation-option-selection" / "reports" ) candidates: list[tuple[int, Path]] = [] for report in reports.glob("final-report-implementation-option-selection-*.md"): match = _SELECTION_REPORT_RE.fullmatch(report.name) if match is None: continue try: validated = validate_task_artifact_path( report, task_root, "selection report" ) except DirectionSelectionError: continue candidates.append((int(match.group("seq")), validated)) return [ _project_relative_path(path, project_root) for _, path in sorted(candidates, reverse=True)[:3] ] def _planning_rerun_selected(state: WizardState) -> bool: if ( not state.clarification_response_path or not state.project_root or not state.task_group or not state.task_id ): return False project_root = Path(state.project_root).resolve() raw_path = Path(state.clarification_response_path).expanduser() path = lexical_absolute_path( raw_path if raw_path.is_absolute() else project_root / raw_path ) task_root = lexical_absolute_path( task_dir(project_root, state.task_group, state.task_id) ) reports = lexical_absolute_path( task_runs_dir(project_root, state.task_group, state.task_id) / "implementation-planning" / "reports" ) try: validate_task_artifact_path(path, task_root, "planning report") except DirectionSelectionError: return False return ( path.is_file() and not path.is_symlink() and re.fullmatch( r"final-report-implementation-planning-\d{3,}\.(?:md|data\.json)", path.name ) is not None and path.parent == reports ) def _build_selected_direction_pick(state: WizardState) -> Prompt: candidates = _selected_direction_candidates(state) t = _p(state.workspace_root, S_SELECTED_DIRECTION_PICK) if not candidates: raise WizardError(t["errors"]["none"]) return Prompt( step=S_SELECTED_DIRECTION_PICK, kind="pick", label=t["label"], options=[_opt(path, path) for path in candidates], echo_template=t["echo_template"], ) def _submit_selected_direction_pick( state: WizardState, value: str ) -> Optional[str]: candidates = _selected_direction_candidates(state) if value not in candidates: raise WizardError( _p(state.workspace_root, S_SELECTED_DIRECTION_PICK)["errors"][ "unknown" ].format(value=value) ) state.selected_direction_path = value return f"selected-direction: {value}" def _classify_approved_plan(path_str: str, project_root: Path) -> tuple[Path, bool]: """Resolve the plan and classify it as fully-approved vs approvable. Returns ``(resolved_path, already_fully_approved)``. Raises WizardError ONLY for failures that approval cannot fix: missing `approved` on the report record (or schema-v1 frontmatter), a blocking plan-body gate, an unparseable §1, or unresolved `Blocks=approval` rows. A plan that is merely not-yet-approved (record `approved: false`, gate ok, no blockers) returns ``already_fully_approved=False`` — the approve-confirm step offers to flip it. """ from okstra_ctl.final_report_paths import require_approved_plan_record resolved = _require_file(path_str, project_root, "approved plan") try: p = require_approved_plan_record(resolved) except ValueError as exc: raise WizardError(str(exc)) from exc loaded = _load_final_report_data_if_present(p) if loaded is not None: planning = loaded[1].get("implementationPlanning") if ( isinstance(planning, dict) and planning.get("planningContract") == "selected-direction" and planning.get("outcome") == "direction-invalidated" ): raise WizardError( "direction-invalidated planning reports are not approvable; " "re-enter implementation-option-selection" ) # A blocking gate or an open Blocks=approval row makes the plan UN-approvable # — these raise regardless of the current flag value. _reject_blocking_plan_body_gate(p, "", action="approved plan validation") scan = scan_approval_gate(p) if scan.unreadable_reason: raise WizardError( f"approved plan §1 approval gate could not be read: {p}\n" f" {scan.unreadable_reason}.\n" " the gate refuses to soft-pass — re-render the report so §1 " "matches the schema." ) blockers = scan.blockers if blockers: lines = [ f"approved plan §1 has {len(blockers)} unresolved `Blocks=approval` " "row(s); resolve them or mark them obsolete before approving:", ] for b in blockers: lines.append(f" - {b.row_id} (Status={b.raw_status})") lines.append(f" file: {p}") raise WizardError("\n".join(lines)) try: record_approved = _record_approved_flag(p) except PrepareError as exc: raise WizardError(str(exc)) from exc return p, record_approved is True def _approve_plan_in_place(plan_path: Path) -> None: """Flip the report record `frontmatter.approved` to true and re-render.""" if not _set_data_json_approved_true_if_present(plan_path): raise WizardError( f"approve-plan: report record could not be updated: {plan_path}" ) def _find_html_approval_sidecar( plan_path: Path, ) -> Optional[tuple[Path, PlanDecisionRecord]]: """plan 의 run 디렉토리 sibling ``user-responses/`` 에서 승인 판정을 담은 sidecar 를 찾는다. source-report 파일명과 seq 가 plan 과 일치해야 하며, 복수면 mtime 최신을 택한다. 승인이 아닌 판정(반려·재작업 요청)은 여기서 걸러진다 — 이 단계가 묻는 것은 "사용자가 이 plan 을 승인해 두었는가" 뿐이고, 반려 사유는 다음 planning run 이 sidecar 를 통째로 읽어 처리한다.""" loaded = _load_final_report_data_if_present(plan_path) if loaded is not None: planning = loaded[1].get("implementationPlanning") if ( isinstance(planning, dict) and planning.get("planningContract") == "selected-direction" ): return None responses_dir = plan_path.parent.parent / "user-responses" if not responses_dir.is_dir(): return None m = re.search(r"-(\d+)\.(?:md|data\.json)$", plan_path.name) plan_seq = m.group(1) if m else "" best: Optional[tuple[float, Path, PlanDecisionRecord]] = None for f in sorted(responses_dir.glob("user-response-*.md")): try: text = f.read_text(encoding="utf-8", errors="replace") except OSError: continue rec = parse_plan_decision(text) if rec is None or not rec.approved or rec.seq != plan_seq: continue from okstra_ctl.final_report_paths import ( final_report_markdown_path, is_report_record_path, ) expected_name = ( final_report_markdown_path(plan_path).name if is_report_record_path(plan_path) else plan_path.name ) if Path(rec.source_report).name != expected_name: continue mtime = f.stat().st_mtime if best is None or mtime > best[0]: best = (mtime, f, rec) return (best[1], best[2]) if best else None def _validate_sidecar_option(plan_path: Path, option_name: str, errors_t: dict) -> None: """sidecar 의 옵션 이름이 plan data.json 의 optionCandidates 에 있는지 검증한다. 없으면 유효 후보를 나열하며 거부한다 (fail-closed).""" data_path = final_report_data_path(plan_path) candidates: list[str] = [] if data_path.is_file(): try: data = load_owned_object(data_path, artifact="planning final report") planning = data.get("implementationPlanning") or {} candidates = [ c.get("name", "") for c in planning.get("optionCandidates") or [] if isinstance(c, dict) and c.get("name") ] except (OSError, JsonBoundaryError): candidates = [] if option_name not in candidates: raise WizardError( errors_t["unknown_option"].format( option=option_name, candidates=", ".join(candidates) if candidates else "(없음)", ) ) def _plan_short_label(candidate: str) -> str: """plan 파일명에서 사용자용 짧은 식별자를 뽑는다. final-report-implementation-planning-002.data.json → implementation-planning-002""" if not candidate: return "" name = Path(candidate).name stem = name[: -len(".data.json")] if name.endswith(".data.json") else Path(name).stem return stem.removeprefix("final-report-") def _stage_plan_for_confirmation( state: WizardState, path_str: str, *, suffix: str = "" ) -> Optional[str]: """Resolve + validate a selected plan, then stage it for the approve-confirm step. Selection NEVER finalizes the plan — the confirm step always runs and asks the user to proceed (approving the plan first if it is not yet approved). `_classify_approved_plan` still raises for failures approval cannot fix.""" p, _ = _classify_approved_plan(path_str, Path(state.project_root)) state.approved_plan_pending_text = False state.approved_plan_path = "" state.approve_plan_candidate = str(p) state.html_approval_sidecar = "" state.html_approval_option = "" found = _find_html_approval_sidecar(p) if found is not None: sidecar_path, record = found state.html_approval_sidecar = str(sidecar_path) state.html_approval_option = record.implementation_option t = _p(state.workspace_root, "approve_plan_confirm", path=str(p)) variants = t["echo_variants"] key = ("selected_final_verification" if state.task_type == "final-verification" and variants.get("selected_final_verification") else "selected") msg = variants[key].format(path=p) return f"{msg} {suffix}".rstrip() if suffix else msg def _git_main_worktree(project_root: Path) -> Path: # main worktree 해소는 worktree 모듈의 public seam(main_worktree_path)에 # 위임한다. base-ref 검증은 git 이 없으면 의미가 없으므로, 자체 메시지로 # 먼저 fail-fast 한다 (과거 자체 `--git-common-dir` 구현을 대체). if not is_git_work_tree(project_root): raise WizardError(f"git unavailable or not a work tree in {project_root}") return main_worktree_path(project_root) def _validate_base_ref(ref: str, project_root: Path) -> str: ref = (ref or "").strip() if not ref: raise WizardError("base ref is empty") main_wt = _git_main_worktree(project_root) rc = subprocess.call( ["git", "-C", str(main_wt), "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) if rc != 0: raise WizardError(f"base ref not found in repository: {ref}") return ref 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] # ---- 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 count_id = f"role-count:{cross_requirement.role}" state.role_selection_order.append(count_id) 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) state.role_selection_order.append( f"role-model:{cross_requirement.role}:{ordinal}" ) 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 } if state.host_entry_mode == "spawn-process": static_roles.add("leader") 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) 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 _role_count_prompt_id(role: str) -> str: return f"role-count:{role}" def _role_add_prompt_id(role: str) -> str: return f"role-add:{role}" def _role_model_prompt_id(role: str, ordinal: int) -> str: return f"role-model:{role}:{ordinal}" def _is_role_selection_step(step_id: str) -> bool: return step_id.startswith(("role-count:", "role-model:", "role-add:")) def _selected_role_count( state: WizardState, requirement: RoleRequirement, ) -> int: if requirement.min_count == requirement.max_count: return requirement.min_count if requirement.role in state.role_counts: return state.role_counts[requirement.role] # min=0 선택 역할(예: critic)은 사용자가 열기 전까지 기본 0 if requirement.min_count == 0: return 0 return requirement.recommended_count def _selectable_static_requirements( state: WizardState, profile: RoleProfile, ) -> tuple[RoleRequirement, ...]: requirements: list[RoleRequirement] = [] if state.host_entry_mode == "spawn-process": requirements.append(RoleRequirement("leader", 1, 1, 1, "lead")) requirements.extend( requirement for requirement in profile.roles if not requirement.dynamic and _selected_role_count(state, requirement) > 0 ) return tuple(requirements) def _count_prompt( state: WizardState, requirement: RoleRequirement, ) -> Prompt: prompt = _p( state.workspace_root, "role_count", role=requirement.role, minimum=str(requirement.min_count), maximum=str(requirement.max_count), default=str(requirement.recommended_count), ) return Prompt( step=_role_count_prompt_id(requirement.role), kind="pick", label=prompt["label"], options=[ _opt( str(count), prompt["options"]["count"].format( count=count, default_suffix=( prompt["options"].get("default_suffix", "") if count == requirement.recommended_count else "" ), ), ) for count in range(requirement.min_count, requirement.max_count + 1) ], echo_template=prompt["echo_template"], ) def _role_add_prompt( state: WizardState, requirement: RoleRequirement, ) -> Prompt: """min=0 선택 역할: 프로필의 적정 수량이 기본이고, 1..max 를 이 스텝에서 고른다. 적정이 0 이면 기본은 추가 안 함이다. 0 보다 큰 적정을 선언한 역할만 기본이 열린 상태로 뜬다 — 어느 쪽이든 사용자는 이 화면에서 바꿀 수 있다. """ prompt = _p( state.workspace_root, "role_add", role=requirement.role, maximum=str(requirement.max_count), ) suffix = prompt["options"].get("default_suffix", "") def _default_suffix(count: int) -> str: return suffix if count == requirement.recommended_count else "" options = [ _opt( "0", prompt["options"]["skip"].format(default_suffix=_default_suffix(0)), ), ] for count in range(1, requirement.max_count + 1): options.append( _opt( str(count), prompt["options"]["add"].format( count=count, default_suffix=_default_suffix(count), ), ) ) return Prompt( step=_role_add_prompt_id(requirement.role), kind="pick", label=prompt["label"], options=options, echo_template=prompt["echo_template"], ) def _role_default_candidates( state: WizardState, role: str, pool: ModelPool, scopes: ModelDefaultScopes | None = None, ) -> tuple[str, ...]: try: selected_scopes = scopes or _model_default_scopes(Path(state.project_root)) scoped = default_candidates(role, selected_scopes) except (PrepareError, ValueError) as exc: raise WizardError(str(exc)) from exc if scoped: for model_ref in scoped: availability = pool.availability( model_ref, role, state.host_runtime, "new-session", ) if not availability.available: raise WizardError( f"configured default model {model_ref!r} is unavailable for " f"role {role!r}: {availability.reason}" ) return tuple(scoped) return tuple( str(model.model_ref) for model in pool.default_candidates(role) ) def _role_model_options( state: WizardState, profile: RoleProfile, role: str, ordinal: int, context: AssignmentContext, scopes: ModelDefaultScopes, ) -> list[Option]: pool = context.pool defaults = _role_default_candidates(state, role, pool, scopes) if defaults: offset = (ordinal - 1) % len(defaults) defaults = (*defaults[offset:], *defaults[:offset]) candidate_refs = [ *defaults, *(str(model.model_ref) for model in pool.list(role=role)), ] ordered_refs = tuple(dict.fromkeys(candidate_refs)) default_refs = frozenset(defaults) prompt = _p( state.workspace_root, "role_model", role=role, ordinal="1", count="1", ) options: list[Option] = [] for model_ref in ordered_refs: availability = pool.availability( model_ref, role, state.host_runtime, "new-session", ) if not availability.available: continue candidate_models = { selected_role: tuple(models) for selected_role, models in state.role_models.items() } selected = list(candidate_models.get(role, ()))[: ordinal - 1] selected.append(model_ref) candidate_models[role] = tuple(selected) if not _role_selection_can_complete( state, profile, context, candidate_models, scopes, ): continue model = pool.resolve(model_ref) options.append(_opt( model_ref, prompt["options"]["model"].format( model_ref=model_ref, display=model.display_name, default_suffix=( prompt["options"].get("default_suffix", "") if model_ref in default_refs else "" ), ), )) if not options: _validate_role_selection_feasibility(state, profile, context, scopes) raise WizardError(f"role {role!r} has no executable model candidates") return options def _model_prompt( state: WizardState, profile: RoleProfile, requirement: RoleRequirement, ordinal: int, context: AssignmentContext, scopes: ModelDefaultScopes, ) -> Prompt: prompt = _p( state.workspace_root, "role_model", role=requirement.role, ordinal=str(ordinal), count=str(_selected_role_count(state, requirement)), ) return Prompt( step=_role_model_prompt_id(requirement.role, ordinal), kind="pick", label=prompt["label"], options=_role_model_options( state, profile, requirement.role, ordinal, context, scopes, ), echo_template=prompt["echo_template"], ) def _rewind_invalid_role_models( state: WizardState, requirements: tuple[RoleRequirement, ...], requirement_index: int, model_index: int, ) -> None: requirement = requirements[requirement_index] del state.role_models[requirement.role][model_index:] invalid_ids: set[str] = set() for later_index, later_requirement in enumerate( requirements[requirement_index:], start=requirement_index, ): first_ordinal = model_index + 1 if later_index == requirement_index else 1 later_count = _selected_role_count(state, later_requirement) invalid_ids.update( _role_model_prompt_id(later_requirement.role, ordinal) for ordinal in range(first_ordinal, later_count + 1) ) if later_index != requirement_index: state.role_models.pop(later_requirement.role, None) state.role_selection_order = [ step_id for step_id in state.role_selection_order if step_id not in invalid_ids ] state.answered = [ step_id for step_id in state.answered if step_id not in invalid_ids ] def _discard_invalid_previous_models( state: WizardState, profile: RoleProfile, requirements: tuple[RoleRequirement, ...], context: AssignmentContext, scopes: ModelDefaultScopes, ) -> None: selected_models = { role: tuple(models) for role, models in state.role_models.items() } if _role_selection_error( state, profile, context, selected_models, scopes, ) is None: return selected_prefix: dict[str, tuple[str, ...]] = {} for requirement_index, requirement in enumerate(requirements): count = _selected_role_count(state, requirement) selected = state.role_models.get(requirement.role, []) if len(selected) > count: raise WizardError( f"role {requirement.role!r} has {len(selected)} models for " f"{count} instances" ) for index, model_ref in enumerate(selected): candidate_prefix = dict(selected_prefix) candidate_prefix[requirement.role] = tuple(selected[: index + 1]) if _role_selection_can_complete( state, profile, context, candidate_prefix, scopes, ): continue _rewind_invalid_role_models( state, requirements, requirement_index, index, ) return if selected: selected_prefix[requirement.role] = tuple(selected) def _host_session_context(state: WizardState) -> HostSessionContext: native_provider = default_host_registry().resolve( state.host_runtime ).descriptor.native_provider_id return HostSessionContext( host_id=state.host_runtime, entry_mode=state.host_entry_mode, available_functions=frozenset(state.available_functions), interaction_surface="wizard", current_model=CurrentSessionModelAttestation.unknown(native_provider), ) def _role_selection_error( state: WizardState, profile: RoleProfile, context: AssignmentContext, role_models: dict[str, tuple[str, ...]], scopes: ModelDefaultScopes, ) -> str | None: try: resolve_assignments( profile=profile, role_counts=dict(state.role_counts), role_models=role_models, scopes=scopes, pool=context.pool, host=_host_session_context(state), environment=context.environment, ) except AssignmentResolutionError as exc: return _assignment_resolution_message(exc) except (PrepareError, ValueError, RuntimeError) as exc: return str(exc) return None def _executable_role_models( state: WizardState, profile: RoleProfile, requirement: RoleRequirement, context: AssignmentContext, ) -> tuple[str, ...]: host = _host_session_context(state) candidates: list[str] = [] instance = RoleInstance(requirement.role, requirement.duty, 1) for model in context.pool.list(role=requirement.role): model_ref = str(model.model_ref) try: resolve_model_assignment( instance=instance, model_ref=model_ref, pool=context.pool, host=host, environment=context.environment, ) except (AssignmentResolutionError, ValueError, RuntimeError): continue candidates.append(model_ref) return tuple(candidates) def _completed_role_models( state: WizardState, profile: RoleProfile, context: AssignmentContext, role_models: dict[str, tuple[str, ...]], ) -> dict[str, tuple[str, ...]] | None: completed = {role: list(models) for role, models in role_models.items()} for requirement in _selectable_static_requirements(state, profile): count = _selected_role_count(state, requirement) selected = completed.setdefault(requirement.role, []) if len(selected) > count: return None candidates = _executable_role_models(state, profile, requirement, context) if not candidates or any(model not in candidates for model in selected): return None # 같은 역할 패널의 model_ref 는 서로 달라야 한다. if len(set(selected)) < len(selected): return None used = set(selected) remaining = [model for model in candidates if model not in used] if len(remaining) < count - len(selected): return None while len(selected) < count: model_ref = next( model for model in candidates if model not in used ) selected.append(model_ref) used.add(model_ref) return {role: tuple(models) for role, models in completed.items()} def _role_selection_can_complete( state: WizardState, profile: RoleProfile, context: AssignmentContext, role_models: dict[str, tuple[str, ...]], scopes: ModelDefaultScopes, ) -> bool: completed = _completed_role_models(state, profile, context, role_models) return completed is not None and _role_selection_error( state, profile, context, completed, scopes, ) is None def _validate_role_selection_feasibility( state: WizardState, profile: RoleProfile, context: AssignmentContext | None = None, scopes: ModelDefaultScopes | None = None, ) -> None: assignment_context = context or load_assignment_context( host_runtime=state.host_runtime, terminal_backend=detect_terminal_backend(), ) selected_scopes = scopes or _model_default_scopes(Path(state.project_root)) error = _role_selection_error( state, profile, assignment_context, { role: tuple(models) for role, models in state.role_models.items() }, selected_scopes, ) if error is not None: raise WizardError(error) def next_role_prompt(state: WizardState) -> Prompt | None: """Ask every adjustable count before one model question per role instance.""" if not _role_selection_enabled(state) or not _identity_ready(state): return None state.use_defaults = False profile = _load_role_profile_for_state(state) for requirement in profile.roles: # 필수 수량: min < max 이고 min > 0 일 때만. if ( requirement.dynamic or requirement.min_count == requirement.max_count or requirement.min_count == 0 ): continue if requirement.role not in state.role_counts: return _count_prompt(state, requirement) count = state.role_counts[requirement.role] if count < requirement.min_count or count > requirement.max_count: raise WizardError( f"role {requirement.role!r} count must be in " f"{requirement.min_count}..{requirement.max_count}: {count}" ) for requirement in profile.roles: # 선택 역할(min=0, max>0): 기본은 추가 안 함. role-add 로만 연다. if ( requirement.dynamic or requirement.min_count != 0 or requirement.max_count == 0 ): continue if requirement.role not in state.role_counts: return _role_add_prompt(state, requirement) count = state.role_counts[requirement.role] if count < 0 or count > requirement.max_count: raise WizardError( f"role {requirement.role!r} count must be in " f"0..{requirement.max_count}: {count}" ) requirements = _selectable_static_requirements(state, profile) try: scopes = _model_default_scopes(Path(state.project_root)) except (PrepareError, ValueError) as exc: raise WizardError(str(exc)) from exc context = load_assignment_context( host_runtime=state.host_runtime, terminal_backend=detect_terminal_backend(), ) _discard_invalid_previous_models( state, profile, requirements, context, scopes, ) for requirement in requirements: count = _selected_role_count(state, requirement) selected = state.role_models.get(requirement.role, []) if len(selected) < count: return _model_prompt( state, profile, requirement, len(selected) + 1, context, scopes, ) _validate_role_selection_feasibility(state, profile, context, scopes) return None def _validate_submitted_role_model( state: WizardState, profile: RoleProfile, ) -> None: try: scopes = _model_default_scopes(Path(state.project_root)) except (PrepareError, ValueError) as exc: raise WizardError(str(exc)) from exc context = load_assignment_context( host_runtime=state.host_runtime, terminal_backend=detect_terminal_backend(), ) role_models = { role: tuple(models) for role, models in state.role_models.items() } requirements = _selectable_static_requirements(state, profile) complete = all( len(role_models.get(requirement.role, ())) == _selected_role_count(state, requirement) for requirement in requirements ) if complete: _validate_role_selection_feasibility(state, profile, context, scopes) return if not _role_selection_can_complete( state, profile, context, role_models, scopes, ): raise WizardError("selected role model leaves no complete role assignment") def _submit_role_prompt(state: WizardState, prompt: Prompt, value: str) -> str: if prompt.step.startswith("role-count:"): role = prompt.step.split(":", 1)[1] profile = _load_role_profile_for_state(state) requirement = next( (row for row in profile.roles if row.role == role), None, ) if ( requirement is None or requirement.dynamic or requirement.min_count == requirement.max_count or requirement.min_count == 0 ): raise WizardError(f"role {role!r} does not accept a count selection") try: count = int(value) except ValueError as exc: raise WizardError(f"role {role!r} count must be an integer") from exc if count < requirement.min_count or count > requirement.max_count: raise WizardError( f"role {role!r} count must be in " f"{requirement.min_count}..{requirement.max_count}: {count}" ) state.role_counts[role] = count state.role_models.pop(role, None) return f"role-count: {role}={count}" if prompt.step.startswith("role-add:"): role = prompt.step.split(":", 1)[1] profile = _load_role_profile_for_state(state) requirement = next( (row for row in profile.roles if row.role == role), None, ) if ( requirement is None or requirement.dynamic or requirement.min_count != 0 or requirement.max_count == 0 ): raise WizardError(f"role {role!r} does not accept an optional add") allowed = {option.value for option in prompt.options} if value not in allowed: raise WizardError( f"role {role!r} add selection must be one of " f"{sorted(allowed, key=int)}: {value}" ) try: count = int(value) except ValueError as exc: raise WizardError(f"role {role!r} count must be an integer") from exc if count < 0 or count > requirement.max_count: raise WizardError( f"role {role!r} count must be in " f"0..{requirement.max_count}: {count}" ) state.role_counts[role] = count state.role_models.pop(role, None) return f"role-add: {role}={count}" _, role, ordinal_raw = prompt.step.split(":", 2) ordinal = int(ordinal_raw) allowed = {option.value for option in prompt.options} if value not in allowed: raise WizardError( f"model {value!r} is not a compatible candidate for role {role!r}" ) selected = state.role_models.setdefault(role, []) if ordinal != len(selected) + 1: raise WizardError( f"role {role!r} model selection is out of order at ordinal {ordinal}" ) selected.append(value) try: profile = _load_role_profile_for_state(state) _validate_submitted_role_model(state, profile) except WizardError: selected.pop() raise return f"role-model: {role}#{ordinal}={value}" def _load_profile_workers(workspace_root: Path, task_type: str) -> list[str]: return resolve_profile_workers(_profile_path(workspace_root, task_type)) def _load_profile_optional_workers( workspace_root: Path, task_type: str ) -> list[str]: return resolve_optional_workers(_profile_path(workspace_root, task_type)) # --------------------------------------------------------------------------- # # Wizard prompt JSON SOT (Phase A1) # --------------------------------------------------------------------------- # _WIZARD_ROOT_CACHE: dict[str, dict] = {} def _load_wizard_root(workspace_root: str) -> dict: """Load and cache the full wizard prompts JSON root for the given workspace_root. Returns the entire parsed JSON object (with `schema_version`, `locale`, `steps`, optional `confirmation` etc. at top level). Callers that only need the steps dict should index `["steps"]` on the return value; `_msg` reads top-level sections like `confirmation`. """ if workspace_root in _WIZARD_ROOT_CACHE: return _WIZARD_ROOT_CACHE[workspace_root] path = Path(workspace_root) / "prompts" / "wizard" / "prompts.ko.json" if not path.is_file(): raise WizardError( f"wizard prompt SOT not found: {path}. " "Re-run `okstra install` or check the workspace_root." ) try: raw = load_owned_object(path, artifact="wizard prompts") except JsonBoundaryError as exc: raise WizardError(f"wizard prompt JSON malformed: {path}: {exc}") from exc if not isinstance(raw, dict): raise WizardError(f"wizard prompt JSON root must be an object: {path}") if raw.get("schema_version") != 1: raise WizardError( f"wizard prompt schema_version mismatch (expected 1): {path}" ) if raw.get("locale") != "ko": raise WizardError( f"wizard prompt locale unsupported (expected 'ko'): {path}" ) steps = raw.get("steps") if not isinstance(steps, dict): raise WizardError(f"wizard prompt 'steps' missing or not a dict: {path}") _WIZARD_ROOT_CACHE[workspace_root] = raw return raw def _p(workspace_root: str, step_id: str, **vars: str) -> dict: """Look up a wizard prompt entry by step_id and interpolate placeholders. Returns a dict with keys: 'label' (str, possibly interpolated), 'echo_template' (str, raw — contains `{value}` for the user's answer), 'options' (dict[value → label], may be empty). Returned dict also includes echo_variants and errors sub-dicts (may be empty). Raises WizardError if the step_id is unknown or a required placeholder is missing from the provided vars. """ steps = _load_wizard_root(workspace_root)["steps"] raw = steps.get(step_id) if raw is None: raise WizardError(f"unknown wizard step_id: {step_id!r}") label_template = raw.get("label", "") fv_label_template = raw.get("label_final_verification", "") unlinked_template = raw.get("label_unlinked", "") try: label = label_template.format(**vars) label_final_verification = fv_label_template.format(**vars) label_unlinked = unlinked_template.format(**vars) except KeyError as exc: missing = exc.args[0] if exc.args else "" raise WizardError( f"missing placeholder {missing!r} for wizard step {step_id!r}" ) from exc return { "label": label, "label_final_verification": label_final_verification, "label_unlinked": label_unlinked, "echo_template": raw.get("echo_template", ""), "options": raw.get("options", {}), "options_final_verification": raw.get("options_final_verification", {}), "options_html_approval": raw.get("options_html_approval", {}), "html_approval_note": raw.get("html_approval_note", ""), "html_approval_note_default_option": raw.get( "html_approval_note_default_option", ""), "echo_variants": raw.get("echo_variants", {}), "errors": raw.get("errors", {}), "labels": raw.get("labels", {}), "sidecar_note": raw.get("sidecar_note", {}), "echo_suffixes": raw.get("echo_suffixes", {}), "recent_label_prefix": raw.get("recent_label_prefix", ""), } def _msg(workspace_root: str, section: str, key: str, **vars: str) -> str: """Look up a top-level message (e.g., `confirmation.header`) and interpolate. Returns the string value at `
.`. Raises WizardError if the section or key is unknown, or if a required placeholder is missing. Use for wizard-level messages (confirmation block etc.). For step-scoped messages (echo_variants, errors), use `_p()` instead. """ root = _load_wizard_root(workspace_root) sect = root.get(section) if not isinstance(sect, dict): raise WizardError(f"unknown wizard section: {section!r}") template = sect.get(key) if template is None: raise WizardError(f"unknown wizard message: {section}.{key!r}") try: return template.format(**vars) except KeyError as exc: missing = exc.args[0] if exc.args else "" raise WizardError( f"missing placeholder {missing!r} for wizard message {section}.{key!r}" ) from exc def _static_options(t: dict) -> list[tuple[str, str]]: """Return (value, label) pairs from the SOT's static options dict, excluding suffix-decoration tokens (`__SUFFIX` / `__LABEL`). Used by hybrid `_build_*` functions that mix dynamic and static options to avoid leaking `_RECOMMENDED_SUFFIX` / `_OPTIONAL_SUFFIX` / `_DEFAULT_SUFFIX` as a literal option entry. """ return [ (k, v) for k, v in t.get("options", {}).items() if not (k.startswith("_") and (k.endswith("_SUFFIX") or k.endswith("_LABEL"))) ] def _resolved_roster(state: WizardState) -> list[str]: """Effective worker list AFTER override. Implementation: profile default (caller never asks for override). Others: override or profile default.""" if state.task_type == "implementation": roster = list(state.profile_workers) # implementation 은 roster override 단계를 건너뛰므로, executor 로 명시 # 선택한 워커(예: antigravity)가 프로필 기본 roster 에 없으면 여기서 # 합류시켜야 render-bundle 의 executor∈roster 가드를 통과한다. if state.executor and state.executor not in roster: roster.append(state.executor) return roster if state.workers_override.strip(): return normalize_workers(state.workers_override) return list(state.profile_workers) # ---- Worktree resolution ------------------------------------------------ def _resolve_reuse_worktree(state: WizardState) -> bool: """For a finalized task identity, is there an active worktree to reuse? New tasks always answer False (no entry possible).""" if state.is_new_task: return False if not (state.project_id and state.task_group and state.task_id): return False entry = worktree_registry.lookup(state.project_id, state.task_group, state.task_id) return bool(entry and entry.status == "active") def _base_ref_required(state: WizardState) -> bool: return state.task_type != "final-verification" and state.reuse_worktree is False def _brief_resolved(state: WizardState) -> bool: """brief 입력이 끝났는가. release-handoff 는 brief 가 없는 phase 라 항상 True.""" return bool(state.brief_path) or state.task_type == "release-handoff" def _base_ref_ready(state: WizardState) -> bool: return not _base_ref_required(state) or S_BASE_REF_PICK in state.answered def _worktree_preview_required(state: WizardState) -> bool: """final-verification 은 stage worktree 를 읽기 전용으로 재사용하므로 별도 worktree 미리보기 줄을 confirmation 블록에 넣지 않는다.""" return state.task_type != "final-verification" def _fix_cycle_confirm_required(state: WizardState) -> bool: """완료(release-handoff) task 에 entry phase 로 재진입하고, 아직 열린 fix cycle 이 없을 때만 묻는다.""" if state.task_type not in fix_cycles.FIX_CYCLE_ENTRY_PHASES: return False task_root = task_dir(Path(state.project_root), state.task_group, state.task_id) workflow = (read_task_manifest(task_root) or {}).get("workflow") or {} if workflow.get("lastCompletedPhase") != "release-handoff": return False return fix_cycles.open_cycle(fix_cycles.read_rows(task_root)) is None def _parse_stage_objects(state: WizardState) -> list: """Return the approved plan's strict Stage Map objects for the picker.""" try: return parse_stage_map_file(Path(state.approved_plan_path)) except StageMapError as exc: raise WizardError( f"approved plan 의 Stage Map 을 신뢰할 수 없습니다 " f"({state.approved_plan_path}): {exc.reason}. plan 의 Stage Map 을 점검하세요." ) from exc def _stage_lifecycle_snapshot( state: WizardState, stages: list, *, reserved_stages: Optional[set] = None, ): """picker 가 보는 Stage Lifecycle Snapshot. 순수 읽기 — carry backfill 금지. backfill_done_from_carry 는 consumers.jsonl 에 done 행을 append 하고 그 부수효과로 worktree-registry stage 점유를 해제한다. 이 스냅샷은 stage picker 표시와 progress 추정(_remaining_screens 가 매 step 반복 호출)에서만 쓰이므로, 화면을 그리는 것만으로 동시 실행 중인 run 의 점유를 풀면 안 된다. carry 기반 done 보정은 prepare 게이트(run.py)가 단일 진입점으로 수행한다. One snapshot per call site: the two ledger reads this replaced ran once per helper, and `_build_stage_pick` sits inside `_remaining_screens`' simulation. """ from .stage_targets import read_stage_lifecycle_snapshot return read_stage_lifecycle_snapshot( [{"stage_number": s.stage_number, "depends_on": list(s.depends_on), "step_count": s.step_count} for s in stages], Path(state.approved_plan_path).resolve().parents[1], recover_from_carry=False, reserved_stages=reserved_stages, ) def _reserved_stage_numbers(state: WizardState) -> set: """worktree-registry 가 active 로 잡고 있는 stage 번호 집합(점유 SSOT). implementation prepare 와 동일 인자로 list_active_stage_numbers 를 호출한다.""" if not (state.project_id and state.task_group and state.task_id): return set() from .worktree_registry import list_active_stage_numbers return list_active_stage_numbers( state.project_id, state.task_group, state.task_id) def _whole_task_allowed( state: WizardState, *, stages: Optional[list] = None, done: Optional[set] = None, ) -> bool: """final-verification 이고 Stage Map 의 모든 stage 가 done 일 때만 True. 위저드는 done 만 본다 — 머지/clean/active 는 prepare 게이트가 강제한다. `stages`/`done` 를 넘기면 재파싱(validator exec_module)·재읽기를 생략한다 — 이미 둘을 계산한 `_build_stage_pick` 의 hot path 중복을 없애기 위한 seam.""" if state.task_type != "final-verification": return False if not state.approved_plan_path: return False if stages is None: stages = _parse_stage_objects(state) if not stages: return False if done is None: done = _stage_lifecycle_snapshot(state, stages).done_stages return all(s.stage_number in done for s in stages) def _existing_task_brief(project_root: Path, task_key: str) -> str: """Read taskBriefPath from manifest for an existing task. Empty if none.""" root = find_task_root(project_root, task_key) if root is None: return "" manifest = read_task_manifest(root) or {} val = manifest.get("taskBriefPath") or "" return val if isinstance(val, str) else "" # ---- 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 _opt(value: str, label: str = "", description: str = "") -> Option: return Option(value=value, label=label or value, description=description) # --- builders --- def _contract_outcome_suffix(entry: dict) -> str: outcome = entry.get("phaseOutcome") implementation = outcome.get("implementation") if isinstance(outcome, dict) else {} if not isinstance(implementation, dict): return "" if implementation.get("state") != "completed": return "" if entry.get("currentStatus") == "contract-violated": return " · implementation complete; contract warnings" return "" def _next_phase_cell(raw: Any) -> str: """task picker 한 줄에 실을 다음 phase 표기. `phase` 만 싣고 `status` 를 버리면 "아직 여기 있다" 가 "다음으로 갈 수 있다" 로 읽힌다. prepare 는 끝나지 않은 run 의 포인터를 `ready` → `pending` 으로 내리면서 `phase` 는 남겨두므로(`render._derive_next_recommended_phase`), 준비만 된 implementation 태스크의 포인터가 `{"phase": "final-verification", "status": "pending"}` 이다. status 를 안 보면 그 줄이 `next: final-verification` 으로 찍혀 착수 가능한 것처럼 읽힌다. 그래서 `ready` 가 아닌 status 는 괄호로 함께 싣는다. `okstra-inspect` 의 status 표는 같은 정보를 `*` 마커로 싣지만(`facets/status.md`), 그쪽은 표 아래 범례 한 줄을 붙일 수 있는 자리다. picker 옵션은 라벨 한 줄이 전부라 범례를 둘 데가 없으므로 status 이름을 그대로 적는다. phase 를 `--` 로 지우지 않는 이유는 그 이름이 "어디로 가는 태스크인가" 라는 정보를 여전히 나르기 때문이다 — 지우는 것은 착수 가능성만이 아니라 목적지까지 지운다. """ pointer = next_phase.promote(raw) cell = pointer["phase"] or "--" if pointer["status"] == next_phase.STATUS_READY: return cell return f"{cell} ({pointer['status']})" def _build_task_pick(state: WizardState) -> Prompt: t = _p(state.workspace_root, "task_pick") project_root = Path(state.project_root) tasks = list_project_tasks(project_root) latest = read_latest_task(project_root) or {} latest_key = latest.get("taskKey") or "" latest_suffix = t["options"].get("_LATEST_SUFFIX", "") remaining = [e for e in tasks if (e.get("workStatus") or "") != "done"] options: list[Option] = [] for entry in remaining[:_RECOMMENDATION_CAP]: key = entry.get("taskKey") or "" ttype = entry.get("taskType") or "" # catalog entries are flat (render_task_catalog_discovery) — there is # no nested "workflow" object here, unlike task-manifest.json. phase = entry.get("currentPhase") or ttype nxt = _next_phase_cell(entry.get("nextRecommendedPhase")) suffix = latest_suffix if key == latest_key else "" label = f"{key} · {phase} · next: {nxt}{_contract_outcome_suffix(entry)}{suffix}" options.append(_opt(value=key, label=label)) for value, label in _static_options(t): options.append(_opt(value=value, label=label)) return Prompt(step=S_TASK_PICK, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"]) def _submit_task_pick(state: WizardState, value: str) -> Optional[str]: value = value.strip() if value == TASK_PICK_NEW_TOKEN: state.is_new_task = True state.task_group = "" state.task_id = "" state.existing_brief_path = "" state.task_type = "" return "task: (brand-new)" # parse "::" parts = value.split(":") if len(parts) != 3 or not all(parts): raise WizardError( f"invalid task-key: {value!r} (expected project-id:task-group:task-id)" ) pid, tg, tid = parts if pid != state.project_id: raise WizardError( f"task-key project-id {pid!r} does not match current project {state.project_id!r}" ) state.is_new_task = False state.task_group = tg state.task_id = tid state.existing_brief_path = _existing_task_brief( Path(state.project_root), value ) # task_type seeded from manifest's nextRecommendedPhase as the recommended default root = find_task_root(Path(state.project_root), value) manifest = (read_task_manifest(root) or {}) if root else {} state.task_type = next_phase.autofill_task_type(manifest) return f"task: {value}" def _suggest_recent_task_groups( state: WizardState, limit: int = _RECOMMENDATION_CAP ) -> list[str]: """최근 task/brief 활동에서 task-group 후보를 limit 개까지 반환.""" if not state.project_root: return [] project_root = Path(state.project_root) scores: dict[str, float] = {} orders: dict[str, int] = {} try: tasks = list_project_tasks(project_root) except (OSError, StateError): tasks = [] for index, entry in enumerate(tasks): tg = entry.get("taskGroup") or "" if not tg: continue scores[tg] = max( scores.get(tg, 0.0), _parse_iso_timestamp(entry.get("updatedAt")), ) orders[tg] = min(orders.get(tg, index), index) brief_root = project_root / ".okstra" / "briefs" if brief_root.is_dir(): next_order = len(orders) for group_dir in brief_root.iterdir(): if not group_dir.is_dir(): continue latest_brief = max( ( _file_recency(path) for path in group_dir.rglob("*.md") if path.is_file() ), default=0.0, ) if latest_brief <= 0.0: continue group = group_dir.name scores[group] = max(scores.get(group, 0.0), latest_brief) orders.setdefault(group, next_order) next_order += 1 return sorted( scores, key=lambda group: (-scores[group], orders[group], group), )[:limit] def _suggest_recent_task_ids( state: WizardState, limit: int = _RECOMMENDATION_CAP ) -> list[str]: """현재 task_group 내의 최근 task-id 후보를 limit 개까지 반환.""" if not state.project_root or not state.task_group: return [] try: tasks = list_project_tasks(Path(state.project_root)) except (OSError, StateError): return [] seen: list[str] = [] for entry in tasks: tg = entry.get("taskGroup") or "" if tg != state.task_group: continue tid = entry.get("taskId") or "" if tid and tid not in seen: seen.append(tid) if len(seen) >= limit: break return seen def _recently_used_brief_times(state: WizardState) -> dict[str, float]: if not state.project_root or not state.task_group: return {} project_root = Path(state.project_root) try: tasks = list_project_tasks(project_root) except (OSError, StateError): return {} out: dict[str, float] = {} for entry in tasks: if entry.get("taskGroup") != state.task_group: continue task_root = entry.get("_resolvedTaskRoot") or "" manifest = read_task_manifest(Path(task_root)) if task_root else None brief_path = "" if isinstance(manifest, dict): value = manifest.get("taskBriefPath") brief_path = value if isinstance(value, str) else "" if not brief_path: value = entry.get("taskBriefPath") brief_path = value if isinstance(value, str) else "" if not brief_path: continue relpath = _project_relative_value(brief_path, project_root) out[relpath] = max( out.get(relpath, 0.0), _parse_iso_timestamp(entry.get("updatedAt")), ) return out def _suggest_group_briefs(state: WizardState, limit: int = 6) -> list[str]: """Return recent okstra-brief-gen outputs for the selected task-group. ``okstra-brief-gen`` writes to ``.okstra/briefs//**/*.md``. The wizard exposes those paths after task-group selection so users can pick a generated brief instead of typing its path. Paths are project-relative. """ if not state.project_root or not state.task_group: return [] project_root = Path(state.project_root) root = project_root / ".okstra" / "briefs" / state.task_group if not root.is_dir(): return [] used_at_by_relpath = _recently_used_brief_times(state) candidates: list[tuple[float, str]] = [] for path in root.rglob("*.md"): if not path.is_file(): continue tg_suggestion, _ = _brief_suggestions(path) if tg_suggestion: try: if _slug_or_die(tg_suggestion, "task_group") != state.task_group: continue except WizardError: continue relpath = _project_relative_path(path, project_root) recency = max(_file_recency(path), used_at_by_relpath.get(relpath, 0.0)) candidates.append((recency, relpath)) candidates.sort(key=lambda item: (-item[0], item[1])) return [rel for _, rel in candidates[:limit]] def _build_task_group(state: WizardState) -> Prompt: sugg = state.task_group_suggestion if sugg: t = _p(state.workspace_root, "task_group_with_suggestion", suggestion=sugg) options = [ _opt(k, v.format(suggestion=sugg)) for k, v in t["options"].items() ] return Prompt( step=S_TASK_GROUP, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) # suggestion 이 없으면 catalog 의 최근 task-group 을 후보로 노출 + 직접 입력 recent = _suggest_recent_task_groups(state) t = _p(state.workspace_root, "task_group_no_suggestion") recent_prefix = t.get("recent_label_prefix", "") options: list[Option] = [] for tg in recent: options.append(_opt(f"{_RECENT_PREFIX}{tg}", f"{recent_prefix}{tg}")) options.append(_opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM])) return Prompt( step=S_TASK_GROUP, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_task_group(state: WizardState, value: str) -> Optional[str]: if state.task_group_suggestion: if value == PICK_USE_SUGGESTED: state.task_group = _slug_or_die( state.task_group_suggestion, "task_group" ) state.task_group_pending_text = False return f"task-group: {state.task_group} (brief)" if value == PICK_TYPE_CUSTOM: state.task_group_pending_text = True t = _p(state.workspace_root, "task_group_with_suggestion", suggestion=state.task_group_suggestion) return t["echo_variants"]["free_input"] raise WizardError( f"expected {PICK_USE_SUGGESTED!r} or {PICK_TYPE_CUSTOM!r}, " f"got: {value!r}" ) # suggestion-없음 분기 (Task 2 신규) if value.startswith(_RECENT_PREFIX): tg = value[len(_RECENT_PREFIX):] state.task_group = _slug_or_die(tg, "task_group") state.task_group_pending_text = False return f"task-group: {state.task_group} (recent)" if value == PICK_TYPE_CUSTOM: state.task_group_pending_text = True t = _p(state.workspace_root, "task_group_no_suggestion") return t["echo_variants"]["free_input"] raise WizardError( f"unexpected task-group value: {value!r} " f"(expected {PICK_USE_SUGGESTED!r}, {PICK_TYPE_CUSTOM!r}, or '{_RECENT_PREFIX}')" ) def _build_task_group_text(state: WizardState) -> Prompt: t = _p(state.workspace_root, "task_group_text") return Prompt(step=S_TASK_GROUP_TEXT, kind="text", label=t["label"], echo_template=t["echo_template"]) def _submit_task_group_text(state: WizardState, value: str) -> Optional[str]: state.task_group = _slug_or_die(value, "task_group") state.task_group_pending_text = False return f"task-group: {state.task_group}" def _build_task_id(state: WizardState) -> Prompt: sugg = state.task_id_suggestion if sugg: t = _p(state.workspace_root, "task_id_with_suggestion", suggestion=sugg) options = [ _opt(k, v.format(suggestion=sugg)) for k, v in t["options"].items() ] return Prompt( step=S_TASK_ID, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) recent = _suggest_recent_task_ids(state) t = _p(state.workspace_root, "task_id_no_suggestion") recent_prefix = t.get("recent_label_prefix", "") options: list[Option] = [] for tid in recent: options.append(_opt(f"{_RECENT_PREFIX}{tid}", f"{recent_prefix}{tid}")) options.append(_opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM])) return Prompt( step=S_TASK_ID, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_task_id(state: WizardState, value: str) -> Optional[str]: if state.task_id_suggestion: if value == PICK_USE_SUGGESTED: state.task_id = _slug_or_die(state.task_id_suggestion, "task_id") state.task_id_pending_text = False return f"task-id: {state.task_id} (brief)" if value == PICK_TYPE_CUSTOM: state.task_id_pending_text = True t = _p(state.workspace_root, "task_id_with_suggestion", suggestion=state.task_id_suggestion) return t["echo_variants"]["free_input"] raise WizardError( f"expected {PICK_USE_SUGGESTED!r} or {PICK_TYPE_CUSTOM!r}, " f"got: {value!r}" ) # suggestion-없음 분기 if value.startswith(_RECENT_PREFIX): tid = value[len(_RECENT_PREFIX):] state.task_id = _slug_or_die(tid, "task_id") state.task_id_pending_text = False return f"task-id: {state.task_id} (recent)" if value == PICK_TYPE_CUSTOM: state.task_id_pending_text = True t = _p(state.workspace_root, "task_id_no_suggestion") return t["echo_variants"]["free_input"] raise WizardError( f"unexpected task-id value: {value!r} " f"(expected {PICK_USE_SUGGESTED!r}, {PICK_TYPE_CUSTOM!r}, " f"or '{_RECENT_PREFIX}')" ) def _build_task_id_text(state: WizardState) -> Prompt: t = _p(state.workspace_root, "task_id_text") return Prompt(step=S_TASK_ID_TEXT, kind="text", label=t["label"], echo_template=t["echo_template"]) def _submit_task_id_text(state: WizardState, value: str) -> Optional[str]: state.task_id = _slug_or_die(value, "task_id") state.task_id_pending_text = False return f"task-id: {state.task_id}" def _existing_task_workflow(state: WizardState) -> dict: """현재 task-key 가 이미 존재하면 그 manifest 의 workflow dict 를 반환한다. picker 로 기존 task 를 고른 경우뿐 아니라, new-task 흐름으로 같은 task-group/task-id 를 다시 입력한 경우(=사실상 이어가기)에도 직전 phase 기반 추천이 끊기지 않게 하는 안전장치. 없으면 {}.""" if not (state.project_id and state.task_group and state.task_id): return {} key = f"{state.project_id}:{state.task_group}:{state.task_id}" root = find_task_root(Path(state.project_root), key) if root is None: return {} workflow = (read_task_manifest(root) or {}).get("workflow") or {} return workflow if isinstance(workflow, dict) else {} def _phase_after(task_type: str) -> str: """라이프사이클(PHASE_SEQUENCE) 상 task_type 바로 다음 단계. 없으면 ''.""" try: idx = PHASE_SEQUENCE.index(task_type) except ValueError: return "" return PHASE_SEQUENCE[idx + 1] if idx + 1 < len(PHASE_SEQUENCE) else "" def _recent_task_types(state: WizardState) -> list[str]: """catalog 최신순으로 이 프로젝트에서 최근 사용된 task-type 목록(중복 제거).""" if not state.project_root: return [] try: tasks = list_project_tasks(Path(state.project_root)) except (OSError, StateError): return [] out: list[str] = [] for entry in tasks: tt = entry.get("taskType") or "" if tt and tt not in out: out.append(tt) return out def _contained_final_report_data_path(report: Path, task_root: Path) -> Path: _resolved_within(report, task_root, "final report path") return _resolved_within( final_report_data_path(report), task_root, "final report data path", ) def _newest_contained_final_report( runs_base: Path, task_root: Path, glob_pattern: str, project_root: Path, ) -> Path | None: candidates: list[Path] = [] for report in runs_base.glob(glob_pattern): if not report.is_file(): continue try: _contained_final_report_data_path(report, task_root) if report.parent.parent.name in ANALYSIS_TASK_TYPES: load_analysis_report_candidate(project_root, report) except AnalysisInputError: continue candidates.append(report) def mtime_safe(report: Path) -> float: try: return report.stat().st_mtime except OSError: return -1.0 return max(candidates, key=mtime_safe) if candidates else None def _analysis_revision_candidate( state: WizardState, report: Path, task_root: Path, expected_task_key: str, ) -> tuple[int, Path] | None: try: _contained_final_report_data_path(report, task_root) candidate = load_analysis_report_candidate( Path(state.project_root), report ) except AnalysisInputError: return None if ( candidate.task_key != expected_task_key or candidate.task_type != state.task_type or candidate.review_status != "revision-requested" ): return None return int(candidate.run_seq), report def _latest_revision_requested_analysis_report( state: WizardState, ) -> Path | None: if state.task_type not in ANALYSIS_TASK_TYPES: return None reports = task_runs_dir( Path(state.project_root), state.task_group, state.task_id ) / state.task_type / "reports" try: task_root = _resolved_within( reports.parents[2], Path(state.project_root).resolve(), "analysis task root", ) except AnalysisInputError: return None expected_task_key = f"{state.project_id}:{state.task_group}:{state.task_id}" candidates: list[tuple[int, Path]] = [] for report in reports.glob("final-report-*.data.json"): candidate = _analysis_revision_candidate( state, report, task_root, expected_task_key, ) if candidate is not None: candidates.append(candidate) if not candidates: return None return max(candidates, key=lambda candidate: candidate[0])[1] def _latest_revision_requested_analysis_type(state: WizardState) -> str: report = _latest_revision_requested_analysis_report(state) return state.task_type if report is not None else "" def _build_task_type(state: WizardState) -> Prompt: t = _p(state.workspace_root, "task_type") recommended_suffix = t["options"].get("_RECOMMENDED_SUFFIX", "") rerun_suffix = t["options"].get("_RERUN_SUFFIX", "") next_suffix = t["options"].get("_NEXT_SUFFIX", "") approve_suffix = t["options"].get("_APPROVE_SUFFIX", recommended_suffix) blocked_rerun_suffix = t["options"].get("_BLOCKED_RERUN_SUFFIX", rerun_suffix) description_by_type = dict(TASK_TYPES) options: list[Option] = [] def add(task_type: str, suffix: str = "") -> None: if task_type not in description_by_type: return if any(o.value == task_type for o in options): return if len(options) >= _RECOMMENDATION_CAP: return options.append(_opt(task_type, f"{task_type}{suffix}", description_by_type[task_type])) workflow = _existing_task_workflow(state) revision_requested = _latest_revision_requested_analysis_type(state) # 포인터에서 phase 를 꺼내는 것은 `status` 가 `ready` 일 때뿐이다. prepare 는 # 실행이 끝나지 않은 run 의 포인터를 일부러 `pending` 으로 내려두는데 # (`render._derive_next_recommended_phase`), 여기서 status 를 안 보고 `phase` # 만 꺼내면 그 방어를 그대로 우회한다 — 기록된 사례가 `implementation` 이 # `prepared` 상태인데 `final-verification` 이 추천으로 뜬 것이다. 판정은 # 셸 진입점이 쓰는 것과 같은 함수(`next_phase.autofill_task_type`)로 한다. # # `ready` 가 아니면 추천은 비고, 아래 `currentPhase` 재실행 옵션이 남는다. # 실패한 run 의 포인터가 `{"phase": "", "status": "blocked"}` 라는 점에서 # 그것이 맞는 제안이다 — 그 옵션은 포인터가 아니라 `currentPhase` 에서 온다. # 계획 승인 대기는 `awaitingApproval` 로 표시한다. 구현이 추천이지만 먼저 # 승인을 받아야 하므로 접미사로 구분한다. 열린 C-NNN 때문에 blocked 면 # 재실행은 답이 기록된 뒤에만 고르라고 접미사로 말한다. recommended = (revision_requested or state.task_type or next_phase.autofill_task_type({"workflow": workflow})) if not recommended and not workflow: recommended = TASK_TYPE_VALUES[0] pointer = next_phase.promote(workflow.get("nextRecommendedPhase")) if workflow.get("awaitingApproval") is True and recommended == "implementation": recommended_suffix = approve_suffix if ( pointer["status"] == next_phase.STATUS_BLOCKED and (workflow.get("currentPhase") or "") == "implementation-planning" ): rerun_suffix = blocked_rerun_suffix add(recommended, recommended_suffix) add(workflow.get("currentPhase") or "", rerun_suffix) add(_phase_after(recommended), next_suffix) for tt in _recent_task_types(state): add(tt) for tt in TASK_TYPE_VALUES: add(tt) options.append(_opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM])) return Prompt(step=S_TASK_TYPE, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"]) def _carry_in_existing_brief(state: WizardState) -> str: """downstream task-type 이 등록된 brief 를 자동 carry-in 한다. 주입한 경로(또는 '').""" if state.task_type in _BRIEF_ENTRY_TASK_TYPES: return "" if state.task_type == "release-handoff": return "" # prepare 가 검증 보고서 인용 input 문서를 자동 생성한다 if state.brief_path or not state.existing_brief_path: return "" p = Path(state.existing_brief_path) if not p.is_absolute(): p = Path(state.project_root) / p if not p.is_file(): return "" # manifest 경로가 깨졌으면 brief_carry 단계가 처리한다 state.brief_path = state.existing_brief_path return state.brief_path def _apply_task_type(state: WizardState, value: str) -> str: if value not in TASK_TYPE_VALUES: raise WizardError( f"unknown task-type: {value!r} " f"(expected one of: {', '.join(TASK_TYPE_VALUES)})" ) state.task_type = value # brief_carry 의 "entry phase 로 전환" 이 task-type 을 리셋한 뒤 submit() 이 # brief_carry 를 answered 로 되돌려 놓는다 — task-type 을 다시 고르는 시점에 # 퍼지해야 downstream 재선택 시 carry 단계가 다시 나온다. state.answered = [a for a in state.answered if a != S_BRIEF_CARRY] carried = _carry_in_existing_brief(state) state.profile_workers = _load_profile_workers( Path(state.workspace_root), value ) state.profile_optional_workers = _load_profile_optional_workers( Path(state.workspace_root), value ) # Reuse-worktree is decided once identity is final. Recompute here so # subsequent base-ref step knows whether to apply. state.reuse_worktree = _resolve_reuse_worktree(state) if carried: return f"task-type: {value}\nbrief (carry-in): {carried}" return f"task-type: {value}" def _submit_task_type(state: WizardState, value: str) -> Optional[str]: # PICK_TYPE_CUSTOM leaves task_type empty while S_TASK_TYPE gets marked # answered — that combination is what gates S_TASK_TYPE_TEXT on. if value == PICK_TYPE_CUSTOM: state.task_type = "" t = _p(state.workspace_root, "task_type") return t["echo_variants"]["free_input"] return _apply_task_type(state, value) def _build_task_type_text(state: WizardState) -> Prompt: t = _p(state.workspace_root, "task_type_text") return Prompt(step=S_TASK_TYPE_TEXT, kind="text", label=t["label"], echo_template=t["echo_template"]) def _submit_task_type_text(state: WizardState, value: str) -> Optional[str]: return _apply_task_type(state, value.strip()) def _build_brief_keep(state: WizardState) -> Prompt: t = _p(state.workspace_root, "brief_keep", existing_brief_path=state.existing_brief_path) return Prompt( step=S_BRIEF_KEEP, kind="pick", label=t["label"], options=[_opt(k, v) for k, v in t["options"].items()], echo_template=t["echo_template"], ) def _submit_brief_keep(state: WizardState, value: str) -> Optional[str]: if value not in ("keep", "change"): raise WizardError(f"expected 'keep' or 'change', got: {value!r}") state.keep_existing_brief = value == "keep" if state.keep_existing_brief: state.brief_path = state.existing_brief_path t = _p(state.workspace_root, "brief_keep", existing_brief_path=state.existing_brief_path) return t["echo_variants"]["kept"].format(brief_path=state.brief_path) return None # next prompt is S_BRIEF_PATH def _suggest_brief_path(state: WizardState) -> tuple[str, str]: """Return (existing_brief_relpath_or_empty, standard_relpath). standard_relpath = ".okstra/tasks///brief.md".""" existing = state.existing_brief_path or "" tg = slugify_task_segment(state.task_group) if state.task_group else "" tid = slugify_task_segment(state.task_id) if state.task_id else "" if tg and tid: standard = str(Path(".okstra") / "tasks" / tg / tid / "brief.md") else: standard = "" return existing, standard def _build_brief_path_pick(state: WizardState) -> Prompt: existing, standard = _suggest_brief_path(state) t = _p(state.workspace_root, "brief_path_pick") options: list[Option] = [] if existing: options.append(_opt("__existing__", t["options"]["__existing__"].format(existing=existing))) if standard and standard != existing: options.append(_opt("__standard__", t["options"]["__standard__"].format(standard=standard))) brief_label = t["labels"].get("brief_candidate", "{path}") for relpath in _suggest_group_briefs(state): if relpath in (existing, standard): continue options.append(_opt(f"{_BRIEF_PREFIX}{relpath}", brief_label.format(path=relpath))) options.append(_opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM])) return Prompt(step=S_BRIEF_PATH_PICK, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"]) def _submit_brief_path_pick(state: WizardState, value: str) -> Optional[str]: existing, standard = _suggest_brief_path(state) if value == "__existing__": if not existing: t = _p(state.workspace_root, "brief_path_pick") raise WizardError(t["errors"]["existing_missing"]) p = _require_file(existing, Path(state.project_root), "task brief") _accept_brief_path(state, p) return f"brief: {p}" if value == "__standard__": p = _require_file(standard, Path(state.project_root), "task brief") _accept_brief_path(state, p) return f"brief: {p}" if value.startswith(_BRIEF_PREFIX): relpath = value[len(_BRIEF_PREFIX):] p = _require_file(relpath, Path(state.project_root), "task brief") _accept_brief_path(state, p) return f"brief: {p}" if value == PICK_TYPE_CUSTOM: state.brief_path_pending_text = True state.brief_path = "" return None raise WizardError( f"expected '__existing__', '__standard__', '{_BRIEF_PREFIX}', " f"or {PICK_TYPE_CUSTOM!r}, " f"got: {value!r}" ) def _build_brief_path(state: WizardState) -> Prompt: t = _p(state.workspace_root, "brief_path") return Prompt( step=S_BRIEF_PATH, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_brief_path(state: WizardState, value: str) -> Optional[str]: p = _require_file(value, Path(state.project_root), "task brief") _accept_brief_path(state, p) return f"brief: {p}" BRIEF_CARRY_SWITCH_ENTRY = "__switch_entry__" BRIEF_CARRY_ABORT = "__abort__" def _build_brief_carry(state: WizardState) -> Prompt: t = _p(state.workspace_root, "brief_carry", task_type=state.task_type) return Prompt( step=S_BRIEF_CARRY, kind="pick", label=t["label"], options=[_opt(k, v) for k, v in t["options"].items()], echo_template=t["echo_template"], ) def _submit_brief_carry(state: WizardState, value: str) -> Optional[str]: t = _p(state.workspace_root, "brief_carry", task_type=state.task_type) if value == BRIEF_CARRY_SWITCH_ENTRY: _reset_from(state, S_TASK_TYPE) return t["echo_variants"]["switch_entry"] if value == PICK_TYPE_CUSTOM: state.brief_path_pending_text = True return None if value == BRIEF_CARRY_ABORT: state.aborted = True return t["echo_variants"]["abort"] raise WizardError( f"expected '{BRIEF_CARRY_SWITCH_ENTRY}', {PICK_TYPE_CUSTOM!r}, " f"or '{BRIEF_CARRY_ABORT}', got: {value!r}" ) def _analysis_current_commit(state: WizardState) -> str: try: return subprocess.check_output( ["git", "-C", state.project_root, "rev-parse", "HEAD"], text=True, stderr=subprocess.DEVNULL, ).strip() except (OSError, subprocess.CalledProcessError) as exc: raise WizardError( f"cannot resolve current project commit for analysis evidence: {state.project_root}" ) from exc def _analysis_evidence_paths(state: WizardState) -> list[Path]: return [ Path(path) for path in (state.feature_evidence_path, state.project_evidence_path) if path ] def _resolve_analysis_evidence(state: WizardState): try: return resolve_evidence_inputs( Path(state.project_root), state.task_type, _analysis_evidence_paths(state), _analysis_current_commit(state), ) except AnalysisInputError as exc: raise WizardError(str(exc)) from exc def _accept_analysis_evidence( state: WizardState, raw_path: str, field_name: str ) -> AnalysisReportCandidate: report_path = _resolve_path(raw_path, Path(state.project_root)) try: candidate = load_analysis_report_candidate( Path(state.project_root), report_path ) except AnalysisInputError as exc: raise WizardError(str(exc)) from exc previous = getattr(state, field_name) setattr(state, field_name, str(candidate.report_path)) try: _resolve_analysis_evidence(state) except WizardError: setattr(state, field_name, previous) raise return candidate def _evidence_option_label(candidate: AnalysisReportCandidate) -> str: return f"{candidate.task_key} · {candidate.task_type} · run {candidate.run_seq}" def _build_analysis_evidence_pick( state: WizardState, *, step: str, relation: str ) -> Prompt: t = _p(state.workspace_root, step) candidates = list_evidence_candidates( Path(state.project_root), state.task_type, relation )[:2] options = [ _opt(str(candidate.report_path), _evidence_option_label(candidate)) for candidate in candidates ] options.extend([ _opt(PICK_SKIP, t["options"][PICK_SKIP]), _opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM]), ]) return Prompt(step=step, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"]) def _submit_analysis_evidence_pick( state: WizardState, value: str, *, step: str, field_name: str, pending_name: str, target_after_skip: bool = False, ) -> Optional[str]: t = _p(state.workspace_root, step) if value == PICK_SKIP: setattr(state, field_name, "") setattr(state, pending_name, False) if target_after_skip: state.analysis_target_pending_text = True return t["echo_variants"]["skip"] if value == PICK_TYPE_CUSTOM: setattr(state, pending_name, True) return None candidate = _accept_analysis_evidence(state, value, field_name) setattr(state, pending_name, False) return t["echo_variants"]["selected"].format(path=candidate.report_path) def _build_feature_evidence_pick(state: WizardState) -> Prompt: return _build_analysis_evidence_pick( state, step=S_FEATURE_EVIDENCE_PICK, relation="feature-baseline" ) def _submit_feature_evidence_pick(state: WizardState, value: str) -> Optional[str]: return _submit_analysis_evidence_pick( state, value, step=S_FEATURE_EVIDENCE_PICK, field_name="feature_evidence_path", pending_name="feature_evidence_pending_text", ) def _build_feature_evidence(state: WizardState) -> Prompt: t = _p(state.workspace_root, S_FEATURE_EVIDENCE) return Prompt(step=S_FEATURE_EVIDENCE, kind="text", label=t["label"], echo_template=t["echo_template"]) def _submit_feature_evidence(state: WizardState, value: str) -> Optional[str]: candidate = _accept_analysis_evidence(state, value, "feature_evidence_path") state.feature_evidence_pending_text = False return f"feature-evidence: {candidate.report_path}" def _build_project_evidence_pick(state: WizardState) -> Prompt: return _build_analysis_evidence_pick( state, step=S_PROJECT_EVIDENCE_PICK, relation="project-context" ) def _submit_project_evidence_pick(state: WizardState, value: str) -> Optional[str]: return _submit_analysis_evidence_pick( state, value, step=S_PROJECT_EVIDENCE_PICK, field_name="project_evidence_path", pending_name="project_evidence_pending_text", target_after_skip=state.task_type == "feature-analysis", ) def _build_project_evidence(state: WizardState) -> Prompt: t = _p(state.workspace_root, S_PROJECT_EVIDENCE) return Prompt(step=S_PROJECT_EVIDENCE, kind="text", label=t["label"], echo_template=t["echo_template"]) def _submit_project_evidence(state: WizardState, value: str) -> Optional[str]: candidate = _accept_analysis_evidence(state, value, "project_evidence_path") state.project_evidence_pending_text = False return f"project-evidence: {candidate.report_path}" def _feature_description(feature: dict[object, object]) -> str: name = str(feature.get("name") or "") summary = str(feature.get("summary") or "") entry_point = str( feature.get("entryPoint") or feature.get("entrypoint") or "" ) return " · ".join(value for value in (name, summary, entry_point) if value) def _build_analysis_target_pick(state: WizardState) -> Prompt: t = _p(state.workspace_root, S_ANALYSIS_TARGET_PICK) try: candidate = load_analysis_report_candidate( Path(state.project_root), Path(state.project_evidence_path) ) except AnalysisInputError as exc: raise WizardError(str(exc)) from exc options = [ _opt(str(feature["id"]), str(feature["id"]), _feature_description(feature)) for feature in candidate.feature_index if isinstance(feature, dict) and isinstance(feature.get("id"), str) ][:3] options.append(_opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM])) return Prompt(step=S_ANALYSIS_TARGET_PICK, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"]) def _accept_analysis_target(state: WizardState, value: str) -> str: candidates: dict[Path, AnalysisReportCandidate] = {} for path in _analysis_evidence_paths(state): try: candidate = load_analysis_report_candidate(Path(state.project_root), path) except AnalysisInputError as exc: raise WizardError(str(exc)) from exc candidates[candidate.report_path] = candidate try: target = resolve_analysis_target( value, _resolve_analysis_evidence(state), candidates ) except AnalysisInputError as exc: raise WizardError(str(exc)) from exc requested_value = target["requestedValue"] if not isinstance(requested_value, str): raise WizardError("analysis target resolver returned an invalid requestedValue") state.analysis_target = requested_value state.analysis_target_pending_text = False return requested_value def _submit_analysis_target_pick(state: WizardState, value: str) -> Optional[str]: if value == PICK_TYPE_CUSTOM: state.analysis_target_pending_text = True return None return f"analysis-target: {_accept_analysis_target(state, value)}" def _build_analysis_target(state: WizardState) -> Prompt: t = _p(state.workspace_root, S_ANALYSIS_TARGET) return Prompt(step=S_ANALYSIS_TARGET, kind="text", label=t["label"], echo_template=t["echo_template"]) def _submit_analysis_target(state: WizardState, value: str) -> Optional[str]: return f"analysis-target: {_accept_analysis_target(state, value)}" def _build_base_ref_pick(state: WizardState) -> Prompt: t = _p(state.workspace_root, "base_ref_pick") recommended_suffix = t["options"].get("_RECOMMENDED_SUFFIX", "") options = [_opt(r, f"main{recommended_suffix}" if r == "main" else r) for r in CANONICAL_BASE_REFS] for value, label in _static_options(t): options.append(_opt(value=value, label=label)) return Prompt( step=S_BASE_REF_PICK, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_base_ref_pick(state: WizardState, value: str) -> Optional[str]: if value == BASE_REF_FREE_INPUT_TOKEN: state.base_ref_pending_text = True state.base_ref = "" return None state.base_ref_pending_text = False ref = _validate_base_ref(value, Path(state.project_root)) state.base_ref = ref return f"base-ref: {ref}" def _build_base_ref_text(state: WizardState) -> Prompt: t = _p(state.workspace_root, "base_ref_text") return Prompt( step=S_BASE_REF_TEXT, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_base_ref_text(state: WizardState, value: str) -> Optional[str]: ref = _validate_base_ref(value, Path(state.project_root)) state.base_ref = ref state.base_ref_pending_text = False return f"base-ref: {ref}" PICK_USE_DEFAULT = "__use_default__" PICK_OTHER = "__other__" PICK_SKIP = "__skip__" _REUSE_LAST_TOKEN = "__reuse_last__" _SIBLINGS_TOKEN = "__siblings__" _LATEST_REPORT_TOKEN = "__latest_report__" _PROJECT_DEFAULT_TOKEN = "__project_default__" ALL_STAGES = "__all_stages__" def _list_implementation_planning_reports( state: WizardState, limit: int = 3 ) -> list[Path]: """task 의 implementation-planning runs 디렉토리에서 최신순으로 final-report 경로를 limit 개까지 반환. Each path is relative to ``project_root`` when possible. """ if not state.task_group or not state.task_id or not state.project_root: return [] # Run seq lives in the filename, not a per-run subdirectory: every # implementation-planning run writes into the same flat `reports/` # dir (see paths.py — `run_reports = runs//reports`). reports_dir = RunRef( project_root=state.project_root, task_group=state.task_group, task_id=state.task_id, task_type="implementation-planning", ).reports_dir if not reports_dir.is_dir(): return [] out: list[Path] = [] for p in list_implementation_planning_reports(reports_dir)[:limit]: try: out.append(p.relative_to(Path(state.project_root))) except ValueError: out.append(p) return out def _latest_implementation_planning_report(state: WizardState) -> Optional[Path]: """task 의 implementation-planning runs 중 가장 최신 final-report 경로 (relpath where possible).""" reports = _list_implementation_planning_reports(state, limit=1) return reports[0] if reports else None def _build_approved_plan_pick(state: WizardState) -> Prompt: reports = _list_implementation_planning_reports(state, limit=3) default = reports[0] if reports else None t = _p(state.workspace_root, "approved_plan_pick", default=str(default) if default is not None else "") is_fv = state.task_type == "final-verification" label = (t["label_final_verification"] or t["label"]) if is_fv else t["label"] other_report_label = t["labels"]["other_report"] options: list[Option] = [] if default is not None: options.append(_opt(PICK_USE_DEFAULT, t["options"][PICK_USE_DEFAULT].format(default=str(default)))) for p in reports[1:]: options.append(_opt(f"{_REPORT_PREFIX}{p}", other_report_label.format(path=str(p)))) options.append(_opt(PICK_OTHER, t["options"][PICK_OTHER])) return Prompt( step=S_APPROVED_PLAN_PICK, kind="pick", label=label, options=options, echo_template=t["echo_template"], ) def _submit_approved_plan_pick(state: WizardState, value: str) -> Optional[str]: t = _p(state.workspace_root, "approved_plan_pick", default="") if value == PICK_USE_DEFAULT: default = _latest_implementation_planning_report(state) if default is None: raise WizardError(t["errors"]["default_not_found"]) return _stage_plan_for_confirmation(state, str(default)) if value.startswith(_REPORT_PREFIX): rel = value[len(_REPORT_PREFIX):] return _stage_plan_for_confirmation( state, rel, suffix=t["echo_suffixes"]["other_report"]) if value == PICK_OTHER: state.approved_plan_pending_text = True state.approved_plan_path = "" return None raise WizardError( f"unexpected approved-plan value: {value!r} " f"(expected {PICK_USE_DEFAULT!r}, {PICK_OTHER!r}, or '{_REPORT_PREFIX}')" ) def _build_approved_plan(state: WizardState) -> Prompt: t = _p(state.workspace_root, "approved_plan") return Prompt( step=S_APPROVED_PLAN, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_approved_plan(state: WizardState, value: str) -> Optional[str]: return _stage_plan_for_confirmation(state, value) def _build_approve_plan_confirm(state: WizardState) -> Prompt: t = _p(state.workspace_root, "approve_plan_confirm", path=state.approve_plan_candidate) is_fv = state.task_type == "final-verification" label = (t["label_final_verification"] or t["label"]) if is_fv else t["label"] options_map = (t["options_final_verification"] or t["options"]) if is_fv else t["options"] if state.html_approval_sidecar: plan_label = _plan_short_label(state.approve_plan_candidate) label += t["html_approval_note"].format( plan_label=plan_label, option=state.html_approval_option or t["html_approval_note_default_option"], ) options_map = { k: v.format(plan_label=plan_label) for k, v in t["options_html_approval"].items() } return Prompt( step=S_APPROVE_PLAN_CONFIRM, kind="pick", label=label, options=[_opt(k, v) for k, v in options_map.items()], echo_template=t["echo_template"], ) def _submit_approve_plan_confirm(state: WizardState, value: str) -> Optional[str]: allowed = ("yes_apply", "yes", "no") if state.html_approval_sidecar else ("yes", "no") if value not in allowed: raise WizardError(f"expected one of {allowed}, got: {value!r}") candidate = state.approve_plan_candidate if not candidate: raise WizardError("approve-plan: no candidate plan to approve") t = _p(state.workspace_root, "approve_plan_confirm", path=candidate) if value == "no": # Declining leaves the candidate set so the confirm step re-prompts; # implementation cannot proceed without choosing to proceed. raise WizardError(t["errors"]["declined"]) apply_option = value == "yes_apply" and bool(state.html_approval_option) p = Path(candidate) if apply_option: # 승인 플립 전에 옵션 유효성부터 검증한다 — 무효 옵션으로 plan 이 # 절반만(승인만) 적용되는 상태를 만들지 않는다. _validate_sidecar_option(p, state.html_approval_option, t["errors"]) resolved, fully_approved = _classify_approved_plan( str(p), Path(state.project_root)) # flip / 옵션 적용은 PrepareError 를 던질 수 있는데, 마법사 디스패처는 # WizardError 만 재프롬프트로 처리한다. raw PrepareError 가 새면 (승인이 # 디스크에 반영된 채) state 저장 없이 traceback 으로 죽으므로 여기서 번역한다. try: if not fully_approved: # Not yet approved → flip data.json (SSOT) + re-render, then re-verify. _approve_plan_in_place(p) resolved, fully_approved = _classify_approved_plan( str(p), Path(state.project_root)) if not fully_approved: raise WizardError( t["errors"]["still_unapproved"].format(path=resolved)) variants = t["echo_variants"] approved_key = ("approved_final_verification" if state.task_type == "final-verification" and variants.get("approved_final_verification") else "approved") echo = variants[approved_key].format(path=resolved) if apply_option: _apply_cli_implementation_option( str(resolved), state.html_approval_option) echo = t["echo_variants"]["approved_with_option"].format( path=resolved, option=state.html_approval_option) except PrepareError as exc: raise WizardError(str(exc)) from exc state.approved_plan_path = str(resolved) state.approve_plan_candidate = "" state.html_approval_sidecar = "" state.html_approval_option = "" return echo def _design_prep_items_by_id(state: WizardState) -> dict[str, dict[str, Any]]: try: items = load_design_prep_items(Path(state.approved_plan_path)) except DesignPrepError as exc: raise WizardError(str(exc)) from exc return {str(item["id"]): item for item in items} def _ensure_design_prep_queue(state: WizardState) -> None: if state.design_prep_current or state.design_prep_queue: return if S_DESIGN_PREP_DECISION in state.answered: return report_path = Path(state.approved_plan_path) if not re.fullmatch( r"final-report-implementation-planning-\d+\.data\.json", report_path.name, ): return if not report_path.is_file(): return try: items = load_design_prep_items(report_path) effective = resolve_design_prep(report_path) except DesignPrepError as exc: raise WizardError(str(exc)) from exc decisions = { str(item["id"]): item.get("decision") for item in effective.effective_items } unresolved = [ item for item in items if item.get("status") in ("blocked", "provisional") and decisions.get(str(item["id"])) == "unanswered" ] unresolved.sort(key=lambda item: item.get("status") != "blocked") state.design_prep_queue = [str(item["id"]) for item in unresolved] _advance_design_prep_item(state) def _advance_design_prep_item(state: WizardState) -> None: state.design_prep_current = ( state.design_prep_queue.pop(0) if state.design_prep_queue else "" ) state.design_prep_decision = "" state.design_prep_overrides_json = "" state.design_prep_notes = "" def _current_design_prep_item(state: WizardState) -> dict[str, Any]: item = _design_prep_items_by_id(state).get(state.design_prep_current) if item is None: raise WizardError( f"design preparation item is missing: {state.design_prep_current}" ) return item def _design_prep_decision_applies(state: WizardState) -> bool: if state.task_type != "implementation" or not state.approved_plan_path: return False _ensure_design_prep_queue(state) return bool(state.design_prep_current and not state.design_prep_decision) def _build_design_prep_decision(state: WizardState) -> Prompt: item = _current_design_prep_item(state) proposal = item.get("aiProposal") or {} t = _p( state.workspace_root, S_DESIGN_PREP_DECISION, item_id=str(item["id"]), title=str(item["title"]), stage_refs=json.dumps(item.get("stageRefs") or [], ensure_ascii=False), proposal_summary=str(proposal.get("summary") or ""), details=json.dumps(proposal.get("details") or [], ensure_ascii=False), assumptions=json.dumps(proposal.get("assumptions") or [], ensure_ascii=False), guardrails=json.dumps(item.get("guardrails") or [], ensure_ascii=False), request_path=str(item.get("requestPath") or ""), ) return Prompt( step=S_DESIGN_PREP_DECISION, kind="pick", label=t["label"], options=[_opt(value, label) for value, label in _static_options(t)], echo_template=t["echo_template"], ) def _submit_design_prep_decision( state: WizardState, value: str, ) -> Optional[str]: t = _p(state.workspace_root, S_DESIGN_PREP_DECISION, item_id="", title="", stage_refs="", proposal_summary="", details="", assumptions="", guardrails="", request_path="") if value not in ("accept-draft", "modify-draft", "reject-draft", "later"): raise WizardError(t["errors"]["invalid_decision"]) if value == "later": item_id = state.design_prep_current _advance_design_prep_item(state) return t["echo_variants"]["later"].format(item_id=item_id) state.design_prep_decision = value state.design_prep_overrides_json = "" state.design_prep_notes = "" return t["echo_template"].format(value=value) def _design_prep_overrides_applies(state: WizardState) -> bool: if state.design_prep_decision == "modify-draft": return not state.design_prep_overrides_json if state.design_prep_decision == "reject-draft": return not state.design_prep_notes return False def _build_design_prep_overrides(state: WizardState) -> Prompt: t = _p( state.workspace_root, S_DESIGN_PREP_OVERRIDES, item_id=state.design_prep_current, decision=state.design_prep_decision, ) return Prompt( step=S_DESIGN_PREP_OVERRIDES, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_design_prep_overrides( state: WizardState, value: str, ) -> Optional[str]: t = _p(state.workspace_root, S_DESIGN_PREP_OVERRIDES, item_id=state.design_prep_current, decision=state.design_prep_decision) if state.design_prep_decision == "reject-draft": if not value.strip(): raise WizardError(t["errors"]["note_required"]) state.design_prep_notes = value.strip() return t["echo_variants"]["note"] try: overrides = json.loads(value) except json.JSONDecodeError as exc: raise WizardError(t["errors"]["invalid_json"].format(error=exc)) from exc if not isinstance(overrides, dict): raise WizardError(t["errors"]["object_required"]) state.design_prep_overrides_json = json.dumps( overrides, ensure_ascii=False, sort_keys=True ) return t["echo_variants"]["overrides"] def _design_prep_confirm_applies(state: WizardState) -> bool: if state.design_prep_decision == "accept-draft": return True if state.design_prep_decision == "modify-draft": return bool(state.design_prep_overrides_json) if state.design_prep_decision == "reject-draft": return bool(state.design_prep_notes) return False def _build_design_prep_confirm(state: WizardState) -> Prompt: item = _current_design_prep_item(state) t = _p( state.workspace_root, S_DESIGN_PREP_CONFIRM, item_id=state.design_prep_current, decision=state.design_prep_decision, item_snapshot=json.dumps(item, ensure_ascii=False, indent=2, sort_keys=True), overrides=state.design_prep_overrides_json or "{}", notes=state.design_prep_notes or "(none)", ) return Prompt( step=S_DESIGN_PREP_CONFIRM, kind="pick", label=t["label"], options=[_opt(value, label) for value, label in _static_options(t)], echo_template=t["echo_template"], ) def _submit_design_prep_confirm( state: WizardState, value: str, ) -> Optional[str]: item_id = state.design_prep_current t = _p(state.workspace_root, S_DESIGN_PREP_CONFIRM, item_id=item_id, decision=state.design_prep_decision, item_snapshot="", overrides="", notes="") if value == "no": state.design_prep_decision = "" state.design_prep_overrides_json = "" state.design_prep_notes = "" return t["echo_variants"]["revise"].format(item_id=item_id) if value != "yes": raise WizardError(t["errors"]["confirmation_required"]) overrides = json.loads(state.design_prep_overrides_json or "{}") try: path = write_design_prep_input( Path(state.approved_plan_path), item_id, state.design_prep_decision, overrides, state.design_prep_notes, captured_by="okstra-wizard", ) except DesignPrepError as exc: raise WizardError(str(exc)) from exc match = re.search(r"-r(\d+)-([0-9a-f-]{36})\.md$", path.name) if match is None: raise WizardError(f"design preparation writer returned invalid path: {path}") echo = t["echo_variants"]["written"].format( item_id=item_id, revision=int(match.group(1)), input_id=match.group(2), path=path, ) _advance_design_prep_item(state) return echo def _build_stage_pick(state: WizardState) -> Prompt: """Parse the Stage Map from the approved plan and build the stage picker.""" t = _p(state.workspace_root, "stage_pick") stages = _parse_stage_objects(state) is_fv = state.task_type == "final-verification" is_impl = state.task_type == "implementation" label = (t["label_final_verification"] or t["label"]) if is_fv else t["label"] snapshot = _stage_lifecycle_snapshot( state, stages, reserved_stages=_reserved_stage_numbers(state) if is_impl else None, ) done = snapshot.done_stages options = [] if _whole_task_allowed(state, stages=stages, done=done): options.append(_opt(WHOLE_TASK_STAGE, t["options"]["whole_task"])) if is_impl: options.append(_opt(ALL_STAGES, t["options"]["all_stages"])) for s in stages: depends = ",".join(map(str, s.depends_on)) or "(none)" suffix = "" if is_fv: suffix = " " + (t["options"]["done_mark"] if s.stage_number in done else t["options"]["undone_mark"]) elif is_impl: suffix = " " + _impl_stage_marker( t, snapshot.lifecycle_for(s.stage_number)) options.append(_opt( str(s.stage_number), f"{s.stage_number}: {s.title} " f"[depends-on: {depends} | steps: {s.step_count}]{suffix}", )) return Prompt( step=S_STAGE_PICK, kind="pick", multi=is_impl, label=label, options=options, echo_template=t["echo_template"], ) # Presentation keys are mapped explicitly rather than interpolated from the # status, so renaming a Stage Lifecycle status is a compile-time-visible edit # here instead of a KeyError raised while drawing the picker. _STAGE_MARKER_KEYS = { "done": "mark_done", "active": "mark_active", "ready": "mark_ready", "blocked": "mark_blocked", } def _impl_stage_marker(t, lifecycle) -> str: return t["options"][_STAGE_MARKER_KEYS[lifecycle.status]] def _submit_stage_pick(state: WizardState, answer: str) -> Optional[str]: if state.task_type == "implementation": return _submit_impl_stage_pick(state, answer) # final-verification: 단일선택 유지 (whole-task 또는 단일 정수; auto 불가) if not answer: raise WizardError("value required") if answer == WHOLE_TASK_STAGE: if not _whole_task_allowed(state): raise WizardError( "whole-task verification requires final-verification " "with all stages done") else: try: int(answer) except ValueError: raise WizardError( f"answer must be whole-task or a stage number, got {answer!r}") state.selected_stage = answer return f"stage: {answer}" def _submit_impl_stage_pick(state: WizardState, answer: str) -> Optional[str]: from .stage_targets import order_stage_closure t = _p(state.workspace_root, "stage_pick") picks = [v.strip() for v in (answer or "").split(",") if v.strip()] if not picks: raise WizardError(t["errors"]["none_selected"]) if WHOLE_TASK_STAGE in picks: raise WizardError(t["errors"]["whole_task_impl"]) stages = _parse_stage_objects(state) all_nums = {s.stage_number for s in stages} snapshot = _stage_lifecycle_snapshot( state, stages, reserved_stages=_reserved_stage_numbers(state)) done = snapshot.done_stages occupied = {lc.stage for lc in snapshot.lifecycles if lc.status in ("done", "active")} chosen = _impl_chosen_stages(t, picks, answer, all_nums, occupied) ordered = order_stage_closure( [(s.stage_number, s.depends_on) for s in stages], chosen, done) state.selected_stages = ",".join(map(str, ordered)) state.selected_stage = str(ordered[0]) if ordered else "auto" added = [n for n in ordered if n not in chosen] if added: return t["echo_variants"]["with_closure"].format( stages=state.selected_stages, added=", ".join(map(str, added))) return t["echo_variants"]["plain"].format(stages=state.selected_stages) def _impl_chosen_stages(t, picks, answer, all_nums, occupied) -> set: if ALL_STAGES in picks: if len(picks) > 1: raise WizardError(t["errors"]["all_exclusive"]) chosen = {n for n in all_nums if n not in occupied} if not chosen: raise WizardError(t["errors"]["nothing_selectable"]) return chosen try: nums = {int(p) for p in picks} except ValueError: raise WizardError(t["errors"]["bad_number"].format(answer=answer)) unknown = sorted(nums - all_nums) if unknown: raise WizardError(t["errors"]["unknown_stage"].format( bad=", ".join(map(str, unknown)))) bad = sorted(nums & occupied) if bad: raise WizardError(t["errors"]["occupied"].format( bad=", ".join(map(str, bad)))) return nums def _handoff_msgs(state: WizardState) -> dict: """handoff_stage_pick 의 JSON 텍스트 묶음 (label 미사용 조회용).""" return _p(state.workspace_root, "handoff_stage_pick", blocked="") def _resolve_handoff_plan(state: WizardState) -> Path: """release-handoff 의 approved plan 을 질문 없이 자동 해소한다. plan 미존재/미승인은 사용자가 고칠 대상이 아니라 라이프사이클 선행 단계 누락이므로 picker 대신 안내 메시지로 즉시 실패한다.""" if state.approved_plan_path: return Path(state.approved_plan_path) t = _handoff_msgs(state) latest = _latest_implementation_planning_report(state) if latest is None: raise WizardError(t["errors"]["no_plan"]) p, fully_approved = _classify_approved_plan( str(latest), Path(state.project_root)) if not fully_approved: raise WizardError(t["errors"]["plan_not_approved"].format(plan=p)) state.approved_plan_path = str(p) return p def _handoff_eligibility(state: WizardState) -> list: """stage 별 PR 자격 — okstra_ctl.handoff 의 SSOT 판정을 그대로 재사용한다.""" from okstra_ctl.handoff import compute_eligibility from okstra_ctl.consumers import read_consumers plan = _resolve_handoff_plan(state) try: stage_map = stage_map_records(parse_stage_map_file(plan)) except StageMapError as exc: raise WizardError(str(exc)) from exc rows = read_consumers(plan.resolve().parents[1]) return compute_eligibility(stage_map, rows) def _latest_whole_task_fv_release_ready(state: WizardState) -> str: """accepted whole-task final-verification 보고서 경로 — handoff 모듈 SSOT 위임.""" from okstra_ctl.handoff import latest_whole_task_fv_release_ready return latest_whole_task_fv_release_ready( state.project_root, state.project_id, state.task_group, state.task_id) def _build_handoff_stage_pick(state: WizardState) -> Prompt: elig = _handoff_eligibility(state) eligible = [e for e in elig if e["eligible"]] blocked = [e for e in elig if not e["eligible"]] whole_task_report = _latest_whole_task_fv_release_ready(state) msgs = _handoff_msgs(state) blocked_summary = ("; ".join( f"stage {e['stage']} ({', '.join(e['reasons'])})" for e in blocked) or msgs["labels"]["blocked_none"]) if not eligible and not whole_task_report: raise WizardError( msgs["errors"]["nothing_eligible"].format(blocked=blocked_summary)) t = _p(state.workspace_root, "handoff_stage_pick", blocked=blocked_summary) options: list[Option] = [] if whole_task_report: options.append(_opt(WHOLE_TASK_STAGE, t["labels"]["whole_task"])) stage_label = t["labels"]["stage"] for e in eligible: deps = ", ".join(str(d) for d in e["depends_on"]) or "-" options.append(_opt(str(e["stage"]), stage_label.format(stage=e["stage"], deps=deps))) return Prompt( step=S_HANDOFF_STAGE_PICK, kind="pick", multi=True, label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_handoff_stage_pick(state: WizardState, value: str) -> Optional[str]: t = _handoff_msgs(state) picks = [v.strip() for v in (value or "").split(",") if v.strip()] if not picks: raise WizardError(t["errors"]["none_selected"]) if WHOLE_TASK_STAGE in picks: if len(picks) > 1: raise WizardError(t["errors"]["whole_task_exclusive"]) if not _latest_whole_task_fv_release_ready(state): raise WizardError(t["errors"]["whole_task_missing"]) state.handoff_mode = "whole-task" state.handoff_stages = "" return t["echo_variants"]["whole_task"] eligible = {str(e["stage"]) for e in _handoff_eligibility(state) if e["eligible"]} bad = [p for p in picks if p not in eligible] if bad: raise WizardError(t["errors"]["not_eligible"].format( bad=", ".join(bad), eligible=", ".join(sorted(eligible)))) nums = sorted({int(p) for p in picks}) state.handoff_mode = "stage-group" state.handoff_stages = ",".join(str(n) for n in nums) return t["echo_variants"]["stage_group"].format( stages=state.handoff_stages) def _same_file(a: Path, b_str: str) -> bool: """Whether two paths resolve to the same file. False when either path cannot be resolved (missing intermediate dir, permission, etc.).""" try: return a.resolve() == Path(b_str).resolve() except OSError: return False def _suggest_latest_final_report(state: WizardState) -> str: """clarification carry-in 으로 추천할 직전 final-report 의 relpath. resume-clarification 은 "같은 phase 를 답변과 함께 재실행" 이므로 재실행 task-type **자신의** 보고서(``runs//reports/final-report-*.md``)를 우선한다. 그 phase 에 보고서가 없을 때만(예: 다음 phase 로 진행) 전체 phase 의 mtime 최신으로 폴백한다 — 과거엔 무조건 전체 mtime 최신이라 더 최근에 재렌더된 다른 phase 보고서를 잘못 집을 수 있었다. 못 찾으면 빈 문자열. """ if not state.task_group or not state.task_id or not state.project_root: return "" runs_base = task_runs_dir(state.project_root, state.task_group, state.task_id) if not runs_base.is_dir(): return "" try: task_root = _resolved_within( runs_base.parent, Path(state.project_root).resolve(), "task root", ) except AnalysisInputError: return "" # A revision request outranks ordinary recency. Carry the newest requested # revision for the selected analysis type into that exact rerun. revision = _latest_revision_requested_analysis_report(state) best = revision if best is None and state.task_type: seg = slugify_task_segment(state.task_type) best = _newest_contained_final_report( runs_base, task_root, f"{seg}/reports/final-report-*.data.json", Path(state.project_root), ) if best is None: best = _newest_contained_final_report( runs_base, task_root, "*/reports/final-report-*.data.json", Path(state.project_root), ) if best is None: return "" # The approved plan is already wired via --approved-plan. On the first run # of an approved-plan consuming phase (final-verification / implementation) # the newest cross-phase report IS that plan, so the fallback would # re-recommend it as a clarification answer — injecting it twice, read as a # user clarification it is not. Never carry the approved plan back in here. if state.approved_plan_path and _same_file(best, state.approved_plan_path): return "" try: return str(best.relative_to(Path(state.project_root))) except ValueError: return str(best) def _clarification_sidecar_note(state: WizardState, suggestion: str) -> str: """추천 final-report 와 함께 carry-in 될 ``user-responses/`` 사이드카 현황 문구. "final-report 만 물어보니 user-responses 파일이 쓰이는지 불확실하다" 는 혼선을 없애기 위해, picker 단계에서 실제 첨부될 답변 파일 개수를 그 자리에서 보여준다. """ p = Path(suggestion) if not p.is_absolute() and state.project_root: p = Path(state.project_root) / p count = len(user_response_sidecars(p)) note = _p(state.workspace_root, "clarification_pick")["sidecar_note"] return note["attached"].format(count=count) if count else note["empty"] def _latest_run_inputs( state: WizardState, *, phase_segment: str = "" ) -> dict: """직전 run 의 입력 스냅샷. 위치/구조 지식은 run_context.latest_run_inputs 가 SSOT.""" return latest_run_inputs( state.project_root, state.task_group, state.task_id, phase_segment=phase_segment) def _suggest_last_directive(state: WizardState) -> str: """같은 task 의 가장 최근 run-inputs-*.json 에서 directive 값을 자동 추출.""" val = _latest_run_inputs(state).get("directive") or "" return val if isinstance(val, str) else "" # --------------------------------------------------------------------------- # "optional cached pick" seam. # # directive / related-tasks / clarification / pr-template 단계는 모두 동일한 # 3-옵션 picker 구조를 갖는다: [건너뛰기 / 추천값(이전 directive·siblings·최근 # 리포트·프로젝트 기본) / 직접 입력]. 추천값은 디스크에서 suggest 하고, 선택 시 # state 의 cache 필드에 담아 submit 에서 꺼낸다. 과거에는 이 구조가 네 곳에 # 기계적으로 복제돼 한 곳을 고치면 나머지가 drift 했다. 변이점만 담은 선언적 # spec + 제네릭 build/submit 으로 단일 seam 화한다. # --------------------------------------------------------------------------- @dataclass(frozen=True) class _OptionalCachedPickSpec: step: str prompt_key: str # _p() 의 step_id recommend_token: str # 추천 옵션의 sentinel value label_key: str # t["labels"] 의 추천 라벨 키 echo_suffix_key: str # t["echo_suffixes"] 의 추천 echo 키 suggest: Callable[[WizardState], str] snippet_style: str # "prefix" (앞 60자+…) | "suffix" (…+뒤 60자) cache_attr: str # 추천값을 담아둘 state 속성 target_attr: str # 확정값을 쓸 state 속성 pending_attr: str # 직접 입력 대기 플래그 state 속성 extra_clear_attrs: tuple[str, ...] = () # skip 시 추가로 비울 속성 # 추천 옵션 라벨·echo 뒤에 덧붙일 동적 안내(예: 함께 첨부될 사이드카 개수). # (state, 추천 relpath) -> note. 미설정 picker 는 영향 없음. recommend_note: Optional[Callable[["WizardState", str], str]] = None def _pick_snippet(value: str, style: str) -> str: if len(value) <= 60: return value return value[:60] + "…" if style == "prefix" else "…" + value[-60:] def _build_optional_cached_pick(state: WizardState, spec: _OptionalCachedPickSpec) -> Prompt: suggestion = spec.suggest(state) t = _p(state.workspace_root, spec.prompt_key) # 추천(이전 directive·siblings·최근 리포트·프로젝트 기본)을 가장 먼저 노출하고, # '건너뛰기'는 중간, '직접 입력'은 항상 마지막에 둔다 (run-prompt 추천 규칙). options: list[Option] = [] if suggestion: snippet = _pick_snippet(suggestion, spec.snippet_style) label = t["labels"][spec.label_key].format(snippet=snippet) if spec.recommend_note: label += spec.recommend_note(state, suggestion) options.append(_opt(spec.recommend_token, label)) setattr(state, spec.cache_attr, suggestion) options.append(_opt(PICK_SKIP, t["options"][PICK_SKIP])) options.append(_opt(PICK_TYPE_CUSTOM, t["options"][PICK_TYPE_CUSTOM])) return Prompt( step=spec.step, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _submit_optional_cached_pick( state: WizardState, value: str, spec: _OptionalCachedPickSpec ) -> Optional[str]: t = _p(state.workspace_root, spec.prompt_key) if value == PICK_SKIP: setattr(state, spec.target_attr, "") for attr in spec.extra_clear_attrs: setattr(state, attr, "") setattr(state, spec.pending_attr, False) return t["echo_suffixes"]["skip"] if value == spec.recommend_token: cached = getattr(state, spec.cache_attr) setattr(state, spec.target_attr, cached) setattr(state, spec.pending_attr, False) echo = t["echo_suffixes"][spec.echo_suffix_key].format(value=cached) if spec.recommend_note: echo += spec.recommend_note(state, cached) return echo if value == PICK_TYPE_CUSTOM: setattr(state, spec.pending_attr, True) return None raise WizardError( f"unexpected {spec.prompt_key} value: {value!r} " f"(expected {PICK_SKIP!r}, {spec.recommend_token!r}, or {PICK_TYPE_CUSTOM!r})" ) _DIRECTIVE_PICK_SPEC = _OptionalCachedPickSpec( step=S_DIRECTIVE_PICK, prompt_key="directive_pick", recommend_token=_REUSE_LAST_TOKEN, label_key="reuse_last", echo_suffix_key="reuse", suggest=_suggest_last_directive, snippet_style="prefix", cache_attr="last_directive_cached", target_attr="directive", pending_attr="directive_pending_text", ) def _build_directive_pick(state: WizardState) -> Prompt: return _build_optional_cached_pick(state, _DIRECTIVE_PICK_SPEC) def _submit_directive_pick(state: WizardState, value: str) -> Optional[str]: return _submit_optional_cached_pick(state, value, _DIRECTIVE_PICK_SPEC) def _suggest_sibling_task_ids(state: WizardState) -> str: """같은 task-group 의 다른 task-id 를 CSV 로 반환 (현재 task 제외, 빈 결과면 '').""" if not state.project_root or not state.task_group: return "" try: tasks = list_project_tasks(Path(state.project_root)) except (OSError, StateError): return "" siblings: list[str] = [] for entry in tasks: tg = entry.get("taskGroup") or "" if tg != state.task_group: continue tid = entry.get("taskId") or "" if not tid or tid == state.task_id: continue if tid not in siblings: siblings.append(tid) return ",".join(siblings) _RELATED_TASKS_PICK_SPEC = _OptionalCachedPickSpec( step=S_RELATED_TASKS_PICK, prompt_key="related_tasks_pick", recommend_token=_SIBLINGS_TOKEN, label_key="siblings", echo_suffix_key="siblings", suggest=_suggest_sibling_task_ids, snippet_style="prefix", cache_attr="last_siblings_cached", target_attr="related_tasks_raw", pending_attr="related_tasks_pending_text", ) def _build_related_tasks_pick(state: WizardState) -> Prompt: return _build_optional_cached_pick(state, _RELATED_TASKS_PICK_SPEC) def _submit_related_tasks_pick(state: WizardState, value: str) -> Optional[str]: return _submit_optional_cached_pick(state, value, _RELATED_TASKS_PICK_SPEC) _CLARIFICATION_PICK_SPEC = _OptionalCachedPickSpec( step=S_CLARIFICATION_PICK, prompt_key="clarification_pick", recommend_token=_LATEST_REPORT_TOKEN, label_key="latest_report", echo_suffix_key="latest_report", suggest=_suggest_latest_final_report, snippet_style="suffix", cache_attr="last_final_report_cached", target_attr="clarification_response_path", pending_attr="clarification_pending_text", recommend_note=_clarification_sidecar_note, ) def _build_clarification_pick(state: WizardState) -> Prompt: return _build_optional_cached_pick(state, _CLARIFICATION_PICK_SPEC) def _submit_clarification_pick(state: WizardState, value: str) -> Optional[str]: return _submit_optional_cached_pick(state, value, _CLARIFICATION_PICK_SPEC) def _carried_planning_report(state: WizardState) -> Optional[Path]: """이번 재실행이 이어받는 직전 implementation-planning 리포트 (없으면 None).""" if state.task_type != "implementation-planning": return None if not state.clarification_response_path or not state.project_root: return None report = _resolve_path( state.clarification_response_path, Path(state.project_root) ) return report if report.is_file() else None def _reverify_scope_preview(state: WizardState) -> Optional[dict]: """답변된 id 가 직전 리포트의 stage 로 되짚어지는지 — 범위 판정의 앞 절반.""" report = _carried_planning_report(state) if report is None: return None return preview_link_availability_for_report( report, set(sidecar_answers(report)) ) def _reverify_scope_pick_required(state: WizardState) -> bool: """좁힐 여지가 있거나, unlinked id 의 stage 번호를 받아야 할 때 묻는다. unlinked 는 full 확정이 아니다. SHA 가 바뀌었거나 입력을 읽을 수 없어 `wouldForceFull` 인 경우만 질문이 의미 없다. """ preview = _reverify_scope_preview(state) return preview is not None and ( not preview["wouldForceFull"] or bool(preview["unlinkedIds"]) ) def _reverify_scope_step_pending(state: WizardState) -> bool: """범위 질문이 아직 안 끝났는가 — confirm 진입을 막는 게이트.""" if state.reverify_scope_pending_text: return True return _reverify_scope_pick_required(state) and not state.reverify_scope def _prior_stage_numbers(state: WizardState) -> set[int]: """직전 리포트 Stage Map 의 stage 번호. 읽을 수 없으면 WizardError.""" t = _p(state.workspace_root, "reverify_scope_stages") report = _carried_planning_report(state) if report is None: raise WizardError( t["errors"]["no_stage_map"].format(reason="carried report not found") ) data_path = final_report_data_path(report) try: data = load_owned_object(data_path, artifact="planning final report") stages = {num for num, _ in parse_stage_graph(data)} except (OSError, ValueError, KeyError, TypeError) as exc: raise WizardError( t["errors"]["no_stage_map"].format(reason=str(exc)) ) from exc if not stages: raise WizardError( t["errors"]["no_stage_map"].format(reason="stage map is empty") ) return stages def _build_reverify_scope_pick(state: WizardState) -> Prompt: t = _p(state.workspace_root, "reverify_scope_pick") opts = t["options"] preview = _reverify_scope_preview(state) or {} unlinked = bool(preview.get("unlinkedIds")) options = [] if not unlinked: options.append(_opt("auto", opts["auto"])) options.append(_opt("full", opts["full"])) options.append(_opt(PICK_TYPE_CUSTOM, opts[PICK_TYPE_CUSTOM])) else: options.append(_opt(PICK_TYPE_CUSTOM, opts[PICK_TYPE_CUSTOM])) options.append(_opt("full", opts["full"])) label = t["label_unlinked"] if unlinked else t["label"] return Prompt( step=S_REVERIFY_SCOPE_PICK, kind="pick", label=label, options=options, echo_template=t["echo_template"]) def _submit_reverify_scope_pick(state: WizardState, value: str) -> Optional[str]: t = _p(state.workspace_root, "reverify_scope_pick") picked = value.strip().lower() if picked == PICK_TYPE_CUSTOM: state.reverify_scope = "" state.reverify_scope_pending_text = True return "reverify-scope: 직접 입력" if picked not in ("auto", "full"): raise WizardError( f"expected 'auto' / 'full' / {PICK_TYPE_CUSTOM!r}, got: {value!r}" ) if picked == "auto": preview = _reverify_scope_preview(state) or {} unlinked = preview.get("unlinkedIds") or [] if unlinked: raise WizardError( t["errors"]["unlinked_auto"].format(ids=", ".join(unlinked)) ) state.reverify_scope = picked state.reverify_scope_pending_text = False return t["echo_suffixes"][picked] def _build_reverify_scope_stages(state: WizardState) -> Prompt: t = _p(state.workspace_root, "reverify_scope_stages") return Prompt( step=S_REVERIFY_SCOPE_STAGES, kind="text", label=t["label"], echo_template=t["echo_template"]) def _submit_reverify_scope_stages(state: WizardState, value: str) -> Optional[str]: t = _p(state.workspace_root, "reverify_scope_stages") tokens = [token.strip() for token in value.split(",") if token.strip()] if not tokens: preview = _reverify_scope_preview(state) or {} unlinked = preview.get("unlinkedIds") or [] if unlinked: raise WizardError( t["errors"]["unlinked_empty"].format(ids=", ".join(unlinked)) ) state.reverify_scope = "auto" state.reverify_scope_pending_text = False return t["echo_suffixes"]["auto"] for token in tokens: if not token.isdigit(): raise WizardError(t["errors"]["not_a_number"].format(token=token)) known = _prior_stage_numbers(state) picked = sorted({int(token) for token in tokens}) unknown = [num for num in picked if num not in known] if unknown: raise WizardError(t["errors"]["unknown_stage"].format( stages=", ".join(str(num) for num in unknown), known=", ".join(str(num) for num in sorted(known)), )) state.reverify_scope = ",".join(str(num) for num in picked) state.reverify_scope_pending_text = False return t["echo_template"].format(value=state.reverify_scope) def _suggest_project_pr_template(state: WizardState) -> str: """project.json 의 prTemplatePath 필드를 읽어 경로 문자열로 반환. 없거나 읽기 실패 시 빈 문자열. """ if not state.project_root: return "" project_json = project_json_path(Path(state.project_root)) if not project_json.is_file(): return "" try: data = load_owned_object(project_json, artifact="project config") except (OSError, JsonBoundaryError): return "" val = data.get("prTemplatePath") or "" return val if isinstance(val, str) else "" _PR_TEMPLATE_PICK_SPEC = _OptionalCachedPickSpec( step=S_PR_TEMPLATE_PICK, prompt_key="pr_template_pick", recommend_token=_PROJECT_DEFAULT_TOKEN, label_key="project_default", echo_suffix_key="project_default", suggest=_suggest_project_pr_template, snippet_style="suffix", cache_attr="last_pr_template_cached", target_attr="pr_template_path", pending_attr="pr_template_pending_text", # skip 시 scope 도 함께 비운다 (원래 _submit_pr_template_pick 의 동작). extra_clear_attrs=("pr_template_scope",), ) def _build_pr_template_pick(state: WizardState) -> Prompt: return _build_optional_cached_pick(state, _PR_TEMPLATE_PICK_SPEC) def _submit_pr_template_pick(state: WizardState, value: str) -> Optional[str]: return _submit_optional_cached_pick(state, value, _PR_TEMPLATE_PICK_SPEC) 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}" # resume 재사용을 제안하는 phase. okstra.sh --resume-clarification 과 동일하게 # §1 clarification 을 갖는 비-implementation phase 로 한정한다(implementation 은 # approved-plan/stage 가 매 run 고유라 재사용 대상이 아니다). _RESUME_REUSE_PHASES = ( "requirements-discovery", "error-analysis", "implementation-planning", *ANALYSIS_TASK_TYPES, ) # YES 선택 시 prefill 로 충족되어 다시 묻지 않는 설정 단계들. _REUSE_FILLED_STEPS = ( S_FEATURE_EVIDENCE_PICK, S_FEATURE_EVIDENCE, S_PROJECT_EVIDENCE_PICK, S_PROJECT_EVIDENCE, S_ANALYSIS_TARGET_PICK, S_ANALYSIS_TARGET, S_DEFAULTS_OR_CUSTOM, S_WORKERS_OVERRIDE, S_LEAD_MODEL, S_EXECUTOR_MODEL, S_CLAUDE_MODEL, S_CODEX_MODEL, S_ANTIGRAVITY_MODEL, S_GROK_MODEL, S_KIMI_MODEL, S_REPORT_WRITER_MODEL, S_DIRECTIVE_PICK, S_DIRECTIVE, S_RELATED_TASKS_PICK, S_RELATED_TASKS, S_CLARIFICATION_PICK, S_CLARIFICATION) def _has_prior_run_inputs(state: WizardState) -> bool: seg = slugify_task_segment(state.task_type) if state.task_type else "" return bool(_latest_run_inputs(state, phase_segment=seg)) def _analysis_inputs_after_reuse(state: WizardState) -> bool: return ( state.reuse_previous is not None or state.task_type not in _RESUME_REUSE_PHASES or not _has_prior_run_inputs(state) ) def _restore_analysis_evidence_inputs( state: WizardState, evidence_raw: object ) -> tuple[tuple[Any, ...], dict[Path, AnalysisReportCandidate]]: if state.task_type not in ANALYSIS_TASK_TYPES or not isinstance(evidence_raw, str): return (), {} report_paths = [ _resolve_path(item.strip(), Path(state.project_root)) for item in evidence_raw.split(",") if item.strip() ] try: candidates = [ load_analysis_report_candidate(Path(state.project_root), path) for path in report_paths ] evidence_inputs = resolve_evidence_inputs( Path(state.project_root), state.task_type, [candidate.report_path for candidate in candidates], _analysis_current_commit(state), ) except AnalysisInputError as exc: raise WizardError(str(exc)) from exc state.feature_evidence_path = "" state.project_evidence_path = "" for evidence in evidence_inputs: if evidence.relation == "feature-baseline": state.feature_evidence_path = str(evidence.report_path) elif evidence.relation == "project-context": state.project_evidence_path = str(evidence.report_path) return evidence_inputs, { candidate.report_path: candidate for candidate in candidates } 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: requested_provider = _safe_provider(inputs.get("leadProvider"), "lead") 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( 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}" def _build_directive(state: WizardState) -> Prompt: t = _p(state.workspace_root, "directive") return Prompt( step=S_DIRECTIVE, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_directive(state: WizardState, value: str) -> Optional[str]: state.directive = (value or "").strip() state.directive_pending_text = False return f"directive: {state.directive or '(none)'}" def _build_related_tasks(state: WizardState) -> Prompt: t = _p(state.workspace_root, "related_tasks") return Prompt( step=S_RELATED_TASKS, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_related_tasks(state: WizardState, value: str) -> Optional[str]: state.related_tasks_raw = (value or "").strip() state.related_tasks_pending_text = False return f"related-tasks: {state.related_tasks_raw or '(none)'}" def _build_clarification(state: WizardState) -> Prompt: t = _p(state.workspace_root, "clarification") return Prompt( step=S_CLARIFICATION, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_clarification(state: WizardState, value: str) -> Optional[str]: val = (value or "").strip() state.clarification_pending_text = False if not val: state.clarification_response_path = "" return "clarification: (none)" p = _require_file(val, Path(state.project_root), "clarification-response") state.clarification_response_path = str(p) return f"clarification: {p}" def _build_pr_template(state: WizardState) -> Prompt: t = _p(state.workspace_root, "pr_template") return Prompt( step=S_PR_TEMPLATE, kind="text", label=t["label"], echo_template=t["echo_template"], ) def _submit_pr_template(state: WizardState, value: str) -> Optional[str]: val = (value or "").strip() state.pr_template_pending_text = False if not val: state.pr_template_path = "" state.pr_template_scope = "" return "pr-template: (auto-resolve)" # Validate by re-using resolve_pr_template_path with override. try: resolved = resolve_pr_template_path( Path(state.project_root), override_path=val ) except PrTemplateError as exc: raise WizardError(str(exc)) state.pr_template_path = str(resolved.path) return f"pr-template: {resolved.path}" def _build_pr_template_scope(state: WizardState) -> Prompt: t = _p(state.workspace_root, "pr_template_scope") return Prompt( step=S_PR_TEMPLATE_SCOPE, kind="pick", label=t["label"], options=[_opt(k, v) for k, v in t["options"].items()], echo_template=t["echo_template"], ) def _submit_pr_template_scope(state: WizardState, value: str) -> Optional[str]: if value not in ("once", "project", "global"): raise WizardError( f"expected 'once' / 'project' / 'global', got: {value!r}" ) state.pr_template_scope = value return f"pr-template-scope: {value}" def _build_fix_cycle_confirm(state: WizardState) -> Prompt: t = _p(state.workspace_root, "fix_cycle_confirm") opts = t["options"] return Prompt( step=S_FIX_CYCLE_CONFIRM, kind="pick", label=t["label"], options=[ _opt("yes", opts["yes"]), _opt("no", opts["no"]), _opt("abort", opts["abort"]), ], echo_template=t["echo_template"]) def _submit_fix_cycle_confirm(state: WizardState, value: str) -> Optional[str]: v = value.strip().lower() if v == "abort": state.aborted = True return "fix-cycle: abort" if v not in ("yes", "no"): raise WizardError(f"expected 'yes' / 'no' / 'abort', got: {value!r}") state.fix_cycle = v return f"fix-cycle: {v}" def _preview_work_category(state: WizardState) -> str: """okstra-run 경로는 --work-category 를 넘기지 않는다. prepare 가 쓰는 것과 같은 resolver 를 태워야 미리보기 브랜치명이 실제 생성 브랜치와 일치한다.""" return resolve_work_category( "", project_root=Path(state.project_root), task_group=state.task_group, task_id=state.task_id, ) def _worktree_preview_line(state: WizardState) -> Optional[str]: """confirm 직전 요약 블록에 넣을 worktree 미리보기 한 줄. final-verification 은 stage worktree 를 읽기 전용 재사용하므로 None.""" if not _worktree_preview_required(state): return None if state.task_type == "implementation": return _worktree_preview_line_impl(state) decision = preview_worktree_decision( project_root=Path(state.project_root), project_id=state.project_id, task_group_segment=state.task_group, task_id_segment=state.task_id, work_category=_preview_work_category(state), base_ref=state.base_ref, ) key = { "new": "worktree_new", "reused": "worktree_reuse", "skipped-in-worktree": "worktree_in_worktree", "skipped-not-git": "worktree_not_git", }[decision.status] return _msg(state.workspace_root, "confirmation", key, branch=decision.branch or "(none)", base_ref=decision.base_ref or "(HEAD)", path=str(decision.path)) def _worktree_preview_line_impl(state: WizardState) -> str: """implementation 은 stage 격리로 동작하므로 task-key 디렉터리가 아니라 이번 run 이 실제로 사용할 stage worktree 관점으로 미리보기를 보여준다.""" stage = (state.selected_stage or "auto").strip() if stage == "auto": parent_dir = compute_worktree_path( project_id=state.project_id, task_group_segment=state.task_group, task_id_segment=state.task_id, ) return _msg(state.workspace_root, "confirmation", "worktree_impl_auto", path=str(parent_dir)) decision = resolve_stage_worktree_decision( project_id=state.project_id, task_group_segment=state.task_group, task_id_segment=state.task_id, work_category=_preview_work_category(state), stage_number=int(stage), ) key = ("worktree_impl_reuse" if decision.status == "reused" else "worktree_impl_new") return _msg(state.workspace_root, "confirmation", key, stage=stage, path=str(decision.path), branch=decision.branch) def _build_confirm(state: WizardState) -> Prompt: t = _p(state.workspace_root, "confirm") return Prompt( step=S_CONFIRM, kind="pick", label=t["label"], options=[_opt(k, v) for k, v in t["options"].items()], echo_template=t["echo_template"], ) def _submit_confirm(state: WizardState, value: str) -> Optional[str]: if value == "abort": state.aborted = True return "confirm: abort" if value not in ("proceed", "edit"): raise WizardError( f"expected 'proceed' / 'edit' / 'abort', got: {value!r}" ) state.confirmed = value == "proceed" return f"confirm: {value}" def _build_edit_target(state: WizardState) -> Prompt: t = _p(state.workspace_root, "edit_target") # offer every step that has been answered. options: list[Option] = [] selection_ids = [ *state.answered, *( step_id for step_id in state.role_selection_order if step_id not in state.answered ), ] for sid in selection_ids: if sid in (S_CONFIRM, S_EDIT_TARGET): continue options.append(_opt(sid, sid)) return Prompt( step=S_EDIT_TARGET, kind="pick", label=t["label"], options=options, echo_template=t["echo_template"], ) def _reset_role_selection_from(state: WizardState, target_step: str) -> None: try: target_index = state.role_selection_order.index(target_step) except ValueError as exc: raise WizardError(f"unknown role selection step: {target_step!r}") from exc kept_ids = state.role_selection_order[:target_index] kept_counts = { step_id.split(":", 1)[1] for step_id in kept_ids if step_id.startswith(("role-count:", "role-add:")) } kept_models: dict[str, int] = {} for step_id in kept_ids: if not step_id.startswith("role-model:"): continue _, role, _ = step_id.split(":", 2) kept_models[role] = kept_models.get(role, 0) + 1 state.role_counts = { role: count for role, count in state.role_counts.items() if role in kept_counts } state.role_models = { role: models[:kept_models[role]] for role, models in state.role_models.items() if kept_models.get(role, 0) > 0 } removed_ids = set(state.role_selection_order[target_index:]) state.role_selection_order = kept_ids state.answered = [ step_id for step_id in state.answered if step_id not in removed_ids ] def _clear_role_selection(state: WizardState) -> None: state.role_counts.clear() state.role_models.clear() state.role_selection_order.clear() state.answered = [ step_id for step_id in state.answered if not _is_role_selection_step(step_id) ] def _submit_edit_target(state: WizardState, value: str) -> Optional[str]: if _is_role_selection_step(value): _reset_role_selection_from(state, value) elif any(s.id == value for s in STEPS): _reset_from(state, value) else: raise WizardError(f"unknown step: {value!r}") state.edit_target = value state.confirmed = None state.edit_target = "" return f"edit-target: {value} (rewinding)" # --- step registry --- STEPS: list[Step] = [ Step(S_TASK_PICK, applies=lambda s: s.is_new_task is None, build=_build_task_pick, submit=_submit_task_pick, owns=("is_new_task", "task_group", "task_id", "task_type", "existing_brief_path", "profile_workers", "profile_optional_workers", "task_group_suggestion", "task_id_suggestion", "task_group_pending_text", "task_id_pending_text")), Step(S_TASK_GROUP, applies=lambda s: (bool(s.is_new_task) and not s.task_group and not s.task_group_pending_text), build=_build_task_group, submit=_submit_task_group, owns=("task_group", "task_group_pending_text")), Step(S_TASK_GROUP_TEXT, applies=lambda s: (bool(s.is_new_task) and not s.task_group and s.task_group_pending_text), build=_build_task_group_text, submit=_submit_task_group_text, owns=("task_group", "task_group_pending_text")), # 신규 task 흐름 순서: task-group → task-type → (entry 면 brief) → task-id. # brief 는 entry phase 전용 입력이므로 task-type 이 정해진 뒤에만 물을 수 있다. Step(S_TASK_TYPE, applies=lambda s: (s.is_new_task is not None and (s.is_new_task is False or bool(s.task_group)) and S_TASK_TYPE not in s.answered), build=_build_task_type, submit=_submit_task_type, owns=("task_type", "profile_workers", "profile_optional_workers", "reuse_worktree")), Step(S_TASK_TYPE_TEXT, applies=lambda s: (S_TASK_TYPE in s.answered and not s.task_type and S_TASK_TYPE_TEXT not in s.answered), build=_build_task_type_text, submit=_submit_task_type_text, owns=("task_type", "profile_workers", "profile_optional_workers", "reuse_worktree")), Step(S_BRIEF_KEEP, applies=lambda s: (not s.is_new_task and s.task_type in _BRIEF_ENTRY_TASK_TYPES and bool(s.existing_brief_path) and s.keep_existing_brief is None and S_TASK_TYPE in s.answered), build=_build_brief_keep, submit=_submit_brief_keep, owns=("keep_existing_brief",)), Step(S_BRIEF_PATH_PICK, applies=lambda s: ( not s.brief_path and not s.brief_path_pending_text and S_BRIEF_PATH_PICK not in s.answered and S_TASK_TYPE in s.answered and s.task_type in _BRIEF_ENTRY_TASK_TYPES and ( s.is_new_task is True or (s.is_new_task is False and (s.keep_existing_brief is False or not s.existing_brief_path)) ) ), build=_build_brief_path_pick, submit=_submit_brief_path_pick, owns=("brief_path_pending_text",)), Step(S_BRIEF_PATH, applies=lambda s: s.brief_path_pending_text and S_BRIEF_PATH not in s.answered, build=_build_brief_path, submit=_submit_brief_path, owns=("brief_path", "task_group_suggestion", "task_id_suggestion")), # downstream task-type 인데 carry-in 할 brief 가 없을 때의 fallback. # release-handoff 는 brief 자체가 없는 phase 라 대상에서 빠진다. Step(S_BRIEF_CARRY, applies=lambda s: (S_TASK_TYPE in s.answered and bool(s.task_type) and s.task_type not in _BRIEF_ENTRY_TASK_TYPES and s.task_type != "release-handoff" and not s.brief_path and not s.brief_path_pending_text), build=_build_brief_carry, submit=_submit_brief_carry, owns=("brief_path_pending_text",)), Step(S_TASK_ID, applies=lambda s: (bool(s.is_new_task) and bool(s.task_group) and _brief_resolved(s) and not s.task_id and not s.task_id_pending_text), build=_build_task_id, submit=_submit_task_id, owns=("task_id", "task_id_pending_text")), Step(S_TASK_ID_TEXT, applies=lambda s: (bool(s.is_new_task) and bool(s.task_group) and _brief_resolved(s) and not s.task_id and s.task_id_pending_text), build=_build_task_id_text, submit=_submit_task_id_text, owns=("task_id", "task_id_pending_text")), Step(S_REUSE_PREVIOUS, applies=lambda s: (_identity_ready(s) and s.use_defaults is None and s.reuse_previous is None and s.task_type in _RESUME_REUSE_PHASES and _has_prior_run_inputs(s)), build=_build_reuse_previous, submit=_submit_reuse_previous, owns=("reuse_previous",)), Step(S_FEATURE_EVIDENCE_PICK, applies=lambda s: (s.task_type == "change-impact-analysis" and not s.feature_evidence_path and not s.feature_evidence_pending_text and S_FEATURE_EVIDENCE_PICK not in s.answered and _brief_resolved(s) and _analysis_inputs_after_reuse(s)), build=_build_feature_evidence_pick, submit=_submit_feature_evidence_pick, owns=("feature_evidence_path", "feature_evidence_pending_text")), Step(S_FEATURE_EVIDENCE, applies=lambda s: (s.task_type == "change-impact-analysis" and s.feature_evidence_pending_text and S_FEATURE_EVIDENCE not in s.answered), build=_build_feature_evidence, submit=_submit_feature_evidence, owns=("feature_evidence_path", "feature_evidence_pending_text")), Step(S_PROJECT_EVIDENCE_PICK, applies=lambda s: (s.task_type in ("feature-analysis", "change-impact-analysis") and not s.project_evidence_path and not s.project_evidence_pending_text and S_PROJECT_EVIDENCE_PICK not in s.answered and _brief_resolved(s) and _analysis_inputs_after_reuse(s) and (s.task_type != "change-impact-analysis" or S_FEATURE_EVIDENCE_PICK in s.answered)), build=_build_project_evidence_pick, submit=_submit_project_evidence_pick, owns=("project_evidence_path", "project_evidence_pending_text", "analysis_target", "analysis_target_pending_text")), Step(S_PROJECT_EVIDENCE, applies=lambda s: (s.task_type in ("feature-analysis", "change-impact-analysis") and s.project_evidence_pending_text and S_PROJECT_EVIDENCE not in s.answered), build=_build_project_evidence, submit=_submit_project_evidence, owns=("project_evidence_path", "project_evidence_pending_text", "analysis_target", "analysis_target_pending_text")), Step(S_ANALYSIS_TARGET_PICK, applies=lambda s: (s.task_type == "feature-analysis" and bool(s.project_evidence_path) and not s.analysis_target and not s.analysis_target_pending_text and S_ANALYSIS_TARGET_PICK not in s.answered), build=_build_analysis_target_pick, submit=_submit_analysis_target_pick, owns=("analysis_target", "analysis_target_pending_text")), Step(S_ANALYSIS_TARGET, applies=lambda s: (s.task_type == "feature-analysis" and not s.analysis_target and s.analysis_target_pending_text and S_ANALYSIS_TARGET not in s.answered), build=_build_analysis_target, submit=_submit_analysis_target, owns=("analysis_target", "analysis_target_pending_text")), Step(S_BASE_REF_PICK, applies=lambda s: (S_TASK_TYPE in s.answered and _base_ref_required(s) and S_BASE_REF_PICK not in s.answered and _brief_resolved(s) and (not s.is_new_task or bool(s.task_id))), build=_build_base_ref_pick, submit=_submit_base_ref_pick, owns=("base_ref", "base_ref_pending_text")), Step(S_BASE_REF_TEXT, applies=lambda s: s.base_ref_pending_text, build=_build_base_ref_text, submit=_submit_base_ref_text, owns=("base_ref", "base_ref_pending_text")), Step(S_APPROVED_PLAN_PICK, applies=lambda s: (s.task_type in _STAGE_SCOPED_TASK_TYPES and not s.approved_plan_path and not s.approved_plan_pending_text and S_APPROVED_PLAN_PICK not in s.answered and bool(s.brief_path) and _base_ref_ready(s) and not s.base_ref_pending_text and _latest_implementation_planning_report(s) is not None), build=_build_approved_plan_pick, submit=_submit_approved_plan_pick, owns=("approved_plan_path", "approved_plan_pending_text")), Step(S_APPROVED_PLAN, applies=lambda s: (s.task_type in _STAGE_SCOPED_TASK_TYPES and not s.approved_plan_path and bool(s.brief_path) and _base_ref_ready(s) and not s.base_ref_pending_text and (s.approved_plan_pending_text or _latest_implementation_planning_report(s) is None)), build=_build_approved_plan, submit=_submit_approved_plan, owns=("approved_plan_path", "approved_plan_pending_text")), Step(S_APPROVE_PLAN_CONFIRM, applies=lambda s: (s.task_type in _STAGE_SCOPED_TASK_TYPES and bool(s.approve_plan_candidate) and not s.approved_plan_path and S_APPROVE_PLAN_CONFIRM not in s.answered), build=_build_approve_plan_confirm, submit=_submit_approve_plan_confirm, owns=("approve_plan_candidate",)), Step(S_DESIGN_PREP_DECISION, applies=_design_prep_decision_applies, build=_build_design_prep_decision, submit=_submit_design_prep_decision, owns=("design_prep_queue", "design_prep_current", "design_prep_decision", "design_prep_overrides_json", "design_prep_notes"), repeatable=True), Step(S_DESIGN_PREP_OVERRIDES, applies=_design_prep_overrides_applies, build=_build_design_prep_overrides, submit=_submit_design_prep_overrides, owns=("design_prep_overrides_json", "design_prep_notes"), repeatable=True), Step(S_DESIGN_PREP_CONFIRM, applies=_design_prep_confirm_applies, build=_build_design_prep_confirm, submit=_submit_design_prep_confirm, owns=("design_prep_decision", "design_prep_overrides_json", "design_prep_notes"), repeatable=True), Step(S_STAGE_PICK, applies=lambda s: (s.task_type in _STAGE_SCOPED_TASK_TYPES and bool(s.approved_plan_path) and S_STAGE_PICK not in s.answered), build=_build_stage_pick, submit=_submit_stage_pick, owns=("selected_stage", "selected_stages")), Step(S_HANDOFF_STAGE_PICK, applies=lambda s: (s.task_type == "release-handoff" and bool(s.task_group) and bool(s.task_id) and S_HANDOFF_STAGE_PICK not in s.answered), build=_build_handoff_stage_pick, submit=_submit_handoff_stage_pick, owns=("handoff_mode", "handoff_stages", "approved_plan_path")), Step(S_EXECUTOR, applies=lambda s: (s.task_type == "implementation" and not _role_selection_enabled(s) and bool(s.approved_plan_path) and not s.executor), build=_build_executor, submit=_submit_executor, owns=("executor",)), Step(S_CRITIC_PICK, applies=lambda s: (not _role_selection_enabled(s) and s.task_type in ("requirements-discovery", "error-analysis", "implementation-planning", "final-verification") and not s.critic and not s.critic_pending_text and S_CRITIC_PICK not in s.answered), build=_build_critic_pick, submit=_submit_critic_pick, owns=("critic", "critic_pending_text")), Step(S_CRITIC_TEXT, applies=lambda s: (not _role_selection_enabled(s) and s.critic_pending_text and S_CRITIC_TEXT not in s.answered), build=_build_critic_text, submit=_submit_critic_text, owns=("critic", "critic_pending_text")), Step(S_DEFAULTS_OR_CUSTOM, applies=lambda s: (_identity_ready(s) and not _role_selection_enabled(s) and s.use_defaults is None), build=_build_defaults_or_custom, submit=_submit_defaults_or_custom, owns=("use_defaults",)), # Worker roster is ALWAYS prompted (independent of the defaults / # customize branch) when the profile has analyser-candidate workers # (required ∪ optional, minus report-writer). Users repeatedly hit # the failure mode where the lead silently picked "Use defaults" # and the worker prompt never appeared — defaults should govern # model choice, not worker selection. `implementation` task-type is # still skipped because it runs lead + executor only. Step(S_WORKERS_OVERRIDE, applies=lambda s: (s.task_type != "implementation" and not _role_selection_enabled(s) and any( w != "report-writer" for w in (s.profile_workers + s.profile_optional_workers) ) and S_WORKERS_OVERRIDE not in s.answered), build=_build_workers_override, submit=_submit_workers_override, owns=("workers_override", "workers_custom_pending")), # "직접 선택" 을 고른 run 에서만 등장한다 — 기본 로스터에서 워커를 빼는 # 축소가 가능한 유일한 화면. Step(S_WORKERS_CUSTOM, applies=lambda s: (not _role_selection_enabled(s) and s.workers_custom_pending and S_WORKERS_CUSTOM not in s.answered), build=_build_workers_custom, submit=_submit_workers_custom, owns=("workers_override", "workers_custom_pending")), Step(S_LEAD_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and S_LEAD_MODEL not in s.answered), build=_build_lead_model, submit=_submit_lead_model, owns=("lead_provider", "lead_model")), Step(S_EXECUTOR_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and s.task_type == "implementation" and S_EXECUTOR_MODEL not in s.answered), build=_build_executor_model, submit=_submit_executor_model, owns=("claude_model", "codex_model", "antigravity_model")), Step(S_CLAUDE_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and s.task_type != "implementation" and "claude" in _resolved_roster(s) and S_CLAUDE_MODEL not in s.answered), build=_build_claude_model, submit=_submit_claude_model, owns=("claude_model",)), Step(S_CODEX_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and s.task_type != "implementation" and "codex" in _resolved_roster(s) and S_CODEX_MODEL not in s.answered), build=_build_codex_model, submit=_submit_codex_model, owns=("codex_model",)), Step(S_ANTIGRAVITY_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and s.task_type != "implementation" and "antigravity" in _resolved_roster(s) and S_ANTIGRAVITY_MODEL not in s.answered), build=_build_antigravity_model, submit=_submit_antigravity_model, owns=("antigravity_model",)), Step(S_GROK_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and s.task_type != "implementation" and "grok" in _resolved_roster(s) and S_GROK_MODEL not in s.answered), build=_build_grok_model, submit=_submit_grok_model, owns=("grok_model",)), Step(S_KIMI_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and s.task_type != "implementation" and "kimi" in _resolved_roster(s) and S_KIMI_MODEL not in s.answered), build=_build_kimi_model, submit=_submit_kimi_model, owns=("kimi_model",)), Step(S_REPORT_WRITER_MODEL, applies=lambda s: (s.use_defaults is False and not _role_selection_enabled(s) and (s.task_type == "implementation" or "report-writer" in _resolved_roster(s)) and S_REPORT_WRITER_MODEL not in s.answered), build=_build_report_writer_model, submit=_submit_report_writer_model, owns=("report_writer_provider", "report_writer_model")), # directive(이번 run 의 추가 지시)는 기본값/커스터마이즈와 무관하게 항상 # 묻는다 — 매 run 마다 줄 수 있는 입력이므로 'Use defaults' 분기 뒤에 숨기지 # 않는다. (use_defaults is not None: defaults_or_custom 답 이후에만 등장) Step(S_DIRECTIVE_PICK, applies=lambda s: (S_DIRECTIVE_PICK not in s.answered and s.use_defaults is not None), build=_build_directive_pick, submit=_submit_directive_pick, owns=("directive", "directive_pending_text")), Step(S_DIRECTIVE, applies=lambda s: (s.directive_pending_text and S_DIRECTIVE not in s.answered), build=_build_directive, submit=_submit_directive, owns=("directive", "directive_pending_text")), Step(S_RELATED_TASKS_PICK, applies=lambda s: (s.use_defaults is False and S_RELATED_TASKS_PICK not in s.answered), build=_build_related_tasks_pick, submit=_submit_related_tasks_pick, owns=("related_tasks_raw", "related_tasks_pending_text")), Step(S_RELATED_TASKS, applies=lambda s: (s.use_defaults is False and s.related_tasks_pending_text and S_RELATED_TASKS not in s.answered), build=_build_related_tasks, submit=_submit_related_tasks, owns=("related_tasks_raw", "related_tasks_pending_text")), # clarification(직전 phase final-report 입력)은 customize 전용이 아니다. # 이어가기에 필수인 직전 final-report 가 존재하면 use_defaults 와 무관하게 # 입력 기회를 노출한다. (과거: use_defaults 게이트 뒤에 숨어 "Use defaults" # 를 고르면 직전 리포트가 통째로 누락됐다.) Step(S_CLARIFICATION_PICK, applies=lambda s: (S_CLARIFICATION_PICK not in s.answered and s.use_defaults is not None and (s.use_defaults is False or bool(_suggest_latest_final_report(s)))), build=_build_clarification_pick, submit=_submit_clarification_pick, owns=("clarification_response_path", "clarification_pending_text")), Step(S_CLARIFICATION, applies=lambda s: (s.clarification_pending_text and S_CLARIFICATION not in s.answered), build=_build_clarification, submit=_submit_clarification, owns=("clarification_response_path", "clarification_pending_text")), # clarification 뒤에 온다. `_planning_rerun_selected` 는 clarification 답을 읽어 # "이 런은 기존 계획서의 재실행" 인지 판정하는데, 이 단계가 clarification 앞에 # 있던 동안 그 값은 언제나 비어 있었다 — 가드가 참이 될 수 없는 자리에 놓여 # 있었다. 그 결과 재실행이어야 할 런이 매번 방향 선택으로 들어갔고, # implementation-option-selection 리포트가 없는 레거시 후보비교 task 는 # 거기서 끝났다. Step(S_SELECTED_DIRECTION_PICK, applies=lambda s: ( s.task_type == "implementation-planning" and not s.selected_direction_path and not _planning_rerun_selected(s) and _brief_resolved(s) and _base_ref_ready(s) and not s.base_ref_pending_text and S_SELECTED_DIRECTION_PICK not in s.answered # 고를 것이 없으면 묻지 않는다. 물으면 build 가 WizardError 를 내고 # `current: null` 로 끝나 재프롬프트조차 불가능해진다 — 사용자가 # 되돌릴 수 없는 막다른 길이다. 방향 없이 진행한 런은 # `run._validate_planning_entry_inputs` 가 두 입력을 모두 이름 붙여 # 거절하므로, 실패는 복구 가능한 자리로 옮겨간다. and bool(_selected_direction_candidates(s)) ), build=_build_selected_direction_pick, submit=_submit_selected_direction_pick, owns=("selected_direction_path",)), Step(S_PR_TEMPLATE_PICK, applies=lambda s: (s.use_defaults is False and s.task_type == "release-handoff" and S_PR_TEMPLATE_PICK not in s.answered), build=_build_pr_template_pick, submit=_submit_pr_template_pick, owns=("pr_template_path", "pr_template_scope", "pr_template_pending_text")), Step(S_PR_TEMPLATE, applies=lambda s: (s.use_defaults is False and s.task_type == "release-handoff" and s.pr_template_pending_text and S_PR_TEMPLATE not in s.answered), build=_build_pr_template, submit=_submit_pr_template, owns=("pr_template_path", "pr_template_scope", "pr_template_pending_text")), Step(S_PR_TEMPLATE_SCOPE, applies=lambda s: (s.use_defaults is False and s.task_type == "release-handoff" and bool(s.pr_template_path) and S_PR_TEMPLATE_SCOPE not in s.answered), build=_build_pr_template_scope, submit=_submit_pr_template_scope, owns=("pr_template_scope",)), # 재검증 범위는 clarification-response 가 정해진 뒤에야 판정할 수 있고, # 확인 블록은 full 재검증 비용을 물기 전 마지막 되돌림 지점이다. 그래서 # 이 질문은 confirm 바로 앞에 선다. Step(S_REVERIFY_SCOPE_PICK, applies=lambda s: (_ready_for_confirm(s) and not s.reverify_scope_pending_text and not s.reverify_scope and _reverify_scope_pick_required(s)), build=_build_reverify_scope_pick, submit=_submit_reverify_scope_pick, owns=("reverify_scope", "reverify_scope_pending_text")), Step(S_REVERIFY_SCOPE_STAGES, applies=lambda s: (s.reverify_scope_pending_text and S_REVERIFY_SCOPE_STAGES not in s.answered), build=_build_reverify_scope_stages, submit=_submit_reverify_scope_stages, owns=("reverify_scope", "reverify_scope_pending_text")), Step(S_FIX_CYCLE_CONFIRM, applies=lambda s: (_ready_for_confirm(s) and not _reverify_scope_step_pending(s) and _fix_cycle_confirm_required(s) and not s.fix_cycle), build=_build_fix_cycle_confirm, submit=_submit_fix_cycle_confirm, owns=("fix_cycle",)), Step(S_CONFIRM, applies=lambda s: (_ready_for_confirm(s) and not _reverify_scope_step_pending(s) and (not _fix_cycle_confirm_required(s) or bool(s.fix_cycle)) and s.confirmed is None), build=_build_confirm, submit=_submit_confirm, owns=("confirmed", "edit_target")), Step(S_EDIT_TARGET, applies=lambda s: s.confirmed is False and not s.edit_target, build=_build_edit_target, submit=_submit_edit_target, owns=()), ] STEP_BY_ID = {s.id: s for s in STEPS} 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 def _ready_for_confirm(s: WizardState) -> bool: if _role_selection_enabled(s): try: if next_role_prompt(s) is not None: return False except WizardError: return False followup_ids = [ S_DIRECTIVE_PICK, S_DIRECTIVE, S_RELATED_TASKS_PICK, S_RELATED_TASKS, S_CLARIFICATION_PICK, S_CLARIFICATION, S_PR_TEMPLATE_PICK, S_PR_TEMPLATE, S_PR_TEMPLATE_SCOPE, ] return not any(STEP_BY_ID[step_id].applies(s) for step_id in followup_ids) if s.use_defaults is None: return False # Worker roster is required regardless of the defaults / customize # branch. Skipping this check let the lead drop into confirm with no # worker prompt ever shown — the very failure mode this gate exists # to prevent. workers_step = STEP_BY_ID[S_WORKERS_OVERRIDE] if workers_step.applies(s): return False if STEP_BY_ID[S_WORKERS_CUSTOM].applies(s): return False if s.use_defaults: return True # customize: every customize-branch step must be answered or not-applicable. custom_ids = [S_LEAD_MODEL, S_EXECUTOR_MODEL, S_CLAUDE_MODEL, S_CODEX_MODEL, S_ANTIGRAVITY_MODEL, S_REPORT_WRITER_MODEL, S_DIRECTIVE, S_RELATED_TASKS, S_CLARIFICATION, S_PR_TEMPLATE, S_PR_TEMPLATE_SCOPE] for sid in custom_ids: step = STEP_BY_ID[sid] if step.applies(s): return False return True def _reset_from(state: WizardState, target_step: str) -> None: """Clear state owned by target_step and all later steps; remove their entries from `answered`. Used when the user picks Edit.""" idx = next((i for i, s in enumerate(STEPS) if s.id == target_step), -1) if idx < 0: return role_boundary = next( i for i, step in enumerate(STEPS) if step.id == S_DEFAULTS_OR_CUSTOM ) if idx < role_boundary: _clear_role_selection(state) # A later step may own a field an earlier answered step also owns — # `handoff_stage_pick` owns `approved_plan_path` because it resolves the # plan on its own. Clearing it while rewinding to a step in front of it # drops an answer the user never revisited; the rewound step then fails its # own `applies` guard, so no step is left to ask and the wizard reports # done while `outcome` still refuses it as incomplete. owned_earlier = { fname for step in STEPS[:idx] if step.id in state.answered for fname in step.owns } cleared_ids: set[str] = set() for position, step in enumerate(STEPS[idx:]): cleared_ids.add(step.id) is_rewind_target = position == 0 for fname in step.owns: # The target's own fields always clear — that is the answer the # user came back to replace, even when an earlier step declares it # too (`task_pick` derives `task_type` for an existing task). if is_rewind_target or fname not in owned_earlier: _reset_field(state, fname) state.answered = [a for a in state.answered if a not in cleared_ids] direct_input_pending = { S_FEATURE_EVIDENCE: "feature_evidence_pending_text", S_PROJECT_EVIDENCE: "project_evidence_pending_text", S_ANALYSIS_TARGET: "analysis_target_pending_text", S_WORKERS_CUSTOM: "workers_custom_pending", } pending_field = direct_input_pending.get(target_step) if pending_field is not None: setattr(state, pending_field, 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": "", } 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) # ---- Public API --------------------------------------------------------- def init_state( *, workspace_root: str, project_root: str, project_id: str, host_runtime: str = "claude-code", host_entry_mode: str = "current-session", available_functions: Optional[list[str]] = None, ) -> WizardState: """Bootstrap a new wizard state.""" return WizardState( execution_identity_version=2, workspace_root=workspace_root, project_root=project_root, project_id=project_id, host_runtime=host_runtime, host_entry_mode=host_entry_mode, available_functions=list(available_functions or []), ) def _build_group_prompt(state: WizardState, group_id: str) -> Prompt: """그룹의 적용가능·미답변 픽 멤버를 최대 GROUP_MAX_TABS 개 모은다. 멤버가 1개뿐이면 멀티탭 UI가 불필요하므로 그 멤버의 평범한 픽을 반환한다. 호출부(next_prompt)는 적용 가능한 멤버가 최소 1개일 때만 진입하므로 빈 그룹은 도달 불가다. """ members: list[Prompt] = [] for sid in PROMPT_GROUPS[group_id]: step = STEP_BY_ID[sid] if sid in state.answered and not step.repeatable: continue if not step.applies(state): continue members.append(step.build(state)) if len(members) >= GROUP_MAX_TABS: break assert members, f"group {group_id!r} reached with no applicable members" if len(members) == 1: return members[0] return Prompt(step=group_id, kind="pick_group", label=GROUP_LABELS[group_id], questions=members) def next_prompt(state: WizardState) -> Prompt: if state.aborted: return Prompt(step=S_ABORTED, kind="aborted") if state.confirmed: return Prompt(step=S_DONE, kind="done") for step in STEPS: if step.id == S_DEFAULTS_OR_CUSTOM: role_prompt = next_role_prompt(state) if role_prompt is not None: return role_prompt if step.id in state.answered and not step.repeatable: continue if step.applies(state): group_id = _STEP_TO_GROUP.get(step.id) if group_id is not None: return _build_group_prompt(state, group_id) return step.build(state) return Prompt(step=S_DONE, kind="done") def _passed_screens(state: WizardState) -> int: """이미 답한 화면 수. pick_group 멤버는 GROUP_MAX_TABS 묶음당 1화면으로 환산.""" count = 0 group_hits: dict[str, int] = {} for sid in state.answered: gid = _STEP_TO_GROUP.get(sid) if gid is None: count += 1 else: group_hits[gid] = group_hits.get(gid, 0) + 1 for hits in group_hits.values(): count += math.ceil(hits / GROUP_MAX_TABS) return count def _sim_answer(prompt: Prompt) -> str: """분모 추정 시뮬레이션의 기본답: pick 은 첫 옵션(추천), text 는 빈 값.""" if prompt.kind == "pick" and prompt.options: return prompt.options[0].value return "" def _sim_advance(state: WizardState, prompt: Prompt) -> None: """기본답으로 한 화면 전진한다. progress 를 재계산하는 submit()/ _submit_group() 은 호출하지 않고 step.submit 만 직접 호출해 재귀를 막는다.""" try: if _is_role_selection_step(prompt.step): _submit_role_prompt(state, prompt, _sim_answer(prompt)) if prompt.step not in state.answered: state.answered.append(prompt.step) if prompt.step not in state.role_selection_order: state.role_selection_order.append(prompt.step) return if prompt.kind == "pick_group": for q in prompt.questions: STEP_BY_ID[q.step].submit(state, _sim_answer(q)) for q in prompt.questions: if q.step not in state.answered: state.answered.append(q.step) return if prompt.step == S_DESIGN_PREP_CONFIRM: _advance_design_prep_item(state) if prompt.step not in state.answered: state.answered.append(prompt.step) return if prompt.step == S_DESIGN_PREP_OVERRIDES: if state.design_prep_decision == "modify-draft": state.design_prep_overrides_json = "{}" else: state.design_prep_notes = "simulation" if prompt.step not in state.answered: state.answered.append(prompt.step) return STEP_BY_ID[prompt.step].submit(state, _sim_answer(prompt)) if prompt.step not in state.answered: state.answered.append(prompt.step) except (WizardError, PrepareError): # 기본답이 거부되는 두 경우만 전진 처리한다 — 입력 검증이 막는 드문 text # 분기(WizardError), 그리고 답 자체는 유효하나 대상의 도메인 상태가 그 # 경로를 막는 경우(PrepareError). 후자가 빠져 있어 승인 게이트가 # `blocked-by-disagreement` 인 계획이 최신 태스크이면 위저드가 첫 화면조차 # 내지 못하고 죽었다: 시뮬레이터가 기본 경로를 따라가다 그 계획을 고르고, # 진행률 라벨 하나 때문에 run 전체가 시작 불가가 됐다. 시뮬레이션은 분모 # 추정이므로 막힌 경로를 만나면 그 화면을 지났다고 치고 계속 세면 된다. # KeyError/AttributeError 등 실제 버그는 여전히 삼키지 않고 그대로 # 전파시켜, progress 라벨이 그럴듯하게 틀리는 대신 시끄럽게 실패한다. members = prompt.questions if prompt.kind == "pick_group" else [prompt] for p in members: if p.step not in state.answered: state.answered.append(p.step) def _remaining_screens(state: WizardState) -> int: """현재 화면부터 confirm(=done 직전)까지 남은 화면 수를 시뮬레이션으로 센다.""" sim = copy.deepcopy(state) screens = 0 for _ in range(len(STEPS) * 2 + 5): # Edit 루프 등에 대한 상한 가드 try: prompt = next_prompt(sim) except Exception: break if prompt.kind in ("done", "aborted"): break screens += 1 _sim_advance(sim, prompt) if prompt.step == S_CONFIRM: # confirm 이후는 done — Edit 는 가정하지 않는다 break return screens def _screen_progress(state: WizardState) -> dict[str, Any]: passed = _passed_screens(state) total = passed + _remaining_screens(state) index = passed + 1 remaining = max(0, total - index) suffix = "마지막 단계" if remaining == 0 else f"앞으로 {remaining} 스텝 남음" return { "index": index, "total": total, "remaining": remaining, "label": f"Step {index}/{total} · {suffix}", } def _domain_prompt(prompt: Prompt) -> WizardPrompt: return WizardPrompt( step=prompt.step, kind=prompt.kind, label=prompt.label, options=tuple( WizardOption(option.value, option.label, option.description) for option in prompt.options ), help=prompt.help, echo_template=prompt.echo_template, multi=prompt.multi, questions=tuple(_domain_prompt(question) for question in prompt.questions), ) def _interaction_plan(state: WizardState, prompt: Prompt) -> InteractionPlan: context = HostSessionContext( host_id=state.host_runtime, entry_mode=state.host_entry_mode, available_functions=frozenset(state.available_functions), interaction_surface="terminal", ) adapter = default_host_registry().resolve(state.host_runtime) return plan_prompt(_domain_prompt(prompt), context, adapter.interaction()).interaction def _interaction_payload(plan: InteractionPlan) -> dict[str, Any]: return { "kind": plan.kind, "answerProtocol": { "kind": plan.answer_protocol.kind, "multi": plan.answer_protocol.multi, }, } def _normalize_interaction_answer( state: WizardState, prompt: Prompt, plan: InteractionPlan, value: str, ) -> str: try: return normalize_planned_answer( _domain_prompt(prompt), plan.answer_protocol, value, ) except WizardAnswerError as exc: is_legacy_raw_answer = ( not state.available_functions and "not an option" in str(exc) and not (value or "").strip().isdecimal() ) if is_legacy_raw_answer: return value raise WizardError(str(exc)) from exc def prompt_payload(state: WizardState, prompt: Prompt) -> dict[str, Any]: """Prompt JSON 에 진행 카운터를 덧붙인다. done/aborted 는 progress 를 생략.""" out = prompt.to_json() plan = _interaction_plan(state, prompt) out["interaction"] = _interaction_payload(plan) if plan.kind in {"numbered-single", "numbered-multi"}: out["presentation"] = "numbered-text" if prompt.kind not in ("done", "aborted"): out["progress"] = _screen_progress(state) return out def _submit_group(state: WizardState, prompt: Prompt, value: str) -> dict[str, Any]: """pick_group 답(JSON 객체)을 각 멤버 submit() 으로 라우팅한다. 멤버 submit 이 WizardError 를 던지면 그대로 전파되어 같은 그룹을 재-프롬프트한다. answered 마킹은 모든 멤버 submit 이 통과한 뒤에만 일괄 수행한다(answered 단위의 전부-아니면-전무). 개별 멤버가 변경한 state 필드는 롤백하지 않지만, 재-프롬프트 시 같은 그룹이 다시 나와 사용자 입력으로 덮어쓰므로 무해하다. """ try: answers = json.loads(value or "{}") except json.JSONDecodeError as exc: raise WizardError(f"pick_group answer must be a JSON object: {exc}") if not isinstance(answers, dict): raise WizardError("pick_group answer must be a JSON object") echoes: list[str] = [] for q in prompt.questions: echo = STEP_BY_ID[q.step].submit(state, str(answers.get(q.step, "") or "")) if echo: echoes.append(echo) for q in prompt.questions: if q.step not in state.answered: state.answered.append(q.step) nxt = next_prompt(state) return {"echo": "; ".join(echoes), "next": prompt_payload(state, nxt)} def submit(state: WizardState, value: str) -> dict[str, Any]: """Validate the answer for the *currently active* step and advance. Returns {"echo": "...", "next": }. Raises WizardError on validation failure (caller may re-prompt). """ prompt = next_prompt(state) if prompt.kind in ("done", "aborted"): return {"echo": "", "next": prompt_payload(state, prompt)} plan = _interaction_plan(state, prompt) value = _normalize_interaction_answer(state, prompt, plan, value) if prompt.kind == "pick_group": return _submit_group(state, prompt, value) if _is_role_selection_step(prompt.step): echo = _submit_role_prompt(state, prompt, value or "") if prompt.step not in state.answered: state.answered.append(prompt.step) if prompt.step not in state.role_selection_order: state.role_selection_order.append(prompt.step) nxt = next_prompt(state) return {"echo": echo, "next": prompt_payload(state, nxt)} step = STEP_BY_ID[prompt.step] echo = step.submit(state, value or "") if prompt.step not in state.answered: state.answered.append(prompt.step) nxt = next_prompt(state) return {"echo": echo or "", "next": prompt_payload(state, nxt)} def _stage_intent(state: WizardState) -> WizardStageIntent: """This state's stage selection, resolved once for every consumer. `render_args` needs the single stage this run prepares; `wizard_outcome` needs the chain the skill drives. Deriving them separately would let the two disagree about which stage the run is for. """ try: return resolve_wizard_stage_intent( task_type=state.task_type, selected_stage=state.selected_stage, selected_stages=state.selected_stages, ) except WizardStageIntentError as exc: raise WizardError(str(exc)) from exc def render_role_args(state: WizardState) -> list[str]: """Return ordered canonical role count/model CLI tokens.""" role_argv: list[str] = [] if _role_selection_enabled(state): profile = _load_role_profile_for_state(state) requirements = _selectable_static_requirements(state, profile) for requirement in profile.roles: if ( requirement.dynamic or requirement.min_count == requirement.max_count ): continue if requirement.role not in state.role_counts: continue count = state.role_counts[requirement.role] if count <= 0: continue role_argv.extend([ "--role-count", f"{requirement.role}={count}", ]) ordered_roles = [requirement.role for requirement in requirements] else: ordered_roles = [] for role in state.role_models: if role not in ordered_roles: ordered_roles.append(role) for role in ordered_roles: for model_ref in state.role_models.get(role, []): role_argv.extend(["--role-model", f"{role}={model_ref}"]) return role_argv def render_args(state: WizardState) -> dict[str, Any]: """Convert finalized state into ``okstra render-bundle`` argument map.""" if state.aborted: raise WizardError( "wizard was aborted by the user — render-args is unavailable" ) base_ref = ( "" if state.reuse_worktree or state.task_type == "final-verification" else state.base_ref ) stage_intent = _stage_intent(state) pr_template = ( state.pr_template_path if state.task_type == "release-handoff" else "" ) evidence_paths = [ path for path in (state.feature_evidence_path, state.project_evidence_path) if path ] rendered: dict[str, Any] = { "project-root": state.project_root, "project-id": state.project_id, "task-group": state.task_group, "task-id": state.task_id, "task-type": state.task_type, "task-brief": state.brief_path, "analysis-target": state.analysis_target, "evidence-inputs": ",".join(evidence_paths), "approved-plan": state.approved_plan_path, "stage": stage_intent.stage, "stages": state.handoff_stages, "base-ref": base_ref, "directive": state.directive, "related-tasks": state.related_tasks_raw, "clarification-response": state.clarification_response_path, "selected-direction": state.selected_direction_path, "reverify-scope": ( state.reverify_scope if state.task_type == "implementation-planning" else "" ), "pr-template-path": pr_template, "fix-cycle": state.fix_cycle, "host-session-context-json": serialize_host_session_context( _host_session_context(state) ), } role_argv = render_role_args(state) rendered["role-count"] = [ role_argv[index + 1] for index, token in enumerate(role_argv) if token == "--role-count" ] rendered["role-model"] = [ role_argv[index + 1] for index, token in enumerate(role_argv) if token == "--role-model" ] return rendered def _render_argv(rendered: dict[str, Any], *, host_runtime: str) -> list[str]: """Flatten render arguments into the canonical render-bundle argv.""" argv = ["--lead-runtime", host_runtime] for name, raw_value in rendered.items(): values = raw_value if isinstance(raw_value, list) else [raw_value] for value in values: if not isinstance(value, str): raise WizardError( f"wizard render arg --{name} must be a string" ) argv.extend([f"--{name}", value]) return argv def _clarification_sidecar_line(state: WizardState) -> Optional[str]: """확인 블록에 찍는 `user-responses/` 첨부 현황. picker 단계의 옵션 라벨에도 같은 사실이 붙지만 그 화면을 지나면 사라지고, 확인 블록에는 final-report 경로만 남았다. 그 줄 바로 밑에 "범위를 좁히지 못함" 이 오니 두 줄이 겹쳐 "답변이 안 붙었다" 로 읽혔다 — 실제로는 첨부돼 반영되고 있었다. 실행 직전 화면에서 답변 id 를 직접 보여 그 오해를 없앤다. """ if not state.clarification_response_path or not state.project_root: return None report = _resolve_path( state.clarification_response_path, Path(state.project_root) ) files = user_response_sidecars(report) if not files: return _msg(state.workspace_root, "confirmation", "clarification_sidecars_empty") answers = sorted(sidecar_answers(report)) return _msg( state.workspace_root, "confirmation", "clarification_sidecars_attached", files=str(len(files)), count=str(len(answers)), ids=", ".join(answers) or _msg( state.workspace_root, "confirmation", "clarification_sidecars_none_parsed"), ) def _reverify_scope_line(state: WizardState) -> Optional[str]: """이번 clarification 재실행이 좁혀질지 — 확인 단계에서 보여주는 줄. 사용자가 범위를 직접 골랐으면 그 선택을 찍는다. 고르지 않았으면(또는 좁힐 수 없어 질문 자체가 안 뜬 경우) 예측을 찍는다. 예측의 절반(답변된 id 가 stage 로 되짚어지는지)은 base SHA 없이 직전 리포트만으로 이미 정해져 있는데, 지금까지는 `okstra recap assemble` 을 따로 돌려야만 보였고 run 이 시작된 뒤 full 로 밝혀지면 두 시간을 물린 뒤였다. 확인 단계는 그 전에 되돌릴 수 있는 마지막 지점이다. """ preview = _reverify_scope_preview(state) if preview is None: return None if state.reverify_scope == "full": return _msg(state.workspace_root, "confirmation", "reverify_scope_user_full") if state.reverify_scope and state.reverify_scope != "auto": return _msg(state.workspace_root, "confirmation", "reverify_scope_user_stages", stages=state.reverify_scope) if preview["unlinkedIds"]: return _msg(state.workspace_root, "confirmation", "reverify_scope_unlinked", ids=", ".join(preview["unlinkedIds"])) if not preview["wouldForceFull"]: return _msg(state.workspace_root, "confirmation", "reverify_scope_incremental") return _msg(state.workspace_root, "confirmation", "reverify_scope_full", reason=preview["reason"]) def confirmation_block(state: WizardState) -> str: """Human-readable echo of the resolved selections (for the Confirm step).""" header = _msg(state.workspace_root, "confirmation", "header") lines: list[str] = [header] lines.append(f" task-type : {state.task_type}") lines.append(f" task-key : {state.task_group}/{state.task_id}") lines.append(f" brief : {state.brief_path or '(none)'}") if state.analysis_target: lines.append(f" analysis-target: {state.analysis_target}") if state.feature_evidence_path or state.project_evidence_path: evidence_status = { str(item.report_path): item.review_status for item in _resolve_analysis_evidence(state) } if state.feature_evidence_path: lines.append( " feature-evidence: " f"{state.feature_evidence_path} " f"({evidence_status.get(state.feature_evidence_path, 'unknown')})" ) if state.project_evidence_path: lines.append( " project-evidence: " f"{state.project_evidence_path} " f"({evidence_status.get(state.project_evidence_path, 'unknown')})" ) if state.task_type == "final-verification": lines.append(" base-ref : (selected stage worktree)") elif state.task_type == "implementation" and state.reuse_worktree: lines.append(_msg(state.workspace_root, "confirmation", "base_ref_stage_isolated")) elif state.reuse_worktree: lines.append(_msg(state.workspace_root, "confirmation", "base_ref_reuse_task_dir", task_key=f"{state.task_group}/{state.task_id}")) else: lines.append(f" base-ref : {state.base_ref}") worktree_line = _worktree_preview_line(state) if worktree_line is not None: lines.append(worktree_line) role_selection = _role_selection_enabled(state) if state.task_type == "implementation" and not role_selection: lines.append(f" executor : {state.executor or '(default)'}") lines.append( _msg(state.workspace_root, "confirmation", "workers_implementation_default") ) elif not role_selection: roster = state.workers_override or ",".join(state.profile_workers) or "(profile default)" lines.append(f" workers : {roster}") if role_selection: profile = _load_role_profile_for_state(state) if state.host_entry_mode == "current-session": lines.append(_msg( state.workspace_root, "confirmation", "static_role", role="leader", ordinal="1", model="current-session", )) for requirement in _selectable_static_requirements(state, profile): count = _selected_role_count(state, requirement) selected = state.role_models.get(requirement.role, []) for ordinal in range(1, count + 1): model_ref = ( selected[ordinal - 1] if ordinal <= len(selected) else "(selection required)" ) lines.append(_msg( state.workspace_root, "confirmation", "static_role", role=requirement.role, ordinal=str(ordinal), model=model_ref, )) for requirement in profile.roles: if not requirement.dynamic: continue lines.append(_msg( state.workspace_root, "confirmation", "dynamic_role", role=requirement.role, )) else: lines.append( f" lead : {state.lead_provider or 'claude'} / " f"{state.lead_model or 'default'}" ) if state.claude_model: lines.append(f" claude-model : {state.claude_model}") if state.codex_model: lines.append(f" codex-model : {state.codex_model}") if state.antigravity_model: lines.append(f" antigravity-model : {state.antigravity_model}") if state.grok_model: lines.append(f" grok-model : {state.grok_model}") if state.kimi_model: lines.append(f" kimi-model : {state.kimi_model}") if state.report_writer_model or state.report_writer_provider: lines.append( f" report-writer : {state.report_writer_provider or 'claude'} / " f"{state.report_writer_model or 'default'}" ) lines.append(f" directive : {state.directive or '(none)'}") if state.related_tasks_raw: lines.append(f" related-tasks : {state.related_tasks_raw}") if (not role_selection and state.task_type in ("requirements-discovery", "error-analysis", "implementation-planning", "final-verification")): lines.append(f" critic : {state.critic or '(off)'}") if state.task_type in _STAGE_SCOPED_TASK_TYPES: lines.append(f" approved-plan : {state.approved_plan_path}") stage = wizard_stage_confirmation_label( task_type=state.task_type, selected_stage=state.selected_stage, whole_task_label=_msg( state.workspace_root, "confirmation", "stage_whole_task" ), ) lines.append(f" stage : {stage}") if state.clarification_response_path: lines.append(f" clarification : {state.clarification_response_path}") sidecar_line = _clarification_sidecar_line(state) if sidecar_line is not None: lines.append(sidecar_line) reverify_line = _reverify_scope_line(state) if reverify_line is not None: lines.append(reverify_line) if state.selected_direction_path: lines.append(f" selected-direction: {state.selected_direction_path}") if state.task_type == "release-handoff" and state.handoff_mode: scope = ( _msg(state.workspace_root, "confirmation", "handoff_scope_whole_task") if state.handoff_mode == "whole-task" else _msg(state.workspace_root, "confirmation", "handoff_scope_stage_group", stages=state.handoff_stages) ) lines.append(f" handoff scope : {scope}") if state.task_type == "release-handoff" and state.pr_template_path: lines.append(f" pr-template : {state.pr_template_path} ({state.pr_template_scope or 'once'})") return "\n".join(lines) def _wizard_persist_actions(state: WizardState) -> list[dict[str, str]]: if state.task_type != "release-handoff": return [] if not state.pr_template_path: return [] if state.pr_template_scope not in ("project", "global"): return [] return [ { "command": "config.set", "key": "pr-template-path", "scope": state.pr_template_scope, "value": state.pr_template_path, } ] def wizard_outcome(state: WizardState) -> dict[str, Any]: """Public outcome for callers that need launch data and follow-up writes. `renderArgs` carries only what `okstra render-bundle` accepts, so a caller can pass every entry through unfiltered — which is exactly what the okstra-run skill is told to do. Signals the skill consumes itself, like the unattended stage chain, live under `orchestration`; mixing them into `renderArgs` made the renderer reject the wizard's own output. """ if state.aborted: raise WizardError("wizard was aborted by the user — outcome is unavailable") if state.confirmed is not True: raise WizardError("wizard is not complete — outcome is unavailable") rendered = render_args(state) return { "renderArgs": rendered, "renderArgv": _render_argv(rendered, host_runtime=state.host_runtime), "orchestration": {"chainStages": _stage_intent(state).chain_stages}, "persistActions": _wizard_persist_actions(state), "confirmationText": confirmation_block(state), } # ---- File I/O helpers (used by CLI) ------------------------------------- _LEGACY_SOURCE_FIELD = "legacySource" def _legacy_v2_companion_path(path: Path) -> Path: return path.with_name(f"{path.name}.v2.json") def _content_addressed_v2_companion_path( path: Path, source_digest: str, suffix: int | None = None, ) -> Path: numbered = f".{suffix}" if suffix is not None else "" return path.with_name(f"{path.name}.v2.{source_digest}{numbered}.json") def _content_addressed_v2_companion_paths( path: Path, source_digest: str, ) -> tuple[Path, ...]: canonical = _content_addressed_v2_companion_path(path, source_digest) numbered_prefix = f"{path.name}.v2.{source_digest}." numbered = tuple(sorted( candidate for candidate in path.parent.iterdir() if candidate.is_file() and candidate.name.startswith(numbered_prefix) and candidate.name.endswith(".json") )) return (canonical, *numbered) def _available_v2_companion_path(path: Path, source_digest: str) -> Path: candidate = _content_addressed_v2_companion_path(path, source_digest) suffix = 1 while candidate.exists(): candidate = _content_addressed_v2_companion_path( path, source_digest, suffix, ) suffix += 1 return candidate def _legacy_source_metadata(path: Path, source: bytes) -> dict[str, str]: return { "path": str(path.resolve()), "sha256": hashlib.sha256(source).hexdigest(), } def _matching_v2_companion( path: Path, source_metadata: dict[str, str], ) -> dict[str, Any] | None: try: payload = load_owned_object(path, artifact="wizard state") except (OSError, UnicodeError, JsonBoundaryError): return None if not isinstance(payload, dict): return None if payload.get("executionIdentityVersion") != 2: return None if payload.get(_LEGACY_SOURCE_FIELD) != source_metadata: return None return payload def _load_matching_v2_state( paths: tuple[Path, ...], source_metadata: dict[str, str], ) -> tuple[WizardState | None, Path | None]: for companion_path in paths: payload = _matching_v2_companion(companion_path, source_metadata) if payload is None: continue try: return WizardState.from_json(payload), companion_path except WizardError: continue return None, None def load_state_file(path: Path) -> WizardState: source_path = Path(path) source = source_path.read_bytes() data = load_owned_object(source_path, artifact="wizard state") source_version = data.get("executionIdentityVersion", 1) source_metadata = _legacy_source_metadata(source_path, source) source_digest = source_metadata["sha256"] resumed_state = None companion_path = None if source_version in (None, 1): content_paths = _content_addressed_v2_companion_paths( source_path, source_digest, ) resumed_state, companion_path = _load_matching_v2_state( content_paths, source_metadata, ) if resumed_state is None: resumed_state, _ = _load_matching_v2_state( (_legacy_v2_companion_path(source_path),), source_metadata, ) state = resumed_state or WizardState.from_json(data) state._source_execution_identity_version = source_version if source_version in (None, 1): state._legacy_source_metadata = source_metadata state._v2_companion_path = companion_path or _available_v2_companion_path( source_path, source_digest, ) return state def _save_resumed_state_file(path: Path, state: WizardState) -> None: """Persist native v2 state while keeping legacy resume sources read-only.""" if getattr(state, "_source_execution_identity_version", 2) in (None, 1): payload = state.to_json() payload[_LEGACY_SOURCE_FIELD] = state._legacy_source_metadata write_json_atomic(state._v2_companion_path, payload) return save_state_file(path, state) def save_state_file(path: Path, state: WizardState) -> None: write_json_atomic(Path(path), state.to_json()) # ---- CLI entrypoint ----------------------------------------------------- def _cli(argv: list[str]) -> int: """``python3 -m okstra_ctl.wizard `` driver. Subcommands: init --state-file PATH --workspace-root P --project-root P --project-id ID step --state-file PATH (--answer VALUE | --no-submit) render-args --state-file PATH confirmation --state-file PATH outcome --state-file PATH """ import argparse parser = argparse.ArgumentParser(prog="okstra_ctl.wizard") sub = parser.add_subparsers(dest="cmd", required=True) p_init = sub.add_parser("init") p_init.add_argument("--state-file", required=True) p_init.add_argument("--workspace-root", required=True) p_init.add_argument("--project-root", required=True) p_init.add_argument("--project-id", required=True) p_init.add_argument( "--host-runtime", default="claude-code", choices=default_host_registry().ids(), ) p_init.add_argument( "--entry-mode", default="current-session", choices=("spawn-process", "current-session"), ) p_init.add_argument("--available-function", action="append", default=[]) p_init.add_argument("--critic", default="") p_step = sub.add_parser("step") p_step.add_argument("--state-file", required=True) p_step.add_argument("--answer", default=None) p_step.add_argument( "--no-submit", action="store_true", help="Fetch the current prompt without submitting an answer.", ) p_render = sub.add_parser("render-args") p_render.add_argument("--state-file", required=True) p_conf = sub.add_parser("confirmation") p_conf.add_argument("--state-file", required=True) p_outcome = sub.add_parser("outcome") p_outcome.add_argument("--state-file", required=True) args = parser.parse_args(argv) state_path = Path(args.state_file) if args.cmd == "init": state = init_state( workspace_root=args.workspace_root, project_root=args.project_root, project_id=args.project_id, host_runtime=args.host_runtime, host_entry_mode=args.entry_mode, available_functions=args.available_function, ) if args.critic: state.critic = args.critic save_state_file(state_path, state) first = next_prompt(state) print(json.dumps({"ok": True, "next": prompt_payload(state, first)}, ensure_ascii=False, indent=2)) return 0 if args.cmd == "step": state = load_state_file(state_path) persisted_state = copy.deepcopy(state) if args.no_submit and args.answer is not None: print(json.dumps( {"ok": False, "error": "--no-submit and --answer are mutually exclusive"}, ensure_ascii=False, indent=2, )) return 2 if not args.no_submit and args.answer is None: print(json.dumps( { "ok": False, "error": ( "step requires --answer VALUE (use --answer '' to submit an " "empty value, or --no-submit to peek at the current prompt)" ), }, ensure_ascii=False, indent=2, )) return 2 try: if args.no_submit: result = {"echo": "", "next": prompt_payload(state, next_prompt(state))} else: result = submit(state, args.answer) except WizardError as exc: _save_resumed_state_file(state_path, persisted_state) try: current = prompt_payload(state, next_prompt(state)) except WizardError: # 현재 step 의 build 자체가 실패하면(예: 손상된 Stage Map 으로 # _build_stage_pick·_build_handoff_stage_pick 가 raise) recovery 재렌더가 # 같은 예외를 다시 던져 이중 실패한다 — try 밖이라 _cli 를 탈출해 # traceback 으로 죽는다. current 를 비워 첫 예외를 깨끗한 envelope 로 낸다. current = None print(json.dumps({"ok": False, "error": str(exc), "current": current}, ensure_ascii=False, indent=2)) return 0 _save_resumed_state_file(state_path, state) print(json.dumps({"ok": True, **result}, ensure_ascii=False, indent=2)) return 0 if args.cmd == "render-args": state = load_state_file(state_path) try: rendered = render_args(state) except WizardError as exc: print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False, indent=2)) return 0 print(json.dumps({"ok": True, "args": rendered}, ensure_ascii=False, indent=2)) return 0 if args.cmd == "confirmation": state = load_state_file(state_path) print(json.dumps({"ok": True, "text": confirmation_block(state)}, ensure_ascii=False, indent=2)) return 0 if args.cmd == "outcome": state = load_state_file(state_path) try: out = wizard_outcome(state) except WizardError as exc: print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False, indent=2)) return 0 print(json.dumps({"ok": True, "outcome": out}, ensure_ascii=False, indent=2)) return 0 return 2 if __name__ == "__main__": import sys raise SystemExit(_cli(sys.argv[1:]))