fix: scope pgy rollover reads by academic year

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Academic block resident rows and anticipated leave generation now resolve PGY from `PersonAcademicYear` for the requested academic year when available, fall back to legacy `Person.pgy_level` only when an AY row is missing, and exclude AY-graduated residents from those rollover-sensitive reads.
 - Annual-planning Playwright coverage now boots isolated local frontend/backend servers on `127.0.0.1:3100` and `127.0.0.1:8010`, so the review-flow slice no longer depends on whatever developer servers are already running on `3000/8000`.
 - Annual shock preview now groups the full repair-block blast radius around the direct absence hits, so coordinators can see which other scheduled assignments sit in the same repair block before generating or publishing repair work.
 - Annual-planning readiness now evaluates effective program policy snapshots, recurring program calendar anchors, and policy-layered institutional events together, so coordinators see the full DB-backed policy stack instead of only institutional-event coverage before annual optimization or repair work.
@@ -104,6 +105,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 **Native Runtime Hardening (#1106)** — Multi-version PG detection, pgvector checks, env var defaults
 
 ### Fixed
+
+- Anticipated leave regeneration now deletes only `anticipated` placeholders that fall inside the requested academic year and uses canonical academic-block date helpers instead of stale `Person.is_active` / `Block.academic_year` assumptions.
 - Repaired missing SQLAlchemy-Continuum version tables for `rotation_templates`, `survey_responses`, and `swap_records`, which could otherwise break audited updates such as template-import publish on drifted databases.
 - Hybrid outpatient target penalties now scale down when a scoped week has fewer solver-owned slots than the published target set assumes, reducing distortion in truncated or overlay-heavy blocks like FMC Block 13.
 - FMC hybrid outpatient consumption now honors week-scoped activity requirements and converts imported template weekdays into the activity solver's Python weekday convention, so published template targets no longer bleed across weeks or land on the wrong days at runtime.

TODO.md

@@ -1,6 +1,6 @@
 # TODO — Actionable Items
 
-> **Updated:** 2026-03-21 (policy-layer and annual-planning browser-confidence actualization in progress)
+> **Updated:** 2026-03-22 (policy-layer actualization merged; PGY rollover batch in progress)
 > **Source:** Extracted from architecture docs, planning docs, code TODOs, and Explore agent audit.
 > **Companion:** `docs/planning/ROADMAP.md` (macro vision), `docs/planning/TECHNICAL_DEBT.md` (debt tracker)
 > **Cutover tracker:** `docs/planning/ROTATION_SHAPE_CUTOVER_STATUS_20260320.md` (merged vs open PR vs untouched rotation-shape/constraint cutover work)
@@ -12,10 +12,11 @@
 
 - [ ] **PII in git history** — Resident names in deleted files still in repo history. Requires `git filter-repo` + force push. All collaborators must re-clone after. **Human-only.** See `docs/security/PII_AUDIT_LOG.md`.
 
-- [ ] **Track A: PGY Graduation Rollover (July 1 deadline)** — `Person.pgy_level` is not academic-year scoped. Updating it on July 1 corrupts historical ACGME queries. Migration exists (`20260224_person_ay.py`), Alembic heads merged (PR #1196), `_sync_academic_year_call_counts()` implemented (PR #1199). **Remaining:**
-  1. Apply migration: `alembic upgrade head` (requires running DB)
-  2. Migrate 67+ consumer files from `Person.pgy_level` → `PersonAcademicYear` per-AY reads
-  3. Create graduation rollover logic for July 1
+- [ ] **Track A: PGY Graduation Rollover (July 1 deadline)** — `Person.pgy_level` is not academic-year scoped. Updating it on July 1 corrupts historical ACGME queries. Migration exists (`20260224_person_ay.py`), Alembic heads merged (PR #1196), `_sync_academic_year_call_counts()` implemented (PR #1199), and the local dev DB is already at Alembic head (`20260321_program_policy_snapshots`). **Remaining:**
+  1. Migrate the remaining `Person.pgy_level` consumers to `PersonAcademicYear` per-AY reads
+  2. Create graduation rollover logic for July 1
+  3. Apply the AY migration chain in any non-upgraded environments before cutover
+  - Current branch progress: `AcademicBlockService` resident rows and `AnticipatedLeaveService` intern selection now prefer AY-scoped PGY rows with safe fallback to legacy `Person.pgy_level`, and they exclude AY-graduated residents.
   - **Doc:** `docs/architecture/excel-stateful-roundtrip-roadmap.md` (Track A, lines 240-312)
 
 ## P1 — High / This Sprint

backend/app/services/academic_block_service.py

@@ -27,6 +27,7 @@
     MatrixCell,
     ResidentRow,
 )
+from app.services.person_academic_year_resolver import list_effective_resident_pgys
 from app.utils.academic_blocks import get_all_block_dates
 from app.validators.advanced_acgme import AdvancedACGMEValidator
 
@@ -76,7 +77,7 @@ def get_block_matrix(
         academic_blocks = self._generate_academic_blocks(start_date, end_date)
 
         # Get residents (rows)
-        residents = self._get_residents(pgy_level)
+        residents = self._get_residents(academic_year=academic_year, pgy_level=pgy_level)
 
         # Get all assignments for the date range
         assignments = self._get_assignments_in_range(start_date, end_date)
@@ -215,27 +216,46 @@ def _generate_academic_blocks(
 
         return blocks
 
-    def _get_residents(self, pgy_level: int | None = None) -> list[ResidentRow]:
+    def _get_residents(
+        self,
+        academic_year: str | None = None,
+        pgy_level: int | None = None,
+    ) -> list[ResidentRow]:
         """
         Get residents for matrix rows.
 
         Args:
+            academic_year: Optional academic year for AY-scoped PGY resolution
             pgy_level: Optional PGY level filter
 
         Returns:
             List of ResidentRow objects
         """
-        # Query residents
-        residents = self.person_repo.list_residents(pgy_level=pgy_level)
+        if academic_year is None:
+            residents = self.person_repo.list_residents(pgy_level=pgy_level)
+            return [
+                ResidentRow(
+                    resident_id=resident.id,
+                    name=resident.name,
+                    pgy_level=resident.pgy_level,
+                )
+                for resident in residents
+            ]
+
+        start_year, _ = self._parse_academic_year(academic_year)
+        resident_rows = list_effective_resident_pgys(
+            self.db,
+            academic_year=start_year.year,
+            pgy_level=pgy_level,
+        )
 
-        # Convert to ResidentRow schema
         return [
             ResidentRow(
-                resident_id=resident.id,
-                name=resident.name,
+                resident_id=resident.person.id,
+                name=resident.person.name,
                 pgy_level=resident.pgy_level,
             )
-            for resident in residents
+            for resident in resident_rows
         ]
 
     def _get_assignments_in_range(

backend/app/services/anticipated_leave_service.py

@@ -1,17 +1,20 @@
 """Service for generating anticipated leave for incoming interns."""
 
 import logging
-from datetime import date, timedelta
+from datetime import timedelta
 from uuid import UUID
 
 from sqlalchemy import select
-from sqlalchemy.orm import Session, selectinload
+from sqlalchemy.orm import Session
 
 from app.models.absence import Absence
-from app.schemas.absence import AbsenceType
 from app.models.person import Person
-from app.models.settings import ApplicationSettings
-from app.utils.academic_blocks import get_all_block_dates as get_academic_block_dates
+from app.schemas.absence import AbsenceType
+from app.services.person_academic_year_resolver import list_effective_resident_pgys
+from app.utils.academic_blocks import (
+    get_academic_year_for_date,
+    get_block_dates,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -34,35 +37,36 @@ def generate_anticipated_leave(
         if weeks_per_intern < 1:
             raise ValueError("weeks_per_intern must be at least 1")
 
-        # Find all PGY-1s for the current cohort.
-        # NOTE: Uses current pgy_level, not year-scoped. Once PersonAcademicYear
-        # table ships (Track A), scope by academic_year to handle past/future AYs.
-        interns = (
-            self.db.query(Person)
-            .filter(Person.pgy_level == 1, Person.is_active == True)  # noqa: E712
-            .all()
-        )
+        interns = [
+            resolved.person
+            for resolved in list_effective_resident_pgys(
+                self.db,
+                academic_year=academic_year,
+                pgy_level=1,
+            )
+        ]
 
         if not interns:
+            logger.info(
+                "No AY-scoped interns found while generating anticipated leave for AY %s",
+                academic_year,
+            )
             return {
                 "interns_processed": 0,
                 "absences_created": 0,
                 "absences_deleted": 0,
             }
 
-        # Clear existing anticipated leave for these interns in this AY
-        # This allows safely re-running the generator
         existing_stmt = select(Absence).where(
-            Absence.person_id.in_([i.id for i in interns]),
+            Absence.person_id.in_([intern.id for intern in interns]),
             Absence.status == "anticipated",
         )
         existing = self.db.execute(existing_stmt).scalars().all()
 
         deleted_count = 0
         for absence in existing:
-            # Simple check if it falls in the academic year
-            start_block, ay1 = get_academic_block_dates(absence.start_date)
-            end_block, ay2 = get_academic_block_dates(absence.end_date)
+            ay1 = get_academic_year_for_date(absence.start_date)
+            ay2 = get_academic_year_for_date(absence.end_date)
             if ay1 == academic_year or ay2 == academic_year:
                 self.db.delete(absence)
                 deleted_count += 1
@@ -88,29 +92,9 @@ def generate_anticipated_leave(
                 block_idx = (i + w * block_spacing) % num_blocks
                 block_num = available_blocks[block_idx]
 
-                # We need the start and end date of that block to put the leave in
-                # We'll just put it in the first week of the chosen block.
-                # In a real system, you might look up the actual Block model.
-                from app.models.block import Block
-
-                block_record = (
-                    self.db.query(Block)
-                    .filter(
-                        Block.block_number == block_num,
-                        Block.academic_year == academic_year,
-                    )
-                    .first()
-                )
-
-                if block_record:
-                    start = block_record.start_date
-                    end = start + timedelta(days=6)  # 1 week
-                else:
-                    # Fallback to rough calculation if Block records aren't generated yet
-                    from app.utils.academic_blocks import get_block_dates
-
-                    start, _ = get_block_dates(block_num, academic_year)  # type: ignore[misc]
-                    end = start + timedelta(days=6)
+                block_dates = get_block_dates(block_num, academic_year)
+                start = block_dates.start_date
+                end = start + timedelta(days=6)
 
                 absence = Absence(
                     person_id=intern.id,
@@ -129,7 +113,10 @@ def generate_anticipated_leave(
         self.db.commit()
 
         logger.info(
-            f"Generated {absences_created} anticipated leave blocks for {len(interns)} interns (AY {academic_year})"
+            "Generated %s anticipated leave blocks for %s interns (AY %s)",
+            absences_created,
+            len(interns),
+            academic_year,
         )
 
         return {

backend/app/services/person_academic_year_resolver.py

@@ -0,0 +1,88 @@
+"""Helpers for resolving AY-scoped resident state with legacy fallback."""
+
+from dataclasses import dataclass
+import logging
+from uuid import UUID
+
+from sqlalchemy import and_, select
+from sqlalchemy.orm import Session
+
+from app.models.person import Person
+from app.models.person_academic_year import PersonAcademicYear
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class EffectiveResidentPgy:
+    """Resolved resident PGY for a specific academic year."""
+
+    person: Person
+    pgy_level: int
+
+
+def list_effective_resident_pgys(
+    db: Session,
+    *,
+    academic_year: int,
+    pgy_level: int | None = None,
+) -> list[EffectiveResidentPgy]:
+    """Resolve resident PGY for an academic year, preferring AY-scoped rows."""
+    stmt = (
+        select(Person, PersonAcademicYear)
+        .outerjoin(
+            PersonAcademicYear,
+            and_(
+                PersonAcademicYear.person_id == Person.id,
+                PersonAcademicYear.academic_year == academic_year,
+            ),
+        )
+        .where(Person.type == "resident")
+        .order_by(Person.name.asc())
+    )
+    people = db.execute(stmt).all()
+
+    person_ids = [person.id for person, _ in people]
+    latest_prior_by_person: dict[UUID, PersonAcademicYear] = {}
+    if person_ids:
+        latest_prior_stmt = (
+            select(PersonAcademicYear)
+            .where(
+                PersonAcademicYear.person_id.in_(person_ids),
+                PersonAcademicYear.academic_year < academic_year,
+            )
+            .order_by(
+                PersonAcademicYear.person_id.asc(),
+                PersonAcademicYear.academic_year.desc(),
+            )
+        )
+        for record in db.execute(latest_prior_stmt).scalars():
+            latest_prior_by_person.setdefault(record.person_id, record)
+
+    residents: list[EffectiveResidentPgy] = []
+    for person, person_academic_year in people:
+        if person_academic_year is not None:
+            if person_academic_year.is_graduated:
+                continue
+            effective_pgy = person_academic_year.pgy_level or person.pgy_level
+        else:
+            latest_prior = latest_prior_by_person.get(person.id)
+            if latest_prior is not None and latest_prior.is_graduated:
+                continue
+            effective_pgy = person.pgy_level
+
+        if effective_pgy not in (1, 2, 3):
+            logger.warning(
+                "Skipping resident %s during AY %s PGY resolution because the effective PGY is %s",
+                person.id,
+                academic_year,
+                effective_pgy,
+            )
+            continue
+
+        if pgy_level is not None and effective_pgy != pgy_level:
+            continue
+
+        residents.append(EffectiveResidentPgy(person=person, pgy_level=effective_pgy))
+
+    return residents