diff --git a/app/services/slot_competition.py b/app/services/slot_competition.py
new file mode 100644
index 0000000..dfb81a6
--- /dev/null
+++ b/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
diff --git a/tests/test_slot_competition.py b/tests/test_slot_competition.py
new file mode 100644
index 0000000..cc806a9
--- /dev/null
+++ b/tests/test_slot_competition.py
@@ -0,0 +1,164 @@
+"""Tests for BE-AGT-003 — multi-slot competition handling.
+
+Covers:
+  - Winner selection (highest priority)
+  - Remaining slots marked Deferred with priority += 1
+  - Priority capping at MAX_PRIORITY (99)
+  - Empty input edge case
+  - Single slot (no competition)
+  - defer_all_slots when agent is not idle
+"""
+
+import pytest
+from datetime import date, time
+
+from app.models.calendar import SlotStatus, SlotType, TimeSlot
+from app.services.slot_competition import (
+    CompetitionResult,
+    MAX_PRIORITY,
+    defer_all_slots,
+    resolve_slot_competition,
+)
+
+
+def _make_slot(db, user_id: int, *, priority: int, status=SlotStatus.NOT_STARTED) -> TimeSlot:
+    """Helper — create a minimal TimeSlot in the test DB."""
+    slot = TimeSlot(
+        user_id=user_id,
+        date=date(2026, 4, 1),
+        slot_type=SlotType.WORK,
+        estimated_duration=30,
+        scheduled_at=time(9, 0),
+        priority=priority,
+        status=status,
+    )
+    db.add(slot)
+    db.flush()
+    return slot
+
+
+# ---------------------------------------------------------------------------
+# resolve_slot_competition
+# ---------------------------------------------------------------------------
+
+class TestResolveSlotCompetition:
+    """Tests for resolve_slot_competition."""
+
+    def test_empty_input(self, db, seed):
+        result = resolve_slot_competition(db, [])
+        assert result.winner is None
+        assert result.deferred == []
+
+    def test_single_slot_no_competition(self, db, seed):
+        slot = _make_slot(db, 1, priority=50)
+        result = resolve_slot_competition(db, [slot])
+
+        assert result.winner is slot
+        assert result.deferred == []
+        # Winner should NOT be modified
+        assert slot.status == SlotStatus.NOT_STARTED
+        assert slot.priority == 50
+
+    def test_winner_is_first_slot(self, db, seed):
+        """Input is pre-sorted by priority desc; first slot wins."""
+        high = _make_slot(db, 1, priority=80)
+        mid = _make_slot(db, 1, priority=50)
+        low = _make_slot(db, 1, priority=10)
+        slots = [high, mid, low]
+
+        result = resolve_slot_competition(db, slots)
+
+        assert result.winner is high
+        assert len(result.deferred) == 2
+        assert mid in result.deferred
+        assert low in result.deferred
+
+    def test_deferred_slots_status_and_priority(self, db, seed):
+        """Deferred slots get status=DEFERRED and priority += 1."""
+        winner = _make_slot(db, 1, priority=80)
+        loser1 = _make_slot(db, 1, priority=50)
+        loser2 = _make_slot(db, 1, priority=10)
+
+        resolve_slot_competition(db, [winner, loser1, loser2])
+
+        # Winner untouched
+        assert winner.status == SlotStatus.NOT_STARTED
+        assert winner.priority == 80
+
+        # Losers deferred + bumped
+        assert loser1.status == SlotStatus.DEFERRED
+        assert loser1.priority == 51
+
+        assert loser2.status == SlotStatus.DEFERRED
+        assert loser2.priority == 11
+
+    def test_priority_capped_at_max(self, db, seed):
+        """Priority bump should not exceed MAX_PRIORITY."""
+        winner = _make_slot(db, 1, priority=99)
+        at_cap = _make_slot(db, 1, priority=99)
+
+        resolve_slot_competition(db, [winner, at_cap])
+
+        assert at_cap.status == SlotStatus.DEFERRED
+        assert at_cap.priority == MAX_PRIORITY  # stays at 99, not 100
+
+    def test_already_deferred_slots_get_bumped(self, db, seed):
+        """Slots that were already DEFERRED still get priority bumped."""
+        winner = _make_slot(db, 1, priority=90)
+        already_deferred = _make_slot(db, 1, priority=40, status=SlotStatus.DEFERRED)
+
+        result = resolve_slot_competition(db, [winner, already_deferred])
+
+        assert already_deferred.status == SlotStatus.DEFERRED
+        assert already_deferred.priority == 41
+
+    def test_tie_breaking_first_wins(self, db, seed):
+        """When priorities are equal, the first in the list wins."""
+        a = _make_slot(db, 1, priority=50)
+        b = _make_slot(db, 1, priority=50)
+
+        result = resolve_slot_competition(db, [a, b])
+
+        assert result.winner is a
+        assert b in result.deferred
+        assert b.status == SlotStatus.DEFERRED
+
+
+# ---------------------------------------------------------------------------
+# defer_all_slots
+# ---------------------------------------------------------------------------
+
+class TestDeferAllSlots:
+    """Tests for defer_all_slots (agent not idle)."""
+
+    def test_empty_input(self, db, seed):
+        result = defer_all_slots(db, [])
+        assert result == []
+
+    def test_all_slots_deferred(self, db, seed):
+        s1 = _make_slot(db, 1, priority=70)
+        s2 = _make_slot(db, 1, priority=30)
+
+        result = defer_all_slots(db, [s1, s2])
+
+        assert len(result) == 2
+        assert s1.status == SlotStatus.DEFERRED
+        assert s1.priority == 71
+        assert s2.status == SlotStatus.DEFERRED
+        assert s2.priority == 31
+
+    def test_priority_cap_in_defer_all(self, db, seed):
+        s = _make_slot(db, 1, priority=99)
+
+        defer_all_slots(db, [s])
+
+        assert s.priority == MAX_PRIORITY
+
+    def test_already_deferred_still_bumped(self, db, seed):
+        """Even if already DEFERRED, priority still increases."""
+        s = _make_slot(db, 1, priority=45, status=SlotStatus.DEFERRED)
+
+        defer_all_slots(db, [s])
+
+        assert s.status == SlotStatus.DEFERRED
+        assert s.priority == 46