"""전원 투표만 모자란 구현 후보와, 그 표를 받아야 할 분석자를 낸다. `implementation-option-selection` 의 1라운드는 설계자들이 병렬로 돌아 서로의 후보를 보지 못한다. 그래서 자기가 낸 후보에만 실현 가능성 표를 남기고, 병합된 집합에는 분석자마다 다른 구멍이 생긴다. 유효성 규칙은 전원 투표를 요구하므로 (`implementation_options._validate_option_feasibility`) 그런 후보는 순위표에 오르지 못하고, 남는 후보가 하나도 없으면 run 이 `routing: blocked` 로 끝난다 — 실측(2026-09-10, dev-10629-4): 설계자 3명 로스터에서 IO-001·IO-002·IO-003 이 각각 `feasible` 2표를 받고도 빠진 분석자가 하나씩 달라 전부 탈락했다. 거절은 그 상태를 알려 줄 뿐 메우지 못한다. 이 명령이 앞으로 가는 길이다: 표만 모자란 후보를 세고, 어느 분석자에게 어떤 후보를 부쳐야 하는지 말한다. 재검증(reverify) 라운드는 주장을 반박하는 라운드이지 후보에 표를 남기는 라운드가 아니므로, 그 구멍을 메우는 디스패치는 리드가 이 목록을 보고 연다. 읽는 자리는 두 가지다. 조립 전이면 작성자 서사(`--narrative`), 이미 발행된 run 이면 리포트 레코드(`--report`). 로스터는 task-manifest 의 `recommendedWorkers` 에서 `report-writer` 를 뺀 것이고, 그것이 검증기가 `participating analysers` 로 쓰는 값과 같은 정의다(`validators/validate-run.py`). """ from __future__ import annotations import argparse import json import sys from collections.abc import Mapping from pathlib import Path from typing import Any from okstra_ctl.final_report_schema import load_schema_version from okstra_ctl.implementation_options import VoteGap, vote_gaps from okstra_ctl.json_boundary import JsonBoundaryError, load_owned_object from okstra_ctl.report_contract import CURRENT_REPORT_SCHEMA_VERSION from okstra_ctl.report_narrative import parse_narrative_structure class OptionVotesError(ValueError): """투표 구멍을 셀 입력이 없거나 읽히지 않는다.""" def _record_from_report(path: Path) -> dict[str, Any]: try: return load_owned_object(path, artifact="final-report data.json") except (JsonBoundaryError, OSError) as exc: raise OptionVotesError(f"report record is unreadable: {exc}") from exc def _record_from_narrative(path: Path) -> dict[str, Any]: """작성자 서사를 레코드 모양으로 읽는다. 값 결함은 무시한다 — 교정 원장이 고칠 자리이고, 표 구멍을 세는 데에는 후보 id 와 `feasibilityVotes` 만 있으면 된다. 여기서 서사 전체를 거절하면 아직 교정 중인 run 은 이 명령을 쓸 수 없다. """ try: markdown = path.read_text(encoding="utf-8") except OSError as exc: raise OptionVotesError(f"narrative is unreadable: {exc}") from exc schema = load_schema_version(CURRENT_REPORT_SCHEMA_VERSION) record, _defects = parse_narrative_structure(markdown, schema) return record def participating_analysers(manifest_path: Path) -> tuple[str, ...]: try: manifest = load_owned_object(manifest_path, artifact="task-manifest") except (JsonBoundaryError, OSError) as exc: raise OptionVotesError(f"task manifest is unreadable: {exc}") from exc roster = manifest.get("recommendedWorkers") if not isinstance(roster, list): raise OptionVotesError("task manifest has no recommendedWorkers roster") return tuple( str(worker) for worker in roster if str(worker) != "report-writer" ) def _selection(record: Mapping[str, Any]) -> Mapping[str, Any]: selection = record.get("implementationOptionSelection") if not isinstance(selection, Mapping): raise OptionVotesError( "the source has no implementationOptionSelection block — " "this command reads an implementation-option-selection run" ) return selection def _dispatch_lines(gaps: list[VoteGap]) -> list[str]: """분석자별로 부칠 후보 목록. 디스패치 단위가 분석자이기 때문이다.""" by_analyser: dict[str, list[str]] = {} for gap in gaps: for analyser in gap.missing: by_analyser.setdefault(analyser, []).append(gap.option_id) return [ f" {analyser}: {', '.join(options)}" for analyser, options in sorted(by_analyser.items()) ] def _render(gaps: list[VoteGap]) -> str: if not gaps: return ( "No candidate is short of votes alone. A blocked run here is " "blocked by something a vote cannot settle — safety blockers, " "unresolved feasibility facts, or too few feasible verdicts." ) lines = [ f"{len(gaps)} candidate(s) need only the missing feasibility votes:", "", ] lines += [ f" {gap.option_id}: {gap.feasible_votes} feasible so far, " f"missing {', '.join(gap.missing)}" for gap in gaps ] lines += ["", "Dispatch one vote-completion assignment per analyser:"] lines += _dispatch_lines(gaps) lines += [ "", "Each assignment asks that analyser for its own feasibility verdict, " "rationale, and counterevidence on the named candidate — nothing else. " "It generates no candidate, so the run stays in `candidate-comparison` " "mode; `preselected-validation` is a whole-run mode that would collapse " "the comparison to one direction.", ] return "\n".join(lines) def _gaps(args: argparse.Namespace) -> int: if bool(args.report) == bool(args.narrative): raise OptionVotesError("pass exactly one of --report or --narrative") record = ( _record_from_report(args.report) if args.report else _record_from_narrative(args.narrative) ) gaps = vote_gaps( _selection(record), participating_analysers(args.task_manifest) ) if args.json: print(json.dumps( { "gaps": [ { "optionId": gap.option_id, "missing": list(gap.missing), "feasibleVotes": gap.feasible_votes, } for gap in gaps ] }, ensure_ascii=False, indent=2, )) else: print(_render(gaps)) return 0 _CLI_DESCRIPTION = ( "Report the implementation candidates that only lack feasibility votes, " "and which analyser owes each one." ) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=_CLI_DESCRIPTION, prog="okstra option-votes" ) subparsers = parser.add_subparsers(dest="command", required=True) gaps_parser = subparsers.add_parser("gaps") gaps_parser.add_argument( "--task-manifest", type=Path, required=True, help="the task's task-manifest.json — its roster names the analysers") gaps_parser.add_argument( "--report", type=Path, help="a published final-report `.data.json`") gaps_parser.add_argument( "--narrative", type=Path, help="the report writer's narrative markdown, before assembly") gaps_parser.add_argument("--json", action="store_true") return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) try: return _gaps(args) except OptionVotesError as exc: print(f"okstra option-votes: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))