Detecting Out-of-Sequence Progress in Schedule Updates

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from typing import Literal, Optional

from pydantic import BaseModel, ConfigDict, Field

logging.basicConfig(level=logging.INFO, format="%(levelname)-8s | %(message)s")
logger = logging.getLogger("schedule.sequence")

RelType = Literal["FS", "SS", "FF", "SF"]
RouteAction = Literal["escalate_alert", "flag_report", "human_review", "quarantine"]

AUTO_ROUTE = Decimal("0.92")     # >= 0.92 -> confident detection
REVIEW_FLOOR = Decimal("0.75")   # 0.75–0.92 -> human review; < 0.75 -> quarantine


@dataclass(frozen=True)
class ActivityProgress:
    """An activity's reported progress from a parsed schedule update."""

    activity_id: str
    is_critical: bool
    actual_start: Optional[datetime]
    actual_finish: Optional[datetime]
    percent_complete: Decimal

    def started(self) -> bool:
        return self.actual_start is not None or self.percent_complete > 0

    def finished(self) -> bool:
        return self.actual_finish is not None or self.percent_complete >= 100


@dataclass(frozen=True)
class Relationship:
    """A precedence link. lag_days may be negative (a lead permitting overlap)."""

    predecessor_id: str
    successor_id: str
    rel_type: RelType
    lag_days: int = 0


class SequenceViolation(BaseModel):
    """One detected out-of-sequence progress event, carrying its routing decision."""

    model_config = ConfigDict(frozen=True)

    successor_id: str
    predecessor_id: str
    rel_type: RelType
    reason: str
    is_critical: bool
    confidence: Decimal = Field(ge=0, le=1, decimal_places=2)

    @property
    def route_state(self) -> RouteAction:
        if self.confidence < REVIEW_FLOOR:
            return "quarantine"
        if self.confidence < AUTO_ROUTE:
            return "human_review"
        return "escalate_alert" if self.is_critical else "flag_report"


def _violation_reason(rel: Relationship, pred: ActivityProgress, succ: ActivityProgress) -> str | None:
    """Return a reason if the relationship's gating event has not occurred, else None."""
    # FS and SS gate the successor's START on a predecessor event.
    if rel.rel_type == "FS" and succ.started() and not pred.finished():
        return "successor started before FS predecessor finished"
    if rel.rel_type == "SS" and succ.started() and not pred.started():
        return "successor started before SS predecessor started"
    # FF and SF gate the successor's FINISH on a predecessor event.
    if rel.rel_type == "FF" and succ.finished() and not pred.finished():
        return "successor finished before FF predecessor finished"
    if rel.rel_type == "SF" and succ.finished() and not pred.started():
        return "successor finished before SF predecessor started"
    return None


def _confidence(rel: Relationship, relogicked_ids: frozenset[str]) -> Decimal:
    """Score how sure we are the detection is a real logic break, not legitimate re-sequencing."""
    if rel.successor_id in relogicked_ids or rel.predecessor_id in relogicked_ids:
        return Decimal("0.70")   # logic changed this period -> likely a deliberate re-sequence
    if rel.lag_days < 0:
        return Decimal("0.80")   # a lead permits overlap -> a human must confirm the dates
    return Decimal("0.97")


def detect_out_of_sequence(
    activities: list[ActivityProgress],
    relationships: list[Relationship],
    relogicked_ids: frozenset[str] = frozenset(),
) -> list[SequenceViolation]:
    """Find every in-progress activity that has run ahead of its predecessors."""
    by_id = {a.activity_id: a for a in activities}
    violations: list[SequenceViolation] = []

    for rel in relationships:
        pred, succ = by_id.get(rel.predecessor_id), by_id.get(rel.successor_id)
        if pred is None or succ is None:
            logger.warning(
                "dangling relationship %s -> %s", rel.predecessor_id, rel.successor_id
            )
            continue

        reason = _violation_reason(rel, pred, succ)
        if reason is None:
            continue

        violations.append(
            SequenceViolation(
                successor_id=rel.successor_id,
                predecessor_id=rel.predecessor_id,
                rel_type=rel.rel_type,
                reason=reason,
                is_critical=succ.is_critical,
                confidence=_confidence(rel, relogicked_ids),
            )
        )
    return violations


if __name__ == "__main__":
    def _dt(y: int, m: int, d: int) -> datetime:
        return datetime(y, m, d, 12, 0, tzinfo=timezone.utc)

    activities = [
        # Formwork not finished, but the pour it gates has already started: FS break.
        ActivityProgress("A100", True, _dt(2026, 4, 6), None, Decimal("60")),
        ActivityProgress("A110", True, _dt(2026, 4, 8), None, Decimal("25")),
        # A wall started before its SS predecessor even began — but it was re-logicked.
        ActivityProgress("A200", False, None, None, Decimal("0")),
        ActivityProgress("A210", False, _dt(2026, 4, 9), None, Decimal("10")),
    ]
    relationships = [
        Relationship("A100", "A110", "FS", lag_days=0),
        Relationship("A200", "A210", "SS", lag_days=0),
    ]
    for v in detect_out_of_sequence(activities, relationships, relogicked_ids=frozenset({"A210"})):
        print(f"{v.successor_id}: {v.reason} (conf {v.confidence}) -> {v.route_state}")