How to Build an AI Project Management Assistant That Tracks Tasks and Deadlines

ai-pm-assistant/
├── app/
│   ├── __init__.py
│   └── cpm_scheduler.py
├── tests/
│   └── test_pm_assistant.py
└── requirements.txt

from graphlib import TopologicalSorter, CycleError
from typing import List, Dict, Set
from pydantic import BaseModel, Field

class ProjectTask(BaseModel):
    task_id: str
    title: str
    owner: str
    duration_days: int = Field(..., ge=1)
    dependencies: List[str] = Field(default_factory=list, description="Predecessor task_ids")
    is_delayed_by_days: int = Field(0, ge=0)

class ScheduledTaskMetrics(BaseModel):
    task_id: str
    title: str
    owner: str
    effective_duration: int
    earliest_start: int
    earliest_finish: int
    latest_start: int
    latest_finish: int
    slack_days: int
    is_critical_path: bool

class ProjectScheduleReport(BaseModel):
    total_project_duration_days: int
    critical_path_task_ids: List[str]
    overallocated_owners: List[str]
    task_metrics: Dict[str, ScheduledTaskMetrics]

def compute_critical_path_schedule(tasks: List[ProjectTask]) -> ProjectScheduleReport:
    """Validates the project DAG for cycles and computes ES, EF, LS, LF, Slack, and Critical Path."""
    task_map: Dict[str, ProjectTask] = {t.task_id: t for t in tasks}
    graph: Dict[str, Set[str]] = {}

    for t in tasks:
        for dep in t.dependencies:
            if dep not in task_map:
                raise ValueError(f"Task '{t.task_id}' depends on non-existent task '{dep}'.")
        graph[t.task_id] = set(t.dependencies)

    sorter = TopologicalSorter(graph)
    try:
        topo_order = list(sorter.static_order())
    except CycleError as exc:
        raise ValueError(f"Circular task dependency detected in project schedule: {exc}") from exc

    # Forward pass: Earliest Start (ES) and Earliest Finish (EF)
    es: Dict[str, int] = {}
    ef: Dict[str, int] = {}
    eff_dur: Dict[str, int] = {}

    for tid in topo_order:
        t = task_map[tid]
        d = t.duration_days + t.is_delayed_by_days
        eff_dur[tid] = d
        earliest_s = max((ef[p] for p in t.dependencies), default=0)
        es[tid] = earliest_s
        ef[tid] = earliest_s + d

    total_duration = max(ef.values(), default=0)

    # Build successor adjacency list for backward pass
    successors: Dict[str, List[str]] = {tid: [] for tid in topo_order}
    for t in tasks:
        for dep in t.dependencies:
            successors[dep].append(t.task_id)

    # Backward pass: Latest Finish (LF) and Latest Start (LS)
    lf: Dict[str, int] = {}
    ls: Dict[str, int] = {}
    for tid in reversed(topo_order):
        latest_f = min((ls[s] for s in successors[tid]), default=total_duration)
        lf[tid] = latest_f
        ls[tid] = latest_f - eff_dur[tid]

    metrics: Dict[str, ScheduledTaskMetrics] = {}
    critical_ids: List[str] = []
    owner_concurrent_critical: Dict[str, int] = {}

    for tid in topo_order:
        t = task_map[tid]
        slack = ls[tid] - es[tid]
        is_crit = slack == 0
        if is_crit:
            critical_ids.append(tid)
            owner_concurrent_critical[t.owner] = owner_concurrent_critical.get(t.owner, 0) + 1

        metrics[tid] = ScheduledTaskMetrics(
            task_id=tid,
            title=t.title,
            owner=t.owner,
            effective_duration=eff_dur[tid],
            earliest_start=es[tid],
            earliest_finish=ef[tid],
            latest_start=ls[tid],
            latest_finish=lf[tid],
            slack_days=slack,
            is_critical_path=is_crit,
        )

    overallocated = [owner for owner, count in owner_concurrent_critical.items() if count > 2]
    return ProjectScheduleReport(
        total_project_duration_days=total_duration,
        critical_path_task_ids=critical_ids,
        overallocated_owners=overallocated,
        task_metrics=metrics,
    )

import pytest
from app.cpm_scheduler import ProjectTask, compute_critical_path_schedule

def test_critical_path_and_slack_calculation():
    tasks = [
        ProjectTask(task_id="T1", title="DB Schema Design", owner="Aniket", duration_days=3, dependencies=[]),
        ProjectTask(task_id="T2", title="Backend API", owner="Aniket", duration_days=5, dependencies=["T1"]),
        ProjectTask(task_id="T3", title="UI Wireframes", owner="Meera", duration_days=2, dependencies=["T1"]),
        ProjectTask(task_id="T4", title="Integration Tests", owner="Rohan", duration_days=2, dependencies=["T2", "T3"]),
    ]
    report = compute_critical_path_schedule(tasks)
    assert report.total_project_duration_days == 10  # T1(3) + T2(5) + T4(2) = 10
    assert report.critical_path_task_ids == ["T1", "T2", "T4"]
    assert report.task_metrics["T3"].slack_days == 3  # UI Wireframes has 3 days of float/slack

def test_circular_dependency_rejected():
    cyclic_tasks = [
        ProjectTask(task_id="A", title="Task A", owner="Dev1", duration_days=2, dependencies=["B"]),
        ProjectTask(task_id="B", title="Task B", owner="Dev2", duration_days=2, dependencies=["A"]),
    ]
    with pytest.raises(ValueError, match="Circular task dependency"):
        compute_critical_path_schedule(cyclic_tasks)