Critical Path Delta Detection

An approved change order or a delayed RFI response does not announce its schedule impact — it hides inside the next update, and by the time a project manager notices the substantial-completion milestone has slipped, the cost is already committed. The precise sub-problem this page solves is how a pipeline takes two parsed schedule updates, computes each one’s critical path with a proper forward/backward pass over the dependency graph, then diffs the results to surface exactly which activities entered or left the critical path and which milestones moved — so an approved change order’s true schedule impact is detected automatically instead of at the monthly update meeting. When that comparison is done by eye in the scheduling tool, near-critical work that just went critical is invisible until it is late, float erosion goes untracked, and the causal link between an approved change and a moved milestone is lost. Inside a deterministic schedule sync integration layer, delta detection is the analytical core: it consumes typed activity records from the parsers, runs the Critical Path Method (CPM) on each update, and emits a classified, routable set of changes. This page details that pipeline end to end — the graph contract, the CPM pass, the diff, and the confidence-scored routing — for Python automation builders, schedulers, and project controls engineers who need change impact detected the day it lands, not weeks later.

Prerequisites

This subsystem sits downstream of the schedule parsers and upstream of the impact ledger and alert router. Before implementing the patterns below, you need:

Python 3.11+ with pydantic v2 for typed validation, plus the standard-library collections, decimal, datetime, enum, uuid, and logging modules. Durations and float are integer working days; every monetary value — such as a liquidated-damages rate — is a Decimal, never a float.
Typed activity and relationship records for each update. The two schedules must arrive as a ScheduleActivity map plus an ActivityRelationship list produced by Primavera P6 export parsing or MS Project schedule parsing; this page assumes those records are already validated and does not re-parse XER or XML.
A stable activity key shared across updates. Delta detection compares activity to activity, so each record must carry the scheduling tool’s immutable activity_id, not its display name. When an activity is bound to a budgeted scope element it uses the same Work Breakdown Structure that drives WBS mapping strategies, so a moved milestone can be traced to a cost account.
A shared project calendar or an explicit note of which calendar each update used. Total float is measured in working days, so two updates computed against different calendars are not directly comparable until reconciled — the troubleshooting section below treats this as a first-class failure mode.
A task queue and dead-letter path so a delta that cannot be trusted — because an activity id churned or the graph will not build — is parked for triage rather than dropped. Escalation of parked and threshold-breaching deltas is owned by fallback alert routing.

The pipeline assumes both updates have already cleared structural validation at ingestion, so the work here is CPM computation, diffing, and routing — not schedule file parsing.

Architecture: two schedules in, classified deltas out

Delta detection is a two-input pipeline: it never looks at one schedule in isolation. Each parsed update is turned into a dependency graph, run through a forward and backward CPM pass to yield early/late dates and total float, and reduced to a critical set — the activities with zero or negative float. The two critical sets, and the two float maps behind them, are then diffed activity by activity to classify every change. The routing decision at the end uses the site-canonical confidence bands, but the confidence here measures identity: how sure the pipeline is that an activity in the current update is the same activity it was in the prior one. A stable activity_id match scores 0.92 or above and auto-routes; a fuzzy name match in the 0.75–0.92 band flags the delta for a scheduler’s review; below 0.75 the record is quarantined rather than reported against a guessed pairing.

Both updates are reduced to a critical set through the same CPM pass; the diff classifies every change, and the identity-match confidence bands decide whether a delta auto-routes to the impact log, waits for a scheduler, or is quarantined on an unstable activity id.

Stage	Input	Output	Error branch
Build DAG	Activity map + relationships	Topologically ordered node list	Cycle in graph → quarantine whole update
Forward/backward pass	Ordered graph + durations	Early/late start & finish per activity	Missing predecessor node → quarantine
Total float + critical set	Early/late dates	`total_float_days` + `is_critical` per activity	Negative float from a hard constraint → flag
Critical-set diff	Prior + current results	Classified `ActivityDelta` list	Activity id present in only one update → churn check
Route deltas	Delta + identity confidence	Auto-route / review / quarantine	`< 0.75` identity → dead-letter queue

Step-by-step implementation

