"""Selection-backed schedule effort, Work Breakdown, and Gantt validation.""" from __future__ import annotations import json import re from dataclasses import dataclass from decimal import Decimal from pathlib import Path from typing import Any from .md_table import is_separator_row, split_pipe_row from .json_boundary import external_user_json_source, load_external_json _ROOT_FIELDS = {"schemaVersion", "tasks"} _TASK_FIELDS = { "taskKey", "taskId", "state", "sourcePlanPath", "selectedStages", "doneStages", "stages", } _STAGE_FIELDS = {"stageNumber", "title", "dependsOn", "stepCount"} # `### . ` — the heading is a title, not an identifier. The # task-id is a machine key; printing it as the reader's heading makes them parse # `nestjs-migration-nlpvibe-to-nestjs-and-org-standard-structure` to learn the # work is a NestJS migration. Blocks bind to At a Glance by their number. _TASK_HEADING_RE = re.compile(r"^###\s+(\d+)\.\s+(\S.*?)\s*$", re.MULTILINE) _WORK_BREAKDOWN_HEADER = ["Stage", "Title", "Steps", "Depends On", "Days"] _WORK_BREAKDOWN_SEPARATOR = ["---:", "---", "---:", "---", "---:"] # No `Task ID` and no `taskType`: the schedule is shared with people who do not # run okstra, so a task is named by its work and a row binds to the selection by # its `#` position — the same rule the `### .` blocks follow. _AT_A_GLANCE_HEADER = [ "#", "작업", "Category", "Priority", "Effort", "Days", "Risk", ] _EFFORT_HEADER = ["Size", "Criteria", "Day(s)"] _EFFORT_SIZES = {"S", "M", "L", "XL", "XXL"} _DAY_NUMBER = r"\d+(?:\.\d+)?" # `(est)` marks a range the schedule derived from `step_count` rather than one # the plan stated. implementation-planning does not estimate duration, so every # range carries it today; the marker is what keeps that visible in the table # instead of in a footnote. _DAY_RANGE_RE = re.compile( rf"^\s*({_DAY_NUMBER})\s*(?:~|-)\s*({_DAY_NUMBER})\s*(?:\(est\))?\s*$" ) _EFFORT_TOTAL_RE = re.compile( rf"\*\*\d+\s+tasks\s+total\s*/\s*estimated\s+effort:\s*" rf"({_DAY_NUMBER})\s*~\s*({_DAY_NUMBER})\s+days?\s*" r"\(Effort\s+sum\)\*\*" ) _PLAIN_FENCE_RE = re.compile( r"^```[ \t]*$\n(.*?)^```[ \t]*$", re.MULTILINE | re.DOTALL ) # A row is `[ ]Stage ` and nothing else. The bar's width IS # the duration — a trailing `days=` annotation restated the Work Breakdown's Days # column, and per-row `! crit` / `est` markers that were identical on every row # carried no information at all. _GANTT_ROW_RE = re.compile( r"^\s*(?:(\S+)\s+)?Stage (\d+)\s+([█]*[░]*)\s*$" ) _GANTT_STAGE_CANDIDATE_RE = re.compile(r"^\s*(?:(\S+)\s+)?Stage (\d+)(?:\s|$)") _GANTT_AXIS_RE = re.compile(r"^\s*Day:\s*(.+)$") _XXL_DAY_RANGE_RE = re.compile(rf"^\s*{_DAY_NUMBER}\s*-\s*$") _TASK_SUBSECTION_PREFIXES = ( "**Problem**:", "**Solution**:", "**Work Breakdown**:", "**Verification Commands**:", "**Rollback**:", ) HALF_DAY = Decimal("0.5") # One column is half a day. A whole-day column cannot draw a 2.5-day stage, which # is how bars silently rounded away from the Work Breakdown they annotate. GANTT_COLUMN_DAYS = HALF_DAY # The axis may overshoot the work by less than one tick interval; more than that # is dead space that misreads as schedule length. _GANTT_AXIS_TICK_DAYS = Decimal("5") class ScheduleSemanticError(ValueError): pass def _round_half_day(value: Decimal) -> Decimal: return (value / HALF_DAY).quantize(Decimal("1")) * HALF_DAY def _half_day_units(value: Decimal) -> int: units = value / HALF_DAY if units != units.to_integral_value(): raise ScheduleSemanticError( "effort: stage totals must use 0.5-day increments" ) return int(units) def _validate_stage_day_allocation( total: tuple[Decimal, Decimal], step_counts: tuple[int, ...], allocated: list[tuple[Decimal, Decimal]], ) -> None: invalid_range = any( lower < 0 or upper < 0 or lower > upper for lower, upper in allocated ) lower_total = sum((lower for lower, _ in allocated), Decimal("0")) upper_total = sum((upper for _, upper in allocated), Decimal("0")) if invalid_range or (lower_total, upper_total) != total: raise ScheduleSemanticError( "effort: cannot represent proportional half-day allocation " f"for step counts {step_counts} within {_format_day_range(total)}" ) for lower, upper in allocated: _half_day_units(lower) _half_day_units(upper) def allocate_stage_days( total: tuple[Decimal, Decimal], step_counts: tuple[int, ...], ) -> tuple[tuple[Decimal, Decimal], ...]: if not step_counts or any(count <= 0 for count in step_counts): raise ValueError("selected stage step counts must be positive") if total[0] < 0 or total[0] > total[1]: raise ScheduleSemanticError( "effort: stage total must satisfy 0 <= lower <= upper" ) _half_day_units(total[0]) _half_day_units(total[1]) total_steps = Decimal(sum(step_counts)) allocated: list[tuple[Decimal, Decimal]] = [] used_lower = Decimal("0") used_upper = Decimal("0") for count in step_counts[:-1]: ratio = Decimal(count) / total_steps current = ( _round_half_day(total[0] * ratio), _round_half_day(total[1] * ratio), ) allocated.append(current) used_lower += current[0] used_upper += current[1] allocated.append((total[0] - used_lower, total[1] - used_upper)) _validate_stage_day_allocation(total, step_counts, allocated) return tuple(allocated) @dataclass(frozen=True) class SelectionStage: stage_number: int title: str depends_on: tuple[int, ...] step_count: int @dataclass(frozen=True) class SelectionTask: task_key: str task_id: str state: str source_plan_path: str selected_stages: tuple[int, ...] done_stages: tuple[int, ...] stages: tuple[SelectionStage, ...] @dataclass(frozen=True) class _BreakdownRow: task_id: str stage_number: int title: str step_count: int depends_on: tuple[tuple[int, bool], ...] days: tuple[Decimal, Decimal] | None @dataclass(frozen=True) class _AtAGlanceRow: task_id: str effort: str days_text: str days: tuple[Decimal, Decimal] | None @dataclass(frozen=True) class _GanttRow: task_id: str stage_number: int filled_cells: int open_cells: int class StageMapSelectionError(ValueError): pass def _require_object(value: Any, context: str) -> dict[str, Any]: if not isinstance(value, dict): raise StageMapSelectionError(f"{context} must be an object") return value def _require_fields( value: dict[str, Any], expected: set[str], context: str, ) -> None: if set(value) == expected: return missing = sorted(expected - set(value)) extra = sorted(set(value) - expected) raise StageMapSelectionError( f"{context} fields mismatch: missing={missing}, extra={extra}" ) def _require_string( value: dict[str, Any], field: str, context: str, ) -> str: result = value.get(field) if not isinstance(result, str): raise StageMapSelectionError(f"{context}.{field} must be a string") return result def _require_positive_int(value: Any, context: str) -> int: if isinstance(value, bool) or not isinstance(value, int) or value < 1: raise StageMapSelectionError(f"{context} must be a positive integer") return value def _require_int_list( value: dict[str, Any], field: str, context: str, ) -> tuple[int, ...]: items = value.get(field) if not isinstance(items, list): raise StageMapSelectionError(f"{context}.{field} must be an array") result = tuple( _require_positive_int(item, f"{context}.{field}[{index}]") for index, item in enumerate(items) ) if len(result) != len(set(result)): raise StageMapSelectionError(f"{context}.{field} must be unique") return result def _parse_stage(value: Any, context: str) -> SelectionStage: record = _require_object(value, context) _require_fields(record, _STAGE_FIELDS, context) title = _require_string(record, "title", context) if not title: raise StageMapSelectionError(f"{context}.title must not be empty") return SelectionStage( stage_number=_require_positive_int( record.get("stageNumber"), f"{context}.stageNumber" ), title=title, depends_on=_require_int_list(record, "dependsOn", context), step_count=_require_positive_int( record.get("stepCount"), f"{context}.stepCount" ), ) def _parse_stages(value: Any, context: str) -> tuple[SelectionStage, ...]: if not isinstance(value, list): raise StageMapSelectionError(f"{context}.stages must be an array") stages = tuple( _parse_stage(stage, f"{context}.stages[{index}]") for index, stage in enumerate(value) ) stage_numbers = [stage.stage_number for stage in stages] if len(stage_numbers) != len(set(stage_numbers)): raise StageMapSelectionError(f"{context}: duplicate stageNumber") return stages def _validate_stage_graph( task_id: str, stages_by_number: dict[int, SelectionStage], ) -> None: for stage in stages_by_number.values(): for dependency in stage.depends_on: if dependency not in stages_by_number: raise StageMapSelectionError( f"task {task_id} stage {stage.stage_number}: " f"dependency {dependency} is absent from stages" ) completed: set[int] = set() current_path: set[int] = set() path: list[int] = [] def visit(stage_number: int) -> None: if stage_number in completed: return if stage_number in current_path: cycle_start = path.index(stage_number) cycle = path[cycle_start:] + [stage_number] rendered = " -> ".join(f"S{item}" for item in cycle) raise ScheduleSemanticError( f"task {task_id}: stage dependency cycle detected: {rendered}" ) current_path.add(stage_number) path.append(stage_number) for dependency in stages_by_number[stage_number].depends_on: visit(dependency) path.pop() current_path.remove(stage_number) completed.add(stage_number) for stage_number in stages_by_number: visit(stage_number) def _validate_task_selection(task: SelectionTask) -> None: selected = set(task.selected_stages) done = set(task.done_stages) overlap = sorted(selected & done) if overlap: raise StageMapSelectionError( f"task {task.task_id}: selected/done stage overlap {overlap}" ) stages_by_number = {stage.stage_number: stage for stage in task.stages} for stage_number in task.selected_stages: if stage_number not in stages_by_number: raise StageMapSelectionError( f"task {task.task_id}: selected stage {stage_number} is absent" ) _validate_stage_graph(task.task_id, stages_by_number) allowed = selected | done pending = list(task.selected_stages) visited: set[int] = set() while pending: stage_number = pending.pop() if stage_number in visited: continue visited.add(stage_number) for dependency in stages_by_number[stage_number].depends_on: if dependency not in allowed: raise StageMapSelectionError( f"task {task.task_id} stage {stage_number}: " f"dependency {dependency} is outside doneStages or selectedStages" ) pending.append(dependency) def _validate_task_state(task: SelectionTask) -> None: if task.state != "missing": return if any(( task.source_plan_path, task.selected_stages, task.done_stages, task.stages, )): raise StageMapSelectionError( f"task {task.task_id}: missing task must have empty sourcePlanPath, " "selectedStages, doneStages, and stages" ) def _parse_task(value: Any, index: int) -> SelectionTask: context = f"tasks[{index}]" record = _require_object(value, context) _require_fields(record, _TASK_FIELDS, context) state = _require_string(record, "state", context) if state not in {"ready", "missing"}: raise StageMapSelectionError( f"{context}.state must be 'ready' or 'missing'" ) task = SelectionTask( task_key=_require_string(record, "taskKey", context), task_id=_require_string(record, "taskId", context), state=state, source_plan_path=_require_string(record, "sourcePlanPath", context), selected_stages=_require_int_list(record, "selectedStages", context), done_stages=_require_int_list(record, "doneStages", context), stages=_parse_stages(record.get("stages"), context), ) if not task.task_key or not task.task_id: raise StageMapSelectionError(f"{context} taskKey/taskId must not be empty") _validate_task_state(task) _validate_task_selection(task) return task def _reject_duplicate_task_identities(tasks: tuple[SelectionTask, ...]) -> None: for field in ("task_key", "task_id"): values = [getattr(task, field) for task in tasks] duplicates = {value for value in values if values.count(value) > 1} if duplicates: label = "taskKey" if field == "task_key" else "taskId" raise StageMapSelectionError( f"duplicate {label}: {sorted(duplicates)[0]}" ) def _strict_json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise StageMapSelectionError(f"duplicate JSON key {key!r}") result[key] = value return result def load_schedule_selection(path: Path) -> tuple[SelectionTask, ...]: # 외부 입력: okstra-schedule 사용자가 선택한 작업 집합 문서다. payload = load_external_json( external_user_json_source(path), artifact="schedule selection", object_pairs_hook=_strict_json_object, ) root = _require_object(payload, "selection") _require_fields(root, _ROOT_FIELDS, "selection") version = root.get("schemaVersion") if type(version) is not int or version != 1: raise StageMapSelectionError("selection.schemaVersion must be exactly 1") values = root.get("tasks") if not isinstance(values, list): raise StageMapSelectionError("selection.tasks must be an array") tasks = tuple(_parse_task(value, index) for index, value in enumerate(values)) _reject_duplicate_task_identities(tasks) return tasks def _parse_dependencies( value: str, task_id: str, stage_number: int, ) -> tuple[tuple[tuple[int, bool], ...], str | None]: if value == "None": return (), None dependencies: list[tuple[int, bool]] = [] for token in value.split(","): normalized = token.strip() match = re.fullmatch(r"Stage (\d+)( \(done\))?", normalized) if match is None: return (), ( f"work breakdown: stage {task_id}/S{stage_number} has invalid " f"Depends On value {value!r}" ) dependencies.append((int(match.group(1)), match.group(2) is not None)) return tuple(dependencies), None def _mask_fenced_lines(lines: list[str]) -> list[str | None]: masked: list[str | None] = [] fence_marker = "" for line in lines: stripped = line.lstrip() marker = stripped[:3] if stripped.startswith(("```", "~~~")) else "" if fence_marker: masked.append(None) if marker == fence_marker: fence_marker = "" elif marker: fence_marker = marker masked.append(None) else: masked.append(line) return masked def _mask_fenced_text(text: str) -> str: lines = text.splitlines(keepends=True) visible_lines = _mask_fenced_lines(lines) return "".join( line if visible is not None else re.sub(r"[^\r\n]", " ", line) for line, visible in zip(lines, visible_lines) ) def _section_bounds( text: str, heading: str, max_level: int, ) -> tuple[int, int] | None: visible_text = _mask_fenced_text(text) match = re.search( rf"^{re.escape(heading)}[ \t]*$", visible_text, re.MULTILINE ) if match is None: return None body_start = match.end() body = visible_text[body_start:] next_heading = re.search( rf"^#{{1,{max_level}}}\s", body, re.MULTILINE ) body_end = ( body_start + next_heading.start() if next_heading else len(text) ) return body_start, body_end def _section_body(text: str, heading: str, max_level: int) -> str: bounds = _section_bounds(text, heading, max_level) if bounds is None: return "" return _mask_fenced_text(text[bounds[0]:bounds[1]]) def _raw_section_body(text: str, heading: str, max_level: int) -> str | None: bounds = _section_bounds(text, heading, max_level) if bounds is None: return None return text[bounds[0]:bounds[1]] def _ordered_day_range( lower: Decimal, upper: Decimal, source: str, ) -> tuple[Decimal, Decimal]: if lower > upper: raise ScheduleSemanticError( f"{source}: lower must not exceed upper" ) return lower, upper def _parse_day_range( value: str, source: str, ) -> tuple[Decimal, Decimal] | None: match = _DAY_RANGE_RE.fullmatch(value) if match is None: return None return _ordered_day_range( Decimal(match.group(1)), Decimal(match.group(2)), source ) def parse_effort_ranges( text: str, ) -> dict[str, tuple[Decimal, Decimal] | None]: section = _section_body(text, "### Effort Sizing Criteria", 3) lines = section.splitlines() header_indexes = [ index for index, line in enumerate(lines) if split_pipe_row(line) == _EFFORT_HEADER ] if len(header_indexes) != 1: raise ScheduleSemanticError( "Effort Sizing Criteria requires exactly one canonical table" ) header_index = header_indexes[0] separator_index = header_index + 1 if ( separator_index >= len(lines) or not is_separator_row(lines[separator_index]) or len(split_pipe_row(lines[separator_index])) != len(_EFFORT_HEADER) ): raise ScheduleSemanticError( "Effort Sizing Criteria requires a canonical separator row" ) ranges: dict[str, tuple[Decimal, Decimal] | None] = {} for line in lines[header_index + 2:]: if not line.strip().startswith("|"): break cells = split_pipe_row(line) size = cells[0].strip().strip("*") if cells else "" if len(cells) != 3 or size not in _EFFORT_SIZES: raise ScheduleSemanticError( f"Effort Sizing Criteria has malformed row {line!r}" ) if size in ranges: raise ScheduleSemanticError( f"Effort Sizing Criteria has duplicate {size} row" ) day_range = _parse_day_range( cells[2], f"Effort Sizing Criteria {size}" ) if size == "XXL": if day_range is not None or _XXL_DAY_RANGE_RE.fullmatch(cells[2]) is None: raise ScheduleSemanticError( "Effort Sizing Criteria XXL has invalid Day(s)" ) ranges[size] = None continue if day_range is None: raise ScheduleSemanticError( f"Effort Sizing Criteria {size} has invalid Day(s)" ) ranges[size] = day_range missing = sorted(_EFFORT_SIZES - set(ranges)) if missing: raise ScheduleSemanticError( "Effort Sizing Criteria is missing size row(s): " + ", ".join(missing) ) return ranges def _parse_at_a_glance_table( lines: list[str], header_index: int, ) -> tuple[list[_AtAGlanceRow], list[str]]: rows: list[_AtAGlanceRow] = [] violations: list[str] = [] for line in lines[header_index + 2:]: if not line.strip().startswith("|"): break cells = split_pipe_row(line) if cells == _AT_A_GLANCE_HEADER: break if len(cells) != len(_AT_A_GLANCE_HEADER): task_id = cells[1] if len(cells) > 1 else "" violations.append( f"At a Glance: noncanonical row for task {task_id} " "requires 7 columns" ) continue effort_tokens = cells[4].split() position = cells[0].strip() rows.append(_AtAGlanceRow( task_id=f"#{position}", effort=effort_tokens[0].strip("*") if effort_tokens else "", days_text=cells[5], days=_parse_day_range(cells[5], f"At a Glance row {position}"), )) return rows, violations def _parse_at_a_glance_rows( text: str, ) -> tuple[list[_AtAGlanceRow], list[str]]: section = _section_body(text, "## At a Glance", 2) lines = section.splitlines() header_indexes = [ index for index, line in enumerate(lines) if split_pipe_row(line) == _AT_A_GLANCE_HEADER ] if not header_indexes: return [], ["At a Glance: missing canonical 7-column table"] rows: list[_AtAGlanceRow] = [] violations: list[str] = [] for header_index in header_indexes: table_rows, table_violations = _parse_at_a_glance_table( lines, header_index ) rows.extend(table_rows) violations.extend(table_violations) return rows, violations def _validated_at_a_glance_rows( text: str, tasks: tuple[SelectionTask, ...], ) -> tuple[dict[str, _AtAGlanceRow], list[str]]: """Rows keyed by task-id, resolved from each row's `#` position. The table names tasks by their work, so position is what ties a row to the selection — the n-th row is the n-th task, the same rule the `### .` blocks follow. """ rows, violations = _parse_at_a_glance_rows(text) seen: dict[str, list[_AtAGlanceRow]] = {} for row in rows: seen.setdefault(row.task_id, []).append(row) validated: dict[str, _AtAGlanceRow] = {} for position, task in enumerate(tasks, start=1): matches = seen.pop(f"#{position}", []) if len(matches) != 1: violations.append( f"At a Glance: task {task.task_id} requires exactly one row " f"at position {position}" ) continue row = matches[0] validated[task.task_id] = _AtAGlanceRow( task_id=task.task_id, effort=row.effort, days_text=row.days_text, days=row.days, ) for leftover, matches in seen.items(): violations.append( f"At a Glance: row {leftover} has no matching task " f"({len(tasks)} tasks selected)" ) return validated, violations def _format_day_range(day_range: tuple[Decimal, Decimal]) -> str: return f"{day_range[0]:.1f} ~ {day_range[1]:.1f}" def _sum_day_ranges( ranges: list[tuple[Decimal, Decimal]], ) -> tuple[Decimal, Decimal]: return ( sum((day_range[0] for day_range in ranges), Decimal("0")), sum((day_range[1] for day_range in ranges), Decimal("0")), ) def _validate_at_a_glance_effort( effort_ranges: dict[str, tuple[Decimal, Decimal] | None], tasks: tuple[SelectionTask, ...], glance_rows: dict[str, _AtAGlanceRow], ) -> list[str]: missing_task_ids = { task.task_id for task in tasks if task.state == "missing" } scheduled_task_ids = { task.task_id for task in tasks if task.selected_stages } violations: list[str] = [] for row in glance_rows.values(): if row.task_id in missing_task_ids: if row.days_text != "[NEEDS-PLANNING]": violations.append( f"effort: {row.task_id} has no planning report and " "requires [NEEDS-PLANNING]" ) continue if row.task_id not in scheduled_task_ids: continue if row.days is None: violations.append( f"effort: {row.task_id} requires a day range in Days — " f"got {row.days_text!r}" ) continue if row.effort not in effort_ranges: violations.append( f"effort: {row.task_id} effort {row.effort} is absent from " "Effort Sizing Criteria" ) continue expected = effort_ranges[row.effort] if expected is None or row.days == expected: continue violations.append( f"effort: {row.task_id} effort {row.effort} requires " f"{_format_day_range(expected)} days" ) return violations def _work_breakdown_lines(block: str) -> list[str | None] | None: lines = _mask_fenced_lines(block.splitlines()) label_index = next( ( index for index, line in enumerate(lines) if line == "**Work Breakdown**:" ), None, ) if label_index is None: return None end = next( ( index for index in range(label_index + 1, len(lines)) if lines[index] is not None and lines[index].startswith(_TASK_SUBSECTION_PREFIXES) ), len(lines), ) return lines[label_index + 1:end] def _parse_breakdown_table( task_id: str, block: str, ) -> tuple[list[_BreakdownRow], list[str]]: lines = _work_breakdown_lines(block) if lines is None: return [], [] violations: list[str] = [] header_indices = [ index for index, line in enumerate(lines) if line is not None and split_pipe_row(line) == _WORK_BREAKDOWN_HEADER ] if len(header_indices) != 1: violations.append( f"work breakdown: task {task_id} must contain exactly one canonical header" ) return [], violations header_index = header_indices[0] if ( header_index + 1 >= len(lines) or lines[header_index + 1] is None or split_pipe_row(lines[header_index + 1]) != _WORK_BREAKDOWN_SEPARATOR ): violations.append( f"work breakdown: task {task_id} must use the canonical separator" ) return [], violations rows: list[_BreakdownRow] = [] for line in lines[header_index + 2:]: if line is None or not line.strip().startswith("|"): break cells = split_pipe_row(line) if len(cells) != 5 or not cells[0].isdigit() or not cells[2].isdigit(): violations.append( f"work breakdown: malformed stage row for task {task_id}: {line!r}" ) continue stage_number = int(cells[0]) depends_on, error = _parse_dependencies(cells[3], task_id, stage_number) if error: violations.append(error) rows.append(_BreakdownRow( task_id=task_id, stage_number=stage_number, title=cells[1], step_count=int(cells[2]), depends_on=depends_on, days=_parse_day_range( cells[4], f"work breakdown {task_id}/S{stage_number}" ), )) return rows, violations def _duplicate_heading_violations( headings: list[re.Match[str]], ) -> list[str]: violations: list[str] = [] values = [heading.group(1) for heading in headings] duplicates = sorted({value for value in values if values.count(value) > 1}) for value in duplicates: violations.append( f"work breakdown: duplicate task-index heading {value}" ) return violations def _task_section_blocks( text: str, ordered_task_ids: list[str], ) -> tuple[list[tuple[str, str]], list[str]]: """Bind each `### .` block to the n-th At a Glance row. The heading carries a human title, so the task it belongs to comes from its position rather than from an identifier printed at the reader. """ visible_text = _mask_fenced_text(text) headings = list(_TASK_HEADING_RE.finditer(visible_text)) blocks: list[tuple[str, str]] = [] for index, heading in enumerate(headings): end = ( headings[index + 1].start() if index + 1 < len(headings) else len(visible_text) ) next_section = re.search( r"^##\s", visible_text[heading.end():end], re.MULTILINE ) if next_section is not None: end = heading.end() + next_section.start() position = int(heading.group(1)) task_id = ( ordered_task_ids[position - 1] if 1 <= position <= len(ordered_task_ids) else f"" ) blocks.append((task_id, visible_text[heading.end():end])) return blocks, _duplicate_heading_violations(headings) def _validate_nonforward_task_section( task: SelectionTask, block: str, ) -> list[str]: if task.selected_stages: return [] marker = ( "[NEEDS-PLANNING]" if task.state == "missing" else "_Complete — no remaining stage_" ) marker_count = sum(line.strip() == marker for line in block.splitlines()) violations: list[str] = [] if marker_count != 1: violations.append( f"task section: task {task.task_id} requires exact marker {marker}" ) if "**Work Breakdown**:" in block: violations.append( f"task section: task {task.task_id} with no selected stages " "must not contain Work Breakdown" ) return violations def _validate_task_section_coverage( blocks: list[tuple[str, str]], tasks: tuple[SelectionTask, ...], ) -> list[str]: indexed: dict[str, list[str]] = {} for task_id, block in blocks: indexed.setdefault(task_id, []).append(block) known = {task.task_id for task in tasks} violations: list[str] = [] for task in tasks: matches = indexed.get(task.task_id, []) if len(matches) != 1: violations.append( f"task section: task {task.task_id} requires exactly one task section" ) continue violations.extend(_validate_nonforward_task_section(task, matches[0])) for task_id in sorted(set(indexed) - known): violations.append(f"task section: unknown task {task_id}") return violations def _extract_breakdown_rows( text: str, tasks: tuple[SelectionTask, ...], ) -> tuple[list[_BreakdownRow], list[str], list[tuple[str, str]]]: # The n-th `### .` block is the n-th selected task; neither the heading # nor the At a Glance row prints an id to match on. blocks, violations = _task_section_blocks( text, [task.task_id for task in tasks] ) if violations: return [], violations, blocks rows: list[_BreakdownRow] = [] for task_id, block in blocks: parsed, errors = _parse_breakdown_table( task_id, block ) rows.extend(parsed) violations.extend(errors) return rows, violations, blocks def _validate_row_fields( row: _BreakdownRow, task: SelectionTask, stage: SelectionStage, ) -> list[str]: prefix = f"work breakdown: stage {task.task_id}/S{stage.stage_number}" violations: list[str] = [] if row.title != stage.title: violations.append( f"{prefix} Title {row.title!r} does not match {stage.title!r}" ) if row.step_count != stage.step_count: violations.append( f"{prefix} Steps {row.step_count} does not match {stage.step_count}" ) done = set(task.done_stages) expected_dependencies: list[tuple[int, bool]] = [] for dependency in stage.depends_on: expected_dependencies.append((dependency, dependency in done)) expected = tuple(expected_dependencies) if row.depends_on != expected: violations.append( f"{prefix} Depends On {row.depends_on!r} does not match {expected!r}" ) return violations def _validate_task_rows( task: SelectionTask, indexed: dict[tuple[str, int], list[tuple[int, _BreakdownRow]]], ) -> list[str]: violations: list[str] = [] stages = {stage.stage_number: stage for stage in task.stages} selected = set(task.selected_stages) done = set(task.done_stages) for stage_number in task.selected_stages: entries = indexed.get((task.task_id, stage_number), []) if not entries: violations.append( f"work breakdown: missing selected stage {task.task_id}/S{stage_number}" ) elif len(entries) > 1: violations.append( f"work breakdown: duplicate stage {task.task_id}/S{stage_number}" ) for _, row in entries: violations.extend(_validate_row_fields(row, task, stages[stage_number])) for stage_number in done: if indexed.get((task.task_id, stage_number)): violations.append( f"work breakdown: done stage {task.task_id}/S{stage_number} must be excluded" ) task_rows = { stage_number for task_id, stage_number in indexed if task_id == task.task_id } for stage_number in sorted(task_rows - selected - done): violations.append( f"work breakdown: stage {task.task_id}/S{stage_number} is not selected" ) return violations def _validate_extra_rows( tasks: tuple[SelectionTask, ...], indexed: dict[tuple[str, int], list[tuple[int, _BreakdownRow]]], ) -> list[str]: known_task_ids = {task.task_id for task in tasks} violations: list[str] = [] for task_id, stage_number in sorted(indexed): if task_id not in known_task_ids: violations.append( f"work breakdown: stage {task_id}/S{stage_number} is not selected" ) return violations def _validate_topological_order( tasks: tuple[SelectionTask, ...], indexed: dict[tuple[str, int], list[tuple[int, _BreakdownRow]]], ) -> list[str]: violations: list[str] = [] for task in tasks: selected = set(task.selected_stages) stages = {stage.stage_number: stage for stage in task.stages} for stage_number in task.selected_stages: positions = indexed.get((task.task_id, stage_number), []) if not positions: continue position = positions[0][0] for dependency in stages[stage_number].depends_on: dependency_rows = indexed.get((task.task_id, dependency), []) if dependency in selected and dependency_rows: if dependency_rows[0][0] >= position: violations.append( f"work breakdown: dependency {task.task_id}/S{dependency} " f"must precede {task.task_id}/S{stage_number}" ) return violations def _validate_work_breakdowns( text: str, tasks: tuple[SelectionTask, ...], ) -> tuple[list[_BreakdownRow], list[str]]: rows, violations, blocks = _extract_breakdown_rows(text, tasks) violations.extend(_validate_task_section_coverage(blocks, tasks)) indexed: dict[tuple[str, int], list[tuple[int, _BreakdownRow]]] = {} for position, row in enumerate(rows): indexed.setdefault((row.task_id, row.stage_number), []).append( (position, row) ) for task in tasks: violations.extend(_validate_task_rows(task, indexed)) violations.extend(_validate_extra_rows(tasks, indexed)) violations.extend(_validate_topological_order(tasks, indexed)) return rows, violations def _expected_selected_stage_days( effort_ranges: dict[str, tuple[Decimal, Decimal] | None], tasks: tuple[SelectionTask, ...], glance_rows: dict[str, _AtAGlanceRow], ) -> dict[tuple[str, int], tuple[Decimal, Decimal]]: expected: dict[tuple[str, int], tuple[Decimal, Decimal]] = {} for task in tasks: glance_row = glance_rows.get(task.task_id) if glance_row is None or not task.selected_stages: continue # XXL has no finite upper bound in the sizing table, so an XXL task is # sized by its own decomposition: the row's own range is the total the # stages divide up. total = effort_ranges.get(glance_row.effort) or glance_row.days if total is None: continue stages = {stage.stage_number: stage for stage in task.stages} step_counts = tuple( stages[number].step_count for number in task.selected_stages ) allocated = allocate_stage_days(total, step_counts) expected.update(zip( ((task.task_id, number) for number in task.selected_stages), allocated, )) return expected def _validate_stage_effort( expected: dict[tuple[str, int], tuple[Decimal, Decimal]], rows: list[_BreakdownRow], ) -> list[str]: violations: list[str] = [] for row in rows: day_range = expected.get((row.task_id, row.stage_number)) if day_range is None or row.days == day_range: continue violations.append( f"effort: {row.task_id}/S{row.stage_number} days require " f"{_format_day_range(day_range)}" ) return violations def _validate_task_stage_sums( tasks: tuple[SelectionTask, ...], rows: list[_BreakdownRow], expected: dict[tuple[str, int], tuple[Decimal, Decimal]], ) -> list[str]: indexed: dict[tuple[str, int], list[_BreakdownRow]] = {} for row in rows: indexed.setdefault((row.task_id, row.stage_number), []).append(row) violations: list[str] = [] for task in tasks: keys = [(task.task_id, number) for number in task.selected_stages] if not keys or any(key not in expected for key in keys): continue entries = [indexed.get(key, []) for key in keys] if any(len(items) != 1 or items[0].days is None for items in entries): continue actual = _sum_day_ranges([items[0].days for items in entries]) required = _sum_day_ranges([expected[key] for key in keys]) if actual != required: violations.append( f"effort: {task.task_id} stage days sum " f"{_format_day_range(actual)} requires " f"{_format_day_range(required)}" ) return violations def _parse_effort_total(text: str) -> tuple[Decimal, Decimal] | None: match = _EFFORT_TOTAL_RE.search(_section_body(text, "## At a Glance", 2)) if match is None: return None return _ordered_day_range( Decimal(match.group(1)), Decimal(match.group(2)), "Effort sum" ) def _scheduled_task_ranges( effort_ranges: dict[str, tuple[Decimal, Decimal] | None], tasks: tuple[SelectionTask, ...], glance_rows: dict[str, _AtAGlanceRow], ) -> list[tuple[Decimal, Decimal]]: """Every in-scope task's day range, including XXL. An XXL row has no finite range in the sizing table; it carries its own, summed from its stages. Dropping those rows is what produced a `0.0 ~ 0.0` Effort sum on a schedule whose stages added up to weeks of work. """ ranges: list[tuple[Decimal, Decimal]] = [] for task in tasks: if task.state == "missing" or not task.selected_stages: continue glance_row = glance_rows.get(task.task_id) if glance_row is None: continue day_range = effort_ranges.get(glance_row.effort) or glance_row.days if day_range is not None: ranges.append(day_range) return ranges def _validate_effort_total( text: str, effort_ranges: dict[str, tuple[Decimal, Decimal] | None], tasks: tuple[SelectionTask, ...], glance_rows: dict[str, _AtAGlanceRow], ) -> list[str]: actual = _parse_effort_total(text) if actual is None: return ["effort: missing Effort sum"] required = _sum_day_ranges(_scheduled_task_ranges( effort_ranges, tasks, glance_rows )) if actual == required: return [] return [ f"effort: Effort sum requires {_format_day_range(required)} days" ] def _validate_effort_semantics( text: str, tasks: tuple[SelectionTask, ...], breakdown_rows: list[_BreakdownRow], ) -> list[str]: effort_ranges = parse_effort_ranges(text) glance_rows, violations = _validated_at_a_glance_rows(text, tasks) violations.extend(_validate_at_a_glance_effort( effort_ranges, tasks, glance_rows )) expected = _expected_selected_stage_days( effort_ranges, tasks, glance_rows ) violations.extend(_validate_stage_effort(expected, breakdown_rows)) violations.extend(_validate_task_stage_sums( tasks, breakdown_rows, expected )) violations.extend(_validate_effort_total( text, effort_ranges, tasks, glance_rows )) return violations def _sole_scheduled_task_id(tasks: tuple[SelectionTask, ...]) -> str: """The one task an unqualified `S` row can only mean; '' when ambiguous.""" scheduled = [task.task_id for task in tasks if task.selected_stages] return scheduled[0] if len(scheduled) == 1 else "" def _parse_gantt_rows( text: str, tasks: tuple[SelectionTask, ...], ) -> tuple[list[_GanttRow] | None, list[str]]: visible_text = _mask_fenced_text(text) headings = list(re.finditer( r"^## Gantt Chart[ \t]*$", visible_text, re.MULTILINE )) if not headings: return None, [] if len(headings) > 1: return None, [ "Gantt: Gantt Chart section must appear at most once" ] section = _raw_section_body(text, "## Gantt Chart", 2) if section is None: return None, [] sole_task_id = _sole_scheduled_task_id(tasks) rows: list[_GanttRow] = [] violations: list[str] = [] for fence in _PLAIN_FENCE_RE.finditer(section): for line in fence.group(1).splitlines(): match = _GANTT_ROW_RE.fullmatch(line) if match is None: candidate = _GANTT_STAGE_CANDIDATE_RE.match(line) if candidate is not None: label = candidate.group(1) or sole_task_id or "" violations.append( f"Gantt: malformed Gantt stage row {label} Stage " f"{candidate.group(2)} — a row is a label and a bar, " "with nothing after it" ) continue stage_number = int(match.group(2)) task_id = match.group(1) or sole_task_id if not task_id: violations.append( f"Gantt: row S{stage_number} omits its task-id, which is " "only allowed when exactly one task is scheduled" ) continue bar = match.group(3) rows.append(_GanttRow( task_id=task_id, stage_number=stage_number, filled_cells=bar.count("█"), open_cells=bar.count("░"), )) return rows, violations def _cells_for_days(days: Decimal) -> Decimal: return days / GANTT_COLUMN_DAYS def _breakdown_days( breakdown_rows: list[_BreakdownRow], ) -> dict[tuple[str, int], tuple[Decimal, Decimal]]: days: dict[tuple[str, int], tuple[Decimal, Decimal]] = {} counts: dict[tuple[str, int], int] = {} for row in breakdown_rows: key = (row.task_id, row.stage_number) counts[key] = counts.get(key, 0) + 1 if row.days is not None: days[key] = row.days return {key: value for key, value in days.items() if counts[key] == 1} def _validate_gantt_geometry( rows: list[_GanttRow], breakdown_rows: list[_BreakdownRow], ) -> list[str]: """A bar's width must be the duration the Work Breakdown gives that stage. The bar is compared against the Days column itself rather than a `days=` label beside it: an annotation restating the table can agree with the label while disagreeing with the plan. """ days = _breakdown_days(breakdown_rows) violations: list[str] = [] for row in rows: day_range = days.get((row.task_id, row.stage_number)) if day_range is None: continue lower, upper = day_range expected_filled = _cells_for_days(lower) expected_open = _cells_for_days(upper - lower) if ( expected_filled != row.filled_cells or expected_open != row.open_cells ): violations.append( f"Gantt: bar for {row.task_id} Stage {row.stage_number} draws " f"{row.filled_cells}█+{row.open_cells}░ but Work Breakdown says " f"{_format_day_range(day_range)} days, which is " f"{expected_filled:.0f}█+{expected_open:.0f}░ at " f"{GANTT_COLUMN_DAYS} day per column" ) return violations def _validate_gantt_axis( text: str, rows: list[_GanttRow], breakdown_rows: list[_BreakdownRow], ) -> list[str]: section = _raw_section_body(text, "## Gantt Chart", 2) if section is None or not rows: return [] ticks: list[Decimal] = [] for fence in _PLAIN_FENCE_RE.finditer(section): for line in fence.group(1).splitlines(): axis = _GANTT_AXIS_RE.match(line) if axis is not None: ticks = [ Decimal(token) for token in re.findall(r"\d+", axis.group(1)) ] if not ticks: return [] days = _breakdown_days(breakdown_rows) span = sum( (days[(row.task_id, row.stage_number)][1] for row in rows if (row.task_id, row.stage_number) in days), Decimal("0"), ) if span and ticks[-1] > span + _GANTT_AXIS_TICK_DAYS: return [ f"Gantt: axis runs to day {ticks[-1]} but the schedule spans at " f"most {span} days — trim the axis to the work" ] return [] def _validate_gantt_coverage( rows: list[_GanttRow], tasks: tuple[SelectionTask, ...], ) -> list[str]: expected = { (task.task_id, number) for task in tasks for number in task.selected_stages } counts: dict[tuple[str, int], int] = {} for row in rows: key = (row.task_id, row.stage_number) counts[key] = counts.get(key, 0) + 1 violations: list[str] = [] for task_id, stage_number in sorted(expected): count = counts.get((task_id, stage_number), 0) if count == 0: violations.append(f"Gantt: missing Gantt row {task_id}/S{stage_number}") elif count > 1: violations.append(f"Gantt: duplicate Gantt row {task_id}/S{stage_number}") for task_id, stage_number in sorted(set(counts) - expected): violations.append( f"Gantt: Gantt row {task_id}/S{stage_number} is not selected" ) return violations def validate_schedule_semantics( text: str, selection_path: Path, ) -> list[str]: try: tasks = load_schedule_selection(selection_path) except (OSError, ValueError, json.JSONDecodeError) as exc: return [f"selection: {exc}"] try: rows, violations = _validate_work_breakdowns(text, tasks) violations.extend(_validate_effort_semantics(text, tasks, rows)) gantt_rows, gantt_violations = _parse_gantt_rows(text, tasks) violations.extend(gantt_violations) if gantt_rows is not None: violations.extend(_validate_gantt_coverage(gantt_rows, tasks)) violations.extend(_validate_gantt_geometry(gantt_rows, rows)) violations.extend(_validate_gantt_axis(text, gantt_rows, rows)) return violations except ScheduleSemanticError as exc: return [str(exc)]