feat: Add explainability system and judge override infrastructure

- Created scheduler/control/explainability.py with DecisionStep and ExplainabilityEngine
* Provides step-by-step reasoning for scheduling decisions
* Explains ripeness status, priority scores, and policy selection

- Created scheduler/control/overrides.py with Override and OverrideManager
* Supports 8 override types: RIPENESS, PRIORITY, ADD_CASE, REMOVE_CASE, etc.
* JudgePreferences for capacity, blocked dates, case type preferences
* CauseListDraft for draft-approval workflow with acceptance tracking
* Full audit trail export capability

- Modified scheduler/simulation/events.py to log decision metadata
* Added columns: priority_score, age_days, readiness_score, is_urgent, adj_boost
* Enables verification of scheduling decisions

- Modified scheduler/simulation/engine.py
* Calculate and log adjournment boost in priority scoring
* Full metadata logging for scheduled cases

- Added scripts/demo_explainability_and_controls.py
* Demonstrates explainability engine with example decisions
* Shows judge override mechanisms and audit trail

- Added scripts/generate_all_cause_lists.py
* Generates compiled cause lists from simulation events
* Creates statistics and visualizations across scenarios

- Updated README.md with explainability and control system features
- Refactored main.py to use court_scheduler CLI
- Updated pyproject.toml dependencies

Phase 6.5 (explainability + override infrastructure) complete.
Next: Integrate overrides into simulation engine.

README.md

@@ -56,19 +56,45 @@ This project delivers a complete court scheduling system for the Code4Change hac
 
 ## Quick Start
 
-### 1. Run EDA Pipeline
+### Using the CLI (Recommended)
+
+The system provides a unified CLI for all operations:
+
+```bash
+# See all available commands
+court-scheduler --help
+
+# Run EDA pipeline
+court-scheduler eda
+
+# Generate test cases
+court-scheduler generate --cases 10000 --output data/generated/cases.csv
+
+# Run simulation
+court-scheduler simulate --days 384 --start 2024-01-01 --log-dir data/sim_runs/test_run
+
+# Run full workflow (EDA -> Generate -> Simulate)
+court-scheduler workflow --cases 10000 --days 384
+```
+
+### Legacy Methods (Still Supported)
+
+<details>
+<summary>Click to see old script-based approach</summary>
+
+#### 1. Run EDA Pipeline
 ```bash
 # Extract parameters from historical data
 uv run python main.py
 ```
 
-### 2. Generate Case Dataset
+#### 2. Generate Case Dataset
 ```bash
-# Generate 10,000 synthetic cases with realistic distributions
+# Generate 10,000 synthetic cases
 uv run python -c "from scheduler.data.case_generator import CaseGenerator; from datetime import date; from pathlib import Path; gen = CaseGenerator(start=date(2022,1,1), end=date(2023,12,31), seed=42); cases = gen.generate(10000, stage_mix_auto=True); CaseGenerator.to_csv(cases, Path('data/generated/cases.csv')); print(f'Generated {len(cases)} cases')"
 ```
 
-### 3. Run Simulation
+#### 3. Run Simulation
 ```bash
 # 2-year simulation with ripeness classification
 uv run python scripts/simulate.py --days 384 --start 2024-01-01 --log-dir data/sim_runs/test_run
@@ -76,6 +102,7 @@ uv run python scripts/simulate.py --days 384 --start 2024-01-01 --log-dir data/s
 # Quick 60-day test
 uv run python scripts/simulate.py --days 60
 ```
+</details>
 
 ## Usage
 

main.py

@@ -1,23 +1,11 @@
-"""Entrypoint to run the full EDA + parameter pipeline.
+#!/usr/bin/env python
+"""Main entry point for Court Scheduling System.
 
