BE-AGT-003: implement multi-slot competition handling

- resolve_slot_competition: selects highest-priority slot as winner,
  marks remaining as Deferred with priority += 1 (capped at 99)
- defer_all_slots: defers all pending slots when agent is not idle
- CompetitionResult dataclass for structured return
- Full test coverage: winner selection, priority bumping, cap, ties,
  empty input, single slot, already-deferred slots

app/services/slot_competition.py

@@ -0,0 +1,125 @@
+"""Multi-slot competition handling — BE-AGT-003.
+
+When multiple slots are pending for an agent at heartbeat time, this
+module resolves the competition:
+
+1. Select the **highest priority** slot for execution.
+2. Mark all other pending slots as ``Deferred``.
+3. Bump ``priority += 1`` on each deferred slot (so deferred slots
+   gradually gain priority and eventually get executed).
+
+Design reference: NEXT_WAVE_DEV_DIRECTION.md §6.3 (Multi-slot competition)
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.calendar import SlotStatus, TimeSlot
+
+
+# Maximum priority cap to prevent unbounded growth
+MAX_PRIORITY = 99
+
+
+@dataclass
+class CompetitionResult:
+    """Outcome of resolving a multi-slot competition.
+
+    Attributes
+    ----------
+    winner : TimeSlot | None
+        The slot selected for execution (highest priority).
+        ``None`` if the input list was empty.
+    deferred : list[TimeSlot]
+        Slots that were marked as ``Deferred`` and had their priority bumped.
+    """
+    winner: Optional[TimeSlot]
+    deferred: list[TimeSlot]
+
+
+def resolve_slot_competition(
+    db: Session,
+    pending_slots: list[TimeSlot],
+) -> CompetitionResult:
+    """Resolve competition among multiple pending slots.
+
+    Parameters
+    ----------
+    db : Session
+        SQLAlchemy database session.  Changes are flushed but not committed
+        — the caller controls the transaction boundary.
+    pending_slots : list[TimeSlot]
+        Actionable slots already filtered and sorted by priority descending
+        (as returned by :func:`agent_heartbeat.get_pending_slots_for_agent`).
+
+    Returns
+    -------
+    CompetitionResult
+        Contains the winning slot (or ``None`` if empty) and the list of
+        deferred slots.
+
+    Notes
+    -----
+    - The input list is assumed to be sorted by priority descending.
+      If two slots share the same priority, the first one in the list wins
+      (stable selection — earlier ``scheduled_at`` or lower id if the
+      heartbeat query doesn't sub-sort, but the caller controls ordering).
+    - Deferred slots have ``priority = min(priority + 1, MAX_PRIORITY)``
+      so they gain urgency over time without exceeding the 0-99 range.
+    - The winner slot is **not** modified by this function — the caller
+      is responsible for setting ``attended``, ``started_at``, ``status``,
+      and transitioning the agent status via ``agent_status.transition_to_busy``.
+    """
+    if not pending_slots:
+        return CompetitionResult(winner=None, deferred=[])
+
+    # The first slot is the winner (highest priority, already sorted)
+    winner = pending_slots[0]
+    deferred: list[TimeSlot] = []
+
+    for slot in pending_slots[1:]:
+        slot.status = SlotStatus.DEFERRED
+        slot.priority = min(slot.priority + 1, MAX_PRIORITY)
+        deferred.append(slot)
+
+    if deferred:
+        db.flush()
+
+    return CompetitionResult(winner=winner, deferred=deferred)
+
+
+def defer_all_slots(
+    db: Session,
+    pending_slots: list[TimeSlot],
+) -> list[TimeSlot]:
+    """Mark ALL pending slots as Deferred (agent is not Idle).
+
+    Used when the agent is busy, exhausted, or otherwise unavailable.
+    Each slot gets ``priority += 1`` (capped at ``MAX_PRIORITY``).
+
+    Parameters
+    ----------
+    db : Session
+        SQLAlchemy database session.
+    pending_slots : list[TimeSlot]
+        Slots to defer.
+
+    Returns
+    -------
+    list[TimeSlot]
+        The deferred slots (same objects, mutated in place).
+    """
+    if not pending_slots:
+        return []
+
+    for slot in pending_slots:
+        if slot.status != SlotStatus.DEFERRED:
+            slot.status = SlotStatus.DEFERRED
+        slot.priority = min(slot.priority + 1, MAX_PRIORITY)
+
+    db.flush()
+    return pending_slots