donna.scheduling.weekly_planner

donna.scheduling.weekly_planner

Weekly planning session — Phase 2.

Fires Monday mornings at a configurable time. Assembles the week's backlog tasks (filtered by deadline proximity, priority, or staleness), resolves dependency order, dry-runs slot assignments, and posts a plan to Discord for user confirmation.

On user reply "confirm" → applies the plan (creates calendar events). On user reply "skip [title]" → removes that task and re-posts.

Pending proposals are held in memory — an in-process dict is sufficient for Phase 2 with a single user.

See docs/scheduling.md and docs/agents.md.

logger module-attribute

logger = get_logger()

WeeklyPlanner

WeeklyPlanner(db: Database, scheduler: Scheduler, recalculator: PriorityRecalculator, service: NotificationService, calendar_client: GoogleCalendarClient, calendar_id: str, user_id: str, fire_hour: int = _DEFAULT_FIRE_HOUR, fire_minute: int = _DEFAULT_FIRE_MINUTE)

Assembles and proposes a weekly schedule every Monday morning.

Usage

planner = WeeklyPlanner(db, scheduler, recalculator, service, calendar_client, calendar_id, user_id) asyncio.create_task(planner.run())

Source code in src/donna/scheduling/weekly_planner.py

def __init__(
    self,
    db: Database,
    scheduler: Scheduler,
    recalculator: PriorityRecalculator,
    service: NotificationService,
    calendar_client: GoogleCalendarClient,
    calendar_id: str,
    user_id: str,
    fire_hour: int = _DEFAULT_FIRE_HOUR,
    fire_minute: int = _DEFAULT_FIRE_MINUTE,
) -> None:
    self._db = db
    self._scheduler = scheduler
    self._recalculator = recalculator
    self._service = service
    self._calendar_client = calendar_client
    self._calendar_id = calendar_id
    self._user_id = user_id
    self._fire_hour = fire_hour
    self._fire_minute = fire_minute

    # In-memory pending proposals: {proposal_id: {"slots": [...], "tasks": [...]}}
    self._pending: dict[str, dict[str, Any]] = {}

run async

run() -> None

Sleep until the next Monday at fire_hour:fire_minute, fire, repeat.

Source code in src/donna/scheduling/weekly_planner.py

async def run(self) -> None:
    """Sleep until the next Monday at fire_hour:fire_minute, fire, repeat."""
    logger.info(
        "weekly_planner_started",
        fire_hour=self._fire_hour,
        fire_minute=self._fire_minute,
        user_id=self._user_id,
    )

    while True:
        now = datetime.now(tz=UTC)
        next_fire = _next_monday_fire(now, self._fire_hour, self._fire_minute)
        wait_seconds = (next_fire - now).total_seconds()

        logger.info(
            "weekly_planner_waiting",
            next_fire=next_fire.isoformat(),
            wait_seconds=int(wait_seconds),
        )
        await asyncio.sleep(max(wait_seconds, 0))

        try:
            await self._fire(datetime.now(tz=UTC))
        except Exception:
            logger.exception("weekly_planner_fire_failed", user_id=self._user_id)

handle_plan_reply async

handle_plan_reply(message_text: str, now: datetime | None = None) -> bool

Handle a user reply to a weekly plan proposal.

Returns True if the reply was handled (matched a pending proposal), False if no pending proposal exists.

Source code in src/donna/scheduling/weekly_planner.py

async def handle_plan_reply(
    self, message_text: str, now: datetime | None = None
) -> bool:
    """Handle a user reply to a weekly plan proposal.

    Returns True if the reply was handled (matched a pending proposal),
    False if no pending proposal exists.
    """
    now = now or datetime.now(tz=UTC)

    # Expire old proposals.
    expired = [pid for pid, p in self._pending.items() if p["expires_at"] < now]
    for pid in expired:
        del self._pending[pid]

    if not self._pending:
        return False

    # Use the most recent pending proposal.
    proposal_id = max(self._pending)
    proposal = self._pending[proposal_id]
    text = message_text.strip().lower()

    if text == "confirm":
        await self._apply_proposal(proposal_id)
        del self._pending[proposal_id]
        return True

    if text.startswith("skip "):
        skip_title = message_text.strip()[5:].strip().strip("'\"")
        tasks = proposal["tasks"]
        slots = proposal["slots"]
        new_pairs = [
            (t, s) for t, s in zip(tasks, slots, strict=False)
            if skip_title.lower() not in t.title.lower()
        ]
        proposal["tasks"] = [t for t, _ in new_pairs]
        proposal["slots"] = [s for _, s in new_pairs]

        # Re-post updated proposal.
        msg = self._format_proposal(
            proposal_id, new_pairs, now
        )
        await self._service.dispatch(
            notification_type="weekly_plan",
            content=f"Updated plan (skipped '{skip_title}'):\n{msg}",
            channel="tasks",
            priority=4,
        )
        return True

    return False