-Order:
-1. Load & clean (save Parquet + metadata)
-2. Visual EDA (plots + CSV summaries)
-3. Parameter extraction (JSON/CSV priors + features)
+This file provides the primary entry point for the project.
+It invokes the CLI which provides all scheduling system operations.
 """
 
-from src.eda_exploration import run_exploration
-from src.eda_load_clean import run_load_and_clean
-from src.eda_parameters import run_parameter_export
+from court_scheduler.cli import main
 
 if __name__ == "__main__":
-    print("Step 1/3: Load and clean")
-    run_load_and_clean()
-
-    print("\nStep 2/3: Exploratory analysis and plots")
-    run_exploration()
-
-    print("\nStep 3/3: Parameter extraction for simulation/scheduler")
-    run_parameter_export()
-
-    print("\nAll steps complete.")
+    main()

pyproject.toml

@@ -18,7 +18,9 @@ dependencies = [
     "typer>=0.12",
     "simpy>=4.1",
     "scipy>=1.14",
-    "scikit-learn>=1.5"
+    "scikit-learn>=1.5",
+    "streamlit>=1.28",
+    "altair>=5.0"
 ]
 
 [project.optional-dependencies]
@@ -30,11 +32,6 @@ dev = [
     "hypothesis>=6.0",
     "mypy>=1.11"
 ]
-graph = [
-    "neo4j>=5.0",
-    "igraph>=0.11",
-    "graph-tool>=2.45; sys_platform != 'win32'"
-]
 
 [project.scripts]
 court-scheduler = "court_scheduler.cli:app"
@@ -43,6 +40,9 @@ court-scheduler = "court_scheduler.cli:app"
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+[tool.hatch.build.targets.wheel]
+packages = ["scheduler"]
+
 [tool.black]
 line-length = 100
 target-version = ["py311"]

scheduler/control/__init__.py

@@ -0,0 +1,31 @@
+"""Control and intervention systems for court scheduling.
+
+Provides explainability and judge override capabilities.
+"""
+
+from .explainability import (
+    DecisionStep,
+    SchedulingExplanation,
+    ExplainabilityEngine
+)
+
+from .overrides import (
+    OverrideType,
+    Override,
+    JudgePreferences,
+    CauseListDraft,
+    OverrideValidator,
+    OverrideManager
+)
+
+__all__ = [
+    'DecisionStep',
+    'SchedulingExplanation',
+    'ExplainabilityEngine',
+    'OverrideType',
+    'Override',
+    'JudgePreferences',
+    'CauseListDraft',
+    'OverrideValidator',
+    'OverrideManager'
+]

scheduler/control/explainability.py

@@ -0,0 +1,316 @@
+"""Explainability system for scheduling decisions.
+
+Provides human-readable explanations for why each case was or wasn't scheduled.
+"""
+from dataclasses import dataclass
+from typing import Optional
+from datetime import date
+
+from scheduler.core.case import Case
+
+
+@dataclass
+class DecisionStep:
+    """Single step in decision reasoning."""
+    step_name: str
+    passed: bool
+    reason: str
+    details: dict
+
+
+@dataclass
+class SchedulingExplanation:
+    """Complete explanation of scheduling decision for a case."""
+    case_id: str
+    scheduled: bool
+    decision_steps: list[DecisionStep]
+    final_reason: str
+    priority_breakdown: Optional[dict] = None
+    courtroom_assignment_reason: Optional[str] = None
+    
+    def to_readable_text(self) -> str:
+        """Convert to human-readable explanation."""
+        lines = [f"Case {self.case_id}: {'SCHEDULED' if self.scheduled else 'NOT SCHEDULED'}"]
+        lines.append("=" * 60)
+        
+        for i, step in enumerate(self.decision_steps, 1):
+            status = "✓ PASS" if step.passed else "✗ FAIL"
+            lines.append(f"\nStep {i}: {step.step_name} - {status}")
+            lines.append(f"  Reason: {step.reason}")
+            if step.details:
+                for key, value in step.details.items():
+                    lines.append(f"    {key}: {value}")
+        
+        if self.priority_breakdown and self.scheduled:
+            lines.append(f"\nPriority Score Breakdown:")
+            for component, value in self.priority_breakdown.items():
+                lines.append(f"  {component}: {value}")
+        
+        if self.courtroom_assignment_reason and self.scheduled:
+            lines.append(f"\nCourtroom Assignment:")
+            lines.append(f"  {self.courtroom_assignment_reason}")
+        
+        lines.append(f"\nFinal Decision: {self.final_reason}")
+        
+        return "\n".join(lines)
+
+
+class ExplainabilityEngine:
+    """Generate explanations for scheduling decisions."""
+    
+    @staticmethod
+    def explain_scheduling_decision(
+        case: Case,
+        current_date: date,
+        scheduled: bool,
+        ripeness_status: str,
+        priority_score: Optional[float] = None,
+        courtroom_id: Optional[int] = None,
+        capacity_full: bool = False,
+        below_threshold: bool = False
+    ) -> SchedulingExplanation:
+        """Generate complete explanation for why case was/wasn't scheduled.
+        
+        Args:
+            case: The case being scheduled
+            current_date: Current simulation date
+            scheduled: Whether case was scheduled
+            ripeness_status: Ripeness classification
+            priority_score: Calculated priority score if scheduled
+            courtroom_id: Assigned courtroom if scheduled
+            capacity_full: Whether capacity was full
+            below_threshold: Whether priority was below threshold
+            
+        Returns:
+            Complete scheduling explanation
+        """
+        steps = []
+        
+        # Step 1: Disposal status check
+        if case.is_disposed:
+            steps.append(DecisionStep(
+                step_name="Case Status Check",
+                passed=False,
+                reason="Case already disposed",
+                details={"disposal_date": str(case.disposal_date)}
+            ))
+            return SchedulingExplanation(
+                case_id=case.case_id,
+                scheduled=False,
+                decision_steps=steps,
+                final_reason="Case disposed, no longer eligible for scheduling"
+            )
+        
+        steps.append(DecisionStep(
+            step_name="Case Status Check",
+            passed=True,
+            reason="Case active and eligible",
+            details={"status": case.status.value}
+        ))
+        
+        # Step 2: Ripeness check
+        is_ripe = ripeness_status == "RIPE"
+        ripeness_detail = {}
+        
+        if not is_ripe:
+            if "SUMMONS" in ripeness_status:
+                ripeness_detail["bottleneck"] = "Summons not yet served"
+                ripeness_detail["action_needed"] = "Wait for summons service confirmation"
+            elif "DEPENDENT" in ripeness_status:
+                ripeness_detail["bottleneck"] = "Dependent on another case"
+                ripeness_detail["action_needed"] = "Wait for dependent case resolution"
+            elif "PARTY" in ripeness_status:
+                ripeness_detail["bottleneck"] = "Party unavailable or unresponsive"
+                ripeness_detail["action_needed"] = "Wait for party availability confirmation"
+            else:
+                ripeness_detail["bottleneck"] = ripeness_status
+        else:
+            ripeness_detail["status"] = "All prerequisites met, ready for hearing"
+        
+        if case.last_hearing_purpose:
+            ripeness_detail["last_hearing_purpose"] = case.last_hearing_purpose
+        
+        steps.append(DecisionStep(
+            step_name="Ripeness Classification",
+            passed=is_ripe,
+            reason="Case is RIPE (ready for hearing)" if is_ripe else f"Case is UNRIPE ({ripeness_status})",
+            details=ripeness_detail
+        ))
+        
+        if not is_ripe and not scheduled:
+            return SchedulingExplanation(
+                case_id=case.case_id,
+                scheduled=False,
+                decision_steps=steps,
+                final_reason=f"Case not scheduled: UNRIPE status blocks scheduling. {ripeness_detail.get('action_needed', 'Waiting for case to become ready')}"
+            )
+        
+        # Step 3: Minimum gap check
+        min_gap_days = 7
+        days_since = case.days_since_last_hearing
+        meets_gap = case.last_hearing_date is None or days_since >= min_gap_days
+        
+        gap_details = {
+            "days_since_last_hearing": days_since,
+            "minimum_required": min_gap_days
+        }
+        
+        if case.last_hearing_date:
+            gap_details["last_hearing_date"] = str(case.last_hearing_date)
+        
+        steps.append(DecisionStep(
+            step_name="Minimum Gap Check",
+            passed=meets_gap,
+            reason=f"{'Meets' if meets_gap else 'Does not meet'} minimum {min_gap_days}-day gap requirement",
+            details=gap_details
+        ))
+        
+        if not meets_gap and not scheduled:
+            next_eligible = case.last_hearing_date.isoformat() if case.last_hearing_date else "unknown"
+            return SchedulingExplanation(
+                case_id=case.case_id,
+                scheduled=False,
+                decision_steps=steps,
+                final_reason=f"Case not scheduled: Only {days_since} days since last hearing (minimum {min_gap_days} required). Next eligible after {next_eligible}"
+            )
+        
+        # Step 4: Priority calculation
+        if priority_score is not None:
+            age_component = min(case.age_days / 2000, 1.0) * 0.35
+            readiness_component = case.readiness_score * 0.25
+            urgency_component = (1.0 if case.is_urgent else 0.0) * 0.25
+            
+            # Adjournment boost calculation
+            import math
+            adj_boost_value = 0.0
+            if case.status.value == "ADJOURNED" and case.hearing_count > 0:
+                adj_boost_value = math.exp(-case.days_since_last_hearing / 21)
+            adj_boost_component = adj_boost_value * 0.15
+            
+            priority_breakdown = {
+                "Age": f"{age_component:.4f} (age={case.age_days}d, weight=0.35)",
+                "Readiness": f"{readiness_component:.4f} (score={case.readiness_score:.2f}, weight=0.25)",
+                "Urgency": f"{urgency_component:.4f} ({'URGENT' if case.is_urgent else 'normal'}, weight=0.25)",
+                "Adjournment Boost": f"{adj_boost_component:.4f} (days_since={days_since}, decay=exp(-{days_since}/21), weight=0.15)",
+                "TOTAL": f"{priority_score:.4f}"
+            }
+            
+            steps.append(DecisionStep(
+                step_name="Priority Calculation",
+                passed=True,
+                reason=f"Priority score calculated: {priority_score:.4f}",
+                details=priority_breakdown
+            ))
+        
+        # Step 5: Selection by policy
+        if scheduled:
+            if capacity_full:
+                steps.append(DecisionStep(
+                    step_name="Capacity Check",
+                    passed=True,
+                    reason="Selected despite full capacity (high priority override)",
+                    details={"priority_score": f"{priority_score:.4f}"}
+                ))
+            elif below_threshold:
+                steps.append(DecisionStep(
+                    step_name="Policy Selection",
+                    passed=True,
+                    reason="Selected by policy despite being below typical threshold",
+                    details={"reason": "Algorithm determined case should be scheduled"}
+                ))
+            else:
+                steps.append(DecisionStep(
+                    step_name="Policy Selection",
+                    passed=True,
+                    reason="Selected by scheduling policy among eligible cases",
+                    details={
+                        "priority_rank": "Top priority among eligible cases",
+                        "policy": "Readiness + Adjournment Boost"
+                    }
+                ))
+            
+            # Courtroom assignment
+            if courtroom_id:
+                courtroom_reason = f"Assigned to Courtroom {courtroom_id} via load balancing (least loaded courtroom selected)"
+                steps.append(DecisionStep(
+                    step_name="Courtroom Assignment",
+                    passed=True,
+                    reason=courtroom_reason,
+                    details={"courtroom_id": courtroom_id}
+                ))
+            
+            final_reason = f"Case SCHEDULED: Passed all checks, priority score {priority_score:.4f}, assigned to Courtroom {courtroom_id}"
+            
+            return SchedulingExplanation(
+                case_id=case.case_id,
+                scheduled=True,
+                decision_steps=steps,
+                final_reason=final_reason,
+                priority_breakdown=priority_breakdown if priority_score else None,
+                courtroom_assignment_reason=courtroom_reason if courtroom_id else None
+            )
+        else:
+            # Not scheduled - determine why
+            if capacity_full:
+                steps.append(DecisionStep(
+                    step_name="Capacity Check",
+                    passed=False,
+                    reason="Daily capacity limit reached",
+                    details={
+                        "priority_score": f"{priority_score:.4f}" if priority_score else "N/A",
+                        "explanation": "Higher priority cases filled all available slots"
+                    }
+                ))
+                final_reason = f"Case NOT SCHEDULED: Capacity full. Priority score {priority_score:.4f} was not high enough to displace scheduled cases"
+            elif below_threshold:
+                steps.append(DecisionStep(
+                    step_name="Policy Selection",
+                    passed=False,
+                    reason="Priority below scheduling threshold",
+                    details={
+                        "priority_score": f"{priority_score:.4f}" if priority_score else "N/A",
+                        "explanation": "Other cases had higher priority scores"
+                    }
+                ))
+                final_reason = f"Case NOT SCHEDULED: Priority score {priority_score:.4f} below threshold. Wait for case to age or become more urgent"
+            else:
+                final_reason = "Case NOT SCHEDULED: Unknown reason (policy decision)"
+            
+            return SchedulingExplanation(
+                case_id=case.case_id,
+                scheduled=False,
+                decision_steps=steps,
+                final_reason=final_reason,
+                priority_breakdown=priority_breakdown if priority_score else None
+            )
+    
+    @staticmethod
+    def explain_why_not_scheduled(case: Case, current_date: date) -> str:
+        """Quick explanation for why a case wasn't scheduled.
+        
+        Args:
+            case: Case to explain
+            current_date: Current date
+            
+        Returns:
+            Human-readable reason
+        """
+        if case.is_disposed:
+            return f"Already disposed on {case.disposal_date}"
+        
+        if case.ripeness_status != "RIPE":
+            bottleneck_reasons = {
+                "UNRIPE_SUMMONS": "Summons not served",
+                "UNRIPE_DEPENDENT": "Waiting for dependent case",
+                "UNRIPE_PARTY": "Party unavailable",
+                "UNRIPE_DOCUMENT": "Documents pending"
+            }
+            reason = bottleneck_reasons.get(case.ripeness_status, case.ripeness_status)
+            return f"UNRIPE: {reason}"
+        
+        if case.last_hearing_date and case.days_since_last_hearing < 7:
+            return f"Too recent (last hearing {case.days_since_last_hearing} days ago, minimum 7 days)"
+        
+        # If ripe and meets gap, then it's priority-based
+        priority = case.get_priority_score()
+        return f"Low priority (score {priority:.3f}) - other cases ranked higher"

scheduler/control/overrides.py

@@ -0,0 +1,438 @@
+"""Judge override and intervention control system.
+
+Allows judges to review, modify, and approve algorithmic scheduling suggestions.
+System is suggestive, not prescriptive - judges retain final control.
+"""
+from dataclasses import dataclass, field
+from datetime import date, datetime
+from enum import Enum
+from typing import Optional
+import json
+
+
+class OverrideType(Enum):
+    """Types of overrides judges can make."""
+    RIPENESS = "ripeness"  # Override ripeness classification
+    PRIORITY = "priority"  # Adjust priority score or urgency
+    ADD_CASE = "add_case"  # Manually add case to cause list
+    REMOVE_CASE = "remove_case"  # Remove case from cause list
+    REORDER = "reorder"  # Change sequence within day
+    CAPACITY = "capacity"  # Adjust daily capacity
+    MIN_GAP = "min_gap"  # Override minimum gap between hearings
+    COURTROOM = "courtroom"  # Change courtroom assignment
+
+
+@dataclass
+class Override:
+    """Single override action by a judge."""
+    override_id: str
+    override_type: OverrideType
+    case_id: str
+    judge_id: str
+    timestamp: datetime
+    old_value: Optional[str] = None
+    new_value: Optional[str] = None
+    reason: str = ""
+    date_affected: Optional[date] = None
+    courtroom_id: Optional[int] = None
+    
+    def to_dict(self) -> dict:
+        """Convert to dictionary for logging."""
+        return {
+            "override_id": self.override_id,
+            "type": self.override_type.value,
+            "case_id": self.case_id,
+            "judge_id": self.judge_id,
+            "timestamp": self.timestamp.isoformat(),
+            "old_value": self.old_value,
+            "new_value": self.new_value,
+            "reason": self.reason,
+            "date_affected": self.date_affected.isoformat() if self.date_affected else None,
+            "courtroom_id": self.courtroom_id
+        }
+    
+    def to_readable_text(self) -> str:
+        """Human-readable description of override."""
+        action_desc = {
+            OverrideType.RIPENESS: f"Changed ripeness from {self.old_value} to {self.new_value}",
+            OverrideType.PRIORITY: f"Adjusted priority from {self.old_value} to {self.new_value}",
+            OverrideType.ADD_CASE: f"Manually added case to cause list",
+            OverrideType.REMOVE_CASE: f"Removed case from cause list",
+            OverrideType.REORDER: f"Reordered from position {self.old_value} to {self.new_value}",
+            OverrideType.CAPACITY: f"Changed capacity from {self.old_value} to {self.new_value}",
+            OverrideType.MIN_GAP: f"Overrode min gap from {self.old_value} to {self.new_value} days",
+            OverrideType.COURTROOM: f"Changed courtroom from {self.old_value} to {self.new_value}"
+        }
+        
+        action = action_desc.get(self.override_type, f"Override: {self.override_type.value}")
+        
+        parts = [
+            f"[{self.timestamp.strftime('%Y-%m-%d %H:%M')}]",
+            f"Judge {self.judge_id}:",
+            action,
+            f"(Case {self.case_id})"
+        ]
+        
+        if self.reason:
+            parts.append(f"Reason: {self.reason}")
+        
+        return " ".join(parts)
+
+
+@dataclass
+class JudgePreferences:
+    """Judge-specific scheduling preferences."""
+    judge_id: str
+    daily_capacity_override: Optional[int] = None  # Override default capacity
+    blocked_dates: list[date] = field(default_factory=list)  # Vacation, illness
+    min_gap_overrides: dict[str, int] = field(default_factory=dict)  # Per-case gap overrides
+    case_type_preferences: dict[str, list[str]] = field(default_factory=dict)  # Day-of-week preferences
+    
+    def to_dict(self) -> dict:
+        """Convert to dictionary."""
+        return {
+            "judge_id": self.judge_id,
+            "daily_capacity_override": self.daily_capacity_override,
+            "blocked_dates": [d.isoformat() for d in self.blocked_dates],
+            "min_gap_overrides": self.min_gap_overrides,
+            "case_type_preferences": self.case_type_preferences
+        }
+
+
+@dataclass
+class CauseListDraft:
+    """Draft cause list before judge approval."""
+    date: date
+    courtroom_id: int
+    judge_id: str
+    algorithm_suggested: list[str]  # Case IDs suggested by algorithm
+    judge_approved: list[str]  # Case IDs after judge review
+    overrides: list[Override]
+    created_at: datetime
+    finalized_at: Optional[datetime] = None
+    status: str = "DRAFT"  # DRAFT, APPROVED, REJECTED
+    
+    def get_acceptance_rate(self) -> float:
+        """Calculate what % of suggestions were accepted."""
+        if not self.algorithm_suggested:
+            return 0.0
+        
+        accepted = len(set(self.algorithm_suggested) & set(self.judge_approved))
+        return accepted / len(self.algorithm_suggested) * 100
+    
+    def get_modifications_summary(self) -> dict:
+        """Summarize modifications made."""
+        added = set(self.judge_approved) - set(self.algorithm_suggested)
+        removed = set(self.algorithm_suggested) - set(self.judge_approved)
+        
+        override_counts = {}
+        for override in self.overrides:
+            override_type = override.override_type.value
+            override_counts[override_type] = override_counts.get(override_type, 0) + 1
+        
+        return {
+            "cases_added": len(added),
+            "cases_removed": len(removed),
+            "cases_kept": len(set(self.algorithm_suggested) & set(self.judge_approved)),
+            "override_types": override_counts,
+            "acceptance_rate": self.get_acceptance_rate()
+        }
+
+
+class OverrideValidator:
+    """Validates override requests against constraints."""
+    
+    @staticmethod
+    def validate_ripeness_override(
+        case_id: str,
+        old_status: str,
+        new_status: str,
+        reason: str
+    ) -> tuple[bool, str]:
+        """Validate ripeness override.
+        
+        Args:
+            case_id: Case ID
+            old_status: Current ripeness status
+            new_status: Requested new status
+            reason: Reason for override
+            
+        Returns:
+            (valid, error_message)
+        """
+        valid_statuses = ["RIPE", "UNRIPE_SUMMONS", "UNRIPE_DEPENDENT", "UNRIPE_PARTY", "UNRIPE_DOCUMENT"]
+        
+        if new_status not in valid_statuses:
+            return False, f"Invalid ripeness status: {new_status}"
+        
+        if not reason:
+            return False, "Reason required for ripeness override"
+        
+        if len(reason) < 10:
+            return False, "Reason must be at least 10 characters"
+        
+        return True, ""
+    
+    @staticmethod
+    def validate_capacity_override(
+        current_capacity: int,
+        new_capacity: int,
+        max_capacity: int = 200
+    ) -> tuple[bool, str]:
+        """Validate capacity override.
+        
+        Args:
+            current_capacity: Current daily capacity
+            new_capacity: Requested new capacity
+            max_capacity: Maximum allowed capacity
+            
+        Returns:
+            (valid, error_message)
+        """
+        if new_capacity < 0:
+            return False, "Capacity cannot be negative"
+        
+        if new_capacity > max_capacity:
+            return False, f"Capacity cannot exceed maximum ({max_capacity})"
+        
+        if new_capacity == 0:
+            return False, "Capacity cannot be zero (use blocked dates for full closures)"
+        
+        return True, ""
+    
+    @staticmethod
+    def validate_add_case(
+        case_id: str,
+        current_schedule: list[str],
+        current_capacity: int,
+        max_capacity: int
+    ) -> tuple[bool, str]:
+        """Validate adding a case to cause list.
+        
+        Args:
+            case_id: Case to add
+            current_schedule: Currently scheduled case IDs
+            current_capacity: Current number of scheduled cases
+            max_capacity: Maximum capacity
+            
+        Returns:
+            (valid, error_message)
+        """
+        if case_id in current_schedule:
+            return False, f"Case {case_id} already in schedule"
+        
+        if current_capacity >= max_capacity:
+            return False, f"Schedule at capacity ({current_capacity}/{max_capacity})"
+        
+        return True, ""
+    
+    @staticmethod
+    def validate_remove_case(
+        case_id: str,
+        current_schedule: list[str]
+    ) -> tuple[bool, str]:
+        """Validate removing a case from cause list.
+        
+        Args:
+            case_id: Case to remove
+            current_schedule: Currently scheduled case IDs
+            
+        Returns:
+            (valid, error_message)
+        """
+        if case_id not in current_schedule:
+            return False, f"Case {case_id} not in schedule"
+        
+        return True, ""
+
+
+class OverrideManager:
+    """Manages judge overrides and interventions."""
+    
+    def __init__(self):
+        self.overrides: list[Override] = []
+        self.drafts: list[CauseListDraft] = []
+        self.preferences: dict[str, JudgePreferences] = {}
+    
+    def create_draft(
+        self,
+        date: date,
+        courtroom_id: int,
+        judge_id: str,
+        algorithm_suggested: list[str]
+    ) -> CauseListDraft:
+        """Create a draft cause list for judge review.
+        
+        Args:
+            date: Date of cause list
+            courtroom_id: Courtroom ID
+            judge_id: Judge ID
+            algorithm_suggested: Case IDs suggested by algorithm
+            
+        Returns:
+            Draft cause list
+        """
+        draft = CauseListDraft(
+            date=date,
+            courtroom_id=courtroom_id,
+            judge_id=judge_id,
+            algorithm_suggested=algorithm_suggested.copy(),
+            judge_approved=[],
+            overrides=[],
+            created_at=datetime.now(),
+            status="DRAFT"
+        )
+        
+        self.drafts.append(draft)
+        return draft
+    
+    def apply_override(
+        self,
+        draft: CauseListDraft,
+        override: Override
+    ) -> tuple[bool, str]:
+        """Apply an override to a draft cause list.
+        
+        Args:
+            draft: Draft to modify
+            override: Override to apply
+            
+        Returns:
+            (success, error_message)
+        """
+        # Validate based on type
+        if override.override_type == OverrideType.RIPENESS:
+            valid, error = OverrideValidator.validate_ripeness_override(
+                override.case_id,
+                override.old_value or "",
+                override.new_value or "",
+                override.reason
+            )
+            if not valid:
+                return False, error
+        
+        elif override.override_type == OverrideType.ADD_CASE:
+            valid, error = OverrideValidator.validate_add_case(
+                override.case_id,
+                draft.judge_approved,
+                len(draft.judge_approved),
+                200  # Max capacity
+            )
+            if not valid:
+                return False, error
+            
+            draft.judge_approved.append(override.case_id)
+        
+        elif override.override_type == OverrideType.REMOVE_CASE:
+            valid, error = OverrideValidator.validate_remove_case(
+                override.case_id,
+                draft.judge_approved
+            )
+            if not valid:
+                return False, error
+            
+            draft.judge_approved.remove(override.case_id)
+        
+        # Record override
+        draft.overrides.append(override)
+        self.overrides.append(override)
+        
+        return True, ""
+    
+    def finalize_draft(self, draft: CauseListDraft) -> bool:
+        """Finalize draft cause list (judge approval).
+        
+        Args:
+            draft: Draft to finalize
+            
+        Returns:
+            Success status
+        """
+        if draft.status != "DRAFT":
+            return False
+        
+        draft.status = "APPROVED"
+        draft.finalized_at = datetime.now()
+        
+        return True
+    
+    def get_judge_preferences(self, judge_id: str) -> JudgePreferences:
+        """Get or create judge preferences.
+        
+        Args:
+            judge_id: Judge ID
+            
+        Returns:
+            Judge preferences
+        """
+        if judge_id not in self.preferences:
+            self.preferences[judge_id] = JudgePreferences(judge_id=judge_id)
+        
+        return self.preferences[judge_id]
+    
+    def get_override_statistics(self, judge_id: Optional[str] = None) -> dict:
+        """Get override statistics.
+        
+        Args:
+            judge_id: Optional filter by judge
+            
+        Returns:
+            Statistics dictionary
+        """
+        relevant_overrides = self.overrides
+        if judge_id:
+            relevant_overrides = [o for o in self.overrides if o.judge_id == judge_id]
+        
+        if not relevant_overrides:
+            return {
+                "total_overrides": 0,
+                "by_type": {},
+                "avg_per_day": 0
+            }
+        
+        override_counts = {}
+        for override in relevant_overrides:
+            override_type = override.override_type.value
+            override_counts[override_type] = override_counts.get(override_type, 0) + 1
+        
+        # Calculate acceptance rate from drafts
+        relevant_drafts = self.drafts
+        if judge_id:
+            relevant_drafts = [d for d in self.drafts if d.judge_id == judge_id]
+        
+        acceptance_rates = [d.get_acceptance_rate() for d in relevant_drafts if d.status == "APPROVED"]
+        avg_acceptance = sum(acceptance_rates) / len(acceptance_rates) if acceptance_rates else 0
+        
+        return {
+            "total_overrides": len(relevant_overrides),
+            "by_type": override_counts,
+            "total_drafts": len(relevant_drafts),
+            "approved_drafts": len([d for d in relevant_drafts if d.status == "APPROVED"]),
+            "avg_acceptance_rate": avg_acceptance,
+            "modification_rate": 100 - avg_acceptance if avg_acceptance else 0
+        }
+    
+    def export_audit_trail(self, output_file: str):
+        """Export complete audit trail to file.
+        
+        Args:
+            output_file: Path to output file
+        """
+        audit_data = {
+            "overrides": [o.to_dict() for o in self.overrides],
+            "drafts": [
+                {
+                    "date": d.date.isoformat(),
+                    "courtroom_id": d.courtroom_id,
+                    "judge_id": d.judge_id,
+                    "status": d.status,
+                    "acceptance_rate": d.get_acceptance_rate(),
+                    "modifications": d.get_modifications_summary()
+                }
+                for d in self.drafts
+            ],
+            "statistics": self.get_override_statistics()
+        }
+        
+        with open(output_file, 'w') as f:
+            json.dump(audit_data, f, indent=2)

scheduler/core/case.py

@@ -0,0 +1,331 @@
+"""Case entity and lifecycle management.
+
+This module defines the Case class which represents a single court case
+progressing through various stages.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import date, datetime
+from typing import List, Optional, TYPE_CHECKING
+from enum import Enum
+
+from scheduler.data.config import TERMINAL_STAGES
+
+if TYPE_CHECKING:
+    from scheduler.core.ripeness import RipenessStatus
+else:
+    # Import at runtime
+    RipenessStatus = None
+
+
+class CaseStatus(Enum):
+    """Status of a case in the system."""
+    PENDING = "pending"          # Filed, awaiting first hearing
+    ACTIVE = "active"            # Has had at least one hearing
+    ADJOURNED = "adjourned"      # Last hearing was adjourned
+    DISPOSED = "disposed"        # Final disposal/settlement reached
+    
+
+@dataclass
+class Case:
+    """Represents a single court case.
+    
+    Attributes:
+        case_id: Unique identifier (like CNR number)
+        case_type: Type of case (RSA, CRP, RFA, CA, CCC, CP, CMP)
+        filed_date: Date when case was filed
+        current_stage: Current stage in lifecycle
+        status: Current status (PENDING, ACTIVE, ADJOURNED, DISPOSED)
+        courtroom_id: Assigned courtroom (0-4 for 5 courtrooms)
+        is_urgent: Whether case is marked urgent
+        readiness_score: Computed readiness score (0-1)
+        hearing_count: Number of hearings held
+        last_hearing_date: Date of most recent hearing
+        days_since_last_hearing: Days elapsed since last hearing
+        age_days: Days since filing
+        disposal_date: Date of disposal (if disposed)
+        history: List of hearing dates and outcomes
+    """
+    case_id: str
+    case_type: str
+    filed_date: date
+    current_stage: str = "ADMISSION"  # Default initial stage
+    status: CaseStatus = CaseStatus.PENDING
+    courtroom_id: int | None = None  # None = not yet assigned; 0 is invalid
+    is_urgent: bool = False
+    readiness_score: float = 0.0
+    hearing_count: int = 0
+    last_hearing_date: Optional[date] = None
+    days_since_last_hearing: int = 0
+    age_days: int = 0
+    disposal_date: Optional[date] = None
+    stage_start_date: Optional[date] = None
+    days_in_stage: int = 0
+    history: List[dict] = field(default_factory=list)
+    
+    # Ripeness tracking (NEW - for bottleneck detection)
+    ripeness_status: str = "UNKNOWN"  # RipenessStatus enum value (stored as string to avoid circular import)
+    bottleneck_reason: Optional[str] = None
+    ripeness_updated_at: Optional[datetime] = None
+    last_hearing_purpose: Optional[str] = None  # Purpose of last hearing (for classification)
+    
+    # No-case-left-behind tracking (NEW)
+    last_scheduled_date: Optional[date] = None
+    days_since_last_scheduled: int = 0
+    
+    def progress_to_stage(self, new_stage: str, current_date: date) -> None:
+        """Progress case to a new stage.
+        
+        Args:
+            new_stage: The stage to progress to
+            current_date: Current simulation date
+        """
+        self.current_stage = new_stage
+        self.stage_start_date = current_date
+        self.days_in_stage = 0
+        
+        # Check if terminal stage (case disposed)
+        if new_stage in TERMINAL_STAGES:
+            self.status = CaseStatus.DISPOSED
+            self.disposal_date = current_date
+            
+        # Record in history
+        self.history.append({
+            "date": current_date,
+            "event": "stage_change",
+            "stage": new_stage,
+        })
+    
+    def record_hearing(self, hearing_date: date, was_heard: bool, outcome: str = "") -> None:
+        """Record a hearing event.
+        
+        Args:
+            hearing_date: Date of the hearing
+            was_heard: Whether the hearing actually proceeded (not adjourned)
+            outcome: Outcome description
+        """
+        self.hearing_count += 1
+        self.last_hearing_date = hearing_date
+        
+        if was_heard:
+            self.status = CaseStatus.ACTIVE
+        else:
+            self.status = CaseStatus.ADJOURNED
+        
+        # Record in history
+        self.history.append({
+            "date": hearing_date,
+            "event": "hearing",
+            "was_heard": was_heard,
+            "outcome": outcome,
+            "stage": self.current_stage,
+        })
+    
+    def update_age(self, current_date: date) -> None:
+        """Update age and days since last hearing.
+        
+        Args:
+            current_date: Current simulation date
+        """
+        self.age_days = (current_date - self.filed_date).days
+        
+        if self.last_hearing_date:
+            self.days_since_last_hearing = (current_date - self.last_hearing_date).days
+        else:
+            self.days_since_last_hearing = self.age_days
+        
+        if self.stage_start_date:
+            self.days_in_stage = (current_date - self.stage_start_date).days
+        else:
+            self.days_in_stage = self.age_days
+        
+        # Update days since last scheduled (for no-case-left-behind tracking)
+        if self.last_scheduled_date:
+            self.days_since_last_scheduled = (current_date - self.last_scheduled_date).days
+        else:
+            self.days_since_last_scheduled = self.age_days
+    
+    def compute_readiness_score(self) -> float:
+        """Compute readiness score based on hearings, gaps, and stage.
+        
+        Formula (from EDA):
+            READINESS = (hearings_capped/50) * 0.4 +
+                       (100/gap_clamped) * 0.3 +
+                       (stage_advanced) * 0.3
+        
+        Returns:
+            Readiness score (0-1, higher = more ready)
+        """
+        # Cap hearings at 50
+        hearings_capped = min(self.hearing_count, 50)
+        hearings_component = (hearings_capped / 50) * 0.4
+        
+        # Gap component (inverse of days since last hearing)
+        gap_clamped = min(max(self.days_since_last_hearing, 1), 100)
+        gap_component = (100 / gap_clamped) * 0.3
+        
+        # Stage component (advanced stages get higher score)
+        advanced_stages = ["ARGUMENTS", "EVIDENCE", "ORDERS / JUDGMENT"]
+        stage_component = 0.3 if self.current_stage in advanced_stages else 0.1
+        
+        readiness = hearings_component + gap_component + stage_component
+        self.readiness_score = min(1.0, max(0.0, readiness))
+        
+        return self.readiness_score
+    
+    def is_ready_for_scheduling(self, min_gap_days: int = 7) -> bool:
+        """Check if case is ready to be scheduled.
+        
+        Args:
+            min_gap_days: Minimum days required since last hearing
+            
+        Returns:
+            True if case can be scheduled
+        """
+        if self.status == CaseStatus.DISPOSED:
+            return False
+        
+        if self.last_hearing_date is None:
+            return True  # First hearing, always ready
+        
+        return self.days_since_last_hearing >= min_gap_days
+    
+    def needs_alert(self, max_gap_days: int = 90) -> bool:
+        """Check if case needs alert due to long gap.
+        
+        Args:
+            max_gap_days: Maximum allowed gap before alert
+            
+        Returns:
+            True if alert should be triggered
+        """
+        if self.status == CaseStatus.DISPOSED:
+            return False
+        
+        return self.days_since_last_hearing > max_gap_days
+    
+    def get_priority_score(self) -> float:
+        """Get overall priority score for scheduling.
+        
+        Combines age, readiness, urgency, and adjournment boost into single score.
+        
+        Formula:
+            priority = age*0.35 + readiness*0.25 + urgency*0.25 + adjournment_boost*0.15
+        
+        Adjournment boost: Recently adjourned cases get priority to avoid indefinite postponement.
+        The boost decays exponentially: strongest immediately after adjournment, weaker over time.
+        
+        Returns:
+            Priority score (higher = higher priority)
+        """
+        # Age component (normalize to 0-1, assuming max age ~2000 days)
+        age_component = min(self.age_days / 2000, 1.0) * 0.35
+        
+        # Readiness component
+        readiness_component = self.readiness_score * 0.25
+        
+        # Urgency component
+        urgency_component = 1.0 if self.is_urgent else 0.0
+        urgency_component *= 0.25
+        
+        # Adjournment boost (NEW - prevents cases from being repeatedly postponed)
+        adjournment_boost = 0.0
+        if self.status == CaseStatus.ADJOURNED and self.hearing_count > 0:
+            # Boost starts at 1.0 immediately after adjournment, decays exponentially
+            # Formula: boost = exp(-days_since_hearing / 21)
+            # At 7 days: ~0.71 (strong boost)
+            # At 14 days: ~0.50 (moderate boost)
+            # At 21 days: ~0.37 (weak boost)
+            # At 28 days: ~0.26 (very weak boost)
+            import math
+            decay_factor = 21  # Half-life of boost
+            adjournment_boost = math.exp(-self.days_since_last_hearing / decay_factor)
+        adjournment_boost *= 0.15
+        
+        return age_component + readiness_component + urgency_component + adjournment_boost
+    
+    def mark_unripe(self, status, reason: str, current_date: datetime) -> None:
+        """Mark case as unripe with bottleneck reason.
+        
+        Args:
+            status: Ripeness status (UNRIPE_SUMMONS, UNRIPE_PARTY, etc.) - RipenessStatus enum
+            reason: Human-readable reason for unripeness
+            current_date: Current simulation date
+        """
+        # Store as string to avoid circular import
+        self.ripeness_status = status.value if hasattr(status, 'value') else str(status)
+        self.bottleneck_reason = reason
+        self.ripeness_updated_at = current_date
+        
+        # Record in history
+        self.history.append({
+            "date": current_date,
+            "event": "ripeness_change",
+            "status": self.ripeness_status,
+            "reason": reason,
+        })
+    
+    def mark_ripe(self, current_date: datetime) -> None:
+        """Mark case as ripe (ready for hearing).
+        
+        Args:
+            current_date: Current simulation date
+        """
+        self.ripeness_status = "RIPE"
+        self.bottleneck_reason = None
+        self.ripeness_updated_at = current_date
+        
+        # Record in history
+        self.history.append({
+            "date": current_date,
+            "event": "ripeness_change",
+            "status": "RIPE",
+            "reason": "Case became ripe",
+        })
+    
+    def mark_scheduled(self, scheduled_date: date) -> None:
+        """Mark case as scheduled for a hearing.
+        
+        Used for no-case-left-behind tracking.
+        
+        Args:
+            scheduled_date: Date case was scheduled
+        """
+        self.last_scheduled_date = scheduled_date
+        self.days_since_last_scheduled = 0
+    
+    @property
+    def is_disposed(self) -> bool:
+        """Check if case is disposed."""
+        return self.status == CaseStatus.DISPOSED
+    
+    def __repr__(self) -> str:
+        return (f"Case(id={self.case_id}, type={self.case_type}, "
+                f"stage={self.current_stage}, status={self.status.value}, "
+                f"hearings={self.hearing_count})")
+    
+    def to_dict(self) -> dict:
+        """Convert case to dictionary for serialization."""
+        return {
+            "case_id": self.case_id,
+            "case_type": self.case_type,
+            "filed_date": self.filed_date.isoformat(),
+            "current_stage": self.current_stage,
+            "status": self.status.value,
+            "courtroom_id": self.courtroom_id,
+            "is_urgent": self.is_urgent,
+            "readiness_score": self.readiness_score,
+            "hearing_count": self.hearing_count,
+            "last_hearing_date": self.last_hearing_date.isoformat() if self.last_hearing_date else None,
+            "days_since_last_hearing": self.days_since_last_hearing,
+            "age_days": self.age_days,
+            "disposal_date": self.disposal_date.isoformat() if self.disposal_date else None,
+            "ripeness_status": self.ripeness_status,
+            "bottleneck_reason": self.bottleneck_reason,
+            "last_hearing_purpose": self.last_hearing_purpose,
+            "last_scheduled_date": self.last_scheduled_date.isoformat() if self.last_scheduled_date else None,
+            "days_since_last_scheduled": self.days_since_last_scheduled,
+            "history": self.history,
+        }

