feat!: add task pinning system and refactor existing systems

README.md

@@ -82,8 +82,10 @@ Yuga Planner follows a **service-oriented architecture** with clear separation o
 | **Live Log Streaming** | Real-time solver progress and status updates in UI | ✅ |
 | **Configurable Parameters** | Adjustable employee count and schedule duration | ✅ |
 | **Mock Project Loading** | Pre-configured sample projects for quick testing | ✅ |
-| **Calendar Parsing** | Extracts tasks from uploaded calendar files (.ics) | ✅ |
-| **MCP Endpoint** | API endpoint for MCP tool integration | ✅ |
+| **Calendar Parsing & Pinning** | Extracts and preserves calendar events from .ics files at original times | ✅ |
+| **Business Hours Enforcement** | Respects 9:00-18:00 working hours with lunch break exclusion | ✅ |
+| **Weekend Scheduling Prevention** | Hard constraint preventing weekend task assignments | ✅ |
+| **MCP Endpoint** | API endpoint for MCP tool integration with calendar support | ✅ |
 
 ## 🎯 Two Usage Modes
 
@@ -123,9 +125,8 @@ available time slots around your existing meetings
 - Designed for seamless chatbot and agent workflow integration
 
 **Current Limitations:**
-- **Weekend constraints:** Tasks can be scheduled on weekends (should respect work-week boundaries)
-- **Working hours:** No enforcement of standard business hours (8 AM - 6 PM)
-- **Calendar pinning:** Tasks from uploaded calendars are solved alongside other tasks but should remain pinned to their original time slots
+- **Cross-system integration:** Gradio web demo and MCP personal tool operate as separate systems
+- **Multi-timezone support:** Currently operates in a single timezone context with UTC conversion for consistency. Calendar events from different timezones are normalized to the same scheduling context.
 
 See the [CHANGELOG.md](CHANGELOG.md) for details on recent MCP-related changes.
 
@@ -138,14 +139,10 @@ See the [CHANGELOG.md](CHANGELOG.md) for details on recent MCP-related changes.
 
 ### Work in Progress
 
-- **Constraint Enhancements:**
-  - Weekend respect (prevent scheduling on weekends)
-  - Working hours enforcement (8 AM - 6 PM business hours)
-  - Calendar task pinning (preserve original time slots for imported calendar events)
-- **Gradio UI overhaul:** Enhanced user experience and visual improvements
-- **Migration to Pydantic models:** Type-safe data validation and serialization
-- **Migrate from violation_analyzer to Timefold dedicated libraries**
-- **Include tests for all constraints using ConstraintVerifier**
+- **🔧 Gradio UI overhaul:** Enhanced user experience and visual improvements
+- **🔍 Migration to Pydantic models:** Type-safe data validation and serialization
+- **📚 Migrate from violation_analyzer to Timefold dedicated libraries**
+- **⚡ Enhanced timezone support:** Multi-timezone calendar integration for international scheduling
 
 ### Future Work
 

src/constraint_solvers/timetable/constraints.py

@@ -6,6 +6,8 @@ from .working_hours import (
     get_working_day_from_slot,
     get_slot_within_day,
     task_spans_lunch_break,
+    is_weekend_slot,
+    get_slot_date,
 )
 
 from timefold.solver.score import HardSoftDecimalScore
@@ -32,37 +34,7 @@ def get_slot_overlap(task1: Task, task2: Task) -> int:
     return max(0, overlap_end - overlap_start)
 
 
-def get_slot_date(slot: int) -> date:
-    """Convert a slot index to a date.
-
-    For compatibility with tests, slot 0 = today, slot 16 = tomorrow, etc.
-    In production, weekends would be filtered out, but for tests we keep simple mapping.
-
-    Args:
-        slot (int): The slot index.
-
-    Returns:
-        date: The date corresponding to the slot.
-    """
-    working_day = get_working_day_from_slot(slot)
-    today = date.today()
-    return today + timedelta(days=working_day)
-
-
-def is_weekend_slot(slot: int) -> bool:
-    """Check if a slot falls on a weekend.
-
-    Since our slot system only includes working days, this should always return False
-    for valid slots, but we keep it for validation purposes.
-
-    Args:
-        slot (int): The slot index.
-
-    Returns:
-        bool: True if the slot would fall on a weekend.
-    """
-    slot_date = get_slot_date(slot)
-    return slot_date.weekday() >= 5  # Saturday=5, Sunday=6
+# Note: get_slot_date and is_weekend_slot are now imported from working_hours
 
 
 def tasks_violate_sequence_order(task1: Task, task2: Task) -> bool:
