Schedule Impact Logging

When a job finishes late, the dispute is never about the calendar — it is about causation. The general contractor and the owner both agree the roof deck slipped eleven days; they violently disagree about why. The specific sub-problem this page solves is how a pipeline produces an immutable, audit-ready log of every schedule impact — the finish-date movement, the critical-path change, and the float consumption that a specific change order or a delayed RFI caused — so a contractor can prove which event drove a delay on a given date. That log is the evidence that decides a time-extension claim: without it, an entitlement argument collapses into two spreadsheets and a deposition. Inside the broader work of schedule sync integration, impact logging is the stage that turns a raw schedule difference into an attributed, tamper-evident fact. It consumes the finish-date and float deltas emitted by critical path delta detection, the approved change orders lifted by automated document ingestion, and the RFI records defined in RFI schema design, then binds them into one append-only record per impact. This page details that record contract, the hashing that makes it defensible, the confidence-scored attribution that decides whether an impact auto-commits or waits for a scheduler, and the severity thresholds that decide when a slip becomes an alert. It targets Python automation builders, project controls engineers, and schedulers who need delay evidence that survives audit.

Prerequisites

Impact logging sits downstream of delta detection and document parsing and upstream of the claims register and the alert router. Before implementing the patterns below, you need:

• Python 3.11+ with pydantic v2 for typed validation, plus the standard-library hashlib, decimal, datetime, uuid, enum, and logging modules. Every finish-date movement and float figure is an integer number of days; any cost carried alongside an impact is a Decimal, never a float.
• A delta-detection feed. Each impact begins as a computed difference between two schedule updates — a moved finish date, a changed is_critical flag, or consumed total_float_days — produced by critical path delta detection. The logger records deltas; it does not recompute the network.
• Parsed source events to attribute against. A slip is only evidence if it names a cause. Approved change orders arrive already typed from automated document ingestion, and RFI records — with their submitted, due, and answered timestamps — follow the contract in RFI schema design.
• A canonical WBS binding so every impact posts against a real scope element. Activities resolve to the same PROJ-NNN-DIV-NN nodes produced by WBS mapping strategies; an impact with no WBS node cannot be rolled into a discipline-level claim.
• An append-only store and an escalation path. The audit log is write-once — an object store with versioning, or an append-only table — and the alert side hands SLA and severity breaches to fallback alert routing so escalation policy lives in one place.

The pipeline assumes each inbound delta already carries the two schedule-update identities it was computed from, so the work here is attribution, hashing, and routing — not schedule parsing.

Architecture: attribution, hashing, and routing

An impact record is not a log line; it is a claim about causation that has to survive cross-examination. Two forces pull on the design. First, the record must be immutable and chained — once an eleven-day slip is attributed to a change order and committed, no later edit can quietly rewrite the delta or the date without breaking a hash that an auditor can recompute. Second, the attribution itself is uncertain — a finish date moved on the same week a change order was approved and an RFI ran late, and the pipeline must score how confident it is that a given event drove the delta before it stamps a cause. Keeping these separate is what lets a low-confidence attribution wait for a scheduler without ever weakening the tamper-evidence of the records that did commit. The flow below runs a delta plus its candidate source through record construction, a confidence gate, severity routing, and into the append-only store.

The routing decision uses the site-canonical confidence bands. A cause-attribution score of 0.92 or above auto-commits the impact with its named cause; a 0.75–0.92 score writes the record but flags it for a scheduler to confirm the attribution; and below 0.75 the record is quarantined as unattributed rather than committed against a guessed cause that a claims consultant would later demolish.

Step-by-step implementation

Step 1 — Define the append-only impact record contract

