diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml
index 0dbf844..43f33a3 100644
--- a/.github/workflows/checks.yml
+++ b/.github/workflows/checks.yml
@@ -20,9 +20,8 @@ jobs:
       - uses: actions/checkout@v4
       - name: Build and test
         run: |
-          make -C src CFLAGS='-std=c11 -O2 -Wall -Wextra -Wpedantic -Werror' test
-          python3 tests/test_controller.py
-          python3 tests/test_metrics.py
+          make -C src CFLAGS='-std=c11 -O2 -Wall -Wextra -Wpedantic -Werror'
+          make test
       - name: Sanitizer checks
         if: runner.os == 'Linux'
         run: |
@@ -30,22 +29,35 @@ jobs:
           make -C src CC=clang CFLAGS='-std=c11 -O1 -g -Wall -Wextra -Wpedantic -Werror -fsanitize=address,undefined -fno-omit-frame-pointer' LDFLAGS='-fsanitize=address,undefined' test
 
   network:
+    strategy:
+      fail-fast: false
+      matrix:
+        suite: [matrix, concurrent]
     runs-on: ubuntu-22.04
-    timeout-minutes: 20
+    timeout-minutes: 45
     steps:
       - uses: actions/checkout@v4
       - name: Install network lab
         run: sudo bash scripts/setup_env.sh
       - name: Exercise normal, loss and link failure
+        if: matrix.suite == 'matrix'
         run: |
           sudo .venv/bin/python scripts/run.py experiment --repeats 1 --output results/ci
-          sudo .venv/bin/python scripts/run.py experiment --scenarios link-failure --modes failover --flows 2 --repeats 1 --output results/ci-concurrent
-          .venv/bin/python scripts/run.py plot results/ci/metrics.csv --output results/ci/plots
+          sudo .venv/bin/python scripts/run.py plot results/ci/metrics.csv --output results/ci/plots
+      - name: Exercise concurrent failover
+        if: matrix.suite == 'concurrent'
+        run: sudo .venv/bin/python scripts/run.py experiment --scenarios link-failure --modes failover --flows 2 --repeats 1 --output results/ci-concurrent
       - name: Save logs and measurements
         if: always()
         uses: actions/upload-artifact@v4
         with:
-          name: network-results
+          name: network-results-${{ matrix.suite }}
+          # Retain measurements and diagnostics, excluding temporary payloads.
           path: |
-            results/ci/
-            results/ci-concurrent/
+            results/ci*/**/*.log
+            results/ci*/**/*.json
+            results/ci*/**/*.jsonl
+            results/ci*/**/*.csv
+            results/ci*/**/*.svg
+            results/ci*/**/*.md
+            results/ci*/**/*.txt
diff --git a/Makefile b/Makefile
index cb5000c..b052bd0 100644
--- a/Makefile
+++ b/Makefile
@@ -9,6 +9,8 @@ clean:
 test: all
 	python3 tests/test_protocol.py
 	python3 tests/test_transport.py
+	python3 tests/test_multipath.py
+	python3 tests/test_probe_schedule.py
 	python3 tests/test_probes.py
 	python3 tests/test_openflow.py
 	python3 tests/test_controller.py
