feat: add driving flag to relationship query responses

Add computed driving flag to all relationship queries (list_relationships,
get_predecessors, get_successors). A relationship is marked as driving when
the predecessor's early end date plus lag determines the successor's early
start date.

Changes:
- Add early_start_date and early_end_date columns to activities schema
- Parse early dates from TASK table in XER files
- Implement is_driving_relationship() helper with 24hr tolerance for
  calendar gaps
- Update all relationship queries to compute and return driving flag
- Add contract and unit tests for driving flag functionality
- Update spec, contracts, and documentation
This commit is contained in:
2026-01-07 07:21:58 -05:00
parent 2255b65ef6
commit af8cdc1d31
17 changed files with 654 additions and 324 deletions

View File

@@ -75,3 +75,22 @@ class TestGetPredecessorsContract:
assert "error" in result
assert result["error"]["code"] == "NO_FILE_LOADED"
async def test_get_predecessors_includes_driving_flag(
self, sample_xer_single_project: Path
) -> None:
"""get_predecessors includes driving flag for each predecessor."""
from xer_mcp.tools.get_predecessors import get_predecessors
from xer_mcp.tools.load_xer import load_xer
await load_xer(file_path=str(sample_xer_single_project))
# A1010 (2002) has one predecessor: A1000 (2001)
result = await get_predecessors(activity_id="2002")
assert "predecessors" in result
assert len(result["predecessors"]) >= 1
# All predecessors should have a driving flag
for pred in result["predecessors"]:
assert "driving" in pred
assert isinstance(pred["driving"], bool)

View File

@@ -74,3 +74,22 @@ class TestGetSuccessorsContract:
assert "error" in result
assert result["error"]["code"] == "NO_FILE_LOADED"
async def test_get_successors_includes_driving_flag(
self, sample_xer_single_project: Path
) -> None:
"""get_successors includes driving flag for each successor."""
from xer_mcp.tools.get_successors import get_successors
from xer_mcp.tools.load_xer import load_xer
await load_xer(file_path=str(sample_xer_single_project))
# A1000 (2001) has one successor: A1010 (2002)
result = await get_successors(activity_id="2001")
assert "successors" in result
assert len(result["successors"]) >= 1
# All successors should have a driving flag
for succ in result["successors"]:
assert "driving" in succ
assert isinstance(succ["driving"], bool)

View File

@@ -76,3 +76,21 @@ class TestListRelationshipsContract:
assert "error" in result
assert result["error"]["code"] == "NO_FILE_LOADED"
async def test_list_relationships_includes_driving_flag(
self, sample_xer_single_project: Path
) -> None:
"""list_relationships returns relationships with driving flag."""
from xer_mcp.tools.list_relationships import list_relationships
from xer_mcp.tools.load_xer import load_xer
await load_xer(file_path=str(sample_xer_single_project))
result = await list_relationships()
assert "relationships" in result
assert len(result["relationships"]) > 0
# All relationships should have a driving flag
for rel in result["relationships"]:
assert "driving" in rel
assert isinstance(rel["driving"], bool)

View File

@@ -109,3 +109,68 @@ class TestActivityQueries:
activity = get_activity_by_id("nonexistent")
assert activity is None
class TestDrivingRelationship:
"""Tests for driving relationship computation."""
def test_is_driving_relationship_fs_driving(self) -> None:
"""FS relationship is driving when pred_end + lag = succ_start."""
from xer_mcp.db.queries import is_driving_relationship
# Pred ends at 2026-01-08T15:00, succ starts at 2026-01-09T07:00
# With 0 lag and overnight gap, this is driving
result = is_driving_relationship(
pred_early_end="2026-01-08T15:00:00",
succ_early_start="2026-01-09T07:00:00",
lag_hours=0.0,
pred_type="FS",
)
assert result is True
def test_is_driving_relationship_fs_not_driving(self) -> None:
"""FS relationship is not driving when there's float."""
from xer_mcp.db.queries import is_driving_relationship
# Pred ends much earlier than succ starts (has float)
result = is_driving_relationship(
pred_early_end="2026-01-01T15:00:00",
succ_early_start="2026-01-10T07:00:00",
lag_hours=0.0,
pred_type="FS",
)
assert result is False
def test_is_driving_relationship_with_lag(self) -> None:
"""FS relationship with lag is driving when pred_end + lag = succ_start."""
from xer_mcp.db.queries import is_driving_relationship
# Pred ends at 2026-01-08T15:00, 16hr lag, succ starts at 2026-01-09T07:00
# 15:00 + 16hrs = next day 07:00 (exactly matches)
result = is_driving_relationship(
pred_early_end="2026-01-08T15:00:00",
succ_early_start="2026-01-09T07:00:00",
lag_hours=16.0,
pred_type="FS",
)
assert result is True
def test_is_driving_relationship_missing_dates(self) -> None:
"""Relationship is not driving when dates are missing."""
from xer_mcp.db.queries import is_driving_relationship
result = is_driving_relationship(
pred_early_end=None,
succ_early_start="2026-01-09T07:00:00",
lag_hours=0.0,
pred_type="FS",
)
assert result is False
result = is_driving_relationship(
pred_early_end="2026-01-08T15:00:00",
succ_early_start=None,
lag_hours=0.0,
pred_type="FS",
)
assert result is False

View File

@@ -87,6 +87,76 @@ class TestTaskHandler:
handler = TaskHandler()
assert handler.table_name == "TASK"
def test_parse_early_dates(self) -> None:
"""Handler should parse early_start_date and early_end_date from TASK row."""
from xer_mcp.parser.table_handlers.task import TaskHandler
handler = TaskHandler()
fields = [
"task_id",
"proj_id",
"wbs_id",
"task_code",
"task_name",
"task_type",
"status_code",
"target_start_date",
"target_end_date",
"early_start_date",
"early_end_date",
"total_float_hr_cnt",
"driving_path_flag",
]
values = [
"2001",
"1001",
"100",
"A1000",
"Site Prep",
"TT_Task",
"TK_NotStart",
"2026-01-02 07:00",
"2026-01-08 15:00",
"2026-01-02 07:00",
"2026-01-08 15:00",
"0",
"Y",
]
result = handler.parse_row(fields, values)
assert result is not None
assert result["early_start_date"] == "2026-01-02T07:00:00"
assert result["early_end_date"] == "2026-01-08T15:00:00"
def test_parse_missing_early_dates(self) -> None:
"""Handler should handle missing early dates gracefully."""
from xer_mcp.parser.table_handlers.task import TaskHandler
handler = TaskHandler()
fields = [
"task_id",
"proj_id",
"task_code",
"task_name",
"task_type",
]
values = [
"2001",
"1001",
"A1000",
"Site Prep",
"TT_Task",
]
result = handler.parse_row(fields, values)
assert result is not None
assert result["early_start_date"] is None
assert result["early_end_date"] is None
class TestTaskpredHandler:
"""Tests for TASKPRED table handler."""