#!/usr/bin/env python3 """Validate the deterministic advanced planning challenge.""" from __future__ import annotations import argparse import json import re from dataclasses import dataclass from pathlib import Path from typing import Any @dataclass(frozen=True) class TaskSpec: duration: int resource: str people: tuple[str, ...] days: tuple[int, ...] TASKS = { "A": TaskSpec(60, "NAS", ("Mia",), (1,)), "B": TaskSpec(90, "DB", ("Mia", "Noah"), (1, 2)), "C": TaskSpec(60, "PROD", ("Noah",), (2,)), "D": TaskSpec(60, "SCANNER", ("Priya",), (1, 2)), "E": TaskSpec(90, "BUILD", ("Leo",), (1,)), "F": TaskSpec(120, "QA", ("Priya", "Leo"), (2,)), "G": TaskSpec(60, "ATLAS", ("Mia", "Noah", "Priya", "Leo"), (3,)), "H": TaskSpec(90, "QUIET", ("Mia",), (2,)), } DEPENDENCIES = ( ("A", "B"), ("B", "C"), ("B", "D"), ("E", "C"), ("C", "F"), ("D", "F"), ("C", "H"), ("F", "G"), ("H", "G"), ) AVAILABILITY = { "Mia": {1: ((540, 720), (780, 1020)), 2: ((540, 720), (780, 960)), 3: ((540, 840),)}, "Noah": {1: ((600, 720), (780, 1020)), 2: ((540, 720), (840, 1020)), 3: ((540, 840),)}, "Priya": {1: ((540, 660), (840, 1020)), 2: ((600, 780), (840, 1020)), 3: ((540, 840),)}, "Leo": {1: ((540, 720), (780, 1020)), 2: ((540, 720), (780, 1020)), 3: ((600, 840),)}, } TIME_RE = re.compile(r"^(?:[01]\d|2[0-3]):[0-5]\d$") def minutes(value: Any) -> int | None: if not isinstance(value, str) or not TIME_RE.fullmatch(value): return None hour, minute = map(int, value.split(":")) if minute not in (0, 30): return None return hour * 60 + minute def absolute(day: int, minute: int) -> int: return (day - 1) * 1440 + minute def validate_payload(payload: Any) -> dict[str, Any]: errors: list[str] = [] checks: dict[str, bool] = {} rows = payload.get("tasks") if isinstance(payload, dict) else None checks["root_shape"] = isinstance(payload, dict) and set(payload) == {"tasks"} and isinstance(rows, list) if not isinstance(rows, list): return {"passed": 0, "total": 1, "percent": 0.0, "checks": checks, "errors": ["Root must contain only a tasks array."]} expected_fields = {"id", "day", "start", "end", "resource", "people"} parsed: dict[str, dict[str, Any]] = {} valid_row_shapes = True for index, row in enumerate(rows): if not isinstance(row, dict) or set(row) != expected_fields: valid_row_shapes = False errors.append(f"Row {index} must contain exactly {sorted(expected_fields)}.") continue task_id = row.get("id") if not isinstance(task_id, str): valid_row_shapes = False errors.append(f"Row {index} has an invalid id.") continue if task_id in parsed: errors.append(f"Task {task_id} appears more than once.") parsed[task_id] = row checks["exact_row_shape"] = valid_row_shapes checks["all_tasks_once"] = len(rows) == len(TASKS) and set(parsed) == set(TASKS) and len(parsed) == len(rows) if not checks["all_tasks_once"]: errors.append(f"Expected tasks A-H exactly once; received {sorted(parsed)} across {len(rows)} rows.") normalized: dict[str, dict[str, Any]] = {} correct_fields = True correct_times = True correct_durations = True correct_days = True availability_ok = True for task_id, spec in TASKS.items(): row = parsed.get(task_id) if row is None: continue day = row.get("day") start = minutes(row.get("start")) end = minutes(row.get("end")) people = row.get("people") if row.get("resource") != spec.resource or not isinstance(people, list) or sorted(people) != sorted(spec.people): correct_fields = False errors.append(f"Task {task_id} has incorrect resource or people.") if not isinstance(day, int) or isinstance(day, bool) or day not in (1, 2, 3) or start is None or end is None: correct_times = False errors.append(f"Task {task_id} has an invalid day or 30-minute time.") continue if end <= start or end - start != spec.duration: correct_durations = False errors.append(f"Task {task_id} must last {spec.duration} minutes.") if day not in spec.days: correct_days = False errors.append(f"Task {task_id} is scheduled on disallowed day {day}.") normalized[task_id] = {"day": day, "start": start, "end": end, "resource": row.get("resource"), "people": tuple(people) if isinstance(people, list) else ()} for person in spec.people: windows = AVAILABILITY[person].get(day, ()) if not any(start >= window_start and end <= window_end for window_start, window_end in windows): availability_ok = False errors.append(f"Task {task_id} falls outside {person}'s availability.") checks["resource_and_people"] = correct_fields checks["valid_30_minute_times"] = correct_times checks["correct_durations"] = correct_durations and len(normalized) == len(TASKS) checks["allowed_days"] = correct_days and len(normalized) == len(TASKS) checks["people_available"] = availability_ok and len(normalized) == len(TASKS) fixed_demo = normalized.get("G") checks["fixed_demo_slot"] = fixed_demo is not None and fixed_demo["day"] == 3 and fixed_demo["start"] == 660 and fixed_demo["end"] == 720 if not checks["fixed_demo_slot"]: errors.append("Task G must be fixed on day 3 from 11:00 to 12:00.") dependency_ok = True for predecessor, successor in DEPENDENCIES: before = normalized.get(predecessor) after = normalized.get(successor) if before is None or after is None or absolute(before["day"], before["end"]) > absolute(after["day"], after["start"]): dependency_ok = False errors.append(f"Dependency violated: {predecessor} must finish before {successor} starts.") checks["dependencies"] = dependency_ok people_overlap_ok = True resource_overlap_ok = True ids = sorted(normalized) for index, left_id in enumerate(ids): left = normalized[left_id] for right_id in ids[index + 1 :]: right = normalized[right_id] overlaps = left["day"] == right["day"] and left["start"] < right["end"] and right["start"] < left["end"] if not overlaps: continue shared_people = sorted(set(left["people"]) & set(right["people"])) if shared_people: people_overlap_ok = False errors.append(f"Tasks {left_id} and {right_id} overlap for {', '.join(shared_people)}.") if left["resource"] == right["resource"]: resource_overlap_ok = False errors.append(f"Tasks {left_id} and {right_id} overlap on resource {left['resource']}.") checks["no_people_overlap"] = people_overlap_ok checks["no_resource_overlap"] = resource_overlap_ok passed = sum(checks.values()) return { "passed": passed, "total": len(checks), "percent": round(100 * passed / len(checks), 1), "checks": checks, "errors": errors, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("session", type=Path) args = parser.parse_args() response_path = args.session / "ollama-response.json" response = json.loads(response_path.read_text(encoding="utf-8")).get("response", "") try: payload = json.loads(response) except json.JSONDecodeError as exc: result = {"session": str(args.session), "passed": 0, "total": 1, "percent": 0.0, "checks": {"strict_json": False}, "errors": [f"Invalid JSON: {exc}"]} else: result = {"session": str(args.session), "strictJson": True, **validate_payload(payload)} print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 if result["passed"] == result["total"] else 1 if __name__ == "__main__": raise SystemExit(main())