Add configurable task reminders for contract end and invoice overdue

tuttle/app/settings/intent.py

@@ -58,3 +58,27 @@ def get_all(self, prefix: str = None) -> IntentResult:
             was_intent_successful=True,
             data=self._app_db.get_all_settings(prefix=prefix),
         )
+
+    # -- Task reminder settings ------------------------------------------------
+
+    TASK_CONTRACT_END_KEY = "tasks.contract_end_lead_days"
+    TASK_INVOICE_OVERDUE_KEY = "tasks.invoice_overdue_lead_days"
+
+    DEFAULT_CONTRACT_END_LEAD = "30, 0"
+    DEFAULT_INVOICE_OVERDUE_LEAD = "0"
+
+    def get_task_reminders(self) -> IntentResult:
+        return IntentResult(
+            was_intent_successful=True,
+            data={
+                "contract_end_lead_days": self._app_db.get_setting(self.TASK_CONTRACT_END_KEY)
+                or self.DEFAULT_CONTRACT_END_LEAD,
+                "invoice_overdue_lead_days": self._app_db.get_setting(self.TASK_INVOICE_OVERDUE_KEY)
+                or self.DEFAULT_INVOICE_OVERDUE_LEAD,
+            },
+        )
+
+    def save_task_reminders(self, contract_end_lead_days: str, invoice_overdue_lead_days: str) -> IntentResult:
+        self._app_db.set_setting(self.TASK_CONTRACT_END_KEY, contract_end_lead_days)
+        self._app_db.set_setting(self.TASK_INVOICE_OVERDUE_KEY, invoice_overdue_lead_days)
+        return IntentResult(was_intent_successful=True, data=None)

tuttle/app/tasks/generator.py