Step 1 — Define the schedule graph contract

The graph is a versioned, typed contract shared by both updates. Each ScheduleActivity carries the tool’s immutable activity_id as its stable key, an integer duration_days, a milestone flag, and the wbs_node that binds it to budgeted scope. Relationships are the four CPM link types expressed as a Literal — Finish-to-Start, Start-to-Start, Finish-to-Finish, Start-to-Finish — each with an integer lag_days that may be negative (a lead). Durations and lags are integer working days because CPM arithmetic is exact day counting; a float here would let rounding drift accumulate across a thousand-activity network. A model_validator rejects any relationship that references an activity outside the map, so a dangling predecessor never reaches the CPM pass as a silent KeyError.

from __future__ import annotations

import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Literal, Optional
from uuid import UUID

from pydantic import BaseModel, Field, field_validator, model_validator

logger = logging.getLogger("schedule.delta")

WBS_PATTERN = r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$"
ACTIVITY_ID_PATTERN = r"^[A-Z]{1,4}\d{3,5}$"  # e.g. STR1010, A1000

RelType = Literal["FS", "SS", "FF", "SF"]


class ScheduleActivity(BaseModel):
    activity_id: str = Field(pattern=ACTIVITY_ID_PATTERN)
    name: str = Field(min_length=1, max_length=200)
    wbs_node: str = Field(pattern=WBS_PATTERN)
    duration_days: int = Field(ge=0)          # int working days; 0 => milestone
    is_milestone: bool = False

    @model_validator(mode="after")
    def milestone_has_zero_duration(self) -> "ScheduleActivity":
        if self.is_milestone and self.duration_days != 0:
            raise ValueError("a milestone must have duration_days == 0")
        return self


class ActivityRelationship(BaseModel):
    predecessor_id: str = Field(pattern=ACTIVITY_ID_PATTERN)
    successor_id: str = Field(pattern=ACTIVITY_ID_PATTERN)
    rel_type: RelType = "FS"
    lag_days: int = 0                          # int days; negative == lead

    @model_validator(mode="after")
    def no_self_loop(self) -> "ActivityRelationship":
        if self.predecessor_id == self.successor_id:
            raise ValueError("a relationship cannot link an activity to itself")
        return self


class ScheduleUpdate(BaseModel):
    schema_version: Literal["1.0"] = "1.0"
    project_uuid: UUID
    update_id: str = Field(pattern=r"^U\d{2}$")     # U00, U01, ...
    data_date: datetime                              # status date of this update
    calendar_id: str = Field(min_length=1)           # which working calendar
    activities: dict[str, ScheduleActivity]
    relationships: list[ActivityRelationship]

    @field_validator("data_date")
    @classmethod
    def require_tz_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("data_date must be timezone-aware (ISO 8601)")
        return v.astimezone(timezone.utc)

    @model_validator(mode="after")
    def relationships_reference_known_activities(self) -> "ScheduleUpdate":
        known = set(self.activities)
        for rel in self.relationships:
            missing = {rel.predecessor_id, rel.successor_id} - known
            if missing:
                raise ValueError(f"relationship references unknown activities: {sorted(missing)}")
        return self

Step 2 — Compute the critical path with a forward and backward pass

CPM runs in three moves: order the graph so every predecessor precedes its successors, sweep forward to find the earliest each activity can start and finish, then sweep backward to find the latest it can slip without pushing project completion. The ordering is a Kahn topological sort, which does double duty — if it cannot consume every node, the leftover set is exactly the cycle, and a cyclic network has no valid critical path. Total float is the gap between the late and early start; an activity with zero or negative float is on the critical path. Each link type shifts the arithmetic: a Start-to-Start link constrains the successor’s start, a Finish-to-Finish link its finish, and lag is added as integer days.

from collections import defaultdict, deque
from dataclasses import dataclass


class ScheduleCycleError(ValueError):
    """The activity graph contains a dependency cycle, so no CPM solution exists."""


@dataclass(frozen=True)
class CpmResult:
    early_start: dict[str, int]
    early_finish: dict[str, int]
    late_start: dict[str, int]
    late_finish: dict[str, int]
    total_float: dict[str, int]
    project_finish: int

    def critical_set(self) -> set[str]:
        # Zero or negative float == on the critical path.
        return {aid for aid, tf in self.total_float.items() if tf <= 0}


