"""Validate one selected implementation direction before planning starts.""" from __future__ import annotations import hashlib import json import os import posixpath import re import stat from collections import Counter from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping from .clarification_items import USER_INPUT_BLOCKS, progress_blocking_ids from .final_report_paths import ( final_report_data_path, final_report_markdown_path, is_full_reading_copy_path, ) from .exact_coverage import ExactCoverageError, calculate_plan_exact_coverage from .final_report_schema import ( SchemaError, load_schema_for_data, validate as validate_schema, ) from .json_boundary import ( JsonBoundaryError, load_owned_object_snapshot, write_owned_object_atomic, ) from .report_views import normalize_direction_selection_identity from .qa_commands import find_denied_tokens from .scope_provenance import brief_end_state_id_sequence from .user_response import UserResponseError, parse_direction_selection _TASK_TYPE = "implementation-option-selection" # 손잡이는 리포트 **기록**이다. 전체 열람본(`.md`)은 `2bcd575` 이후 완료 산출물이 # 아니라 요청 시 렌더라(`dispatch_state._required_worker_artifacts`) run 의 reports # 디렉터리에 없다. `--approved-plan` 이 이미 같은 이유로 기록만 받는다 # (`final_report_paths.require_approved_plan_record`). _REPORT_RE = re.compile( r"^final-report-implementation-option-selection-(?P\d{3,})\.data\.json$" ) _FRONTMATTER_RE = re.compile( r"\A---[ \t]*\r?\n(?P.*?)(?:\r?\n)---[ \t]*(?:\r?\n|\Z)", re.DOTALL, ) class DirectionSelectionError(ValueError): """Raised when a direction report cannot authorize planning.""" @dataclass(frozen=True) class SelectedDirection: mode: str task_key: str source_report: str source_data_sha256: str selection_evidence: str option_id: str option_name: str selection_note: str user_constraints: tuple[str, ...] direction: dict[str, Any] requirement_coverage: tuple[dict[str, Any], ...] planning_invariants: tuple[dict[str, Any], ...] @dataclass(frozen=True) class SelectedDirectionSnapshot: data: Mapping[str, Any] path: Path relative_path: str sha256: str def lexical_absolute_path(path: Path) -> Path: """Make a user path absolute without following symbolic links.""" return Path(os.path.abspath(os.fspath(Path(path).expanduser()))) def _strict_regular_file(path: Path, label: str) -> Path: path = lexical_absolute_path(path) try: mode = path.lstat().st_mode except OSError as exc: raise DirectionSelectionError(f"{label} file not found: {path}") from exc if stat.S_ISLNK(mode): raise DirectionSelectionError(f"{label} path contains a symlink: {path}") if not stat.S_ISREG(mode): raise DirectionSelectionError(f"{label} must be a real regular file: {path}") return path def validate_task_artifact_path( path: Path, task_root: Path, label: str ) -> Path: """Reject links from the task root through one regular artifact file.""" path = lexical_absolute_path(path) task_root = lexical_absolute_path(task_root) try: relative = path.relative_to(task_root) except ValueError as exc: raise DirectionSelectionError( f"{label} must stay under the selected task root" ) from exc current = task_root for part in ("", *relative.parts): current = current if not part else current / part try: mode = current.lstat().st_mode except OSError as exc: raise DirectionSelectionError( f"{label} file not found: {path}" ) from exc if stat.S_ISLNK(mode): raise DirectionSelectionError( f"{label} path contains a symlink: {current}" ) if not stat.S_ISREG(mode): raise DirectionSelectionError(f"{label} must be a real regular file: {path}") try: path.resolve(strict=True).relative_to(task_root.resolve(strict=True)) except (OSError, ValueError) as exc: raise DirectionSelectionError( f"{label} must resolve inside the selected task root" ) from exc return path def _selection_layout(report: Path) -> tuple[Path, str]: match = _REPORT_RE.fullmatch(report.name) if ( match is None or report.parent.name != "reports" or report.parent.parent.name != _TASK_TYPE or report.parent.parent.parent.name != "runs" ): raise DirectionSelectionError( "report path task type must be implementation-option-selection" ) task_root = report.parents[3] return task_root, match.group("seq") def _parse_data(data_path: Path) -> tuple[dict[str, Any], bytes]: try: snapshot = load_owned_object_snapshot( data_path, artifact="implementation option data" ) except JsonBoundaryError as exc: raise DirectionSelectionError(f"data JSON is invalid: {data_path}") from exc return snapshot.value, snapshot.raw_bytes def _validate_report_identity(data: Mapping[str, Any], expected_task_key: str) -> None: header = data.get("header") frontmatter = data.get("frontmatter") if not isinstance(header, Mapping) or not isinstance(frontmatter, Mapping): raise DirectionSelectionError("report data is missing taskType metadata") if ( header.get("taskType") != _TASK_TYPE or frontmatter.get("taskType") != _TASK_TYPE ): raise DirectionSelectionError( "report taskType must be implementation-option-selection" ) if header.get("taskKey") != expected_task_key: raise DirectionSelectionError( "report taskKey does not match the requested planning task" ) def _validate_schema(data: dict[str, Any]) -> None: try: errors = validate_schema(data, load_schema_for_data(data)) except SchemaError as exc: raise DirectionSelectionError( f"selection report schema validation failed: {exc}" ) from exc if errors: raise DirectionSelectionError( "selection report schema validation failed: " + "; ".join(errors) ) def _relative_source(path: Path, task_root: Path, label: str) -> str: try: return path.relative_to(task_root).as_posix() except ValueError as exc: raise DirectionSelectionError( f"{label} must stay under the selected task root" ) from exc def _sidecar_metadata(sidecar_text: str, key: str) -> str: frontmatter = _FRONTMATTER_RE.match(sidecar_text) if frontmatter is None: return "" match = re.search( rf"^{re.escape(key)}:[ \t]*(\S.*?)[ \t]*$", frontmatter.group("body"), re.MULTILINE, ) return match.group(1) if match else "" def _comparison_selection( *, report: Path, data_path: Path, data_bytes: bytes, task_root: Path, seq: str, expected_task_key: str, ) -> tuple[str, str, str, tuple[str, ...]]: sidecar = _strict_regular_file( report.parent.parent / "user-responses" / f"user-response-implementation-option-selection-{seq}.md", "sidecar", ) sidecar = validate_task_artifact_path(sidecar, task_root, "sidecar") try: sidecar_text = sidecar.read_text(encoding="utf-8") parsed = parse_direction_selection(sidecar_text) except (OSError, UnicodeError, UserResponseError) as exc: raise DirectionSelectionError(str(exc)) from exc if parsed is None: raise DirectionSelectionError("sidecar has no DIRECTION SELECTION block") if _sidecar_metadata(sidecar_text, "task-key") != expected_task_key: raise DirectionSelectionError("sidecar taskKey does not match the planning task") if _sidecar_metadata(sidecar_text, "task-type") != _TASK_TYPE: raise DirectionSelectionError( "sidecar taskType must be implementation-option-selection" ) expected_report = _relative_source(report, task_root, "source report") expected_data = _relative_source(data_path, task_root, "source data") if parsed.source_report != expected_report: raise DirectionSelectionError("sidecar source report does not match report path") if parsed.source_data != expected_data: raise DirectionSelectionError("sidecar source data does not match data path") digest = hashlib.sha256(data_bytes).hexdigest() if parsed.source_data_sha256 != digest: raise DirectionSelectionError("sidecar source data digest does not match bytes") if parsed.seq != seq: raise DirectionSelectionError("sidecar sequence does not match report sequence") constraints = tuple( line.strip() for line in parsed.constraints.splitlines() if line.strip() ) return ( _relative_source(sidecar, task_root, "selection evidence"), parsed.option_id, parsed.selection_note, constraints, ) def _preselected_selection( selection: Mapping[str, Any], ) -> tuple[str, str, str, tuple[str, ...]]: if selection.get("routing") != "implementation-planning": raise DirectionSelectionError( "preselected direction routing must be implementation-planning" ) evidence = selection.get("preselectedDirection") if not isinstance(evidence, Mapping): raise DirectionSelectionError("preselected direction evidence is missing") confirmation = str(evidence.get("confirmationEvidence") or "").strip() citation = str(evidence.get("citation") or "").strip() option_id = str(evidence.get("optionId") or "").strip() if not confirmation or not citation: raise DirectionSelectionError( "preselected direction evidence and citation are required" ) return f"{confirmation} ({citation})", option_id, "", () def _selected_option( selection: Mapping[str, Any], option_id: str ) -> tuple[Mapping[str, Any], str]: """선택된 후보를 **id 로만** 특정한다. 사이드카의 `Option-Name` 은 대조하지 않는다. HTML 열람본은 번역 오버레이를 적용한 사본으로 렌더되므로(`report_html/render.py` `_localize`) Export 가 적는 이름은 독자의 언어다 — `reportLanguage` 가 `en` 이 아닌 런에서 영어 정본과의 바이트 비교는 통과할 수 있는 값이 하나도 없었다. id 는 번역되지 않으므로 신원은 id 가 온전히 결정하고, 하류로 나가는 이름은 아래에서 리포트 값으로 정규화된다. """ options = selection.get("rankedOptions") if not isinstance(options, list): raise DirectionSelectionError("ranked candidate list is missing") option = next( (row for row in options if isinstance(row, Mapping) and row.get("id") == option_id), None, ) if option is None: raise DirectionSelectionError("selected option is not a ranked candidate") try: report_option_id, report_option_name = normalize_direction_selection_identity( str(option.get("id") or ""), str(option.get("name") or "") ) except ValueError as exc: raise DirectionSelectionError(str(exc)) from exc if report_option_id != option_id: raise DirectionSelectionError("selected candidate id does not match report") coverage = option.get("coverageSummary") if not isinstance(coverage, Mapping) or coverage.get("coverageVerdict") != "exact": raise DirectionSelectionError("selected candidate must have exact coverage") return option, report_option_name def _validate_no_blockers(data: Mapping[str, Any], option: Mapping[str, Any]) -> None: if option.get("safetyBlockers") or option.get("unresolvedFeasibilityFacts"): raise DirectionSelectionError("selected candidate has a safety blocker") blockers = progress_blocking_ids( data.get("clarificationItems"), USER_INPUT_BLOCKS ) if blockers: raise DirectionSelectionError( "selection report has an unresolved next-phase blocker" ) def _direction_payload(option: Mapping[str, Any]) -> dict[str, Any]: fields = ( "goal", "coreMechanism", "architectureBoundaries", "expectedChangeAreas", "codeEvidence", "scopeCommitments", ) return _copy_json({field: option.get(field) for field in fields if field in option}) def _copy_json(value: Any) -> Any: """Return a JSON-only deep copy without sharing mutable report data.""" return json.loads(json.dumps(value, ensure_ascii=False)) def resolve_selected_direction( report_path: Path, expected_task_key: str, ) -> SelectedDirection: """Validate a selection report record in the documented fail-closed order. 손잡이는 `.data.json` 이다. 전체 열람본은 어디서도 열리지 않는다 — 사이드카의 `source-report` 와 대조할 이름으로만 쓰인다. 그 파일을 실물로 요구하는 동안 이 게이트는 통과할 수 있는 입력이 0개였고, 후보 비교가 끝나도 계획으로 넘어갈 길이 없었다. 실측: fontsninja-v3-site 의 선택 리포트 3쌍 전부 열람본이 없다. """ handle = lexical_absolute_path(Path(report_path)) if is_full_reading_copy_path(handle): raise DirectionSelectionError( "selected direction must be the report record (.data.json), not the " f"full reading copy: {report_path}\n" f" use: {final_report_data_path(handle)}" ) data_path = _strict_regular_file(handle, "data") task_root, seq = _selection_layout(data_path) data_path = validate_task_artifact_path(data_path, task_root, "data") # 사이드카의 `source-report` 는 열람본 이름을 적는다. 그 파일을 열지는 않는다 — # 대조할 이름일 뿐이다. report = final_report_markdown_path(data_path) data, data_bytes = _parse_data(data_path) _validate_report_identity(data, expected_task_key) _validate_schema(data) selection = data.get("implementationOptionSelection") if not isinstance(selection, Mapping): raise DirectionSelectionError("implementation option selection data is missing") mode = str(selection.get("mode") or "") if mode == "candidate-comparison": evidence, option_id, note, constraints = _comparison_selection( report=report, data_path=data_path, data_bytes=data_bytes, task_root=task_root, seq=seq, expected_task_key=expected_task_key, ) elif mode == "preselected-validation": evidence, option_id, note, constraints = _preselected_selection(selection) else: raise DirectionSelectionError(f"unsupported selection mode: {mode}") option, normalized_option_name = _selected_option(selection, option_id) _validate_no_blockers(data, option) return SelectedDirection( mode=mode, task_key=expected_task_key, source_report=_relative_source(report, task_root, "source report"), source_data_sha256=hashlib.sha256(data_bytes).hexdigest(), selection_evidence=evidence, option_id=option_id, option_name=normalized_option_name, selection_note=note, user_constraints=constraints, direction=_direction_payload(option), requirement_coverage=tuple(_copy_json(option.get("requirementCoverage") or ())), planning_invariants=tuple(_copy_json(option.get("planningInvariants") or ())), ) def write_selected_direction_snapshot( direction: SelectedDirection, destination: Path, ) -> Path: """Write the normalized planning input using stable JSON bytes.""" destination = Path(destination) destination.parent.mkdir(parents=True, exist_ok=True) payload = { "schemaVersion": "1.0", "taskKey": direction.task_key, "sourceReport": direction.source_report, "sourceDataSha256": direction.source_data_sha256, "selectionEvidence": direction.selection_evidence, "optionId": direction.option_id, "optionName": direction.option_name, "selectionNote": direction.selection_note, "userConstraints": list(direction.user_constraints), "direction": direction.direction, "requirementCoverage": list(direction.requirement_coverage), "planningInvariants": list(direction.planning_invariants), } write_owned_object_atomic( destination, payload, artifact="selected implementation direction", ) return destination _PLAN_EXECUTION_FIELDS = ( "directionRealization", "stageMap", "stages", "designPreparation", "dependencyMigrationRisk", "validationChecklist", "rollbackStrategy", "requirementCoverage", "coverageSummary", "variationPointAnalysis", "planBodyVerification", "stepwiseExecution", "supersessionLedger", "crossProjectDependencies", "decisionDrafts", "skippedAdrCandidates", "incrementalDecision", "userNarrative", ) def load_selected_direction_snapshot( snapshot_path: Path, ) -> tuple[SelectedDirectionSnapshot | None, list[str]]: path = lexical_absolute_path(Path(snapshot_path)) if path.name != "selected-direction.json" or path.parent.name != "instruction-set": return None, [ "selectedDirectionRef.snapshotPath must name " "instruction-set/selected-direction.json" ] task_root = path.parent.parent try: path = validate_task_artifact_path(path, task_root, "selected direction snapshot") loaded = load_owned_object_snapshot( path, artifact="selected direction snapshot" ) content = loaded.raw_bytes data = loaded.value except DirectionSelectionError as exc: return None, [str(exc)] except OSError as exc: return None, [f"selected direction snapshot is unreadable: {exc}"] except (UnicodeError, json.JSONDecodeError, JsonBoundaryError) as exc: return None, [f"selected direction snapshot JSON is invalid: {exc}"] if not isinstance(data, Mapping): return None, ["selected direction snapshot JSON must contain an object"] return SelectedDirectionSnapshot( data=data, path=path, relative_path=path.relative_to(task_root).as_posix(), sha256=hashlib.sha256(content).hexdigest(), ), [] def validate_selected_direction_ref( ref: Mapping[str, Any], snapshot: SelectedDirectionSnapshot ) -> list[str]: """Match a planning reference to one byte-verified direction snapshot.""" failures: list[str] = [] for field in ("sourceReport", "sourceDataSha256", "optionId"): if ref.get(field) != snapshot.data.get(field): failures.append( f"selectedDirectionRef.{field} is missing or does not match " "the selected-direction snapshot" ) if ref.get("snapshotPath") != snapshot.relative_path: # 기대값을 이름으로 댄다. `sourceReport` / `sourceDataSha256` / `optionId` # 는 스냅샷 파일에 그대로 실려 있어 저자가 옮겨 적으면 되지만, # `snapshotPath` 는 저자가 직접 만들어야 하고 그 기준점(태스크 루트)은 # 스키마에도 프롬프트에도 적혀 있지 않았다. 무엇과 다른지 말하지 않는 # 실패는 실행이 스스로 고칠 수 없다. failures.append( "selectedDirectionRef.snapshotPath is missing or does not match the " "actual selected-direction snapshot path; expected the task-root " f"relative {snapshot.relative_path!r}, got " f"{ref.get('snapshotPath')!r}" ) if ref.get("snapshotSha256") != snapshot.sha256: failures.append( "selectedDirectionRef.snapshotSha256 is missing or does not match " "the actual selected-direction snapshot bytes" ) return failures def _direction_realization_errors( realization: object, snapshot: SelectedDirectionSnapshot ) -> list[str]: if not isinstance(realization, Mapping): return ["directionRealization is missing from a plan-ready result"] direction = snapshot.data.get("direction") if not isinstance(direction, Mapping): return ["selected direction snapshot is missing its direction object"] comparisons = { "coreMechanism": direction.get("coreMechanism"), "architectureBoundaries": direction.get("architectureBoundaries"), "planningInvariants": snapshot.data.get("planningInvariants"), "userConstraints": snapshot.data.get("userConstraints"), } return [ f"directionRealization.{field} must exactly preserve the selected-direction snapshot" for field, expected in comparisons.items() if realization.get(field) != expected ] def _plan_reference_sets( planning: Mapping[str, Any], ) -> tuple[set[int], set[str], set[str], tuple[str, ...]]: stages = [row for row in planning.get("stages") or () if isinstance(row, Mapping)] stage_ids = { row["stage"] for row in stages if isinstance(row.get("stage"), int) } step_ids = { f"{stage['stage']}.{step['step']}" for stage in stages if isinstance(stage.get("stage"), int) for step in stage.get("stepwiseExecution") or () if isinstance(step, Mapping) and isinstance(step.get("step"), int) } validation_ids = { str(row.get("id")) for row in planning.get("validationChecklist") or () if isinstance(row, Mapping) and row.get("id") } realization = planning.get("directionRealization") file_rows = ( realization.get("fileStructure") or () if isinstance(realization, Mapping) else () ) file_paths = tuple( str(row.get("path")) for row in file_rows if isinstance(row, Mapping) and row.get("path") ) return stage_ids, step_ids, validation_ids, file_paths def _duplicate_values(values: list[object]) -> tuple[object, ...]: return tuple(value for value, count in Counter(values).items() if count > 1) def _normalized_plan_file_path(value: object) -> str: return posixpath.normpath(str(value).replace("\\", "/")) def _plan_structure_errors(planning: Mapping[str, Any]) -> list[str]: failures: list[str] = [] stage_map = [ row for row in planning.get("stageMap") or () if isinstance(row, Mapping) ] stages = [ row for row in planning.get("stages") or () if isinstance(row, Mapping) ] stage_map_numbers = [ row["stage"] for row in stage_map if isinstance(row.get("stage"), int) ] stage_numbers = [ row["stage"] for row in stages if isinstance(row.get("stage"), int) ] duplicate_stage_map = set(_duplicate_values(stage_map_numbers)) duplicate_stages = set(_duplicate_values(stage_numbers)) failures.extend( f"stageMap contains duplicate stage {number}" for number in sorted(duplicate_stage_map) ) failures.extend( f"stages contains duplicate stage {number}" for number in sorted(duplicate_stages) ) stage_map_set = set(stage_map_numbers) stage_set = set(stage_numbers) failures.extend( f"stageMap is missing stages stage {number}" for number in sorted(stage_set - stage_map_set) ) failures.extend( f"stages is missing stageMap stage {number}" for number in sorted(stage_map_set - stage_set) ) unique_stage_map = { row["stage"]: row for row in stage_map if isinstance(row.get("stage"), int) and row["stage"] not in duplicate_stage_map } unique_stages = { row["stage"]: row for row in stages if isinstance(row.get("stage"), int) and row["stage"] not in duplicate_stages } failures.extend( f"stageMap and stages stage {number} title must match" for number in sorted(unique_stage_map.keys() & unique_stages.keys()) if unique_stage_map[number].get("title") != unique_stages[number].get("title") ) for stage in stages: stage_number = stage.get("stage") step_numbers = [ step["step"] for step in stage.get("stepwiseExecution") or () if isinstance(step, Mapping) and isinstance(step.get("step"), int) ] failures.extend( f"stages stage {stage_number} step {step_number} is duplicate" for step_number in sorted(_duplicate_values(step_numbers)) ) validation_ids = [ str(row["id"]) for row in planning.get("validationChecklist") or () if isinstance(row, Mapping) and row.get("id") ] failures.extend( f"validationChecklist id {validation_id} is duplicate" for validation_id in sorted(_duplicate_values(validation_ids)) ) realization = planning.get("directionRealization") file_rows = ( [row for row in realization.get("fileStructure") or () if isinstance(row, Mapping)] if isinstance(realization, Mapping) else [] ) file_ids = [str(row["id"]) for row in file_rows if row.get("id")] failures.extend( f"directionRealization.fileStructure id {file_id} is duplicate" for file_id in sorted(_duplicate_values(file_ids)) ) normalized_paths = [ _normalized_plan_file_path(row["path"]) for row in file_rows if row.get("path") ] failures.extend( f"directionRealization.fileStructure path {path} is duplicate after normalization" for path in sorted(_duplicate_values(normalized_paths)) ) return failures def _coverage_reference_errors( rows: list[Mapping[str, Any]], stage_ids: set[int], step_ids: set[str], validation_ids: set[str], file_paths: tuple[str, ...], dependencies: list[Mapping[str, Any]], ) -> list[str]: failures: list[str] = [] reference_sets = { "stageRefs": stage_ids, "stepRefs": step_ids, "validationRefs": validation_ids, "fileRefs": set(file_paths), "crossProjectDependencyRefs": {row.get("id") for row in dependencies}, } for row in rows: requirement_id = str(row.get("originalRequirementId") or "") external_refs = row.get("crossProjectDependencyRefs") or () if row.get("status") == "externally-tracked": if not external_refs: failures.append( f"requirementCoverage {requirement_id} requires crossProjectDependencyRefs" ) for ref in external_refs: matches = [item for item in dependencies if item.get("id") == ref] if ( len(matches) != 1 or any( not isinstance(matches[0].get(field), str) or not matches[0][field].strip() for field in ( "project", "requiredWork", "verificationSignal", "linkedWork", "howToStart", ) ) or matches[0].get("direction") not in {"upstream-precondition", "downstream-carry"} ): failures.append( f"requirementCoverage {requirement_id} crossProjectDependencyRefs must resolve to one complete dependency: {ref}" ) elif external_refs: failures.append( f"requirementCoverage {requirement_id} external references require externally-tracked status" ) elif any( not row.get(field) for field in ("stageRefs", "stepRefs", "validationRefs", "fileRefs") ): failures.append( f"requirementCoverage {requirement_id} requires local stage, step, validation and file references" ) for field, valid_values in reference_sets.items(): dangling = [ value for value in row.get(field) or () if value not in valid_values ] if dangling: failures.append( f"requirementCoverage {requirement_id} {field} contains " f"dangling reference(s): {dangling}" ) return failures def _requirement_ids_for_scope( rows: list[Mapping[str, Any]], field: str, value: object ) -> tuple[str, ...]: return tuple( str(row.get("originalRequirementId")) for row in rows if value in (row.get(field) or ()) and row.get("originalRequirementId") ) def _coverage_summary_errors( planning: Mapping[str, Any], original_ids: tuple[str, ...] ) -> list[str]: rows = [ row for row in planning.get("requirementCoverage") or () if isinstance(row, Mapping) ] row_ids = tuple(str(row.get("originalRequirementId") or "") for row in rows) failures: list[str] = [] if not original_ids: return ["original requirement denominator from taskBriefPath is empty"] if row_ids != original_ids: failures.append( "original requirement rows must match the taskBriefPath requirement " "sequence exactly once" ) return failures structure_failures = _plan_structure_errors(planning) failures.extend(structure_failures) stage_ids, step_ids, validation_ids, file_paths = _plan_reference_sets(planning) failures.extend( _coverage_reference_errors( rows, stage_ids, step_ids, validation_ids, file_paths, [ row for row in planning.get("crossProjectDependencies") or () if isinstance(row, Mapping) ], ) ) if failures: return failures # 검증된 외부 작업 연결은 계획 범위에만 포함하며 실제 완료 상태는 바꾸지 않는다. statuses = { row_id: "covered" if row.get("status") == "externally-tracked" else str(row.get("status") or "") for row_id, row in zip(row_ids, rows) } stage_requirements = { stage: _requirement_ids_for_scope(rows, "stageRefs", stage) for stage in sorted(stage_ids) } file_requirements = { path: _requirement_ids_for_scope(rows, "fileRefs", path) for path in file_paths } try: result, unmapped_stages, unmapped_files = calculate_plan_exact_coverage( original_ids, statuses, stage_requirements, file_requirements ) except ExactCoverageError as exc: return [*failures, f"plan-ready requires exact 100% coverage: {exc}"] failures.extend( f"unmapped stage {stage} has no original requirement" for stage in unmapped_stages ) failures.extend( f"unmapped file change {path} has no original requirement" for path in unmapped_files ) failures.extend( _declared_coverage_summary_errors( planning.get("coverageSummary"), result, unmapped_stages, unmapped_files ) ) if result.verdict != "exact": failures.append("plan-ready requires exact 100% coverage") return failures def _declared_coverage_summary_errors( summary: object, result: Any, unmapped_stages: tuple[int, ...], unmapped_files: tuple[str, ...], ) -> list[str]: if not isinstance(summary, Mapping): return ["coverageSummary is missing from a plan-ready result"] expected = { "coveragePercent": result.coverage_percent, "scopePrecisionPercent": result.scope_precision_percent, "coverageVerdict": result.verdict, "unmappedStages": list(unmapped_stages), "unmappedFileChanges": list(unmapped_files), } return [ f"coverageSummary.{field} does not match recalculated plan coverage" for field, value in expected.items() if summary.get(field) != value ] # stage 를 게이트하는 검증 행이 implementation run 안에서 실행 가능함을 강제하는 # 마커. 실측 실패값(dev-10341 VC-013/VC-014)에서 왔다: 수동 워크스루·배포 전제 # 문구가 stageRefs 를 단 채 실리면, implementation run 은 배포 금지·자격증명 # 부재라 그 행을 영원히 통과시킬 수 없고 코드 결함 0건인 stage 가 차단된다. # 그런 검증은 브리프의 External Gates / final-verification 의 user-owned # external QA 소속이다. _MANUAL_VALIDATION_MARKERS = ( "by hand", "manually", "redeploy", "deployment", "preproduction", "signed in", ) def stage_validation_executability_errors( planning: Mapping[str, Any], ) -> list[str]: failures: list[str] = [] for row in planning.get("validationChecklist") or (): if not isinstance(row, Mapping): continue command = str(row.get("commandOrObservation") or "") denied = find_denied_tokens(command) if re.search(r"(?\s*/(?:tmp|private/tmp|var/tmp)/", command): denied.append( "output outside the worktree (use a worktree-local output path)" ) if denied: failures.append( f"validationChecklist {row.get('id')} conflicts with verifier command rules: " f"{'; '.join(denied)}; correct and approve the plan before implementation" ) if not row.get("stageRefs"): continue lowered = command.lower() matched = [ marker for marker in _MANUAL_VALIDATION_MARKERS if marker in lowered ] if matched: failures.append( f"validationChecklist {row.get('id')} gates stage(s) " f"{list(row.get('stageRefs') or ())} but is not executable by the " f"implementation run (marker(s): {matched}); move manual or " "deployed-environment verification to the brief's External Gates " "or final-verification's user-owned external QA" ) return failures def _is_config_only_path(path: str) -> bool: name = posixpath.basename(posixpath.normpath(str(path).replace("\\", "/"))) if name == "Dockerfile" or name.startswith("Dockerfile."): return True if name.startswith("docker-compose") or name.startswith(".env"): return True return name.endswith((".yml", ".yaml")) def _micro_stage_fold_errors(planning: Mapping[str, Any]) -> list[str]: """설정 파일 전용 종속 stage 는 독립 stage 가 아니다 — 커밋으로만 나눈다. 실측(dev-10341): 다른 stage 가 도입한 플래그를 Dockerfile/compose 2줄, workflow 2줄로 전파하는 작업이 stage 2·3 을 각각 차지해 run 2회(회당 30분+)와 검증 차단 1회를 낳았다. 값 전파는 도입 stage 의 스텝이고, 분리가 필요하면 커밋을 나눈다. 첫 stage(의존 없음)는 예외다 — 설정만 바꾸는 태스크 자체는 정당하다. """ dependent: dict[int, bool] = {} for row in planning.get("stageMap") or (): if isinstance(row, Mapping) and isinstance(row.get("stage"), int): raw = str(row.get("dependsOn") or "").strip().lower() dependent[row["stage"]] = raw not in ("", "(none)", "none") failures: list[str] = [] for stage in planning.get("stages") or (): if not isinstance(stage, Mapping): continue number = stage.get("stage") if not isinstance(number, int) or not dependent.get(number): continue paths = [ str(path) for step in stage.get("stepwiseExecution") or () if isinstance(step, Mapping) for path in step.get("plannedPaths") or () ] if paths and all(_is_config_only_path(path) for path in paths): failures.append( f"stage {number} is a config-only propagation stage " f"({sorted(set(paths))}); fold it into the stage that introduces " "the value and split the work by commits, not stages" ) return failures def _direction_invalidation_errors(planning: Mapping[str, Any]) -> list[str]: failures: list[str] = [] invalidation = planning.get("directionInvalidation") if not isinstance(invalidation, Mapping): failures.append("direction-invalidated requires directionInvalidation evidence") else: for field in ("reasons", "codeEvidence"): values = invalidation.get(field) if not isinstance(values, list) or not values or any( not isinstance(value, str) or not value.strip() for value in values ): failures.append(f"directionInvalidation.{field} must contain evidence") if planning.get("routing") != "implementation-option-selection": failures.append( "direction-invalidated routing must be implementation-option-selection" ) for field in _PLAN_EXECUTION_FIELDS: if field in planning: failures.append( f"direction-invalidated must not contain execution field {field}" ) return failures def validate_selected_direction_plan( data: Mapping[str, Any], brief_path: Path, snapshot_path: Path ) -> list[str]: """Validate selected-direction planning semantics against independent inputs.""" planning = data.get("implementationPlanning") if not isinstance(planning, Mapping): return ["implementationPlanning is missing"] if planning.get("planningContract") != "selected-direction": return ["implementationPlanning.planningContract must be selected-direction"] snapshot, failures = load_selected_direction_snapshot(snapshot_path) if snapshot is None: return failures header = data.get("header") plan_task_key = header.get("taskKey") if isinstance(header, Mapping) else None snapshot_task_key = snapshot.data.get("taskKey") if not isinstance(plan_task_key, str) or not plan_task_key: failures.append("plan header.taskKey must be a non-empty string") if not isinstance(snapshot_task_key, str) or not snapshot_task_key: failures.append("selected direction snapshot taskKey must be a non-empty string") elif isinstance(plan_task_key, str) and snapshot_task_key != plan_task_key: failures.append( "selected direction snapshot taskKey must exactly match plan header.taskKey" ) ref = planning.get("selectedDirectionRef") if not isinstance(ref, Mapping): failures.append("selectedDirectionRef is missing") else: failures.extend(validate_selected_direction_ref(ref, snapshot)) outcome = planning.get("outcome") if outcome == "direction-invalidated": failures.extend(_direction_invalidation_errors(planning)) return failures if outcome != "plan-ready": return [*failures, f"selected-direction outcome is invalid: {outcome!r}"] failures.extend( _direction_realization_errors(planning.get("directionRealization"), snapshot) ) failures.extend( _coverage_summary_errors( planning, brief_end_state_id_sequence(Path(brief_path)) ) ) failures.extend(stage_validation_executability_errors(planning)) failures.extend(_micro_stage_fold_errors(planning)) return failures