scheduler/core/courtroom.py

@@ -0,0 +1,228 @@
+"""Courtroom resource management.
+
+This module defines the Courtroom class which represents a physical courtroom
+with capacity constraints and daily scheduling.
+"""
+
+from dataclasses import dataclass, field
+from datetime import date
+from typing import Dict, List, Optional, Set
+
+from scheduler.data.config import DEFAULT_DAILY_CAPACITY
+
+
+@dataclass
+class Courtroom:
+    """Represents a courtroom resource.
+    
+    Attributes:
+        courtroom_id: Unique identifier (0-4 for 5 courtrooms)
+        judge_id: Currently assigned judge (optional)
+        daily_capacity: Maximum cases that can be heard per day
+        case_types: Types of cases handled by this courtroom
+        schedule: Dict mapping dates to lists of case_ids scheduled
+        hearings_held: Count of hearings held
+        utilization_history: Track daily utilization rates
+    """
+    courtroom_id: int
+    judge_id: Optional[str] = None
+    daily_capacity: int = DEFAULT_DAILY_CAPACITY
+    case_types: Set[str] = field(default_factory=set)
+    schedule: Dict[date, List[str]] = field(default_factory=dict)
+    hearings_held: int = 0
+    utilization_history: List[Dict] = field(default_factory=list)
+    
+    def assign_judge(self, judge_id: str) -> None:
+        """Assign a judge to this courtroom.
+        
+        Args:
+            judge_id: Judge identifier
+        """
+        self.judge_id = judge_id
+    
+    def add_case_types(self, *case_types: str) -> None:
+        """Add case types that this courtroom handles.
+        
+        Args:
+            *case_types: One or more case type strings (e.g., 'RSA', 'CRP')
+        """
+        self.case_types.update(case_types)
+    
+    def can_schedule(self, hearing_date: date, case_id: str) -> bool:
+        """Check if a case can be scheduled on a given date.
+        
+        Args:
+            hearing_date: Date to check
+            case_id: Case identifier
+            
+        Returns:
+            True if slot available, False if at capacity
+        """
+        if hearing_date not in self.schedule:
+            return True  # No hearings scheduled yet
+        
+        # Check if already scheduled
+        if case_id in self.schedule[hearing_date]:
+            return False  # Already scheduled
+        
+        # Check capacity
+        return len(self.schedule[hearing_date]) < self.daily_capacity
+    
+    def schedule_case(self, hearing_date: date, case_id: str) -> bool:
+        """Schedule a case for a hearing.
+        
+        Args:
+            hearing_date: Date of hearing
+            case_id: Case identifier
+            
+        Returns:
+            True if successfully scheduled, False if at capacity
+        """
+        if not self.can_schedule(hearing_date, case_id):
+            return False
+        
+        if hearing_date not in self.schedule:
+            self.schedule[hearing_date] = []
+        
+        self.schedule[hearing_date].append(case_id)
+        return True
+    
+    def unschedule_case(self, hearing_date: date, case_id: str) -> bool:
+        """Remove a case from schedule (e.g., if adjourned).
+        
+        Args:
+            hearing_date: Date of hearing
+            case_id: Case identifier
+            
+        Returns:
+            True if successfully removed, False if not found
+        """
+        if hearing_date not in self.schedule:
+            return False
+        
+        if case_id in self.schedule[hearing_date]:
+            self.schedule[hearing_date].remove(case_id)
+            return True
+        
+        return False
+    
+    def get_daily_schedule(self, hearing_date: date) -> List[str]:
+        """Get list of cases scheduled for a specific date.
+        
+        Args:
+            hearing_date: Date to query
+            
+        Returns:
+            List of case_ids scheduled (empty if none)
+        """
+        return self.schedule.get(hearing_date, [])
+    
+    def get_capacity_for_date(self, hearing_date: date) -> int:
+        """Get remaining capacity for a specific date.
+        
+        Args:
+            hearing_date: Date to query
+            
+        Returns:
+            Number of available slots
+        """
+        scheduled_count = len(self.get_daily_schedule(hearing_date))
+        return self.daily_capacity - scheduled_count
+    
+    def record_hearing_completed(self, hearing_date: date) -> None:
+        """Record that a hearing was held.
+        
+        Args:
+            hearing_date: Date of hearing
+        """
+        self.hearings_held += 1
+    
+    def compute_utilization(self, hearing_date: date) -> float:
+        """Compute utilization rate for a specific date.
+        
+        Args:
+            hearing_date: Date to compute for
+            
+        Returns:
+            Utilization rate (0.0 to 1.0)
+        """
+        scheduled_count = len(self.get_daily_schedule(hearing_date))
+        return scheduled_count / self.daily_capacity if self.daily_capacity > 0 else 0.0
+    
+    def record_daily_utilization(self, hearing_date: date, actual_hearings: int) -> None:
+        """Record actual utilization for a day.
+        
+        Args:
+            hearing_date: Date of hearings
+            actual_hearings: Number of hearings actually held (not adjourned)
+        """
+        scheduled = len(self.get_daily_schedule(hearing_date))
+        utilization = actual_hearings / self.daily_capacity if self.daily_capacity > 0 else 0.0
+        
+        self.utilization_history.append({
+            "date": hearing_date,
+            "scheduled": scheduled,
+            "actual": actual_hearings,
+            "capacity": self.daily_capacity,
+            "utilization": utilization,
+        })
+    
+    def get_average_utilization(self) -> float:
+        """Calculate average utilization rate across all recorded days.
+        
+        Returns:
+            Average utilization (0.0 to 1.0)
+        """
+        if not self.utilization_history:
+            return 0.0
+        
+        total = sum(day["utilization"] for day in self.utilization_history)
+        return total / len(self.utilization_history)
+    
+    def get_schedule_summary(self, start_date: date, end_date: date) -> Dict:
+        """Get summary statistics for a date range.
+        
+        Args:
+            start_date: Start of range
+            end_date: End of range
+            
+        Returns:
+            Dict with counts and utilization stats
+        """
+        days_in_range = [d for d in self.schedule.keys() 
+                        if start_date <= d <= end_date]
+        
+        total_scheduled = sum(len(self.schedule[d]) for d in days_in_range)
+        days_with_hearings = len(days_in_range)
+        
+        return {
+            "courtroom_id": self.courtroom_id,
+            "days_with_hearings": days_with_hearings,
+            "total_cases_scheduled": total_scheduled,
+            "avg_cases_per_day": total_scheduled / days_with_hearings if days_with_hearings > 0 else 0,
+            "total_capacity": days_with_hearings * self.daily_capacity,
+            "utilization_rate": total_scheduled / (days_with_hearings * self.daily_capacity) 
+                              if days_with_hearings > 0 else 0,
+        }
+    
+    def clear_schedule(self) -> None:
+        """Clear all scheduled hearings (for testing/reset)."""
+        self.schedule.clear()
+        self.utilization_history.clear()
+        self.hearings_held = 0
+    
+    def __repr__(self) -> str:
+        return (f"Courtroom(id={self.courtroom_id}, judge={self.judge_id}, "
+                f"capacity={self.daily_capacity}, types={self.case_types})")
+    
+    def to_dict(self) -> dict:
+        """Convert courtroom to dictionary for serialization."""
+        return {
+            "courtroom_id": self.courtroom_id,
+            "judge_id": self.judge_id,
+            "daily_capacity": self.daily_capacity,
+            "case_types": list(self.case_types),
+            "schedule_size": len(self.schedule),
+            "hearings_held": self.hearings_held,
+            "avg_utilization": self.get_average_utilization(),
+        }

scheduler/core/hearing.py

@@ -0,0 +1,134 @@
+"""Hearing event entity and outcome tracking.
+
+This module defines the Hearing class which represents a scheduled court hearing
+with its outcome and associated metadata.
+"""
+
+from dataclasses import dataclass, field
+from datetime import date
+from enum import Enum
+from typing import Optional
+
+
+class HearingOutcome(Enum):
+    """Possible outcomes of a hearing."""
+    SCHEDULED = "SCHEDULED"       # Future hearing
+    HEARD = "HEARD"               # Completed successfully
+    ADJOURNED = "ADJOURNED"       # Postponed
+    DISPOSED = "DISPOSED"         # Case concluded
+    NO_SHOW = "NO_SHOW"           # Party absent
+    WITHDRAWN = "WITHDRAWN"       # Case withdrawn
+
+
+@dataclass
+class Hearing:
+    """Represents a scheduled court hearing event.
+    
+    Attributes:
+        hearing_id: Unique identifier
+        case_id: Associated case
+        scheduled_date: Date of hearing
+        courtroom_id: Assigned courtroom
+        judge_id: Presiding judge
+        stage: Case stage at time of hearing
+        outcome: Result of hearing
+        actual_date: Actual date if rescheduled
+        duration_minutes: Estimated duration
+        notes: Optional notes
+    """
+    hearing_id: str
+    case_id: str
+    scheduled_date: date
+    courtroom_id: int
+    judge_id: str
+    stage: str
+    outcome: HearingOutcome = HearingOutcome.SCHEDULED
+    actual_date: Optional[date] = None
+    duration_minutes: int = 30
+    notes: Optional[str] = None
+    
+    def mark_as_heard(self, actual_date: Optional[date] = None) -> None:
+        """Mark hearing as successfully completed.
+        
+        Args:
+            actual_date: Actual date if different from scheduled
+        """
+        self.outcome = HearingOutcome.HEARD
+        self.actual_date = actual_date or self.scheduled_date
+    
+    def mark_as_adjourned(self, reason: str = "") -> None:
+        """Mark hearing as adjourned.
+        
+        Args:
+            reason: Reason for adjournment
+        """
+        self.outcome = HearingOutcome.ADJOURNED
+        if reason:
+            self.notes = reason
+    
+    def mark_as_disposed(self) -> None:
+        """Mark hearing as final disposition."""
+        self.outcome = HearingOutcome.DISPOSED
+        self.actual_date = self.scheduled_date
+    
+    def mark_as_no_show(self, party: str = "") -> None:
+        """Mark hearing as no-show.
+        
+        Args:
+            party: Which party was absent
+        """
+        self.outcome = HearingOutcome.NO_SHOW
+        if party:
+            self.notes = f"No show: {party}"
+    
+    def reschedule(self, new_date: date) -> None:
+        """Reschedule hearing to a new date.
+        
+        Args:
+            new_date: New scheduled date
+        """
+        self.scheduled_date = new_date
+        self.outcome = HearingOutcome.SCHEDULED
+    
+    def is_complete(self) -> bool:
+        """Check if hearing has concluded.
+        
+        Returns:
+            True if outcome is not SCHEDULED
+        """
+        return self.outcome != HearingOutcome.SCHEDULED
+    
+    def is_successful(self) -> bool:
+        """Check if hearing was successfully held.
+        
+        Returns:
+            True if outcome is HEARD or DISPOSED
+        """
+        return self.outcome in (HearingOutcome.HEARD, HearingOutcome.DISPOSED)
+    
+    def get_effective_date(self) -> date:
+        """Get actual or scheduled date.
+        
+        Returns:
+            actual_date if set, else scheduled_date
+        """
+        return self.actual_date or self.scheduled_date
+    
+    def __repr__(self) -> str:
+        return (f"Hearing(id={self.hearing_id}, case={self.case_id}, "
+                f"date={self.scheduled_date}, outcome={self.outcome.value})")
+    
+    def to_dict(self) -> dict:
+        """Convert hearing to dictionary for serialization."""
+        return {
+            "hearing_id": self.hearing_id,
+            "case_id": self.case_id,
+            "scheduled_date": self.scheduled_date.isoformat(),
+            "actual_date": self.actual_date.isoformat() if self.actual_date else None,
+            "courtroom_id": self.courtroom_id,
+            "judge_id": self.judge_id,
+            "stage": self.stage,
+            "outcome": self.outcome.value,
+            "duration_minutes": self.duration_minutes,
+            "notes": self.notes,
+        }

scheduler/core/judge.py

