"""`okstra.sh` 대화형 입력 경로가 쓰는 조회 헬퍼들. 이 여섯 갈래는 원래 `scripts/lib/okstra/interactive.sh` 의 파이썬 heredoc 안에 있었다(docs/coding-rules.md R1 위반). 셸 문자열 안에 있는 동안에는 import 도 테스트도 할 수 없었다. 로직과 출력 형식은 그대로 옮겼다 — 호출부의 파서를 건드리지 않기 위해서다. """ from __future__ import annotations import argparse import json import os import sys from pathlib import Path from . import next_phase from .ids import slugify_task_segment from .implementation_outcome import reconcile_implementation_outcome from .json_boundary import JsonBoundaryError, load_owned_object from .paths import RunRef, task_dir from .run_context import latest_run_inputs def _resolve_task_root(args: argparse.Namespace) -> int: """`OK\\t` 또는 `MISSING` + 진단 줄들. task root 의 slug 경로 구성은 `paths.task_dir` 이 단일 기준점이다. 예전에 이 코드는 slugify 와 `.okstra/tasks//` 구조를 자체 재구현해 규칙이 바뀌면 조용히 어긋날 수 있었다. """ project_root = Path(args.project_root) requested_key = f"{args.project_id}:{args.task_group}:{args.task_id}" requested_key_ci = requested_key.lower() candidates: list[str] = [] catalog_keys: list[str] = [] catalog_path = project_root / ".okstra" / "discovery" / "task-catalog.json" if catalog_path.is_file(): try: catalog = load_owned_object(catalog_path, artifact="task catalog") except JsonBoundaryError as exc: sys.stderr.write(f"resolve_task_root: failed to parse {catalog_path}: {exc}\n") catalog = None if isinstance(catalog, dict): catalog_project_id = catalog.get("projectId") for entry in catalog.get("tasks") or []: if not isinstance(entry, dict): continue entry_key = entry.get("taskKey") or "" if not isinstance(entry_key, str) or not entry_key: continue belongs = ( catalog_project_id == args.project_id or entry.get("projectId") == args.project_id or entry_key.startswith(f"{args.project_id}:") ) if belongs: catalog_keys.append(entry_key) if entry_key.lower() != requested_key_ci: continue rel = entry.get("taskRootPath") or entry.get("taskRoot") if isinstance(rel, str) and rel: abs_path = ( Path(rel) if os.path.isabs(rel) else (project_root / rel).resolve() ) if abs_path.is_dir(): print(f"OK\t{abs_path}") return 0 candidates.append(str(abs_path)) slug_path = task_dir(project_root, args.task_group, args.task_id) if slug_path.is_dir(): print(f"OK\t{slug_path}") return 0 candidates.append(str(slug_path)) print("MISSING") print(f"REQUESTED\t{requested_key}") for candidate in candidates: print(f"TRIED\t{candidate}") for key in sorted(set(catalog_keys)): print(f"AVAILABLE\t{key}") return 0 def _reconcile_outcome(args: argparse.Namespace) -> int: """구현 결과 화해. 편의 기능이라 실패해도 run 을 끊지 않는다.""" try: reconcile_implementation_outcome(Path(args.project_root), Path(args.task_root)) except Exception as exc: # noqa: BLE001 - 아래에서 신호를 남긴다 # expected-miss: 화해는 편의 기능이고 호출부(okstra.sh)는 `set -e` 아래 # 있다. 여기서 던지면 평범한 run 이 통째로 중단된다. 대신 stderr 로 # 남겨 삼킨 사실이 보이게 한다. sys.stderr.write(f"reconcile-outcome: skipped ({exc})\n") return 0 def _autofill(args: argparse.Namespace) -> int: """`BRIEF=` / `TYPE=` 두 줄. 매니페스트를 못 읽으면 빈 값.""" try: data = load_owned_object(Path(args.manifest), artifact="task manifest") except JsonBoundaryError as exc: sys.stderr.write(f"autofill_from_manifest: failed to parse {args.manifest}: {exc}\n") return 0 brief = "" task_type = "" if args.need_brief: brief = (data.get("taskBriefPath") or "").strip() if args.need_type: # 포인터 판정은 파이썬이 갖는다 — `ready` 가 아니면 빈 문자열이라 # 셸이 센티널 목록을 따로 들고 있을 이유가 없다. task_type = next_phase.autofill_task_type(data) print(f"BRIEF={brief}") print(f"TYPE={task_type}") return 0 def _latest_report(args: argparse.Namespace) -> int: """`\\t` 한 줄. 없으면 아무것도 내지 않는다.""" types = ( (args.task_type,) if args.task_type else ("error-analysis", "requirements-discovery") ) ref = RunRef.latest_under(args.task_root, types) if ref is not None: sys.stdout.write(f"{ref.report}\t{ref.task_type}") return 0 def _manifest_brief(args: argparse.Namespace) -> int: """매니페스트의 `taskBriefPath` 한 줄.""" try: data = load_owned_object(Path(args.manifest), artifact="task manifest") except JsonBoundaryError as exc: sys.stderr.write(f"manifest-brief: failed to parse {args.manifest}: {exc}\n") return 1 print((data.get("taskBriefPath") or "").strip()) return 0 def _resume_args(args: argparse.Namespace) -> int: """직전 run 의 인자들을 NUL 로 구분해 흘린다(셸의 `read -d ''` 용). `default` 와 빈 값은 phase 기본값을 뜻하므로 넘기지 않는다 — 명시 플래그로 고정하면 다음 run 이 기본값 변경을 못 받는다. """ inputs = latest_run_inputs( args.project_root, args.task_group, args.task_id, phase_segment=slugify_task_segment(args.task_type), ) def emit(flag: str, value: str) -> None: if value: sys.stdout.write(flag + "\0" + value + "\0") def model(value: object) -> str: return value if isinstance(value, str) and value not in ("", "default") else "" workers = inputs.get("workers") if isinstance(workers, list): workers = ",".join(w for w in workers if isinstance(w, str)) elif not isinstance(workers, str): workers = "" emit("--workers", workers) emit("--lead-model", model(inputs.get("leadModel"))) emit("--claude-model", model(inputs.get("claudeModel"))) emit("--codex-model", model(inputs.get("codexModel"))) emit("--antigravity-model", model(inputs.get("antigravityModel"))) emit("--report-writer-model", model(inputs.get("reportWriterModel"))) directive = inputs.get("directive") emit("--directive", directive if isinstance(directive, str) else "") related = inputs.get("relatedTasks") emit("--related-tasks", related if isinstance(related, str) else "") return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="okstra interactive") sub = parser.add_subparsers(dest="op", required=True) resolve = sub.add_parser("resolve-task-root") resolve.add_argument("--project-root", required=True) resolve.add_argument("--project-id", required=True) resolve.add_argument("--task-group", required=True) resolve.add_argument("--task-id", required=True) resolve.set_defaults(handler=_resolve_task_root) reconcile = sub.add_parser("reconcile-outcome") reconcile.add_argument("--project-root", required=True) reconcile.add_argument("--task-root", required=True) reconcile.set_defaults(handler=_reconcile_outcome) autofill = sub.add_parser("autofill") autofill.add_argument("--manifest", required=True) autofill.add_argument("--need-brief", action="store_true") autofill.add_argument("--need-type", action="store_true") autofill.set_defaults(handler=_autofill) latest = sub.add_parser("latest-report") latest.add_argument("--task-root", required=True) latest.add_argument("--task-type", default="") latest.set_defaults(handler=_latest_report) brief = sub.add_parser("manifest-brief") brief.add_argument("--manifest", required=True) brief.set_defaults(handler=_manifest_brief) resume = sub.add_parser("resume-args") resume.add_argument("--project-root", required=True) resume.add_argument("--task-group", required=True) resume.add_argument("--task-id", required=True) resume.add_argument("--task-type", required=True) resume.set_defaults(handler=_resume_args) args = parser.parse_args(argv) return args.handler(args) if __name__ == "__main__": raise SystemExit(main())