def _topological_order(
    activity_ids: list[str], relationships: list[ActivityRelationship]
) -> list[str]:
    indegree: dict[str, int] = {aid: 0 for aid in activity_ids}
    successors: dict[str, list[str]] = defaultdict(list)
    for rel in relationships:
        successors[rel.predecessor_id].append(rel.successor_id)
        indegree[rel.successor_id] += 1

    queue = deque(aid for aid in activity_ids if indegree[aid] == 0)
    order: list[str] = []
    while queue:
        node = queue.popleft()
        order.append(node)
        for succ in successors[node]:
            indegree[succ] -= 1
            if indegree[succ] == 0:
                queue.append(succ)

    if len(order) != len(activity_ids):
        stuck = sorted(set(activity_ids) - set(order))
        raise ScheduleCycleError(f"dependency cycle involves: {stuck}")
    return order


def compute_cpm(update: ScheduleUpdate) -> CpmResult:
    durations = {aid: act.duration_days for aid, act in update.activities.items()}
    order = _topological_order(list(update.activities), update.relationships)

    preds_of: dict[str, list[ActivityRelationship]] = defaultdict(list)
    succs_of: dict[str, list[ActivityRelationship]] = defaultdict(list)
    for rel in update.relationships:
        preds_of[rel.successor_id].append(rel)
        succs_of[rel.predecessor_id].append(rel)

    # Forward pass: earliest start/finish honouring every incoming link type.
    early_start: dict[str, int] = {}
    early_finish: dict[str, int] = {}
    for aid in order:
        dur = durations[aid]
        es = 0
        for rel in preds_of[aid]:
            p = rel.predecessor_id
            if rel.rel_type == "FS":
                cand = early_finish[p] + rel.lag_days
            elif rel.rel_type == "SS":
                cand = early_start[p] + rel.lag_days
            elif rel.rel_type == "FF":
                cand = early_finish[p] + rel.lag_days - dur
            else:  # SF
                cand = early_start[p] + rel.lag_days - dur
            es = max(es, cand)
        early_start[aid] = es
        early_finish[aid] = es + dur

    project_finish = max(early_finish.values()) if early_finish else 0

    # Backward pass: latest finish/start without moving project completion.
    late_finish: dict[str, int] = {}
    late_start: dict[str, int] = {}
    for aid in reversed(order):
        dur = durations[aid]
        lf = project_finish
        for rel in succs_of[aid]:
            s = rel.successor_id
            if rel.rel_type == "FS":
                cand = late_start[s] - rel.lag_days
            elif rel.rel_type == "SS":
                cand = late_start[s] - rel.lag_days + dur
            elif rel.rel_type == "FF":
                cand = late_finish[s] - rel.lag_days
            else:  # SF
                cand = late_finish[s] - rel.lag_days + dur
            lf = min(lf, cand)
        late_finish[aid] = lf
        late_start[aid] = lf - dur

    total_float = {aid: late_start[aid] - early_start[aid] for aid in order}
    return CpmResult(
        early_start=early_start,
        early_finish=early_finish,
        late_start=late_start,
        late_finish=late_finish,
        total_float=total_float,
        project_finish=project_finish,
    )

Step 3 — Diff the two updates into classified deltas

With a CpmResult for each update, the diff walks the union of activity ids and classifies every change against a stable key. An activity that was off the critical path and is now on it is entered_cp; the reverse is left_cp; one that stayed critical but lost float is still_critical; a non-critical activity whose float shrank past a near-critical threshold is float_eroded; and a milestone whose finish day moved is milestone_moved. An id present in only one update is not a delta but a churn signal — it feeds the identity check in Step 4 rather than being reported as a real schedule change. Each delta carries both float values and both finish days so the downstream log records the magnitude, not just the direction.

DeltaKind = Literal[
    "entered_cp", "left_cp", "still_critical", "float_eroded", "milestone_moved"
]

NEAR_CRITICAL_DAYS = 5  # float at or below this (but > 0) counts as eroding toward critical


