feat(docker): expose the eBPF time-window filter on kernel links

uBridge's window mode gained the semantics the B.2 spec asked for —
`tc window_drop <if> <start> <outage> <pct> [<period> [<jitter>]]`: a
single outage that traffic passes before *and* after, recurring flaps
with period >= outage, and a per-cycle jitter — so the kernel-only
`window_drop` filter type is now exposed:

- validation mirrors the tc grammar (3-5 params, ranges, period >=
  outage, jitter cap) and adds window_drop to KERNEL_ONLY_FILTERS;
  filter_inactive_filters keeps start=0 — unlike every other filter a
  zero first parameter means "the outage starts now", not "disabled".
- `_ubridge_apply_ebpf_drops` sends `tc window_drop` (positional
  pass-through) and turns the mode off explicitly when absent, under the
  same ebpf=1 gate.
- FILTERS entry (Start/Outage/Chance/Period/Jitter) + available_filters.

The schedule is relative to the apply: reconcile re-sets the classifier
on every apply, so any filter change, node restart or link reset restarts
the window.

Verified on a two-container kernel link, 15/15: a single outage
[1000, 1500, 100] arrives per-seq as [0..4, 13..19] (passing before the
window, dropped inside, passing after); recurring 1000/2000 ms = 52 %
round trip (duty cycle); period == outage = permanent outage; period <
outage = 409 before uBridge; jittered flap keeps the ~50 % duty cycle;
30 % chance on a permanent window = 53 % round trip against the
per-direction theory 1-0.7^2 = 51 %; coexist with netem; node restart
restores; available_filters kernel vs relay (409 + hidden); teardown
clean. P6b re-run as a regression on the same uBridge build: 13/13.

Worth recording for the next rebuild: the first delivered window program
loaded as root but reported ebpf=0 through gns3-server — the verifier's
non-root path loses the loop counter's bound, treats the catch-up loop as
unbounded and rejects the *whole* program (E2BIG, 8193 pending states),
which also took frequency_drop and quota down, with ebpf=0 as the only
symptom. Smaller trip caps did not help; the loop had to go (uBridge
572ca0c), on top of a Makefile header-dependency fix (53e34bb) that had
been shipping a stale program object.

tests/compute/docker/test_docker_kernel_datapath.py

@@ -497,6 +497,7 @@ async def test_update_nio_kernel_turns_ebpf_modes_off(vm):
 
     vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" off')
     vm._ubridge_send.assert_any_call(f'tc quota_drop "{host_ifc}" off')
+    vm._ubridge_send.assert_any_call(f'tc window_drop "{host_ifc}" off')
     for c in vm._ubridge_send.call_args_list:
         assert "nth_drop" not in c.args[0] or "off" in c.args[0]
 
@@ -520,6 +521,54 @@ async def test_apply_ebpf_drops_requires_ebpf_capability(vm):
     vm._ubridge_send.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_update_nio_kernel_applies_window_drop(vm):
+    """
+    window_drop becomes the eBPF time-window mode: the positional
+    [start, outage, chance, period, jitter] maps 1:1 onto the tc command
+    (trailing optionals omitted), and an absent window is turned off.
+    """
+
+    vm._ubridge_hypervisor = _caps_hypervisor(ebpf="1")
+    vm._ubridge_send = AsyncioMagicMock()
+    vm.status = "started"
+    host_ifc, _ = vm._veth_names(0, 0)
+    vm._kernel_veths[(0, 0)] = host_ifc
+    nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
+    vm._ethernet_adapters[0].add_nio(0, nio)
+
+    nio.filters = {"window_drop": [0, 2000, 100]}
+    await vm.adapter_update_nio_binding(0, nio)
+    vm._ubridge_send.assert_any_call(f'tc window_drop "{host_ifc}" 0 2000 100')
+
+    vm._ubridge_send.reset_mock()
+    nio.filters = {"window_drop": [500, 800, 100, 2400, 300]}
+    await vm.adapter_update_nio_binding(0, nio)
+    vm._ubridge_send.assert_any_call(f'tc window_drop "{host_ifc}" 500 800 100 2400 300')
+
+    # only the window mode present: nth/quota are explicitly off
+    vm._ubridge_send.assert_any_call(f'tc nth_drop "{host_ifc}" off')
+    vm._ubridge_send.assert_any_call(f'tc quota_drop "{host_ifc}" off')
+
+
+@pytest.mark.asyncio
+async def test_apply_window_drop_requires_ebpf_capability(vm):
+    """
+    The window mode shares the eBPF capability gate (error names setcap).
+    """
+
+    hyp = MagicMock()
+    hyp.is_running.return_value = True
+    hyp.send = AsyncioMagicMock(return_value=["netem=delay;ebpf=0;cbpf=1"])
+    vm._ubridge_hypervisor = hyp
+    vm._ubridge_send = AsyncioMagicMock()
+    host_ifc, _ = vm._veth_names(0, 0)
+
+    with pytest.raises(DockerError, match="setcap cap_bpf"):
+        await vm._ubridge_apply_ebpf_drops(host_ifc, {"window_drop": [0, 2000, 100]})
+    vm._ubridge_send.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_apply_ebpf_drops_skips_old_ubridge_without_filters(vm):
     """

tests/controller/test_kernel_datapath_link.py

@@ -428,7 +428,7 @@ async def test_update_suspend_kernel_link_sends_no_synthetic_filter(project):
 async def test_kernel_datapath_eligible_with_netem_extension_filters(project):
     """
     All netem-extension types have a tc netem equivalent — they keep the