The record is a versioned, frozen contract: once constructed it cannot be mutated, which is the whole point of an audit log. A schema_version at the root lets a field addition ship without breaking downstream consumers. The event_time — when the impact actually occurred on the schedule — is timezone-aware per the ISO 8601 date and time standard, because a naive timestamp would corrupt delay math across sites in different zones and hand the other side an easy attack on the claim. The cause is a closed Literal vocabulary, delta_days and float_consumed_days are integer days (finish movement, positive for a slip), and the source linkage is enforced by a model validator so a change_order cause can never be logged without a change_order_id.

from __future__ import annotations

import hashlib
import logging
from datetime import datetime, timezone
from typing import Literal, Optional
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("schedule.impact")

WBS_PATTERN = r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$"

ImpactCause = Literal["change_order", "rfi_delay", "resequence", "weather", "other"]
RoutingState = Literal["auto_commit", "human_review", "quarantine"]


class ScheduleImpactRecord(BaseModel):
    """One immutable, hash-chained record of a single schedule impact."""

    model_config = ConfigDict(frozen=True)

    schema_version: Literal["1.0"] = "1.0"
    record_id: UUID
    project_uuid: UUID
    activity_id: str = Field(min_length=1, max_length=40)
    wbs_node: str = Field(pattern=WBS_PATTERN)
    event_time: datetime          # when the impact occurred on the schedule
    logged_at: datetime           # when this record was written
    cause: ImpactCause
    delta_days: int               # finish-date movement; positive = slip
    float_consumed_days: int = Field(ge=0)
    critical_path_changed: bool
    change_order_id: Optional[str] = Field(default=None, max_length=40)
    rfi_id: Optional[str] = Field(default=None, max_length=40)
    link_confidence: float = Field(ge=0.0, le=1.0)
    prev_hash: str = Field(pattern=r"^[0-9a-f]{64}$")

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

    @model_validator(mode="after")
    def cause_requires_linkage(self) -> "ScheduleImpactRecord":
        # A named cause must carry the id that proves it, or the record is
        # not evidence — it is an assertion. Quarantine unattributed causes.
        if self.cause == "change_order" and not self.change_order_id:
            raise ValueError("change_order cause requires change_order_id")
        if self.cause == "rfi_delay" and not self.rfi_id:
            raise ValueError("rfi_delay cause requires rfi_id")
        return self

    def content_digest(self) -> str:
        """Deterministic hash over the record's evidentiary content + prev_hash."""
        payload = self.model_dump_json(exclude={"link_confidence"})
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()

The digest deliberately excludes link_confidence — a scheduler may confirm an attribution and nudge the confidence without that being a substantive edit to the delay evidence — but includes every date, delta, and linkage id that a claim would rely on.

Step 2 — Chain records for tamper-evidence

An append-only log is only defensible if a reader can prove no record was altered or removed after the fact. Each record’s prev_hash points at the previous record’s content_digest, so the log is a hash chain: change one delta_days value in the middle and every subsequent digest fails to verify. The genesis record chains from a fixed zero hash. verify_chain recomputes the whole chain and returns the first break, which is exactly what an auditor runs.

GENESIS_HASH = "0" * 64


class ImpactLog:
    """Append-only, hash-chained store of ScheduleImpactRecords."""

    def __init__(self) -> None:
        self._records: list[ScheduleImpactRecord] = []

    @property
    def head_hash(self) -> str:
        return self._records[-1].content_digest() if self._records else GENESIS_HASH

    def append(self, record: ScheduleImpactRecord) -> str:
        # The caller must have built the record with prev_hash == head_hash.
        if record.prev_hash != self.head_hash:
            raise ValueError(
                f"chain break: prev_hash {record.prev_hash[:12]} "
                f"does not match head {self.head_hash[:12]}"
            )
        self._records.append(record)
        digest = record.content_digest()
        logger.info("impact.appended | activity=%s digest=%s", record.activity_id, digest[:12])
        return digest

    def verify_chain(self) -> Optional[int]:
        """Return the index of the first tampered record, or None if intact."""
        expected = GENESIS_HASH
        for idx, rec in enumerate(self._records):
            if rec.prev_hash != expected:
                return idx
            expected = rec.content_digest()
        return None