class ActivityDelta(BaseModel):
    activity_id: str = Field(pattern=ACTIVITY_ID_PATTERN)
    kind: DeltaKind
    prior_float_days: Optional[int] = None
    current_float_days: Optional[int] = None
    prior_finish_day: Optional[int] = None
    current_finish_day: Optional[int] = None
    identity_confidence: Decimal = Field(ge=0, le=1)

    @property
    def slip_days(self) -> int:
        if self.prior_finish_day is None or self.current_finish_day is None:
            return 0
        return self.current_finish_day - self.prior_finish_day


def diff_schedules(
    prior: ScheduleUpdate,
    current: ScheduleUpdate,
    prior_cpm: CpmResult,
    current_cpm: CpmResult,
    identity: dict[str, Decimal],
) -> list[ActivityDelta]:
    prior_crit = prior_cpm.critical_set()
    curr_crit = current_cpm.critical_set()
    deltas: list[ActivityDelta] = []

    for aid in current.activities:
        if aid not in prior.activities:
            continue  # new or churned id — handled by the identity check, not a delta
        conf = identity.get(aid, Decimal("1.0"))
        was, now = aid in prior_crit, aid in curr_crit
        pf, cf = prior_cpm.total_float[aid], current_cpm.total_float[aid]
        pfin, cfin = prior_cpm.early_finish[aid], current_cpm.early_finish[aid]

        kind: Optional[DeltaKind] = None
        if not was and now:
            kind = "entered_cp"
        elif was and not now:
            kind = "left_cp"
        elif was and now and cf < pf:
            kind = "still_critical"
        elif not now and 0 < cf <= NEAR_CRITICAL_DAYS and cf < pf:
            kind = "float_eroded"
        elif current.activities[aid].is_milestone and cfin != pfin:
            kind = "milestone_moved"

        if kind is not None:
            deltas.append(
                ActivityDelta(
                    activity_id=aid, kind=kind,
                    prior_float_days=pf, current_float_days=cf,
                    prior_finish_day=pfin, current_finish_day=cfin,
                    identity_confidence=conf,
                )
            )
    return deltas

Step 4 — Route each delta by identity confidence and impact

The final stage turns a classified delta into an action. Identity confidence gates first: a delta whose activity pairing scored below 0.75 is quarantined, because reporting a critical-path change against the wrong activity is worse than reporting nothing. For a trusted delta, magnitude decides escalation. A milestone slip is priced with an integer day count times a Decimal liquidated-damages rate; when an approved change order is in play and that impact crosses the Decimal("5000.00") threshold, the delta auto-routes as a change-order schedule impact. Everything else in the 0.75–0.92 band waits for a scheduler. The function returns an action rather than writing anything, so the queue can retry it safely; the actual write goes through schedule impact logging, and the baseline comparison it feeds is owned by schedule baseline variance.

AUTO_ROUTE = Decimal("0.92")     # >= 0.92 identity -> auto-route the delta
REVIEW_FLOOR = Decimal("0.75")   # 0.75-0.92 -> scheduler review; < 0.75 -> quarantine
IMPACT_THRESHOLD = Decimal("5000.00")  # change-order schedule-impact escalation

RouteAction = Literal["auto_route", "human_review", "quarantine", "file_for_record"]


def route_delta(
    delta: ActivityDelta,
    daily_delay_cost: Decimal,
    change_order_pending: bool,
) -> RouteAction:
    if delta.identity_confidence < REVIEW_FLOOR:
        logger.warning(
            "delta.quarantine",
            extra={"activity": delta.activity_id, "conf": str(delta.identity_confidence)},
        )
        return "quarantine"

    slip = max(delta.slip_days, 0)
    cost_impact = daily_delay_cost * Decimal(slip)
    escalate = (
        delta.kind in ("entered_cp", "milestone_moved")
        and change_order_pending
        and cost_impact >= IMPACT_THRESHOLD
    )

    if delta.identity_confidence >= AUTO_ROUTE and escalate:
        logger.info(
            "delta.change_order_impact",
            extra={"activity": delta.activity_id, "impact": str(cost_impact), "slip": slip},
        )
        return "auto_route"
    if delta.identity_confidence >= AUTO_ROUTE and delta.kind in ("entered_cp", "milestone_moved"):
        return "auto_route"      # critical-path change we trust, log it without a CO
    if delta.identity_confidence < AUTO_ROUTE:
        return "human_review"    # trusted pairing but fuzzy — a scheduler confirms
    return "file_for_record"     # left_cp / still_critical with no dollar impact

