#!/usr/bin/env python3 """Read-only staging attestation for the Jan--June 2026 Phase 1 E2E term. The script is intentionally incapable of authenticating to an API, publishing Kafka work, or changing domain state. The workflow executes it inside a PostgreSQL READ ONLY transaction against the attested Timetabler checkout. """ from __future__ import annotations import datetime import hashlib import json import os import re import sys from collections import Counter from pathlib import Path from typing import Any WINDOW_START = datetime.date(2026, 1, 1) WINDOW_END = datetime.date(2026, 6, 30) EXPECTED_ACTIVITY_COUNT = 46 GIT_SHA = re.compile(r"^[0-9a-f]{40}$") class TermAttestationError(RuntimeError): """A fail-closed staging precondition was not met.""" def __init__(self, message: str, *, details: dict[str, Any] | None = None): super().__init__(message) self.details = details or {} def _sha256(value: Any) -> str: encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), default=str ).encode() return hashlib.sha256(encoded).hexdigest() def _emit_bounded_fact(path: list[str], value: Any, *, max_bytes: int = 6_000) -> None: record = {"term_attestation_fact": {"path": path, "value": value}} encoded = json.dumps(record, sort_keys=True, separators=(",", ":")) if len(encoded.encode()) <= max_bytes: print(encoded) return if isinstance(value, dict): for key in sorted(value): _emit_bounded_fact([*path, str(key)], value[key], max_bytes=max_bytes) return if isinstance(value, list): start = 0 while start < len(value): end = min(len(value), start + 100) while end > start: chunk_record = { "term_attestation_fact": { "path": path, "slice_start": start, "slice_end": end, "value": value[start:end], } } chunk = json.dumps( chunk_record, sort_keys=True, separators=(",", ":") ) if len(chunk.encode()) <= max_bytes: print(chunk) start = end break end -= 1 else: raise TermAttestationError("one audit list item exceeds output bound") return raise TermAttestationError("one audit scalar exceeds output bound") def _ids(relation: Any) -> list[int]: return sorted(int(value) for value in relation.values_list("id", flat=True)) def _activity_weeks(activity: Any) -> list[Any]: relation = activity.week_pattern.week if activity.week_pattern_id else activity.week return list(relation.order_by("week", "id")) def select_term_candidate(candidates: list[dict[str, Any]]) -> dict[str, Any]: """Resolve one active Jan--June term; count is attested independently.""" matches = [ candidate for candidate in candidates if candidate["start_date"].year == 2026 and candidate["start_date"].month == 1 and candidate["end_date"].year == 2026 and candidate["end_date"].month == 6 and int(candidate["status"]) == 1 ] if len(matches) != 1: raise TermAttestationError( "expected exactly one active Jan--June 2026 academic term", details={ "overlapping_term_candidates": [ { **candidate, "start_date": candidate["start_date"].isoformat(), "end_date": candidate["end_date"].isoformat(), } for candidate in candidates ], "strict_match_count": len(matches), }, ) return matches[0] def _requirement_candidate_ids(activity: Any, kind: str) -> list[int]: requirement_type = int(getattr(activity, f"{kind}_requirement_type") or 0) preset = getattr(activity, f"{kind}_preset") if requirement_type == 2: return _ids(preset) if requirement_type != 1: return [] suitability_ids = _ids(getattr(activity, f"{kind}_suitability")) model_name = "TtStaff" if kind == "staff" else "TtLocation" from api import models model = getattr(models, model_name) queryset = model.objects.filter(status=1) if suitability_ids: queryset = queryset.filter(suitability__id__in=suitability_ids) return sorted(set(int(value) for value in queryset.values_list("id", flat=True))) def _resource_map_coverage( *, kind: str, resource_ids: list[int], week_ids: list[int] ) -> dict[str, Any]: from api import models map_model = getattr( models, "TtStaffResourceMap" if kind == "staff" else "TtLocationResourceMap" ) resource_field = f"{kind}_id" rows = list( map_model.objects.filter( **{f"{resource_field}__in": resource_ids, "week_id__in": week_ids} ).values_list(resource_field, "week_id", "pattern") ) pairs = {(int(resource_id), int(week_id)) for resource_id, week_id, _ in rows} expected_pairs = { (resource_id, week_id) for resource_id in resource_ids for week_id in week_ids } busy_slots = sum(str(pattern or "").count("1") for _, _, pattern in rows) return { "candidate_count": len(resource_ids), "candidate_ids_sha256": _sha256(resource_ids), "expected_map_rows": len(expected_pairs), "observed_map_rows": len(pairs), "missing_map_rows": len(expected_pairs - pairs), "busy_slot_marks": busy_slots, "complete": pairs == expected_pairs, } def _activity_inventory(activity: Any, scope_ids: set[int]) -> dict[str, Any]: weeks = _activity_weeks(activity) week_ids = [int(week.id) for week in weeks] staff_candidates = _requirement_candidate_ids(activity, "staff") location_candidates = _requirement_candidate_ids(activity, "location") allocated_staff_ids = _ids(activity.staff) allocated_location_ids = _ids(activity.location) from api.models import TtActivity foreign_staff_users = sorted( set( int(value) for value in TtActivity.objects.filter( scheduled=1, staff__id__in=staff_candidates ) .exclude(id__in=scope_ids) .values_list("id", flat=True) ) ) foreign_location_users = sorted( set( int(value) for value in TtActivity.objects.filter( scheduled=1, location__id__in=location_candidates ) .exclude(id__in=scope_ids) .values_list("id", flat=True) ) ) staff_required = ( len(staff_candidates) if int(activity.staff_requirement_type or 0) == 2 else int(activity.staff_requirement or 0) ) location_required = ( len(location_candidates) if int(activity.location_requirement_type or 0) == 2 else int(activity.location_requirement or 0) ) staff_maps = _resource_map_coverage( kind="staff", resource_ids=staff_candidates, week_ids=week_ids ) location_maps = _resource_map_coverage( kind="location", resource_ids=location_candidates, week_ids=week_ids ) input_blockers: list[str] = [] if not week_ids: input_blockers.append("no_occurrence_weeks") if not activity.slot_required or int(activity.slot_required) <= 0: input_blockers.append("invalid_slot_requirement") if staff_required < 1 or len(staff_candidates) < staff_required: input_blockers.append("insufficient_staff_candidates") if location_required < 1 or len(location_candidates) < location_required: input_blockers.append("insufficient_location_candidates") if not staff_maps["complete"]: input_blockers.append("incomplete_staff_resource_maps") if not location_maps["complete"]: input_blockers.append("incomplete_location_resource_maps") return { "activity_id": int(activity.id), "status": int(activity.status), "scheduled": bool(activity.scheduled), "is_booking": bool(activity.is_booking), "is_jta": bool(activity.is_jta), "is_variant": bool(activity.is_variant), "variant_parent_id": ( int(activity.variant_parent_id) if activity.variant_parent_id else None ), "slot_required": int(activity.slot_required or 0), "week_ids": week_ids, "week_start_dates": [ week.start_date.isoformat() if week.start_date else None for week in weeks ], "scheduled_day": activity.scheduled_day, "scheduled_start_slot": activity.scheduled_start_slot, "allocated_staff_ids": allocated_staff_ids, "allocated_location_ids": allocated_location_ids, "staff_requirement": { "type": int(activity.staff_requirement_type or 0), "count": staff_required, "preset_ids": _ids(activity.staff_preset), "suitability_ids": _ids(activity.staff_suitability), "candidate_ids": staff_candidates, "resource_maps": staff_maps, }, "location_requirement": { "type": int(activity.location_requirement_type or 0), "count": location_required, "preset_ids": _ids(activity.location_preset), "suitability_ids": _ids(activity.location_suitability), "candidate_ids": location_candidates, "resource_maps": location_maps, }, "foreign_scheduled_activity_ids_using_candidate_staff": foreign_staff_users, "foreign_scheduled_activity_ids_using_candidate_locations": foreign_location_users, "engine_input_blockers": input_blockers, "engine_input_ready": not input_blockers, "would_replace_existing_allocation": bool( activity.scheduled or allocated_staff_ids or allocated_location_ids ), } def _database_attestation(connection: Any) -> dict[str, Any]: database = connection.settings_dict fingerprint = { "vendor": connection.vendor, "engine": str(database.get("ENGINE") or ""), "name": str(database.get("NAME") or ""), "host": str(database.get("HOST") or ""), "port": str(database.get("PORT") or ""), } with connection.cursor() as cursor: cursor.execute("SELECT current_database(), current_user") current_database, current_user = cursor.fetchone() return { "alias": connection.alias, "vendor": connection.vendor, "configuration_sha256": _sha256(fingerprint), "current_database_sha256": hashlib.sha256( str(current_database).encode() ).hexdigest(), "current_user_sha256": hashlib.sha256(str(current_user).encode()).hexdigest(), "other_database_connection_used": False, } def _replacement_states(row: Any) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: committed = (row.payload or {}).get("committed_state") or {} replacement = committed.get("replacement") or {} return replacement.get("previous"), replacement.get("current") def summarize_activity_projection( rows: list[Any], *, watermark: int, academic_term_id: int ) -> dict[str, Any]: """Mirror the fixed-watermark latest-aggregate selection for activities.""" latest: dict[str, Any] = {} for row in sorted( (row for row in rows if int(row.transport_sequence) <= watermark), key=lambda item: ( int(item.transport_sequence), int(item.event_version), int(item.id), ), ): latest[str(row.aggregate_id)] = row live_all_ids: list[int] = [] live_term_ids: list[int] = [] term_tombstone_ids: list[int] = [] term_departed_ids: list[int] = [] for row in latest.values(): previous, current = _replacement_states(row) if current: live_all_ids.append(int(current["id"])) if int(current.get("academic_term_id") or 0) == academic_term_id: live_term_ids.append(int(current["id"])) if ( previous and int(previous.get("academic_term_id") or 0) == academic_term_id and not current ): term_tombstone_ids.append(int(previous["id"])) elif ( previous and int(previous.get("academic_term_id") or 0) == academic_term_id and current and int(current.get("academic_term_id") or 0) != academic_term_id ): term_departed_ids.append(int(previous["id"])) return { "watermark": watermark, "latest_activity_aggregate_count": len(latest), "live_activity_aggregate_count": len(set(live_all_ids)), "live_activity_ids_sha256": _sha256(sorted(set(live_all_ids))), "term_live_activity_count": len(set(live_term_ids)), "term_live_activity_ids": sorted(set(live_term_ids)), "term_live_activity_ids_sha256": _sha256(sorted(set(live_term_ids))), "term_latest_tombstone_ids": sorted(set(term_tombstone_ids)), "term_latest_departed_ids": sorted(set(term_departed_ids)), } def _term_history(rows: list[Any], *, academic_term_id: int) -> dict[str, Any]: events: list[Any] = [] activity_ids: set[int] = set() for row in rows: previous, current = _replacement_states(row) states = [state for state in (previous, current) if state] matching = [ state for state in states if int(state.get("academic_term_id") or 0) == academic_term_id ] if matching: events.append(row) activity_ids.update(int(state["id"]) for state in matching) return { "ever_attributed_activity_count": len(activity_ids), "ever_attributed_activity_ids": sorted(activity_ids), "ever_attributed_activity_ids_sha256": _sha256(sorted(activity_ids)), "event_count": len(events), "event_types": dict(Counter(str(row.event_type) for row in events)), "first_sequence": min( (int(row.transport_sequence) for row in events), default=None ), "last_sequence": max( (int(row.transport_sequence) for row in events), default=None ), "event_lineage_sha256": _sha256( [ { "aggregate_id": str(row.aggregate_id), "event_type": str(row.event_type), "event_version": int(row.event_version), "transport_sequence": int(row.transport_sequence), } for row in events ] ), } def build_attestation() -> dict[str, Any]: from django.conf import settings from django.db import connection from api.models import ( IntegrationOutbox, TtAcademicTerm, TtActivity, TtActivityTemplate, TtSetting, ) from deploy.phase1_load_harness import _assert_source_safety expected_watermark = int(os.environ["TT_PHASE1_TERM_EXPECTED_WATERMARK"]) deployed_sha = os.environ.get("TT_PHASE1_TERM_DEPLOYED_SHA", "") if not GIT_SHA.fullmatch(deployed_sha): raise TermAttestationError("deployed Timetabler SHA is not attested") if connection.vendor != "postgresql": raise TermAttestationError("staging attestation requires PostgreSQL") integration = settings.RESOURCE_BOOKING_INTEGRATION configuration = { "capture_enabled": bool(integration["CAPTURE_ENABLED"]), "publish_enabled": bool(integration["PUBLISH_ENABLED"]), "activation_approved": bool(integration["ACTIVATION_APPROVED"]), "phase": str(integration["PHASE"]), "reverse_delivery_enabled": bool(integration["REVERSE_DELIVERY_ENABLED"]), "transport": str(integration["TRANSPORT"]), "schema_version": int(integration["SCHEMA_VERSION"]), "source_scope": str(integration["SOURCE_SCOPE"]), "topic_sha256": hashlib.sha256( str(integration.get("KAFKA_TOPIC") or "").encode() ).hexdigest(), } if configuration != { **configuration, "capture_enabled": True, "publish_enabled": True, "activation_approved": True, "phase": "phase1", "reverse_delivery_enabled": False, "transport": "kafka", "schema_version": 2, }: raise TermAttestationError("Timetabler Phase 1 configuration is not safe") with connection.cursor() as cursor: cursor.execute("SET TRANSACTION READ ONLY") source = _assert_source_safety(expected_watermark=expected_watermark) database = _database_attestation(connection) candidate_rows = list( TtAcademicTerm.objects.filter( start_date__lte=WINDOW_END, end_date__gte=WINDOW_START ) .values("id", "start_date", "end_date", "status") .order_by("start_date", "end_date", "id") ) candidates = [ { **candidate, "activity_count": TtActivity.objects.filter( academic_term_id=candidate["id"] ).count(), } for candidate in candidate_rows ] term = select_term_candidate(candidates) activities = list( TtActivity.objects.filter(academic_term_id=term["id"]) .select_related("week_pattern") .prefetch_related( "week", "week_pattern__week", "staff", "location", "staff_preset", "location_preset", "staff_suitability", "location_suitability", ) .order_by("id") ) activity_count_matches = len(activities) == EXPECTED_ACTIVITY_COUNT scope_ids = {int(activity.id) for activity in activities} inventory = [_activity_inventory(activity, scope_ids) for activity in activities] scheduled_ids = [item["activity_id"] for item in inventory if item["scheduled"]] overwrite_ids = [ item["activity_id"] for item in inventory if item["would_replace_existing_allocation"] ] input_blocked_ids = [ item["activity_id"] for item in inventory if not item["engine_input_ready"] ] foreign_conflict_ids = sorted( { activity_id for item in inventory for activity_id in ( item["foreign_scheduled_activity_ids_using_candidate_staff"] + item["foreign_scheduled_activity_ids_using_candidate_locations"] ) } ) candidate_staff_ids = sorted( { resource_id for item in inventory for resource_id in item["staff_requirement"]["candidate_ids"] } ) candidate_location_ids = sorted( { resource_id for item in inventory for resource_id in item["location_requirement"]["candidate_ids"] } ) settings_map = TtSetting.get_multiple_setting( {"slot_per_week", "slot_per_day", "minute_per_slot"} ) variants = [item for item in inventory if item["is_variant"]] activity_rows = list( IntegrationOutbox.objects.filter( source_scope=configuration["source_scope"], aggregate_type="activity", transaction_finalized=True, transport_sequence__lte=expected_watermark, ).order_by("transport_sequence", "transaction_index", "id") ) projection_fences = sorted({42, 117, expected_watermark}) projection = [ summarize_activity_projection( activity_rows, watermark=watermark, academic_term_id=int(term["id"]) ) for watermark in projection_fences if watermark <= expected_watermark ] history = _term_history(activity_rows, academic_term_id=int(term["id"])) current_ids = sorted(scope_ids) historical_missing_ids = sorted( set(history["ever_attributed_activity_ids"]) - scope_ids ) current_categories = { "raw_count": len(activities), "active_count": sum(int(activity.status) == 1 for activity in activities), "archived_or_inactive_count": sum( int(activity.status) != 1 for activity in activities ), "non_jta_count": sum(not bool(activity.is_jta) for activity in activities), "jta_parent_count": sum( bool(activity.is_jta) and not activity.jta_parent_id for activity in activities ), "jta_child_count": sum( bool(activity.is_jta) and bool(activity.jta_parent_id) for activity in activities ), "variant_count": sum(bool(activity.is_variant) for activity in activities), "variant_parent_reference_count": len( { int(activity.variant_parent_id) for activity in activities if activity.variant_parent_id } ), "admin_activity_list_visible_count": TtActivity.objects.filter( academic_term_id=term["id"] ) .exclude(is_jta=1, jta_parent_id__isnull=False) .count(), "activity_template_count": TtActivityTemplate.objects.filter( academic_term_id=term["id"] ).count(), "active_activity_template_count": TtActivityTemplate.objects.filter( academic_term_id=term["id"], status=1 ).count(), } result = { "schema_version": 1, "profile": "JAN_JUN_2026_TERM_ATTESTATION", "service": "timetabler", "repository": "Mayvins/timetabler-be", "deployed_git_sha": deployed_sha, "captured_at": datetime.datetime.now(datetime.UTC).isoformat(), "expected_source_watermark": expected_watermark, "source": source, "database": database, "configuration": configuration, "configuration_sha256": _sha256(configuration), "academic_term": { "id": int(term["id"]), "start_date": term["start_date"].isoformat(), "end_date": term["end_date"].isoformat(), "status": int(term["status"]), "activity_count": len(inventory), "expected_activity_count": EXPECTED_ACTIVITY_COUNT, "activity_count_matches": activity_count_matches, }, "activity_count_mismatch_analysis": { "current_categories": current_categories, "current_activity_ids": current_ids, "current_activity_ids_sha256": _sha256(current_ids), "source_projection_at_fixed_watermarks": projection, "source_history": history, "historical_activity_ids_absent_from_domain": historical_missing_ids, "historical_activity_ids_absent_from_domain_sha256": _sha256( historical_missing_ids ), "deleted_rows_are_not_countable_from_domain_table": True, "deletion_or_departure_is_attributed_only_from_durable_outbox": True, "historical_46_baseline_claimed_count": 46, "historical_46_baseline_attribution_proven": any( item["term_live_activity_count"] == 46 for item in projection ), "ui_46_explanation": ( "activity_template_count" if current_categories["activity_template_count"] == 46 else "not_explained_by_current_admin_activity_list_or_templates" ), }, "overlapping_term_candidates": [ { **candidate, "start_date": candidate["start_date"].isoformat(), "end_date": candidate["end_date"].isoformat(), } for candidate in candidates ], "scheduler_settings": {key: int(value) for key, value in settings_map.items()}, "activities": inventory, "summary": { "activity_ids": sorted(scope_ids), "activity_ids_sha256": _sha256(sorted(scope_ids)), "scheduled_count": len(scheduled_ids), "scheduled_activity_ids": scheduled_ids, "unscheduled_count": len(inventory) - len(scheduled_ids), "overwrite_activity_ids": overwrite_ids, "engine_input_blocked_activity_ids": input_blocked_ids, "foreign_scheduled_activity_ids_using_candidate_resources": foreign_conflict_ids, "candidate_resource_universe": { "staff_ids": candidate_staff_ids, "staff_ids_sha256": _sha256(candidate_staff_ids), "location_ids": candidate_location_ids, "location_ids_sha256": _sha256(candidate_location_ids), }, "requirement_type_counts": dict( Counter( f"staff:{item['staff_requirement']['type']}/location:{item['location_requirement']['type']}" for item in inventory ) ), }, "engine_attestation": { "advisory_preschedule_dispatched": False, "terminal_engine_result_observed": False, "input_model_ready_count": len(inventory) - len(input_blocked_ids), "input_model_blocked_count": len(input_blocked_ids), "actual_engine_feasibility_proven": False, "reason": "a read-only audit cannot enqueue an advisory engine request", }, "prospective_transport": { "expected_source_transactions_for_one_atomic_engine_response": 1, "requested_activity_count": len(inventory), "current_variant_member_count": len(variants), "exact_event_member_count": None, "event_member_count_reason": ( "the final member count is fixed only after the engine response is parsed; " "no-slot items and variant create/merge/tombstones change the member set" ), "required_terminal_evidence": [ "engine_request_id", "engine_response_hash", "applied_engine_response_receipt", "change_set_id/source_transaction_id", "final_source_watermark", "affected_activity_count", "absolute_occurrence_count_and_sha256", "staff_location_allocation_sha256", "publisher_last_sequence", "resource_booking_receipt_for_same_source_transaction", ], }, "bulk_schedule_preconditions": { "exact_scope_count": len(inventory) == EXPECTED_ACTIVITY_COUNT, "all_activities_unscheduled": not scheduled_ids, "no_existing_allocations_to_overwrite": not overwrite_ids, "static_engine_inputs_ready": not input_blocked_ids, "foreign_candidate_resource_use_requires_final_slot_conflict_check": bool( foreign_conflict_ids ), "engine_advisory_result_required_before_execution": True, "resource_booking_overlap_attestation_required_before_execution": True, "ready_to_schedule": False, }, "attestation_passed": activity_count_matches, "execution_authorized": False, "mutation_executed": False, "schedule_dispatched": False, "kafka_published": False, "resource_booking_accessed": False, "other_database_accessed": False, "reverse_delivery_enabled": False, "phase2_enabled": False, "dr_executed": False, "failed_load02_reclassified": False, } result["evidence_sha256"] = _sha256(result) return result def main() -> int: app_dir = Path(os.environ.get("TT_PHASE1_APP_DIR", ".")).resolve() if str(app_dir) not in sys.path: sys.path.insert(0, str(app_dir)) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings") import django from django.db import transaction django.setup() try: with transaction.atomic(): result = build_attestation() transaction.set_rollback(True) except Exception as error: details = getattr(error, "details", {}) print( json.dumps( { "term_attestation": "blocked", "reason": str(error), "details": details, "mutation_executed": False, "schedule_dispatched": False, }, sort_keys=True, ) ) return 2 summary = {key: value for key, value in result.items() if key != "activities"} for key in sorted(summary): _emit_bounded_fact([key], summary[key]) for activity in result["activities"]: print( json.dumps( {"term_activity_inventory": activity}, sort_keys=True, separators=(",", ":"), ) ) print( json.dumps( { "term_attestation_complete": True, "activity_records_emitted": len(result["activities"]), "attestation_passed": result["attestation_passed"], "evidence_sha256": result["evidence_sha256"], "mutation_executed": False, "schedule_dispatched": False, }, sort_keys=True, separators=(",", ":"), ) ) return 0 if result["attestation_passed"] else 2 if __name__ == "__main__": raise SystemExit(main())