"""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 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 .scope_provenance import brief_end_state_id_sequence from .user_response import UserResponseError, parse_direction_selection _TASK_TYPE = "implementation-option-selection" _REPORT_RE = re.compile( r"^final-report-implementation-option-selection-(?P\d{3,})\.md$" ) _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 _sibling_data_path(report: Path) -> Path: data_path = _strict_regular_file(final_report_data_path(report), "data") if data_path.parent != report.parent: raise DirectionSelectionError("data file must be a sibling of the report") return data_path 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, 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.option_name, parsed.selection_note, constraints, ) def _preselected_selection( selection: Mapping[str, Any], ) -> tuple[str, 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, option_name: str ) -> tuple[Mapping[str, Any], str]: 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 or ( option_name and report_option_name != option_name ): raise DirectionSelectionError("selected candidate name 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 in the documented fail-closed order.""" report = _strict_regular_file(Path(report_path), "report") data_path = _sibling_data_path(report) task_root, seq = _selection_layout(report) report = validate_task_artifact_path(report, task_root, "report") data_path = validate_task_artifact_path(data_path, task_root, "data") 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, option_name, 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, option_name, note, constraints = _preselected_selection( selection ) else: raise DirectionSelectionError(f"unsupported selection mode: {mode}") option, normalized_option_name = _selected_option( selection, option_id, option_name ) _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: failures.append( "selectedDirectionRef.snapshotPath is missing or does not match the " "actual selected-direction snapshot path" ) 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, ...], ) -> list[str]: failures: list[str] = [] reference_sets = { "stageRefs": stage_ids, "stepRefs": step_ids, "validationRefs": validation_ids, "fileRefs": set(file_paths), } for row in rows: requirement_id = str(row.get("originalRequirementId") or "") 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 ) ) if structure_failures: return failures statuses = {row_id: 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 ] 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)) ) ) return failures