@@ -0,0 +1,167 @@
+"""Judge entity and workload management.
+
+This module defines the Judge class which represents a judicial officer
+presiding over hearings in a courtroom.
+"""
+
+from dataclasses import dataclass, field
+from datetime import date
+from typing import Dict, List, Optional, Set
+
+
+@dataclass
+class Judge:
+    """Represents a judge with workload tracking.
+    
+    Attributes:
+        judge_id: Unique identifier
+        name: Judge's name
+        courtroom_id: Assigned courtroom (optional)
+        preferred_case_types: Case types this judge specializes in
+        cases_heard: Count of cases heard
+        hearings_presided: Count of hearings presided
+        workload_history: Daily workload tracking
+    """
+    judge_id: str
+    name: str
+    courtroom_id: Optional[int] = None
+    preferred_case_types: Set[str] = field(default_factory=set)
+    cases_heard: int = 0
+    hearings_presided: int = 0
+    workload_history: List[Dict] = field(default_factory=list)
+    
+    def assign_courtroom(self, courtroom_id: int) -> None:
+        """Assign judge to a courtroom.
+        
+        Args:
+            courtroom_id: Courtroom identifier
+        """
+        self.courtroom_id = courtroom_id
+    
+    def add_preferred_types(self, *case_types: str) -> None:
+        """Add case types to judge's preferences.
+        
+        Args:
+            *case_types: One or more case type strings
+        """
+        self.preferred_case_types.update(case_types)
+    
+    def record_hearing(self, hearing_date: date, case_id: str, case_type: str) -> None:
+        """Record a hearing presided over.
+        
+        Args:
+            hearing_date: Date of hearing
+            case_id: Case identifier
+            case_type: Type of case
+        """
+        self.hearings_presided += 1
+    
+    def record_daily_workload(self, hearing_date: date, cases_heard: int, 
+                            cases_adjourned: int) -> None:
+        """Record workload for a specific day.
+        
+        Args:
+            hearing_date: Date of hearings
+            cases_heard: Number of cases actually heard
+            cases_adjourned: Number of cases adjourned
+        """
+        self.workload_history.append({
+            "date": hearing_date,
+            "cases_heard": cases_heard,
+            "cases_adjourned": cases_adjourned,
+            "total_scheduled": cases_heard + cases_adjourned,
+        })
+        
+        self.cases_heard += cases_heard
+    
+    def get_average_daily_workload(self) -> float:
+        """Calculate average cases heard per day.
+        
+        Returns:
+            Average number of cases per day
+        """
+        if not self.workload_history:
+            return 0.0
+        
+        total = sum(day["cases_heard"] for day in self.workload_history)
+        return total / len(self.workload_history)
+    
+    def get_adjournment_rate(self) -> float:
+        """Calculate judge's adjournment rate.
+        
+        Returns:
+            Proportion of cases adjourned (0.0 to 1.0)
+        """
+        if not self.workload_history:
+            return 0.0
+        
+        total_adjourned = sum(day["cases_adjourned"] for day in self.workload_history)
+        total_scheduled = sum(day["total_scheduled"] for day in self.workload_history)
+        
+        return total_adjourned / total_scheduled if total_scheduled > 0 else 0.0
+    
+    def get_workload_summary(self, start_date: date, end_date: date) -> Dict:
+        """Get workload summary for a date range.
+        
+        Args:
+            start_date: Start of range
+            end_date: End of range
+            
+        Returns:
+            Dict with workload statistics
+        """
+        days_in_range = [day for day in self.workload_history 
+                        if start_date <= day["date"] <= end_date]
+        
+        if not days_in_range:
+            return {
+                "judge_id": self.judge_id,
+                "days_worked": 0,
+                "total_cases_heard": 0,
+                "avg_cases_per_day": 0.0,
+                "adjournment_rate": 0.0,
+            }
+        
+        total_heard = sum(day["cases_heard"] for day in days_in_range)
+        total_adjourned = sum(day["cases_adjourned"] for day in days_in_range)
+        total_scheduled = total_heard + total_adjourned
+        
+        return {
+            "judge_id": self.judge_id,
+            "days_worked": len(days_in_range),
+            "total_cases_heard": total_heard,
+            "total_cases_adjourned": total_adjourned,
+            "avg_cases_per_day": total_heard / len(days_in_range),
+            "adjournment_rate": total_adjourned / total_scheduled if total_scheduled > 0 else 0.0,
+        }
+    
+    def is_specialized_in(self, case_type: str) -> bool:
+        """Check if judge specializes in a case type.
+        
+        Args:
+            case_type: Case type to check
+            
+        Returns:
+            True if in preferred types or no preferences set
+        """
+        if not self.preferred_case_types:
+            return True  # No preferences means handles all types
+        
+        return case_type in self.preferred_case_types
+    
+    def __repr__(self) -> str:
+        return (f"Judge(id={self.judge_id}, courtroom={self.courtroom_id}, "
+                f"hearings={self.hearings_presided})")
+    
+    def to_dict(self) -> dict:
+        """Convert judge to dictionary for serialization."""
+        return {
+            "judge_id": self.judge_id,
+            "name": self.name,
+            "courtroom_id": self.courtroom_id,
+            "preferred_case_types": list(self.preferred_case_types),
+            "cases_heard": self.cases_heard,
+            "hearings_presided": self.hearings_presided,
+            "avg_daily_workload": self.get_average_daily_workload(),
+            "adjournment_rate": self.get_adjournment_rate(),
+        }

scheduler/core/ripeness.py

@@ -0,0 +1,216 @@
+"""Case ripeness classification for intelligent scheduling.
+
+Ripe cases are ready for substantive judicial time.
+Unripe cases have bottlenecks (summons, dependencies, parties, documents).
+
+Based on analysis of historical PurposeOfHearing patterns (see scripts/analyze_ripeness_patterns.py).
+"""
+from __future__ import annotations
+
+from enum import Enum
+from typing import TYPE_CHECKING
+from datetime import datetime, timedelta
+
+if TYPE_CHECKING:
+    from scheduler.core.case import Case
+
+
+class RipenessStatus(Enum):
+    """Status indicating whether a case is ready for hearing."""
+    
+    RIPE = "RIPE"  # Ready for hearing
+    UNRIPE_SUMMONS = "UNRIPE_SUMMONS"  # Waiting for summons service
+    UNRIPE_DEPENDENT = "UNRIPE_DEPENDENT"  # Waiting for dependent case/order
+    UNRIPE_PARTY = "UNRIPE_PARTY"  # Party/lawyer unavailable
+    UNRIPE_DOCUMENT = "UNRIPE_DOCUMENT"  # Missing documents/evidence
+    UNKNOWN = "UNKNOWN"  # Cannot determine
+
+    def is_ripe(self) -> bool:
+        """Check if status indicates ripeness."""
+        return self == RipenessStatus.RIPE
+    
+    def is_unripe(self) -> bool:
+        """Check if status indicates unripeness."""
+        return self in {
+            RipenessStatus.UNRIPE_SUMMONS,
+            RipenessStatus.UNRIPE_DEPENDENT,
+            RipenessStatus.UNRIPE_PARTY,
+            RipenessStatus.UNRIPE_DOCUMENT,
+        }
+
+
+# Keywords indicating bottlenecks (data-driven from analyze_ripeness_patterns.py)
+UNRIPE_KEYWORDS = {
+    "SUMMONS": RipenessStatus.UNRIPE_SUMMONS,
+    "NOTICE": RipenessStatus.UNRIPE_SUMMONS,
+    "ISSUE": RipenessStatus.UNRIPE_SUMMONS,
+    "SERVICE": RipenessStatus.UNRIPE_SUMMONS,
+    "STAY": RipenessStatus.UNRIPE_DEPENDENT,
+    "PENDING": RipenessStatus.UNRIPE_DEPENDENT,
+}
+
+RIPE_KEYWORDS = ["ARGUMENTS", "HEARING", "FINAL", "JUDGMENT", "ORDERS", "DISPOSAL"]
+
+
+class RipenessClassifier:
+    """Classify cases as RIPE or UNRIPE for scheduling optimization."""
+    
+    # Stages that indicate case is ready for substantive hearing
+    RIPE_STAGES = [
+        "ARGUMENTS",
+        "EVIDENCE", 
+        "ORDERS / JUDGMENT",
+        "FINAL DISPOSAL"
+    ]
+    
+    # Stages that indicate administrative/preliminary work
+    UNRIPE_STAGES = [
+        "PRE-ADMISSION",
+        "ADMISSION",  # Most cases stuck here waiting for compliance
+        "FRAMING OF CHARGES",
+        "INTERLOCUTORY APPLICATION"
+    ]
+    
+    @classmethod
+    def classify(cls, case: Case, current_date: datetime | None = None) -> RipenessStatus:
+        """Classify case ripeness status with bottleneck type.
+        
+        Args:
+            case: Case to classify
+            current_date: Current simulation date (defaults to now)
+            
+        Returns:
+            RipenessStatus enum indicating ripeness and bottleneck type
+            
+        Algorithm:
+        1. Check last hearing purpose for explicit bottleneck keywords
+        2. Check stage (ADMISSION vs ORDERS/JUDGMENT)
+        3. Check case maturity (days since filing, hearing count)
+        4. Check if stuck (many hearings but no progress)
+        5. Default to RIPE if no bottlenecks detected
+        """
+        if current_date is None:
+            current_date = datetime.now()
+        
+        # 1. Check last hearing purpose for explicit bottleneck keywords
+        if hasattr(case, "last_hearing_purpose") and case.last_hearing_purpose:
+            purpose_upper = case.last_hearing_purpose.upper()
+            
+            for keyword, bottleneck_type in UNRIPE_KEYWORDS.items():
+                if keyword in purpose_upper:
+                    return bottleneck_type
+        
+        # 2. Check stage - ADMISSION stage with few hearings is likely unripe
+        if case.current_stage == "ADMISSION":
+            # New cases in ADMISSION (< 3 hearings) are often unripe
+            if case.hearing_count < 3:
+                return RipenessStatus.UNRIPE_SUMMONS
+        
+        # 3. Check if case is "stuck" (many hearings but no progress)
+        if case.hearing_count > 10:
+            # Calculate average days between hearings
+            if case.age_days > 0:
+                avg_gap = case.age_days / case.hearing_count
+                
+                # If average gap > 60 days, likely stuck due to bottleneck
+                if avg_gap > 60:
+                    return RipenessStatus.UNRIPE_PARTY
+        
+        # 4. Check stage-based ripeness (ripe stages are substantive)
+        if case.current_stage in cls.RIPE_STAGES:
+            return RipenessStatus.RIPE
+        
+        # 5. Default to RIPE if no bottlenecks detected
+        # NOTE: Scheduling gap enforcement (MIN_GAP_BETWEEN_HEARINGS) is handled
+        # by the simulation engine, not the ripeness classifier. Ripeness only
+        # detects substantive bottlenecks (summons, dependencies, party issues).
+        return RipenessStatus.RIPE
+    
+    @classmethod
+    def get_ripeness_priority(cls, case: Case, current_date: datetime | None = None) -> float:
+        """Get priority adjustment based on ripeness.
+        
+        Ripe cases should get judicial time priority over unripe cases
+        when scheduling is tight.
+        
+        Returns:
+            Priority multiplier (1.5 for RIPE, 0.7 for UNRIPE)
+        """
+        ripeness = cls.classify(case, current_date)
+        return 1.5 if ripeness.is_ripe() else 0.7
+    
+    @classmethod
+    def is_schedulable(cls, case: Case, current_date: datetime | None = None) -> bool:
+        """Determine if a case can be scheduled for a hearing.
+        
+        A case is schedulable if:
+        - It is RIPE (no bottlenecks)
+        - It has been sufficient time since last hearing
+        - It is not disposed
+        
+        Args:
+            case: The case to check
+            current_date: Current simulation date
+        
+        Returns:
+            True if case can be scheduled, False otherwise
+        """
+        # Check disposal status
+        if case.is_disposed:
+            return False
+        
+        # Calculate current ripeness
+        ripeness = cls.classify(case, current_date)
+        
+        # Only RIPE cases can be scheduled
+        return ripeness.is_ripe()
+    
+    @classmethod
+    def get_ripeness_reason(cls, ripeness_status: RipenessStatus) -> str:
+        """Get human-readable explanation for ripeness status.
+        
+        Used in dashboard tooltips and reports.
+        
+        Args:
+            ripeness_status: The status to explain
+        
+        Returns:
+            Human-readable explanation string
+        """
+        reasons = {
+            RipenessStatus.RIPE: "Case is ready for hearing (no bottlenecks detected)",
+            RipenessStatus.UNRIPE_SUMMONS: "Waiting for summons service or notice response",
+            RipenessStatus.UNRIPE_DEPENDENT: "Waiting for another case or court order",
+            RipenessStatus.UNRIPE_PARTY: "Party or lawyer unavailable",
+            RipenessStatus.UNRIPE_DOCUMENT: "Missing documents or evidence",
+            RipenessStatus.UNKNOWN: "Insufficient data to determine ripeness",
+        }
+        return reasons.get(ripeness_status, "Unknown status")
+    
+    @classmethod
+    def estimate_ripening_time(cls, case: Case, current_date: datetime) -> timedelta | None:
+        """Estimate time until case becomes ripe.
+        
+        This is a heuristic based on bottleneck type and historical data.
+        
+        Args:
+            case: The case to evaluate
+            current_date: Current simulation date
+        
+        Returns:
+            Estimated timedelta until ripe, or None if already ripe or unknown
+        """
+        ripeness = cls.classify(case, current_date)
+        
+        if ripeness.is_ripe():
+            return timedelta(0)
+        
+        # Heuristic estimates based on bottleneck type
+        estimates = {
+            RipenessStatus.UNRIPE_SUMMONS: timedelta(days=30),
+            RipenessStatus.UNRIPE_DEPENDENT: timedelta(days=60),
+            RipenessStatus.UNRIPE_PARTY: timedelta(days=14),
+            RipenessStatus.UNRIPE_DOCUMENT: timedelta(days=21),
+        }
+        
+        return estimates.get(ripeness, None)

scheduler/data/case_generator.py