Schema and configuration reference

Field	Type / pattern	Rule	Why it matters
`activity_id`	`^[A-Z]{1,4}\d{3,5}$`	Immutable across updates	The only reliable key to diff activity to activity
`duration_days`	`int`, `ge=0`	Working days, `0` for milestones	CPM is exact day arithmetic; a float would drift
`lag_days`	`int`	May be negative (lead)	Shifts each link type’s start/finish constraint
`rel_type`	`Literal[FS,SS,FF,SF]`	Closed set	Each type changes the forward/backward equation
`calendar_id`	`str`	Recorded per update	Two updates on different calendars are not comparable
`total_float`	`int` days	`<= 0` ⇒ critical	Defines the critical set on each side of the diff
`kind`	`DeltaKind` Literal	One classification per changed activity	Drives whether a delta escalates
`identity_confidence`	`Decimal` 0–1	Stable id ⇒ `1.0`; fuzzy ⇒ lower	Gates the routing decision
`daily_delay_cost`	`Decimal`	Never a float	Prices a milestone slip against the threshold

Routing constants are site-canonical and appear identically wherever delta logic runs: AUTO_ROUTE = Decimal("0.92"), REVIEW_FLOOR = Decimal("0.75"), IMPACT_THRESHOLD = Decimal("5000.00"), and a per-project NEAR_CRITICAL_DAYS window (default 5) that defines float erosion.

Verification and testing

CPM and the diff are pure functions of their inputs, so they test cleanly against a hand-computed network. Build a tiny three-activity chain, assert the float and critical set, then push one activity’s duration up and confirm the diff reports the successor entering the critical path.

from decimal import Decimal
from uuid import uuid4
from datetime import datetime, timezone


def _update(update_id: str, dur_b: int) -> ScheduleUpdate:
    acts = {
        "A1000": ScheduleActivity(activity_id="A1000", name="Mobilize",
                                  wbs_node="PROJ-014-CIV-01", duration_days=5),
        "A1010": ScheduleActivity(activity_id="A1010", name="Foundations",
                                  wbs_node="PROJ-014-STR-02", duration_days=dur_b),
        "A1020": ScheduleActivity(activity_id="A1020", name="Substantial Completion",
                                  wbs_node="PROJ-014-STR-02", duration_days=0, is_milestone=True),
    }
    rels = [
        ActivityRelationship(predecessor_id="A1000", successor_id="A1010"),
        ActivityRelationship(predecessor_id="A1010", successor_id="A1020"),
    ]
    return ScheduleUpdate(project_uuid=uuid4(), update_id=update_id,
                          data_date=datetime(2026, 7, 1, tzinfo=timezone.utc),
                          calendar_id="STD-5D", activities=acts, relationships=rels)


def test_linear_chain_is_all_critical():
    cpm = compute_cpm(_update("U01", dur_b=10))
    assert cpm.project_finish == 15
    assert cpm.critical_set() == {"A1000", "A1010", "A1020"}


def test_cycle_raises():
    up = _update("U01", dur_b=10)
    up.relationships.append(ActivityRelationship(predecessor_id="A1020", successor_id="A1000"))
    try:
        compute_cpm(up)
        assert False, "expected a cycle error"
    except ScheduleCycleError as exc:
        assert "A1000" in str(exc)


def test_milestone_slip_is_detected_and_priced():
    prior, current = _update("U01", dur_b=10), _update("U02", dur_b=14)
    pc, cc = compute_cpm(prior), compute_cpm(current)
    identity = {aid: Decimal("1.0") for aid in current.activities}
    deltas = {d.activity_id: d for d in diff_schedules(prior, current, pc, cc, identity)}
    milestone = deltas["A1020"]
    assert milestone.kind == "milestone_moved"
    assert milestone.slip_days == 4
    action = route_delta(milestone, daily_delay_cost=Decimal("2000.00"), change_order_pending=True)
    assert action == "auto_route"  # 4 days * 2000 = 8000 >= 5000

