#!/usr/bin/env python3 """Idempotently provision Phase 1 load fixtures through supported TT APIs.""" from __future__ import annotations import hashlib import json from pathlib import Path import sys import time from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) import phase1_e2e_entry_gate as identity_gate # noqa: E402 import phase1_load_harness as harness # noqa: E402 class ProvisionError(harness.LoadHarnessError): pass def _module_week_ids(template: Any) -> set[int]: """Return the module's effective teaching-week ids exactly as the engine sees them.""" module = template.module if getattr(module, "week_pattern_id", None): relation = module.week_pattern.week else: relation = module.week return {int(value) for value in relation.values_list("id", flat=True)} def _module_availability_pattern(template: Any) -> str: module = template.module if getattr(module, "availability_id", None): return str(module.availability.pattern or "") return str(module.availability_pattern or "") def _available_reference_slots( academic_pattern: str, module_pattern: str, slot_per_day: int ) -> list[int]: """Return slots accepted by both the term and its module. The engine pads a short module pattern with available slots, so the same normalization is used here. End-of-day starts are excluded because even a one-slot activity must not cross a day boundary. """ return [ index for index, academic_value in enumerate(academic_pattern or "") if academic_value == "0" and (module_pattern[index] if index < len(module_pattern) else "0") == "0" and index % slot_per_day < slot_per_day - 1 ] def _select_reference_template( templates: Any, approved_week_count: int, *, approved_academic_term_id: int, require_resource_free: bool = True, slot_per_day: int | None = None, ) -> tuple[Any, list[Any]]: """Select a module/template donor inside the approved academic term.""" for candidate in templates: if candidate.academic_term_id != approved_academic_term_id: continue if candidate.academic_term.week.count() < approved_week_count: continue if require_resource_free and any( relation.exists() for relation in ( candidate.staff, candidate.location, candidate.staff_suitability, candidate.location_suitability, ) ): continue weeks = list( candidate.academic_term.week.order_by("week", "id")[:approved_week_count] ) if len(weeks) != approved_week_count: continue module = candidate.module if getattr(module, "status", 1) != 1: continue module_term_id = getattr(module, "academic_term_id", None) if module_term_id != candidate.academic_term_id: continue if not {int(week.id) for week in weeks}.issubset(_module_week_ids(candidate)): continue if slot_per_day is not None and len( _available_reference_slots( candidate.academic_term.default_availability or "", _module_availability_pattern(candidate), slot_per_day, ) ) < 2: continue return candidate, weeks raise ProvisionError( "no engine-feasible maximum-recurrence template donor is available " "in the approved academic term" ) def _approved_term_scope(manifest: dict[str, Any]) -> dict[str, Any]: baseline = manifest.get("accepted_baseline") authoritative_term = ( baseline.get("authoritative_term") if isinstance(baseline, dict) else None ) if not isinstance(authoritative_term, dict): raise ProvisionError( "accepted baseline lacks the authoritative academic-term scope" ) academic_term_id = authoritative_term.get("academic_term_id") if ( isinstance(academic_term_id, bool) or not isinstance(academic_term_id, int) or academic_term_id < 1 ): raise ProvisionError("authoritative academic-term id is invalid") start_date = authoritative_term.get("start_date") end_date = authoritative_term.get("end_date") if not isinstance(start_date, str) or not isinstance(end_date, str): raise ProvisionError("authoritative academic-term dates are invalid") return { "academic_term_id": academic_term_id, "start_date": start_date, "end_date": end_date, } def _reference_facts(manifest: dict[str, Any]) -> dict[str, Any]: from api.models import TtActivityTemplate, TtSetting approved_term = _approved_term_scope(manifest) settings = TtSetting.get_multiple_setting( {"minute_per_slot", "slot_per_day", "slot_per_week"} ) minute_per_slot = int(settings["minute_per_slot"]) slot_per_day = int(settings["slot_per_day"]) templates = ( TtActivityTemplate.objects.filter( status=1, academic_term__status=1, academic_term_id=approved_term["academic_term_id"], academic_term__start_date=approved_term["start_date"], academic_term__end_date=approved_term["end_date"], activity_type__booking=False, module__isnull=False, module__status=1, ) .select_related( "academic_term", "activity_type", "module", "module__availability", "module__week_pattern", ) .prefetch_related( "academic_term__week", "module__week", "module__week_pattern__week", "staff", "location", "staff_suitability", "location_suitability", ) .order_by("id") ) approved_week_count = int( manifest["provision_plan"]["maximum_recurrence_week_count"] ) template, weeks = _select_reference_template( templates, approved_week_count, approved_academic_term_id=approved_term["academic_term_id"], require_resource_free=False, slot_per_day=slot_per_day, ) actual_term_scope = { "academic_term_id": int(template.academic_term_id), "start_date": template.academic_term.start_date.isoformat(), "end_date": template.academic_term.end_date.isoformat(), } if actual_term_scope != approved_term: raise ProvisionError( "selected fixture template differs from the approved academic-term scope" ) available = _available_reference_slots( template.academic_term.default_availability or "", _module_availability_pattern(template), slot_per_day, ) if len(available) < 2: raise ProvisionError( "approved academic term and module lack two deterministic common slots" ) return { "academic_term_id": int(template.academic_term_id), "academic_term_start_date": actual_term_scope["start_date"], "academic_term_end_date": actual_term_scope["end_date"], "activity_template_id": int(template.id), "donor_activity_template_id": int(template.id), "module_id": int(template.module_id), "activity_type_id": int(template.activity_type_id), "all_week_ids": [int(week.id) for week in weeks], "short_week_ids": [int(week.id) for week in weeks[:2]], "minute_per_slot": minute_per_slot, "slot_a": int(available[0]), "slot_b": int(available[1]), } def _parse_preschedule_feasibility_response( raw: Any, *, delivery_id: str, activity_id: int, required_slots: tuple[int, ...], ) -> dict[str, Any]: response_sha256 = hashlib.sha256(str(raw).encode("utf-8")).hexdigest() try: response = json.loads(raw) if isinstance(raw, str) else raw if not isinstance(response, dict): raise ValueError("response is not an object") schedule = response.get("schedule", []) schedule = json.loads(schedule) if isinstance(schedule, str) else schedule if not isinstance(schedule, list) or len(schedule) != 1: raise ValueError("preschedule result is not one activity") result = schedule[0] if not isinstance(result, dict) or int(result["activity"]) != activity_id: raise ValueError("preschedule activity differs from the canary") preferences = result.get("preference", []) preferences = ( json.loads(preferences) if isinstance(preferences, str) else preferences ) if not isinstance(preferences, list): raise ValueError("preschedule preferences are not an array") required_preferences = [int(preferences[slot]) for slot in required_slots] protocol_valid = ( str(response.get("request_id") or "") == delivery_id and str(response.get("status") or "").lower() == "success" ) return { "protocol_valid": protocol_valid, "required_slots_available": all(value > 0 for value in required_preferences), "required_preferences": required_preferences, "response_sha256": response_sha256, } except (IndexError, KeyError, TypeError, ValueError, json.JSONDecodeError): return { "protocol_valid": False, "required_slots_available": False, "required_preferences": [], "response_sha256": response_sha256, } def _assert_manifest_slots_match_reference( manifest: dict[str, Any], facts: dict[str, Any] ) -> dict[str, int]: """Fail before fixture writes when load slots differ from engine preflight slots.""" declared = { "slot_a": int(manifest["variables"]["slot_a"]), "slot_b": int(manifest["variables"]["slot_b"]), } actual = { "slot_a": int(facts["slot_a"]), "slot_b": int(facts["slot_b"]), } if declared != actual: raise ProvisionError( "approved manifest slots differ from the engine-feasible reference slots: " f"declared={declared} actual={actual}" ) return actual class Provisioner: def __init__( self, *, config: harness.LoadConfig, manifest: dict[str, Any], session: identity_gate.TimetablerAdminSession, ): self.config = config self.manifest = manifest self.session = session self.actor_id = int(session.user_id or 0) self.facts = _reference_facts(manifest) _assert_manifest_slots_match_reference(manifest, self.facts) self.created = {key: 0 for key in manifest["provision_plan"]["counts"]} self.reused = {key: 0 for key in manifest["provision_plan"]["counts"]} self.request_ids: set[str] = set() self.engine_feasibility_preflight: dict[str, dict[str, Any]] = {} def _expected_event(self, request_id: str) -> bool: from api.models import IntegrationOutbox rows = list( IntegrationOutbox.objects.filter(request_id=request_id) .values_list("transport_sequence", flat=True) .distinct()[:2] ) if len(rows) > 1: raise ProvisionError( "provision request id maps to multiple source transactions" ) return len(rows) == 1 def _create(self, *, path: str, payload: dict[str, Any], label: str) -> None: request_id = f"{self.config.run_id}-provision-{label}" if self._expected_event(request_id): raise ProvisionError( "fixture is absent but its stable provision transaction exists" ) before = harness._current_watermark() self.session.post(path, payload, request_id=request_id) deadline = time.monotonic() + 90 while harness._current_watermark() == before and time.monotonic() < deadline: time.sleep(0.2) if harness._current_watermark() != before + 1 or not self._expected_event( request_id ): raise ProvisionError( "supported fixture create did not emit exactly one source transaction" ) self.request_ids.add(request_id) def _owned(self, model, code: str, request_id: str): rows = list(model.objects.filter(code=code).order_by("id")[:2]) if len(rows) > 1: raise ProvisionError(f"duplicate run-owned fixture code: {code}") if not rows: return None row = rows[0] if row.created_by != self.actor_id or row.status != 1: raise ProvisionError( f"fixture collision is not owned by the load identity: {code}" ) if not self._expected_event(request_id): raise ProvisionError( f"reused fixture lacks its stable source transaction: {code}" ) self.request_ids.add(request_id) return row def assert_engine_feasibility( self, activity: Any, *, label: str, required_slots: tuple[int, ...] ) -> dict[str, Any]: """Prove required slots through the real engine without scheduling. Preschedule is advice-only: it produces a durable engine delivery and response but no Timetabler outbox transaction or Resource Booking write. A stable correlation id makes interrupted provision retries idempotent. """ from api.models import EngineResponseQuarantine, KafkaLog, PostCommitDelivery request_id = f"{self.config.run_id}-provision-engine-feasibility-{label}" before = harness._current_watermark() deliveries = list( PostCommitDelivery.objects.filter(correlation_id=request_id).order_by("id")[:2] ) if len(deliveries) > 1: raise ProvisionError("engine feasibility canary has duplicate deliveries") if not deliveries: self.session.post( "/api/admin/preschedule-request", {"activity_ids": [int(activity.id)]}, request_id=request_id, ) deadline = time.monotonic() + 120 while time.monotonic() < deadline: deliveries = list( PostCommitDelivery.objects.filter(correlation_id=request_id).order_by( "id" )[:2] ) if len(deliveries) > 1: raise ProvisionError("engine feasibility canary has duplicate deliveries") if not deliveries: time.sleep(0.2) continue delivery = deliveries[0] if ( delivery.channel != "kafka" or delivery.method != "preschedule" or not isinstance(delivery.payload, dict) or [int(value) for value in delivery.payload.get("activities", [])] != [int(activity.id)] ): raise ProvisionError("engine feasibility delivery contract is invalid") if delivery.status == PostCommitDelivery.Status.DEAD_LETTER: raise ProvisionError("engine feasibility delivery reached dead letter") if delivery.status != PostCommitDelivery.Status.PUBLISHED: time.sleep(0.2) continue delivery_id = str(delivery.delivery_id) if EngineResponseQuarantine.objects.filter(request_id=delivery_id).exists(): raise ProvisionError("engine feasibility response was quarantined") kafka_rows = list(KafkaLog.objects.filter(request_id=delivery_id)[:2]) if len(kafka_rows) > 1: raise ProvisionError("engine feasibility Kafka correlation is not unique") if not kafka_rows or not kafka_rows[0].response_at: time.sleep(0.2) continue parsed = _parse_preschedule_feasibility_response( kafka_rows[0].response_data, delivery_id=delivery_id, activity_id=int(activity.id), required_slots=required_slots, ) if not parsed["protocol_valid"]: raise ProvisionError("engine feasibility response contract is invalid") if not parsed["required_slots_available"]: raise ProvisionError( "Scheduling Engine rejected one or both approved load slots" ) if harness._current_watermark() != before: raise ProvisionError( "advice-only engine feasibility canary changed the source watermark" ) return { "verified": True, "advice_only": True, "required_slots": list(required_slots), "required_preferences": parsed["required_preferences"], "response_sha256": parsed["response_sha256"], "source_watermark_unchanged": True, } raise ProvisionError("timed out waiting for engine feasibility advice") def ensure_staff(self, category: str, index: int, count_key: str): from api.models import TtStaff code = f"{self.config.run_id}-{category}-{index:04d}" label = f"{category}-{index:04d}" request_id = f"{self.config.run_id}-provision-{label}" row = self._owned(TtStaff, code, request_id) if row is None: self._create( path="/api/admin/staff/create", payload={ "code": code, "name": f"{self.config.run_id} staff {index:04d}", "maximum_period": 1, "contract_period": 1, }, label=label, ) row = self._owned(TtStaff, code, request_id) self.created[count_key] += 1 else: self.reused[count_key] += 1 return row def ensure_location(self, category: str, index: int, count_key: str): from api.models import TtLocation code = f"{self.config.run_id}-{category}-{index:04d}" label = f"{category}-{index:04d}" request_id = f"{self.config.run_id}-provision-{label}" row = self._owned(TtLocation, code, request_id) if row is None: self._create( path="/api/admin/location/create", payload={ "code": code, "name": f"{self.config.run_id} location {index:04d}", "maximum_period": 1, "contract_period": 1, "capacity": 1, }, label=label, ) row = self._owned(TtLocation, code, request_id) self.created[count_key] += 1 else: self.reused[count_key] += 1 return row def ensure_activity( self, *, category: str, count_key: str, index: int, staff_id: int, location_id: int, week_ids: list[int], ): from api.models import TtActivity code = f"{self.config.run_id}-{category}-{index:04d}" label = f"{category}-{index:04d}" request_id = f"{self.config.run_id}-provision-{label}" row = self._owned(TtActivity, code, request_id) expected_weeks = week_ids if row is None: self._create( path="/api/admin/activity/create", payload={ "code": code, "name": f"{self.config.run_id} {category} {index:04d}", "duration": self.facts["minute_per_slot"], "planned_size": 1, "academic_term_id": self.facts["academic_term_id"], "activity_template_id": self.facts["activity_template_id"], "week_pattern": expected_weeks, }, label=label, ) row = self._owned(TtActivity, code, request_id) self.created[count_key] += 1 else: self.reused[count_key] += 1 if row is None or row.scheduled: raise ProvisionError( f"activity fixture is absent or already scheduled: {code}" ) if list(row.week.order_by("id").values_list("id", flat=True)) != sorted( expected_weeks ): raise ProvisionError( f"activity fixture has an incompatible recurrence: {code}" ) if row.staff.exists() or row.location.exists(): raise ProvisionError( f"unscheduled activity fixture already has a final allocation: {code}" ) candidate_state_is_exact = ( set(row.staff_preset.values_list("id", flat=True)) == {staff_id} and set(row.location_preset.values_list("id", flat=True)) == {location_id} and not row.staff_suitability.exists() and not row.location_suitability.exists() and int(row.staff_requirement or 0) == 1 and int(row.location_requirement or 0) == 1 and row.staff_requirement_type == TtActivity.STAFF_REQUIREMENT_TYPE["preset"] and row.location_requirement_type == TtActivity.LOCATION_REQUIREMENT_TYPE["preset"] ) if not candidate_state_is_exact: before = harness._current_watermark() self.session.post( "/api/admin/resources/update-requirement", { "activity_id": int(row.id), "staff_requirement": 1, "staff_preset": [staff_id], "staff_suitability": [], "location_requirement": 1, "location_preset": [location_id], "location_suitability": [], }, request_id=f"{request_id}-requirements", ) if harness._current_watermark() != before: raise ProvisionError( "advisory requirement preparation unexpectedly emitted a source transaction" ) row.refresh_from_db() if ( row.scheduled or row.staff.exists() or row.location.exists() or set(row.staff_preset.values_list("id", flat=True)) != {staff_id} or set(row.location_preset.values_list("id", flat=True)) != {location_id} or row.staff_suitability.exists() or row.location_suitability.exists() or int(row.staff_requirement or 0) != 1 or int(row.location_requirement or 0) != 1 or row.staff_requirement_type != TtActivity.STAFF_REQUIREMENT_TYPE["preset"] or row.location_requirement_type != TtActivity.LOCATION_REQUIREMENT_TYPE["preset"] ): raise ProvisionError( f"activity fixture lacks exact candidate requirements: {code}" ) return row def run(self) -> None: counts = self.manifest["provision_plan"]["counts"] for index in range(1, counts["lifecycle_staff"] + 1): self.ensure_staff("lifecycle-staff", index, "lifecycle_staff") for index in range(1, counts["lifecycle_location"] + 1): self.ensure_location("lifecycle-location", index, "lifecycle_location") staff = [ self.ensure_staff("allocation-staff", index, "allocation_staff") for index in range(1, counts["allocation_staff"] + 1) ] locations = [ self.ensure_location("allocation-location", index, "allocation_location") for index in range(1, counts["allocation_location"] + 1) ] category_contracts = ( ("traffic", "traffic_activity", 0, self.facts["short_week_ids"]), ( "rate-recurrence", "rate_recurrence_activity", counts["traffic_activity"], self.facts["short_week_ids"], ), ("recurrence", "recurrence_activity", 0, self.facts["all_week_ids"]), ("bulk", "bulk_activity", 0, self.facts["short_week_ids"]), ("boundary", "boundary_activity", 0, self.facts["short_week_ids"]), ) for category, count_key, shard_offset, week_ids in category_contracts: total = counts[count_key] for index in range(1, total + 1): shard = (shard_offset + index - 1) % len(staff) activity = self.ensure_activity( category=category, count_key=count_key, index=index, staff_id=int(staff[shard].id), location_id=int(locations[shard].id), week_ids=week_ids, ) if category == "traffic" and index == 1: self.engine_feasibility_preflight["ordinary"] = ( self.assert_engine_feasibility( activity, label="ordinary", required_slots=(self.facts["slot_a"],), ) ) if category == "rate-recurrence" and index == 1: self.engine_feasibility_preflight["rate_recurrence"] = ( self.assert_engine_feasibility( activity, label="rate-recurrence", required_slots=(self.facts["slot_a"],), ) ) if set(self.engine_feasibility_preflight) != {"ordinary", "rate_recurrence"}: raise ProvisionError("both engine feasibility canaries must execute") def execute(config: harness.LoadConfig) -> dict[str, Any]: identity_gate.bootstrap_django_runtime() manifest = harness.load_manifest( config.manifest_json, run_id=config.run_id, environment=config.environment, source_scope=config.source_scope, require_approval=True, ) if config.action != "provision": raise ProvisionError("provisioner accepts only the provision profile") if not config.mutation_confirmed or config.operational_confirmed: raise ProvisionError("provision confirmation flags are invalid") if config.expected_sha != manifest["commits"]["timetabler"]: raise ProvisionError( "deployed Timetabler SHA differs from the canonical manifest" ) harness.assert_evidence_exchange_runtime(manifest, run_id=config.run_id) predecessor = harness.verify_predecessor_evidence( run_id=config.run_id, stage="provision", manifest=manifest, supplied_sha256=config.prior_evidence_sha256, ) harness.assert_approved_window(manifest, "PROVISION") start_fence = manifest["expected_stage_start_watermarks"]["provision"] final_fence = manifest["expected_stage_final_watermarks"]["provision"] before_metrics = harness._assert_source_safety() current = before_metrics["transport_watermark"] if not start_fence <= current <= final_fence: raise ProvisionError( "partial provision watermark is outside approved retry fences" ) session = identity_gate.TimetablerAdminSession( base_url=identity_gate.validate_base_url( config.base_url, "Timetabler base URL" ), email=config.email, password=config.password, run_id=config.run_id, ) session.login_and_attest() try: provisioner = Provisioner(config=config, manifest=manifest, session=session) provisioner.run() finally: session.logout() expected_requests = { f"{config.run_id}-provision-{category}-{index:04d}" for category, count_key in ( ("lifecycle-staff", "lifecycle_staff"), ("lifecycle-location", "lifecycle_location"), ("allocation-staff", "allocation_staff"), ("allocation-location", "allocation_location"), ("traffic", "traffic_activity"), ("rate-recurrence", "rate_recurrence_activity"), ("recurrence", "recurrence_activity"), ("bulk", "bulk_activity"), ("boundary", "boundary_activity"), ) for index in range(1, manifest["provision_plan"]["counts"][count_key] + 1) } final_metrics, transactions = harness._wait_for_stage_publication( start=start_fence, expected_delta=manifest["provision_plan"]["expected_source_transactions"], request_ids=expected_requests, timeout_seconds=manifest["profile_timing"]["PROVISION"]["drain_seconds"], ) if final_metrics["transport_watermark"] != final_fence: raise ProvisionError("provision did not reach its exact final watermark") for transaction in transactions: transaction["workload_classification"] = "fixture_provision" fixture_manifest = { "counts": manifest["provision_plan"]["counts"], "academic_term_id": provisioner.facts["academic_term_id"], "academic_term_start_date": provisioner.facts["academic_term_start_date"], "academic_term_end_date": provisioner.facts["academic_term_end_date"], "authoritative_term_scope_enforced": True, "activity_template_id": provisioner.facts["activity_template_id"], "donor_activity_template_id": provisioner.facts[ "donor_activity_template_id" ], "term_template_donor_only": True, "template_candidate_defaults_authoritative": False, "candidate_defaults_replaced_before_scheduling": True, "run_owned_activity_final_allocations_empty_at_provision": True, "advisory_requirement_source_watermark_unchanged": True, "maximum_recurrence_week_count": len(provisioner.facts["all_week_ids"]), "slot_a": provisioner.facts["slot_a"], "slot_b": provisioner.facts["slot_b"], "complete_staff_location_presets": True, "engine_feasibility_preflight": provisioner.engine_feasibility_preflight, } stage_assertions = { "created_counts": provisioner.created, "reused_counts": provisioner.reused, "fixture_manifest": fixture_manifest, "engine_feasibility_preflight": provisioner.engine_feasibility_preflight, "idempotent_retry_inspected": True, "supported_timetabler_apis_only": True, } return harness.seal_evidence( { "schema_version": 1, "profile": "PROVISION", "service": "timetabler", "repository": "Mayvins/timetabler-be", "run_id": config.run_id, "environment": "staging", "production": False, "git_sha": config.expected_sha, "resource_booking_git_sha_attestation": manifest["commits"][ "resource_booking" ], "github_run_id": config.github_run_id, "source_scope": config.source_scope, "entry_watermark": start_fence, "resume_watermark": current, "final_watermark": final_fence, "prior_evidence_sha256": config.prior_evidence_sha256, "predecessor_evidence": predecessor, "configuration_snapshot_sha256": manifest["configuration_snapshot"][ "sha256" ], "evidence_exchange": manifest["evidence_exchange"], "workload_parameters": manifest["load"], "profile_timing": manifest["profile_timing"]["PROVISION"], "approved_window": manifest["approved_window"], "thresholds": manifest["thresholds"], "stage_assertions": stage_assertions, "created_counts": provisioner.created, "reused_counts": provisioner.reused, "source_transactions": transactions, "source_metric_samples": [before_metrics, final_metrics], "host_metric_samples": [], "final_source_metrics": final_metrics, "fixture_manifest": fixture_manifest, "idempotent_retry_inspected": True, "load_executed": False, "mutation_executed": True, "restart_executed": False, "reverse_delivery_enabled": False, "phase2_enabled": False, "dr_executed": False, "resource_booking_database_touched": False, "resource_booking_process_touched": False, "direct_database_mutation": False, "manual_outbox_edit": False, } ) def main() -> int: try: config = harness.LoadConfig.from_environment() evidence = execute(config) persisted = harness.persist_evidence(evidence) print( json.dumps( { "phase1_load": "evidence_persisted", "profile": "PROVISION", "run_id": config.run_id, "entry_watermark": evidence["entry_watermark"], "resume_watermark": evidence["resume_watermark"], "final_watermark": evidence["final_watermark"], "load_executed": False, "dr_executed": False, "reverse_delivery_enabled": False, **persisted, }, sort_keys=True, ) ) return 0 except ( ProvisionError, harness.LoadHarnessError, identity_gate.EntryGateError, ) as error: print( json.dumps( {"phase1_load_provision": "blocked", "reason": str(error)}, sort_keys=True, ), file=sys.stderr, ) return 3 if __name__ == "__main__": raise SystemExit(main())