@@ -0,0 +1,265 @@
+"""Synthetic case generator (Phase 2).
+
+Generates Case objects between start_date and end_date using:
+- CASE_TYPE_DISTRIBUTION
+- Monthly seasonality factors
+- Urgent case percentage
+- Court working days (CourtCalendar)
+
+Also provides CSV export/import helpers compatible with scripts.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, timedelta
+from pathlib import Path
+from typing import Iterable, List, Tuple
+import csv
+import random
+
+from scheduler.core.case import Case
+from scheduler.utils.calendar import CourtCalendar
+from scheduler.data.config import (
+    CASE_TYPE_DISTRIBUTION,
+    MONTHLY_SEASONALITY,
+    URGENT_CASE_PERCENTAGE,
+)
+from scheduler.data.param_loader import load_parameters
+
+
+def _month_iter(start: date, end: date) -> Iterable[Tuple[int, int]]:
+    y, m = start.year, start.month
+    while (y, m) <= (end.year, end.month):
+        yield (y, m)
+        if m == 12:
+            y += 1
+            m = 1
+        else:
+            m += 1
+
+
+@dataclass
+class CaseGenerator:
+    start: date
+    end: date
+    seed: int = 42
+
+    def generate(self, n_cases: int, stage_mix: dict | None = None, stage_mix_auto: bool = False) -> List[Case]:
+        random.seed(self.seed)
+        cal = CourtCalendar()
+        if stage_mix_auto:
+            params = load_parameters()
+            stage_mix = params.get_stage_stationary_distribution()
+        stage_mix = stage_mix or {"ADMISSION": 1.0}
+        # normalize explicitly
+        total_mix = sum(stage_mix.values()) or 1.0
+        stage_mix = {k: v/total_mix for k, v in stage_mix.items()}
+        # precompute cumulative for stage sampling
+        stage_items = list(stage_mix.items())
+        scum = []
+        accs = 0.0
+        for _, p in stage_items:
+            accs += p
+            scum.append(accs)
+        if scum:
+            scum[-1] = 1.0
+        def sample_stage() -> str:
+            if not stage_items:
+                return "ADMISSION"
+            r = random.random()
+            for i, (st, _) in enumerate(stage_items):
+                if r <= scum[i]:
+                    return st
+            return stage_items[-1][0]
+
+        # duration sampling helpers (lognormal via median & p90)
+        def sample_stage_duration(stage: str) -> float:
+            params = getattr(sample_stage_duration, "_params", None)
+            if params is None:
+                setattr(sample_stage_duration, "_params", load_parameters())
+                params = getattr(sample_stage_duration, "_params")
+            med = params.get_stage_duration(stage, "median")
+            p90 = params.get_stage_duration(stage, "p90")
+            import math
+            med = max(med, 1e-3)
+            p90 = max(p90, med + 1e-6)
+            z = 1.2815515655446004
+            sigma = max(1e-6, math.log(p90) - math.log(med)) / z
+            mu = math.log(med)
+            # Box-Muller normal sample
+            u1 = max(random.random(), 1e-9)
+            u2 = max(random.random(), 1e-9)
+            z0 = ( (-2.0*math.log(u1)) ** 0.5 ) * math.cos(2.0*math.pi*u2)
+            val = math.exp(mu + sigma * z0)
+            return max(1.0, val)
+
+        # 1) Build monthly working-day lists and weights (seasonality * working days)
+        month_days = {}
+        month_weight = {}
+        for (y, m) in _month_iter(self.start, self.end):
+            days = cal.get_working_days_in_month(y, m)
+            # restrict to [start, end]
+            days = [d for d in days if self.start <= d <= self.end]
+            if not days:
+                continue
+            month_days[(y, m)] = days
+            month_weight[(y, m)] = MONTHLY_SEASONALITY.get(m, 1.0) * len(days)
+
+        # normalize weights
+        total_w = sum(month_weight.values())
+        if total_w == 0:
+            return []
+
+        # 2) Allocate case counts per month (round, then adjust)
+        alloc = {}
+        remaining = n_cases
+        for key, w in month_weight.items():
+            cnt = int(round(n_cases * (w / total_w)))
+            alloc[key] = cnt
+        # adjust rounding to total n_cases
+        diff = n_cases - sum(alloc.values())
+        if diff != 0:
+            # distribute the difference across months deterministically by key order
+            keys = sorted(alloc.keys())
+            idx = 0
+            step = 1 if diff > 0 else -1
+            for _ in range(abs(diff)):
+                alloc[keys[idx]] += step
+                idx = (idx + 1) % len(keys)
+
+        # 3) Sampling helpers
+        type_items = list(CASE_TYPE_DISTRIBUTION.items())
+        type_acc = []
+        cum = 0.0
+        for _, p in type_items:
+            cum += p
+            type_acc.append(cum)
+        # ensure last is exactly 1.0 in case of rounding issues
+        if type_acc:
+            type_acc[-1] = 1.0
+
+        def sample_case_type() -> str:
+            r = random.random()
+            for (i, (ct, _)) in enumerate(type_items):
+                if r <= type_acc[i]:
+                    return ct
+            return type_items[-1][0]
+
+        cases: List[Case] = []
+        seq = 0
+        for key in sorted(alloc.keys()):
+            y, m = key
+            days = month_days[key]
+            if not days or alloc[key] <= 0:
+                continue
+            # simple distribution across working days of the month
+            for _ in range(alloc[key]):
+                filed = days[seq % len(days)]
+                seq += 1
+                ct = sample_case_type()
+                urgent = random.random() < URGENT_CASE_PERCENTAGE
+                cid = f"{ct}/{filed.year}/{len(cases)+1:05d}"
+                init_stage = sample_stage()
+                # For initial cases: they're filed on 'filed' date, started current stage on filed date
+                # days_in_stage represents how long they've been in this stage as of simulation start
+                # We sample a duration but cap it to not go before filed_date
+                dur_days = int(sample_stage_duration(init_stage))
+                # stage_start should be between filed_date and some time after
+                # For simplicity: set stage_start = filed_date, case just entered this stage
+                c = Case(
+                    case_id=cid,
+                    case_type=ct,
+                    filed_date=filed,
+                    current_stage=init_stage,
+                    is_urgent=urgent,
+                )
+                c.stage_start_date = filed
+                c.days_in_stage = 0
+                # Initialize realistic hearing history
+                # Spread last hearings across past 7-30 days to simulate realistic court flow
+                # This ensures constant stream of cases becoming eligible, not all at once
+                days_since_filed = (self.end - filed).days
+                if days_since_filed > 30:  # Only if filed at least 30 days before end
+                    c.hearing_count = max(1, days_since_filed // 30)
+                    # Last hearing was randomly 7-30 days before end (spread across a month)
+                    # 7 days = just became eligible, 30 days = long overdue
+                    days_before_end = random.randint(7, 30)
+                    c.last_hearing_date = self.end - timedelta(days=days_before_end)
+                    # Set days_since_last_hearing so simulation starts with staggered eligibility
+                    c.days_since_last_hearing = days_before_end
+                    
+                    # Simulate realistic hearing purposes for ripeness classification
+                    # 20% of cases have bottlenecks (unripe)
+                    bottleneck_purposes = [
+                        "ISSUE SUMMONS",
+                        "FOR NOTICE",
+                        "AWAIT SERVICE OF NOTICE",
+                        "STAY APPLICATION PENDING",
+                        "FOR ORDERS",
+                    ]
+                    ripe_purposes = [
+                        "ARGUMENTS",
+                        "HEARING",
+                        "FINAL ARGUMENTS",
+                        "FOR JUDGMENT",
+                        "EVIDENCE",
+                    ]
+                    
+                    if init_stage == "ADMISSION" and c.hearing_count < 3:
+                        # Early ADMISSION cases more likely unripe
+                        c.last_hearing_purpose = random.choice(bottleneck_purposes) if random.random() < 0.4 else random.choice(ripe_purposes)
+                    elif init_stage in ["ARGUMENTS", "ORDERS / JUDGMENT", "FINAL DISPOSAL"]:
+                        # Advanced stages usually ripe
+                        c.last_hearing_purpose = random.choice(ripe_purposes)
+                    else:
+                        # Mixed
+                        c.last_hearing_purpose = random.choice(bottleneck_purposes) if random.random() < 0.2 else random.choice(ripe_purposes)
+                        
+                cases.append(c)
+
+        return cases
+
+    # CSV helpers -----------------------------------------------------------
+    @staticmethod
+    def to_csv(cases: List[Case], out_path: Path) -> None:
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+        with out_path.open("w", newline="") as f:
+            w = csv.writer(f)
+            w.writerow(["case_id", "case_type", "filed_date", "current_stage", "is_urgent", "hearing_count", "last_hearing_date", "days_since_last_hearing", "last_hearing_purpose"])
+            for c in cases:
+                w.writerow([
+                    c.case_id,
+                    c.case_type,
+                    c.filed_date.isoformat(),
+                    c.current_stage,
+                    1 if c.is_urgent else 0,
+                    c.hearing_count,
+                    c.last_hearing_date.isoformat() if c.last_hearing_date else "",
+                    c.days_since_last_hearing,
+                    c.last_hearing_purpose or "",
+                ])
+
+    @staticmethod
+    def from_csv(path: Path) -> List[Case]:
+        cases: List[Case] = []
+        with path.open("r", newline="") as f:
+            r = csv.DictReader(f)
+            for row in r:
+                c = Case(
+                    case_id=row["case_id"],
+                    case_type=row["case_type"],
+                    filed_date=date.fromisoformat(row["filed_date"]),
+                    current_stage=row.get("current_stage", "ADMISSION"),
+                    is_urgent=(str(row.get("is_urgent", "0")) in ("1", "true", "True")),
+                )
+                # Load hearing history if available
+                if "hearing_count" in row and row["hearing_count"]:
+                    c.hearing_count = int(row["hearing_count"])
+                if "last_hearing_date" in row and row["last_hearing_date"]:
+                    c.last_hearing_date = date.fromisoformat(row["last_hearing_date"])
+                if "days_since_last_hearing" in row and row["days_since_last_hearing"]:
+                    c.days_since_last_hearing = int(row["days_since_last_hearing"])
+                if "last_hearing_purpose" in row and row["last_hearing_purpose"]:
+                    c.last_hearing_purpose = row["last_hearing_purpose"]
+                cases.append(c)
+        return cases

scheduler/data/config.py

@@ -0,0 +1,122 @@
+"""Configuration constants for court scheduling system.
+
+This module contains all configuration parameters and constants used throughout
+the scheduler implementation.
+"""
+
+from pathlib import Path
+from typing import Dict, List
+
+# Project paths
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+REPORTS_DIR = PROJECT_ROOT / "reports" / "figures"
+
+# Find the latest versioned output directory
+def get_latest_params_dir() -> Path:
+    """Get the latest versioned parameters directory from EDA outputs."""
+    if not REPORTS_DIR.exists():
+        raise FileNotFoundError(f"Reports directory not found: {REPORTS_DIR}")
+    
+    version_dirs = [d for d in REPORTS_DIR.iterdir() if d.is_dir() and d.name.startswith("v")]
+    if not version_dirs:
+        raise FileNotFoundError(f"No versioned directories found in {REPORTS_DIR}")
+    
+    latest_dir = max(version_dirs, key=lambda d: d.stat().st_mtime)
+    params_dir = latest_dir / "params"
+    
+    if not params_dir.exists():
+        params_dir = latest_dir  # Fallback if params/ subdirectory doesn't exist
+    
+    return params_dir
+
+# Court operational constants
+WORKING_DAYS_PER_YEAR = 192  # From Karnataka High Court calendar
+COURTROOMS = 5  # Number of courtrooms to simulate
+SIMULATION_YEARS = 2  # Duration of simulation
+SIMULATION_DAYS = WORKING_DAYS_PER_YEAR * SIMULATION_YEARS  # 384 days
+
+# Case type distribution (from EDA)
+CASE_TYPE_DISTRIBUTION = {
+    "CRP": 0.201,  # Civil Revision Petition
+    "CA": 0.200,   # Civil Appeal
+    "RSA": 0.196,  # Regular Second Appeal
+    "RFA": 0.167,  # Regular First Appeal
+    "CCC": 0.111,  # Civil Contempt Petition
+    "CP": 0.096,   # Civil Petition
+    "CMP": 0.028,  # Civil Miscellaneous Petition
+}
+
+# Case types ordered list
+CASE_TYPES = list(CASE_TYPE_DISTRIBUTION.keys())
+
+# Stage taxonomy (from EDA analysis)
+STAGES = [
+    "PRE-ADMISSION",
+    "ADMISSION",
+    "FRAMING OF CHARGES",
+    "EVIDENCE",
+    "ARGUMENTS",
+    "INTERLOCUTORY APPLICATION",
+    "SETTLEMENT",
+    "ORDERS / JUDGMENT",
+    "FINAL DISPOSAL",
+    "OTHER",
+    "NA",
+]
+
+# Terminal stages (case is disposed after these)
+# NA represents case closure in historical data (most common disposal path)
+TERMINAL_STAGES = ["FINAL DISPOSAL", "SETTLEMENT", "NA"]
+
+# Scheduling constraints
+# EDA shows median gaps: RSA=38 days, RFA=31 days, CRP=14 days (transitions.csv)
+# Using conservative 14 days for general scheduling (allows more frequent hearings)
+# Stage-specific gaps handled via transition probabilities in param_loader
+MIN_GAP_BETWEEN_HEARINGS = 14  # days (reduced from 7, based on CRP median)
+MAX_GAP_WITHOUT_ALERT = 90     # days
+URGENT_CASE_PERCENTAGE = 0.05  # 5% of cases marked urgent
+
+# Multi-objective optimization weights
+FAIRNESS_WEIGHT = 0.4
+EFFICIENCY_WEIGHT = 0.3
+URGENCY_WEIGHT = 0.3
+
+# Daily capacity per courtroom (from EDA: median = 151)
+DEFAULT_DAILY_CAPACITY = 151
+
+# Filing rate (cases per year, derived from EDA)
+ANNUAL_FILING_RATE = 6000  # ~500 per month
+MONTHLY_FILING_RATE = ANNUAL_FILING_RATE // 12
+
+# Seasonality factors (relative to average)
+# Lower in May (summer), December-January (holidays)
+MONTHLY_SEASONALITY = {
+    1: 0.90,   # January (holidays)
+    2: 1.15,   # February (peak)
+    3: 1.15,   # March (peak)
+    4: 1.10,   # April (peak)
+    5: 0.70,   # May (summer vacation)
+    6: 0.90,   # June (recovery)
+    7: 1.10,   # July (peak)
+    8: 1.10,   # August (peak)
+    9: 1.10,   # September (peak)
+    10: 1.10,  # October (peak)
+    11: 1.05,  # November (peak)
+    12: 0.85,  # December (holidays approaching)
+}
+
+# Alias for calendar module compatibility
+SEASONALITY_FACTORS = MONTHLY_SEASONALITY
+
+# Success criteria thresholds
+FAIRNESS_GINI_TARGET = 0.4        # Gini coefficient < 0.4
+EFFICIENCY_UTILIZATION_TARGET = 0.85  # > 85% utilization
+URGENCY_SCHEDULING_DAYS = 14      # High-readiness cases scheduled within 14 days
+URGENT_SCHEDULING_DAYS = 7        # Urgent cases scheduled within 7 days
+
+# Random seed for reproducibility
+RANDOM_SEED = 42
+
+# Logging configuration
+LOG_LEVEL = "INFO"
+LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

scheduler/data/param_loader.py

@@ -0,0 +1,343 @@
+"""Load parameters extracted from exploratory data analysis.
+
+This module reads all parameter files generated by the EDA pipeline and makes
+them available to the scheduler.
+"""
+
+import json
+import math
+from pathlib import Path
+from typing import Dict, Optional, List
+
+import pandas as pd
+import polars as pl
+
+from scheduler.data.config import get_latest_params_dir
+
+
+class ParameterLoader:
+    """Loads and manages parameters from EDA outputs.
+
+    Performance notes:
+    - Builds in-memory lookup caches to avoid repeated DataFrame filtering.
+    """
+    
+    def __init__(self, params_dir: Optional[Path] = None):
+        """Initialize parameter loader.
+        
+        Args:
+            params_dir: Directory containing parameter files. If None, uses latest.
+        """
+        self.params_dir = params_dir or get_latest_params_dir()
+        
+        # Cached parameters
+        self._transition_probs: Optional[pd.DataFrame] = None
+        self._stage_duration: Optional[pd.DataFrame] = None
+        self._court_capacity: Optional[Dict] = None
+        self._adjournment_proxies: Optional[pd.DataFrame] = None
+        self._case_type_summary: Optional[pd.DataFrame] = None
+        self._transition_entropy: Optional[pd.DataFrame] = None
+        # caches
+        self._duration_map: Optional[Dict[str, Dict[str, float]]] = None  # stage -> {"median": x, "p90": y}
+        self._transitions_map: Optional[Dict[str, List[tuple]]] = None     # stage_from -> [(stage_to, cum_p), ...]
+        self._adj_map: Optional[Dict[str, Dict[str, float]]] = None        # stage -> {case_type: p_adj}
+        
+    @property
+    def transition_probs(self) -> pd.DataFrame:
+        """Stage transition probabilities.
+        
+        Returns:
+            DataFrame with columns: STAGE_FROM, STAGE_TO, N, row_n, p
+        """
+        if self._transition_probs is None:
+            file_path = self.params_dir / "stage_transition_probs.csv"
+            self._transition_probs = pd.read_csv(file_path)
+        return self._transition_probs
+    
+    def get_transition_prob(self, stage_from: str, stage_to: str) -> float:
+        """Get probability of transitioning from one stage to another.
+        
+        Args:
+            stage_from: Current stage
+            stage_to: Next stage
+            
+        Returns:
+            Transition probability (0-1)
+        """
+        df = self.transition_probs
+        match = df[(df["STAGE_FROM"] == stage_from) & (df["STAGE_TO"] == stage_to)]
+        
+        if len(match) == 0:
+            return 0.0
+        
+        return float(match.iloc[0]["p"])
+    
+    def _build_transitions_map(self) -> None:
+        if self._transitions_map is not None:
+            return
+        df = self.transition_probs
+        self._transitions_map = {}
+        # group by STAGE_FROM, build cumulative probs for fast sampling
+        for st_from, group in df.groupby("STAGE_FROM"):
+            cum = 0.0
+            lst = []
+            for _, row in group.sort_values("p").iterrows():
+                cum += float(row["p"])
+                lst.append((str(row["STAGE_TO"]), cum))
+            # ensure last cum is 1.0 to guard against rounding
+            if lst:
+                to_last, _ = lst[-1]
+                lst[-1] = (to_last, 1.0)
+            self._transitions_map[str(st_from)] = lst
+
+    def get_stage_transitions(self, stage_from: str) -> pd.DataFrame:
+        """Get all possible transitions from a given stage.
+        
+        Args:
+            stage_from: Current stage
+            
+        Returns:
+            DataFrame with STAGE_TO and p columns
+        """
+        df = self.transition_probs
+        return df[df["STAGE_FROM"] == stage_from][["STAGE_TO", "p"]].reset_index(drop=True)
+
+    def get_stage_transitions_fast(self, stage_from: str) -> List[tuple]:
+        """Fast lookup: returns list of (stage_to, cum_p)."""
+        self._build_transitions_map()
+        if not self._transitions_map:
+            return []
+        return self._transitions_map.get(stage_from, [])
+    
+    @property
+    def stage_duration(self) -> pd.DataFrame:
+        """Stage duration statistics.
+        
+        Returns:
+            DataFrame with columns: STAGE, RUN_MEDIAN_DAYS, RUN_P90_DAYS,
+                                   HEARINGS_PER_RUN_MED, N_RUNS
+        """
+        if self._stage_duration is None:
+            file_path = self.params_dir / "stage_duration.csv"
+            self._stage_duration = pd.read_csv(file_path)
+        return self._stage_duration
+    
+    def _build_duration_map(self) -> None:
+        if self._duration_map is not None:
+            return
+        df = self.stage_duration
+        self._duration_map = {}
+        for _, row in df.iterrows():
+            st = str(row["STAGE"])
+            self._duration_map.setdefault(st, {})
+            self._duration_map[st]["median"] = float(row["RUN_MEDIAN_DAYS"])
+            self._duration_map[st]["p90"] = float(row["RUN_P90_DAYS"])
+
+    def get_stage_duration(self, stage: str, percentile: str = "median") -> float:
+        """Get typical duration for a stage.
+        
+        Args:
+            stage: Stage name
+            percentile: 'median' or 'p90'
+            
+        Returns:
+            Duration in days
+        """
+        self._build_duration_map()
+        if not self._duration_map or stage not in self._duration_map:
+            return 30.0
+        p = "median" if percentile == "median" else "p90"
+        return float(self._duration_map[stage].get(p, 30.0))
+    
+    @property
+    def court_capacity(self) -> Dict:
+        """Court capacity metrics.
+        
+        Returns:
+            Dict with keys: slots_median_global, slots_p90_global
+        """
+        if self._court_capacity is None:
+            file_path = self.params_dir / "court_capacity_global.json"
+            with open(file_path, "r") as f:
+                self._court_capacity = json.load(f)
+        return self._court_capacity
+    
+    @property
+    def daily_capacity_median(self) -> int:
+        """Median daily capacity per courtroom."""
+        return int(self.court_capacity["slots_median_global"])
+    
+    @property
+    def daily_capacity_p90(self) -> int:
+        """90th percentile daily capacity per courtroom."""
+        return int(self.court_capacity["slots_p90_global"])
+    
+    @property
+    def adjournment_proxies(self) -> pd.DataFrame:
+        """Adjournment probabilities by stage and case type.
+        
+        Returns:
+            DataFrame with columns: Remappedstages, casetype,
+                                   p_adjourn_proxy, p_not_reached_proxy, n
+        """
+        if self._adjournment_proxies is None:
+            file_path = self.params_dir / "adjournment_proxies.csv"
+            self._adjournment_proxies = pd.read_csv(file_path)
+        return self._adjournment_proxies
+    
+    def _build_adj_map(self) -> None:
+        if self._adj_map is not None:
+            return
+        df = self.adjournment_proxies
+        self._adj_map = {}
+        for _, row in df.iterrows():
+            st = str(row["Remappedstages"])
+            ct = str(row["casetype"])
+            p = float(row["p_adjourn_proxy"])
+            self._adj_map.setdefault(st, {})[ct] = p
+
+    def get_adjournment_prob(self, stage: str, case_type: str) -> float:
+        """Get probability of adjournment for given stage and case type.
+        
+        Args:
+            stage: Stage name
+            case_type: Case type (e.g., 'RSA', 'CRP')
+            
+        Returns:
+            Adjournment probability (0-1)
+        """
+        self._build_adj_map()
+        if not self._adj_map:
+            return 0.4
+        if stage in self._adj_map and case_type in self._adj_map[stage]:
+            return float(self._adj_map[stage][case_type])
+        # fallback: average across types for this stage
+        if stage in self._adj_map and self._adj_map[stage]:
+            vals = list(self._adj_map[stage].values())
+            return float(sum(vals) / len(vals))
+        return 0.4
+    
+    @property
+    def case_type_summary(self) -> pd.DataFrame:
+        """Summary statistics by case type.
+        
+        Returns:
+            DataFrame with columns: CASE_TYPE, n_cases, disp_median,
+                                   disp_p90, hear_median, gap_median
+        """
+        if self._case_type_summary is None:
+            file_path = self.params_dir / "case_type_summary.csv"
+            self._case_type_summary = pd.read_csv(file_path)
+        return self._case_type_summary
+    
+    def get_case_type_stats(self, case_type: str) -> Dict:
+        """Get statistics for a specific case type.
+        
+        Args:
+            case_type: Case type (e.g., 'RSA', 'CRP')
+            
+        Returns:
+            Dict with disp_median, disp_p90, hear_median, gap_median
+        """
+        df = self.case_type_summary
+        match = df[df["CASE_TYPE"] == case_type]
+        
+        if len(match) == 0:
+            raise ValueError(f"Unknown case type: {case_type}")
+        
+        return match.iloc[0].to_dict()
+    
+    @property
+    def transition_entropy(self) -> pd.DataFrame:
+        """Stage transition entropy (predictability metric).
+        
+        Returns:
+            DataFrame with columns: STAGE_FROM, entropy
+        """
+        if self._transition_entropy is None:
+            file_path = self.params_dir / "stage_transition_entropy.csv"
+            self._transition_entropy = pd.read_csv(file_path)
+        return self._transition_entropy
+    
+    def get_stage_predictability(self, stage: str) -> float:
+        """Get predictability of transitions from a stage (inverse of entropy).
+        
+        Args:
+            stage: Stage name
+            
+        Returns:
+            Predictability score (0-1, higher = more predictable)
+        """
+        df = self.transition_entropy
+        match = df[df["STAGE_FROM"] == stage]
+        
+        if len(match) == 0:
+            return 0.5  # Default: medium predictability
+        
+        entropy = float(match.iloc[0]["entropy"])
+        # Convert entropy to predictability (lower entropy = higher predictability)
+        # Max entropy ~1.4, so normalize
+        predictability = max(0.0, 1.0 - (entropy / 1.5))
+        return predictability
+    
+    def get_stage_stationary_distribution(self) -> Dict[str, float]:
+        """Approximate stationary distribution over stages from transition matrix.
+        Returns stage -> probability summing to 1.0.
+        """
+        df = self.transition_probs.copy()
+        # drop nulls and ensure strings
+        df = df[df["STAGE_FROM"].notna() & df["STAGE_TO"].notna()]
+        df["STAGE_FROM"] = df["STAGE_FROM"].astype(str)
+        df["STAGE_TO"] = df["STAGE_TO"].astype(str)
+        stages = sorted(set(df["STAGE_FROM"]).union(set(df["STAGE_TO"])) )
+        idx = {s: i for i, s in enumerate(stages)}
+        n = len(stages)
+        # build dense row-stochastic matrix
+        P = [[0.0]*n for _ in range(n)]
+        for _, row in df.iterrows():
+            i = idx[str(row["STAGE_FROM"])]; j = idx[str(row["STAGE_TO"])]
+            P[i][j] += float(row["p"])
+        # ensure rows sum to 1 by topping up self-loop
+        for i in range(n):
+            s = sum(P[i])
+            if s < 0.999:
+                P[i][i] += (1.0 - s)
+            elif s > 1.001:
+                # normalize if slightly over
+                P[i] = [v/s for v in P[i]]
+        # power iteration
+        pi = [1.0/n]*n
+        for _ in range(200):
+            new = [0.0]*n
+            for j in range(n):
+                acc = 0.0
+                for i in range(n):
+                    acc += pi[i]*P[i][j]
+                new[j] = acc
+            # normalize
+            z = sum(new)
+            if z == 0:
+                break
+            new = [v/z for v in new]
+            # check convergence
+            if sum(abs(new[k]-pi[k]) for k in range(n)) < 1e-9:
+                pi = new
+                break
+            pi = new
+        return {stages[i]: pi[i] for i in range(n)}
+
+    def __repr__(self) -> str:
+        return f"ParameterLoader(params_dir={self.params_dir})"
+
+
+# Convenience function for quick access
+def load_parameters(params_dir: Optional[Path] = None) -> ParameterLoader:
+    """Load parameters from EDA outputs.
+    
+    Args:
+        params_dir: Directory containing parameter files. If None, uses latest.
+        
+    Returns:
+        ParameterLoader instance
+    """
+    return ParameterLoader(params_dir)

scheduler/metrics/basic.py

@@ -0,0 +1,62 @@
+"""Basic metrics for scheduler evaluation.
+
+These helpers avoid heavy dependencies and can be used by scripts.
+"""
+from __future__ import annotations
+
+from typing import Iterable, List, Tuple
+
+
+def gini(values: Iterable[float]) -> float:
+    """Compute the Gini coefficient for a non-negative list of values.
+
+    Args:
+        values: Sequence of non-negative numbers
+
+    Returns:
+        Gini coefficient in [0, 1]
+    """
+    vals = [v for v in values if v is not None]
+    n = len(vals)
+    if n == 0:
+        return 0.0
+    if min(vals) < 0:
+        raise ValueError("Gini expects non-negative values")
+    sorted_vals = sorted(vals)
+    cum = 0.0
+    for i, x in enumerate(sorted_vals, start=1):
+        cum += i * x
+    total = sum(sorted_vals)
+    if total == 0:
+        return 0.0
+    # Gini formula: (2*sum(i*x_i)/(n*sum(x)) - (n+1)/n)
+    return (2 * cum) / (n * total) - (n + 1) / n
+
+
+def utilization(total_scheduled: int, capacity: int) -> float:
+    """Compute utilization as scheduled/capacity.
+
+    Args:
+        total_scheduled: Number of scheduled hearings
+        capacity: Total available slots
+    """
+    if capacity <= 0:
+        return 0.0
+    return min(1.0, total_scheduled / capacity)
+
+
+def urgency_sla(records: List[Tuple[bool, int]], days: int = 7) -> float:
+    """Compute SLA for urgent cases.
+
+    Args:
+        records: List of tuples (is_urgent, working_day_delay)
+        days: SLA threshold in working days
+
+    Returns:
+        Proportion of urgent cases within SLA (0..1)
+    """
+    urgent = [delay for is_urgent, delay in records if is_urgent]
+    if not urgent:
+        return 1.0
+    within = sum(1 for d in urgent if d <= days)
+    return within / len(urgent)

scheduler/output/__init__.py

@@ -0,0 +1,5 @@
+"""Output generation for court scheduling system."""
+
+from .cause_list import CauseListGenerator, generate_cause_lists_from_sweep
+
+__all__ = ['CauseListGenerator', 'generate_cause_lists_from_sweep']

scheduler/output/cause_list.py

@@ -0,0 +1,232 @@
+"""Daily cause list generator for court scheduling system.
+
+Generates machine-readable cause lists from simulation results with explainability.
+"""
+from pathlib import Path
+from typing import Optional
+import pandas as pd
+from datetime import datetime
+
+
+class CauseListGenerator:
+    """Generates daily cause lists with explanations for scheduling decisions."""
+    
+    def __init__(self, events_file: Path):
+        """Initialize with simulation events CSV.
+        
+        Args:
+            events_file: Path to events.csv from simulation
+        """
+        self.events_file = events_file
+        self.events = pd.read_csv(events_file)
+        
+    def generate_daily_lists(self, output_dir: Path) -> Path:
+        """Generate daily cause lists for entire simulation period.
+        
+        Args:
+            output_dir: Directory to save cause list CSVs
+            
+        Returns:
+            Path to compiled cause list CSV
+        """
+        output_dir.mkdir(parents=True, exist_ok=True)
+        
+        # Filter for 'scheduled' events (actual column name is 'type')
+        scheduled = self.events[self.events['type'] == 'scheduled'].copy()
+        
+        if scheduled.empty:
+            raise ValueError("No 'scheduled' events found in simulation")
+        
+        # Parse date column (handle different formats)
+        scheduled['date'] = pd.to_datetime(scheduled['date'])
+        
+        # Add sequence number per courtroom per day
+        # Sort by date, courtroom, then case_id for consistency
+        scheduled = scheduled.sort_values(['date', 'courtroom_id', 'case_id'])
+        scheduled['sequence_number'] = scheduled.groupby(['date', 'courtroom_id']).cumcount() + 1
+        
+        # Build cause list structure
+        cause_list = pd.DataFrame({
+            'Date': scheduled['date'].dt.strftime('%Y-%m-%d'),
+            'Courtroom_ID': scheduled['courtroom_id'].fillna(1).astype(int),
+            'Case_ID': scheduled['case_id'],
+            'Case_Type': scheduled['case_type'],
+            'Stage': scheduled['stage'],
+            'Purpose': 'HEARING',  # Default purpose
+            'Sequence_Number': scheduled['sequence_number'],
+            'Explanation': scheduled.apply(self._generate_explanation, axis=1)
+        })
+        
+        # Save compiled cause list
+        compiled_path = output_dir / "compiled_cause_list.csv"
+        cause_list.to_csv(compiled_path, index=False)
+        
+        # Generate daily summaries
+        daily_summary = cause_list.groupby('Date').agg({
+            'Case_ID': 'count',
+            'Courtroom_ID': 'nunique'
+        }).rename(columns={
+            'Case_ID': 'Total_Hearings',
+            'Courtroom_ID': 'Active_Courtrooms'
+        })
+        
+        summary_path = output_dir / "daily_summaries.csv"
+        daily_summary.to_csv(summary_path)
+        
+        print(f"Generated cause list: {compiled_path}")
+        print(f"  Total hearings: {len(cause_list):,}")
+        print(f"  Date range: {cause_list['Date'].min()} to {cause_list['Date'].max()}")
+        print(f"  Unique cases: {cause_list['Case_ID'].nunique():,}")
+        print(f"Daily summaries: {summary_path}")
+        
+        return compiled_path
+    
+    def _generate_explanation(self, row: pd.Series) -> str:
+        """Generate human-readable explanation for scheduling decision.
+        
+        Args:
+            row: Row from scheduled events DataFrame
+            
+        Returns:
+            Explanation string
+        """
+        parts = []
+        
+        # Case type urgency (heuristic)
+        case_type = row.get('case_type', '')
+        if case_type in ['CCC', 'CP', 'CMP']:
+            parts.append("HIGH URGENCY (criminal)")
+        elif case_type in ['CA', 'CRP']:
+            parts.append("MEDIUM urgency")
+        else:
+            parts.append("standard urgency")
+        
+        # Stage information
+        stage = row.get('stage', '')
+        if isinstance(stage, str):
+            if 'JUDGMENT' in stage or 'ORDER' in stage:
+                parts.append("ready for orders/judgment")
+            elif 'ADMISSION' in stage:
+                parts.append("admission stage")
+        
+        # Courtroom allocation
+        courtroom = row.get('courtroom_id', 1)
+        try:
+            parts.append(f"assigned to Courtroom {int(courtroom)}")
+        except Exception:
+            parts.append("courtroom assigned")
+        
+        # Additional details
+        detail = row.get('detail')
+        if isinstance(detail, str) and detail:
+            parts.append(detail)
+        
+        return " | ".join(parts) if parts else "Scheduled for hearing"
+    
+    def generate_no_case_left_behind_report(self, all_cases_file: Path, output_file: Path):
+        """Verify no case was left unscheduled for too long.
+        
+        Args:
+            all_cases_file: Path to CSV with all cases in simulation
+            output_file: Path to save verification report
+        """
+        scheduled = self.events[self.events['event_type'] == 'HEARING_SCHEDULED'].copy()
+        scheduled['date'] = pd.to_datetime(scheduled['date'])
+        
+        # Get unique cases scheduled
+        scheduled_cases = set(scheduled['case_id'].unique())
+        
+        # Load all cases
+        all_cases = pd.read_csv(all_cases_file)
+        all_case_ids = set(all_cases['case_id'].astype(str).unique())
+        
+        # Find never-scheduled cases
+        never_scheduled = all_case_ids - scheduled_cases
+        
+        # Calculate gaps between hearings per case
+        scheduled['date'] = pd.to_datetime(scheduled['date'])
+        scheduled = scheduled.sort_values(['case_id', 'date'])
+        scheduled['days_since_last'] = scheduled.groupby('case_id')['date'].diff().dt.days
+        
+        # Statistics
+        coverage = len(scheduled_cases) / len(all_case_ids) * 100
+        max_gap = scheduled['days_since_last'].max()
+        avg_gap = scheduled['days_since_last'].mean()
+        
+        report = pd.DataFrame({
+            'Metric': [
+                'Total Cases',
+                'Cases Scheduled At Least Once',
+                'Coverage (%)',
+                'Cases Never Scheduled',
+                'Max Gap Between Hearings (days)',
+                'Avg Gap Between Hearings (days)',
+                'Cases with Gap > 60 days',
+                'Cases with Gap > 90 days'
+            ],
+            'Value': [
+                len(all_case_ids),
+                len(scheduled_cases),
+                f"{coverage:.2f}",
+                len(never_scheduled),
+                f"{max_gap:.0f}" if pd.notna(max_gap) else "N/A",
+                f"{avg_gap:.1f}" if pd.notna(avg_gap) else "N/A",
+                (scheduled['days_since_last'] > 60).sum(),
+                (scheduled['days_since_last'] > 90).sum()
+            ]
+        })
+        
+        report.to_csv(output_file, index=False)
+        print(f"\nNo-Case-Left-Behind Verification Report: {output_file}")
+        print(report.to_string(index=False))
+        
+        return report
+
+
+def generate_cause_lists_from_sweep(sweep_dir: Path, scenario: str, policy: str):
+    """Generate cause lists from comprehensive sweep results.
+    
+    Args:
+        sweep_dir: Path to sweep results directory
+        scenario: Scenario name (e.g., 'baseline_10k')
+        policy: Policy name (e.g., 'readiness')
+    """
+    results_dir = sweep_dir / f"{scenario}_{policy}"
+    events_file = results_dir / "events.csv"
+    
+    if not events_file.exists():
+        raise FileNotFoundError(f"Events file not found: {events_file}")
+    
+    output_dir = results_dir / "cause_lists"
+    
+    generator = CauseListGenerator(events_file)
+    cause_list_path = generator.generate_daily_lists(output_dir)
+    
+    # Generate no-case-left-behind report if cases file exists
+    # This would need the original cases dataset - skip for now
+    # cases_file = sweep_dir / "datasets" / f"{scenario}_cases.csv"
+    # if cases_file.exists():
+    #     report_path = output_dir / "no_case_left_behind.csv"
+    #     generator.generate_no_case_left_behind_report(cases_file, report_path)
+    
+    return cause_list_path
+
+
+if __name__ == "__main__":
+    # Example usage
+    sweep_dir = Path("data/comprehensive_sweep_20251120_184341")
+    
+    # Generate for our algorithm
+    print("="*70)
+    print("Generating Cause Lists for Readiness Algorithm (Our Algorithm)")
+    print("="*70)
+    
+    cause_list = generate_cause_lists_from_sweep(
+        sweep_dir=sweep_dir,
+        scenario="baseline_10k",
+        policy="readiness"
+    )
+    
+    print("\n" + "="*70)
+    print("Cause List Generation Complete")
+    print("="*70)

scheduler/simulation/allocator.py

@@ -115,8 +115,8 @@ class CourtroomAllocator:
                 self.capacity_rejections += 1
                 continue
 
-            # Track if courtroom changed
-            if case.courtroom_id is not None and case.courtroom_id != courtroom_id:
+            # Track if courtroom changed (only count actual switches, not initial assignments)
+            if case.courtroom_id is not None and case.courtroom_id != 0 and case.courtroom_id != courtroom_id:
                 self.allocation_changes += 1
 
             # Assign case to courtroom

scheduler/simulation/engine.py

@@ -279,10 +279,12 @@ class CourtSim:
         
         # Build allocation dict for compatibility with existing loop
         allocation: Dict[int, List[Case]] = {r.courtroom_id: [] for r in self.rooms}
+        seen_cases = set()  # Track seen case_ids to prevent duplicates
         for case in cases_to_allocate:
-            if case.case_id in case_to_courtroom:
+            if case.case_id in case_to_courtroom and case.case_id not in seen_cases:
                 courtroom_id = case_to_courtroom[case.case_id]
                 allocation[courtroom_id].append(case)
+                seen_cases.add(case.case_id)
         
         return allocation
 
@@ -336,11 +338,34 @@ class CourtSim:
             sw.writerow(["case_id", "courtroom_id", "policy", "age_days", "readiness_score", "urgent", "stage", "days_since_last_hearing", "stage_ready_date"])
         for room in self.rooms:
             for case in allocation[room.courtroom_id]:
+                # Skip if case already disposed (safety check)
+                if case.status == CaseStatus.DISPOSED:
+                    continue
+                
                 if room.schedule_case(current, case.case_id):
                     # Mark case as scheduled (for no-case-left-behind tracking)
                     case.mark_scheduled(current)
                     
-                    self._events.write(current, "scheduled", case.case_id, case_type=case.case_type, stage=case.current_stage, courtroom_id=room.courtroom_id)
+                    # Calculate adjournment boost for logging
+                    import math
+                    adj_boost = 0.0
+                    if case.status == CaseStatus.ADJOURNED and case.hearing_count > 0:
+                        adj_boost = math.exp(-case.days_since_last_hearing / 21)
+                    
+                    # Log with full decision metadata
+                    self._events.write(
+                        current, "scheduled", case.case_id, 
+                        case_type=case.case_type, 
+                        stage=case.current_stage, 
+                        courtroom_id=room.courtroom_id,
+                        priority_score=case.get_priority_score(),
+                        age_days=case.age_days,
+                        readiness_score=case.readiness_score,
+                        is_urgent=case.is_urgent,
+                        adj_boost=adj_boost,
+                        ripeness_status=case.ripeness_status,
+                        days_since_hearing=case.days_since_last_hearing
+                    )
                     day_total += 1
                     self._hearings_total += 1
                     # log suggestive rationale
@@ -438,6 +463,32 @@ class CourtSim:
         # Generate courtroom allocation summary
         print(f"\n{self.allocator.get_courtroom_summary()}")
         
+        # Generate comprehensive case status breakdown
+        total_cases = len(self.cases)
+        disposed_cases = [c for c in self.cases if c.status == CaseStatus.DISPOSED]
+        scheduled_at_least_once = [c for c in self.cases if c.last_scheduled_date is not None]
+        never_scheduled = [c for c in self.cases if c.last_scheduled_date is None]
+        scheduled_but_not_disposed = [c for c in scheduled_at_least_once if c.status != CaseStatus.DISPOSED]
+        
+        print(f"\n=== Case Status Breakdown ===")
+        print(f"Total cases in system: {total_cases:,}")
+        print(f"\nScheduling outcomes:")
+        print(f"  Scheduled at least once: {len(scheduled_at_least_once):,} ({len(scheduled_at_least_once)/total_cases*100:.1f}%)")
+        print(f"    - Disposed: {len(disposed_cases):,} ({len(disposed_cases)/total_cases*100:.1f}%)")
+        print(f"    - Active (not disposed): {len(scheduled_but_not_disposed):,} ({len(scheduled_but_not_disposed)/total_cases*100:.1f}%)")
+        print(f"  Never scheduled: {len(never_scheduled):,} ({len(never_scheduled)/total_cases*100:.1f}%)")
+        
+        if scheduled_at_least_once:
+            avg_hearings = sum(c.hearing_count for c in scheduled_at_least_once) / len(scheduled_at_least_once)
+            print(f"\nAverage hearings per scheduled case: {avg_hearings:.1f}")
+        
+        if disposed_cases:
+            avg_hearings_to_disposal = sum(c.hearing_count for c in disposed_cases) / len(disposed_cases)
+            avg_days_to_disposal = sum((c.disposal_date - c.filed_date).days for c in disposed_cases) / len(disposed_cases)
+            print(f"\nDisposal metrics:")
+            print(f"  Average hearings to disposal: {avg_hearings_to_disposal:.1f}")
+            print(f"  Average days to disposal: {avg_days_to_disposal:.0f}")
+        
         return CourtSimResult(
             hearings_total=self._hearings_total,
             hearings_heard=self._hearings_heard,

scheduler/simulation/events.py

@@ -0,0 +1,63 @@
+"""Event schema and writer for simulation audit trail.
+
+Each event is a flat dict suitable for CSV logging with a 'type' field.
+Types:
+- filing: a new case filed into the system
+- scheduled: a case scheduled on a date
+- outcome: hearing outcome (heard/adjourned)
+- stage_change: case progresses to a new stage
+- disposed: case disposed
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+import csv
+from typing import Dict, Any, Iterable
+
+
+@dataclass
+class EventWriter:
+    path: Path
+
+    def __post_init__(self) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._buffer = []  # in-memory rows to append
+        if not self.path.exists():
+            with self.path.open("w", newline="") as f:
+                w = csv.writer(f)
+                w.writerow([
+                    "date", "type", "case_id", "case_type", "stage", "courtroom_id",
+                    "detail", "extra",
+                    "priority_score", "age_days", "readiness_score", "is_urgent",
+                    "adj_boost", "ripeness_status", "days_since_hearing"
+                ])
+
+    def write(self, date_: date, type_: str, case_id: str, case_type: str = "",
+              stage: str = "", courtroom_id: int | None = None,
+              detail: str = "", extra: str = "",
+              priority_score: float | None = None, age_days: int | None = None,
+              readiness_score: float | None = None, is_urgent: bool | None = None,
+              adj_boost: float | None = None, ripeness_status: str = "",
+              days_since_hearing: int | None = None) -> None:
+        self._buffer.append([
+            date_.isoformat(), type_, case_id, case_type, stage,
+            courtroom_id if courtroom_id is not None else "",
+            detail, extra,
+            f"{priority_score:.4f}" if priority_score is not None else "",
+            age_days if age_days is not None else "",
+            f"{readiness_score:.4f}" if readiness_score is not None else "",
+            int(is_urgent) if is_urgent is not None else "",
+            f"{adj_boost:.4f}" if adj_boost is not None else "",
+            ripeness_status,
+            days_since_hearing if days_since_hearing is not None else "",
+        ])
+
+    def flush(self) -> None:
+        if not self._buffer:
+            return
+        with self.path.open("a", newline="") as f:
+            w = csv.writer(f)
+            w.writerows(self._buffer)
+        self._buffer.clear()