Run the suite with pytest -q. For a manual smoke test, serialize a delta list with model_dump_json() and confirm each slip_days and identity_confidence round-trips through ActivityDelta.model_validate_json().

Troubleshooting

ScheduleCycleError on an update that opens fine in the scheduling tool. A redundant or miskeyed relationship closed a loop the tool tolerated but CPM cannot. The exception names the stuck activities; inspect their in-edges, drop the relationship that reverses the intended flow, and re-run. Never “fix” a cycle by dropping a random edge — quarantine the whole update so a scheduler removes the correct link, because a guessed removal silently changes the critical path.
Every activity’s float shifts by the same amount between updates. The two updates were computed against different calendars — one counted a holiday week as working days, the other did not — so their day indices are not on the same scale. Reconcile to a single calendar_id before diffing; comparing float across calendars produces phantom float_eroded deltas for activities that never actually moved.
A real critical-path change is reported as churn and dropped. The activity’s id changed between updates — a common P6 artifact when an activity is deleted and re-added rather than edited. Because the diff keys on activity_id, the pair looks like one deletion plus one addition. Feed a fuzzy name-and-WBS match into identity_confidence so the pairing lands in the review band instead of vanishing; comparing by display name alone, covered in the sibling page on detecting critical path changes between schedule updates, is what causes the drop.
Negative total float appears after an approved change order. A hard constraint — a “finish no later than” date the change order did not move — now sits earlier than the CPM-computed finish, so late dates fall behind early dates and float goes negative. That is correct output, not a bug: negative float is the network telling you the constraint is already unachievable. Route those activities to review rather than clamping float at zero, which would hide the overrun.
Duplicate impact records from one pair of updates. A redelivered broker message re-ran the diff and the routing published twice. Make the outer task idempotent with a commit keyed on the two update_id values plus activity_id, so a replay is a no-op even though the pure functions happily return the same verdict.

Frequently Asked Questions

Why compute the critical path from scratch instead of trusting the tool's flag?

Primavera P6 and MS Project each store a longest-path or critical flag, but they compute it under their own calendar, constraint, and multiple-float-path settings, which differ between the two files you are comparing. Running one deterministic CPM pass over both updates guarantees the critical sets are measured the same way, so a diff reflects a real schedule change rather than a difference in tool configuration.

How do the confidence bands apply to a critical-path delta?

They gate identity, not the schedule math. A delta whose activity was matched by a stable activity_id scores 0.92 or above and auto-routes. A pairing recovered by a fuzzy name-and-WBS match lands in the 0.75–0.92 band and waits for a scheduler to confirm. Below 0.75 the delta is quarantined, because reporting a critical-path change against the wrong activity is worse than reporting none.

Why are durations and float integers rather than Decimals?

CPM is exact whole-day arithmetic over a working calendar: an activity takes a whole number of working days, and float is a difference of two day indices. Integers keep that arithmetic exact across a large network with no rounding drift. Decimal is reserved for money — the liquidated-damages rate that prices a milestone slip — where fractional currency precision genuinely matters.

What counts as a milestone move versus float erosion?

A milestone move is a change in the computed finish day of a zero-duration milestone activity — the date a contractual event lands. Float erosion is a non-critical activity losing total float toward zero without yet crossing it; the NEAR_CRITICAL_DAYS window (default five days) flags it early so a path trending critical surfaces before it actually goes critical, which the companion page on calculating total float erosion across schedule updates develops in full.

Why route a delta instead of writing the impact directly?

Keeping route_delta a pure function that returns an action makes it testable and safe to retry: the broker can redeliver a message and the verdict is identical. The side effects — appending to the impact log, triggering a change-order draft, paging a scheduler — happen in a thin outer task that commits idempotently, so one pair of updates can never generate two impact records.

← Back to Schedule Sync & Critical Path Integration

Critical Path Delta Detection

Explore in this section