Because records are frozen, tampering means replacing an object in the list — which verify_chain catches at the first altered digest. A production store swaps the in-memory list for an append-only table or a versioned object bucket, but the chaining contract is identical.

Step 3 — Attribute a delta to a source by confidence

A delta event arrives with the finish movement and float change; a candidate source arrives with its own timeline — a change order’s approval date, or an RFI’s overdue answer. Attribution scores how well the source explains the delta: an event whose effect lands squarely on the impacted activity within the expected window scores high; a coincidence a month away scores low. This is where the site-canonical bands govern behavior — 0.92 and above auto-commits the cause, 0.75–0.92 writes the record but flags a scheduler to confirm, and below 0.75 the record is built with cause "other" and quarantined so no delay is ever attributed to an event on a guess.

from datetime import timedelta

AUTO_COMMIT = 0.92      # >= 0.92  -> commit with named cause
REVIEW_FLOOR = 0.75     # 0.75-0.92 -> scheduler confirms; < 0.75 -> quarantine


def attribution_confidence(
    delta_time: datetime,
    source_time: datetime,
    delta_on_impacted_activity: bool,
    expected_window_days: int = 14,
) -> float:
    """Score how well a source event explains a schedule delta."""
    # Temporal proximity: a source acting well before the delta is a weak cause.
    gap_days = abs((delta_time - source_time).total_seconds()) / 86_400
    if gap_days > expected_window_days:
        proximity = 0.4
    else:
        proximity = 1.0 - (gap_days / expected_window_days) * 0.3  # 1.0 -> 0.7
    # A source that touches the very activity that moved is far stronger evidence.
    linkage = 1.0 if delta_on_impacted_activity else 0.55
    return round(min(proximity, linkage), 4)


def classify(score: float) -> RoutingState:
    if score >= AUTO_COMMIT:
        return "auto_commit"
    if score >= REVIEW_FLOOR:
        return "human_review"
    return "quarantine"

The score takes the weaker of temporal proximity and structural linkage, so a change order approved on exactly the right day still fails attribution if it never touched the activity that actually slipped — which is precisely the mistake a claims consultant looks for.

Step 4 — Route by severity and trigger alerts

Once a record is committed to the chain, severity decides whether anyone needs to know now. A slip that consumes float without moving the project finish is logged quietly; a slip that changes the critical path, or that exceeds the project’s alert_threshold_days, raises a delay-watch alert. Alerts are not sent from here — the function returns an action and the escalation itself is owned by the fallback alert router, so notification policy lives in one place rather than scattered across every logger.

ALERT_THRESHOLD_DAYS = 5        # per-project configurable slip that warrants a heads-up

SeverityAction = Literal["file_only", "raise_alert"]


def severity_action(record: ScheduleImpactRecord) -> SeverityAction:
    # Any critical-path movement is always alert-worthy: it moves the finish date.
    if record.critical_path_changed:
        logger.warning(
            "impact.critical | activity=%s delta_days=%d cause=%s",
            record.activity_id, record.delta_days, record.cause,
        )
        return "raise_alert"
    if record.delta_days >= ALERT_THRESHOLD_DAYS:
        logger.info(
            "impact.threshold | activity=%s delta_days=%d",
            record.activity_id, record.delta_days,
        )
        return "raise_alert"
    return "file_only"

Returning an action rather than firing a notification keeps the function pure and testable, and lets the queue retry the outer task without double-alerting the project team.

Schema and configuration reference

Routing and severity constants are site-canonical and appear identically wherever impact logic runs: AUTO_COMMIT = 0.92, REVIEW_FLOOR = 0.75, a per-project ALERT_THRESHOLD_DAYS (default 5), and — where an impact carries a cost delta — the Decimal("5000.00") change-order draft threshold shared with the rest of the automation.

Verification and testing

