"""확인 화면 — `confirmation_block` 과 confirm 단계.""" from __future__ import annotations from pathlib import Path from typing import Optional from okstra_ctl.registry.host_registry import default_host_registry from okstra_ctl.clarification_items import sidecar_answers, user_response_sidecars from okstra_ctl.incremental_scope import CARRY_ALL_SCOPE from okstra_ctl.wizard_stage_intent import wizard_stage_confirmation_label, resolve_wizard_stage_intent from okstra_ctl.worktree import ( compute_worktree_path, preview_worktree_decision, resolve_stage_worktree_decision, ) from okstra_ctl.work_categories import resolve_work_category from .render import render_args from .ids import S_CONFIRM, _STAGE_SCOPED_TASK_TYPES from .state import ( Prompt, WizardError, WizardState, _load_role_profile_for_state, _role_selection_enabled, ) from .prompts import _msg, _opt, _p from .sources import _resolve_path, _reverify_scope_preview from .roles import _selectable_static_requirements, _selected_role_count from .steps_analysis import _resolve_analysis_evidence def _worktree_preview_required(state: WizardState) -> bool: """final-verification 은 stage worktree 를 읽기 전용으로 재사용하므로 별도 worktree 미리보기 줄을 confirmation 블록에 넣지 않는다.""" return state.task_type != "final-verification" 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: """확인 질문의 본문이 곧 선택 요약이다. 요약을 `okstra wizard confirmation` 의 별도 텍스트로만 두면 리드가 그것을 자기 표로 다시 쓰면서 줄을 빠뜨린다(실측 2026-09-08, jobs implementation: 리드가 "추가 지시·관련 작업·추가 응답 문서는 모두 없음" 한 줄로 뭉개고 directive·base-ref·brief 줄을 뺀 표를 냈다). 네이티브 질문 카드는 질문 텍스트를 반드시 그리므로 요약을 질문 텍스트에 싣는다 — 리드가 빼놓을 자리가 없다. `okstra wizard confirmation` 은 텍스트 호스트와 재표시용으로 같은 블록을 낸다. """ t = _p(state.workspace_root, "confirm") if not state.confirmation_prompt: state.confirmation_stages = resolve_wizard_stage_intent( task_type=state.task_type, selected_stage=state.selected_stage, selected_stages=state.selected_stages, ).chain_stages state.confirmation_scope = render_args(state) state.confirmation_prompt = f"{confirmation_block(state)}\n\n{t['label']}" return Prompt( step=S_CONFIRM, kind="pick", label=state.confirmation_prompt, 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 state.user_authorization = {} return "confirm: abort" if value not in ("proceed", "edit"): raise WizardError( f"expected 'proceed' / 'edit' / 'abort', got: {value!r}" ) if value == "proceed": stages = resolve_wizard_stage_intent( task_type=state.task_type, selected_stage=state.selected_stage, selected_stages=state.selected_stages, ).chain_stages if (not state.confirmation_prompt or state.confirmation_scope != render_args(state) or state.confirmation_stages != stages): raise WizardError("confirmation scope changed; display the confirmation again before proceeding") state.user_authorization = { "schemaVersion": "1.0", "source": "wizard-confirmation", "response": value, "prompt": state.confirmation_prompt, "scope": state.confirmation_scope.copy(), "stageScope": state.confirmation_stages.split(",") if state.confirmation_stages else [], } else: state.user_authorization = {} state.confirmation_prompt = "" state.confirmation_scope = {} state.confirmation_stages = "" state.confirmed = value == "proceed" return f"confirm: {value}" 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 == CARRY_ALL_SCOPE: return _msg(state.workspace_root, "confirmation", "reverify_scope_user_carry_all") 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) native_provider = default_host_registry().resolve( state.host_runtime ).descriptor.native_provider_id leader_model = ( "current-session" if state.host_entry_mode == "current-session" else f"{native_provider} (host runtime default)" ) lines.append(_msg( state.workspace_root, "confirmation", "static_role", role="leader", ordinal="1", model=leader_model, )) 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'})") if state.fix_cycle: lines.append(f" fix-cycle : {state.fix_cycle}") report_writer_models = state.role_models.get("report-writer", []) translator_model = ( report_writer_models[0] if report_writer_models else f"{state.report_writer_provider or 'claude'}/{state.report_writer_model or 'default'}" ) lines.append(_msg( state.workspace_root, "confirmation", "translation_scope", model=translator_model, )) lines.append(_msg( state.workspace_root, "confirmation", "provider_data_scope", project_root=state.project_root, )) return "\n".join(lines)