diff --git a/.github/workflows/windows-prerelease.yml b/.github/workflows/windows-prerelease.yml index ddc9797..2a9013c 100644 --- a/.github/workflows/windows-prerelease.yml +++ b/.github/workflows/windows-prerelease.yml @@ -125,7 +125,11 @@ jobs: if ($actual -ne $expected -or $actual -ne $manifest.setup.sha256) { throw "Setup checksum receipt mismatch." } if ((Get-Item -LiteralPath $setup).Length -ge 2GB) { throw "Setup exceeds GitHub's asset limit." } if (@($manifest.profiles.profile | Sort-Object) -join ',' -ne 'classroom-oneroster,core') { throw "Both profiles were not embedded." } - if ($manifest.gam_version -ne '7.47.02') { throw "Bundled GAM pin mismatch." } + $gamVersionOutput = & .venv\Scripts\python.exe -c "from gamgui.core.gam.commands import EXPECTED_GAM_VERSION; print(EXPECTED_GAM_VERSION)" + $gamVersionExitCode = $LASTEXITCODE + $expectedGamVersion = ($gamVersionOutput | Out-String).Trim() + if ($gamVersionExitCode -ne 0 -or $expectedGamVersion -notmatch '^\d+\.\d+\.\d+$') { throw "Repository GAM source pin is invalid." } + if ([string]$manifest.gam_version -cne $expectedGamVersion) { throw "Bundled GAM pin mismatch." } - name: Exercise fail-closed setup, both profiles, signatures, and uninstall shell: pwsh run: | diff --git a/gamgui/components/oneroster/executor.py b/gamgui/components/oneroster/executor.py index a39b314..4b02bed 100644 --- a/gamgui/components/oneroster/executor.py +++ b/gamgui/components/oneroster/executor.py @@ -26,11 +26,14 @@ GateState, ImportAction, LivePlanningResult, + ManagedCourseDirty, + ManagedCourseVerification, OneRosterError, StudentEnrollmentGate, canonical_hash, ) from .planner import OneRosterPlanner +from .semantic import metadata_hash, student_hash, teacher_hash from .store import OneRosterStore, action_sequence_hash @@ -93,6 +96,12 @@ def observe( self.clean_batches = 0 +@dataclass(frozen=True) +class _VerificationRead: + results: Mapping[str, bool] + verifications: tuple[ManagedCourseVerification, ...] + + class ExecutableClassroomConnector(Protocol): async def run_classroom_batch( self, @@ -209,6 +218,7 @@ async def execute( manifest.import_id, limited_import=manifest.plan_kind == "limited", now=now, + required_metadata_aliases=tuple(manifest.live_evidence), ) self.store.record_execution_planning_receipt( run.id, @@ -403,6 +413,7 @@ async def observe() -> LivePlanningResult: manifest.import_id, limited_import=manifest.plan_kind == "limited", now=now, + required_metadata_aliases=tuple(manifest.live_evidence), ) first = await observe() @@ -474,6 +485,7 @@ async def revalidate_scheduled_gate( manifest.import_id, limited_import=manifest.plan_kind == "limited", now=now, + required_metadata_aliases=tuple(manifest.live_evidence), ) await self._validate_preflight( manifest, @@ -535,7 +547,22 @@ async def reconcile_interrupted( actions = self.store.get_manifest_actions_by_id( manifest_id, batch.action_ids ) - verified = await self._verify(actions) + verification_read = await self._verify(actions) + verified = verification_read.results + self.store.record_managed_course_verifications( + verification_read.verifications + ) + self.store.mark_managed_course_dirty( + _dirty_for_actions( + tuple( + action + for action in actions + if not verified.get(action.id, False) + ) + ), + error_code="OR-RECOVERY-VERIFY-INCOMPLETE", + recovery_required=True, + ) for action in actions: if not verified.get(action.id, False): continue @@ -910,6 +937,11 @@ async def _apply_chunk( durable_batch.id, worker_count=worker_count, ) + self.store.mark_managed_course_dirty( + _dirty_for_actions(chunk), + error_code="OR-EXECUTION-UNCERTAIN", + recovery_required=True, + ) commands = [_command_for(action) for action in chunk] batch_failed = False throttling_count = 0 @@ -951,8 +983,12 @@ async def pulse() -> None: if batch_receipt is not None: throttling_count = int(batch_receipt.throttling_count) try: - verified, verification_attempts, verification_seconds = ( - await self._verify_with_retries(chunk, owner_ids) + verified, verification_attempts, verification_seconds, verifications = ( + await self._verify_with_retries( + chunk, + owner_ids, + include_verifications=True, + ) ) applied = guarded_applied failed = guarded_failed @@ -980,6 +1016,9 @@ async def pulse() -> None: }: blocked_courses.add(action.subject.casefold()) if durable_batch is not None: + dirty = _dirty_for_actions( + tuple(action for action in chunk if not verified.get(action.id)) + ) completed_batch = self.store.complete_verified_batch( durable_batch.id, manifest_id, @@ -990,10 +1029,19 @@ async def pulse() -> None: verification_attempts=verification_attempts, worker_count=worker_count, throttling_count=throttling_count, + managed_verifications=verifications, + managed_dirty=dirty, ) persistence_seconds = completed_batch.persistence_seconds else: persistence_started = time.perf_counter() + self.store.record_managed_course_verifications(verifications) + self.store.mark_managed_course_dirty( + _dirty_for_actions( + tuple(action for action in chunk if not verified.get(action.id)) + ), + error_code="OR-BATCH-VERIFY-FAILED", + ) for action_id, (status, detail) in results.items(): self.store.mark_action_result( manifest_id, @@ -1076,10 +1124,13 @@ async def _verify_with_retries( self, actions: Sequence[ImportAction], owner_ids: Optional[Mapping[str, str]], - ) -> tuple[dict[str, bool], int, float]: + *, + include_verifications: bool = False, + ) -> Any: started = time.perf_counter() remaining = list(actions) verified = {action.id: False for action in actions} + verification_by_alias: dict[str, ManagedCourseVerification] = {} attempts = 0 retry_offsets = (0.0, *self.stabilization_delays) for retry_offset in retry_offsets: @@ -1088,16 +1139,37 @@ async def _verify_with_retries( wait_seconds = retry_offset - (time.perf_counter() - started) if wait_seconds > 0: await asyncio.sleep(wait_seconds) - result = await self._verify(remaining, owner_ids) + raw_read = await self._verify(remaining, owner_ids) + if isinstance(raw_read, _VerificationRead): + read_results = raw_read.results + read_verifications = raw_read.verifications + else: + # Keep the focused private-method test seam compatible with + # mapping-only verification fakes. + read_results = raw_read + read_verifications = () + for verification in read_verifications: + key = verification.alias.casefold() + verification_by_alias[key] = _merge_verification( + verification_by_alias.get(key), + verification, + ) attempts += 1 next_remaining: list[ImportAction] = [] for action in remaining: - if bool(result.get(action.id)): + if bool(read_results.get(action.id)): verified[action.id] = True else: next_remaining.append(action) remaining = next_remaining - return verified, attempts, time.perf_counter() - started + result = ( + verified, + attempts, + time.perf_counter() - started, + ) + if include_verifications: + return (*result, tuple(verification_by_alias.values())) + return result def _record_performance_audit( self, @@ -1221,7 +1293,7 @@ async def _verify( self, actions: Sequence[ImportAction], owner_ids: Optional[Mapping[str, str]] = None, - ) -> dict[str, bool]: + ) -> _VerificationRead: aliases = sorted({action.subject for action in actions}) bulk_courses = getattr( self.connector, @@ -1262,12 +1334,21 @@ async def read(alias: str) -> tuple[str, Optional[Any]]: return alias.casefold(), detail details = dict(await asyncio.gather(*(read(alias) for alias in aliases))) - roster_aliases = { + teacher_aliases = { + action.subject.casefold() + for action in actions + if action.kind in {"teacher_add", "teacher_remove"} + } + student_aliases = { action.subject.casefold() for action in actions - if action.kind in {"teacher_add", "teacher_remove", "student_add", "student_remove"} + if action.kind in {"student_add", "student_remove"} } - rosters: dict[str, tuple[set[str], set[str]]] = {} + roster_aliases = teacher_aliases | student_aliases + rosters: dict[ + str, + tuple[Optional[set[str]], Optional[set[str]]], + ] = {} roster_details = { alias: detail for alias in sorted(roster_aliases) @@ -1280,42 +1361,63 @@ async def read(alias: str) -> tuple[str, Optional[Any]]: None, ) if roster_details and callable(bulk_rosters): - requested_course_ids = [ - _text(detail, "id") for detail in roster_details.values() - ] - try: - participants = await bulk_rosters( - requested_course_ids, - "all", - ) - if ( - not isinstance(participants, CourseRosterSnapshot) - or not participants.covers(requested_course_ids) - ): + by_course = { + _text(detail, "id"): alias + for alias, detail in roster_details.items() + } + both_ids = sorted( + course_id + for course_id, alias in by_course.items() + if alias in teacher_aliases and alias in student_aliases + ) + teacher_ids = sorted( + course_id + for course_id, alias in by_course.items() + if alias in teacher_aliases and alias not in student_aliases + ) + student_ids = sorted( + course_id + for course_id, alias in by_course.items() + if alias in student_aliases and alias not in teacher_aliases + ) + for requested_course_ids, role in ( + (both_ids, "all"), + (teacher_ids, "teachers"), + (student_ids, "students"), + ): + if not requested_course_ids: + continue + try: + participants = await bulk_rosters(requested_course_ids, role) + complete = bool( + isinstance(participants, CourseRosterSnapshot) + and participants.covers(requested_course_ids) + ) + except Exception: + complete = False participants = None - except Exception: - participants = None - if participants is not None: - by_course = { - _text(detail, "id"): alias - for alias, detail in roster_details.items() - } - rosters.update( - { - alias: ( - set(participants.for_course(course_id)[0]), - set(participants.for_course(course_id)[1]), - ) - for course_id, alias in by_course.items() - } - ) + if not complete or participants is None: + continue + for course_id in requested_course_ids: + alias = by_course[course_id] + teachers, students = participants.for_course(course_id) + rosters[alias] = ( + set(teachers) if role in {"all", "teachers"} else None, + set(students) if role in {"all", "students"} else None, + ) elif roster_details: for alias, detail in roster_details.items(): course_id = _text(detail, "id") + teachers_needed = alias in teacher_aliases + students_needed = alias in student_aliases try: teachers, students = await asyncio.gather( - self.connector.list_course_participants(course_id, "teachers"), - self.connector.list_course_participants(course_id, "students"), + self.connector.list_course_participants(course_id, "teachers") + if teachers_needed + else asyncio.sleep(0, result=()), + self.connector.list_course_participants(course_id, "students") + if students_needed + else asyncio.sleep(0, result=()), ) except Exception: continue @@ -1326,18 +1428,124 @@ async def read(alias: str) -> tuple[str, Optional[Any]]: # A course-only GAM row is not proof of a complete roster. continue rosters[alias] = ( - {_participant_email(item) for item in teachers if _participant_email(item)}, - {_participant_email(item) for item in students if _participant_email(item)}, + {_participant_email(item) for item in teachers if _participant_email(item)} + if teachers_needed + else None, + {_participant_email(item) for item in students if _participant_email(item)} + if students_needed + else None, ) - return { - action.id: _verify_action( + action_results: dict[str, bool] = {} + for action in actions: + alias = action.subject.casefold() + roster = rosters.get(alias) + if action.kind in {"teacher_add", "teacher_remove"} and ( + roster is None or roster[0] is None + ): + action_roster = None + elif action.kind in {"student_add", "student_remove"} and ( + roster is None or roster[1] is None + ): + action_roster = None + else: + action_roster = ( + set(roster[0] or ()), + set(roster[1] or ()), + ) if roster is not None else None + action_results[action.id] = _verify_action( action, - details.get(action.subject.casefold()), - rosters.get(action.subject.casefold()), + details.get(alias), + action_roster, owner_ids or {}, ) - for action in actions + + states = self.store.get_managed_course_states(aliases) + kinds_by_alias: dict[str, set[str]] = {} + for action in actions: + kinds_by_alias.setdefault(action.subject.casefold(), set()).add(action.kind) + owners_by_id = { + str(user_id): str(email).casefold() + for email, user_id in (owner_ids or {}).items() + if str(user_id) + } + verifications: list[ManagedCourseVerification] = [] + metadata_kinds = { + "course_create", + "course_update", + "course_activate", + "course_archive", + "owner_transfer", } + for alias in aliases: + key = alias.casefold() + if key not in states: + continue + detail = details.get(key) + if detail is None or not _has_alias(detail, alias): + continue + course_id = _text(detail, "id") + if not course_id: + continue + kinds = kinds_by_alias.get(key, set()) + metadata_needed = bool(kinds & metadata_kinds) + teachers_needed = bool(kinds & {"teacher_add", "teacher_remove"}) + students_needed = bool(kinds & {"student_add", "student_remove"}) + roster = rosters.get(key) + owner_email = _text(detail, "owner_email", "ownerEmail").casefold() + if not owner_email: + owner_email = owners_by_id.get( + _text(detail, "owner_id", "ownerId"), + "", + ) + state = _text(detail, "course_state", "courseState").upper() + metadata_covered = bool(metadata_needed and owner_email and state) + teacher_members = ( + tuple(sorted(roster[0])) + if teachers_needed and roster is not None and roster[0] is not None + else None + ) + student_members = ( + tuple(sorted(roster[1])) + if students_needed and roster is not None and roster[1] is not None + else None + ) + if not (metadata_covered or teacher_members is not None or student_members is not None): + continue + verifications.append( + ManagedCourseVerification( + alias=alias, + course_id=course_id, + metadata_hash=( + metadata_hash( + alias, + _text(detail, "name"), + owner_email, + _text(detail, "room"), + _text(detail, "section"), + state, + ) + if metadata_covered + else None + ), + teacher_hash=( + teacher_hash(teacher_members) + if teacher_members is not None + else None + ), + student_hash=( + student_hash(student_members) + if student_members is not None + else None + ), + teacher_members=teacher_members, + student_members=student_members, + course_state=state if metadata_covered else None, + ) + ) + return _VerificationRead( + results=action_results, + verifications=tuple(verifications), + ) def _command_for(action: ImportAction) -> list[str]: @@ -1440,6 +1648,77 @@ def _verify_action( return False +def _dirty_for_actions( + actions: Sequence[ImportAction], +) -> tuple[ManagedCourseDirty, ...]: + by_alias: dict[str, dict[str, bool]] = {} + metadata_kinds = { + "course_create", + "course_update", + "course_activate", + "course_archive", + "owner_transfer", + } + for action in actions: + scopes = by_alias.setdefault( + action.subject, + {"metadata": False, "teachers": False, "students": False}, + ) + if action.kind in metadata_kinds: + scopes["metadata"] = True + elif action.kind in {"teacher_add", "teacher_remove"}: + scopes["teachers"] = True + elif action.kind in {"student_add", "student_remove"}: + scopes["students"] = True + return tuple( + ManagedCourseDirty(alias=alias, **scopes) + for alias, scopes in sorted(by_alias.items(), key=lambda item: item[0].casefold()) + if any(scopes.values()) + ) + + +def _merge_verification( + previous: Optional[ManagedCourseVerification], + current: ManagedCourseVerification, +) -> ManagedCourseVerification: + if previous is None: + return current + return ManagedCourseVerification( + alias=current.alias, + course_id=current.course_id or previous.course_id, + source_class_id=current.source_class_id or previous.source_class_id, + last_seen_import_id=current.last_seen_import_id or previous.last_seen_import_id, + metadata_hash=( + current.metadata_hash + if current.metadata_hash is not None + else previous.metadata_hash + ), + teacher_hash=( + current.teacher_hash + if current.teacher_hash is not None + else previous.teacher_hash + ), + student_hash=( + current.student_hash + if current.student_hash is not None + else previous.student_hash + ), + teacher_members=( + current.teacher_members + if current.teacher_members is not None + else previous.teacher_members + ), + student_members=( + current.student_members + if current.student_members is not None + else previous.student_members + ), + course_state=current.course_state or previous.course_state, + verified_at=max(previous.verified_at, current.verified_at), + clear_recovery=previous.clear_recovery and current.clear_recovery, + ) + + def _gate_allows( gate: StudentEnrollmentGate, manifest: ClassroomImportManifest, diff --git a/gamgui/components/oneroster/models.py b/gamgui/components/oneroster/models.py index 12f53c0..f4a4b7b 100644 --- a/gamgui/components/oneroster/models.py +++ b/gamgui/components/oneroster/models.py @@ -463,6 +463,11 @@ class LivePlanningResult: compare=False, repr=False, ) + desired_course_hashes: Mapping[str, "ManagedCourseDesired"] = field( + default_factory=dict, + compare=False, + repr=False, + ) def actions_for(self, plan_kind: str) -> Tuple[ImportAction, ...]: kind = str(plan_kind or "").strip().casefold() @@ -608,6 +613,104 @@ class PlanningPerformanceReceipt: directory_snapshot_seconds: float = 0.0 classroom_snapshot_seconds: float = 0.0 roster_snapshot_seconds: float = 0.0 + total_managed_aliases: int = 0 + candidate_aliases: int = 0 + unchanged_aliases: int = 0 + metadata_reads_requested: int = 0 + teacher_rosters_requested: int = 0 + student_rosters_requested: int = 0 + cached_metadata_scopes: int = 0 + cached_teacher_scopes: int = 0 + cached_student_scopes: int = 0 + audit_courses_requested: int = 0 + metadata_chunk_count: int = 0 + teacher_roster_chunk_count: int = 0 + student_roster_chunk_count: int = 0 + completed_chunk_count: int = 0 + retried_chunk_count: int = 0 + failed_chunk_count: int = 0 + rate_limit_count: int = 0 + timeout_retry_count: int = 0 + incomplete_coverage_count: int = 0 + latency_regression_count: int = 0 + largest_metadata_chunk: int = 0 + largest_roster_chunk: int = 0 + maximum_observed_read_concurrency: int = 0 + final_recommended_read_concurrency: int = 1 + metadata_final_read_concurrency: int = 1 + teacher_final_read_concurrency: int = 1 + student_final_read_concurrency: int = 1 + metadata_chunk_worker_levels: Tuple[int, ...] = () + teacher_roster_chunk_worker_levels: Tuple[int, ...] = () + student_roster_chunk_worker_levels: Tuple[int, ...] = () + + +@dataclass(frozen=True) +class ManagedCourseState: + """Durable, independently scoped verification state for one managed alias.""" + + domain: str + alias: str + course_id: str + source_class_id: str + last_seen_import_id: str + desired_metadata_hash: str = "" + desired_teacher_hash: str = "" + desired_student_hash: str = "" + verified_metadata_hash: str = "" + verified_teacher_hash: str = "" + verified_student_hash: str = "" + last_metadata_verified_at: float = 0.0 + last_teacher_verified_at: float = 0.0 + last_student_verified_at: float = 0.0 + verified_course_state: str = "" + metadata_dirty: bool = True + teacher_roster_dirty: bool = True + student_roster_dirty: bool = True + recovery_required: bool = False + last_error_code: str = "" + version: int = 1 + updated_at: float = 0.0 + + +@dataclass(frozen=True) +class ManagedCourseDesired: + """Accepted desired hashes; recording these never establishes live proof.""" + + alias: str + import_id: str + metadata_hash: str + teacher_hash: str + student_hash: str + source_class_id: str = "" + + +@dataclass(frozen=True) +class ManagedCourseVerification: + """Fresh live evidence for only the explicitly populated verification scopes.""" + + alias: str + course_id: str + source_class_id: str = "" + last_seen_import_id: str = "" + metadata_hash: Optional[str] = None + teacher_hash: Optional[str] = None + student_hash: Optional[str] = None + teacher_members: Optional[Tuple[str, ...]] = None + student_members: Optional[Tuple[str, ...]] = None + course_state: Optional[str] = None + verified_at: float = 0.0 + clear_recovery: bool = True + + +@dataclass(frozen=True) +class ManagedCourseDirty: + """Exact registry scopes whose live outcome is incomplete or uncertain.""" + + alias: str + metadata: bool = False + teachers: bool = False + students: bool = False @dataclass(frozen=True) diff --git a/gamgui/components/oneroster/planner.py b/gamgui/components/oneroster/planner.py index 5a90bd2..f83aae5 100644 --- a/gamgui/components/oneroster/planner.py +++ b/gamgui/components/oneroster/planner.py @@ -11,15 +11,20 @@ import asyncio import hashlib import json +import random import sqlite3 import time from collections import Counter, defaultdict from contextlib import closing -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime -from typing import Any, Iterable, Mapping, Optional, Protocol, Sequence +from typing import Any, Awaitable, Callable, Iterable, Mapping, Optional, Protocol, Sequence from gamgui.core.classroom.models import CourseRosterSnapshot +from gamgui.core.connectors.gam_connector import ( + ONEROSTER_MANAGED_ALIAS_CHUNK_CAP, + ONEROSTER_ROSTER_CHUNK_CAP, +) from gamgui.core.gam.errors import GAMError, GAMErrorKind from .ingest import district_roster_date @@ -28,10 +33,15 @@ ImportIssue, IssueSeverity, LivePlanningResult, + ManagedCourseDesired, + ManagedCourseDirty, + ManagedCourseState, + ManagedCourseVerification, OneRosterError, PlanningPerformanceReceipt, canonical_hash, ) +from .semantic import metadata_hash, student_hash, teacher_hash from .store import OneRosterStore from .thresholds import evaluate_thresholds @@ -39,6 +49,14 @@ PLANNER_SCHEMA_VERSION = 3 DIRECTORY_CONCURRENCY = 12 COURSE_CONCURRENCY = 8 +DEFAULT_AUDIT_SAMPLE_SIZE = 25 +ONEROSTER_READ_MAX_CONCURRENCY = 4 +ONEROSTER_READ_MAX_ATTEMPTS = 3 +ONEROSTER_READ_BACKOFF_BASE_SECONDS = 1.0 +ONEROSTER_READ_BACKOFF_CAP_SECONDS = 8.0 +ONEROSTER_READ_BACKOFF_JITTER_RATIO = 0.25 +ONEROSTER_READ_LATENCY_REGRESSION_FACTOR = 2.0 +ONEROSTER_READ_LATENCY_REGRESSION_FLOOR_SECONDS = 1.0 # The current planner uses compact in-memory participant/action models. Keep # district planning comfortably below the measured memory cliff; larger valid # snapshots remain available for inspection and CSV export. @@ -93,6 +111,17 @@ class _LiveCourse: students: tuple[str, ...] = () owner_email: str = "" unresolved_members: tuple[str, ...] = () + metadata_loaded: bool = True + teachers_loaded: bool = True + students_loaded: bool = True + + +@dataclass(frozen=True, slots=True) +class _CourseReadIntent: + metadata: bool + teachers: bool + students: bool + reasons: tuple[str, ...] @dataclass(frozen=True, slots=True) @@ -107,6 +136,109 @@ class _ActionPayload: live_evidence: Mapping[str, Any] +class _IncompleteReadCoverage(ValueError): + """A bounded read returned data that cannot prove its requested scope.""" + + +@dataclass(frozen=True, slots=True) +class _CompletedReadChunk: + value: Any + duration_seconds: float + clean: bool + + +@dataclass(slots=True) +class _ReadTuner: + maximum: int + level: int = 1 + clean_streak: int = 0 + baseline_seconds: Optional[float] = None + + def penalize(self) -> None: + self.level = max(1, self.level - 1) + self.clean_streak = 0 + + def completed_cleanly(self, duration_seconds: float) -> bool: + duration = max(0.0, float(duration_seconds)) + regressed = bool( + self.baseline_seconds is not None + and duration + > max( + ONEROSTER_READ_LATENCY_REGRESSION_FLOOR_SECONDS, + self.baseline_seconds * ONEROSTER_READ_LATENCY_REGRESSION_FACTOR, + ) + ) + if regressed: + self.penalize() + else: + self.clean_streak += 1 + if self.clean_streak >= 2 and self.level < self.maximum: + self.level += 1 + self.clean_streak = 0 + self.baseline_seconds = ( + duration + if self.baseline_seconds is None + else (self.baseline_seconds * 0.75) + (duration * 0.25) + ) + return regressed + + +@dataclass(slots=True) +class _ReadPerformance: + metadata_chunk_count: int = 0 + teacher_roster_chunk_count: int = 0 + student_roster_chunk_count: int = 0 + completed_chunk_count: int = 0 + retried_chunk_count: int = 0 + failed_chunk_count: int = 0 + rate_limit_count: int = 0 + timeout_retry_count: int = 0 + incomplete_coverage_count: int = 0 + latency_regression_count: int = 0 + largest_metadata_chunk: int = 0 + largest_roster_chunk: int = 0 + active_chunks: int = 0 + maximum_observed_read_concurrency: int = 0 + metadata_final_read_concurrency: int = 1 + teacher_final_read_concurrency: int = 1 + student_final_read_concurrency: int = 1 + metadata_chunk_worker_levels: list[int] = field(default_factory=list) + teacher_roster_chunk_worker_levels: list[int] = field(default_factory=list) + student_roster_chunk_worker_levels: list[int] = field(default_factory=list) + + def scheduled(self, category: str, size: int, worker_level: int) -> None: + if category == "metadata": + self.metadata_chunk_count += 1 + self.largest_metadata_chunk = max(self.largest_metadata_chunk, size) + self.metadata_chunk_worker_levels.append(worker_level) + elif category == "teachers": + self.teacher_roster_chunk_count += 1 + self.largest_roster_chunk = max(self.largest_roster_chunk, size) + self.teacher_roster_chunk_worker_levels.append(worker_level) + else: + self.student_roster_chunk_count += 1 + self.largest_roster_chunk = max(self.largest_roster_chunk, size) + self.student_roster_chunk_worker_levels.append(worker_level) + + def set_final_level(self, category: str, level: int) -> None: + if category == "metadata": + self.metadata_final_read_concurrency = level + elif category == "teachers": + self.teacher_final_read_concurrency = level + else: + self.student_final_read_concurrency = level + + def final_recommendation(self) -> int: + levels: list[int] = [] + if self.metadata_chunk_count: + levels.append(self.metadata_final_read_concurrency) + if self.teacher_roster_chunk_count: + levels.append(self.teacher_final_read_concurrency) + if self.student_roster_chunk_count: + levels.append(self.student_final_read_concurrency) + return min(levels) if levels else 1 + + def planner_configuration_hash( *, limited_import: bool, @@ -135,11 +267,85 @@ def __init__( *, directory_concurrency: int = DIRECTORY_CONCURRENCY, course_concurrency: int = COURSE_CONCURRENCY, + audit_sample_size: Optional[int] = None, + read_max_concurrency: Optional[int] = None, + read_max_attempts: Optional[int] = None, + read_backoff_base_seconds: Optional[float] = None, + read_backoff_cap_seconds: Optional[float] = None, + read_sleep: Optional[Callable[[float], Awaitable[None]]] = None, + read_jitter: Optional[Callable[[], float]] = None, ) -> None: self.store = store self.connector = connector self.directory_concurrency = max(1, min(int(directory_concurrency), 32)) self.course_concurrency = max(1, min(int(course_concurrency), 16)) + configured_audit = ( + getattr(connector, "oneroster_audit_sample_size", DEFAULT_AUDIT_SAMPLE_SIZE) + if audit_sample_size is None + else audit_sample_size + ) + self.audit_sample_size = max(0, min(int(configured_audit), 25)) + configured_read_concurrency = ( + getattr( + connector, + "oneroster_read_max_concurrency", + ONEROSTER_READ_MAX_CONCURRENCY, + ) + if read_max_concurrency is None + else read_max_concurrency + ) + configured_read_attempts = ( + getattr( + connector, + "oneroster_read_max_attempts", + ONEROSTER_READ_MAX_ATTEMPTS, + ) + if read_max_attempts is None + else read_max_attempts + ) + configured_backoff_base = ( + getattr( + connector, + "oneroster_read_backoff_base_seconds", + ONEROSTER_READ_BACKOFF_BASE_SECONDS, + ) + if read_backoff_base_seconds is None + else read_backoff_base_seconds + ) + configured_backoff_cap = ( + getattr( + connector, + "oneroster_read_backoff_cap_seconds", + ONEROSTER_READ_BACKOFF_CAP_SECONDS, + ) + if read_backoff_cap_seconds is None + else read_backoff_cap_seconds + ) + self.read_max_concurrency = max( + 1, + min(int(configured_read_concurrency), ONEROSTER_READ_MAX_CONCURRENCY), + ) + self.read_max_attempts = max(1, min(int(configured_read_attempts), 6)) + self.read_backoff_base_seconds = max( + 0.0, + min(float(configured_backoff_base), 60.0), + ) + self.read_backoff_cap_seconds = max( + self.read_backoff_base_seconds, + min(float(configured_backoff_cap), 120.0), + ) + self._read_sleep = read_sleep or getattr( + connector, + "oneroster_read_sleep", + asyncio.sleep, + ) + self._read_jitter = read_jitter or getattr( + connector, + "oneroster_read_jitter", + random.random, + ) + self._read_performance = _ReadPerformance() + self._checkpointed_scopes: set[tuple[str, str]] = set() async def plan( self, @@ -147,8 +353,11 @@ async def plan( *, limited_import: bool = False, now: Optional[datetime] = None, + required_metadata_aliases: Sequence[str] = (), ) -> LivePlanningResult: planning_started = time.perf_counter() + self._read_performance = _ReadPerformance() + self._checkpointed_scopes = set() snapshot = self.store.refresh_schedule_scope( import_id, today=district_roster_date(now), @@ -183,7 +392,6 @@ async def plan( protected_aliases = { alias.casefold() for alias in protected_alias_values } - previous_id = self.store.previous_accepted_import_id(import_id) previous_aliases = self.store.previous_accepted_aliases(import_id) relevant_aliases = tuple( sorted( @@ -191,24 +399,43 @@ async def plan( key=str.casefold, ) ) + managed_states = self.store.get_managed_course_states(relevant_aliases) + bootstrap_metadata_aliases = tuple( + alias + for alias in relevant_aliases + if (state := managed_states.get(alias.casefold())) is None + or not state.course_id + or state.metadata_dirty + or state.recovery_required + ) async def timed_directory_snapshot() -> tuple[Optional[dict[str, Any]], float]: started = time.perf_counter() value = await self._read_directory_snapshot() return value, time.perf_counter() - started - async def timed_classroom_snapshot() -> tuple[Optional[dict[str, Any]], float]: + async def timed_bootstrap_snapshot() -> tuple[Optional[dict[str, Any]], float]: started = time.perf_counter() - value = await self._read_managed_course_snapshot(relevant_aliases) + value = await self._read_managed_course_snapshot( + bootstrap_metadata_aliases + ) return value, time.perf_counter() - started - # Independent live reads use separate private GAMCFGDIRs. GAMRunner - # serializes only refreshed-token persistence, not the read processes. - directory_result, classroom_result = await asyncio.gather( - timed_directory_snapshot(), - timed_classroom_snapshot(), + bootstrap_read_concurrently = bool( + 0 < len(bootstrap_metadata_aliases) <= ONEROSTER_MANAGED_ALIAS_CHUNK_CAP ) - directory_snapshot, directory_snapshot_seconds = directory_result - managed_courses, classroom_snapshot_seconds = classroom_result + if bootstrap_read_concurrently: + directory_result, bootstrap_result = await asyncio.gather( + timed_directory_snapshot(), + timed_bootstrap_snapshot(), + ) + directory_snapshot, directory_snapshot_seconds = directory_result + bootstrap_courses, bootstrap_snapshot_seconds = bootstrap_result + else: + directory_snapshot, directory_snapshot_seconds = ( + await timed_directory_snapshot() + ) + bootstrap_courses = {} + bootstrap_snapshot_seconds = 0.0 resolved, resolution_issues = await self._resolve_directory( desired, directory_snapshot, @@ -222,11 +449,149 @@ async def timed_classroom_snapshot() -> tuple[Optional[dict[str, Any]], float]: ) issues.extend(course_issues) + desired_states = _desired_course_states(eligible, import_id) + read_plan, unchanged_aliases = _build_course_read_plan( + eligible, + desired_states, + managed_states, + previous_aliases=previous_aliases, + protected_aliases=protected_alias_values, + ) + audit_aliases = self.store.select_managed_audit_aliases( + tuple( + course.alias + for course in eligible + if course.alias.casefold() in unchanged_aliases + ), + limit=self.audit_sample_size, + ) + for alias in audit_aliases: + key = alias.casefold() + read_plan[key] = _CourseReadIntent( + metadata=True, + teachers=True, + students=True, + reasons=("bounded_audit",), + ) + unchanged_aliases.discard(key) + required_metadata_keys = { + key + for alias in required_metadata_aliases + if (key := _managed_alias_key(alias)) + } + for alias in relevant_aliases: + key = alias.casefold() + if key not in required_metadata_keys: + continue + current = read_plan.get( + key, + _CourseReadIntent(False, False, False, ()), + ) + read_plan[key] = _CourseReadIntent( + metadata=True, + teachers=current.teachers, + students=current.students, + reasons=(*current.reasons, "execution_identity_guard"), + ) + unchanged_aliases.discard(key) + + # Desired hashes establish intent, not live proof. New aliases enter as + # dirty stubs until exact live evidence supplies a stable course ID. + self.store.record_managed_course_desired(tuple(desired_states.values())) + managed_states = self.store.get_managed_course_states(relevant_aliases) + cached_members = self.store.verified_managed_members_many( + tuple(course.alias for course in eligible) + ) + + metadata_aliases = tuple( + alias + for alias in relevant_aliases + if (intent := read_plan.get(alias.casefold())) is not None + and intent.metadata + ) + eligible_by_alias = {course.alias.casefold(): course for course in eligible} + aliases_by_key = {alias.casefold(): alias for alias in relevant_aliases} + known_course_ids: dict[str, str] = {} + if bootstrap_read_concurrently and bootstrap_courses is not None: + await self._persist_metadata_chunk( + import_id, + bootstrap_metadata_aliases, + bootstrap_courses, + eligible_by_alias, + directory_snapshot, + required_metadata_keys, + ) + known_course_ids.update( + (_text(detail, "id"), key) + for key, detail in bootstrap_courses.items() + if _text(detail, "id") + ) + + async def checkpoint_metadata( + chunk: tuple[str, ...], + indexed: Mapping[str, Any], + ) -> Mapping[str, Any]: + for key, detail in indexed.items(): + course_id = _text(detail, "id") + previous_key = known_course_ids.get(course_id) if course_id else None + if previous_key is not None and previous_key != key: + self.store.mark_managed_course_dirty( + tuple( + ManagedCourseDirty(aliases_by_key[item], metadata=True) + for item in (previous_key, key) + ), + error_code="OR-ALIAS-COLLISION", + recovery_required=True, + ) + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one managed alias resolves to the same live Classroom course.", + ) + await self._persist_metadata_chunk( + import_id, + chunk, + indexed, + eligible_by_alias, + directory_snapshot, + required_metadata_keys, + ) + known_course_ids.update( + (_text(detail, "id"), key) + for key, detail in indexed.items() + if _text(detail, "id") + ) + return indexed + + bootstrap_keys = { + alias.casefold() for alias in bootstrap_metadata_aliases + } if bootstrap_read_concurrently else set() + remaining_metadata_aliases = tuple( + alias + for alias in metadata_aliases + if alias.casefold() not in bootstrap_keys + ) + managed_courses = bootstrap_courses + classroom_snapshot_seconds = bootstrap_snapshot_seconds + if remaining_metadata_aliases and managed_courses is not None: + classroom_started = time.perf_counter() + additional_courses = await self._read_managed_course_snapshot( + remaining_metadata_aliases, + on_chunk=checkpoint_metadata, + ) + classroom_snapshot_seconds += time.perf_counter() - classroom_started + if additional_courses is None: + managed_courses = None + else: + managed_courses = {**managed_courses, **additional_courses} roster_started = time.perf_counter() live_courses = await self._read_live_courses( + import_id, eligible, managed_courses, directory_snapshot, + read_plan=read_plan, + managed_states=managed_states, + cached_members=cached_members, ) live_courses, live_resolution_issues = await self._resolve_live_participants( live_courses, @@ -235,6 +600,15 @@ async def timed_classroom_snapshot() -> tuple[Optional[dict[str, Any]], float]: ) issues.extend(live_resolution_issues) roster_snapshot_seconds = time.perf_counter() - roster_started + # Optional legacy connectors without exact bulk capabilities checkpoint + # their bounded fallback reads here. Normal GAM bulk scopes were already + # committed chunk-by-chunk and are skipped by this final compatibility pass. + self._persist_live_verification( + import_id, + eligible, + live_courses, + read_plan, + ) archive_actions, archive_basis, archive_issues = await self._archive_actions( import_id, @@ -293,7 +667,84 @@ async def timed_classroom_snapshot() -> tuple[Optional[dict[str, Any]], float]: directory_snapshot_seconds=directory_snapshot_seconds, classroom_snapshot_seconds=classroom_snapshot_seconds, roster_snapshot_seconds=roster_snapshot_seconds, + total_managed_aliases=len(relevant_aliases), + candidate_aliases=len(read_plan), + unchanged_aliases=len(unchanged_aliases), + metadata_reads_requested=len(metadata_aliases), + teacher_rosters_requested=sum( + intent.teachers for intent in read_plan.values() + ), + student_rosters_requested=sum( + intent.students for intent in read_plan.values() + ), + cached_metadata_scopes=sum( + not bool( + (intent := read_plan.get(course.alias.casefold())) + and intent.metadata + ) + for course in eligible + ), + cached_teacher_scopes=sum( + not bool( + (intent := read_plan.get(course.alias.casefold())) + and intent.teachers + ) + for course in eligible + ), + cached_student_scopes=sum( + not bool( + (intent := read_plan.get(course.alias.casefold())) + and intent.students + ) + for course in eligible + ), + audit_courses_requested=len(audit_aliases), + metadata_chunk_count=self._read_performance.metadata_chunk_count, + teacher_roster_chunk_count=( + self._read_performance.teacher_roster_chunk_count + ), + student_roster_chunk_count=( + self._read_performance.student_roster_chunk_count + ), + completed_chunk_count=self._read_performance.completed_chunk_count, + retried_chunk_count=self._read_performance.retried_chunk_count, + failed_chunk_count=self._read_performance.failed_chunk_count, + rate_limit_count=self._read_performance.rate_limit_count, + timeout_retry_count=self._read_performance.timeout_retry_count, + incomplete_coverage_count=( + self._read_performance.incomplete_coverage_count + ), + latency_regression_count=( + self._read_performance.latency_regression_count + ), + largest_metadata_chunk=self._read_performance.largest_metadata_chunk, + largest_roster_chunk=self._read_performance.largest_roster_chunk, + maximum_observed_read_concurrency=( + self._read_performance.maximum_observed_read_concurrency + ), + final_recommended_read_concurrency=( + self._read_performance.final_recommendation() + ), + metadata_final_read_concurrency=( + self._read_performance.metadata_final_read_concurrency + ), + teacher_final_read_concurrency=( + self._read_performance.teacher_final_read_concurrency + ), + student_final_read_concurrency=( + self._read_performance.student_final_read_concurrency + ), + metadata_chunk_worker_levels=tuple( + self._read_performance.metadata_chunk_worker_levels + ), + teacher_roster_chunk_worker_levels=tuple( + self._read_performance.teacher_roster_chunk_worker_levels + ), + student_roster_chunk_worker_levels=tuple( + self._read_performance.student_roster_chunk_worker_levels + ), ), + desired_course_hashes=desired_states, ) async def _read_directory_snapshot(self) -> Optional[dict[str, Any]]: bulk = getattr(self.connector, "list_oneroster_directory", None) @@ -313,25 +764,367 @@ async def _read_directory_snapshot(self) -> Optional[dict[str, Any]]: async def _read_managed_course_snapshot( self, aliases: Sequence[str], + *, + on_chunk: Optional[ + Callable[ + [tuple[str, ...], Mapping[str, Any]], + Awaitable[Mapping[str, Any]], + ] + ] = None, ) -> Optional[dict[str, Any]]: + selected_by_key: dict[str, str] = {} + for raw_alias in aliases: + alias = str(raw_alias or "").strip() + if alias.startswith("d:"): + alias = alias[2:] + key = _managed_alias_key(alias) + if not key: + raise OneRosterError( + "OR-ALIAS-INVALID", + "A managed Classroom alias must use the exact Section_ form.", + ) + selected_by_key.setdefault(key, alias) + selected = tuple(sorted(selected_by_key.values(), key=str.casefold)) + if not selected: + return {} bulk = getattr(self.connector, "list_oneroster_managed_courses", None) if not callable(bulk): return None + chunks = tuple( + tuple(selected[offset : offset + ONEROSTER_MANAGED_ALIAS_CHUNK_CAP]) + for offset in range(0, len(selected), ONEROSTER_MANAGED_ALIAS_CHUNK_CAP) + ) + merged: dict[str, Any] = {} + course_ids: dict[str, str] = {} + + async def read_chunk(chunk: tuple[str, ...]) -> Mapping[str, Any]: + raw = await bulk(chunk) + return _index_managed_courses(raw, requested_aliases=chunk) + + async def accept_chunk( + chunk: tuple[str, ...], + indexed: Mapping[str, Any], + ) -> None: + for alias_key, detail in indexed.items(): + course_id = _text(detail, "id") + previous_alias = course_ids.get(course_id) if course_id else None + if previous_alias is not None and previous_alias != alias_key: + affected = tuple( + ManagedCourseDirty(alias, metadata=True) + for alias in ( + selected_by_key[previous_alias], + selected_by_key[alias_key], + ) + ) + self.store.mark_managed_course_dirty( + affected, + error_code="OR-ALIAS-COLLISION", + recovery_required=True, + ) + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one managed alias resolves to the same live Classroom course.", + ) + if course_id: + course_ids[course_id] = alias_key + if on_chunk is not None: + await on_chunk(chunk, indexed) + merged.update(indexed) + + async def fail_chunk(chunk: tuple[str, ...], error_code: str) -> None: + self.store.mark_managed_course_dirty( + tuple(ManagedCourseDirty(alias, metadata=True) for alias in chunk), + error_code=error_code, + ) + + await self._coordinate_read_chunks( + chunks, + category="metadata", + read_chunk=read_chunk, + accept_chunk=accept_chunk, + fail_chunk=fail_chunk, + ) + return merged + + async def _coordinate_read_chunks( + self, + chunks: Sequence[tuple[str, ...]], + *, + category: str, + read_chunk: Callable[[tuple[str, ...]], Awaitable[Any]], + accept_chunk: Callable[[tuple[str, ...], Any], Awaitable[None]], + fail_chunk: Callable[[tuple[str, ...], str], Awaitable[None]], + ) -> None: + tuner = _ReadTuner(self.read_max_concurrency) + offset = 0 try: - raw = await bulk(aliases) - return _index_managed_courses(raw) - except GAMError: - # The exact-set bulk command is an optimization. Fall back to - # bounded per-alias reads, which distinguish absent new courses - # from genuine authentication and permission failures. - return None - except OneRosterError: - raise - except Exception as exc: - raise OneRosterError( + while offset < len(chunks): + worker_level = tuner.level + wave = tuple(chunks[offset : offset + worker_level]) + for chunk in wave: + self._read_performance.scheduled( + category, + len(chunk), + worker_level, + ) + outcomes = await asyncio.gather( + *( + self._attempt_bounded_read(chunk, read_chunk, tuner) + for chunk in wave + ), + return_exceptions=True, + ) + first_failure: Optional[tuple[OneRosterError, BaseException]] = None + for chunk, outcome in zip(wave, outcomes): + if isinstance(outcome, asyncio.CancelledError): + raise outcome + failure: Optional[BaseException] = ( + outcome if isinstance(outcome, BaseException) else None + ) + completed: Optional[_CompletedReadChunk] = ( + outcome if isinstance(outcome, _CompletedReadChunk) else None + ) + if completed is not None: + try: + await accept_chunk(chunk, completed.value) + except BaseException as exc: + if isinstance(exc, asyncio.CancelledError): + raise + failure = exc + if isinstance(exc, _IncompleteReadCoverage): + self._read_performance.incomplete_coverage_count += 1 + tuner.penalize() + if failure is not None: + self._read_performance.failed_chunk_count += 1 + safe_error = self._safe_read_error(failure, category) + await fail_chunk(chunk, safe_error.code) + if first_failure is None: + first_failure = (safe_error, failure) + continue + assert completed is not None + self._read_performance.completed_chunk_count += 1 + if completed.clean and tuner.completed_cleanly( + completed.duration_seconds + ): + self._read_performance.latency_regression_count += 1 + if first_failure is not None: + safe_error, cause = first_failure + if safe_error is cause: + raise safe_error + raise safe_error from cause + offset += len(wave) + finally: + self._read_performance.set_final_level(category, tuner.level) + + async def _attempt_bounded_read( + self, + chunk: tuple[str, ...], + read_chunk: Callable[[tuple[str, ...]], Awaitable[Any]], + tuner: _ReadTuner, + ) -> _CompletedReadChunk: + self._read_performance.active_chunks += 1 + self._read_performance.maximum_observed_read_concurrency = max( + self._read_performance.maximum_observed_read_concurrency, + self._read_performance.active_chunks, + ) + retried = False + try: + for attempt in range(self.read_max_attempts): + started = time.perf_counter() + try: + value = await read_chunk(chunk) + except _IncompleteReadCoverage: + self._read_performance.incomplete_coverage_count += 1 + tuner.penalize() + raise + except GAMError as exc: + retryable = exc.kind in { + GAMErrorKind.RATE_LIMITED, + GAMErrorKind.TIMEOUT, + } + if not retryable: + raise + tuner.penalize() + if exc.kind is GAMErrorKind.RATE_LIMITED: + self._read_performance.rate_limit_count += 1 + if attempt + 1 >= self.read_max_attempts: + raise + if exc.kind is GAMErrorKind.TIMEOUT: + self._read_performance.timeout_retry_count += 1 + if not retried: + retried = True + self._read_performance.retried_chunk_count += 1 + await self._read_sleep(self._read_backoff_delay(attempt)) + continue + return _CompletedReadChunk( + value=value, + duration_seconds=time.perf_counter() - started, + clean=not retried, + ) + raise RuntimeError("bounded OneRoster read exhausted unexpectedly") + finally: + self._read_performance.active_chunks -= 1 + + def _read_backoff_delay(self, retry_index: int) -> float: + jitter = max(0.0, min(float(self._read_jitter()), 1.0)) + exponential = self.read_backoff_base_seconds * (2 ** max(0, retry_index)) + return min( + self.read_backoff_cap_seconds, + exponential * (1.0 + (ONEROSTER_READ_BACKOFF_JITTER_RATIO * jitter)), + ) + + @staticmethod + def _safe_read_error(error: BaseException, category: str) -> OneRosterError: + if isinstance(error, OneRosterError): + return error + if isinstance(error, GAMError): + if error.kind is GAMErrorKind.RATE_LIMITED: + return OneRosterError( + "OR-ONEROSTER-READ-RATE-LIMITED", + "Google continued to rate-limit a bounded OneRoster read; completed chunks were preserved.", + ) + if error.kind is GAMErrorKind.TIMEOUT: + return OneRosterError( + "OR-ONEROSTER-READ-TIMEOUT", + "A bounded OneRoster read timed out after finite retries; completed chunks were preserved.", + ) + if error.kind in { + GAMErrorKind.AUTH_EXPIRED, + GAMErrorKind.NOT_AUTHENTICATED, + }: + return OneRosterError( + "OR-ONEROSTER-READ-AUTH", + "OneRoster live reads require renewed Google authentication.", + ) + if error.kind in { + GAMErrorKind.PERMISSION_DENIED, + GAMErrorKind.SCOPE_MISSING, + }: + return OneRosterError( + "OR-ONEROSTER-READ-PERMISSION", + "The authorized account cannot safely complete OneRoster live reads.", + ) + if category == "metadata": + return OneRosterError( "OR-CLASSROOM-READ", "Live Classroom course resolution failed; no import plan was created.", + ) + return OneRosterError( + "OR-CLASSROOM-ROSTER-READ", + "Live Classroom rosters could not be read safely.", + ) + + async def _persist_metadata_chunk( + self, + import_id: str, + chunk: Sequence[str], + indexed: Mapping[str, Any], + courses_by_alias: Mapping[str, _DesiredCourse], + directory_snapshot: Optional[Mapping[str, Any]], + manifest_guard_aliases: set[str], + ) -> None: + states = self.store.get_managed_course_states(chunk) + aliases_by_key = { + _managed_alias_key(alias): str(alias).removeprefix("d:") + for alias in chunk + } + verifications: list[ManagedCourseVerification] = [] + verified_keys: list[str] = [] + for key, detail in indexed.items(): + alias = aliases_by_key.get(key, "") + state = states.get(key) + if state is None and key not in courses_by_alias: + # A pre-registry accepted alias can still supply fresh archive + # evidence, but there is no durable Branch 1 row to checkpoint. + continue + if not alias or state is None: + raise _IncompleteReadCoverage( + "A returned managed course has no registered exact alias." + ) + course_id = _text(detail, "id") + if state.course_id and state.course_id != course_id: + self.store.mark_managed_course_dirty( + (ManagedCourseDirty(alias, metadata=True),), + error_code="OR-MANAGED-COURSE-ID-DRIFT", + recovery_required=True, + ) + if key in manifest_guard_aliases: + # Executor/gate revalidation must retain the fresh live basis + # so its manifest comparison can persist an exact drift report. + continue + raise OneRosterError( + "OR-MANAGED-COURSE-ID-DRIFT", + "The exact managed alias resolved to a different Classroom course.", + ) + owner_email = await self._normalize_live_identity( + _text(detail, "owner_email", "ownerEmail"), + _text(detail, "owner_id", "ownerId"), + directory_snapshot, + ) + course_state = _text(detail, "course_state", "courseState").upper() + if not ( + course_id + and _has_exact_alias(detail, alias) + and owner_email + and course_state + ): + raise _IncompleteReadCoverage( + "A managed-course metadata chunk returned incomplete exact coverage." + ) + desired_course = courses_by_alias.get(key) + verifications.append( + ManagedCourseVerification( + alias=alias, + course_id=course_id, + source_class_id=( + desired_course.class_id + if desired_course is not None + else state.source_class_id + ), + last_seen_import_id=import_id, + metadata_hash=metadata_hash( + alias, + _text(detail, "name"), + owner_email, + _text(detail, "room"), + _text(detail, "section"), + course_state, + ), + course_state=course_state, + ) + ) + verified_keys.append(key) + self.store.record_managed_course_verifications(tuple(verifications)) + self._checkpointed_scopes.update( + (key, "metadata") for key in verified_keys + ) + + async def _normalize_live_identity( + self, + email: str, + user_id: str, + directory_snapshot: Optional[Mapping[str, Any]], + ) -> str: + email_key = _directory_identifier(email) + id_key = _directory_id_key(user_id) + if directory_snapshot is not None: + return _snapshot_primary( + directory_snapshot.get(email_key) + or directory_snapshot.get(id_key) + ) + reference = email_key or str(user_id or "").strip() + if not reference: + return "" + try: + user = await self.connector.get_user(reference) + except Exception as exc: + if _is_not_found(exc): + return "" + raise OneRosterError( + "OR-DIRECTORY-READ", + "Live Directory resolution failed; no import plan was created.", ) from exc + return _snapshot_primary(user) async def _resolve_directory( self, @@ -375,19 +1168,48 @@ async def resolve(email: str) -> tuple[str, str, str]: async def _read_live_courses( self, + import_id: str, courses: Sequence[_DesiredCourse], managed_courses: Optional[Mapping[str, Any]] = None, directory_snapshot: Optional[Mapping[str, Any]] = None, + *, + read_plan: Optional[Mapping[str, _CourseReadIntent]] = None, + managed_states: Optional[Mapping[str, ManagedCourseState]] = None, + cached_members: Optional[ + Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]] + ] = None, ) -> dict[str, Optional[_LiveCourse]]: + intents = read_plan or {} + states = managed_states or {} + retained_members = cached_members or {} if managed_courses is not None: details = { - course.alias.casefold(): managed_courses.get(course.alias.casefold()) + course.alias.casefold(): ( + managed_courses.get(course.alias.casefold()) + if bool( + (intent := intents.get(course.alias.casefold())) + and intent.metadata + ) + else _cached_course_detail( + course, + states.get(course.alias.casefold()), + ) + ) for course in courses } else: semaphore = asyncio.Semaphore(self.course_concurrency) async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: + intent = intents.get(course.alias.casefold()) + if not intent or not intent.metadata: + return ( + course.alias.casefold(), + _cached_course_detail( + course, + states.get(course.alias.casefold()), + ), + ) async with semaphore: try: detail = await self.connector.get_course( @@ -406,8 +1228,13 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: return course.alias.casefold(), detail details = dict(await asyncio.gather(*(read(course) for course in courses))) - present = [detail for detail in details.values() if detail is not None] - rosters = await self._read_rosters(present) + rosters = await self._read_rosters( + import_id, + courses, + details, + intents, + directory_snapshot, + ) result: dict[str, Optional[_LiveCourse]] = {} for course in courses: key = course.alias.casefold() @@ -416,7 +1243,21 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: result[key] = None continue course_id = _text(detail, "id") - teachers, students = rosters.get(course_id, ((), ())) + cached_teachers, cached_students = retained_members.get(key, ((), ())) + read_teachers, read_students, unresolved_members = rosters.get( + course_id, + (None, None, ()), + ) + teachers = ( + tuple(read_teachers) + if read_teachers is not None + else tuple(cached_teachers) + ) + students = ( + tuple(read_students) + if read_students is not None + else tuple(cached_students) + ) owner_email = _text( detail, "owner_email", @@ -435,6 +1276,12 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: teachers, students, owner_email, + unresolved_members=tuple(unresolved_members), + metadata_loaded=bool( + (intent := intents.get(key)) and intent.metadata + ), + teachers_loaded=bool(intent and intent.teachers), + students_loaded=bool(intent and intent.students), ) result[key] = live return result @@ -488,7 +1335,8 @@ async def resolve(email: str) -> tuple[str, str]: continue unresolved = tuple( sorted( - { + set(live.unresolved_members) + | { email for email in (*live.teachers, *live.students) if not mapping.get(email) @@ -513,68 +1361,507 @@ async def resolve(email: str) -> tuple[str, str]: ), owner_email=mapping.get(live.owner_email, ""), unresolved_members=unresolved, + metadata_loaded=live.metadata_loaded, + teachers_loaded=live.teachers_loaded, + students_loaded=live.students_loaded, ) return normalized, issues async def _read_rosters( self, - details: Sequence[Any], - ) -> dict[str, tuple[tuple[str, ...], tuple[str, ...]]]: - if not details: + import_id: str, + courses: Sequence[_DesiredCourse], + details: Mapping[str, Optional[Any]], + read_plan: Mapping[str, _CourseReadIntent], + directory_snapshot: Optional[Mapping[str, Any]], + ) -> dict[ + str, + tuple[ + Optional[tuple[str, ...]], + Optional[tuple[str, ...]], + tuple[str, ...], + ], + ]: + courses_by_alias = {course.alias.casefold(): course for course in courses} + manifest_guard_aliases = { + alias + for alias, intent in read_plan.items() + if "execution_identity_guard" in intent.reasons + } + requested: dict[ + str, + tuple[str, _DesiredCourse, Any, bool, bool], + ] = {} + for alias, detail in details.items(): + intent = read_plan.get(alias) + course_id = _text(detail, "id") if detail is not None else "" + if not intent or not course_id or not (intent.teachers or intent.students): + continue + course = courses_by_alias.get(alias) + if course is None: + continue + previous = requested.get(course_id) + if previous is not None and previous[0] != alias: + self.store.mark_managed_course_dirty( + ( + ManagedCourseDirty( + previous[1].alias, + teachers=previous[3], + students=previous[4], + ), + ManagedCourseDirty( + course.alias, + teachers=intent.teachers, + students=intent.students, + ), + ), + error_code="OR-ALIAS-COLLISION", + recovery_required=True, + ) + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one managed alias resolves to the same live Classroom course.", + ) + requested[course_id] = ( + alias, + course, + detail, + intent.teachers, + intent.students, + ) + if not requested: return {} - course_ids = [_text(detail, "id") for detail in details if _text(detail, "id")] + teacher_ids = sorted( + course_id + for course_id, (_alias, _course, _detail, teachers, _students) in requested.items() + if teachers + ) + student_ids = sorted( + course_id + for course_id, (_alias, _course, _detail, _teachers, students) in requested.items() + if students + ) + result: dict[ + str, + tuple[ + Optional[tuple[str, ...]], + Optional[tuple[str, ...]], + tuple[str, ...], + ], + ] = {} bulk = getattr(self.connector, "list_course_participants_many", None) if callable(bulk): - try: - participants = await bulk(course_ids, "all") - if ( - not isinstance(participants, CourseRosterSnapshot) - or not participants.covers(course_ids) - ): - raise ValueError( - "The Classroom roster snapshot did not prove complete " - "coverage of the requested courses." + for course_ids, role in ( + (teacher_ids, "teachers"), + (student_ids, "students"), + ): + if not course_ids: + continue + chunks = tuple( + tuple(course_ids[offset : offset + ONEROSTER_ROSTER_CHUNK_CAP]) + for offset in range(0, len(course_ids), ONEROSTER_ROSTER_CHUNK_CAP) + ) + + async def read_chunk( + chunk: tuple[str, ...], + *, + selected_role: str = role, + ) -> Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]]: + participants = await bulk(chunk, selected_role) + requested_ids = frozenset(chunk) + if ( + not isinstance(participants, CourseRosterSnapshot) + or participants.seen_course_ids != requested_ids + or not set(participants.rosters) <= requested_ids + ): + raise _IncompleteReadCoverage( + "The Classroom roster snapshot did not prove exact complete coverage." + ) + normalized: dict[ + str, + tuple[tuple[str, ...], tuple[str, ...]], + ] = {} + for course_id in chunk: + teachers, students = participants.for_course(course_id) + selected_members = ( + teachers if selected_role == "teachers" else students + ) + unexpected_members = ( + students if selected_role == "teachers" else teachers + ) + if unexpected_members: + raise _IncompleteReadCoverage( + "A role-specific Classroom roster read returned another role." + ) + normalized[course_id] = await self._normalize_roster_members( + selected_members, + directory_snapshot, + ) + return normalized + + async def accept_chunk( + chunk: tuple[str, ...], + normalized: Mapping[ + str, + tuple[tuple[str, ...], tuple[str, ...]], + ], + *, + selected_role: str = role, + ) -> None: + self._persist_roster_chunk( + import_id, + selected_role, + chunk, + normalized, + requested, + manifest_guard_aliases, ) - except Exception as exc: - raise OneRosterError( - "OR-CLASSROOM-ROSTER-READ", - "Live Classroom rosters could not be read safely.", - ) from exc - return { - course_id: ( - tuple(sorted(participants.for_course(course_id)[0])), - tuple(sorted(participants.for_course(course_id)[1])), + for course_id in chunk: + members, unresolved = normalized[course_id] + old_teachers, old_students, old_unresolved = result.get( + course_id, + (None, None, ()), + ) + result[course_id] = ( + members if selected_role == "teachers" else old_teachers, + members if selected_role == "students" else old_students, + tuple(sorted({*old_unresolved, *unresolved})), + ) + + async def fail_chunk( + chunk: tuple[str, ...], + error_code: str, + *, + selected_role: str = role, + ) -> None: + self.store.mark_managed_course_dirty( + tuple( + ManagedCourseDirty( + requested[course_id][1].alias, + teachers=selected_role == "teachers", + students=selected_role == "students", + ) + for course_id in chunk + ), + error_code=error_code, + ) + + await self._coordinate_read_chunks( + chunks, + category=role, + read_chunk=read_chunk, + accept_chunk=accept_chunk, + fail_chunk=fail_chunk, ) - for course_id in course_ids - } + return result semaphore = asyncio.Semaphore(self.course_concurrency) - async def read(course_id: str) -> tuple[str, tuple[str, ...], tuple[str, ...]]: + async def read( + course_id: str, + teachers_needed: bool, + students_needed: bool, + ) -> tuple[ + str, + Optional[tuple[str, ...]], + Optional[tuple[str, ...]], + tuple[str, ...], + ]: async with semaphore: try: teachers, students = await asyncio.gather( - self.connector.list_course_participants(course_id, "teachers"), - self.connector.list_course_participants(course_id, "students"), + self.connector.list_course_participants(course_id, "teachers") + if teachers_needed + else asyncio.sleep(0, result=()), + self.connector.list_course_participants(course_id, "students") + if students_needed + else asyncio.sleep(0, result=()), ) except Exception as exc: raise OneRosterError( "OR-CLASSROOM-ROSTER-READ", "Live Classroom rosters could not be read safely.", ) from exc + normalized_teachers: Optional[tuple[str, ...]] = None + normalized_students: Optional[tuple[str, ...]] = None + unresolved: set[str] = set() + if teachers_needed: + normalized_teachers, missing = await self._normalize_roster_members( + tuple( + _participant_email(item) + for item in teachers + if _participant_email(item) + ), + directory_snapshot, + ) + unresolved.update(missing) + self._persist_roster_chunk( + import_id, + "teachers", + (course_id,), + {course_id: (normalized_teachers, missing)}, + requested, + manifest_guard_aliases, + ) + if students_needed: + normalized_students, missing = await self._normalize_roster_members( + tuple( + _participant_email(item) + for item in students + if _participant_email(item) + ), + directory_snapshot, + ) + unresolved.update(missing) + self._persist_roster_chunk( + import_id, + "students", + (course_id,), + {course_id: (normalized_students, missing)}, + requested, + manifest_guard_aliases, + ) return ( course_id, - tuple(sorted({_participant_email(item) for item in teachers if _participant_email(item)})), - tuple(sorted({_participant_email(item) for item in students if _participant_email(item)})), + normalized_teachers, + normalized_students, + tuple(sorted(unresolved)), ) return { - course_id: (teachers, students) - for course_id, teachers, students in await asyncio.gather( - *(read(course_id) for course_id in course_ids) + course_id: (teachers, students, unresolved) + for course_id, teachers, students, unresolved in await asyncio.gather( + *( + read(course_id, teachers, students) + for course_id, ( + _alias, + _course, + _detail, + teachers, + students, + ) in requested.items() + ) ) } + async def _normalize_roster_members( + self, + members: Iterable[str], + directory_snapshot: Optional[Mapping[str, Any]], + ) -> tuple[tuple[str, ...], tuple[str, ...]]: + normalized: set[str] = set() + unresolved: set[str] = set() + for raw_email in sorted( + {_directory_identifier(email) for email in members} - {""} + ): + primary = await self._normalize_live_identity( + raw_email, + "", + directory_snapshot, + ) + if primary: + normalized.add(primary) + else: + unresolved.add(raw_email) + return tuple(sorted(normalized)), tuple(sorted(unresolved)) + + def _persist_roster_chunk( + self, + import_id: str, + role: str, + chunk: Sequence[str], + normalized: Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]], + requested: Mapping[ + str, + tuple[str, _DesiredCourse, Any, bool, bool], + ], + manifest_guard_aliases: set[str], + ) -> None: + aliases = tuple(requested[course_id][1].alias for course_id in chunk) + states = self.store.get_managed_course_states(aliases) + verifications: list[ManagedCourseVerification] = [] + dirty: list[ManagedCourseDirty] = [] + verified_keys: list[str] = [] + for course_id in chunk: + alias_key, course, detail, _teachers, _students = requested[course_id] + state = states.get(alias_key) + if ( + state is None + or not _has_exact_alias(detail, course.alias) + or state.course_id not in {"", course_id} + ): + if state is not None and state.course_id not in {"", course_id}: + self.store.mark_managed_course_dirty( + ( + ManagedCourseDirty( + course.alias, + metadata=True, + teachers=role == "teachers", + students=role == "students", + ), + ), + error_code="OR-MANAGED-COURSE-ID-DRIFT", + recovery_required=True, + ) + if alias_key in manifest_guard_aliases: + # Do not adopt the rebound ID. The executor still needs + # this live roster basis to produce OR-MANIFEST-DRIFT. + continue + raise OneRosterError( + "OR-MANAGED-COURSE-ID-DRIFT", + "The exact managed alias resolved to a different Classroom course.", + ) + raise _IncompleteReadCoverage( + "A roster chunk did not retain exact managed-course identity." + ) + members, unresolved = normalized[course_id] + if unresolved: + dirty.append( + ManagedCourseDirty( + course.alias, + teachers=role == "teachers", + students=role == "students", + ) + ) + continue + verifications.append( + ManagedCourseVerification( + alias=course.alias, + course_id=course_id, + source_class_id=course.class_id, + last_seen_import_id=import_id, + teacher_hash=teacher_hash(members) if role == "teachers" else None, + student_hash=student_hash(members) if role == "students" else None, + teacher_members=members if role == "teachers" else None, + student_members=members if role == "students" else None, + ) + ) + verified_keys.append(alias_key) + self.store.record_managed_course_verifications(tuple(verifications)) + self.store.mark_managed_course_dirty( + tuple(dirty), + error_code="OR-LIVE-PARTICIPANT-UNRESOLVED", + ) + self._checkpointed_scopes.update((key, role) for key in verified_keys) + + def _persist_live_verification( + self, + import_id: str, + courses: Sequence[_DesiredCourse], + live_courses: Mapping[str, Optional[_LiveCourse]], + read_plan: Mapping[str, _CourseReadIntent], + ) -> None: + verified: list[ManagedCourseVerification] = [] + dirty: list[ManagedCourseDirty] = [] + identity_dirty: list[ManagedCourseDirty] = [] + known_states = self.store.get_managed_course_states( + tuple(course.alias for course in courses) + ) + for course in courses: + key = course.alias.casefold() + intent = read_plan.get(key) + live = live_courses.get(key) + if intent is None or live is None: + continue + metadata_requested = bool( + intent.metadata and (key, "metadata") not in self._checkpointed_scopes + ) + teachers_requested = bool( + intent.teachers and (key, "teachers") not in self._checkpointed_scopes + ) + students_requested = bool( + intent.students and (key, "students") not in self._checkpointed_scopes + ) + if not (metadata_requested or teachers_requested or students_requested): + continue + course_id = _text(live.detail, "id") + known = known_states.get(key) + if known is not None and known.course_id and course_id != known.course_id: + identity_dirty.append( + ManagedCourseDirty( + course.alias, + metadata=metadata_requested, + teachers=teachers_requested, + students=students_requested, + ) + ) + continue + exact_alias = _has_exact_alias(live.detail, course.alias) + state = _text(live.detail, "course_state", "courseState").upper() + metadata_covered = bool( + metadata_requested + and course_id + and exact_alias + and live.owner_email + and state + ) + roster_covered = bool( + course_id and exact_alias and not live.unresolved_members + ) + if metadata_requested and not metadata_covered: + dirty.append(ManagedCourseDirty(course.alias, metadata=True)) + if teachers_requested and not roster_covered: + dirty.append(ManagedCourseDirty(course.alias, teachers=True)) + if students_requested and not roster_covered: + dirty.append(ManagedCourseDirty(course.alias, students=True)) + if not ( + metadata_covered + or (teachers_requested and roster_covered) + or (students_requested and roster_covered) + ): + continue + verified.append( + ManagedCourseVerification( + alias=course.alias, + course_id=course_id, + source_class_id=course.class_id, + last_seen_import_id=import_id, + metadata_hash=( + metadata_hash( + course.alias, + _text(live.detail, "name"), + live.owner_email, + _text(live.detail, "room"), + _text(live.detail, "section"), + state, + ) + if metadata_covered + else None + ), + teacher_hash=( + teacher_hash(live.teachers) + if teachers_requested and roster_covered + else None + ), + student_hash=( + student_hash(live.students) + if students_requested and roster_covered + else None + ), + teacher_members=( + tuple(live.teachers) + if teachers_requested and roster_covered + else None + ), + student_members=( + tuple(live.students) + if students_requested and roster_covered + else None + ), + course_state=state if metadata_covered else None, + ) + ) + self.store.record_managed_course_verifications(tuple(verified)) + self.store.mark_managed_course_dirty( + tuple(dirty), + error_code="OR-LIVE-READ-INCOMPLETE", + ) + self.store.mark_managed_course_dirty( + tuple(identity_dirty), + error_code="OR-MANAGED-COURSE-ID-DRIFT", + recovery_required=True, + ) + async def _archive_actions( self, import_id: str, @@ -958,7 +2245,11 @@ def _resolve_snapshot_user( ) -def _index_managed_courses(raw: Any) -> dict[str, Any]: +def _index_managed_courses( + raw: Any, + *, + requested_aliases: Sequence[str], +) -> dict[str, Any]: if ( isinstance(raw, (str, bytes, Mapping)) or not isinstance(raw, Sequence) @@ -967,30 +2258,46 @@ def _index_managed_courses(raw: Any) -> dict[str, Any]: "OR-CLASSROOM-READ", "The live Classroom snapshot had an invalid shape; no import plan was created.", ) + requested = { + _managed_alias_key(alias): str(alias).removeprefix("d:") + for alias in requested_aliases + if _managed_alias_key(alias) + } + if len(requested) != len(requested_aliases): + raise OneRosterError( + "OR-ALIAS-COLLISION", + "The managed-course metadata chunk did not contain unique exact aliases.", + ) indexed: dict[str, Any] = {} course_ids: dict[str, str] = {} for detail in raw: - aliases = { + matches = { key for alias in _aliases(detail) if (key := _managed_alias_key(alias)) + and key in requested } - for key in aliases: - if key in indexed: + if len(matches) != 1: + raise OneRosterError( + "OR-ALIAS-COLLISION", + "A live Classroom course did not map to exactly one requested managed alias.", + ) + key = next(iter(matches)) + if key in indexed: + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one live Classroom course claims the same managed alias.", + ) + indexed[key] = detail + course_id = _text(detail, "id") + if course_id: + existing_alias = course_ids.get(course_id) + if existing_alias is not None and existing_alias != key: raise OneRosterError( "OR-ALIAS-COLLISION", - "More than one live Classroom course claims the same managed alias.", + "More than one managed alias resolves to the same live Classroom course.", ) - indexed[key] = detail - course_id = _text(detail, "id") - if course_id: - existing_alias = course_ids.get(course_id) - if existing_alias is not None and existing_alias != key: - raise OneRosterError( - "OR-ALIAS-COLLISION", - "More than one managed alias resolves to the same live Classroom course.", - ) - course_ids[course_id] = key + course_ids[course_id] = key return indexed @@ -1007,6 +2314,126 @@ def _managed_alias_key(value: Any) -> str: return alias.casefold() +def _desired_course_states( + courses: Sequence[_DesiredCourse], + import_id: str, +) -> dict[str, ManagedCourseDesired]: + return { + course.alias.casefold(): ManagedCourseDesired( + alias=course.alias, + import_id=import_id, + source_class_id=course.class_id, + metadata_hash=metadata_hash( + course.alias, + course.name, + course.owner_email, + course.room, + course.section, + "ACTIVE", + ), + teacher_hash=teacher_hash(_desired_members(course, "teacher")), + student_hash=student_hash(_desired_members(course, "student")), + ) + for course in courses + } + + +def _build_course_read_plan( + courses: Sequence[_DesiredCourse], + desired: Mapping[str, ManagedCourseDesired], + managed: Mapping[str, ManagedCourseState], + *, + previous_aliases: Sequence[str], + protected_aliases: Sequence[str], +) -> tuple[dict[str, _CourseReadIntent], set[str]]: + plan: dict[str, _CourseReadIntent] = {} + unchanged: set[str] = set() + for course in courses: + key = course.alias.casefold() + target = desired[key] + state = managed.get(key) + reasons: list[str] = [] + if state is None: + plan[key] = _CourseReadIntent( + metadata=True, + teachers=True, + students=True, + reasons=("new_course",), + ) + continue + metadata_changed = target.metadata_hash != state.desired_metadata_hash + teachers_changed = target.teacher_hash != state.desired_teacher_hash + students_changed = target.student_hash != state.desired_student_hash + metadata_dirty = bool( + state.metadata_dirty + or not state.course_id + or target.metadata_hash != state.verified_metadata_hash + ) + teachers_dirty = bool( + state.teacher_roster_dirty + or target.teacher_hash != state.verified_teacher_hash + ) + students_dirty = bool( + state.student_roster_dirty + or target.student_hash != state.verified_student_hash + ) + if metadata_changed: + reasons.append("metadata_changed") + if teachers_changed: + reasons.append("teachers_changed") + if students_changed: + reasons.append("students_changed") + if metadata_dirty and not metadata_changed: + reasons.append("dirty_metadata") + if teachers_dirty and not teachers_changed: + reasons.append("dirty_teachers") + if students_dirty and not students_changed: + reasons.append("dirty_students") + if state.recovery_required: + reasons.append("recovery_required") + metadata_dirty = teachers_dirty = students_dirty = True + intent = _CourseReadIntent( + metadata=metadata_changed or metadata_dirty, + teachers=teachers_changed or teachers_dirty, + students=students_changed or students_dirty, + reasons=tuple(reasons), + ) + if intent.metadata or intent.teachers or intent.students: + plan[key] = intent + else: + unchanged.add(key) + + protected = {alias.casefold() for alias in protected_aliases} + for alias in previous_aliases: + key = alias.casefold() + if key not in protected: + plan[key] = _CourseReadIntent( + metadata=True, + teachers=False, + students=False, + reasons=("removed_from_source",), + ) + unchanged.discard(key) + return plan, unchanged + + +def _cached_course_detail( + course: _DesiredCourse, + state: Optional[ManagedCourseState], +) -> Optional[dict[str, Any]]: + if state is None or not state.course_id: + return None + return { + "id": state.course_id, + "aliases": (course.alias, f"d:{course.alias}"), + "name": course.name, + "section": course.section, + "room": course.room, + "owner_email": course.owner_email, + "course_state": state.verified_course_state or "ACTIVE", + } + + def _read_desired_courses( path: Any, domain: str, @@ -1336,7 +2763,11 @@ def _append_existing_course_actions( "section": course.section, "room": course.room, } - if not limited_import and current_metadata != desired_metadata: + if ( + live.metadata_loaded + and not limited_import + and current_metadata != desired_metadata + ): _append_action( ordinary, _action( @@ -1349,7 +2780,7 @@ def _append_existing_course_actions( ) state = _text(detail, "course_state", "courseState").upper() - if not limited_import and state != "ACTIVE": + if live.metadata_loaded and not limited_import and state != "ACTIVE": _append_action( ordinary, _action( @@ -1367,27 +2798,30 @@ def _append_existing_course_actions( current_students = set(live.students) current_owner = live.owner_email - for teacher in sorted(desired_teachers - current_teachers): - _append_action(ordinary, _action("teacher_add", course.alias, teacher)) - if not limited_import: - for teacher in sorted(current_teachers - desired_teachers): - if teacher == current_owner: - continue - _append_action( - ordinary, - _action("teacher_remove", course.alias, teacher), - ) - for student in sorted(desired_students - current_students): - _append_action(ordinary, _action("student_add", course.alias, student)) - if not limited_import: - for student in sorted(current_students - desired_students): - _append_action( - ordinary, - _action("student_remove", course.alias, student), - ) + if live.teachers_loaded: + for teacher in sorted(desired_teachers - current_teachers): + _append_action(ordinary, _action("teacher_add", course.alias, teacher)) + if not limited_import: + for teacher in sorted(current_teachers - desired_teachers): + if teacher == current_owner: + continue + _append_action( + ordinary, + _action("teacher_remove", course.alias, teacher), + ) + if live.students_loaded: + for student in sorted(desired_students - current_students): + _append_action(ordinary, _action("student_add", course.alias, student)) + if not limited_import: + for student in sorted(current_students - desired_students): + _append_action( + ordinary, + _action("student_remove", course.alias, student), + ) if ( - not limited_import + live.metadata_loaded + and not limited_import and current_owner and current_owner != course.owner_email ): @@ -1448,7 +2882,7 @@ def _live_basis(alias: str, live: _LiveCourse) -> dict[str, Any]: "alias": alias, "exists": True, "id": _text(detail, "id"), - "aliases": sorted(_aliases(detail)), + "aliases": [alias] if _has_exact_alias(detail, alias) else sorted(_aliases(detail)), "name": _text(detail, "name"), "section": _text(detail, "section"), "room": _text(detail, "room"), diff --git a/gamgui/components/oneroster/semantic.py b/gamgui/components/oneroster/semantic.py new file mode 100644 index 0000000..ffefa07 --- /dev/null +++ b/gamgui/components/oneroster/semantic.py @@ -0,0 +1,62 @@ +"""Pure canonical semantic hashing for OneRoster managed-course state.""" + +from __future__ import annotations + +from typing import Any, Iterable + +from .models import canonical_hash + + +def normalize_email_values(values: Iterable[object]) -> tuple[str, ...]: + """Return deterministic membership semantics without retaining identities twice.""" + + return tuple( + sorted( + { + str(value or "").strip().casefold() + for value in values + if str(value or "").strip() + } + ) + ) + + +def metadata_basis( + alias: object, + name: object, + owner_email: object, + room: object, + section: object, + state: object = "ACTIVE", +) -> dict[str, Any]: + """Build the exact desired metadata basis without touching authored descriptions.""" + + return { + "alias": str(alias or "").strip(), + "name": str(name or "").strip(), + "owner_email": str(owner_email or "").strip().casefold(), + "room": str(room or "").strip(), + "section": str(section or "").strip(), + "state": str(state or "").strip().upper(), + } + + +def metadata_hash( + alias: object, + name: object, + owner_email: object, + room: object, + section: object, + state: object = "ACTIVE", +) -> str: + return canonical_hash( + metadata_basis(alias, name, owner_email, room, section, state) + ) + + +def teacher_hash(values: Iterable[object]) -> str: + return canonical_hash(normalize_email_values(values)) + + +def student_hash(values: Iterable[object]) -> str: + return canonical_hash(normalize_email_values(values)) diff --git a/gamgui/components/oneroster/store.py b/gamgui/components/oneroster/store.py index 26383fe..75f3b2f 100644 --- a/gamgui/components/oneroster/store.py +++ b/gamgui/components/oneroster/store.py @@ -44,6 +44,10 @@ ImportAction, ImportIssue, IssueSeverity, + ManagedCourseDesired, + ManagedCourseDirty, + ManagedCourseState, + ManagedCourseVerification, ManifestPage, MAX_PAGE_SIZE, OneRosterError, @@ -67,6 +71,7 @@ preview_total, source_rows, ) +from .semantic import normalize_email_values, student_hash, teacher_hash from .thresholds import evaluation_hash @@ -97,7 +102,6 @@ def default_component_data_root() -> Path: ), ) - def _progress_phase(kind: str) -> tuple[str, str]: normalized = str(kind or "").casefold() for key, label, kinds in _PROGRESS_PHASES: @@ -560,6 +564,346 @@ def mark_accepted(self, import_id: str, *, when: Optional[float] = None) -> None (accepted_at, self.domain, import_id), ) + def get_managed_course_states( + self, + aliases: Optional[Sequence[str]] = None, + ) -> dict[str, ManagedCourseState]: + """Return durable verification state keyed by case-folded exact alias.""" + + requested = ( + {_managed_alias(alias).casefold() for alias in aliases} + if aliases is not None + else None + ) + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT * FROM managed_course_state + WHERE domain = ? ORDER BY alias COLLATE NOCASE + """, + (self.domain,), + ).fetchall() + return { + str(row["alias"]).casefold(): _managed_course_state_from_row(row) + for row in rows + if requested is None or str(row["alias"]).casefold() in requested + } + + def record_managed_course_desired( + self, + values: Sequence[ManagedCourseDesired], + *, + now: Optional[float] = None, + ) -> None: + """Persist authoritative desired hashes without claiming live verification.""" + + timestamp = float(now if now is not None else time.time()) + normalized: list[ManagedCourseDesired] = [] + for value in values: + normalized.append( + ManagedCourseDesired( + alias=_managed_alias(value.alias), + import_id=str(value.import_id or "").strip(), + metadata_hash=_validate_digest(value.metadata_hash), + teacher_hash=_validate_digest(value.teacher_hash), + student_hash=_validate_digest(value.student_hash), + source_class_id=str(value.source_class_id or "").strip(), + ) + ) + if not normalized: + return + with closing(self._conn()) as conn, conn: + conn.executemany( + """ + INSERT INTO managed_course_state ( + domain, alias, course_id, source_class_id, last_seen_import_id, + desired_metadata_hash, desired_teacher_hash, desired_student_hash, + verified_metadata_hash, verified_teacher_hash, verified_student_hash, + last_metadata_verified_at, last_teacher_verified_at, + last_student_verified_at, verified_course_state, + metadata_dirty, teacher_roster_dirty, student_roster_dirty, + recovery_required, last_error_code, version, updated_at + ) VALUES ( + ?, ?, '', ?, ?, ?, ?, ?, '', '', '', 0, 0, 0, '', + 1, 1, 1, 0, '', 1, ? + ) + ON CONFLICT(domain, alias) DO UPDATE SET + source_class_id = excluded.source_class_id, + last_seen_import_id = excluded.last_seen_import_id, + metadata_dirty = CASE + WHEN managed_course_state.verified_metadata_hash != excluded.desired_metadata_hash + THEN 1 ELSE managed_course_state.metadata_dirty END, + teacher_roster_dirty = CASE + WHEN managed_course_state.verified_teacher_hash != excluded.desired_teacher_hash + THEN 1 ELSE managed_course_state.teacher_roster_dirty END, + student_roster_dirty = CASE + WHEN managed_course_state.verified_student_hash != excluded.desired_student_hash + THEN 1 ELSE managed_course_state.student_roster_dirty END, + desired_metadata_hash = excluded.desired_metadata_hash, + desired_teacher_hash = excluded.desired_teacher_hash, + desired_student_hash = excluded.desired_student_hash, + version = managed_course_state.version + 1, + updated_at = excluded.updated_at + """, + ( + ( + self.domain, + value.alias, + value.source_class_id, + value.import_id, + value.metadata_hash, + value.teacher_hash, + value.student_hash, + timestamp, + ) + for value in normalized + ), + ) + + def record_managed_course_verifications( + self, + values: Sequence[ManagedCourseVerification], + *, + now: Optional[float] = None, + ) -> None: + """Persist only explicitly covered live scopes in one transaction.""" + + if not values: + return + timestamp = float(now if now is not None else time.time()) + with closing(self._conn()) as conn, conn: + for value in values: + self._record_managed_course_verification( + conn, + value, + timestamp=timestamp, + ) + + def mark_managed_course_dirty( + self, + values: Sequence[ManagedCourseDirty], + *, + error_code: str, + recovery_required: bool = False, + now: Optional[float] = None, + ) -> None: + """Mark only uncertain scopes dirty; unrelated verification stays valid.""" + + if not values: + return + timestamp = float(now if now is not None else time.time()) + with closing(self._conn()) as conn, conn: + for value in values: + alias = _managed_alias(value.alias) + conn.execute( + """ + UPDATE managed_course_state SET + metadata_dirty = CASE WHEN ? THEN 1 ELSE metadata_dirty END, + teacher_roster_dirty = CASE WHEN ? THEN 1 ELSE teacher_roster_dirty END, + student_roster_dirty = CASE WHEN ? THEN 1 ELSE student_roster_dirty END, + recovery_required = CASE WHEN ? THEN 1 ELSE recovery_required END, + last_error_code = ?, version = version + 1, updated_at = ? + WHERE domain = ? AND alias = ? + """, + ( + bool(value.metadata), + bool(value.teachers), + bool(value.students), + bool(recovery_required), + str(error_code or "")[:100], + timestamp, + self.domain, + alias, + ), + ) + + def verified_managed_members(self, alias: str, role: str) -> tuple[str, ...]: + managed_alias = _managed_alias(alias) + normalized_role = str(role or "").strip().casefold() + if normalized_role not in {"teachers", "students"}: + raise ValueError("Managed-course role must be teachers or students.") + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT email FROM managed_course_members + WHERE domain = ? AND alias = ? AND role = ? + ORDER BY email + """, + (self.domain, managed_alias, normalized_role), + ).fetchall() + return tuple(str(row["email"]) for row in rows) + + def verified_managed_members_many( + self, + aliases: Sequence[str], + ) -> dict[str, tuple[tuple[str, ...], tuple[str, ...]]]: + requested = {_managed_alias(alias).casefold() for alias in aliases} + result: dict[str, tuple[list[str], list[str]]] = { + alias: ([], []) for alias in requested + } + if not requested: + return {} + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT alias, role, email FROM managed_course_members + WHERE domain = ? ORDER BY alias COLLATE NOCASE, role, email + """, + (self.domain,), + ).fetchall() + for row in rows: + key = str(row["alias"]).casefold() + if key not in result: + continue + teachers, students = result[key] + (teachers if str(row["role"]) == "teachers" else students).append( + str(row["email"]) + ) + return { + alias: (tuple(teachers), tuple(students)) + for alias, (teachers, students) in result.items() + } + + def select_managed_audit_aliases( + self, + aliases: Sequence[str], + *, + limit: int = 25, + ) -> tuple[str, ...]: + """Choose the oldest fully verified eligible aliases deterministically.""" + + cap = max(0, min(int(limit), 25)) + if not cap: + return () + requested = {_managed_alias(alias).casefold() for alias in aliases} + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT alias, + MIN(last_metadata_verified_at, + last_teacher_verified_at, + last_student_verified_at) AS oldest + FROM managed_course_state + WHERE domain = ? + AND metadata_dirty = 0 + AND teacher_roster_dirty = 0 + AND student_roster_dirty = 0 + AND recovery_required = 0 + ORDER BY oldest ASC, alias COLLATE NOCASE ASC + """, + (self.domain,), + ).fetchall() + return tuple( + str(row["alias"]) + for row in rows + if str(row["alias"]).casefold() in requested + )[:cap] + + def _record_managed_course_verification( + self, + conn: sqlite3.Connection, + value: ManagedCourseVerification, + *, + timestamp: float, + ) -> None: + alias = _managed_alias(value.alias) + row = conn.execute( + "SELECT * FROM managed_course_state WHERE domain = ? AND alias = ?", + (self.domain, alias), + ).fetchone() + if row is None: + raise OneRosterError( + "OR-MANAGED-STATE-MISSING", + "Live verification cannot adopt an unregistered Classroom course.", + ) + course_id = str(value.course_id or "").strip() + existing_id = str(row["course_id"] or "").strip() + if existing_id and course_id and existing_id != course_id: + raise OneRosterError( + "OR-MANAGED-COURSE-ID-DRIFT", + "The exact managed alias resolved to a different Classroom course.", + ) + effective_course_id = course_id or existing_id + verified_at = float(value.verified_at or timestamp) + updates: dict[str, Any] = { + "course_id": effective_course_id, + "source_class_id": str(value.source_class_id or row["source_class_id"] or ""), + "last_seen_import_id": str(value.last_seen_import_id or row["last_seen_import_id"] or ""), + "last_error_code": "", + "updated_at": timestamp, + } + if value.metadata_hash is not None: + updates.update( + verified_metadata_hash=_validate_digest(value.metadata_hash), + last_metadata_verified_at=verified_at, + verified_course_state=str(value.course_state or "").strip().upper(), + metadata_dirty=0, + ) + for role, supplied_hash, supplied_members, hash_function, dirty_column, time_column, hash_column in ( + ( + "teachers", + value.teacher_hash, + value.teacher_members, + teacher_hash, + "teacher_roster_dirty", + "last_teacher_verified_at", + "verified_teacher_hash", + ), + ( + "students", + value.student_hash, + value.student_members, + student_hash, + "student_roster_dirty", + "last_student_verified_at", + "verified_student_hash", + ), + ): + if supplied_hash is None: + if supplied_members is not None: + raise ValueError("Managed-course members require a matching verified hash.") + continue + if supplied_members is None: + raise ValueError("A verified roster hash requires covered member evidence.") + members = normalize_email_values(supplied_members) + digest = _validate_digest(supplied_hash) + if hash_function(members) != digest: + raise ValueError("Managed-course roster members do not match the verified hash.") + conn.execute( + "DELETE FROM managed_course_members WHERE domain = ? AND alias = ? AND role = ?", + (self.domain, alias, role), + ) + conn.executemany( + """ + INSERT INTO managed_course_members(domain, alias, role, email, verified_at) + VALUES (?, ?, ?, ?, ?) + """, + ((self.domain, alias, role, email, verified_at) for email in members), + ) + updates[hash_column] = digest + updates[time_column] = verified_at + updates[dirty_column] = 0 + assignments = ", ".join(f"{column} = ?" for column in updates) + conn.execute( + f""" + UPDATE managed_course_state SET {assignments}, version = version + 1 + WHERE domain = ? AND alias = ? + """, + (*updates.values(), self.domain, alias), + ) + if value.clear_recovery: + conn.execute( + """ + UPDATE managed_course_state + SET recovery_required = CASE + WHEN metadata_dirty = 0 AND teacher_roster_dirty = 0 + AND student_roster_dirty = 0 THEN 0 + ELSE recovery_required END + WHERE domain = ? AND alias = ? + """, + (self.domain, alias), + ) + def maybe_mark_import_accepted( self, manifest_id: str, @@ -1519,6 +1863,8 @@ def complete_verified_batch( verification_attempts: int, worker_count: int, throttling_count: int = 0, + managed_verifications: Sequence[ManagedCourseVerification] = (), + managed_dirty: Sequence[ManagedCourseDirty] = (), now: Optional[float] = None, ) -> ExecutionBatch: """Atomically save exact verified results and complete their durable batch.""" @@ -1605,6 +1951,33 @@ def complete_verified_batch( "OR-BATCH-ACTIONS-CHANGED", "Not every immutable batch action could be updated.", ) + for verification in managed_verifications: + self._record_managed_course_verification( + conn, + verification, + timestamp=timestamp, + ) + for dirty in managed_dirty: + alias = _managed_alias(dirty.alias) + conn.execute( + """ + UPDATE managed_course_state SET + metadata_dirty = CASE WHEN ? THEN 1 ELSE metadata_dirty END, + teacher_roster_dirty = CASE WHEN ? THEN 1 ELSE teacher_roster_dirty END, + student_roster_dirty = CASE WHEN ? THEN 1 ELSE student_roster_dirty END, + last_error_code = 'OR-BATCH-VERIFY-FAILED', + version = version + 1, updated_at = ? + WHERE domain = ? AND alias = ? + """, + ( + bool(dirty.metadata), + bool(dirty.teachers), + bool(dirty.students), + timestamp, + self.domain, + alias, + ), + ) failed = any(status == "failed" for status, _detail in normalized.values()) persistence_seconds = time.perf_counter() - persist_started conn.execute( @@ -2866,6 +3239,52 @@ def _init_state(self) -> None: ); CREATE INDEX IF NOT EXISTS accepted_aliases_domain_time ON accepted_managed_aliases(domain, accepted_at DESC, alias); + CREATE TABLE IF NOT EXISTS managed_course_state ( + domain TEXT NOT NULL, + alias TEXT NOT NULL, + course_id TEXT NOT NULL DEFAULT '', + source_class_id TEXT NOT NULL DEFAULT '', + last_seen_import_id TEXT NOT NULL DEFAULT '', + desired_metadata_hash TEXT NOT NULL DEFAULT '', + desired_teacher_hash TEXT NOT NULL DEFAULT '', + desired_student_hash TEXT NOT NULL DEFAULT '', + verified_metadata_hash TEXT NOT NULL DEFAULT '', + verified_teacher_hash TEXT NOT NULL DEFAULT '', + verified_student_hash TEXT NOT NULL DEFAULT '', + last_metadata_verified_at REAL NOT NULL DEFAULT 0, + last_teacher_verified_at REAL NOT NULL DEFAULT 0, + last_student_verified_at REAL NOT NULL DEFAULT 0, + verified_course_state TEXT NOT NULL DEFAULT '', + metadata_dirty INTEGER NOT NULL DEFAULT 1, + teacher_roster_dirty INTEGER NOT NULL DEFAULT 1, + student_roster_dirty INTEGER NOT NULL DEFAULT 1, + recovery_required INTEGER NOT NULL DEFAULT 0, + last_error_code TEXT NOT NULL DEFAULT '', + version INTEGER NOT NULL DEFAULT 1, + updated_at REAL NOT NULL DEFAULT 0, + PRIMARY KEY(domain, alias) + ); + CREATE UNIQUE INDEX IF NOT EXISTS managed_course_state_course_id + ON managed_course_state(domain, course_id) + WHERE course_id != ''; + CREATE INDEX IF NOT EXISTS managed_course_state_audit + ON managed_course_state( + domain, metadata_dirty, teacher_roster_dirty, + student_roster_dirty, recovery_required, + last_metadata_verified_at, last_teacher_verified_at, + last_student_verified_at, alias + ); + CREATE TABLE IF NOT EXISTS managed_course_members ( + domain TEXT NOT NULL, + alias TEXT NOT NULL, + role TEXT NOT NULL CHECK(role IN ('teachers', 'students')), + email TEXT NOT NULL, + verified_at REAL NOT NULL, + PRIMARY KEY(domain, alias, role, email), + FOREIGN KEY(domain, alias) + REFERENCES managed_course_state(domain, alias) + ON DELETE CASCADE + ); """ ) _ensure_column( @@ -3070,6 +3489,47 @@ def _snapshot_write_conn(path: Path) -> sqlite3.Connection: raise +def _managed_alias(value: object) -> str: + alias = str(value or "").strip() + if ( + not alias.startswith("Section_") + or len(alias) == len("Section_") + or any(character in alias for character in "\r\n\x00") + ): + raise OneRosterError( + "OR-ALIAS-INVALID", + "A managed Classroom alias must use the exact Section_ form.", + ) + return alias + + +def _managed_course_state_from_row(row: sqlite3.Row) -> ManagedCourseState: + return ManagedCourseState( + domain=str(row["domain"]), + alias=str(row["alias"]), + course_id=str(row["course_id"] or ""), + source_class_id=str(row["source_class_id"] or ""), + last_seen_import_id=str(row["last_seen_import_id"] or ""), + desired_metadata_hash=str(row["desired_metadata_hash"] or ""), + desired_teacher_hash=str(row["desired_teacher_hash"] or ""), + desired_student_hash=str(row["desired_student_hash"] or ""), + verified_metadata_hash=str(row["verified_metadata_hash"] or ""), + verified_teacher_hash=str(row["verified_teacher_hash"] or ""), + verified_student_hash=str(row["verified_student_hash"] or ""), + last_metadata_verified_at=float(row["last_metadata_verified_at"] or 0), + last_teacher_verified_at=float(row["last_teacher_verified_at"] or 0), + last_student_verified_at=float(row["last_student_verified_at"] or 0), + verified_course_state=str(row["verified_course_state"] or ""), + metadata_dirty=bool(row["metadata_dirty"]), + teacher_roster_dirty=bool(row["teacher_roster_dirty"]), + student_roster_dirty=bool(row["student_roster_dirty"]), + recovery_required=bool(row["recovery_required"]), + last_error_code=str(row["last_error_code"] or ""), + version=int(row["version"] or 1), + updated_at=float(row["updated_at"] or 0), + ) + + def _managed_aliases_from_snapshot(path: Path, domain: str) -> tuple[str, ...]: with closing(_snapshot_conn(path)) as conn: rows = conn.execute( diff --git a/gamgui/core/connectors/gam_connector.py b/gamgui/core/connectors/gam_connector.py index b4b34a6..1cea9ce 100644 --- a/gamgui/core/connectors/gam_connector.py +++ b/gamgui/core/connectors/gam_connector.py @@ -64,6 +64,8 @@ from .person import ConnectorAccount, Person MAX_ONEROSTER_BULK_ROSTER_MEMBERS = 500_000 +ONEROSTER_MANAGED_ALIAS_CHUNK_CAP = 200 +ONEROSTER_ROSTER_CHUNK_CAP = 100 def _parse_signature(text: str) -> str: @@ -851,6 +853,10 @@ async def list_oneroster_managed_courses( selected.append(alias) if not selected: return [] + if len(selected) > ONEROSTER_MANAGED_ALIAS_CHUNK_CAP: + raise ValueError( + "managed Classroom alias request exceeds the OneRoster chunk cap" + ) selector_path = _write_private_selector( selected, @@ -1041,6 +1047,10 @@ async def list_course_participants_many( raise ValueError("invalid Classroom roster role") if not selected: return CourseRosterSnapshot.empty() + if len(selected) > ONEROSTER_ROSTER_CHUNK_CAP: + raise ValueError( + "Classroom roster request exceeds the OneRoster chunk cap" + ) if any( len(course_id) > 512 or any(character in course_id for character in "\r\n\x00") @@ -1071,7 +1081,7 @@ async def list_course_participants_many( normalized_role, ), timeout=timeout, - serialize=True, + serialize=False, ) as result: participants = await _parse_private_spool_off_loop( _read_oneroster_participants_spool, diff --git a/scripts/build_windows_release.ps1 b/scripts/build_windows_release.ps1 index 8d8bc82..0e1faee 100644 --- a/scripts/build_windows_release.ps1 +++ b/scripts/build_windows_release.ps1 @@ -16,6 +16,12 @@ $repoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path Set-Location $repoRoot $python = Join-Path $repoRoot ".venv\Scripts\python.exe" if (-not (Test-Path -LiteralPath $python)) { throw "Run uv sync with the dev, desktop, and build extras first." } +$gamVersionOutput = & $python -c "from gamgui.core.gam.commands import EXPECTED_GAM_VERSION; print(EXPECTED_GAM_VERSION)" +$gamVersionExitCode = $LASTEXITCODE +$expectedGamVersion = ($gamVersionOutput | Out-String).Trim() +if ($gamVersionExitCode -ne 0 -or $expectedGamVersion -notmatch '^\d+\.\d+\.\d+$') { + throw "The repository GAM source pin is invalid." +} $packagedPaths = @( "main.py", "gamgui", "gamgui.spec", "gamgui-updater.spec", "pyproject.toml", "uv.lock", @@ -44,7 +50,7 @@ if ($Bootstrap) { } } -& (Join-Path $PSScriptRoot "fetch_gam_windows.ps1") +& (Join-Path $PSScriptRoot "fetch_gam_windows.ps1") -Tag "v$expectedGamVersion" if ($LASTEXITCODE) { throw "Pinned GAM vendoring failed." } & $python -c "import PyInstaller, webview, keyring" if ($LASTEXITCODE) { throw "Locked build dependencies are unavailable." } @@ -88,8 +94,8 @@ if ( [string]$embeddedProfile.artifact.toolchain_manifest_digest -ne $toolchainDigest -or [string]$embeddedProfile.artifact.signer_thumbprint -ne $CertificateSha256.ToLowerInvariant() ) { throw "The embedded Windows artifact identity did not match the exact build inputs." } -$embeddedGamVersion = (Get-Content -Raw -LiteralPath (Join-Path $bundle "_internal\resources\gam7\VERSION")).Trim() -if ($embeddedGamVersion -notmatch '7\.47\.02') { throw "The embedded Windows GAM version did not match the tested pin." } +$embeddedGamTag = (Get-Content -Raw -LiteralPath (Join-Path $bundle "_internal\resources\gam7\VERSION")).Trim() +if ($embeddedGamTag -cne "v$expectedGamVersion") { throw "The embedded Windows GAM version did not match the repository source pin." } & $python -c "import struct,sys; p=open(sys.argv[1],'rb'); p.seek(0x3c); p.seek(struct.unpack('