Treat attribution, severity, and chaining as independently testable units. The chain’s tamper-evidence is part of the test surface, not an afterthought: build a short log, mutate one record, and assert verify_chain finds it.

from datetime import datetime, timezone
from uuid import uuid4


def _record(prev_hash: str, **overrides) -> ScheduleImpactRecord:
    base = dict(
        record_id=uuid4(),
        project_uuid=uuid4(),
        activity_id="A1200",
        wbs_node="PROJ-014-STR-02",
        event_time=datetime(2026, 7, 3, 12, 0, tzinfo=timezone.utc),
        logged_at=datetime(2026, 7, 4, 9, 0, tzinfo=timezone.utc),
        cause="change_order",
        delta_days=11,
        float_consumed_days=0,
        critical_path_changed=True,
        change_order_id="CO-0042",
        link_confidence=0.95,
        prev_hash=prev_hash,
    )
    base.update(overrides)
    return ScheduleImpactRecord(**base)


def test_change_order_cause_requires_id():
    import pytest
    with pytest.raises(ValueError):
        _record(GENESIS_HASH, change_order_id=None)


def test_naive_timestamp_rejected():
    import pytest
    with pytest.raises(ValueError):
        _record(GENESIS_HASH, event_time=datetime(2026, 7, 3, 12, 0))


def test_critical_move_raises_alert():
    rec = _record(GENESIS_HASH)
    assert severity_action(rec) == "raise_alert"


def test_confidence_bands():
    assert classify(0.97) == "auto_commit"
    assert classify(0.80) == "human_review"
    assert classify(0.60) == "quarantine"


def test_chain_detects_tampering():
    log = ImpactLog()
    first = _record(GENESIS_HASH)
    log.append(first)
    second = _record(first.content_digest(), delta_days=3, critical_path_changed=False)
    log.append(second)
    assert log.verify_chain() is None
    # Simulate a retroactive edit to the first record's delta.
    log._records[0] = _record(GENESIS_HASH, record_id=first.record_id, delta_days=99)
    assert log.verify_chain() == 1  # the break surfaces at the next record

Run the suite with pytest -q. For a manual smoke test, append two records and print log.verify_chain(); it must return None before any edit and an index after one.

Troubleshooting

• Naive timestamps silently shift a delay across sites. A field record logged in local time without a zone gets read as UTC, moving an impact by hours and, at a day boundary, by a whole delta_day. The event_time validator rejects naive datetimes at the boundary; fix the upstream parser to attach the site’s zone before the record is built, never after it is chained.
• Duplicate impacts on broker replay. A redelivered delta message re-runs construction and appends a second record for the same event, inflating the claim. Make the append idempotent: key the commit on project_uuid plus activity_id plus event_time plus cause, so a replay is a no-op even though construction happily rebuilds an identical record.
• A retroactive edit breaks the audit chain. Someone “corrects” a delta_days value on a committed record, and every downstream digest now fails verify_chain. That is the system working — never edit in place. Record the correction as a new record (a resequence or other cause referencing the original record_id) so the chain stays intact and the correction is itself auditable.
• Over-alerting drowns the real slip. Setting ALERT_THRESHOLD_DAYS to 1 fires on routine float consumption and trains the team to ignore alerts. Alert unconditionally on critical_path_changed, but tune the day threshold to the project’s reporting cadence and route non-critical slips to a digest through fallback alert routing rather than an immediate page.
• Every impact quarantines as unattributed. Attribution collapses below 0.75 because the delta feed and the source events use different activity identifiers, so delta_on_impacted_activity is always false. Reconcile the identifiers against the shared WBS binding from schedule baseline variance before scoring, rather than lowering REVIEW_FLOOR and letting weak attributions through.

Frequently Asked Questions

Related

• Critical Path Delta Detection
• Schedule Baseline Variance
• RFI Schema Design
• Automated Document Ingestion & Parsing
• Fallback Alert Routing
• WBS Mapping Strategies

← Back to Schedule Sync & Critical Path Integration