-    link kernel-eligible (only frequency_drop still forces the relay).
+    link kernel-eligible (kernel/relay is a purely topological choice).
     """
 
     link, node1, node2 = await _kernel_link(project)
@@ -441,6 +441,8 @@ async def test_kernel_datapath_eligible_with_netem_extension_filters(project):
         "gemodel": [100, 0, 30],
         "seed": [42],
         "limit": [5000],
+        "quota": [1000000, 100],
+        "window_drop": [0, 800, 100, 2400, 200],
     }
     # gemodel/packet_loss are mutually exclusive — drop packet_loss for the
     # eligibility check (validation rejects the mix elsewhere)
@@ -469,9 +471,9 @@ async def test_update_netem_extension_filters_accepted_on_kernel_link(project):
 @pytest.mark.asyncio
 async def test_update_kernel_only_filters_rejected_on_relay_link(project):
     """
-    Netem-extension / quota filters have no relay equivalent: setting them
-    on a created relay link is a 409. (The link is relay-wired because the
-    endpoints are mixed node types.)
+    Netem-extension / quota / window_drop filters have no relay equivalent:
+    setting them on a created relay link is a 409. (The link is relay-wired
+    because the endpoints are mixed node types.)
     """
 
     link, _n1, _n2 = await _relay_link(project)
@@ -481,6 +483,11 @@ async def test_update_kernel_only_filters_rejected_on_relay_link(project):
 
     with pytest.raises(ControllerError, match="kernel-datapath"):
         await link.update_filters({"rate": ["512kbit"]})
+    with pytest.raises(ControllerError, match="kernel-datapath"):
+        await link.update_filters({"window_drop": [0, 800, 100, 2400]})
+    with pytest.raises(ControllerError, match="period must be greater than or equal"):
+        # cross-parameter rule mirrors the tc window grammar
+        await link.update_filters({"window_drop": [0, 2000, 100, 1000]})
     # the stored filters are untouched
     assert link.filters == {}
 
@@ -489,6 +496,24 @@ async def test_update_kernel_only_filters_rejected_on_relay_link(project):
     assert link.filters == {"frequency_drop": [7]}
 
 
+@pytest.mark.asyncio
+async def test_update_window_drop_accepted_on_kernel_link(project):
+    """
+    window_drop rides the kernel NIO to both endpoints — including the
+    start=0 form (an immediate outage is NOT an "inactive" filter).
+    """
+
+    link, node1, node2 = await _kernel_link(project)
+    await link._prepare()
+
+    filters = {"window_drop": [0, 800, 100, 2400, 200]}
+    await link.update_filters(filters)
+    assert link.filters == filters
+    by_node = {entry[0].id: entry[3] for entry in (await link._prepare())}
+    assert by_node[node1.id]["filters"] == filters
+    assert by_node[node2.id]["filters"] == filters
+
+
 @pytest.mark.asyncio
 async def test_update_kernel_only_filters_accepted_before_creation(project):
     """

tests/utils/test_packet_filter_validation.py

@@ -383,3 +383,36 @@ class TestNetemExtensionFilters:
         # frequency_drop runs on both datapaths (relay userspace filter /
         # eBPF every-Nth) and stays
         assert clean == {"frequency_drop": [7]}
+
+    def test_window_drop_valid_and_invalid(self):
+        # the three delivered forms: single outage, recurring, jittered
+        validate_filter_parameters("window_drop", [0, 2000, 100])
+        validate_filter_parameters("window_drop", [500, 800, 100, 2400])
+        validate_filter_parameters("window_drop", [500, 800, 100, 2400, 300])
+        with pytest.raises(FilterValidationError, match="expects 3 to 5"):
+            validate_filter_parameters("window_drop", [500, 800])
+        with pytest.raises(FilterValidationError, match="expects 3 to 5"):
+            validate_filter_parameters("window_drop", [500, 800, 100, 2400, 300, 1])
+        with pytest.raises(FilterValidationError, match="Outage"):
+            validate_filter_parameters("window_drop", [0, 0, 100])
+        with pytest.raises(FilterValidationError, match="Chance"):
+            validate_filter_parameters("window_drop", [0, 2000, 101])
+        with pytest.raises(FilterValidationError, match="Jitter"):
+            validate_filter_parameters("window_drop", [0, 2000, 100, 2400, 10**10])
+
+    def test_window_drop_is_kernel_only(self):
+        clean, dropped = split_kernel_only_features({"window_drop": [0, 800, 100, 2400], "delay": [10, 0]})
+        assert dropped == {"window_drop"}
+        assert clean == {"delay": [10, 0]}
+
+    def test_window_drop_period_must_cover_outage(self):
+        with pytest.raises(FilterValidationError, match="period must be greater than or equal"):
+            validate_all_filters({"window_drop": [0, 2000, 100, 1000]})
+        # period == outage is accepted (uBridge allows it); shorter is not
+        validate_all_filters({"window_drop": [0, 2000, 100, 2000]})
+
+    def test_window_drop_start_zero_is_active(self):
+        # start=0 means "the outage starts now" — unlike other filters a
+        # zero first parameter must NOT be treated as "disabled"
+        active = filter_inactive_filters({"window_drop": [0, 2000, 100]})
+        assert active == {"window_drop": [0, 2000, 100]}