"""options 단계 — directive, related tasks, clarification, pr template 의 "optional cached pick" seam 과 build/submit.""" from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Callable, Optional from okstra_ctl.clarification_items import user_response_sidecars from okstra_ctl.json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic from okstra_ctl.report_language import is_language_tag from okstra_ctl.pr_template import PrTemplateError, resolve_pr_template_path from okstra_project.dirs import project_json_path from okstra_project.state import StateError, list_project_tasks from .ids import ( PICK_SKIP, PICK_TYPE_CUSTOM, S_CLARIFICATION, S_CLARIFICATION_PICK, S_DIRECTIVE, S_DIRECTIVE_PICK, S_PR_TEMPLATE, S_PR_TEMPLATE_PICK, S_PR_TEMPLATE_SCOPE, S_RELATED_TASKS, S_RELATED_TASKS_PICK, S_REPORT_LANGUAGE, _LATEST_REPORT_TOKEN, _PROJECT_DEFAULT_TOKEN, _REUSE_LAST_TOKEN, _SIBLINGS_TOKEN, ) from .state import Option, Prompt, WizardError, WizardState from .prompts import _opt, _p, _pick_snippet from .sources import _latest_run_inputs, _require_file, _suggest_latest_final_report def _report_language_required(state: WizardState) -> bool: path = project_json_path(Path(state.project_root)) if not path.is_file(): return False try: data = load_owned_object(path, artifact="project config") except JsonBoundaryError as exc: raise WizardError(str(exc)) from exc return data.get("reportLanguage") in (None, "") def _build_report_language(state: WizardState) -> Prompt: text = _p(state.workspace_root, "report_language") return Prompt( step=S_REPORT_LANGUAGE, kind="text", label=text["label"], echo_template=text["echo_template"], ) def _submit_report_language(state: WizardState, value: str) -> Optional[str]: language = value.strip() if not is_language_tag(language): raise WizardError("보고서 언어 태그를 입력하세요: en, ko, fr, pt-BR") path = project_json_path(Path(state.project_root)) try: data = load_owned_object(path, artifact="project config") if data.get("reportLanguage") in (None, ""): data["reportLanguage"] = language write_owned_object_atomic(path, data, artifact="project config") except (OSError, JsonBoundaryError) as exc: raise WizardError(str(exc)) from exc return f"report-language: {data['reportLanguage']}" 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 _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 _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, recommended=True)) 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 _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 _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}"