diff --git a/controller/failover.py b/controller/failover.py
index 656bfbf..16b53d9 100644
--- a/controller/failover.py
+++ b/controller/failover.py
@@ -1,3 +1,4 @@
+
 """Proactive OpenFlow 1.3 routes with port-status-driven path recovery.
 
 Run with osken-manager --ofp-tcp-listen-port 6653 controller/failover.py.
@@ -32,11 +33,15 @@
 
 # The manager may load this file by path without adding the repository root.
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
 from controller.routing import PathPolicy, SWITCHES, forwarding_rules
 from controller.probes import ProbeTracker, ENDPOINTS, ETHERTYPE
+from controller.probe_schedule import AdaptiveProbeSchedule
+from controller.multipath import eligible_paths
 
 
 COOKIE = 0x5255445000000001
+MULTIPATH_COOKIE = 0x5255445000000010
 
 
 class FailoverController(AppBase):
@@ -44,7 +49,12 @@ class FailoverController(AppBase):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.policy = PathPolicy(os.environ.get("RUDP_MODE", "failover"), os.environ.get("RUDP_INITIAL_PATH", "upper"))
+
+        self.policy = PathPolicy(
+            os.environ.get("RUDP_MODE", "failover"),
+            os.environ.get("RUDP_INITIAL_PATH", "upper"),
+        )
+
         self.datapaths = {}
         self.discovered = set()
         self.started = False
@@ -53,135 +63,337 @@ def __init__(self, *args, **kwargs):
         self.pending = {}
         self.failed_generation = False
         self.cause = "startup"
+
         event_path = os.environ.get("RUDP_EVENT_LOG")
-        self.event_file = open(event_path, "a", buffering=1) if event_path else None
+        self.event_file = (
+            open(event_path, "a", buffering=1) if event_path else None
+        )
+
         self._event("controller_started", mode=self.policy.mode)
+
         self.probes = ProbeTracker()
+
+        self.probe_schedules = {
+            key: AdaptiveProbeSchedule() for key in ENDPOINTS
+        }
+
+        # Each path/direction has its own next-probe deadline.
+        self.next_probe_ns = {key: 0 for key in ENDPOINTS}
+
+        self.multipath_paths = frozenset()
         self.last_route_change_ns = 0
         self.monitor = hub.spawn(self._monitor)
 
     def _refresh_health(self, now):
         self.policy.health = self.probes.scores(now)
 
+        # Adapt each direction's schedule according to its own path's health.
+        # A degraded upper path should not automatically accelerate the lower
+        # path's probes.
+        for (path, direction), schedule in self.probe_schedules.items():
+            path_score = self.policy.health.get(path)
+            degraded = path_score is None or path_score > 0.10
+            schedule.set_degraded(degraded, now)
+
     def _monitor(self):
         while True:
             now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
+
+            # Record timed-out probes and make their next probe eligible now.
             for observation in self.probes.expire(now):
                 self._event("probe_result", **observation)
+
+                key = (
+                    observation["path"],
+                    observation["direction"],
+                )
+
+                self.probe_schedules[key].record_result(False, now)
+                self.next_probe_ns[key] = now
+
             self._refresh_health(now)
+
             if self.started:
-                self._event("path_health", scores=self.policy.health,
-                            source="bidirectional_ethernet_probes", samples=self.probes.samples_as_json(),
-                            hysteresis=.02, hold_down_seconds=2)
+                self._event(
+                    "path_health",
+                    scores=self.policy.health,
+                    source="bidirectional_ethernet_probes",
+                    samples=self.probes.samples_as_json(),
+                    hysteresis=0.02,
+                    hold_down_seconds=2,
+                )
+
                 self._reconcile("health_update")
-                for (path, direction), (source, out_port, _, _) in ENDPOINTS.items():
+
+                # Send only the probes whose individual deadlines have arrived.
+                for (
+                    (path, direction),
+                    (source, out_port, _, _),
+                ) in ENDPOINTS.items():
+
                     if not self.policy.available(path):
                         continue
+
                     dp = self.datapaths.get(source)
+
                     if dp is None:
                         continue
+
+                    key = (path, direction)
+
+                    if now < self.next_probe_ns[key]:
+                        continue
+
                     frame = self.probes.send(path, direction, now)
+
                     dp.send_msg(dp.ofproto_parser.OFPPacketOut(
-                        datapath=dp, buffer_id=dp.ofproto.OFP_NO_BUFFER,
+                        datapath=dp,
+                        buffer_id=dp.ofproto.OFP_NO_BUFFER,
                         in_port=dp.ofproto.OFPP_CONTROLLER,
-                        actions=[dp.ofproto_parser.OFPActionOutput(out_port)], data=frame))
-            hub.sleep(.25)
+                        actions=[
+                            dp.ofproto_parser.OFPActionOutput(out_port)
+                        ],
+                        data=frame,
+                    ))
+
+                    interval_ns = int(
+                        self.probe_schedules[key].interval_seconds(now)
+                        * 1_000_000_000
+                    )
+
+                    self.next_probe_ns[key] = now + interval_ns
+
+            # Wake at the next probe deadline or pending-probe timeout.
+            # Cap sleep at 250 ms so health and controller state are refreshed.
+            deadlines = []
+
+            if self.started:
+                for (
+                    (path, direction),
+                    (source, _, _, _),
+                ) in ENDPOINTS.items():
+
+                    if not self.policy.available(path):
+                        continue
+
+                    if source not in self.datapaths:
+                        continue
+
+                    deadline = self.next_probe_ns[(path, direction)]
+
+                    if deadline > now:
+                        deadlines.append(deadline)
+
+                for _, _, sent_ns in self.probes.pending.values():
+                    timeout_deadline = sent_ns + self.probes.timeout_ns
+
+                    if timeout_deadline > now:
+                        deadlines.append(timeout_deadline)
+
+            if deadlines:
+                sleep_ns = max(1_000_000, min(deadlines) - now)
+                interval = min(sleep_ns / 1_000_000_000, 0.25)
+            else:
+                interval = 0.25
+
+            hub.sleep(interval)
 
     @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
     def probe_received(self, event):
         msg = event.msg
         now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
-        observation = self.probes.receive(bytes(msg.data), msg.datapath.id, msg.match.get("in_port"), now)
+
+        observation = self.probes.receive(
+            bytes(msg.data),
+            msg.datapath.id,
+            msg.match.get("in_port"),
+            now,
+        )
+
         if observation:
             self._event("probe_result", **observation)
+
+            key = (
+                observation["path"],
+                observation["direction"],
+            )
+
+            schedule = self.probe_schedules[key]
+            schedule.record_result(True, now)
+
+            # If the new result calls for faster probing, bring the deadline
+            # forward. Never delay a probe that is already due.
+            interval_ns = int(
+                schedule.interval_seconds(now) * 1_000_000_000
+            )
+
+            proposed_deadline = now + interval_ns
+
+            self.next_probe_ns[key] = min(
+                self.next_probe_ns[key],
+                proposed_deadline,
+            )
+
             self._refresh_health(now)
 
     def _event(self, event, **fields):
-        record = dict(event=event, monotonic_ns=time.clock_gettime_ns(time.CLOCK_MONOTONIC),
-                      wall_time=datetime.datetime.now(datetime.timezone.utc).isoformat(),
-                      **fields)
+        record = dict(
+            event=event,
+            monotonic_ns=time.clock_gettime_ns(time.CLOCK_MONOTONIC),
+            wall_time=datetime.datetime.now(
+                datetime.timezone.utc
+            ).isoformat(),
+            **fields,
+        )
+
         line = json.dumps(record, sort_keys=True)
         self.logger.info(line)
+
         if self.event_file:
             self.event_file.write(line + "\n")
 
     @set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)
     def features(self, event):
         dp = event.msg.datapath
+
         if dp.id not in SWITCHES:
             self._event("unexpected_switch", dpid=dp.id)
             return
+
         self.probes.invalidate_switch(dp.id)
+
         self.datapaths[dp.id] = dp
         self.policy.connected.add(dp.id)
         self.discovered.discard(dp.id)
+
         parser, ofp = dp.ofproto_parser, dp.ofproto
-        # This is a dedicated lab controller. Clear stale flows on reconnect,
-        # then keep table-miss as drop rather than learning/flooding the loop.
+
+        # Clear stale flows on reconnect, then keep table-miss as drop.
         dp.send_msg(parser.OFPFlowMod(
-            datapath=dp, command=ofp.OFPFC_DELETE, table_id=ofp.OFPTT_ALL,
-            out_port=ofp.OFPP_ANY, out_group=ofp.OFPG_ANY,
-            match=parser.OFPMatch()))
+            datapath=dp,
+            command=ofp.OFPFC_DELETE,
+            table_id=ofp.OFPTT_ALL,
+            out_port=ofp.OFPP_ANY,
+            out_group=ofp.OFPG_ANY,
+            match=parser.OFPMatch(),
+        ))
+
         dp.send_msg(parser.OFPFlowMod(
-            datapath=dp, cookie=COOKIE, priority=0,
-            match=parser.OFPMatch(), instructions=[]))
-        # Probe rules are independent of the chosen application path. Transit
-        # crosses the branch; edge ingress punts to the controller, never hosts.
+            datapath=dp,
+            cookie=COOKIE,
+            priority=0,
+            match=parser.OFPMatch(),
+            instructions=[],
+        ))
+
+        # Probe rules are independent of the selected application path.
         probe_ports = (1, 2) if dp.id in (2, 3, 4) else (2, 3)
+
         for in_port in probe_ports:
-            out_port = 3 - in_port if dp.id in (2, 3) else ofp.OFPP_CONTROLLER
-            action = parser.OFPActionOutput(out_port, ofp.OFPCML_NO_BUFFER)
+            out_port = (
+                3 - in_port if dp.id in (2, 3)
+                else ofp.OFPP_CONTROLLER
+            )
+
+            action = parser.OFPActionOutput(
+                out_port,
+                ofp.OFPCML_NO_BUFFER,
+            )
+
             dp.send_msg(parser.OFPFlowMod(
-                datapath=dp, cookie=COOKIE, priority=200,
-                match=parser.OFPMatch(in_port=in_port, eth_type=ETHERTYPE),
-                instructions=[parser.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [action])]))
+                datapath=dp,
+                cookie=COOKIE,
+                priority=200,
+                match=parser.OFPMatch(
+                    in_port=in_port,
+                    eth_type=ETHERTYPE,
+                ),
+                instructions=[
+                    parser.OFPInstructionActions(
+                        ofp.OFPIT_APPLY_ACTIONS,
+                        [action],
+                    )
+                ],
+            ))
+
         dp.send_msg(parser.OFPPortDescStatsRequest(dp, 0))
         self._event("switch_connected", dpid=dp.id)
 
     @staticmethod
     def _port_up(dp, port):
         ofp = dp.ofproto
-        return not (port.state & ofp.OFPPS_LINK_DOWN
-                    or port.config & ofp.OFPPC_PORT_DOWN)
+
+        return not (
+            port.state & ofp.OFPPS_LINK_DOWN
+            or port.config & ofp.OFPPC_PORT_DOWN
+        )
 
     @set_ev_cls(ofp_event.EventOFPPortDescStatsReply, MAIN_DISPATCHER)
     def port_description(self, event):
         msg, dp = event.msg, event.msg.datapath
+
         if dp.id not in self.datapaths:
             return
+
         for port in msg.body:
-            self.policy.ports[(dp.id, port.port_no)] = self._port_up(dp, port)
+            self.policy.ports[(dp.id, port.port_no)] = (
+                self._port_up(dp, port)
+            )
+
         if msg.flags & dp.ofproto.OFPMPF_REPLY_MORE:
             return
+
         self.discovered.add(dp.id)
-        self._reconcile("startup" if not self.started else "switch_up", force=True)
+
+        self._reconcile(
+            "startup" if not self.started else "switch_up",
+            force=True,
+        )
 
     @set_ev_cls(ofp_event.EventOFPPortStatus, MAIN_DISPATCHER)
     def port_status(self, event):
         msg, dp = event.msg, event.msg.datapath
+
         if dp.id not in self.datapaths:
             return
+
         key = (dp.id, msg.desc.port_no)
-        up = msg.reason != dp.ofproto.OFPPR_DELETE and self._port_up(dp, msg.desc)
+
+        up = (
+            msg.reason != dp.ofproto.OFPPR_DELETE
+            and self._port_up(dp, msg.desc)
+        )
+
         previous = self.policy.ports.get(key)
         self.policy.ports[key] = up
+
         if previous == up:
             return
+
         cause = "link_up" if up else "link_down"
+
         self.probes.invalidate_port(dp.id, msg.desc.port_no)
-        self._refresh_health(time.clock_gettime_ns(time.CLOCK_MONOTONIC))
+
+        self._refresh_health(
+            time.clock_gettime_ns(time.CLOCK_MONOTONIC)
+        )
+
         self._event(cause, dpid=dp.id, port=msg.desc.port_no)
         self._reconcile(cause)
 
     @set_ev_cls(ofp_event.EventOFPStateChange, DEAD_DISPATCHER)
     def disconnected(self, event):
         dp = event.datapath
+
         if self.datapaths.get(dp.id) is not dp:
             return
+
         self.probes.invalidate_switch(dp.id)
+
         del self.datapaths[dp.id]
         self.policy.connected.discard(dp.id)
         self.discovered.discard(dp.id)
+
         self._event("switch_down", dpid=dp.id)
         self._reconcile("switch_down", force=True)
 
@@ -189,44 +401,155 @@ def _reconcile(self, cause, force=False):
         if not self.started:
             if self.discovered != SWITCHES:
                 return
+
             self.started = True
+
         path = self.policy.choose(self.path)
-        if not force and path == self.path:
+
+        multipath_paths = eligible_paths(self.policy, path)
+
+        if not force and path == self.path and multipath_paths == self.multipath_paths:
             return
+
         now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
+
         if cause == "health_update" and path != self.path:
-            if self.pending or now - self.last_route_change_ns < 2_000_000_000:
-                self._event("route_decision", selected=self.path, candidate=path, reason="hold_down_or_barrier",
-                            health=self.policy.health)
+            if (
+                self.pending
+                or now - self.last_route_change_ns < 2_000_000_000
+            ):
+                self._event(
+                    "route_decision",
+                    selected=self.path,
+                    candidate=path,
+                    reason="hold_down_or_barrier",
+                    health=self.policy.health,
+                )
                 return
-        self.last_route_change_ns = now
-        self._event("route_decision", previous=self.path, selected=path, reason=cause,
-                    health=self.policy.health)
+
+        if path != self.path:
+            self.last_route_change_ns = now
+        self.multipath_paths = multipath_paths
+
+        self._event(
+            "route_decision",
+            previous=self.path,
+            selected=path,
+            reason=cause,
+            health=self.policy.health,
+        )
+
         self.path = path
         self.generation += 1
         self.cause = cause
         self.pending.clear()
         self.failed_generation = False
-        self._event("route_installing", path=path, cause=cause,
-                    generation=self.generation, health=self.policy.health)
+
+        self._event(
+            "route_installing",
+            path=path,
+            cause=cause,
+            generation=self.generation,
+            health=self.policy.health,
+        )
+
         rules = forwarding_rules(path)
-        # Each ADD replaces the previous identical match/priority atomically
-        # within a switch. During an update, asymmetrical forward/reverse paths
-        # still reach their host: branch ingress never sends into another branch.
+
+        # Install the selected route and remove stale multipath rules.
         for dpid, dp in sorted(self.datapaths.items()):
             parser, ofp = dp.ofproto_parser, dp.ofproto
+
             for rule in rules[dpid]:
-                actions = ([] if rule.out_port is None else
-                           [parser.OFPActionOutput(rule.out_port)])
-                instructions = [parser.OFPInstructionActions(
-                    ofp.OFPIT_APPLY_ACTIONS, actions)]
+                actions = (
+                    [] if rule.out_port is None
+                    else [parser.OFPActionOutput(rule.out_port)]
+                )
+
+                instructions = [
+                    parser.OFPInstructionActions(
+                        ofp.OFPIT_APPLY_ACTIONS,
+                        actions,
+                    )
+                ]
+
                 dp.send_msg(parser.OFPFlowMod(
-                    datapath=dp, cookie=COOKIE, priority=100,
-                    match=parser.OFPMatch(in_port=rule.in_port,
-                                          eth_type=rule.eth_type),
-                    instructions=instructions))
+                    datapath=dp,
+                    cookie=COOKIE,
+                    priority=100,
+                    match=parser.OFPMatch(
+                        in_port=rule.in_port,
+                        eth_type=rule.eth_type,
+                    ),
+                    instructions=instructions,
+                ))
+
+            # Remove this controller's previous DSCP-specific rules.
+            dp.send_msg(parser.OFPFlowMod(
+                datapath=dp,
+                cookie=MULTIPATH_COOKIE,
+                cookie_mask=0xffffffffffffffff,
+                table_id=0,
+                command=ofp.OFPFC_DELETE,
+                out_port=ofp.OFPP_ANY,
+                out_group=ofp.OFPG_ANY,
+                match=parser.OFPMatch(),
+            ))
+
+            # DATA uses DSCP 8 for upper and DSCP 16 for lower.
+            multipath_rules = {
+                "upper": (
+                    8,
+                    ((1, 1, 2), (2, 1, 2), (4, 1, 3)),
+                ),
+                "lower": (
+                    16,
+                    ((1, 1, 3), (3, 1, 2), (4, 2, 3)),
+                ),
+            }
+
+            for path_name, (dscp, entries) in multipath_rules.items():
+                details = self.policy.availability_details(path_name)
+
+                self._event(
+                    "multipath_path_check",
+                    dpid=dpid,
+                    path=path_name,
+                    dscp=dscp,
+                    available=details["available"],
+                    missing_switches=details["missing_switches"],
+                    unavailable_ports=details["unavailable_ports"],
+                )
+
+                # Health changes can remove a branch even while its ports stay up.
+                if path_name not in multipath_paths:
+                    continue
+
+                for switch_id, ingress, output in entries:
+                    if switch_id != dpid:
+                        continue
+
+                    instructions = [
+                        parser.OFPInstructionActions(
+                            ofp.OFPIT_APPLY_ACTIONS,
+                            [parser.OFPActionOutput(output)],
+                        )
+                    ]
+
+                    dp.send_msg(parser.OFPFlowMod(
+                        datapath=dp,
+                        cookie=MULTIPATH_COOKIE,
+                        priority=150,
+                        match=parser.OFPMatch(
+                            in_port=ingress,
+                            eth_type=0x0800,
+                            ip_dscp=dscp,
+                        ),
+                        instructions=instructions,
+                    ))
+
             barrier = parser.OFPBarrierRequest(dp)
             dp.set_xid(barrier)
+
             self.pending[(dpid, barrier.xid)] = self.generation
             dp.send_msg(barrier)
 
@@ -234,20 +557,41 @@ def _reconcile(self, cause, force=False):
     def barrier_reply(self, event):
         key = (event.msg.datapath.id, event.msg.xid)
         generation = self.pending.pop(key, None)
-        if generation != self.generation or self.pending or self.failed_generation:
+
+        if (
+            generation != self.generation
+            or self.pending
+            or self.failed_generation
+        ):
             return
-        self._event("route_ready" if self.path else "route_unavailable",
-                    path=self.path, cause=self.cause, generation=self.generation)
 
-    @set_ev_cls(ofp_event.EventOFPErrorMsg, [CONFIG_DISPATCHER, MAIN_DISPATCHER])
+        self._event(
+            "route_ready" if self.path else "route_unavailable",
+            path=self.path,
+            cause=self.cause,
+            generation=self.generation,
+        )
+
+    @set_ev_cls(
+        ofp_event.EventOFPErrorMsg,
+        [CONFIG_DISPATCHER, MAIN_DISPATCHER],
+    )
     def openflow_error(self, event):
         msg = event.msg
         self.failed_generation = True
-        self._event("openflow_error", dpid=msg.datapath.id,
-                    error_type=msg.type, code=msg.code, xid=msg.xid)
+
+        self._event(
+            "openflow_error",
+            dpid=msg.datapath.id,
+            error_type=msg.type,
+            code=msg.code,
+            xid=msg.xid,
+        )
 
     def close(self):
         hub.kill(self.monitor)
+
         if self.event_file:
             self.event_file.close()
+
         super().close()
diff --git a/controller/multipath.py b/controller/multipath.py
new file mode 100644
index 0000000..d955702
--- /dev/null
+++ b/controller/multipath.py
@@ -0,0 +1,16 @@
+"""Choose branches eligible for DSCP DATA rules using the routing policy."""
+
+
+def eligible_paths(policy, selected):
+    # Static experiments must retain their intentionally fixed route.
+    if policy.mode != "failover" or selected is None:
+        return frozenset()
+    paths = {selected} if policy.available(selected) else set()
+    selected_score = policy.health.get(selected)
+    if selected_score is not None:
+        for path in ("upper", "lower"):
+            score = policy.health.get(path)
+            if (policy.available(path) and score is not None
+                    and score <= selected_score + .02):
+                paths.add(path)
+    return frozenset(paths)
diff --git a/controller/probe_schedule.py b/controller/probe_schedule.py
new file mode 100644
index 0000000..23d6ca7
--- /dev/null
+++ b/controller/probe_schedule.py
@@ -0,0 +1,53 @@
+"""Adaptive per-path, per-direction probe scheduling."""
+import time
+
+
+class AdaptiveProbeSchedule:
+    def __init__(
+        self,
+        min_interval=0.1,
+        # Leave the 500 ms response deadline inside the tracker's 2 s freshness.
+        max_interval=1.5,
+        initial_interval=0.25,
+        healthy_threshold=0.05,
+        degraded_threshold=0.10,
+    ):
+        if not 0 < min_interval <= initial_interval <= max_interval:
+            raise ValueError("Invalid probe interval bounds")
+        if not 0 <= healthy_threshold < degraded_threshold:
+            raise ValueError("Invalid health thresholds")
+
+        self.min_interval = min_interval
+        self.max_interval = max_interval
+        self.healthy_threshold = healthy_threshold
+        self.degraded_threshold = degraded_threshold
+        self.interval = initial_interval
+        self.degraded = True
+        self.last_result_ns = None
+
+    def set_degraded(self, degraded, now_ns=None):
+        """Update health state and optionally record its update time."""
+        self.degraded = bool(degraded)
+        if now_ns is not None:
+            self.last_result_ns = now_ns
+
+    def record_result(self, success, now_ns=None):
+        """Record a probe result and adapt the next interval."""
+        if now_ns is None:
+            now_ns = time.monotonic_ns()
+        self.last_result_ns = now_ns
+
+        if success:
+            if not self.degraded:
+                self.interval = min(self.max_interval, self.interval * 1.25)
+        else:
+            self.degraded = True
+            self.interval = max(self.min_interval, self.interval / 2)
+
+        return self.interval
+
+    def interval_seconds(self, now_ns=None):
+        """Return the interval used by the controller's next deadline."""
+        if self.degraded:
+            return max(self.min_interval, min(self.interval, 0.25))
+        return min(self.max_interval, self.interval)
diff --git a/controller/routing.py b/controller/routing.py
index 7fff25b..c1af4dc 100644
--- a/controller/routing.py
+++ b/controller/routing.py
@@ -1,3 +1,4 @@
+
 """Small routing policy shared by the controller and its unit tests.
 
 The topology deliberately uses explicit ports. Both branches carry transit
@@ -28,18 +29,23 @@ def forwarding_rules(path):
     """Return exact-port IPv4/ARP forwarding rules; None blocks host ingress."""
     if path not in (None, "upper", "lower"):
         raise ValueError("unknown path")
+
     first = {"upper": 2, "lower": 3, None: None}[path]
     last = {"upper": 1, "lower": 2, None: None}[path]
+
     port_pairs = {
         1: ((1, first), (2, 1), (3, 1)),
         2: ((1, 2), (2, 1)),
         3: ((1, 2), (2, 1)),
         4: ((1, 3), (2, 3), (3, last)),
     }
+
     return {
-        dpid: [Flow(source, eth_type, destination)
-               for source, destination in pairs
-               for eth_type in (0x0800, 0x0806)]
+        dpid: [
+            Flow(source, eth_type, destination)
+            for source, destination in pairs
+            for eth_type in (0x0800, 0x0806)
+        ]
         for dpid, pairs in port_pairs.items()
     }
 
@@ -48,36 +54,78 @@ class PathPolicy:
     def __init__(self, mode="failover", initial="upper"):
         if mode not in ("static", "failover"):
             raise ValueError("mode must be static or failover")
+
         if initial not in PATH_PORTS:
             raise ValueError("initial path must be upper or lower")
+
         self.initial = initial
         self.mode = mode
         self.connected = set()
         self.ports = {}
         self.health = {}
 
+    def availability_details(self, path):
+        missing_switches = [
+            dpid for dpid in PATH_SWITCHES[path]
+            if dpid not in self.connected
+        ]
+
+        unavailable_ports = [
+            port for port in HOST_PORTS + PATH_PORTS[path]
+            if not self.ports.get(port, False)
+        ]
+
+        return {
+            "available": not missing_switches and not unavailable_ports,
+            "missing_switches": missing_switches,
+            "unavailable_ports": unavailable_ports,
+        }
+
     def available(self, path):
-        return (all(dpid in self.connected for dpid in PATH_SWITCHES[path])
-                and all(self.ports.get(port, False)
-                        for port in HOST_PORTS + PATH_PORTS[path]))
+        return self.availability_details(path)["available"]
 
     def choose(self, current):
         # A static route intentionally does not recover from a branch failure.
         # Host loss, however, leaves no usable endpoint in either mode.
         if not all(dpid in self.connected for dpid in (1, 4)):
             return None
+
         if not all(self.ports.get(port, False) for port in HOST_PORTS):
             return None
+
         if self.mode == "static":
             return self.initial
-        candidates = [path for path in (self.initial, "lower" if self.initial == "upper" else "upper") if self.available(path)]
+
+        candidates = [
+            path
+            for path in (
+                self.initial,
+                "lower" if self.initial == "upper" else "upper",
+            )
+            if self.available(path)
+        ]
+
         if not candidates:
             return None
-        if current in candidates and not all(path in self.health for path in candidates):
+
+        if current in candidates and not all(
+            path in self.health for path in candidates
+        ):
             return current
-        best = min(candidates, key=lambda path: self.health.get(path, 0))
+
+        best = min(
+            candidates,
+            key=lambda path: self.health.get(path, 0),
+        )
+
         # Require a score improvement greater than .02; this score combines
         # smoothed loss and delay, so the threshold is not a loss percentage.
-        if current in candidates and self.health.get(current, 0) <= self.health.get(best, 0) + .02:
+        if (
+            current in candidates
+            and self.health.get(current, 0)
+            <= self.health.get(best, 0) + 0.02
+        ):
             return current
+
         return best
+
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index 792d5a5..87c8420 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -116,7 +116,7 @@ failback. Initial branch defaults to upper; RUDP_INITIAL_PATH=lower permits the
 reverse-direction recovery check.
 
 `controller/probes.py` tracks independent Ethernet probes in both directions on
-both branches, including the idle branch. Every 250 ms the controller sends a
+both branches, including the idle branch. At each adaptive deadline the controller sends a
 60-byte experimental-EtherType 0x88b5 frame with a random session token and sequence
 through the selected branch port. High-priority transit rules cross s2/s3; edge
 rules punt the arriving frame to the controller and never to hosts. Validation
@@ -129,7 +129,8 @@ Latency includes controller/OpenFlow overhead; it is not a pure link RTT or loss
 probability. Both directions need four samples, each at most 2 s old. Health-only
 moves require an improvement exceeding .02 and a 2 s route hold-down, and wait for
 outstanding barriers. Hard port failures bypass that hold-down. Link changes clear
-old branch health. Approximately 16 small frames/s probe both live branches.
+old branch health. At the initial 250 ms spacing, both branches use approximately
+16 small frames/s; healthy spacing grows and loss accelerates probes.
 
 ## Socket–SDN observability and recovery definition
 
@@ -150,3 +151,23 @@ No route or no later useful ACK is censored/N/A, never zero. Repeated-fault
 attribution ends at the next intervention; a failure on the inactive branch is
 labeled separately. The joined SVG and recovery JSON preserve all three timestamps.
 See PROGRESS_LOG.md, DEMO.md and scripts/evaluate.py.
+
+## Optional multipath DATA and adaptive probes
+
+`client --multipath --upper-weight 3 --lower-weight 1` marks DATA with DSCP 8
+or 16 using sequence modulo total weight. Retries retain the sequence's mark;
+control packets reset to DSCP 0. The controller maps those marks onto eligible
+branches. This is forward DATA distribution; ACKs use the selected reverse route.
+It is not independent congestion control per path.
+
+A branch must have working ports and measured health within .02 of the selected
+path's score to join it. Unmeasured alternates are excluded. Removing an unhealthy
+mark-specific rule makes that traffic use the selected base route. Static mode
+installs no multipath overrides, preserving the permanent-failure comparison.
+
+Probe deadlines are independent per branch/direction. Successful healthy probes
+increase spacing up to 1.5 seconds; loss halves it to a 100 ms minimum. Degraded
+paths are capped at 250 ms. The tracker still expires outstanding probes at
+500 ms; the 1.5 s spacing leaves that deadline inside the 2 s health freshness.
+These changes need fresh Linux measurements; historical fixed-interval
+results do not measure their overhead or recovery behavior.
diff --git a/docs/DEMO.md b/docs/DEMO.md
index 34f20d9..913596b 100644
--- a/docs/DEMO.md
+++ b/docs/DEMO.md
@@ -81,3 +81,16 @@ docker run --rm --privileged -v "$PWD/results:/project/results" rudp-lab python
 The console's `recovery=` is injection-to-progress, not link-down-to-progress.
 The CSV and joined events distinguish both. Never label a startup failure or
 uninjected transfer as successful failover.
+
+## Two simultaneous uploads
+
+Each flow uses a separate receiver process and UDP port on h2; the receiver is
+one upload per process, rather than a persistent multi-session service. Both
+senders run on h1. This demonstrates concurrent uploads, not separate sender hosts.
+
+```sh
+sudo .venv/bin/python scripts/run.py experiment --scenarios link-failure --modes failover --rtos adaptive --sacks bitmap --ccs off --flows 2 --repeats 1 --output results/live-concurrent
+```
+
+Check both flow rows for success, matching hashes and recovery. Rehearse this
+before assessment; use retained output during the ten-minute presentation.
diff --git a/docs/VALIDATION.md b/docs/VALIDATION.md
index f00372f..4931047 100644
--- a/docs/VALIDATION.md
+++ b/docs/VALIDATION.md
@@ -121,3 +121,29 @@ Historical material is preserved in [the archive](../results/archive/README.md).
 These measurements are retained from commit
 `79f097dea2ae8f61e70dd4558816531dfbbb97a5`. This newer client/codec interface has separate localhost checks. The recorded Linux
 measurements describe the earlier commit and its original source fingerprints.
+
+
+## Validate the final submission revision
+
+After merging, freeze the code and rebuild the Linux image from that checkout.
+Run the preflight and evaluation into fresh directories. Record the code commit
+alongside the output so later documentation commits cannot obscure its identity.
+
+```sh
+git rev-parse HEAD > results/submission-source.txt
+docker build -t rudp-lab:submission .
+sudo docker run --rm --privileged -v "$PWD/results:/project/results" rudp-lab:submission python scripts/run.py demo --mode failover --datapath user --output results/submission-preflight
+sudo docker run --rm --privileged -v "$PWD/results:/project/results" rudp-lab:submission python scripts/run.py evaluate --output results/submission-evaluation
+```
+
+The standard evaluation does not enable the optional multipath sender. Its
+successful completion alone cannot establish simultaneous use of both branches.
+Multipath needs a Linux trial with marked DATA, per-branch packet counters,
+matching hashes and hard/soft failure interventions. Localhost and recording
+OpenFlow tests cover implementation behavior, not that Mininet evidence gap.
+
+For an optional performance comparison, existing experiment switches support
+`--file-size`, `--delay`, `--flows` and `--ccs off on`. Change one factor at a time,
+use three repeats and fresh directories, and compare per-flow goodput and total
+goodput. A fairness claim requires competition measurements; it cannot be inferred
+from a clean single-flow transfer. No new performance measurements are claimed here.
diff --git a/src/client.c b/src/client.c
index 8596184..497087b 100644
--- a/src/client.c
+++ b/src/client.c
@@ -1,5 +1,11 @@
+#ifdef __APPLE__
+/* IP_TOS is a Darwin extension hidden by the POSIX feature profile. */
+#define _DARWIN_C_SOURCE
+#endif
 #include "utils.h"
 #include <arpa/inet.h>
+#include <errno.h>
+#include <netinet/in.h>
 #include <getopt.h>
 #include <math.h>
 #include <time.h>
@@ -68,17 +74,46 @@ static int exchange(int fd, const struct sockaddr_in *server,
 }
 
 struct transport_options {
-    int adaptive, sack, congestion;
+    int adaptive, sack, congestion, multipath;
+    unsigned int upper_weight, lower_weight;
     double drop_rate;
 };
 
 /* Simulate loss after the send decision so normal ACK/retry logic observes
  * exactly the same silence it would see after network DATA loss. */
 static int send_data(int fd, const struct packet *packet,
-                     const struct sockaddr_in *peer, double drop_rate)
+                     const struct sockaddr_in *peer, double drop_rate,
+                     int multipath, unsigned int upper_weight,
+                     unsigned int lower_weight)
 {
     if (drop_rate > 0 && rand() / (RAND_MAX + 1.0) < drop_rate) return 0;
-    return send_packet(fd, packet, peer);
+
+    if (!multipath)
+        return send_packet(fd, packet, peer);
+
+    /* Weighted sequence mapping is deterministic, so retransmissions of
+     * a sequence number stay on the same path. */
+    if (upper_weight == 0 || lower_weight == 0 ||
+        upper_weight > 1000000U || lower_weight > 1000000U ||
+        upper_weight > 1000000U - lower_weight) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    unsigned int total_weight = upper_weight + lower_weight;
+    unsigned int position = packet->sequence % total_weight;
+    int dscp = position < upper_weight ? 8 : 16;
+    int tos = dscp << 2;
+    if (setsockopt(fd, IPPROTO_IP, IP_TOS, &tos, sizeof(tos)) != 0)
+        return -1;
+
+    int result = send_packet(fd, packet, peer);
+    int reset_tos = 0;
+    if (setsockopt(fd, IPPROTO_IP, IP_TOS,
+                   &reset_tos, sizeof(reset_tos)) != 0 && result == 0)
+        return -1;
+
+    return result;
 }
 
 static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
@@ -109,7 +144,9 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                 fprintf(stderr, "Input file changed or could not be read.\n");
                 return -1;
             }
-            if (send_data(fd, &slot->packet, server, options.drop_rate) != 0) {
+            if (send_data(fd, &slot->packet, server, options.drop_rate,
+                           options.multipath, options.upper_weight,
+                           options.lower_weight) != 0) {
                 perror("send DATA");
                 return -1;
             }
@@ -133,7 +170,9 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                     fprintf(stderr, "Retry limit reached for DATA %" PRIu32 ".\n", sequence);
                     return -1;
                 }
-                if (send_data(fd, &slot->packet, server, options.drop_rate) != 0) {
+                if (send_data(fd, &slot->packet, server, options.drop_rate,
+                           options.multipath, options.upper_weight,
+                           options.lower_weight) != 0) {
                     perror("retransmit DATA");
                     return -1;
                 }
@@ -235,7 +274,9 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                     struct send_slot *slot = &slots[seq % WINDOW_SIZE];
                     if (slot->acknowledged || slot->fast_sent || slot->gap_reports < 3) continue;
                     if (slot->retries >= MAX_RETRIES) return -1;
-                    if (send_data(fd, &slot->packet, server, options.drop_rate) != 0) return -1;
+                    if (send_data(fd, &slot->packet, server, options.drop_rate,
+                           options.multipath, options.upper_weight,
+                           options.lower_weight) != 0) return -1;
                     slot->fast_sent = 1;
                     slot->sent_at = monotonic_ms();
                     ++slot->retries; ++stats->data_sends; ++stats->data_retries;
@@ -280,13 +321,16 @@ int main(int argc, char *argv[])
     uint16_t port;
     const char *host = NULL, *port_text = NULL, *filename = NULL;
     const char *rto_mode = getenv("RUDP_RTO");
-    struct transport_options options = {0, option_enabled("RUDP_SACK"), option_enabled("RUDP_CC"), 0};
+    struct transport_options options = {0, option_enabled("RUDP_SACK"), option_enabled("RUDP_CC"), option_enabled("RUDP_MULTIPATH"), 1, 1, 0};
     static const struct option cli[] = {
         {"host", required_argument, NULL, 'H'}, {"port", required_argument, NULL, 'p'},
         {"file", required_argument, NULL, 'f'}, {"sack", no_argument, NULL, 's'},
         {"no-sack", no_argument, NULL, 'S'}, {"cc", no_argument, NULL, 'c'},
-        {"no-cc", no_argument, NULL, 'C'}, {"adaptive-rto", no_argument, NULL, 'a'},
-        {"fixed-rto", no_argument, NULL, 'F'}, {"drop-rate", required_argument, NULL, 'd'},
+        {"no-cc", no_argument, NULL, 'C'}, {"multipath", no_argument, NULL, 'm'}, {"adaptive-rto", no_argument, NULL, 'a'},
+        {"fixed-rto", no_argument, NULL, 'F'},
+        {"upper-weight", required_argument, NULL, 'u'},
+        {"lower-weight", required_argument, NULL, 'l'},
+        {"drop-rate", required_argument, NULL, 'd'},
         {"help", no_argument, NULL, 'h'}, {NULL, 0, NULL, 0}
     };
     int opt;
@@ -299,6 +343,22 @@ int main(int argc, char *argv[])
         case 'F': rto_mode = "fixed"; break;
         case 's': case 'S': options.sack = opt == 's'; break;
         case 'c': case 'C': options.congestion = opt == 'c'; break;
+        case 'm': options.multipath = 1; break;
+        case 'u': case 'l': {
+            char *end;
+            errno = 0;
+            unsigned long weight = strtoul(optarg, &end, 10);
+            if (errno || end == optarg || *end ||
+                weight < 1 || weight > 1000000UL) {
+                fprintf(stderr, "Path weights must be integers from 1 to 1000000\n");
+                return EXIT_FAILURE;
+            }
+            if (opt == 'u')
+                options.upper_weight = (unsigned int)weight;
+            else
+                options.lower_weight = (unsigned int)weight;
+            break;
+        }
         case 'd': {
             char *end;
             options.drop_rate = strtod(optarg, &end);
@@ -313,6 +373,9 @@ int main(int argc, char *argv[])
         case 'h':
             puts("Usage: client HOST PORT FILE | client --host IPv4 --port PORT --file FILE\n"
                  "Options: --sack | --no-sack, --adaptive-rto | --fixed-rto, --cc | --no-cc\n"
+                 "         --multipath    distribute DATA across upper/lower SDN paths\n"
+                 "         --upper-weight N  upper path weight (default 1)\n"
+                 "         --lower-weight N  lower path weight (default 1)\n"
                  "         --drop-rate P   simulate loss on each outgoing DATA send (0..1)");
             return EXIT_SUCCESS;
         default: return EXIT_FAILURE;
@@ -324,6 +387,12 @@ int main(int argc, char *argv[])
         fprintf(stderr, "Usage: client HOST PORT FILE (see --help)\n");
         return EXIT_FAILURE;
     }
+    if (options.upper_weight < 1 || options.lower_weight < 1 ||
+        options.lower_weight > 1000000U ||
+        options.upper_weight > 1000000U - options.lower_weight) {
+        fprintf(stderr, "Combined path weights must not exceed 1000000\n");
+        return EXIT_FAILURE;
+    }
     if (rto_mode && strcmp(rto_mode, "fixed") && strcmp(rto_mode, "adaptive")) {
         fprintf(stderr, "RUDP_RTO must be fixed or adaptive.\n");
         return EXIT_FAILURE;
diff --git a/tests/multipath_sender.c b/tests/multipath_sender.c
new file mode 100644
index 0000000..d480c79
--- /dev/null
+++ b/tests/multipath_sender.c
@@ -0,0 +1,44 @@
+/* Exercise the real sender's DSCP selection without requiring an SDN switch. */
+#define main client_main
+#define setsockopt record_setsockopt
+#define send_packet record_send_packet
+#include "../src/client.c"
+#undef main
+#undef setsockopt
+#undef send_packet
+#include <assert.h>
+
+static int current_tos, observed_tos;
+
+int record_setsockopt(int fd, int level, int option, const void *value, socklen_t length)
+{
+    (void)fd;
+    assert(level == IPPROTO_IP && option == IP_TOS && length == sizeof(int));
+    current_tos = *(const int *)value;
+    return 0;
+}
+
+int record_send_packet(int fd, const struct packet *packet, const struct sockaddr_in *peer)
+{
+    (void)fd; (void)packet; (void)peer;
+    observed_tos = current_tos;
+    return 0;
+}
+
+int main(void)
+{
+    struct packet packet = {0};
+    struct sockaddr_in peer = {0};
+    for (unsigned int sequence = 0; sequence < 16; ++sequence) {
+        packet.sequence = sequence;
+        for (int attempt = 0; attempt < 2; ++attempt) {
+            assert(send_data(0, &packet, &peer, 0, 1, 3, 1) == 0);
+            assert(observed_tos == ((sequence % 4 < 3 ? 8 : 16) << 2));
+            assert(current_tos == 0); /* Control traffic must not inherit DATA's mark. */
+        }
+    }
+    assert(send_data(0, &packet, &peer, 0, 0, 1, 1) == 0);
+    assert(observed_tos == 0);
+    assert(send_data(0, &packet, &peer, 0, 1, 0, 1) == -1);
+    return 0;
+}
diff --git a/tests/test_local.py b/tests/test_local.py
index d4508f2..26d98eb 100644
--- a/tests/test_local.py
+++ b/tests/test_local.py
@@ -60,6 +60,24 @@ def stop(process):
 
 
 class LocalTests(unittest.TestCase):
+    def test_multipath_client_transfers_with_weights(self):
+        with tempfile.TemporaryDirectory() as folder:
+            source, output = Path(folder)/'input', Path(folder)/'output'
+            data = os.urandom(32769)
+            source.write_bytes(data)
+            receiver, address = server(output)
+            try:
+                result = subprocess.run(
+                    [str(CLIENT), address[0], str(address[1]), str(source),
+                     '--multipath', '--upper-weight', '3', '--lower-weight', '1'],
+                    capture_output=True, text=True, timeout=15)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                _, stderr = receiver.communicate(timeout=12)
+                self.assertEqual(receiver.returncode, 0, stderr)
+                self.assertEqual(output.read_bytes(), data)
+            finally:
+                stop(receiver)
+
     def test_binary_sizes(self):
         def transfer(size):
             with tempfile.TemporaryDirectory() as folder:
diff --git a/tests/test_multipath.py b/tests/test_multipath.py
new file mode 100644
index 0000000..b61e0cc
--- /dev/null
+++ b/tests/test_multipath.py
@@ -0,0 +1,53 @@
+"""Exercise the policy used by the real DSCP rule installer."""
+from pathlib import Path
+import sys
+import unittest
+import os
+import shlex
+import subprocess
+import tempfile
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from controller.multipath import eligible_paths
+from test_controller import connected_policy
+
+
+class MultipathTests(unittest.TestCase):
+    def setUp(self):
+        self.policy = connected_policy()
+        self.policy.health = {"upper": .01, "lower": .01}
+
+    def test_real_sender_marks_weights_retries_and_resets(self):
+        root = Path(__file__).resolve().parents[1]
+        with tempfile.TemporaryDirectory() as folder:
+            binary = Path(folder)/'sender-test'
+            subprocess.run(shlex.split(os.environ.get('CC', 'cc')) + [
+                '-std=c11', '-D_POSIX_C_SOURCE=200809L', '-Wall', '-Wextra',
+                '-Wpedantic', '-Werror', str(root/'tests/multipath_sender.c'),
+                str(root/'src/utils.c'), '-o', str(binary)], check=True)
+            subprocess.run([str(binary)], check=True, timeout=5)
+
+    def test_healthy_paths_share_data(self):
+        self.assertEqual(eligible_paths(self.policy, "upper"), {"upper", "lower"})
+
+    def test_degraded_path_is_excluded(self):
+        self.policy.health["upper"] = .5
+        self.assertEqual(eligible_paths(self.policy, "lower"), {"lower"})
+
+    def test_unmeasured_alternate_is_excluded(self):
+        del self.policy.health["lower"]
+        self.assertEqual(eligible_paths(self.policy, "upper"), {"upper"})
+
+    def test_failed_path_is_excluded(self):
+        self.policy.ports[(1, 2)] = False
+        self.assertEqual(eligible_paths(self.policy, "lower"), {"lower"})
+
+    def test_static_mode_has_no_overrides(self):
+        self.policy.mode = "static"
+        self.assertFalse(eligible_paths(self.policy, "upper"))
+
+    def test_no_selected_route_has_no_overrides(self):
+        self.assertFalse(eligible_paths(self.policy, None))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_openflow.py b/tests/test_openflow.py
index 252dc7c..30b7f7f 100644
--- a/tests/test_openflow.py
+++ b/tests/test_openflow.py
@@ -65,6 +65,49 @@ def test_soft_hold_down_never_blocks_hard_failure(self):
             c._reconcile('link_down')
             self.assertEqual(c.path, 'lower')
 
+    def test_multipath_rules_follow_health_without_route_change(self):
+        c = self.controller
+        c.policy = connected_policy()
+        c.policy.health = {'upper': .01, 'lower': .01}
+        c.started = True
+        parser = MagicMock()
+        parser.OFPMatch.side_effect = lambda **kwargs: kwargs
+        dp = types.SimpleNamespace(
+            id=1, ofproto_parser=parser,
+            ofproto=types.SimpleNamespace(OFPIT_APPLY_ACTIONS=4, OFPFC_DELETE=3,
+                                         OFPP_ANY=0xffffffff, OFPG_ANY=0xffffffff),
+            send_msg=MagicMock(), set_xid=MagicMock())
+        c.datapaths = {1: dp}
+
+        def installed_dscps():
+            return {call.kwargs['match']['ip_dscp']
+                    for call in parser.OFPFlowMod.call_args_list
+                    if call.kwargs.get('priority') == 150}
+
+        c._reconcile('startup')
+        self.assertEqual(installed_dscps(), {8, 16})
+        parser.reset_mock()
+        c.pending.clear()
+        # The alternate degrades while the selected route remains upper.
+        c.policy.health['lower'] = .5
+        c._reconcile('health_update')
+        self.assertEqual(c.path, 'upper')
+        self.assertEqual(installed_dscps(), {8})
+        self.assertTrue(any(call.kwargs.get('cookie') == self.module.MULTIPATH_COOKIE
+                            and call.kwargs.get('command') == 3
+                            for call in parser.OFPFlowMod.call_args_list))
+        parser.reset_mock()
+        c.pending.clear()
+        c.last_route_change_ns = 0
+        c.policy.health = {'upper': .5, 'lower': .01}
+        c._reconcile('health_update')
+        self.assertEqual(c.path, 'lower')
+        self.assertEqual(installed_dscps(), {16})
+        parser.reset_mock()
+        c.policy.mode = 'static'
+        c._reconcile('startup', force=True)
+        self.assertFalse(installed_dscps())
+
     def test_old_barrier_never_marks_new_generation_ready(self):
         c = self.controller; c.generation = 2; c.pending = {(1, 10): 1}; c.path = 'lower'
         c.barrier_reply(types.SimpleNamespace(msg=types.SimpleNamespace(datapath=types.SimpleNamespace(id=1), xid=10)))
diff --git a/tests/test_probe_schedule.py b/tests/test_probe_schedule.py
new file mode 100644
index 0000000..bf0af0b
--- /dev/null
+++ b/tests/test_probe_schedule.py
@@ -0,0 +1,62 @@
+from pathlib import Path
+import sys
+import unittest
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from controller.probe_schedule import AdaptiveProbeSchedule
+
+
+class AdaptiveProbeScheduleTests(unittest.TestCase):
+    def test_starts_at_250ms(self):
+        scheduler = AdaptiveProbeSchedule()
+        self.assertEqual(scheduler.interval, 0.25)
+
+    def test_degraded_schedule_uses_fast_interval(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(True)
+        self.assertEqual(scheduler.interval_seconds(0), 0.25)
+
+    def test_failed_probe_halves_interval(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.record_result(False, now_ns=1_000_000_000)
+        self.assertEqual(scheduler.interval, 0.125)
+
+    def test_healthy_schedule_can_increase_interval(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(False)
+        scheduler.record_result(True, now_ns=1_000_000_000)
+        self.assertAlmostEqual(scheduler.interval, 0.3125)
+
+    def test_interval_never_exceeds_maximum(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(False)
+        for i in range(30):
+            scheduler.record_result(True, now_ns=i * 1_000_000_000)
+        self.assertLessEqual(scheduler.interval, scheduler.max_interval)
+
+    def test_interval_never_falls_below_minimum(self):
+        scheduler = AdaptiveProbeSchedule()
+        for i in range(30):
+            scheduler.record_result(False, now_ns=i * 1_000_000_000)
+        self.assertGreaterEqual(scheduler.interval, scheduler.min_interval)
+
+    def test_degraded_interval_is_capped_at_250ms(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(True)
+        scheduler.interval = 1.0
+        self.assertEqual(scheduler.interval_seconds(0), 0.25)
+
+    def test_maximum_spacing_leaves_time_for_response(self):
+        from controller.probes import ProbeTracker
+        schedule, tracker = AdaptiveProbeSchedule(), ProbeTracker()
+        self.assertLessEqual(schedule.max_interval * 1e9 + tracker.timeout_ns,
+                             tracker.freshness_ns)
+
+    def test_initial_interval_is_within_bounds(self):
+        scheduler = AdaptiveProbeSchedule()
+        self.assertGreaterEqual(scheduler.interval, scheduler.min_interval)
+        self.assertLessEqual(scheduler.interval, scheduler.max_interval)
+
+
+if __name__ == "__main__":
+    unittest.main()