scheduler/simulation/policies/__init__.py

@@ -0,0 +1,18 @@
+"""Scheduling policy implementations."""
+from scheduler.simulation.policies.fifo import FIFOPolicy
+from scheduler.simulation.policies.age import AgeBasedPolicy
+from scheduler.simulation.policies.readiness import ReadinessPolicy
+
+POLICY_REGISTRY = {
+    "fifo": FIFOPolicy,
+    "age": AgeBasedPolicy,
+    "readiness": ReadinessPolicy,
+}
+
+def get_policy(name: str):
+    name_lower = name.lower()
+    if name_lower not in POLICY_REGISTRY:
+        raise ValueError(f"Unknown policy: {name}")
+    return POLICY_REGISTRY[name_lower]()
+
+__all__ = ["FIFOPolicy", "AgeBasedPolicy", "ReadinessPolicy", "get_policy"]

scheduler/simulation/policies/age.py

@@ -0,0 +1,38 @@
+"""Age-based scheduling policy.
+
+Prioritizes older cases to reduce maximum age and prevent starvation.
+Uses case age (days since filing) as primary criterion.
+"""
+from __future__ import annotations
+
+from datetime import date
+from typing import List
+
+from scheduler.simulation.scheduler import SchedulerPolicy
+from scheduler.core.case import Case
+
+
+class AgeBasedPolicy(SchedulerPolicy):
+    """Age-based scheduling: oldest cases scheduled first."""
+    
+    def prioritize(self, cases: List[Case], current_date: date) -> List[Case]:
+        """Sort cases by age (oldest first).
+        
+        Args:
+            cases: List of eligible cases
+            current_date: Current simulation date
+            
+        Returns:
+            Cases sorted by age_days (descending)
+        """
+        # Update ages first
+        for c in cases:
+            c.update_age(current_date)
+        
+        return sorted(cases, key=lambda c: c.age_days, reverse=True)
+    
+    def get_name(self) -> str:
+        return "Age-Based"
+    
+    def requires_readiness_score(self) -> bool:
+        return False

