Building a Construction Project Status Agent: Progress Updates and Delay Notifications

Why Construction Projects Need AI Status Agents

Construction projects are notoriously difficult to track. A typical commercial build involves dozens of subcontractors, hundreds of milestones, weather dependencies, permit approvals, and material deliveries — all interconnected. When a concrete pour slips by three days, the cascading impact on framing, electrical rough-in, and inspection schedules is hard to calculate manually. An AI agent can monitor all these dependencies, calculate schedule impact in real time, and notify the right stakeholders before small delays become major problems.

The difference between a reactive and proactive construction manager is information latency. An AI agent reduces that latency from days to minutes.

Modeling the Project Schedule

Construction schedules are dependency graphs. Each milestone depends on predecessors, and delays propagate through the critical path.

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional

class MilestoneStatus(Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    DELAYED = "delayed"
    BLOCKED = "blocked"

@dataclass
class Milestone:
    id: str
    name: str
    planned_start: datetime
    planned_end: datetime
    actual_start: Optional[datetime] = None
    actual_end: Optional[datetime] = None
    status: MilestoneStatus = MilestoneStatus.NOT_STARTED
    dependencies: list[str] = field(default_factory=list)
    assigned_contractor: str = ""
    completion_percentage: float = 0.0

class ProjectSchedule:
    def __init__(self, milestones: list[Milestone]):
        self.milestones = {m.id: m for m in milestones}

    def calculate_delay_impact(self, delayed_milestone_id: str, delay_days: int) -> list[dict]:
        affected = []
        visited = set()
        queue = [delayed_milestone_id]

        while queue:
            current_id = queue.pop(0)
            if current_id in visited:
                continue
            visited.add(current_id)

            for mid, milestone in self.milestones.items():
                if current_id in milestone.dependencies and mid not in visited:
                    new_start = milestone.planned_start + timedelta(days=delay_days)
                    new_end = milestone.planned_end + timedelta(days=delay_days)
                    affected.append({
                        "milestone_id": mid,
                        "milestone_name": milestone.name,
                        "original_start": milestone.planned_start.isoformat(),
                        "new_start": new_start.isoformat(),
                        "delay_days": delay_days,
                        "contractor": milestone.assigned_contractor,
                    })
                    queue.append(mid)

        return affected

Photo Documentation Processing

Field crews submit daily photos. The agent logs them against milestones and extracts metadata for progress tracking.

from datetime import datetime

class PhotoDocumentation:
    def __init__(self, storage_client, db):
        self.storage = storage_client
        self.db = db

    async def process_site_photo(
        self, image_data: bytes, milestone_id: str,
        uploaded_by: str, notes: str = "",
    ) -> dict:
        timestamp = datetime.now()
        filename = f"{milestone_id}/{timestamp.strftime('%Y%m%d_%H%M%S')}.jpg"
        url = await self.storage.upload(filename, image_data)

        record = {
            "milestone_id": milestone_id,
            "photo_url": url,
            "uploaded_by": uploaded_by,
            "timestamp": timestamp.isoformat(),
            "notes": notes,
        }
        await self.db.execute(
            """INSERT INTO site_photos
               (milestone_id, photo_url, uploaded_by, captured_at, notes)
               VALUES ($1, $2, $3, $4, $5)""",
            milestone_id, url, uploaded_by, timestamp, notes,
        )
        return record

    async def get_milestone_photos(self, milestone_id: str) -> list[dict]:
        rows = await self.db.fetch(
            """SELECT photo_url, uploaded_by, captured_at, notes
               FROM site_photos
               WHERE milestone_id = $1
               ORDER BY captured_at DESC""",
            milestone_id,
        )
        return [dict(r) for r in rows]

Delay Alert System

The agent monitors schedule variances and sends targeted notifications to affected stakeholders.

from dataclasses import dataclass

@dataclass
class StakeholderAlert:
    recipient: str
    role: str
    milestone_name: str
    delay_days: int
    impact_summary: str
    action_required: str

class DelayAlertEngine:
    def __init__(self, notification_service):
        self.notifier = notification_service

    async def evaluate_and_alert(
        self, schedule: "ProjectSchedule", milestone_id: str, delay_days: int,
    ) -> list[StakeholderAlert]:
        affected = schedule.calculate_delay_impact(milestone_id, delay_days)
        source = schedule.milestones[milestone_id]
        alerts = []

        # Always alert the project manager
        alerts.append(StakeholderAlert(
            recipient="project_manager",
            role="Project Manager",
            milestone_name=source.name,
            delay_days=delay_days,
            impact_summary=f"{len(affected)} downstream milestones affected",
            action_required="Review updated schedule and approve revised timeline",
        ))

        # Alert affected contractors
        contractors_notified = set()
        for item in affected:
            contractor = item["contractor"]
            if contractor and contractor not in contractors_notified:
                alerts.append(StakeholderAlert(
                    recipient=contractor,
                    role="Subcontractor",
                    milestone_name=item["milestone_name"],
                    delay_days=delay_days,
                    impact_summary=f"Your start date shifts to {item['new_start']}",
                    action_required="Confirm availability for revised schedule",
                ))
                contractors_notified.add(contractor)

        # Alert owner/client for delays over 5 days
        if delay_days > 5:
            alerts.append(StakeholderAlert(
                recipient="client",
                role="Property Owner",
                milestone_name=source.name,
                delay_days=delay_days,
                impact_summary=f"Project completion may shift by {delay_days} days",
                action_required="No action needed — team is developing mitigation plan",
            ))

        for alert in alerts:
            await self.notifier.send(
                to=alert.recipient,
                subject=f"Schedule Update: {alert.milestone_name}",
                body=f"{alert.impact_summary}. {alert.action_required}",
            )
        return alerts

Progress Report Generation

The agent compiles daily and weekly progress reports from milestone data, photos, and schedule variances.

class ProgressReportGenerator:
    def __init__(self, schedule: "ProjectSchedule", photo_docs: PhotoDocumentation):
        self.schedule = schedule
        self.photos = photo_docs

    async def generate_weekly_report(self, project_name: str) -> dict:
        completed = []
        in_progress = []
        delayed = []

        for mid, ms in self.schedule.milestones.items():
            if ms.status == MilestoneStatus.COMPLETED:
                completed.append(ms.name)
            elif ms.status == MilestoneStatus.DELAYED:
                delayed.append({"name": ms.name, "contractor": ms.assigned_contractor})
            elif ms.status == MilestoneStatus.IN_PROGRESS:
                photos = await self.photos.get_milestone_photos(mid)
                in_progress.append({
                    "name": ms.name,
                    "completion": ms.completion_percentage,
                    "photo_count": len(photos),
                })

        total = len(self.schedule.milestones)
        done = len(completed)
        return {
            "project": project_name,
            "overall_progress": f"{(done / total * 100):.1f}%",
            "completed_this_week": completed,
            "in_progress": in_progress,
            "delayed": delayed,
            "schedule_health": "on_track" if not delayed else "at_risk",
        }

FAQ

How does the agent handle weather-related delays?

The agent integrates with weather APIs to monitor forecasts at the job site location. When conditions will prevent work (heavy rain for concrete pours, high winds for crane operations), it proactively flags the risk before the delay occurs. This gives the project manager time to reschedule or adjust the sequence of work.

Can the agent work with existing project management tools like Procore?

Yes. The agent is designed with an integration layer that connects to Procore, PlanGrid, or Buildertrend via their APIs. It pulls schedule data, pushes status updates, and syncs photo documentation — acting as an intelligent layer on top of whatever tools the team already uses.

How do you calculate the critical path automatically?

The agent uses topological sorting on the milestone dependency graph to identify the longest path through the project. Any milestone on this path with zero float is critical — a one-day delay there means a one-day delay for the entire project. The calculate_delay_impact method performs a breadth-first traversal of downstream dependencies to quantify the ripple effect.

---

#Construction #ProjectManagement #MilestoneTracking #DelayNotifications #StakeholderCommunication #AgenticAI #LearnAI #AIEngineering