@@ -8,8 +8,11 @@
 holds.  Business tasks require explicit user action.
 """
 
+import datetime
+
 from sqlmodel import Session, select
 
+from ...app_db import AppDatabase
 from ...model import Client, Contact, Contract, Invoice, Project, Task
 
 # ---------------------------------------------------------------------------
@@ -66,6 +69,24 @@
 ]
 
 
+def _parse_lead_days(raw: str) -> list[int]:
+    """Parse a comma-separated string of non-negative integers."""
+    result = []
+    for part in raw.split(","):
+        part = part.strip()
+        if part.isdigit():
+            result.append(int(part))
+    return sorted(set(result), reverse=True)
+
+
+def _load_lead_days() -> tuple[list[int], list[int]]:
+    """Read task reminder lead-day settings from the app database."""
+    app_db = AppDatabase()
+    contract_raw = app_db.get_setting("tasks.contract_end_lead_days") or "30, 0"
+    overdue_raw = app_db.get_setting("tasks.invoice_overdue_lead_days") or "0"
+    return _parse_lead_days(contract_raw), _parse_lead_days(overdue_raw)
+
+
 # ---------------------------------------------------------------------------
 # Public API
 # ---------------------------------------------------------------------------
@@ -75,7 +96,9 @@ def generate_tasks(session: Session) -> None:
     """Refresh task rows based on current business state."""
     _generate_tutorial_tasks(session)
     _generate_manual_tutorial_tasks(session)
-    _generate_overdue_tasks(session)
+    contract_leads, overdue_leads = _load_lead_days()
+    _generate_overdue_tasks(session, overdue_leads)
+    _generate_contract_end_tasks(session, contract_leads)
     session.commit()
 
 
@@ -124,22 +147,67 @@ def _generate_manual_tutorial_tasks(session: Session) -> None:
 # ---------------------------------------------------------------------------
 
 
-def _generate_overdue_tasks(session: Session) -> None:
+def _generate_overdue_tasks(session: Session, lead_days: list[int]) -> None:
+    if not lead_days:
+        return
     invoices = session.exec(select(Invoice)).all()
+    today = datetime.date.today()
     for inv in invoices:
         if inv.status != "overdue":
             continue
-        # Skip if a reminder already exists for this invoice
         if inv.reminders:
             continue
+        due = inv.effective_due_date
+        if not due:
+            continue
+        days_overdue = (today - due).days
+        for d in lead_days:
+            if days_overdue < d:
+                continue
+            key = f"overdue:{inv.id}" if d == 0 else f"overdue:{inv.id}:{d}"
+            existing = session.exec(select(Task).where(Task.key == key)).first()
+            if not existing:
+                suffix = f" — {d} days overdue" if d > 0 else ""
+                session.add(
+                    Task(
+                        key=key,
+                        title=f"Invoice {inv.number} is overdue{suffix} — send a reminder",
+                        description=f"This invoice was due on {due}.",
+                    )
+                )
 
-        key = f"overdue:{inv.id}"
-        existing = session.exec(select(Task).where(Task.key == key)).first()
-        if not existing:
-            session.add(
-                Task(
-                    key=key,
-                    title=f"Invoice {inv.number} is overdue — send a reminder",
-                    description=f"This invoice was due on {inv.effective_due_date}.",
+
+# ---------------------------------------------------------------------------
+# Contract-end tasks
+# ---------------------------------------------------------------------------
+
+
+def _generate_contract_end_tasks(session: Session, lead_days: list[int]) -> None:
+    if not lead_days:
+        return
+    contracts = session.exec(select(Contract)).all()
+    today = datetime.date.today()
+    for c in contracts:
+        if not c.end_date or c.is_completed:
+            continue
+        days_until_end = (c.end_date - today).days
+        for d in lead_days:
+            if days_until_end > d:
+                continue
+            key = f"contract_ending:{c.id}:{d}"
+            existing = session.exec(select(Task).where(Task.key == key)).first()
+            if not existing:
+                if d == 0:
+                    title = f'Contract "{c.title}" ends today'
+                else:
+                    title = f'Contract "{c.title}" ends in {d} days'
+                session.add(
+                    Task(
+                        key=key,
+                        title=title,
+                        description=f"End date: {c.end_date}.",
+                    )
                 )
-            )
+            elif existing.status == "pending" and c.is_completed:
+                existing.status = "done"
+                session.add(existing)

tuttle_tests/test_tasks.py

@@ -2,11 +2,12 @@
 
 import datetime
 from decimal import Decimal
+from unittest.mock import patch
 
 import pytest
 import sqlmodel
 
-from tuttle.app.tasks.generator import generate_tasks
+from tuttle.app.tasks.generator import _parse_lead_days, generate_tasks
 from tuttle.model import (
     Address,
     Client,
@@ -28,6 +29,25 @@ def session():
         yield s
 
 
+# ---------------------------------------------------------------------------
+# _parse_lead_days
+# ---------------------------------------------------------------------------
+
+
+class TestParseLeadDays:
+    def test_simple(self):
+        assert _parse_lead_days("30, 7, 0") == [30, 7, 0]
+
+    def test_deduplicates(self):
+        assert _parse_lead_days("7, 7, 0") == [7, 0]
+
+    def test_ignores_garbage(self):
+        assert _parse_lead_days("30, abc, , -1, 0") == [30, 0]
+
+    def test_empty_string(self):
+        assert _parse_lead_days("") == []
+
+
 # ---------------------------------------------------------------------------
 # Tutorial tasks
 # ---------------------------------------------------------------------------
@@ -200,3 +220,140 @@ def test_does_not_create_for_paid_invoice(self, session, overdue_invoice):
         generate_tasks(session)
         task = session.exec(sqlmodel.select(Task).where(Task.key == f"overdue:{overdue_invoice.id}")).first()
         assert task is None
+
+
+# ---------------------------------------------------------------------------
+# Contract-end tasks
+# ---------------------------------------------------------------------------
+
+
+class TestContractEndTasks:
+    @pytest.fixture
+    def ending_contract(self, session):
+        client = Client(name="EndCorp")
+        contract = Contract(
+            title="Ending Contract",
+            client=client,
+            rate=Decimal("100"),
+            currency="EUR",
+            unit=TimeUnit.hour,
+            units_per_workday=8,
+            term_of_payment=14,
+            billing_cycle=Cycle.monthly,
+            start_date=datetime.date(2025, 1, 1),
+            end_date=datetime.date.today() + datetime.timedelta(days=10),
+        )
+        session.add(contract)
+        session.commit()
+        return contract
+
+    def test_creates_task_within_lead_window(self, session, ending_contract):
+        """Contract ending in 10 days triggers a 30-day lead task."""
+        generate_tasks(session)
+        task = session.exec(sqlmodel.select(Task).where(Task.key == f"contract_ending:{ending_contract.id}:30")).first()
+        assert task is not None
+        assert "Ending Contract" in task.title
+
+    def test_no_task_outside_lead_window(self, session):
+        """Contract ending in 60 days should not trigger 30-day lead task."""
+        client = Client(name="FarCorp")
+        contract = Contract(
+            title="Far Contract",
+            client=client,
+            rate=Decimal("100"),
+            currency="EUR",
+            unit=TimeUnit.hour,
+            units_per_workday=8,
+            term_of_payment=14,
+            billing_cycle=Cycle.monthly,
+            start_date=datetime.date(2025, 1, 1),
+            end_date=datetime.date.today() + datetime.timedelta(days=60),
+        )
+        session.add(contract)
+        session.commit()
+        generate_tasks(session)
+        tasks = session.exec(
+            sqlmodel.select(Task).where(Task.key.startswith("contract_ending:"))  # type: ignore[union-attr]
+        ).all()
+        assert len(tasks) == 0
+
+    def test_no_task_for_completed_contract(self, session, ending_contract):
+        ending_contract.is_completed = True
+        session.add(ending_contract)
+        session.commit()
+        generate_tasks(session)
+        tasks = session.exec(
+            sqlmodel.select(Task).where(Task.key.startswith("contract_ending:"))  # type: ignore[union-attr]
+        ).all()
+        assert len(tasks) == 0
+
+    def test_no_task_for_open_ended_contract(self, session):
+        client = Client(name="OpenCorp")
+        contract = Contract(
+            title="Open Contract",
+            client=client,
+            rate=Decimal("100"),
+            currency="EUR",
+            unit=TimeUnit.hour,
+            units_per_workday=8,
+            term_of_payment=14,
+            billing_cycle=Cycle.monthly,
+            start_date=datetime.date(2025, 1, 1),
+        )
+        session.add(contract)
+        session.commit()
+        generate_tasks(session)
+        tasks = session.exec(
+            sqlmodel.select(Task).where(Task.key.startswith("contract_ending:"))  # type: ignore[union-attr]
+        ).all()
+        assert len(tasks) == 0
+
+    def test_idempotent(self, session, ending_contract):
+        generate_tasks(session)
+        generate_tasks(session)
+        tasks = session.exec(sqlmodel.select(Task).where(Task.key == f"contract_ending:{ending_contract.id}:30")).all()
+        assert len(tasks) == 1
+
+    @patch("tuttle.app.tasks.generator._load_lead_days")
+    def test_custom_lead_days(self, mock_load, session):
+        mock_load.return_value = ([7], [0])
+        client = Client(name="CustomCorp")
+        contract = Contract(
+            title="Custom Lead",
+            client=client,
+            rate=Decimal("100"),
+            currency="EUR",
+            unit=TimeUnit.hour,
+            units_per_workday=8,
+            term_of_payment=14,
+            billing_cycle=Cycle.monthly,
+            start_date=datetime.date(2025, 1, 1),
+            end_date=datetime.date.today() + datetime.timedelta(days=5),
+        )
+        session.add(contract)
+        session.commit()
+        generate_tasks(session)
+        task = session.exec(sqlmodel.select(Task).where(Task.key == f"contract_ending:{contract.id}:7")).first()
+        assert task is not None
+
+    def test_today_end_date_task(self, session):
+        """Contract ending today triggers the 0-day lead task."""
+        client = Client(name="TodayCorp")
+        contract = Contract(
+            title="Today Contract",
+            client=client,
+            rate=Decimal("100"),
+            currency="EUR",
+            unit=TimeUnit.hour,
+            units_per_workday=8,
+            term_of_payment=14,
+            billing_cycle=Cycle.monthly,
+            start_date=datetime.date(2025, 1, 1),
+            end_date=datetime.date.today(),
+        )
+        session.add(contract)
+        session.commit()
+        generate_tasks(session)
+        task = session.exec(sqlmodel.select(Task).where(Task.key == f"contract_ending:{contract.id}:0")).first()
+        assert task is not None
+        assert "ends today" in task.title