"""Deterministic bidirectional resource forecast for employee transfers. The kernel is datastore-free and receives anonymous people/resources plus the current and target policy state. It never grants or revokes anything: output is a versioned explanation input for Helix recommendations, whose execution continues through the existing provisioning and human-review gates. """ from __future__ import annotations from dataclasses import dataclass from datetime import date, timedelta from typing import Literal from app.intelligence.demand_forecast import ResourceKey TRANSFER_FORECAST_VERSION = "transfer-forecast-v1" TRANSFER_HORIZONS = (30, 60, 90) Confidence = Literal["insufficient", "low", "medium", "high"] Direction = Literal["add", "recover", "retain"] Action = Literal["grant", "recover", "retain", "review"] @dataclass(frozen=True) class CurrentHolding: resource: ResourceKey retain_on_transfer: bool = False review_on_transfer: bool = False @dataclass(frozen=True) class TransferCandidate: """One effective or planned transfer with anonymous local references.""" ref: str effective_date: date current_resources: tuple[CurrentHolding, ...] target_resources: frozenset[ResourceKey] target_confidence: Confidence = "high" target_complete: bool = True employment_status: str = "active" cancelled: bool = False anomalous: bool = False @dataclass(frozen=True) class TransferDelta: resource: ResourceKey direction: Direction action: Action confidence: Confidence reason: str @dataclass(frozen=True) class TransferProjection: candidate_ref: str effective_date: date overdue: bool review_required: bool deltas: tuple[TransferDelta, ...] @dataclass(frozen=True) class TransferResourceSummary: resource: ResourceKey add_count: int recover_count: int retain_count: int review_count: int @dataclass(frozen=True) class TransferHorizon: horizon_days: int cutoff_date: date transfer_count: int overdue_count: int incomplete_count: int add_count: int recover_count: int retain_count: int review_count: int transfers: tuple[TransferProjection, ...] resources: tuple[TransferResourceSummary, ...] @dataclass(frozen=True) class TransferForecast: forecast_version: str as_of_date: date horizons: tuple[TransferHorizon, ...] def _clean(value: str, field: str) -> str: cleaned = value.strip().casefold() if not cleaned: raise ValueError(f"{field} must not be empty") if len(cleaned) > 128: raise ValueError(f"{field} must be at most 128 characters") return cleaned def _resource(value: ResourceKey) -> ResourceKey: if value.scope_mode not in ("global", "location"): raise ValueError("resource scope_mode must be global or location") return ResourceKey( kind=_clean(value.kind, "resource kind"), key=_clean(value.key, "resource key"), scope_mode=value.scope_mode, ) def _normalise(rows: tuple[TransferCandidate, ...]) -> tuple[TransferCandidate, ...]: refs: set[str] = set() result: list[TransferCandidate] = [] for row in rows: ref = _clean(row.ref, "transfer ref") if ref in refs: raise ValueError(f"duplicate transfer ref: {ref}") refs.add(ref) if row.target_confidence not in ("insufficient", "low", "medium", "high"): raise ValueError("invalid target confidence") current: list[CurrentHolding] = [] current_keys: set[ResourceKey] = set() for holding in row.current_resources: resource = _resource(holding.resource) if resource in current_keys: raise ValueError(f"duplicate current resource for {ref}: {resource}") current_keys.add(resource) if holding.retain_on_transfer and holding.review_on_transfer: raise ValueError("holding cannot be both retained and manually reviewed") current.append(CurrentHolding( resource, holding.retain_on_transfer, holding.review_on_transfer, )) target = frozenset(_resource(resource) for resource in row.target_resources) result.append(TransferCandidate( ref=ref, effective_date=row.effective_date, current_resources=tuple(sorted(current, key=lambda item: item.resource)), target_resources=target, target_confidence=row.target_confidence, target_complete=row.target_complete, employment_status=row.employment_status.strip().casefold(), cancelled=row.cancelled, anomalous=row.anomalous, )) return tuple(result) def _project(row: TransferCandidate, *, as_of_date: date) -> TransferProjection: if not row.target_complete or row.target_confidence == "insufficient": return TransferProjection( row.ref, row.effective_date, row.effective_date < as_of_date, True, (), ) current = {holding.resource: holding for holding in row.current_resources} additions = row.target_resources - current.keys() obsolete = current.keys() - row.target_resources review_only = row.target_confidence == "low" deltas: list[TransferDelta] = [] for resource in sorted(additions): deltas.append(TransferDelta( resource=resource, direction="add", action="review" if review_only else "grant", confidence=row.target_confidence, reason="target_policy_missing", )) for resource in sorted(obsolete): holding = current[resource] # Existing Helix policy keeps physical assets with a mover. The # kernel records that fact explicitly instead of silently proposing # an unsafe automatic recovery. manual_review = holding.review_on_transfer retained = not manual_review and ( holding.retain_on_transfer or resource.kind == "hardware" ) deltas.append(TransferDelta( resource=resource, direction="retain" if retained else "recover", action=( "review" if review_only or manual_review else ("retain" if retained else "recover") ), confidence=row.target_confidence, reason=( "transfer_manual_review" if manual_review else "transfer_retention_policy" if retained else "target_policy_unmatched" ), )) return TransferProjection( candidate_ref=row.ref, effective_date=row.effective_date, overdue=row.effective_date < as_of_date, review_required=review_only or any(item.action == "review" for item in deltas), deltas=tuple(deltas), ) def build_transfer_forecast( *, as_of_date: date, transfers: tuple[TransferCandidate, ...], ) -> TransferForecast: """Build cumulative 30/60/90-day add/recover/retain projections.""" normalised = _normalise(transfers) eligible = tuple( row for row in normalised if not row.cancelled and not row.anomalous and row.employment_status == "active" and as_of_date - timedelta(days=90) <= row.effective_date <= as_of_date + timedelta(days=90) ) horizons: list[TransferHorizon] = [] for days in TRANSFER_HORIZONS: cutoff = as_of_date + timedelta(days=days) rows = tuple(sorted( (row for row in eligible if row.effective_date <= cutoff), key=lambda row: (row.effective_date, row.ref), )) projections = tuple(_project(row, as_of_date=as_of_date) for row in rows) counts: dict[ResourceKey, dict[Direction | Literal["review"], int]] = {} incomplete = sum(not row.target_complete or row.target_confidence == "insufficient" for row in rows) for projection in projections: for delta in projection.deltas: bucket = counts.setdefault(delta.resource, {"add": 0, "recover": 0, "retain": 0, "review": 0}) bucket[delta.direction] += 1 if delta.action == "review": bucket["review"] += 1 summaries = tuple( TransferResourceSummary( resource=resource, add_count=bucket["add"], recover_count=bucket["recover"], retain_count=bucket["retain"], review_count=bucket["review"], ) for resource, bucket in sorted(counts.items()) ) horizons.append(TransferHorizon( horizon_days=days, cutoff_date=cutoff, transfer_count=len(rows), overdue_count=sum(projection.overdue for projection in projections), incomplete_count=incomplete, add_count=sum(item.add_count for item in summaries), recover_count=sum(item.recover_count for item in summaries), retain_count=sum(item.retain_count for item in summaries), review_count=( sum(projection.review_required and not projection.deltas for projection in projections) + sum(item.review_count for item in summaries) ), transfers=projections, resources=summaries, )) return TransferForecast(TRANSFER_FORECAST_VERSION, as_of_date, tuple(horizons)) __all__ = [ "CurrentHolding", "TRANSFER_FORECAST_VERSION", "TRANSFER_HORIZONS", "TransferCandidate", "TransferDelta", "TransferForecast", "TransferHorizon", "TransferProjection", "TransferResourceSummary", "build_transfer_forecast", ]