"""identity 단계 — task pick/group/id/type, brief, base ref 의 build/submit.""" from __future__ import annotations import subprocess from pathlib import Path from collections.abc import Mapping from typing import Any, Optional from okstra_ctl import next_phase from okstra_ctl.ids import slugify_task_segment from okstra_ctl.group_context import ( QUEUE_DONE, QUEUE_IN_PROGRESS, QUEUE_NOT_STARTED, group_context_file, group_queue, is_group_context_file, parse_memory_entries, split_memory_region, ) from okstra_project.state import ( StateError, list_project_tasks, read_latest_task, read_task_manifest, find_task_root, ) from .ids import ( BASE_REF_FREE_INPUT_TOKEN, CANONICAL_BASE_REFS, PICK_TYPE_CUSTOM, PICK_USE_SUGGESTED, S_BASE_REF_PICK, S_BASE_REF_TEXT, S_BRIEF_CARRY, S_BRIEF_KEEP, S_BRIEF_PATH, S_BRIEF_PATH_PICK, S_TASK_GROUP, S_TASK_GROUP_TEXT, S_TASK_ID, S_TASK_ID_TEXT, S_TASK_PICK, S_TASK_TYPE, S_TASK_TYPE_TEXT, TASK_PICK_NEW_TOKEN, TASK_TYPES, TASK_TYPE_VALUES, _BRIEF_ENTRY_TASK_TYPES, _BRIEF_PREFIX, _RECENT_PREFIX, _RECOMMENDATION_CAP, ) from .state import Option, Prompt, WizardError, WizardState, _slug_or_die from .prompts import _opt, _p, _recommendation_budget, _static_options from .sources import ( _accept_brief_path, _brief_recommended_phase, _brief_suggestions, _existing_task_brief, _existing_task_manifest, _file_recency, _git_main_worktree, _latest_revision_requested_analysis_type, _load_profile_optional_workers, _load_profile_workers, _parse_iso_timestamp, _phase_after, _project_relative_path, _recent_task_types, _recently_used_brief_times, _report_data_for_pointer, _require_file, _resolve_reuse_worktree, ) 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 # --- 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 _recommended_task_entry( remaining: list[dict], latest_key: str, preferred_key: str = "", ) -> dict | None: """남은 task 중 이 run 이 권하는 하나. 현재 대화에서 전달한 task 를 우선한다. 전달값이 없으면 포인터가 `ready` 인 task 는 지금 바로 다음 phase 를 시작할 수 있는 task 다 — 그중 최신(카탈로그는 updatedAt 내림차순)을 권한다. `ready` 가 없으면 마지막으로 만진 task(`latest-task.json`)를, 그것도 목록에 없으면 최신 task 를 권한다. 실측(2026-09-09): 세 줄 전부가 `(추천)` 이었고 리드는 산문에서 2번을 권했다 — 추천이 여럿이면 라벨은 아무것도 고르지 않은 것이다. """ if preferred_key: for entry in remaining: if entry.get("taskKey") == preferred_key: return entry raise WizardError(f"preferred task not found in project catalog: {preferred_key}") if not remaining: return None for entry in remaining: pointer = next_phase.promote(entry.get("nextRecommendedPhase")) if pointer["status"] == next_phase.STATUS_READY: return entry for entry in remaining: if latest_key and entry.get("taskKey") == latest_key: return entry return remaining[0] 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" or e.get("taskKey") == state.preferred_task_key] # 추천은 하나이고 1번이다. 나머지는 최신순 그대로. recommended = _recommended_task_entry(remaining, latest_key, state.preferred_task_key) ordered = ( [recommended, *(e for e in remaining if e is not recommended)] if recommended is not None else remaining ) options: list[Option] = [] for entry in ordered[:_recommendation_budget(state)]: 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, recommended=entry is recommended)) 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.preferred_task_key = "" 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.preferred_task_key = value 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 = _recommended_task_type(state, 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 _suggest_group_briefs(state: WizardState, limit: int = 6) -> list[str]: """Return the selected task-group's okstra-brief-gen outputs in start order. ``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. The order is the group's start order (`group_context.group_queue`: brief ordinal) with tasks not yet started first, then the ones in progress, then the finished ones — the queue the group's `group-context.md` memory region shows. Briefs the queue does not know (no `type: brief` frontmatter) follow by recency. 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() or is_group_context_file(path): 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])) by_recency = [rel for _, rel in candidates] queue = _group_start_order(project_root, state.task_group) rank = {QUEUE_NOT_STARTED: 0, QUEUE_IN_PROGRESS: 1, QUEUE_DONE: 2} ordered = [ row.brief for row in sorted( (row for row in queue if row.brief in by_recency), key=lambda row: rank.get(row.status, 3), ) ] ordered.extend(rel for rel in by_recency if rel not in ordered) return ordered[:limit] def _group_start_order(project_root: Path, task_group: str) -> list: """`group-context.md` 의 okstra 영역에 기록된 항목으로 상태를 매긴 시작 순서.""" target = group_context_file(project_root, task_group) region = "" if target.is_file(): _, region, _ = split_memory_region(target.read_text(encoding="utf-8")) return group_queue(project_root, task_group, parse_memory_entries(region)) 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, _recommendation_budget(state)) t = _p(state.workspace_root, "task_group_no_suggestion") recent_prefix = t.get("recent_label_prefix", "") options: list[Option] = [] # 추천은 하나뿐이다: 가장 최근에 task·brief 활동이 있던 group — 새 task 는 # 진행 중인 작업 흐름에 속할 가능성이 가장 크다. 나머지는 후보다. for index, tg in enumerate(recent): options.append(_opt(f"{_RECENT_PREFIX}{tg}", f"{recent_prefix}{tg}", recommended=index == 0)) 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, _recommendation_budget(state)) t = _p(state.workspace_root, "task_id_no_suggestion") recent_prefix = t.get("recent_label_prefix", "") options: list[Option] = [] # 새 task 의 id 로 같은 group 의 기존 id 를 권할 근거는 없다 — 후보로만 # 싣고 추천은 비운다. 추천이 있는 경우는 brief frontmatter 의 제안뿐이고, # 그것은 위의 `task_id_with_suggestion` 분기다. 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 _brief_recommended_task_type(state: WizardState) -> str: """선택된 brief 가 지목한 진입 phase. brief 가 없거나 값이 없으면 ''.""" if not state.brief_path: return "" path = Path(state.brief_path) if not path.is_absolute(): path = Path(state.project_root) / path return _brief_recommended_phase(path) def _recommended_task_type( state: WizardState, manifest: Mapping[str, Any], ) -> str: """매니페스트 포인터가 ready 면 그 phase, 끝난 계획의 낡은 pending 이면 리포트 투영.""" filled = next_phase.autofill_task_type(manifest) if filled: return filled workflow = manifest.get("workflow") if not isinstance(workflow, Mapping): return "" if ( workflow.get("currentPhase") != "implementation-planning" or workflow.get("currentPhaseState") != "completed" ): return "" pointer = next_phase.project(_report_data_for_pointer(state, manifest)) if pointer["status"] != next_phase.STATUS_READY: return "" return pointer["phase"] def _implementation_needs_approval( state: WizardState, manifest: Mapping[str, Any], workflow: Mapping[str, Any], ) -> bool: if workflow.get("awaitingApproval") is True: return True if ( workflow.get("currentPhase") != "implementation-planning" or workflow.get("currentPhaseState") != "completed" ): return False frontmatter = _report_data_for_pointer(state, manifest).get("frontmatter") if not isinstance(frontmatter, dict): return True return frontmatter.get("approved") is not True 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] = [] budget = _recommendation_budget(state) def add(task_type: str, suffix: str = "", *, recommended: bool = False) -> None: if task_type not in description_by_type: return if any(o.value == task_type for o in options): return if len(options) >= budget: return options.append(_opt(task_type, f"{task_type}{suffix}", description_by_type[task_type], recommended=recommended)) manifest = _existing_task_manifest(state) workflow = manifest.get("workflow") if isinstance(manifest.get("workflow"), dict) else {} 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` 에서 온다. # 끝난 계획의 낡은 pending 은 최신 리포트에서 다시 투영한다. 승인 대기는 # `awaitingApproval` 또는 미승인 계획 리포트로 접미사를 붙인다. 열린 # C-NNN 때문에 blocked 면 재실행은 답이 기록된 뒤에만 고르라고 말한다. # 이어갈 매니페스트가 없는 새 task 는 포인터도 이력도 없다 — 이번 task 가 # 무슨 일인지 아는 유일한 입력은 방금 고른 brief 다. brief 가 자기 진입 # phase 를 적어 두었으면(okstra-brief-gen 의 "Recommended next phase") 그것을 # 추천으로 쓰고, 없으면 라이프사이클 첫 단계로 떨어진다. fresh_task = bool(state.is_new_task) and not workflow brief_phase = _brief_recommended_task_type(state) if fresh_task else "" recommended = (revision_requested or state.task_type or _recommended_task_type(state, manifest)) if not recommended and not workflow: recommended = brief_phase or TASK_TYPE_VALUES[0] if brief_phase and recommended == brief_phase: recommended_suffix = t["options"].get("_BRIEF_SUFFIX", recommended_suffix) pointer = next_phase.promote(workflow.get("nextRecommendedPhase")) if ( recommended == "implementation" and _implementation_needs_approval(state, manifest, workflow) ): 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, recommended=True) add(workflow.get("currentPhase") or "", rerun_suffix) if not fresh_task: add(_phase_after(recommended), next_suffix) # 새 task 의 남은 후보는 entry phase 로 한정한다. implementation 은 승인된 # 계획서를, final-verification 은 구현을, release-handoff 는 커밋을 요구하므로 # 선행 산출물이 없는 새 키에서는 어차피 진입할 수 없다 — 목록에 두면 고르는 # 순간 실패하는 함정이다. 그래도 필요하면 "직접 입력" 이 전체 화이트리스트를 # 받는다(`_submit_task_type_text`). fillers = _BRIEF_ENTRY_TASK_TYPES if fresh_task else TASK_TYPE_VALUES for tt in _recent_task_types(state): if tt in fillers: add(tt) for tt in fillers: 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] = [] # 후보는 `_RECOMMENDATION_CAP` 개까지다. 그룹에 brief 가 여럿이면 목록이 # 호스트 네이티브 선택기의 옵션 한도(claude-code AskUserQuestion 은 4개, # `adapters/hosts/claude-code/relay.md` 의 `nativeLimits`)를 넘고, 그때 # `CapabilityInteractionPort.plan` 이 `numbered-single` 로 내려 선택지가 # 아니라 번호 목록 텍스트로 나온다. 잘린 후보는 "직접 입력" 으로 간다. budget = _recommendation_budget(state) def add(value: str, label: str) -> None: if len(options) >= budget: return # 첫 후보만 추천이다: 이 task 의 기존 brief → (있으면) 표준 경로의 # brief → 그룹 시작 순서상 첫 미착수 brief. 각각이 그 자리에서 가장 # 타당한 입력이고, 그 뒤 줄들은 대안이다. options.append(_opt(value, label, recommended=not options)) if existing: add("__existing__", t["options"]["__existing__"].format(existing=existing)) # 표준 경로는 파일이 실제로 있을 때만 후보다. 없는 파일을 고르면 # `_require_file` 이 거절하므로, 그 줄은 선택지가 아니라 함정이고 추천은 # 더더욱 아니다. if ( standard and standard != existing and (Path(state.project_root) / standard).is_file() ): add("__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 add(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}" 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 _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}"