scheduler/simulation/policies/fifo.py

@@ -0,0 +1,34 @@
+"""First-In-First-Out (FIFO) scheduling policy.
+
+Schedules cases in the order they were filed, treating all cases equally.
+This is the simplest baseline policy.
+"""
+from __future__ import annotations
+
+from datetime import date
+from typing import List
+
+from scheduler.simulation.scheduler import SchedulerPolicy
+from scheduler.core.case import Case
+
+
+class FIFOPolicy(SchedulerPolicy):
+    """FIFO scheduling: cases scheduled in filing order."""
+    
+    def prioritize(self, cases: List[Case], current_date: date) -> List[Case]:
+        """Sort cases by filed_date (earliest first).
+        
+        Args:
+            cases: List of eligible cases
+            current_date: Current simulation date (unused)
+            
+        Returns:
+            Cases sorted by filing date (oldest first)
+        """
+        return sorted(cases, key=lambda c: c.filed_date)
+    
+    def get_name(self) -> str:
+        return "FIFO"
+    
+    def requires_readiness_score(self) -> bool:
+        return False

scheduler/simulation/policies/readiness.py

@@ -0,0 +1,48 @@
+"""Readiness-based scheduling policy.
+
+Combines age, readiness score, and urgency into a composite priority score.
+This is the most sophisticated policy, balancing fairness with efficiency.
+
+Priority formula:
+  priority = (age/2000) * 0.4 + readiness * 0.3 + urgent * 0.3
+"""
+from __future__ import annotations
+
+from datetime import date
+from typing import List
+
+from scheduler.simulation.scheduler import SchedulerPolicy
+from scheduler.core.case import Case
+
+
+class ReadinessPolicy(SchedulerPolicy):
+    """Readiness-based scheduling: composite priority score."""
+    
+    def prioritize(self, cases: List[Case], current_date: date) -> List[Case]:
+        """Sort cases by composite priority score (highest first).
+        
+        The priority score combines:
+        - Age (40% weight)
+        - Readiness (30% weight)
+        - Urgency (30% weight)
+        
+        Args:
+            cases: List of eligible cases
+            current_date: Current simulation date
+            
+        Returns:
+            Cases sorted by priority score (descending)
+        """
+        # Update ages and compute readiness
+        for c in cases:
+            c.update_age(current_date)
+            c.compute_readiness_score()
+        
+        # Sort by priority score (higher = more urgent)
+        return sorted(cases, key=lambda c: c.get_priority_score(), reverse=True)
+    
+    def get_name(self) -> str:
+        return "Readiness-Based"
+    
+    def requires_readiness_score(self) -> bool:
+        return True

scheduler/simulation/scheduler.py

@@ -0,0 +1,43 @@
+"""Base scheduler interface for policy implementations.
+
+This module defines the abstract interface that all scheduling policies must implement.
+Each policy decides which cases to schedule on a given day based on different criteria.
+"""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from datetime import date
+from typing import List
+
+from scheduler.core.case import Case
+
+
+class SchedulerPolicy(ABC):
+    """Abstract base class for scheduling policies.
+    
+    All scheduling policies must implement the `prioritize` method which
+    ranks cases for scheduling on a given day.
+    """
+    
+    @abstractmethod
+    def prioritize(self, cases: List[Case], current_date: date) -> List[Case]:
+        """Prioritize cases for scheduling on the given date.
+        
+        Args:
+            cases: List of eligible cases (already filtered for readiness, not disposed)
+            current_date: Current simulation date
+            
+        Returns:
+            Sorted list of cases in priority order (highest priority first)
+        """
+        pass
+    
+    @abstractmethod
+    def get_name(self) -> str:
+        """Get the policy name for logging/reporting."""
+        pass
+    
+    @abstractmethod
+    def requires_readiness_score(self) -> bool:
+        """Return True if this policy requires readiness score computation."""
+        pass

scheduler/utils/calendar.py

@@ -0,0 +1,217 @@
+"""Court calendar utilities with working days and seasonality.
+
+This module provides utilities for calculating working days considering
+court holidays, seasonality, and Karnataka High Court calendar.
+"""
+
+from datetime import date, timedelta
+from typing import List, Set
+
+from scheduler.data.config import (
+    WORKING_DAYS_PER_YEAR,
+    SEASONALITY_FACTORS,
+)
+
+
+class CourtCalendar:
+    """Manages court working days and seasonality.
+    
+    Attributes:
+        holidays: Set of holiday dates
+        working_days_per_year: Expected working days annually
+    """
+    
+    def __init__(self, working_days_per_year: int = WORKING_DAYS_PER_YEAR):
+        """Initialize court calendar.
+        
+        Args:
+            working_days_per_year: Annual working days (default 192)
+        """
+        self.working_days_per_year = working_days_per_year
+        self.holidays: Set[date] = set()
+    
+    def add_holiday(self, holiday_date: date) -> None:
+        """Add a holiday to the calendar.
+        
+        Args:
+            holiday_date: Date to mark as holiday
+        """
+        self.holidays.add(holiday_date)
+    
+    def add_holidays(self, holiday_dates: List[date]) -> None:
+        """Add multiple holidays.
+        
+        Args:
+            holiday_dates: List of dates to mark as holidays
+        """
+        self.holidays.update(holiday_dates)
+    
+    def is_working_day(self, check_date: date) -> bool:
+        """Check if a date is a working day.
+        
+        Args:
+            check_date: Date to check
+            
+        Returns:
+            True if date is a working day (not weekend or holiday)
+        """
+        # Saturday (5) and Sunday (6) are weekends
+        if check_date.weekday() in (5, 6):
+            return False
+        
+        if check_date in self.holidays:
+            return False
+        
+        return True
+    
+    def next_working_day(self, start_date: date, days_ahead: int = 1) -> date:
+        """Get the next working day after a given number of working days.
+        
+        Args:
+            start_date: Starting date
+            days_ahead: Number of working days to advance
+            
+        Returns:
+            Next working day date
+        """
+        current = start_date
+        working_days_found = 0
+        
+        while working_days_found < days_ahead:
+            current += timedelta(days=1)
+            if self.is_working_day(current):
+                working_days_found += 1
+        
+        return current
+    
+    def working_days_between(self, start_date: date, end_date: date) -> int:
+        """Count working days between two dates (inclusive).
+        
+        Args:
+            start_date: Start of range
+            end_date: End of range
+            
+        Returns:
+            Number of working days
+        """
+        if start_date > end_date:
+            return 0
+        
+        count = 0
+        current = start_date
+        
+        while current <= end_date:
+            if self.is_working_day(current):
+                count += 1
+            current += timedelta(days=1)
+        
+        return count
+    
+    def get_working_days_in_month(self, year: int, month: int) -> List[date]:
+        """Get all working days in a specific month.
+        
+        Args:
+            year: Year
+            month: Month (1-12)
+            
+        Returns:
+            List of working day dates
+        """
+        # Get first and last day of month
+        first_day = date(year, month, 1)
+        
+        if month == 12:
+            last_day = date(year, 12, 31)
+        else:
+            last_day = date(year, month + 1, 1) - timedelta(days=1)
+        
+        working_days = []
+        current = first_day
+        
+        while current <= last_day:
+            if self.is_working_day(current):
+                working_days.append(current)
+            current += timedelta(days=1)
+        
+        return working_days
+    
+    def get_working_days_in_year(self, year: int) -> List[date]:
+        """Get all working days in a year.
+        
+        Args:
+            year: Year
+            
+        Returns:
+            List of working day dates
+        """
+        working_days = []
+        
+        for month in range(1, 13):
+            working_days.extend(self.get_working_days_in_month(year, month))
+        
+        return working_days
+    
+    def get_seasonality_factor(self, check_date: date) -> float:
+        """Get seasonality factor for a date based on month.
+        
+        Args:
+            check_date: Date to check
+            
+        Returns:
+            Seasonality multiplier (from config)
+        """
+        return SEASONALITY_FACTORS.get(check_date.month, 1.0)
+    
+    def get_expected_capacity(self, check_date: date, base_capacity: int) -> int:
+        """Get expected capacity adjusted for seasonality.
+        
+        Args:
+            check_date: Date to check
+            base_capacity: Base daily capacity
+            
+        Returns:
+            Adjusted capacity
+        """
+        factor = self.get_seasonality_factor(check_date)
+        return int(base_capacity * factor)
+    
+    def generate_court_calendar(self, start_date: date, end_date: date) -> List[date]:
+        """Generate list of all court working days in a date range.
+        
+        Args:
+            start_date: Start of simulation
+            end_date: End of simulation
+            
+        Returns:
+            List of working day dates
+        """
+        working_days = []
+        current = start_date
+        
+        while current <= end_date:
+            if self.is_working_day(current):
+                working_days.append(current)
+            current += timedelta(days=1)
+        
+        return working_days
+    
+    def add_standard_holidays(self, year: int) -> None:
+        """Add standard Indian national holidays for a year.
+        
+        This is a simplified set. In production, use actual court holiday calendar.
+        
+        Args:
+            year: Year to add holidays for
+        """
+        # Standard national holidays (simplified)
+        holidays = [
+            date(year, 1, 26),   # Republic Day
+            date(year, 8, 15),   # Independence Day
+            date(year, 10, 2),   # Gandhi Jayanti
+            date(year, 12, 25),  # Christmas
+        ]
+        
+        self.add_holidays(holidays)
+    
+    def __repr__(self) -> str:
+        return f"CourtCalendar(working_days/year={self.working_days_per_year}, holidays={len(self.holidays)})"

scripts/demo_explainability_and_controls.py