@@ -183,9 +155,11 @@ def task_fits_in_schedule(constraint_factory: ConstraintFactory):
 def unavailable_employee(constraint_factory: ConstraintFactory):
     return (
         constraint_factory.for_each(Task)
+        .join(ScheduleInfo)
         .filter(
-            lambda task: task.employee is not None
-            and get_slot_date(task.start_slot) in task.employee.unavailable_dates
+            lambda task, schedule_info: task.employee is not None
+            and get_slot_date(task.start_slot, schedule_info.base_date)
+            in task.employee.unavailable_dates
         )
         .penalize(HardSoftDecimalScore.ONE_HARD)
         .as_constraint("Unavailable employee")
@@ -215,9 +189,11 @@ def no_weekend_scheduling(constraint_factory: ConstraintFactory):
 def undesired_day_for_employee(constraint_factory: ConstraintFactory):
     return (
         constraint_factory.for_each(Task)
+        .join(ScheduleInfo)
         .filter(
-            lambda task: task.employee is not None
-            and get_slot_date(task.start_slot) in task.employee.undesired_dates
+            lambda task, schedule_info: task.employee is not None
+            and get_slot_date(task.start_slot, schedule_info.base_date)
+            in task.employee.undesired_dates
         )
         .penalize(HardSoftDecimalScore.ONE_SOFT)
         .as_constraint("Undesired day for employee")
@@ -227,9 +203,11 @@ def undesired_day_for_employee(constraint_factory: ConstraintFactory):
 def desired_day_for_employee(constraint_factory: ConstraintFactory):
     return (
         constraint_factory.for_each(Task)
+        .join(ScheduleInfo)
         .filter(
-            lambda task: task.employee is not None
-            and get_slot_date(task.start_slot) in task.employee.desired_dates
+            lambda task, schedule_info: task.employee is not None
+            and get_slot_date(task.start_slot, schedule_info.base_date)
+            in task.employee.desired_dates
         )
         .reward(HardSoftDecimalScore.ONE_SOFT)
         .as_constraint("Desired day for employee")

src/constraint_solvers/timetable/domain.py

@@ -2,7 +2,7 @@ from timefold.solver import SolverStatus
 from timefold.solver.domain import *
 from timefold.solver.score import HardSoftDecimalScore
 
-from datetime import date
+from datetime import date, timezone
 from typing import Annotated
 from dataclasses import dataclass, field
 
@@ -99,13 +99,35 @@ class Task:
 @dataclass
 class ScheduleInfo:
     total_slots: int  # Total number of 30-minute slots in the schedule
+    base_date: date = None  # Base date for slot 0 (optional, defaults to today)
+    base_timezone: timezone = None  # Timezone for datetime conversions (optional)
 
     def to_dict(self):
-        return {"total_slots": self.total_slots}
+        return {
+            "total_slots": self.total_slots,
+            "base_date": self.base_date.isoformat() if self.base_date else None,
+            "base_timezone": str(self.base_timezone) if self.base_timezone else None,
+        }
 
     @staticmethod
     def from_dict(d):
-        return ScheduleInfo(total_slots=d["total_slots"])
+        base_date = None
+        if d.get("base_date"):
+            base_date = date.fromisoformat(d["base_date"])
+
+        base_timezone = None
+        if d.get("base_timezone"):
+            # Simple timezone parsing - extend as needed
+            tz_str = d["base_timezone"]
+            if tz_str == "UTC" or "+00:00" in tz_str:
+                base_timezone = timezone.utc
+            # Add more timezone parsing as needed
+
+        return ScheduleInfo(
+            total_slots=d["total_slots"],
+            base_date=base_date,
+            base_timezone=base_timezone,
+        )
 
 
 @planning_solution
@@ -124,8 +146,17 @@ class EmployeeSchedule:
     def get_start_slot_range(
         self,
     ) -> Annotated[list[int], ValueRangeProvider(id="startSlotRange")]:
-        """Returns all possible start slots."""
-        return list(range(self.schedule_info.total_slots))
+        """Returns all possible start slots, including slots used by pinned tasks."""
+        max_slot = self.schedule_info.total_slots
+
+        # Ensure all pinned task slots are included in the range
+        for task in self.tasks:
+            if getattr(task, "pinned", False):
+                task_end_slot = task.start_slot + task.duration_slots
+                if task_end_slot > max_slot:
+                    max_slot = task_end_slot
+
+        return list(range(max_slot))
 
     def to_dict(self):
         return {

src/constraint_solvers/timetable/working_hours.py

@@ -2,10 +2,46 @@
 #     WORKING HOURS CONFIG
 # =========================
 
-# Working hours: 9:00-13:00 (8 slots) + 14:00-18:00 (8 slots) = 16 slots per working day
-SLOTS_PER_WORKING_DAY = 16
+# Working hours: 9:00-18:00 (20 slots) = 20 slots per working day
+# Each slot is 30 minutes, starting at 9:00 AM
+SLOTS_PER_WORKING_DAY = 20  # 9:00-18:00 (9 hours * 2 slots/hour)
 MORNING_SLOTS = 8  # 9:00-13:00 (4 hours * 2 slots/hour)
-AFTERNOON_SLOTS = 8  # 14:00-18:00 (4 hours * 2 slots/hour)
+AFTERNOON_SLOTS = 10  # 14:00-18:00 (4 hours * 2 slots/hour)
+LUNCH_BREAK_START_SLOT = 8  # 13:00-14:00
+LUNCH_BREAK_END_SLOT = 10  # 14:00
+
+from datetime import datetime, date, time, timezone, timedelta
+
+
+def slot_to_datetime(slot: int, base_date: date = None, base_timezone=None) -> datetime:
+    """
+    Convert a slot index to a naive datetime in local time, accounting for working days.
+
+    Args:
+        slot: The slot index (each slot = 30 minutes within working hours)
+        base_date: Base date for slot 0 (defaults to today)
+        base_timezone: Ignored (kept for API compatibility)
+
+    Returns:
+        datetime: The corresponding naive datetime in local time
+    """
+    if base_date is None:
+        base_date = date.today()
+
+    # Calculate which working day and slot within that day
+    working_day = get_working_day_from_slot(slot)
+    slot_within_day = get_slot_within_day(slot)
+
+    # Get the actual calendar date for this working day
+    target_date = base_date + timedelta(days=working_day)
+
+    # Calculate time within the working day (9:00 AM + slot_within_day * 30 minutes)
+    minutes_from_9am = slot_within_day * 30
+    target_time = datetime.combine(
+        target_date, datetime.min.time().replace(hour=9)
+    ) + timedelta(minutes=minutes_from_9am)
+
+    return target_time
 
 
 def get_working_day_from_slot(slot: int) -> int:
@@ -21,19 +57,19 @@ def get_working_day_from_slot(slot: int) -> int:
 
 
 def get_slot_within_day(slot: int) -> int:
-    """Get the slot position within a working day (0-15).
+    """Get the slot position within a working day (0-19).
 
     Args:
         slot (int): The slot index.
 
     Returns:
-        int: The slot position within the day (0-15).
+        int: The slot position within the day (0-19).
     """
     return slot % SLOTS_PER_WORKING_DAY
 
 
 def task_spans_lunch_break(task) -> bool:
-    """Check if a task spans across the lunch break period.
+    """Check if a task spans across the lunch break period (13:00-14:00).
 
     Args:
         task: The task to check.
@@ -44,5 +80,41 @@ def task_spans_lunch_break(task) -> bool:
     start_slot_in_day = get_slot_within_day(task.start_slot)
     end_slot_in_day = start_slot_in_day + task.duration_slots - 1
 
-    # If task starts in morning (0-7) and ends in afternoon (8-15), it spans lunch
-    return start_slot_in_day < MORNING_SLOTS and end_slot_in_day >= MORNING_SLOTS
+    # Check if task overlaps with lunch break slots (8-9, which is 13:00-14:00)
+    return (
+        start_slot_in_day <= LUNCH_BREAK_END_SLOT - 1
+        and end_slot_in_day >= LUNCH_BREAK_START_SLOT
+    )
+
+
+def is_weekend_slot(slot: int) -> bool:
+    """Check if a slot falls on a weekend.
+
+    Args:
+        slot: The slot index
+
+    Returns:
+        bool: True if the slot is on a weekend
+    """
+    working_day = get_working_day_from_slot(slot)
+    # For simplicity, assume every 7th day starting from day 5 and 6 are weekends
+    # This is a simplification - in practice you'd want to use actual calendar logic
+    day_of_week = working_day % 7
+    return day_of_week >= 5  # Saturday (5) and Sunday (6)
+
+
+def get_slot_date(slot: int, base_date: date = None) -> date:
+    """Get the date for a given slot.
+
+    Args:
+        slot: The slot index
+        base_date: Base date for slot 0 (defaults to today)
+
+    Returns:
+        date: The date for this slot
+    """
+    if base_date is None:
+        base_date = date.today()
+
+    working_days = get_working_day_from_slot(slot)
+    return base_date + timedelta(days=working_days)

src/factory/data/formatters.py

@@ -5,45 +5,10 @@ from factory.data.generators import earliest_monday_on_or_after
 from constraint_solvers.timetable.working_hours import (
     SLOTS_PER_WORKING_DAY,
     MORNING_SLOTS,
+    slot_to_datetime,
 )
 
 
-def slot_to_datetime(slot: int, base_date: date = None) -> datetime:
-    """Convert a slot index to actual datetime, respecting working hours.
-
-    Args:
-        slot (int): The slot index (0-based).
-        base_date (date, optional): Base date to start from. Defaults to today.
-
-    Returns:
-        datetime: The actual datetime for this slot.
-    """
-    if base_date is None:
-        base_date = date.today()
-
-    # Calculate which working day this slot falls on
-    working_day = slot // SLOTS_PER_WORKING_DAY
-    slot_within_day = slot % SLOTS_PER_WORKING_DAY
-
-    # Calculate the actual calendar date
-    actual_date = base_date + timedelta(days=working_day)
-
-    # Convert slot within day to actual time
-    if slot_within_day < MORNING_SLOTS:
-        # Morning session: 9:00-13:00 (slots 0-7)
-        hour = 9 + (slot_within_day // 2)
-        minute = (slot_within_day % 2) * 30
-    else:
-        # Afternoon session: 14:00-18:00 (slots 8-15)
-        afternoon_slot = slot_within_day - MORNING_SLOTS
-        hour = 14 + (afternoon_slot // 2)
-        minute = (afternoon_slot % 2) * 30
-
-    return datetime.combine(
-        actual_date, datetime.min.time().replace(hour=hour, minute=minute)
-    )
-
-
 def schedule_to_dataframe(schedule) -> pd.DataFrame:
     """
     Convert an EmployeeSchedule to a pandas DataFrame.
@@ -56,14 +21,22 @@ def schedule_to_dataframe(schedule) -> pd.DataFrame:
     """
     data: list[dict[str, str]] = []
 
+    # Get base date from schedule info if available
+    base_date = None
+    if hasattr(schedule, "schedule_info"):
+        if hasattr(schedule.schedule_info, "base_date"):
+            base_date = schedule.schedule_info.base_date
+
     # Process each task in the schedule
     for task in schedule.tasks:
         # Get employee name or "Unassigned" if no employee assigned
         employee: str = task.employee.name if task.employee else "Unassigned"
 
-        # Calculate start and end times using working hours
-        start_time: datetime = slot_to_datetime(task.start_slot)
-        end_time: datetime = slot_to_datetime(task.start_slot + task.duration_slots)
+        # Calculate start and end times (naive local time)
+        start_time: datetime = slot_to_datetime(task.start_slot, base_date)
+        end_time: datetime = slot_to_datetime(
+            task.start_slot + task.duration_slots, base_date
+        )
 
         # Add task data to list with availability flags
         data.append(
@@ -76,6 +49,7 @@ def schedule_to_dataframe(schedule) -> pd.DataFrame:
                 "End": end_time,
                 "Duration (hours)": task.duration_slots / 2,  # Convert slots to hours
                 "Required Skill": task.required_skill,
+                "Pinned": getattr(task, "pinned", False),  # Include pinned status
                 # Check if task falls on employee's unavailable date
                 "Unavailable": employee != "Unassigned"
                 and hasattr(task.employee, "unavailable_dates")

src/factory/data/generators.py

@@ -5,6 +5,7 @@ from itertools import product
 
 from factory.data.models import *
 from constraint_solvers.timetable.domain import *
+from utils.extract_calendar import datetime_to_slot, calculate_duration_slots
 
 
 ### EMPLOYEES ###
@@ -209,9 +210,11 @@ def generate_tasks_from_calendar(
     parameters: TimeTableDataParameters,
     random: Random,
     calendar_entries: list[dict],
+    base_date: date = None,
 ) -> list[Task]:
     """
     Generate Task objects from calendar entries with Skills.
+    Calendar tasks are pinned to their original datetime slots.
     """
     tasks: list[Task] = []
     ids = generate_task_ids()
@@ -225,13 +228,27 @@ def generate_tasks_from_calendar(
             else:
                 required_skill = random.choice(parameters.skill_set.optional_skills)
 
+        # Calculate start_slot and duration_slots from calendar datetime info
+        start_datetime = entry.get("start_datetime")
+        end_datetime = entry.get("end_datetime")
+
+        if start_datetime and end_datetime and base_date:
+            # Calculate actual slot and duration from calendar times
+            start_slot = datetime_to_slot(start_datetime, base_date)
+            duration_slots = calculate_duration_slots(start_datetime, end_datetime)
+        else:
+            # Fallback to default values if datetime info is missing
+            start_slot = entry.get("start_slot", 0)
+            duration_slots = entry.get("duration_slots", 2)  # Default 1 hour
+
         tasks.append(
             Task(
                 id=next(ids),
                 description=entry["summary"],
-                duration_slots=entry.get("duration_slots", 2),  # Default 1 hour
-                start_slot=entry.get("start_slot", 0),
+                duration_slots=duration_slots,
+                start_slot=start_slot,
                 required_skill=required_skill,
+                pinned=True,  # Pin calendar tasks to their original times
             )
         )
 

src/factory/data/provider.py

@@ -15,6 +15,11 @@ from factory.agents.task_composer_agent import TaskComposerAgent
 from constraint_solvers.timetable.domain import *
 
 from utils.logging_config import setup_logging, get_logger
+from utils.extract_calendar import (
+    get_earliest_calendar_date,
+    datetime_to_slot,
+    validate_calendar_working_hours,
+)
 
 # Initialize logging
 setup_logging()
@@ -141,12 +146,65 @@ async def generate_mcp_data(
     days_in_schedule: int = None,
 ):
     parameters = MCP_PARAMS
-    if employee_count is not None or days_in_schedule is not None:
+
+    # --- DETERMINE START DATE AND REQUIRED SCHEDULE LENGTH FROM CALENDAR ---
+
+    # Validate calendar entries are within working hours first
+    if calendar_entries:
+        is_valid, error_msg = validate_calendar_working_hours(calendar_entries)
+        if not is_valid:
+            logger.error(f"❌ Calendar validation failed: {error_msg}")
+            raise ValueError(
+                f"Calendar entries violate working hours constraints:\n{error_msg}"
+            )
+        else:
+            logger.info(
+                f"✅ All {len(calendar_entries)} calendar entries are within working hours (8:00-18:00)"
+            )
+
+    # Use earliest calendar date as the base, or fall back to next Monday if no calendar
+    earliest_calendar_date = (
+        get_earliest_calendar_date(calendar_entries) if calendar_entries else None
+    )
+
+    if earliest_calendar_date:
+        start_date: date = earliest_calendar_date
+
+        # Calculate required schedule length to accommodate all calendar entries
+        if calendar_entries and days_in_schedule is None:
+            # Find the latest calendar date to determine required schedule length
+            latest_date = earliest_calendar_date
+            for entry in calendar_entries:
+                end_dt = entry.get("end_datetime")
+                if end_dt and end_dt.date() > latest_date:
+                    latest_date = end_dt.date()
+
+            # Calculate days needed plus buffer for LLM tasks
+            calendar_days_span = (latest_date - earliest_calendar_date).days + 1
+            min_required_days = (
+                calendar_days_span + 30
+            )  # Add 30 days buffer for LLM tasks
+
+            # Use the larger of user-specified or calculated requirement
+            calculated_days = max(min_required_days, parameters.days_in_schedule)
+            logger.info(
+                f"📊 Calendar span: {calendar_days_span} days, using {calculated_days} total schedule days"
+            )
+        else:
+            calculated_days = (
+                days_in_schedule if days_in_schedule else parameters.days_in_schedule
+            )
+    else:
+        start_date: date = earliest_monday_on_or_after(date.today())
+        calculated_days = (
+            days_in_schedule if days_in_schedule else parameters.days_in_schedule
+        )
+
+    # Update parameters with calculated values
+    if employee_count is not None or calculated_days != parameters.days_in_schedule:
         parameters = TimeTableDataParameters(
             skill_set=parameters.skill_set,
-            days_in_schedule=days_in_schedule
-            if days_in_schedule is not None
-            else parameters.days_in_schedule,
+            days_in_schedule=calculated_days,
             employee_count=employee_count
             if employee_count is not None
             else parameters.employee_count,
@@ -155,14 +213,25 @@ async def generate_mcp_data(
             random_seed=parameters.random_seed,
         )
 
-    start_date: date = earliest_monday_on_or_after(date.today())
     randomizer: Random = Random(parameters.random_seed)
     total_slots: int = parameters.days_in_schedule * SLOTS_PER_WORKING_DAY
 
     # --- CALENDAR TASKS ---
     calendar_tasks = generate_tasks_from_calendar(
-        parameters, randomizer, calendar_entries
+        parameters, randomizer, calendar_entries, base_date=start_date
     )
+
+    # Validate that all calendar tasks have valid slot assignments
+    for task in calendar_tasks:
+        if task.start_slot >= total_slots:
+            logger.error(
+                f"Calendar task '{task.description}' has slot {task.start_slot} >= {total_slots}"
+            )
+            raise ValueError(
+                f"Calendar task slot {task.start_slot} exceeds schedule length {total_slots}. "
+                f"Increase days_in_schedule or check calendar dates."
+            )
+
     # Assign project_id 'EXISTING' to all calendar tasks
     for t in calendar_tasks:
         t.sequence_number = 0  # will be overwritten later
@@ -257,7 +326,9 @@ async def generate_mcp_data(
     schedule = EmployeeSchedule(
         employees=employees,
         tasks=all_tasks,
-        schedule_info=ScheduleInfo(total_slots=total_slots),
+        schedule_info=ScheduleInfo(
+            total_slots=total_slots, base_date=start_date, base_timezone=None
+        ),
     )
 
     final_df = schedule_to_dataframe(schedule)

src/services/data.py

@@ -2,6 +2,7 @@ import os
 import uuid
 from io import StringIO
 from typing import Dict, List, Tuple, Union, Optional, Any
+from datetime import datetime, date, timezone
 
 import pandas as pd
 
@@ -22,6 +23,7 @@ from constraint_solvers.timetable.domain import (
 from factory.data.formatters import schedule_to_dataframe, employees_to_dataframe
 from .mock_projects import MockProjectService
 from utils.logging_config import setup_logging, get_logger
+from utils.extract_calendar import datetime_to_slot, get_earliest_calendar_date
 
 # Initialize logging
 setup_logging()
@@ -214,7 +216,8 @@ class DataService:
             employees=list(combined_employees.values()),
             tasks=combined_tasks,
             schedule_info=ScheduleInfo(
-                total_slots=parameters.days_in_schedule * SLOTS_PER_WORKING_DAY
+                total_slots=parameters.days_in_schedule * SLOTS_PER_WORKING_DAY,
+                base_date=None,  # Use default base_date for regular data loading
             ),
         )
 
@@ -238,6 +241,7 @@ class DataService:
                 "End",
                 "Duration (hours)",
                 "Required Skill",
+                "Pinned",
             ]
         ].sort_values(["Project", "Sequence"])
 
@@ -289,12 +293,15 @@ class DataService:
             raise ValueError(f"Error parsing task data: {str(e)}")
 
     @staticmethod
-    def convert_dataframe_to_tasks(task_df: pd.DataFrame) -> List[Task]:
+    def convert_dataframe_to_tasks(
+        task_df: pd.DataFrame, base_date: date = None
+    ) -> List[Task]:
         """
         Convert a DataFrame to a list of Task objects.
 
         Args:
             task_df: DataFrame containing task data
+            base_date: Base date for slot calculations (for pinned tasks)
 
         Returns:
             List of Task objects
@@ -302,19 +309,131 @@ class DataService:
         logger.info("🆔 Generating task IDs and converting to solver format...")
         ids = (str(i) for i in range(len(task_df)))
 
+        # Determine base_date if not provided
+        if base_date is None:
+            # Try to get from pinned tasks' dates
+            pinned_tasks = task_df[task_df.get("Pinned", False) == True]
+            if not pinned_tasks.empty:
+                earliest_date = None
+                for _, row in pinned_tasks.iterrows():
+                    start_time = row.get("Start")
+                    if start_time is not None:
+                        try:
+                            if isinstance(start_time, str):
+                                dt = datetime.fromisoformat(
+                                    start_time.replace("Z", "+00:00")
+                                )
+                            elif isinstance(start_time, pd.Timestamp):
+                                dt = start_time.to_pydatetime()
+                            elif isinstance(start_time, datetime):
+                                dt = start_time
+                            elif isinstance(start_time, (int, float)):
+                                # Handle Unix timestamp (milliseconds or seconds)
+                                if start_time > 1e10:
+                                    dt = datetime.fromtimestamp(
+                                        start_time / 1000, tz=timezone.utc
+                                    ).replace(tzinfo=None)
+                                else:
+                                    dt = datetime.fromtimestamp(
+                                        start_time, tz=timezone.utc
+                                    ).replace(tzinfo=None)
+                            else:
+                                logger.debug(
+                                    f"Unhandled start_time type for base_date: {type(start_time)} = {start_time}"
+                                )
+                                continue
+
+                            if earliest_date is None or dt.date() < earliest_date:
+                                earliest_date = dt.date()
+                        except Exception as e:
+                            logger.debug(f"Error parsing start_time for base_date: {e}")
+                            continue
+
+                if earliest_date:
+                    base_date = earliest_date
+                    logger.info(f"Determined base_date from pinned tasks: {base_date}")
+                else:
+                    base_date = date.today()
+                    logger.warning(
+                        "Could not determine base_date from pinned tasks, using today"
+                    )
+            else:
+                base_date = date.today()
+
         tasks = []
         for _, row in task_df.iterrows():
+            # Check if task is pinned and should preserve its start_slot
+            is_pinned = row.get("Pinned", False)
+
+            # For pinned tasks, calculate start_slot from the Start datetime
+            if is_pinned and "Start" in row and row["Start"] is not None:
+                try:
+                    start_time = row["Start"]
+
+                    # Handle different datetime formats
+                    if isinstance(start_time, str):
+                        # Parse ISO string
+                        start_time = datetime.fromisoformat(
+                            start_time.replace("Z", "+00:00")
+                        )
+                    elif isinstance(start_time, pd.Timestamp):
+                        # Convert pandas Timestamp to datetime
+                        start_time = start_time.to_pydatetime()
+                    elif isinstance(start_time, (int, float)):
+                        # Handle Unix timestamp (milliseconds or seconds)
+                        try:
+                            # If it's a large number, assume milliseconds
+                            if start_time > 1e10:
+                                start_time = datetime.fromtimestamp(
+                                    start_time / 1000, tz=timezone.utc
+                                ).replace(tzinfo=None)
+                            else:
+                                start_time = datetime.fromtimestamp(
+                                    start_time, tz=timezone.utc
+                                ).replace(tzinfo=None)
+                        except (ValueError, OSError) as e:
+                            logger.warning(
+                                f"Cannot convert timestamp {start_time} to datetime: {e}"
+                            )
+                            start_slot = 0
+                    elif not isinstance(start_time, datetime):
+                        # Skip conversion if we can't parse the datetime
+                        logger.warning(
+                            f"Cannot parse start time for pinned task: {start_time} (type: {type(start_time)})"
+                        )
+                        start_slot = 0
+
+                    if isinstance(start_time, datetime):
+                        start_slot = datetime_to_slot(start_time, base_date)
+                        logger.info(
+                            f"Converted datetime {start_time} to slot {start_slot} for pinned task (base: {base_date})"
+                        )
+                    else:
+                        start_slot = 0
+
+                except Exception as e:
+                    logger.warning(
+                        f"Error converting datetime to slot for pinned task: {e}"
+                    )
+                    start_slot = 0
+            else:
+                start_slot = 0  # Will be assigned by solver for non-pinned tasks
+
             tasks.append(
                 Task(
                     id=next(ids),
                     description=row["Task"],
                     duration_slots=int(float(row["Duration (hours)"]) * 2),
-                    start_slot=0,
+                    start_slot=start_slot,
                     required_skill=row["Required Skill"],
                     project_id=row.get("Project", ""),
                     sequence_number=int(row.get("Sequence", 0)),
+                    pinned=is_pinned,
+                    employee=None,  # Will be assigned in generate_schedule_for_solving
                 )
             )
 
-        logger.info(f"✅ Converted {len(tasks)} tasks for solver")
+        logger.info(
+            f"✅ Converted {len(tasks)} tasks for solver (base_date: {base_date})"
+        )
         return tasks

src/services/schedule.py

@@ -1,5 +1,5 @@
 import os, uuid, random
-from datetime import datetime
+from datetime import datetime, date, timezone
 from typing import Tuple, Dict, Any, Optional
 
 import pandas as pd
@@ -82,8 +82,52 @@ class ScheduleService:
             # Parse task data
             task_df = DataService.parse_task_data_from_json(task_df_json, debug)
 
+            # Extract base_date from pinned tasks for consistent slot calculations
+            base_date = None
+            pinned_tasks = task_df[task_df.get("Pinned", False) == True]
+            if not pinned_tasks.empty:
+                # Try to determine base_date from earliest pinned task
+                earliest_date = None
+                for _, row in pinned_tasks.iterrows():
+                    start_time = row.get("Start")
+                    if start_time is not None:
+                        try:
+                            if isinstance(start_time, str):
+                                dt = datetime.fromisoformat(
+                                    start_time.replace("Z", "+00:00")
+                                )
+                            elif isinstance(start_time, pd.Timestamp):
+                                dt = start_time.to_pydatetime()
+                            elif isinstance(start_time, datetime):
+                                dt = start_time
+                            elif isinstance(start_time, (int, float)):
+                                # Handle Unix timestamp (milliseconds or seconds)
+                                if start_time > 1e10:
+                                    dt = datetime.fromtimestamp(
+                                        start_time / 1000, tz=timezone.utc
+                                    ).replace(tzinfo=None)
+                                else:
+                                    dt = datetime.fromtimestamp(
+                                        start_time, tz=timezone.utc
+                                    ).replace(tzinfo=None)
+                            else:
+                                logger.debug(
+                                    f"Unhandled start_time type for base_date: {type(start_time)} = {start_time}"
+                                )
+                                continue
+
+                            if earliest_date is None or dt.date() < earliest_date:
+                                earliest_date = dt.date()
+                        except Exception as e:
+                            logger.debug(f"Error parsing start_time for base_date: {e}")
+                            continue
+
+                if earliest_date:
+                    base_date = earliest_date
+                    logger.info(f"🗓️ Determined base_date for schedule: {base_date}")
+
             # Convert DataFrame to tasks
-            tasks = DataService.convert_dataframe_to_tasks(task_df)
+            tasks = DataService.convert_dataframe_to_tasks(task_df, base_date)
 
             # Debug: Log task information if debug is enabled
             if debug:
@@ -97,7 +141,7 @@ class ScheduleService:
 
             # Generate schedule
             schedule = ScheduleService.generate_schedule_for_solving(
-                tasks, employee_count, days_in_schedule
+                tasks, employee_count, days_in_schedule, base_date
             )
 
             # Start solving
@@ -106,7 +150,7 @@ class ScheduleService:
                 solved_task_df,
                 new_job_id,
                 status,
-            ) = await ScheduleService.solve_schedule(schedule, debug)
+            ) = ScheduleService.solve_schedule(schedule, debug)
 
             logger.info("📈 Solver process initiated successfully")
             return emp_df, solved_task_df, new_job_id, status, state_data
@@ -124,7 +168,10 @@ class ScheduleService:
 
     @staticmethod
     def generate_schedule_for_solving(
-        tasks: list, employee_count: Optional[int], days_in_schedule: Optional[int]
+        tasks: list,
+        employee_count: Optional[int],
+        days_in_schedule: Optional[int],
+        base_date: date = None,
     ) -> EmployeeSchedule:
         """Generate a complete schedule ready for solving"""
         parameters: TimeTableDataParameters = DATA_PARAMS
@@ -177,16 +224,45 @@ class ScheduleService:
 
         logger.info(f"✅ Generated {len(employees)} employees")
 
+        # Assign employees to all tasks (both pinned and non-pinned)
+        # For single employee scenarios, assign the single employee to all tasks
+        if parameters.employee_count == 1 and len(employees) == 1:
+            main_employee = employees[0]
+            for task in tasks:
+                task.employee = main_employee
+                logger.debug(
+                    f"Assigned {main_employee.name} to task: {task.description[:30]}..."
+                )
+        else:
+            # For multi-employee scenarios, assign employees based on skills and availability
+            # This is a simple assignment - the solver will optimize later
+            for task in tasks:
+                # Find an employee with the required skill
+                suitable_employees = [
+                    emp for emp in employees if task.required_skill in emp.skills
+                ]
+                if suitable_employees:
+                    task.employee = suitable_employees[0]  # Simple assignment
+                else:
+                    # Fallback: assign the first employee
+                    task.employee = employees[0]
+                    logger.warning(
+                        f"No employee found with skill '{task.required_skill}' for task '{task.description[:30]}...', assigned {employees[0].name}"
+                    )
+
+        logger.info(f"✅ Assigned employees to {len(tasks)} tasks")
+
         return EmployeeSchedule(
             employees=employees,
             tasks=tasks,
             schedule_info=ScheduleInfo(
-                total_slots=parameters.days_in_schedule * SLOTS_PER_WORKING_DAY
+                total_slots=parameters.days_in_schedule * SLOTS_PER_WORKING_DAY,
+                base_date=base_date,
             ),
         )
 
     @staticmethod
-    async def solve_schedule(
+    def solve_schedule(
         schedule: EmployeeSchedule, debug: bool = False
     ) -> Tuple[pd.DataFrame, pd.DataFrame, str, str]:
         """
@@ -223,6 +299,7 @@ class ScheduleService:
                 "End",
                 "Duration (hours)",
                 "Required Skill",
+                "Pinned",
             ]
         ].sort_values(["Project", "Sequence"])
 
@@ -268,6 +345,7 @@ class ScheduleService:
                     "End",
                     "Duration (hours)",
                     "Required Skill",
+                    "Pinned",
                 ]
             ].sort_values(["Start"])
 

src/utils/extract_calendar.py

@@ -1,4 +1,10 @@
 from icalendar import Calendar
+from datetime import datetime, date, timezone, timedelta
+from typing import Optional, Tuple, List, Dict, Any
+from constraint_solvers.timetable.working_hours import (
+    SLOTS_PER_WORKING_DAY,
+    MORNING_SLOTS,
+)
 
 
 def extract_ical_entries(file_bytes):
@@ -23,15 +29,178 @@ def extract_ical_entries(file_bytes):
 
                     return str(val)
 
-                entries.append(
-                    {
-                        "summary": summary,
-                        "dtstart": to_iso(dtstart),
-                        "dtend": to_iso(dtend),
-                    }
-                )
+                def to_datetime(val):
+                    """Convert icalendar datetime to Python datetime object, normalized to current timezone."""
+                    if hasattr(val, "dt"):
+                        dt = val.dt
+                        if isinstance(dt, datetime):
+                            # If timezone-aware, convert to current timezone, then make naive
+                            if dt.tzinfo is not None:
+                                # Convert to local timezone then strip timezone info
+                                local_dt = dt.astimezone()
+                                return local_dt.replace(tzinfo=None)
+                            else:
+                                # Already naive, return as-is
+                                return dt
+                        elif isinstance(dt, date):
+                            # Convert date to datetime at 9 AM (naive)
+                            return datetime.combine(
+                                dt, datetime.min.time().replace(hour=9)
+                            )
+                    return None
+
+                # Parse datetime objects for slot calculation (now normalized to current timezone)
+                start_datetime = to_datetime(dtstart)
+                end_datetime = to_datetime(dtend)
+
+                entry = {
+                    "summary": summary,
+                    "dtstart": to_iso(dtstart),
+                    "dtend": to_iso(dtend),
+                }
+
+                # Add datetime objects for slot calculation
+                if start_datetime:
+                    entry["start_datetime"] = start_datetime
+                if end_datetime:
+                    entry["end_datetime"] = end_datetime
+
+                entries.append(entry)
 
         return entries, None
 
     except Exception as e:
         return None, str(e)
+
+
+def get_earliest_calendar_date(
+    calendar_entries: List[Dict[str, Any]]
+) -> Optional[date]:
+    """
+    Find the earliest date from calendar entries to use as base_date for scheduling.
+
+    Args:
+        calendar_entries: List of calendar entry dictionaries
+
+    Returns:
+        The earliest date found, or None if no valid dates found
+    """
+    earliest_date = None
+
+    for entry in calendar_entries:
+        start_datetime = entry.get("start_datetime")
+        if start_datetime and isinstance(start_datetime, datetime):
+            entry_date = start_datetime.date()
+            if earliest_date is None or entry_date < earliest_date:
+                earliest_date = entry_date
+
+    return earliest_date
+
+
+def validate_calendar_working_hours(
+    calendar_entries: List[Dict[str, Any]]
+) -> Tuple[bool, str]:
+    """
+    Validate that all calendar entries fall within standard working hours (9:00-18:00) and don't span lunch break (13:00-14:00).
+
+    Args:
+        calendar_entries: List of calendar entry dictionaries
+
+    Returns:
+        Tuple of (is_valid, error_message)
+    """
+    if not calendar_entries:
+        return True, ""
+
+    violations = []
+
+    for entry in calendar_entries:
+        summary = entry.get("summary", "Unknown Event")
+        start_datetime = entry.get("start_datetime")
+        end_datetime = entry.get("end_datetime")
+
+        if start_datetime and isinstance(start_datetime, datetime):
+            if start_datetime.hour < 9:
+                violations.append(
+                    f"'{summary}' starts at {start_datetime.hour:02d}:{start_datetime.minute:02d} (before 9:00)"
+                )
+
+        if end_datetime and isinstance(end_datetime, datetime):
+            if end_datetime.hour > 18 or (
+                end_datetime.hour == 18 and end_datetime.minute > 0
+            ):
+                violations.append(
+                    f"'{summary}' ends at {end_datetime.hour:02d}:{end_datetime.minute:02d} (after 18:00)"
+                )
+
+        # Check for lunch break spanning (13:00-14:00)
+        if (
+            start_datetime
+            and end_datetime
+            and isinstance(start_datetime, datetime)
+            and isinstance(end_datetime, datetime)
+        ):
+            start_hour_min = start_datetime.hour + start_datetime.minute / 60.0
+            end_hour_min = end_datetime.hour + end_datetime.minute / 60.0
+
+            # Check if task spans across lunch break (13:00-14:00)
+            if start_hour_min < 14.0 and end_hour_min > 13.0:
+                violations.append(
+                    f"'{summary}' ({start_datetime.hour:02d}:{start_datetime.minute:02d}-{end_datetime.hour:02d}:{end_datetime.minute:02d}) spans lunch break (13:00-14:00)"
+                )
+
+    if violations:
+        error_msg = "Calendar entries violate working constraints:\n" + "\n".join(
+            violations
+        )
+        return False, error_msg
+
+    return True, ""
+
+
+def datetime_to_slot(dt: datetime, base_date: date) -> int:
+    """
+    Convert a datetime to a 30-minute slot index within working days.
+
+    Args:
+        dt: The datetime to convert (should be naive local time)
+        base_date: The base date (slot 0 = base_date at 9:00 AM local time)
+
+    Returns:
+        The slot index (each slot = 30 minutes within working hours)
+    """
+    # Calculate which working day this datetime falls on
+    days_from_base = (dt.date() - base_date).days
+
+    # Calculate time within the working day (minutes from 9:00 AM)
+    minutes_from_9am = (dt.hour - 9) * 60 + dt.minute
+
+    # Convert to slot within the day (each slot = 30 minutes)
+    slot_within_day = round(minutes_from_9am / 30)
+
+    # Calculate total slot index
+    total_slot = days_from_base * SLOTS_PER_WORKING_DAY + slot_within_day
+
+    # Ensure non-negative slot
+    return max(0, total_slot)
+
+
+def calculate_duration_slots(start_dt: datetime, end_dt: datetime) -> int:
+    """
+    Calculate duration in 30-minute slots between two datetimes (naive local time).
+
+    Args:
+        start_dt: Start datetime (naive local time)
+        end_dt: End datetime (naive local time)
+
+    Returns:
+        Duration in 30-minute slots (minimum 1 slot)
+    """
+    # Calculate difference in minutes (both should be naive local time)
+    time_diff = end_dt - start_dt
+    total_minutes = time_diff.total_seconds() / 60
+
+    # Convert to 30-minute slots, rounding up to ensure task duration is preserved
+    duration_slots = max(1, round(total_minutes / 30))
+
+    return duration_slots

tests/data/calendar.ics

@@ -15,8 +15,8 @@ END:VEVENT
 BEGIN:VEVENT
 UID:recur-meeting-2@mock
 DTSTAMP:20240523T000000Z
-DTSTART;TZID=UTC:20250602T140000
-DTEND;TZID=UTC:20250602T150000
+DTSTART;TZID=UTC:20250602T143000
+DTEND;TZID=UTC:20250602T153000
 RRULE:FREQ=WEEKLY;BYDAY=MO,WE,FR
 SUMMARY:Project Review
 END:VEVENT
@@ -24,16 +24,16 @@ END:VEVENT
 BEGIN:VEVENT
 UID:single-event-1@mock
 DTSTAMP:20240523T000000Z
-DTSTART;TZID=UTC:20250605T130000
-DTEND;TZID=UTC:20250605T140000
+DTSTART;TZID=UTC:20250605T133000
+DTEND;TZID=UTC:20250605T143000
 SUMMARY:Client Call
 END:VEVENT
 
 BEGIN:VEVENT
 UID:single-event-2@mock
 DTSTAMP:20240523T000000Z
-DTSTART;TZID=UTC:20250616T110000
-DTEND;TZID=UTC:20250616T120000
+DTSTART;TZID=UTC:20250616T143000
+DTEND;TZID=UTC:20250616T153000
 SUMMARY:Workshop
 END:VEVENT
 
@@ -41,15 +41,15 @@ BEGIN:VEVENT
 UID:single-event-3@mock
 DTSTAMP:20240523T000000Z
 DTSTART;TZID=UTC:20250707T150000
-DTEND;TZID=UTC:20250707T163000
+DTEND;TZID=UTC:20250707T160000
 SUMMARY:Planning Session
 END:VEVENT
 
 BEGIN:VEVENT
 UID:single-event-4@mock
 DTSTAMP:20240523T000000Z
-DTSTART;TZID=UTC:20250722T093000
-DTEND;TZID=UTC:20250722T103000
+DTSTART;TZID=UTC:20250722T100000
+DTEND;TZID=UTC:20250722T110000
 SUMMARY:Demo
 END:VEVENT
 

tests/data/calendar_wrong.ics

@@ -0,0 +1,54 @@
+BEGIN:VCALENDAR
+VERSION:2.0
+PRODID:-//Mock Calendar//EN
+CALSCALE:GREGORIAN
+
+BEGIN:VEVENT
+UID:early-meeting@mock
+DTSTAMP:20240523T000000Z
+DTSTART:20250603T050000Z
+DTEND:20250603T060000Z
+SUMMARY:Early Morning Meeting
+END:VEVENT
+
+BEGIN:VEVENT
+UID:late-meeting@mock
+DTSTAMP:20240523T000000Z
+DTSTART;TZID=UTC:20250602T180000
+DTEND;TZID=UTC:20250602T190000
+SUMMARY:Evening Meeting
+END:VEVENT
+
+BEGIN:VEVENT
+UID:lunch-meeting@mock
+DTSTAMP:20240523T000000Z
+DTSTART;TZID=UTC:20250605T130000
+DTEND;TZID=UTC:20250605T140000
+SUMMARY:Lunch Meeting
+END:VEVENT
+
+BEGIN:VEVENT
+UID:long-lunch-meeting@mock
+DTSTAMP:20240523T000000Z
+DTSTART;TZID=UTC:20250616T123000
+DTEND;TZID=UTC:20250616T143000
+SUMMARY:Long Lunch Meeting
+END:VEVENT
+
+BEGIN:VEVENT
+UID:very-late-meeting@mock
+DTSTAMP:20240523T000000Z
+DTSTART;TZID=UTC:20250707T190000
+DTEND;TZID=UTC:20250707T200000
+SUMMARY:Very Late Meeting
+END:VEVENT
+
+BEGIN:VEVENT
+UID:valid-meeting@mock
+DTSTAMP:20240523T000000Z
+DTSTART;TZID=UTC:20250722T100000
+DTEND;TZID=UTC:20250722T110000
+SUMMARY:Valid Meeting
+END:VEVENT
+
+END:VCALENDAR

tests/test_constraints.py

@@ -215,8 +215,8 @@ class TestConstraints:
         task = create_task(
             task_id="task1",
             description="Overlong Task",
-            duration_slots=10,  # Task extends to slot 64 (beyond 59)
-            start_slot=55,
+            duration_slots=10,  # Task extends to slot 65 (beyond 59)
+            start_slot=56,  # Start at slot 56, end at slot 65 (beyond schedule)
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -245,11 +245,11 @@ class TestConstraints:
 
     def test_unavailable_employee_constraint_violation(self):
         """Test that tasks assigned to unavailable employees are penalized."""
-        # Assuming 16 slots per working day, tomorrow starts at slot 16
+        # With 20 slots per working day, tomorrow starts at slot 20
         task = create_task(
             task_id="task1",
             description="Task on unavailable day",
-            start_slot=16,  # Tomorrow (when Alice is unavailable)
+            start_slot=20,  # Tomorrow (when Alice is unavailable)
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -261,11 +261,11 @@ class TestConstraints:
         )
 
     def test_unavailable_employee_constraint_satisfied(self):
-        """Test that tasks assigned on available days are not penalized."""
+        """Test that tasks not on unavailable days are not penalized."""
         task = create_task(
             task_id="task1",
             description="Task on available day",
-            start_slot=0,  # Today (when Alice is available)
+            start_slot=0,  # Today (Alice is available)
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -371,8 +371,8 @@ class TestConstraints:
         task = create_task(
             task_id="task1",
             description="Task spanning lunch",
-            start_slot=6,  # Starts in morning (slot 6)
-            duration_slots=4,  # Ends in afternoon (slot 10), spans lunch
+            start_slot=7,  # Starts at 12:30 (slot 7), spans lunch break
+            duration_slots=4,  # 2 hours, ends at 14:30 (slot 11)
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -383,30 +383,13 @@ class TestConstraints:
             .penalizes_by(1)
         )
 
-    def test_no_lunch_break_spanning_constraint_satisfied_morning(self):
-        """Test that tasks contained in morning session are not penalized."""
+    def test_no_lunch_break_spanning_constraint_satisfied(self):
+        """Test that tasks not spanning lunch break are not penalized."""
         task = create_task(
             task_id="task1",
-            description="Morning task",
-            start_slot=2,  # Morning session
-            duration_slots=4,  # Stays in morning (slots 2-5)
-            required_skill="Python",
-            employee=self.employee_alice,
-        )
-
-        (
-            self.constraint_verifier.verify_that(no_lunch_break_spanning)
-            .given(task, self.employee_alice, self.schedule_info)
-            .penalizes_by(0)
-        )
-
-    def test_no_lunch_break_spanning_constraint_satisfied_afternoon(self):
-        """Test that tasks contained in afternoon session are not penalized."""
-        task = create_task(
-            task_id="task1",
-            description="Afternoon task",
-            start_slot=10,  # Afternoon session (slot 10 = 3rd hour of afternoon)
-            duration_slots=4,  # Stays in afternoon (slots 10-13)
+            description="Task before lunch",
+            start_slot=0,  # Starts at 9:00
+            duration_slots=4,  # 2 hours, ends at 11:00
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -441,11 +424,11 @@ class TestConstraints:
 
     def test_undesired_day_for_employee_constraint_violation(self):
         """Test that tasks on undesired days incur soft penalty."""
-        # Assuming 16 slots per working day, day after tomorrow starts at slot 32
+        # With 20 slots per working day, day after tomorrow starts at slot 40
         task = create_task(
             task_id="task1",
             description="Task on undesired day",
-            start_slot=32,  # Day after tomorrow (Alice's undesired date)
+            start_slot=40,  # Day after tomorrow (Alice's undesired date)
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -457,11 +440,11 @@ class TestConstraints:
         )
 
     def test_undesired_day_for_employee_constraint_satisfied(self):
-        """Test that tasks on neutral days don't incur undesired day penalty."""
+        """Test that tasks not on undesired days are not penalized."""
         task = create_task(
             task_id="task1",
             description="Task on neutral day",
-            start_slot=0,  # Today (neutral for Alice, though it's also desired)
+            start_slot=0,  # Today (neutral for Alice)
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -473,7 +456,8 @@ class TestConstraints:
         )
 
     def test_desired_day_for_employee_constraint_reward(self):
-        """Test that tasks on desired days provide soft reward."""
+        """Test that tasks on desired days provide reward."""
+        # Alice's desired day is today (slot 0-19)
         task = create_task(
             task_id="task1",
             description="Task on desired day",
@@ -493,7 +477,7 @@ class TestConstraints:
         task = create_task(
             task_id="task1",
             description="Task on neutral day",
-            start_slot=16,  # Tomorrow (neutral for Alice)
+            start_slot=20,  # Tomorrow (neutral for Alice)
             required_skill="Python",
             employee=self.employee_alice,
         )
@@ -606,7 +590,7 @@ class TestConstraints:
             create_task(
                 "task2",
                 "Valid Java Task",
-                start_slot=8,  # Afternoon session, non-overlapping
+                start_slot=10,  # After lunch break (14:00), non-overlapping
                 required_skill="Java",
                 project_id="project1",
                 sequence_number=2,
@@ -615,7 +599,7 @@ class TestConstraints:
             create_task(
                 "task3",
                 "Bob's Valid Task",
-                start_slot=12,
+                start_slot=14,  # After lunch break (14:00)
                 required_skill="Java",
                 project_id="project2",
                 sequence_number=1,
@@ -675,7 +659,7 @@ class TestConstraints:
                 self.employee_bob,
                 self.schedule_info,
             )
-            .scores(HardSoftDecimalScore.of(Decimal("-5"), Decimal("-0.12132")))
+            .scores(HardSoftDecimalScore.of(Decimal("-4"), Decimal("-0.12132")))
         )
 
 
@@ -728,11 +712,11 @@ def create_task(
     )
 
 
-def create_schedule_info(total_slots=48):
+def create_schedule_info(total_slots=60):
     """Create a schedule info object with specified total slots.
-    Default is 48 slots = 3 working days * 16 slots per working day.
+    Default is 60 slots = 3 working days * 20 slots per working day.
     """
-    return ScheduleInfo(total_slots=total_slots)
+    return ScheduleInfo(total_slots=total_slots, base_date=date.today())
 
 
 def create_standard_employees(dates):

tests/test_factory.py

@@ -1,4 +1,10 @@
 import pytest
+import time
+import pandas as pd
+import traceback
+from io import StringIO
+from datetime import datetime, date, timedelta
+from typing import List, Dict, Tuple, Optional, Any
 
 from src.utils.load_secrets import load_secrets
 
@@ -7,8 +13,422 @@ load_secrets("tests/secrets/creds.py")
 
 import factory.data.provider as data_provider
 from src.utils.extract_calendar import extract_ical_entries
+from src.handlers.mcp_backend import process_message_and_attached_file
+from src.services import ScheduleService, StateService
+from src.services.data import DataService
+from src.factory.data.formatters import schedule_to_dataframe
 
+# Add cleanup fixture for proper solver shutdown
+@pytest.fixture(scope="session", autouse=True)
+def cleanup_solver():
+    """Automatically cleanup solver resources after all tests complete."""
+    yield  # Run tests
 
+    # Cleanup: Terminate all active solver jobs and shutdown solver manager
+    try:
+        from constraint_solvers.timetable.solver import solver_manager
+        from src.state import app_state
+
+        # Clear all stored schedules first
+        app_state.clear_solved_schedules()
+
+        # Terminate all active solver jobs gracefully using the Timefold terminateEarly method
+        if hasattr(solver_manager, "terminateEarly"):
+            # According to Timefold docs, terminateEarly() affects all jobs for this manager
+            try:
+                solver_manager.terminateEarly()
+                print("🧹 Terminated all active solver jobs")
+            except Exception as e:
+                print(f"⚠️ Error terminating solver jobs: {e}")
+
+        # Try additional cleanup methods if available
+        if hasattr(solver_manager, "close"):
+            solver_manager.close()
+            print("🔒 Closed solver manager")
+        elif hasattr(solver_manager, "shutdown"):
+            solver_manager.shutdown()
+            print("🔒 Shutdown solver manager")
+        else:
+            print("⚠️ No explicit close/shutdown method found on solver manager")
+
+        print("✅ Solver cleanup completed successfully")
+
+    except Exception as e:
+        print(f"⚠️ Error during solver cleanup: {e}")
+        # Don't fail tests if cleanup fails, but log it
+
+
+# Test Configuration
+TEST_CONFIG = {
+    "valid_calendar": "tests/data/calendar.ics",
+    "invalid_calendar": "tests/data/calendar_wrong.ics",
+    "default_employee_count": 1,
+    "default_project_id": "PROJECT",
+    "solver_max_polls": 30,
+    "solver_poll_interval": 1,
+    "datetime_tolerance_seconds": 60,
+}
+
+
+# Fixtures and Helper Functions
+@pytest.fixture
+def valid_calendar_entries():
+    """Load valid calendar entries for testing."""
+    return load_calendar_entries(TEST_CONFIG["valid_calendar"])
+
+
+@pytest.fixture
+def invalid_calendar_entries():
+    """Load invalid calendar entries for testing."""
+    return load_calendar_entries(TEST_CONFIG["invalid_calendar"])
+
+
+def load_calendar_entries(file_path: str) -> List[Dict]:
+    """Load and extract calendar entries from an iCS file."""
+    with open(file_path, "rb") as f:
+        file_bytes = f.read()
+
+    entries, error = extract_ical_entries(file_bytes)
+    assert error is None, f"Calendar extraction failed: {error}"
+    assert len(entries) > 0, "No calendar entries found"
+
+    return entries
+
+
+def print_calendar_entries(entries: List[Dict], title: str = "Calendar Entries"):
+    """Print calendar entries in a formatted way."""
+    print(f"\n📅 {title} ({len(entries)} entries):")
+    for i, entry in enumerate(entries):
+        start_dt = entry.get("start_datetime")
+        end_dt = entry.get("end_datetime")
+        print(f"  {i+1}. {entry['summary']}: {start_dt} → {end_dt}")
+
+
+def calculate_required_schedule_days(
+    calendar_entries: List[Dict], buffer_days: int = 30
+) -> int:
+    """Calculate required schedule days based on calendar entries."""
+    if not calendar_entries:
+        return 60  # Default
+
+    earliest_date = None
+    latest_date = None
+
+    for entry in calendar_entries:
+        for dt_key in ["start_datetime", "end_datetime"]:
+            dt = entry.get(dt_key)
+            if dt and isinstance(dt, datetime):
+                entry_date = dt.date()
+                if earliest_date is None or entry_date < earliest_date:
+                    earliest_date = entry_date
+                if latest_date is None or entry_date > latest_date:
+                    latest_date = entry_date
+
+    if earliest_date and latest_date:
+        calendar_span = (latest_date - earliest_date).days + 1
+        return calendar_span + buffer_days
+    else:
+        return 60  # Fallback
+
+
+async def generate_mcp_data_helper(
+    calendar_entries: List[Dict],
+    user_message: str,
+    project_id: str = None,
+    employee_count: int = None,
+    days_in_schedule: int = None,
+) -> pd.DataFrame:
+    """Helper function to generate MCP data with consistent defaults."""
+    project_id = project_id or TEST_CONFIG["default_project_id"]
+    employee_count = employee_count or TEST_CONFIG["default_employee_count"]
+
+    if days_in_schedule is None:
+        days_in_schedule = calculate_required_schedule_days(calendar_entries)
+
+    return await data_provider.generate_mcp_data(
+        calendar_entries=calendar_entries,
+        user_message=user_message,
+        project_id=project_id,
+        employee_count=employee_count,
+        days_in_schedule=days_in_schedule,
+    )
+
+
+async def solve_schedule_with_polling(
+    initial_df: pd.DataFrame, employee_count: int = None
+) -> Optional[pd.DataFrame]:
+    """Solve schedule with polling and return the result."""
+    employee_count = employee_count or TEST_CONFIG["default_employee_count"]
+    required_days = calculate_required_schedule_days([])  # Use default
+
+    # Extract date range from pinned tasks for better schedule length calculation
+    pinned_tasks = initial_df[initial_df.get("Pinned", False) == True]
+    if not pinned_tasks.empty:
+        required_days = calculate_required_schedule_days_from_df(pinned_tasks)
+
+    state_data = {
+        "task_df_json": initial_df.to_json(orient="split"),
+        "employee_count": employee_count,
+        "days_in_schedule": required_days,
+    }
+
+    # Start solving
+    (
+        emp_df,
+        task_df,
+        job_id,
+        status,
+        state_data,
+    ) = await ScheduleService.solve_schedule_from_state(
+        state_data=state_data, job_id=None, debug=True
+    )
+
+    print(f"Solver started with job_id: {job_id}")
+    print(f"Initial status: {status}")
+
+    # Poll for solution using the correct StateService methods
+    max_polls = TEST_CONFIG["solver_max_polls"]
+    poll_interval = TEST_CONFIG["solver_poll_interval"]
+
+    final_df = None
+
+    try:
+        for poll_count in range(1, max_polls + 1):
+            print(f"  Polling {poll_count}/{max_polls}...")
+            time.sleep(poll_interval)
+
+            # Use StateService to check for completed solution
+            if StateService.has_solved_schedule(job_id):
+                solved_schedule = StateService.get_solved_schedule(job_id)
+
+                if solved_schedule is not None:
+                    print(f"✅ Schedule solved after {poll_count} polls!")
+
+                    # Convert solved schedule to DataFrame
+                    final_df = schedule_to_dataframe(solved_schedule)
+
+                    # Generate status message to check for failures
+                    status_message = ScheduleService.generate_status_message(
+                        solved_schedule
+                    )
+
+                    if "CONSTRAINTS VIOLATED" in status_message:
+                        print(f"❌ Solver failed: {status_message}")
+                        final_df = None
+                    else:
+                        print(f"✅ Solver succeeded: {status_message}")
+
+                    break
+
+        if final_df is None:
+            print("⏰ Solver timed out after max polls")
+
+    finally:
+        # Clean up: Ensure solver job is terminated
+        try:
+            from constraint_solvers.timetable.solver import solver_manager
+
+            # Terminate the specific job to free resources using Timefold's terminateEarly
+            if hasattr(solver_manager, "terminateEarly"):
+                try:
+                    solver_manager.terminateEarly(job_id)
+                    print(f"🧹 Terminated solver job: {job_id}")
+                except Exception as e:
+                    # If specific job termination fails, try to terminate all jobs
+                    print(f"⚠️ Error terminating specific job {job_id}: {e}")
+                    try:
+                        solver_manager.terminateEarly()
+                        print(
+                            f"🧹 Terminated all solver jobs after specific termination failed"
+                        )
+                    except Exception as e2:
+                        print(f"⚠️ Could not terminate any solver jobs: {e2}")
+            else:
+                print(f"⚠️ terminateEarly method not available on solver_manager")
+        except Exception as e:
+            print(f"⚠️ Could not access solver_manager for cleanup: {e}")
+
+    return final_df
+
+
+def calculate_required_schedule_days_from_df(
+    pinned_df: pd.DataFrame, buffer_days: int = 30
+) -> int:
+    """Calculate required schedule days from DataFrame with pinned tasks."""
+    earliest_date = None
+    latest_date = None
+
+    for _, row in pinned_df.iterrows():
+        for date_col in ["Start", "End"]:
+            date_val = row.get(date_col)
+            if date_val is not None:
+                try:
+                    if isinstance(date_val, str):
+                        dt = datetime.fromisoformat(date_val.replace("Z", "+00:00"))
+                    else:
+                        dt = pd.to_datetime(date_val).to_pydatetime()
+
+                    if earliest_date is None or dt.date() < earliest_date:
+                        earliest_date = dt.date()
+                    if latest_date is None or dt.date() > latest_date:
+                        latest_date = dt.date()
+                except:
+                    continue
+
+    if earliest_date and latest_date:
+        calendar_span = (latest_date - earliest_date).days + 1
+        return calendar_span + buffer_days
+    else:
+        return 60  # Default
+
+
+def analyze_schedule_dataframe(
+    df: pd.DataFrame, title: str = "Schedule Analysis"
+) -> Dict[str, Any]:
+    """Analyze a schedule DataFrame and return summary information."""
+    existing_tasks = df[df["Project"] == "EXISTING"]
+    project_tasks = df[df["Project"] == "PROJECT"]
+
+    analysis = {
+        "total_tasks": len(df),
+        "existing_tasks": len(existing_tasks),
+        "project_tasks": len(project_tasks),
+        "existing_df": existing_tasks,
+        "project_df": project_tasks,
+    }
+
+    print(f"\n📊 {title} ({analysis['total_tasks']} tasks):")
+    print(f"  - EXISTING (calendar): {analysis['existing_tasks']} tasks")
+    print(f"  - PROJECT (LLM): {analysis['project_tasks']} tasks")
+
+    return analysis
+
+
+def verify_calendar_tasks_pinned(existing_tasks_df: pd.DataFrame) -> bool:
+    """Verify that all calendar tasks are pinned."""
+    print(f"\n🔒 Verifying calendar tasks are pinned:")
+    all_pinned = True
+
+    for _, task in existing_tasks_df.iterrows():
+        is_pinned = task.get("Pinned", False)
+        task_name = task["Task"]
+        print(f"  - {task_name}: pinned = {is_pinned}")
+
+        if not is_pinned:
+            all_pinned = False
+            print(f"    ❌ Calendar task should be pinned!")
+        else:
+            print(f"    ✅ Calendar task properly pinned")
+
+    return all_pinned
+
+
+def verify_time_preservation(
+    original_times: Dict, final_tasks_df: pd.DataFrame
+) -> bool:
+    """Verify that calendar tasks preserved their original times."""
+    print(f"\n🔍 Verifying calendar tasks preserved their original times:")
+    time_preserved = True
+
+    for _, task in final_tasks_df.iterrows():
+        task_name = task["Task"]
+        final_start = task["Start"]
+
+        original = original_times.get(task_name)
+        if original is None:
+            print(f"  - {task_name}: ❌ Not found in original data")
+            time_preserved = False
+            continue
+
+        # Normalize and compare times
+        preserved = compare_datetime_values(original["start"], final_start)
+
+        print(f"  - {task_name}:")
+        print(f"    Original: {original['start']}")
+        print(f"    Final:    {final_start}")
+        print(f"    Preserved: {'✅' if preserved else '❌'}")
+
+        if not preserved:
+            time_preserved = False
+
+    return time_preserved
+
+
+def compare_datetime_values(dt1: Any, dt2: Any, tolerance_seconds: int = None) -> bool:
+    """Compare two datetime values with tolerance for timezone differences."""
+    tolerance = tolerance_seconds or TEST_CONFIG["datetime_tolerance_seconds"]
+
+    # Convert to comparable datetime objects
+    try:
+        if isinstance(dt1, str):
+            dt1 = datetime.fromisoformat(dt1.replace("Z", "+00:00"))
+
+        if isinstance(dt2, str):
+            dt2 = datetime.fromisoformat(dt2.replace("Z", "+00:00"))
+
+        # Normalize timezones for comparison
+        if dt1.tzinfo is not None and dt2.tzinfo is None:
+            dt1 = dt1.replace(tzinfo=None)
+        elif dt1.tzinfo is None and dt2.tzinfo is not None:
+            dt2 = dt2.replace(tzinfo=None)
+
+        return abs((dt1 - dt2).total_seconds()) < tolerance
+    except:
+        return False
+
+
+def store_original_calendar_times(existing_tasks_df: pd.DataFrame) -> Dict[str, Dict]:
+    """Store original calendar task times for later comparison."""
+    original_times = {}
+
+    for _, task in existing_tasks_df.iterrows():
+        original_times[task["Task"]] = {
+            "start": task["Start"],
+            "end": task["End"],
+            "pinned": task.get("Pinned", False),
+        }
+
+    print("\n📌 Original calendar task times:")
+    for task_name, times in original_times.items():
+        print(
+            f"  - {task_name}: {times['start']} → {times['end']} (pinned: {times['pinned']})"
+        )
+
+    return original_times
+
+
+def verify_llm_tasks_scheduled(project_tasks_df: pd.DataFrame) -> bool:
+    """Verify that LLM tasks are properly scheduled and not pinned."""
+    print(f"\n🔄 Verifying LLM tasks were properly scheduled:")
+    all_scheduled = True
+
+    for _, task in project_tasks_df.iterrows():
+        task_name = task["Task"]
+        start_time = task["Start"]
+        is_pinned = task.get("Pinned", False)
+
+        print(f"  - {task_name}:")
+        print(f"    Scheduled at: {start_time}")
+        print(f"    Pinned: {is_pinned}")
+
+        # LLM tasks should not be pinned
+        if is_pinned:
+            all_scheduled = False
+            print(f"    ❌ LLM task should not be pinned!")
+        else:
+            print(f"    ✅ LLM task properly unpinned")
+
+        # LLM tasks should have been scheduled to actual times
+        if start_time is None or start_time == "":
+            all_scheduled = False
+            print(f"    ❌ LLM task was not scheduled!")
+        else:
+            print(f"    ✅ LLM task was scheduled")
+
+    return all_scheduled
+
+
+# Test Functions
 @pytest.mark.asyncio
 async def test_factory_demo_agent():
     # Use a simple string as the project description
@@ -44,23 +464,14 @@ async def test_factory_demo_agent():
 
 
 @pytest.mark.asyncio
-async def test_factory_mcp():
-    # Load the real calendar.ics file
-    with open("tests/data/calendar.ics", "rb") as f:
-        file_bytes = f.read()
-    entries, err = extract_ical_entries(file_bytes)
-    assert err is None
-    assert entries is not None
-    assert len(entries) > 0
-
-    print("\nEntries:")
-    print(entries)
+async def test_factory_mcp(valid_calendar_entries):
+    print_calendar_entries(valid_calendar_entries, "Loaded Calendar Entries")
 
     # Use a made-up user message
     user_message = "Create a new AWS VPC."
 
     # Call generate_mcp_data directly
-    df = await data_provider.generate_mcp_data(entries, user_message)
+    df = await generate_mcp_data_helper(valid_calendar_entries, user_message)
 
     # Assert the DataFrame is not empty
     assert df is not None
@@ -68,3 +479,297 @@ async def test_factory_mcp():
 
     # Print the DataFrame for debug
     print(df)
+
+
+@pytest.mark.asyncio
+async def test_mcp_workflow_calendar_pinning(valid_calendar_entries):
+    """
+    Test that verifies calendar tasks (EXISTING) remain pinned to their original times
+    while LLM tasks (PROJECT) are rescheduled around them in the MCP workflow.
+    """
+    print("\n" + "=" * 60)
+    print("Testing MCP Workflow: Calendar Task Pinning vs LLM Task Scheduling")
+    print("=" * 60)
+
+    print_calendar_entries(valid_calendar_entries, "Loaded Calendar Entries")
+
+    # Generate initial MCP data
+    user_message = "Set up CI/CD pipeline and configure monitoring system"
+    initial_df = await generate_mcp_data_helper(valid_calendar_entries, user_message)
+
+    # Analyze initial schedule
+    analysis = analyze_schedule_dataframe(initial_df, "Generated Initial Data")
+
+    # Store original calendar task times and verify they're pinned
+    original_times = store_original_calendar_times(analysis["existing_df"])
+    calendar_pinned = verify_calendar_tasks_pinned(analysis["existing_df"])
+    assert calendar_pinned, "Calendar tasks should be pinned!"
+
+    # Solve the schedule
+    print(f"\n🔧 Running MCP workflow to solve schedule...")
+    solved_schedule_df = await solve_schedule_with_polling(initial_df)
+
+    if solved_schedule_df is None:
+        print("⏰ Solver timed out - this might be due to complex constraints")
+        print("⚠️  Skipping verification steps for timeout case")
+        return
+
+    # Analyze final schedule (solved_schedule_df is already a DataFrame)
+    final_analysis = analyze_schedule_dataframe(solved_schedule_df, "Final Schedule")
+
+    # Verify calendar tasks preserved their times
+    time_preserved = verify_time_preservation(
+        original_times, final_analysis["existing_df"]
+    )
+
+    # Verify LLM tasks were properly scheduled
+    llm_scheduled = verify_llm_tasks_scheduled(final_analysis["project_df"])
+
+    # Final assertions
+    assert time_preserved, "Calendar tasks did not preserve their original times!"
+    assert llm_scheduled, "LLM tasks were not properly scheduled!"
+
+    print(f"\n🎉 MCP Workflow Test Results:")
+    print(f"✅ Calendar tasks preserved original times: {time_preserved}")
+    print(f"✅ LLM tasks were properly scheduled: {llm_scheduled}")
+    print(
+        "🎯 MCP workflow test passed! Calendar tasks are pinned, LLM tasks are flexible."
+    )
+
+
+@pytest.mark.asyncio
+async def test_calendar_validation_rejects_invalid_entries(invalid_calendar_entries):
+    """
+    Test that calendar validation properly rejects entries that violate working hours constraints.
+    """
+    print("\n" + "=" * 60)
+    print("Testing Calendar Validation: Constraint Violations")
+    print("=" * 60)
+
+    print_calendar_entries(invalid_calendar_entries, "Invalid Calendar Entries")
+
+    # Test that generate_mcp_data raises an error due to validation failure
+    user_message = "Simple test task"
+
+    print(f"\n❌ Attempting to generate MCP data with invalid calendar (should fail)...")
+
+    with pytest.raises(ValueError) as exc_info:
+        await generate_mcp_data_helper(invalid_calendar_entries, user_message)
+
+    error_message = str(exc_info.value)
+    print(f"\n✅ Validation correctly rejected invalid calendar:")
+    print(f"Error: {error_message}")
+
+    # Verify the error message contains expected constraint violations
+    assert "Calendar entries violate working constraints" in error_message
+    # Check for specific violations that should be detected
+    assert (
+        "Early Morning Meeting" in error_message
+        or "07:00" in error_message
+        or "before 9:00" in error_message
+    ), f"Should detect early morning violation in: {error_message}"
+    assert (
+        "Evening Meeting" in error_message
+        or "21:00" in error_message
+        or "after 18:00" in error_message
+    ), f"Should detect evening violation in: {error_message}"
+    assert (
+        "Very Late Meeting" in error_message or "22:00" in error_message
+    ), f"Should detect very late violation in: {error_message}"
+
+    print("✅ All expected constraint violations were detected!")
+
+
+@pytest.mark.asyncio
+async def test_calendar_validation_accepts_valid_entries(valid_calendar_entries):
+    """
+    Test that calendar validation accepts valid entries and processing continues normally.
+    """
+    print("\n" + "=" * 60)
+    print("Testing Calendar Validation: Valid Entries")
+    print("=" * 60)
+
+    print_calendar_entries(valid_calendar_entries, "Valid Calendar Entries")
+
+    # Test that generate_mcp_data succeeds with valid calendar
+    user_message = "Simple test task"
+
+    print(
+        f"\n✅ Attempting to generate MCP data with valid calendar (should succeed)..."
+    )
+
+    try:
+        initial_df = await generate_mcp_data_helper(
+            valid_calendar_entries, user_message
+        )
+
+        print(f"✅ Validation passed! Generated {len(initial_df)} tasks successfully")
+
+        # Analyze and verify the result
+        analysis = analyze_schedule_dataframe(initial_df, "Generated Schedule")
+
+        assert analysis["existing_tasks"] > 0, "Should have calendar tasks"
+        assert analysis["project_tasks"] > 0, "Should have LLM tasks"
+
+        # Verify all calendar tasks are pinned
+        calendar_pinned = verify_calendar_tasks_pinned(analysis["existing_df"])
+        assert calendar_pinned, "All calendar tasks should be properly pinned!"
+
+    except Exception as e:
+        pytest.fail(f"Valid calendar should not raise an error, but got: {e}")
+
+
+@pytest.mark.asyncio
+async def test_mcp_backend_end_to_end():
+    """
+    Test the complete MCP backend workflow using the actual handler function.
+    This tests the full process_message_and_attached_file flow.
+    """
+    print("\n" + "=" * 50)
+    print("Testing MCP Backend End-to-End")
+    print("=" * 50)
+
+    # Test message for LLM tasks
+    message_body = "Implement user authentication and setup database migrations"
+    file_path = TEST_CONFIG["valid_calendar"]
+
+    # Run the MCP backend handler
+    print(f"📨 Processing message: '{message_body}'")
+    print(f"📁 Using calendar file: {file_path}")
+
+    result = await process_message_and_attached_file(file_path, message_body)
+
+    # Verify the result structure
+    assert isinstance(result, dict), "Result should be a dictionary"
+    assert result.get("status") in [
+        "success",
+        "timeout",
+    ], f"Unexpected status: {result.get('status')}"
+
+    if result.get("status") == "success":
+        print("✅ MCP backend completed successfully!")
+
+        # Verify result contains expected fields
+        assert "schedule" in result, "Result should contain schedule data"
+        assert "calendar_entries" in result, "Result should contain calendar entries"
+        assert "file_info" in result, "Result should contain file info"
+
+        schedule = result["schedule"]
+        calendar_entries = result["calendar_entries"]
+
+        print(f"📅 Calendar entries processed: {len(calendar_entries)}")
+        print(f"📋 Total scheduled tasks: {len(schedule)}")
+
+        # Analyze the schedule
+        existing_tasks = [t for t in schedule if t.get("Project") == "EXISTING"]
+        project_tasks = [t for t in schedule if t.get("Project") == "PROJECT"]
+
+        print(f"🔒 EXISTING (calendar) tasks: {len(existing_tasks)}")
+        print(f"🔧 PROJECT (LLM) tasks: {len(project_tasks)}")
+
+        # Verify we have both types of tasks
+        assert len(existing_tasks) > 0, "Should have calendar tasks"
+        assert len(project_tasks) > 0, "Should have LLM-generated tasks"
+
+        # Check that project tasks exist and are scheduled
+        for task in project_tasks:
+            task_name = task.get("Task", "Unknown")
+            start_time = task.get("Start")
+            print(f"⏰ LLM task '{task_name}': scheduled at {start_time}")
+            assert (
+                start_time is not None
+            ), f"LLM task '{task_name}' should have a scheduled start time"
+
+        print("🎯 MCP backend end-to-end test passed!")
+
+    elif result.get("status") == "timeout":
+        print("⏰ MCP backend timed out - this is acceptable for testing")
+        print("The solver may need more time for complex schedules")
+
+        # Still verify basic structure
+        assert "calendar_entries" in result, "Result should contain calendar entries"
+        assert "file_info" in result, "Result should contain file info"
+
+    else:
+        # Handle error cases
+        error_msg = result.get("error", "Unknown error")
+        print(f"❌ MCP backend failed: {error_msg}")
+        assert False, f"MCP backend failed: {error_msg}"
+
+    print("✅ MCP backend structure and behavior verified!")
+
+
+@pytest.mark.asyncio
+async def test_mcp_datetime_debug(valid_calendar_entries):
+    """
+    Debug test to isolate the datetime conversion issue in MCP workflow.
+    """
+    print("\n" + "=" * 50)
+    print("Testing MCP Datetime Conversion Debug")
+    print("=" * 50)
+
+    print(f"\n📅 Calendar entries debug:")
+    for i, entry in enumerate(valid_calendar_entries):
+        print(f"  {i+1}. {entry['summary']}:")
+        print(
+            f"     start_datetime: {entry.get('start_datetime')} (type: {type(entry.get('start_datetime'))})"
+        )
+        print(
+            f"     end_datetime: {entry.get('end_datetime')} (type: {type(entry.get('end_datetime'))})"
+        )
+
+    # Generate MCP data and check the DataFrame structure
+    user_message = "Simple test task"
+
+    try:
+        # Generate data with calculated schedule length
+        required_days = calculate_required_schedule_days(
+            valid_calendar_entries, buffer_days=10
+        )
+        print(f"📊 Using {required_days} total schedule days")
+
+        initial_df = await generate_mcp_data_helper(
+            valid_calendar_entries, user_message, days_in_schedule=required_days
+        )
+
+        print(f"\n📊 Generated DataFrame columns: {list(initial_df.columns)}")
+        print(f"📊 DataFrame shape: {initial_df.shape}")
+        print(f"📊 DataFrame dtypes:\n{initial_df.dtypes}")
+
+        # Check the Start and End column formats
+        print(f"\n🕒 Start column sample:")
+        for i, row in initial_df.head(3).iterrows():
+            start_val = row.get("Start")
+            print(f"  Row {i}: {start_val} (type: {type(start_val)})")
+
+        # Test conversion to JSON and back
+        json_str = initial_df.to_json(orient="split")
+        print(f"\n📄 JSON conversion successful")
+
+        # Test parsing back
+        task_df_back = pd.read_json(StringIO(json_str), orient="split")
+        print(f"📄 JSON parsing back successful")
+        print(f"📄 Parsed dtypes:\n{task_df_back.dtypes}")
+
+        # Test task conversion with minimal error handling
+        print(f"\n🔄 Testing task conversion...")
+
+        # Only try with the first task to isolate issues
+        single_task_df = task_df_back.head(1)
+        print(f"Single task for testing:\n{single_task_df}")
+
+        tasks = DataService.convert_dataframe_to_tasks(single_task_df)
+        print(f"✅ Successfully converted {len(tasks)} tasks")
+
+        for task in tasks:
+            print(f"  Task: {task.description}")
+            print(f"    start_slot: {task.start_slot} (type: {type(task.start_slot)})")
+            print(f"    pinned: {task.pinned}")
+            print(f"    project_id: {task.project_id}")
+
+    except Exception as e:
+        print(f"❌ Error in MCP data generation/conversion: {e}")
+        traceback.print_exc()
+        raise
+
+    print("🎯 MCP datetime debug test completed!")