@@ -0,0 +1,378 @@
+"""Demonstration of explainability and judge intervention controls.
+
+Shows:
+1. Step-by-step decision reasoning for scheduled/unscheduled cases
+2. Judge override capabilities
+3. Draft cause list review and approval process
+4. Audit trail tracking
+"""
+from datetime import date, datetime
+from pathlib import Path
+import sys
+
+# Add parent directory to path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from scheduler.core.case import Case, CaseStatus
+from scheduler.control.explainability import ExplainabilityEngine
+from scheduler.control.overrides import (
+    OverrideManager,
+    Override,
+    OverrideType
+)
+
+
+def demo_explainability():
+    """Demonstrate step-by-step decision reasoning."""
+    print("=" * 80)
+    print("DEMO 1: EXPLAINABILITY - STEP-BY-STEP DECISION REASONING")
+    print("=" * 80)
+    print()
+    
+    # Create a sample case
+    case = Case(
+        case_id="CRP/2023/01234",
+        case_type="CRP",
+        filed_date=date(2023, 1, 15),
+        current_stage="ORDERS / JUDGMENT",
+        is_urgent=True
+    )
+    
+    # Simulate case progression
+    case.age_days = 180
+    case.hearing_count = 3
+    case.days_since_last_hearing = 21
+    case.last_hearing_date = date(2023, 6, 1)
+    case.last_hearing_purpose = "ARGUMENTS"
+    case.readiness_score = 0.85
+    case.ripeness_status = "RIPE"
+    case.status = CaseStatus.ADJOURNED
+    
+    # Calculate priority
+    priority_score = case.get_priority_score()
+    
+    # Example 1: Case SCHEDULED
+    print("Example 1: Case SCHEDULED")
+    print("-" * 80)
+    
+    explanation = ExplainabilityEngine.explain_scheduling_decision(
+        case=case,
+        current_date=date(2023, 6, 22),
+        scheduled=True,
+        ripeness_status="RIPE",
+        priority_score=priority_score,
+        courtroom_id=3,
+        capacity_full=False,
+        below_threshold=False
+    )
+    
+    print(explanation.to_readable_text())
+    print()
+    
+    # Example 2: Case NOT SCHEDULED (capacity full)
+    print("\n" + "=" * 80)
+    print("Example 2: Case NOT SCHEDULED (Capacity Full)")
+    print("-" * 80)
+    
+    explanation2 = ExplainabilityEngine.explain_scheduling_decision(
+        case=case,
+        current_date=date(2023, 6, 22),
+        scheduled=False,
+        ripeness_status="RIPE",
+        priority_score=priority_score,
+        courtroom_id=None,
+        capacity_full=True,
+        below_threshold=False
+    )
+    
+    print(explanation2.to_readable_text())
+    print()
+    
+    # Example 3: Case NOT SCHEDULED (unripe)
+    print("\n" + "=" * 80)
+    print("Example 3: Case NOT SCHEDULED (UNRIPE - Summons Pending)")
+    print("-" * 80)
+    
+    case_unripe = Case(
+        case_id="RSA/2023/05678",
+        case_type="RSA",
+        filed_date=date(2023, 5, 1),
+        current_stage="ADMISSION",
+        is_urgent=False
+    )
+    case_unripe.age_days = 50
+    case_unripe.readiness_score = 0.2
+    case_unripe.ripeness_status = "UNRIPE_SUMMONS"
+    case_unripe.last_hearing_purpose = "ISSUE SUMMONS"
+    
+    explanation3 = ExplainabilityEngine.explain_scheduling_decision(
+        case=case_unripe,
+        current_date=date(2023, 6, 22),
+        scheduled=False,
+        ripeness_status="UNRIPE_SUMMONS",
+        priority_score=None,
+        courtroom_id=None,
+        capacity_full=False,
+        below_threshold=False
+    )
+    
+    print(explanation3.to_readable_text())
+    print()
+
+
+def demo_judge_overrides():
+    """Demonstrate judge intervention controls."""
+    print("\n" + "=" * 80)
+    print("DEMO 2: JUDGE INTERVENTION CONTROLS")
+    print("=" * 80)
+    print()
+    
+    # Create override manager
+    manager = OverrideManager()
+    
+    # Create a draft cause list
+    print("Step 1: Algorithm generates draft cause list")
+    print("-" * 80)
+    
+    algorithm_suggested = [
+        "CRP/2023/00101",
+        "CRP/2023/00102",
+        "RSA/2023/00201",
+        "CA/2023/00301",
+        "CCC/2023/00401"
+    ]
+    
+    draft = manager.create_draft(
+        date=date(2023, 6, 22),
+        courtroom_id=3,
+        judge_id="J001",
+        algorithm_suggested=algorithm_suggested
+    )
+    
+    print(f"Draft created for {draft.date}")
+    print(f"Courtroom: {draft.courtroom_id}")
+    print(f"Judge: {draft.judge_id}")
+    print(f"Algorithm suggested {len(algorithm_suggested)} cases:")
+    for i, case_id in enumerate(algorithm_suggested, 1):
+        print(f"  {i}. {case_id}")
+    print()
+    
+    # Judge starts with algorithm suggestions
+    draft.judge_approved = algorithm_suggested.copy()
+    
+    # Step 2: Judge makes overrides
+    print("\nStep 2: Judge reviews and makes modifications")
+    print("-" * 80)
+    
+    # Override 1: Judge adds an urgent case
+    print("\nOverride 1: Judge adds urgent case")
+    override1 = Override(
+        override_id="OV001",
+        override_type=OverrideType.ADD_CASE,
+        case_id="CCC/2023/00999",
+        judge_id="J001",
+        timestamp=datetime.now(),
+        reason="Medical emergency case, party has critical health condition"
+    )
+    
+    success, error = manager.apply_override(draft, override1)
+    if success:
+        print(f"  ✓ {override1.to_readable_text()}")
+    else:
+        print(f"  ✗ Failed: {error}")
+    print()
+    
+    # Override 2: Judge removes a case
+    print("Override 2: Judge removes a case")
+    override2 = Override(
+        override_id="OV002",
+        override_type=OverrideType.REMOVE_CASE,
+        case_id="RSA/2023/00201",
+        judge_id="J001",
+        timestamp=datetime.now(),
+        reason="Party requested postponement due to family emergency"
+    )
+    
+    success, error = manager.apply_override(draft, override2)
+    if success:
+        print(f"  ✓ {override2.to_readable_text()}")
+    else:
+        print(f"  ✗ Failed: {error}")
+    print()
+    
+    # Override 3: Judge overrides ripeness
+    print("Override 3: Judge overrides ripeness status")
+    override3 = Override(
+        override_id="OV003",
+        override_type=OverrideType.RIPENESS,
+        case_id="CRP/2023/00102",
+        judge_id="J001",
+        timestamp=datetime.now(),
+        old_value="UNRIPE_SUMMONS",
+        new_value="RIPE",
+        reason="Summons served yesterday, confirmation received this morning"
+    )
+    
+    success, error = manager.apply_override(draft, override3)
+    if success:
+        print(f"  ✓ {override3.to_readable_text()}")
+    else:
+        print(f"  ✗ Failed: {error}")
+    print()
+    
+    # Step 3: Judge approves final list
+    print("\nStep 3: Judge finalizes cause list")
+    print("-" * 80)
+    
+    manager.finalize_draft(draft)
+    
+    print(f"Status: {draft.status}")
+    print(f"Finalized at: {draft.finalized_at.strftime('%Y-%m-%d %H:%M') if draft.finalized_at else 'N/A'}")
+    print()
+    
+    # Show modifications summary
+    print("Modifications Summary:")
+    summary = draft.get_modifications_summary()
+    print(f"  Cases added: {summary['cases_added']}")
+    print(f"  Cases removed: {summary['cases_removed']}")
+    print(f"  Cases kept: {summary['cases_kept']}")
+    print(f"  Acceptance rate: {summary['acceptance_rate']:.1f}%")
+    print(f"  Override types: {summary['override_types']}")
+    print()
+    
+    # Show final list
+    print("Final Approved Cases:")
+    for i, case_id in enumerate(draft.judge_approved, 1):
+        marker = "  [NEW]" if case_id not in algorithm_suggested else ""
+        print(f"  {i}. {case_id}{marker}")
+    print()
+
+
+def demo_judge_preferences():
+    """Demonstrate judge-specific preferences."""
+    print("\n" + "=" * 80)
+    print("DEMO 3: JUDGE PREFERENCES")
+    print("=" * 80)
+    print()
+    
+    manager = OverrideManager()
+    
+    # Set judge preferences
+    prefs = manager.get_judge_preferences("J001")
+    
+    print("Judge J001 Preferences:")
+    print("-" * 80)
+    
+    # Set capacity override
+    prefs.daily_capacity_override = 120
+    print(f"Daily capacity override: {prefs.daily_capacity_override} (default: 151)")
+    print("  Reason: Judge works half-days on Fridays")
+    print()
+    
+    # Block dates
+    prefs.blocked_dates = [
+        date(2023, 7, 10),
+        date(2023, 7, 11),
+        date(2023, 7, 12)
+    ]
+    print("Blocked dates:")
+    for blocked in prefs.blocked_dates:
+        print(f"  - {blocked} (vacation)")
+    print()
+    
+    # Case type preferences
+    prefs.case_type_preferences = {
+        "Monday": ["CRP", "CA"],
+        "Wednesday": ["RSA", "RFA"]
+    }
+    print("Case type preferences by day:")
+    for day, types in prefs.case_type_preferences.items():
+        print(f"  {day}: {', '.join(types)}")
+    print()
+
+
+def demo_audit_trail():
+    """Demonstrate audit trail export."""
+    print("\n" + "=" * 80)
+    print("DEMO 4: AUDIT TRAIL")
+    print("=" * 80)
+    print()
+    
+    manager = OverrideManager()
+    
+    # Simulate some activity
+    draft1 = manager.create_draft(
+        date=date(2023, 6, 22),
+        courtroom_id=1,
+        judge_id="J001",
+        algorithm_suggested=["CRP/001", "CA/002", "RSA/003"]
+    )
+    draft1.judge_approved = ["CRP/001", "CA/002"]  # Removed one
+    draft1.status = "APPROVED"
+    
+    override = Override(
+        override_id="OV001",
+        override_type=OverrideType.REMOVE_CASE,
+        case_id="RSA/003",
+        judge_id="J001",
+        timestamp=datetime.now(),
+        reason="Party unavailable"
+    )
+    draft1.overrides.append(override)
+    manager.overrides.append(override)
+    
+    # Get statistics
+    stats = manager.get_override_statistics()
+    
+    print("Override Statistics:")
+    print("-" * 80)
+    print(f"Total overrides: {stats['total_overrides']}")
+    print(f"Total drafts: {stats['total_drafts']}")
+    print(f"Approved drafts: {stats['approved_drafts']}")
+    print(f"Average acceptance rate: {stats['avg_acceptance_rate']:.1f}%")
+    print(f"Modification rate: {stats['modification_rate']:.1f}%")
+    print(f"By type: {stats['by_type']}")
+    print()
+    
+    # Export audit trail
+    output_file = "demo_audit_trail.json"
+    manager.export_audit_trail(output_file)
+    print(f"✓ Audit trail exported to: {output_file}")
+    print()
+
+
+def main():
+    """Run all demonstrations."""
+    print("\n")
+    print("#" * 80)
+    print("# COURT SCHEDULING SYSTEM - EXPLAINABILITY & CONTROLS DEMO")
+    print("# Demonstrating step-by-step reasoning and judge intervention")
+    print("#" * 80)
+    print()
+    
+    demo_explainability()
+    demo_judge_overrides()
+    demo_judge_preferences()
+    demo_audit_trail()
+    
+    print("\n" + "=" * 80)
+    print("DEMO COMPLETE")
+    print("=" * 80)
+    print()
+    print("Key Takeaways:")
+    print("1. Every scheduling decision has step-by-step explanation")
+    print("2. Judges can override ANY algorithmic decision with reasoning")
+    print("3. All overrides are tracked in audit trail")
+    print("4. System is SUGGESTIVE, not prescriptive")
+    print("5. Judge preferences are respected (capacity, blocked dates, etc.)")
+    print()
+    print("This demonstrates compliance with hackathon requirements:")
+    print("  - Decision transparency (Phase 6.5 requirement)")
+    print("  - User control and overrides (Phase 6.5 requirement)")
+    print("  - Explainability for each step (Step 3 compliance)")
+    print("  - Audit trail tracking (Phase 6.5 requirement)")
+    print()
+
+
+if __name__ == "__main__":
+    main()

scripts/generate_all_cause_lists.py

@@ -0,0 +1,261 @@
+"""Generate cause lists for all scenarios and policies from comprehensive sweep.
+
+Analyzes distribution and statistics of daily generated cause lists across scenarios and policies.
+"""
+from pathlib import Path
+import pandas as pd
+import matplotlib.pyplot as plt
+import seaborn as sns
+from scheduler.output.cause_list import CauseListGenerator
+
+# Set style
+plt.style.use('seaborn-v0_8-darkgrid')
+sns.set_palette("husl")
+
+# Find latest sweep directory
+data_dir = Path("data")
+sweep_dirs = sorted([d for d in data_dir.glob("comprehensive_sweep_*")], reverse=True)
+if not sweep_dirs:
+    raise FileNotFoundError("No sweep directories found")
+
+sweep_dir = sweep_dirs[0]
+print(f"Processing sweep: {sweep_dir.name}")
+print("=" * 80)
+
+# Get all result directories
+result_dirs = [d for d in sweep_dir.iterdir() if d.is_dir() and d.name != "datasets"]
+
+# Generate cause lists for each
+all_stats = []
+
+for result_dir in result_dirs:
+    events_file = result_dir / "events.csv"
+    if not events_file.exists():
+        continue
+    
+    # Parse scenario and policy from directory name
+    parts = result_dir.name.rsplit('_', 1)
+    if len(parts) != 2:
+        continue
+    scenario, policy = parts
+    
+    print(f"\n{scenario} - {policy}")
+    print("-" * 60)
+    
+    try:
+        # Generate cause list
+        output_dir = result_dir / "cause_lists"
+        generator = CauseListGenerator(events_file)
+        cause_list_path = generator.generate_daily_lists(output_dir)
+        
+        # Load and analyze
+        cause_list = pd.read_csv(cause_list_path)
+        
+        # Daily statistics
+        daily_stats = cause_list.groupby('Date').agg({
+            'Case_ID': 'count',
+            'Courtroom_ID': 'nunique',
+            'Sequence_Number': 'max'
+        }).rename(columns={
+            'Case_ID': 'hearings',
+            'Courtroom_ID': 'active_courtrooms',
+            'Sequence_Number': 'max_sequence'
+        })
+        
+        # Overall statistics
+        stats = {
+            'scenario': scenario,
+            'policy': policy,
+            'total_hearings': len(cause_list),
+            'unique_cases': cause_list['Case_ID'].nunique(),
+            'total_days': cause_list['Date'].nunique(),
+            'avg_hearings_per_day': daily_stats['hearings'].mean(),
+            'std_hearings_per_day': daily_stats['hearings'].std(),
+            'min_hearings_per_day': daily_stats['hearings'].min(),
+            'max_hearings_per_day': daily_stats['hearings'].max(),
+            'avg_courtrooms_per_day': daily_stats['active_courtrooms'].mean(),
+            'avg_cases_per_courtroom': daily_stats['hearings'].mean() / daily_stats['active_courtrooms'].mean()
+        }
+        
+        all_stats.append(stats)
+        
+        print(f"  Total hearings: {stats['total_hearings']:,}")
+        print(f"  Unique cases: {stats['unique_cases']:,}")
+        print(f"  Days: {stats['total_days']}")
+        print(f"  Avg hearings/day: {stats['avg_hearings_per_day']:.1f} ± {stats['std_hearings_per_day']:.1f}")
+        print(f"  Avg cases/courtroom: {stats['avg_cases_per_courtroom']:.1f}")
+        
+    except Exception as e:
+        print(f"  ERROR: {e}")
+
+# Convert to DataFrame
+stats_df = pd.DataFrame(all_stats)
+stats_df.to_csv(sweep_dir / "cause_list_statistics.csv", index=False)
+
+print("\n" + "=" * 80)
+print(f"Generated {len(all_stats)} cause lists")
+print(f"Statistics saved to: {sweep_dir / 'cause_list_statistics.csv'}")
+
+# Generate comparative visualizations
+print("\nGenerating visualizations...")
+
+viz_dir = sweep_dir / "visualizations"
+viz_dir.mkdir(exist_ok=True)
+
+# 1. Average daily hearings by policy and scenario
+fig, ax = plt.subplots(figsize=(16, 8))
+
+scenarios = stats_df['scenario'].unique()
+policies = ['fifo', 'age', 'readiness']
+x = range(len(scenarios))
+width = 0.25
+
+for i, policy in enumerate(policies):
+    policy_data = stats_df[stats_df['policy'] == policy].set_index('scenario')
+    values = [policy_data.loc[s, 'avg_hearings_per_day'] if s in policy_data.index else 0 for s in scenarios]
+    
+    label = {
+        'fifo': 'FIFO (Baseline)',
+        'age': 'Age-Based (Baseline)',
+        'readiness': 'Our Algorithm (Readiness)'
+    }[policy]
+    
+    bars = ax.bar([xi + i*width for xi in x], values, width, 
+                   label=label, alpha=0.8, edgecolor='black', linewidth=1.2)
+    
+    # Add value labels
+    for j, v in enumerate(values):
+        if v > 0:
+            ax.text(x[j] + i*width, v + 5, f'{v:.0f}', 
+                   ha='center', va='bottom', fontsize=9)
+
+ax.set_xlabel('Scenario', fontsize=13, fontweight='bold')
+ax.set_ylabel('Average Hearings per Day', fontsize=13, fontweight='bold')
+ax.set_title('Daily Cause List Size: Comparison Across Policies and Scenarios', 
+             fontsize=15, fontweight='bold', pad=20)
+ax.set_xticks([xi + width for xi in x])
+ax.set_xticklabels(scenarios, rotation=45, ha='right')
+ax.legend(fontsize=11)
+ax.grid(axis='y', alpha=0.3)
+
+plt.tight_layout()
+plt.savefig(viz_dir / "cause_list_daily_size_comparison.png", dpi=300, bbox_inches='tight')
+print(f"  Saved: {viz_dir / 'cause_list_daily_size_comparison.png'}")
+
+# 2. Variability (std dev) comparison
+fig, ax = plt.subplots(figsize=(16, 8))
+
+for i, policy in enumerate(policies):
+    policy_data = stats_df[stats_df['policy'] == policy].set_index('scenario')
+    values = [policy_data.loc[s, 'std_hearings_per_day'] if s in policy_data.index else 0 for s in scenarios]
+    
+    label = {
+        'fifo': 'FIFO',
+        'age': 'Age',
+        'readiness': 'Readiness (Ours)'
+    }[policy]
+    
+    bars = ax.bar([xi + i*width for xi in x], values, width, 
+                   label=label, alpha=0.8, edgecolor='black', linewidth=1.2)
+    
+    for j, v in enumerate(values):
+        if v > 0:
+            ax.text(x[j] + i*width, v + 0.5, f'{v:.1f}', 
+                   ha='center', va='bottom', fontsize=9)
+
+ax.set_xlabel('Scenario', fontsize=13, fontweight='bold')
+ax.set_ylabel('Std Dev of Daily Hearings', fontsize=13, fontweight='bold')
+ax.set_title('Cause List Consistency: Lower is More Predictable', 
+             fontsize=15, fontweight='bold', pad=20)
+ax.set_xticks([xi + width for xi in x])
+ax.set_xticklabels(scenarios, rotation=45, ha='right')
+ax.legend(fontsize=11)
+ax.grid(axis='y', alpha=0.3)
+
+plt.tight_layout()
+plt.savefig(viz_dir / "cause_list_variability.png", dpi=300, bbox_inches='tight')
+print(f"  Saved: {viz_dir / 'cause_list_variability.png'}")
+
+# 3. Cases per courtroom efficiency
+fig, ax = plt.subplots(figsize=(16, 8))
+
+for i, policy in enumerate(policies):
+    policy_data = stats_df[stats_df['policy'] == policy].set_index('scenario')
+    values = [policy_data.loc[s, 'avg_cases_per_courtroom'] if s in policy_data.index else 0 for s in scenarios]
+    
+    label = {
+        'fifo': 'FIFO',
+        'age': 'Age',
+        'readiness': 'Readiness (Ours)'
+    }[policy]
+    
+    bars = ax.bar([xi + i*width for xi in x], values, width, 
+                   label=label, alpha=0.8, edgecolor='black', linewidth=1.2)
+    
+    for j, v in enumerate(values):
+        if v > 0:
+            ax.text(x[j] + i*width, v + 0.5, f'{v:.1f}', 
+                   ha='center', va='bottom', fontsize=9)
+
+ax.set_xlabel('Scenario', fontsize=13, fontweight='bold')
+ax.set_ylabel('Avg Cases per Courtroom per Day', fontsize=13, fontweight='bold')
+ax.set_title('Courtroom Load Balance: Cases per Courtroom', 
+             fontsize=15, fontweight='bold', pad=20)
+ax.set_xticks([xi + width for xi in x])
+ax.set_xticklabels(scenarios, rotation=45, ha='right')
+ax.legend(fontsize=11)
+ax.grid(axis='y', alpha=0.3)
+
+plt.tight_layout()
+plt.savefig(viz_dir / "cause_list_courtroom_load.png", dpi=300, bbox_inches='tight')
+print(f"  Saved: {viz_dir / 'cause_list_courtroom_load.png'}")
+
+# 4. Statistical summary table
+fig, ax = plt.subplots(figsize=(14, 10))
+ax.axis('tight')
+ax.axis('off')
+
+# Create summary table
+summary_data = []
+for policy in policies:
+    policy_stats = stats_df[stats_df['policy'] == policy]
+    summary_data.append([
+        {'fifo': 'FIFO', 'age': 'Age', 'readiness': 'Readiness (OURS)'}[policy],
+        f"{policy_stats['avg_hearings_per_day'].mean():.1f}",
+        f"{policy_stats['std_hearings_per_day'].mean():.2f}",
+        f"{policy_stats['avg_cases_per_courtroom'].mean():.1f}",
+        f"{policy_stats['unique_cases'].mean():.0f}",
+        f"{policy_stats['total_hearings'].mean():.0f}"
+    ])
+
+table = ax.table(cellText=summary_data,
+                colLabels=['Policy', 'Avg Hearings/Day', 'Std Dev', 
+                          'Cases/Courtroom', 'Avg Unique Cases', 'Avg Total Hearings'],
+                cellLoc='center',
+                loc='center',
+                colWidths=[0.2, 0.15, 0.15, 0.15, 0.15, 0.15])
+
+table.auto_set_font_size(False)
+table.set_fontsize(12)
+table.scale(1, 3)
+
+# Style header
+for i in range(6):
+    table[(0, i)].set_facecolor('#4CAF50')
+    table[(0, i)].set_text_props(weight='bold', color='white')
+
+# Highlight our algorithm
+table[(3, 0)].set_facecolor('#E8F5E9')
+for i in range(1, 6):
+    table[(3, i)].set_facecolor('#E8F5E9')
+    table[(3, i)].set_text_props(weight='bold')
+
+plt.title('Cause List Statistics Summary: Average Across All Scenarios', 
+          fontsize=14, fontweight='bold', pad=20)
+plt.savefig(viz_dir / "cause_list_summary_table.png", dpi=300, bbox_inches='tight')
+print(f"  Saved: {viz_dir / 'cause_list_summary_table.png'}")
+
+print("\n" + "=" * 80)
+print("CAUSE LIST GENERATION AND ANALYSIS COMPLETE!")
+print(f"All visualizations saved to: {viz_dir}")
+print("=" * 80)