{
  lib,
  pkgs,
  inputs,
  ...
}:
let
  jfLib = import ./jellyfin-test-lib.nix { inherit pkgs lib; };
  webhookPlugin = import ../services/jellyfin/jellyfin-webhook-plugin.nix { inherit pkgs lib; };
  configureWebhook = webhookPlugin.mkConfigureScript {
    jellyfinUrl = "http://localhost:8096";
    webhooks = [
      {
        name = "qBittorrent Monitor";
        uri = "http://127.0.0.1:9898/";
        notificationTypes = [
          "PlaybackStart"
          "PlaybackProgress"
          "PlaybackStop"
        ];
      }
    ];
  };
in
pkgs.testers.runNixOSTest {
  name = "jellyfin-qbittorrent-monitor";

  nodes = {
    server =
      { ... }:
      {
        imports = [
          jfLib.jellyfinTestConfig
          inputs.vpn-confinement.nixosModules.default
        ];

        # Real qBittorrent service
        services.qbittorrent = {
          enable = true;
          webuiPort = 8080;
          openFirewall = true;

          serverConfig.LegalNotice.Accepted = true;

          serverConfig.Preferences = {
            WebUI = {
              # Disable authentication for testing
              AuthSubnetWhitelist = "0.0.0.0/0,::/0";
              AuthSubnetWhitelistEnabled = true;
              LocalHostAuth = false;
            };

            Downloads = {
              SavePath = "/var/lib/qbittorrent/downloads";
              TempPath = "/var/lib/qbittorrent/incomplete";
            };
          };

          serverConfig.BitTorrent.Session = {
            # Normal speed - unlimited
            GlobalUPSpeedLimit = 0;
            GlobalDLSpeedLimit = 0;

            # Alternate speed limits for when Jellyfin is streaming
            AlternativeGlobalUPSpeedLimit = 100;
            AlternativeGlobalDLSpeedLimit = 100;
          };
        };

        networking.firewall.allowedTCPPorts = [
          8096
          8080
        ];
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce [
          {
            address = "192.168.1.1";
            prefixLength = 24;
          }
        ];
        networking.interfaces.eth1.ipv4.routes = [
          {
            address = "203.0.113.0";
            prefixLength = 24;
          }
        ];

        # Create directories for qBittorrent.
        systemd.tmpfiles.rules = [
          "d /var/lib/qbittorrent/downloads 0755 qbittorrent qbittorrent"
          "d /var/lib/qbittorrent/incomplete 0755 qbittorrent qbittorrent"
        ];

        # Install the Jellyfin Webhook plugin before Jellyfin starts, mirroring
        # the production module. Jellyfin rewrites meta.json at runtime so a
        # read-only nix-store symlink would fail — we materialise a writable copy.
        systemd.services."jellyfin-webhook-install" = {
          description = "Install Jellyfin Webhook plugin files";
          before = [ "jellyfin.service" ];
          wantedBy = [ "jellyfin.service" ];
          serviceConfig = {
            Type = "oneshot";
            RemainAfterExit = true;
            User = "jellyfin";
            Group = "jellyfin";
            UMask = "0077";
            ExecStart = webhookPlugin.mkInstallScript {
              pluginsDir = "/var/lib/jellyfin/plugins";
            };
          };
        };
      };

    # Public test IP (RFC 5737 TEST-NET-3) so Jellyfin sees it as external
    client = {
      environment.systemPackages = [ pkgs.curl ];
      networking.interfaces.eth1.ipv4.addresses = lib.mkForce [
        {
          address = "203.0.113.10";
          prefixLength = 24;
        }
      ];
      networking.interfaces.eth1.ipv4.routes = [
        {
          address = "192.168.1.0";
          prefixLength = 24;
        }
      ];
    };
  };

  testScript = ''
    import json
    import time

    import importlib.util
    _spec = importlib.util.spec_from_file_location("jf_helpers", "${jfLib.helpers}")
    assert _spec and _spec.loader
    _jf = importlib.util.module_from_spec(_spec)
    _spec.loader.exec_module(_jf)
    setup_jellyfin = _jf.setup_jellyfin
    jellyfin_api = _jf.jellyfin_api

    auth_header = 'MediaBrowser Client="NixOS Test", DeviceId="test-1337", Device="TestDevice", Version="1.0"'

    def is_throttled():
        return server.succeed("curl -s http://localhost:8080/api/v2/transfer/speedLimitsMode").strip() == "1"

    def get_alt_dl_limit():
        prefs = json.loads(server.succeed("curl -s http://localhost:8080/api/v2/app/preferences"))
        return prefs["alt_dl_limit"]

    def get_alt_up_limit():
        prefs = json.loads(server.succeed("curl -s http://localhost:8080/api/v2/app/preferences"))
        return prefs["alt_up_limit"]

    def are_torrents_paused():
        torrents = json.loads(server.succeed("curl -s 'http://localhost:8080/api/v2/torrents/info'"))
        if not torrents:
            return False
        return all(t["state"].startswith("stopped") for t in torrents)

    start_all()
    server.wait_for_unit("qbittorrent.service")
    server.wait_for_open_port(8080)
    server.wait_until_succeeds("curl -sf http://localhost:8080/api/v2/app/version", timeout=30)

    token, user_id, movie_id, media_source_id = setup_jellyfin(
        server, retry, auth_header,
        "${jfLib.payloads.auth}", "${jfLib.payloads.empty}",
    )

    with subtest("Start monitor service"):
        python = "${pkgs.python3.withPackages (ps: [ ps.requests ])}/bin/python"
        monitor = "${../services/jellyfin/jellyfin-qbittorrent-monitor.py}"
        server.succeed(f"""
          systemd-run --unit=monitor-test \
            --setenv=JELLYFIN_URL=http://localhost:8096 \
            --setenv=JELLYFIN_API_KEY={token} \
            --setenv=QBITTORRENT_URL=http://localhost:8080 \
            --setenv=CHECK_INTERVAL=1 \
            --setenv=STREAMING_START_DELAY=1 \
            --setenv=STREAMING_STOP_DELAY=1 \
            --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
            --setenv=SERVICE_BUFFER=2000000 \
            --setenv=DEFAULT_STREAM_BITRATE=10000000 \
            --setenv=MIN_TORRENT_SPEED=100 \
            {python} {monitor}
        """)
        time.sleep(2)
        assert not is_throttled(), "Should start unthrottled"

    client_auth = 'MediaBrowser Client="External Client", DeviceId="external-9999", Device="ExternalDevice", Version="1.0"'
    client_auth2 = 'MediaBrowser Client="External Client 2", DeviceId="external-8888", Device="ExternalDevice2", Version="1.0"'
    server_ip = "192.168.1.1"

    with subtest("Client authenticates from external network"):
        auth_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}'"
        client_auth_result = json.loads(client.succeed(auth_cmd))
        client_token = client_auth_result["AccessToken"]

    with subtest("Second client authenticates from external network"):
        auth_cmd2 = f"curl -sf -X POST 'http://{server_ip}:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth2}'"
        client_auth_result2 = json.loads(client.succeed(auth_cmd2))
        client_token2 = client_auth_result2["AccessToken"]

    with subtest("External video playback triggers throttling"):
        playback_start = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-1",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_start)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        time.sleep(2)
        assert is_throttled(), "Should throttle for external video playback"

    with subtest("Pausing disables throttling"):
        playback_progress = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-1",
            "IsPaused": True,
            "PositionTicks": 10000000,
        }
        progress_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Progress' -d '{json.dumps(playback_progress)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(progress_cmd)
        time.sleep(2)

        assert not is_throttled(), "Should unthrottle when paused"

    with subtest("Resuming re-enables throttling"):
        playback_progress["IsPaused"] = False
        playback_progress["PositionTicks"] = 20000000
        progress_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Progress' -d '{json.dumps(playback_progress)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(progress_cmd)
        time.sleep(2)

        assert is_throttled(), "Should re-throttle when resumed"

    with subtest("Stopping playback disables throttling"):
        playback_stop = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-1",
            "PositionTicks": 50000000,
        }
        stop_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(playback_stop)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(stop_cmd)
        time.sleep(2)

        assert not is_throttled(), "Should unthrottle when playback stops"

    with subtest("Single stream sets proportional alt speed limits"):
        playback_start = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-proportional",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_start)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        time.sleep(3)

        assert is_throttled(), "Should be in alt speed mode during streaming"
        dl_limit = get_alt_dl_limit()
        ul_limit = get_alt_up_limit()
        # Both upload and download should get remaining bandwidth (proportional)
        assert dl_limit > 0, f"Download limit should be > 0, got {dl_limit}"
        assert ul_limit == dl_limit, f"Upload limit ({ul_limit}) should equal download limit ({dl_limit})"

        # Stop playback
        playback_stop = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-proportional",
            "PositionTicks": 50000000,
        }
        stop_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(playback_stop)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(stop_cmd)
        time.sleep(3)

    with subtest("Multiple streams reduce available bandwidth"):
        # Start first stream
        playback1 = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-multi-1",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd1 = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback1)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd1)
        time.sleep(3)

        single_dl_limit = get_alt_dl_limit()

        # Start second stream with different client identity
        playback2 = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-multi-2",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd2 = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback2)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth2}, Token={client_token2}'"
        client.succeed(start_cmd2)
        time.sleep(3)

        dual_dl_limit = get_alt_dl_limit()
        # Two streams should leave less bandwidth than one stream
        assert dual_dl_limit < single_dl_limit, f"Two streams ({dual_dl_limit}) should have lower limit than one ({single_dl_limit})"

        # Stop both streams
        stop1 = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-multi-1",
            "PositionTicks": 50000000,
        }
        stop_cmd1 = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(stop1)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(stop_cmd1)

        stop2 = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-multi-2",
            "PositionTicks": 50000000,
        }
        stop_cmd2 = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(stop2)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth2}, Token={client_token2}'"
        client.succeed(stop_cmd2)
        time.sleep(3)

    with subtest("Budget exhaustion pauses all torrents"):
        # Stop current monitor
        server.succeed("systemctl stop monitor-test || true")
        time.sleep(1)

        # Add a dummy torrent so we can check pause state
        server.succeed("curl -sf -X POST 'http://localhost:8080/api/v2/torrents/add' -d 'urls=magnet:?xt=urn:btih:0000000000000000000000000000000000000001%26dn=test-torrent'")
        time.sleep(2)

        # Start monitor with impossibly low budget
        server.succeed(f"""
          systemd-run --unit=monitor-exhaust \
            --setenv=JELLYFIN_URL=http://localhost:8096 \
            --setenv=JELLYFIN_API_KEY={token} \
            --setenv=QBITTORRENT_URL=http://localhost:8080 \
            --setenv=CHECK_INTERVAL=1 \
            --setenv=STREAMING_START_DELAY=1 \
            --setenv=STREAMING_STOP_DELAY=1 \
            --setenv=TOTAL_BANDWIDTH_BUDGET=1000 \
            --setenv=SERVICE_BUFFER=500 \
            --setenv=DEFAULT_STREAM_BITRATE=10000000 \
            --setenv=MIN_TORRENT_SPEED=100 \
            {python} {monitor}
        """)
        time.sleep(2)

        # Start a stream - this will exceed the tiny budget
        playback_start = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-exhaust",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_start)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        time.sleep(3)

        assert are_torrents_paused(), "Torrents should be paused when budget is exhausted"

    with subtest("Recovery from pause restores unlimited"):
        # Stop the stream
        playback_stop = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-exhaust",
            "PositionTicks": 50000000,
        }
        stop_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(playback_stop)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(stop_cmd)
        time.sleep(3)

        assert not is_throttled(), "Should return to unlimited after streams stop"
        assert not are_torrents_paused(), "Torrents should be resumed after streams stop"

        # Clean up: stop exhaust monitor, restart normal monitor
        server.succeed("systemctl stop monitor-exhaust || true")
        time.sleep(1)
        server.succeed(f"""
          systemd-run --unit=monitor-test \
            --setenv=JELLYFIN_URL=http://localhost:8096 \
            --setenv=JELLYFIN_API_KEY={token} \
            --setenv=QBITTORRENT_URL=http://localhost:8080 \
            --setenv=CHECK_INTERVAL=1 \
            --setenv=STREAMING_START_DELAY=1 \
            --setenv=STREAMING_STOP_DELAY=1 \
            --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
            --setenv=SERVICE_BUFFER=2000000 \
            --setenv=DEFAULT_STREAM_BITRATE=10000000 \
            --setenv=MIN_TORRENT_SPEED=100 \
            {python} {monitor}
        """)
        time.sleep(2)

    with subtest("Local playback does NOT trigger throttling"):
        local_auth = 'MediaBrowser Client="Local Client", DeviceId="local-1111", Device="LocalDevice", Version="1.0"'
        local_auth_result = json.loads(server.succeed(
            f"curl -sf -X POST 'http://localhost:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{local_auth}'"
        ))
        local_token = local_auth_result["AccessToken"]

        local_playback = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-local",
            "CanSeek": True,
            "IsPaused": False,
        }
        server.succeed(f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing' -d '{json.dumps(local_playback)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{local_auth}, Token={local_token}'")
        time.sleep(2)
        assert not is_throttled(), "Should NOT throttle for local playback"

        local_playback["PositionTicks"] = 50000000
        server.succeed(f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing/Stopped' -d '{json.dumps(local_playback)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{local_auth}, Token={local_token}'")

    with subtest("Hairpin'd LAN session (source IP = configured gateway) DOES throttle"):
        # Simulates a LAN client reaching Jellyfin via the public hostname:
        # the router SNATs the source to itself, so Jellyfin sees the gateway
        # IP and IsInLocalNetwork=True even though WAN bandwidth is in play.
        # We use 127.0.0.1 as the "gateway" in this VM because the localhost
        # curl below produces source 127.0.0.1 from Jellyfin's view.
        server.succeed("systemctl stop monitor-test || true")
        time.sleep(1)
        server.succeed(f"""
          systemd-run --unit=monitor-hairpin \
            --setenv=JELLYFIN_URL=http://localhost:8096 \
            --setenv=JELLYFIN_API_KEY={token} \
            --setenv=QBITTORRENT_URL=http://localhost:8080 \
            --setenv=CHECK_INTERVAL=1 \
            --setenv=STREAMING_START_DELAY=1 \
            --setenv=STREAMING_STOP_DELAY=1 \
            --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
            --setenv=SERVICE_BUFFER=2000000 \
            --setenv=DEFAULT_STREAM_BITRATE=10000000 \
            --setenv=MIN_TORRENT_SPEED=100 \
            --setenv=LAN_GATEWAY_IP=127.0.0.1 \
            {python} {monitor}
        """)
        time.sleep(2)
        assert not is_throttled(), "Should start unthrottled (no streams yet)"

        hairpin_auth = 'MediaBrowser Client="Hairpin Client", DeviceId="hairpin-2222", Device="HairpinDevice", Version="1.0"'
        hairpin_auth_result = json.loads(server.succeed(
            f"curl -sf -X POST 'http://localhost:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{hairpin_auth}'"
        ))
        hairpin_token = hairpin_auth_result["AccessToken"]

        hairpin_playback = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-hairpin",
            "CanSeek": True,
            "IsPaused": False,
        }
        server.succeed(f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing' -d '{json.dumps(hairpin_playback)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{hairpin_auth}, Token={hairpin_token}'")
        time.sleep(3)
        assert is_throttled(), "Hairpin'd session (source=gateway) should throttle even though source is RFC1918"

        # Cleanup: stop the playback and the override-monitor, restore the normal one.
        hairpin_playback["PositionTicks"] = 50000000
        server.succeed(f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing/Stopped' -d '{json.dumps(hairpin_playback)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{hairpin_auth}, Token={hairpin_token}'")
        time.sleep(2)
        assert not is_throttled(), "Should unthrottle after hairpin'd playback stops"

        server.succeed("systemctl stop monitor-hairpin || true")
        time.sleep(1)
        server.succeed(f"""
          systemd-run --unit=monitor-test \
            --setenv=JELLYFIN_URL=http://localhost:8096 \
            --setenv=JELLYFIN_API_KEY={token} \
            --setenv=QBITTORRENT_URL=http://localhost:8080 \
            --setenv=CHECK_INTERVAL=1 \
            --setenv=STREAMING_START_DELAY=1 \
            --setenv=STREAMING_STOP_DELAY=1 \
            --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
            --setenv=SERVICE_BUFFER=2000000 \
            --setenv=DEFAULT_STREAM_BITRATE=10000000 \
            --setenv=MIN_TORRENT_SPEED=100 \
            {python} {monitor}
        """)
        time.sleep(2)

    # === WEBHOOK TESTS ===
    #
    # Configure the Jellyfin Webhook plugin to target the monitor, then verify
    # the real Jellyfin → plugin → monitor path reacts faster than any possible
    # poll. CHECK_INTERVAL=30 rules out polling as the cause.

    WEBHOOK_PORT = 9898
    WEBHOOK_CREDS = "/tmp/webhook-creds"

    # Start a webhook-enabled monitor with long poll interval.
    server.succeed("systemctl stop monitor-test || true")
    time.sleep(1)
    server.succeed(f"""
      systemd-run --unit=monitor-webhook \
        --setenv=JELLYFIN_URL=http://localhost:8096 \
        --setenv=JELLYFIN_API_KEY={token} \
        --setenv=QBITTORRENT_URL=http://localhost:8080 \
        --setenv=CHECK_INTERVAL=30 \
        --setenv=STREAMING_START_DELAY=1 \
        --setenv=STREAMING_STOP_DELAY=1 \
        --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
        --setenv=SERVICE_BUFFER=2000000 \
        --setenv=DEFAULT_STREAM_BITRATE=10000000 \
        --setenv=MIN_TORRENT_SPEED=100 \
        --setenv=WEBHOOK_PORT={WEBHOOK_PORT} \
        --setenv=WEBHOOK_BIND=127.0.0.1 \
        {python} {monitor}
    """)
    server.wait_until_succeeds(f"ss -ltn | grep -q ':{WEBHOOK_PORT}'", timeout=15)
    time.sleep(2)
    assert not is_throttled(), "Should start unthrottled"

    # Drop the admin token where the configure script expects it (production uses agenix).
    server.succeed(f"mkdir -p {WEBHOOK_CREDS} && echo '{token}' > {WEBHOOK_CREDS}/jellyfin-api-key")
    server.succeed(
        f"systemd-run --wait --unit=webhook-configure-test "
        f"--setenv=CREDENTIALS_DIRECTORY={WEBHOOK_CREDS} "
        f"${configureWebhook}"
    )

    with subtest("Real PlaybackStart event throttles via the plugin"):
        playback_start = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-plugin-start",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_start)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        server.wait_until_succeeds(
            "curl -sf http://localhost:8080/api/v2/transfer/speedLimitsMode | grep -q '^1$'",
            timeout=5,
        )
        # Let STREAMING_STOP_DELAY (1s) elapse so the upcoming stop is not swallowed by hysteresis.
        time.sleep(2)

    with subtest("Real PlaybackStop event unthrottles via the plugin"):
        playback_stop = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-plugin-start",
            "PositionTicks": 50000000,
        }
        stop_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(playback_stop)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(stop_cmd)
        server.wait_until_succeeds(
            "curl -sf http://localhost:8080/api/v2/transfer/speedLimitsMode | grep -q '^0$'",
            timeout=10,
        )

    # Restore fast-polling monitor for the service-restart tests below.
    server.succeed("systemctl stop monitor-webhook || true")
    time.sleep(1)
    server.succeed(f"""
      systemd-run --unit=monitor-test \
        --setenv=JELLYFIN_URL=http://localhost:8096 \
        --setenv=JELLYFIN_API_KEY={token} \
        --setenv=QBITTORRENT_URL=http://localhost:8080 \
        --setenv=CHECK_INTERVAL=1 \
        --setenv=STREAMING_START_DELAY=1 \
        --setenv=STREAMING_STOP_DELAY=1 \
        --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
        --setenv=SERVICE_BUFFER=2000000 \
        --setenv=DEFAULT_STREAM_BITRATE=10000000 \
        --setenv=MIN_TORRENT_SPEED=100 \
        {python} {monitor}
    """)
    time.sleep(2)


    # === FLICKER TEST ===
    #
    # Session flicker (brief appear→disappear→reappear cycles) is the
    # production reality when xmrig starves the transcode CPU and the
    # client's buffer oscillates. The hysteresis timer must measure
    # CONTIGUOUS absence, not cumulative gaps across flickers. A 5s
    # stop delay with flickers of 2s absent / 2s present must NOT
    # accumulate into a spurious unlimited transition.

    server.succeed("systemctl stop monitor-test || true")
    time.sleep(1)
    server.succeed(f"""
      systemd-run --unit=monitor-flicker \
        --setenv=JELLYFIN_URL=http://localhost:8096 \
        --setenv=JELLYFIN_API_KEY={token} \
        --setenv=QBITTORRENT_URL=http://localhost:8080 \
        --setenv=CHECK_INTERVAL=1 \
        --setenv=STREAMING_START_DELAY=1 \
        --setenv=STREAMING_STOP_DELAY=5 \
        --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
        --setenv=SERVICE_BUFFER=2000000 \
        --setenv=DEFAULT_STREAM_BITRATE=10000000 \
        --setenv=MIN_TORRENT_SPEED=100 \
        {python} {monitor}
    """)
    time.sleep(2)
    assert not is_throttled(), "Should start unthrottled"

    with subtest("Session flicker does not accumulate hysteresis timer"):
        # Start playback
        playback_flicker = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-flicker",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_flicker)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        time.sleep(2)
        assert is_throttled(), "Should throttle when streaming starts"

        # Flicker 1: stop 2s, restart 2s
        playback_flicker["PositionTicks"] = 50000000
        stop_flicker = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(playback_flicker)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(stop_flicker)
        time.sleep(2)
        client.succeed(start_cmd)
        time.sleep(2)

        # Flicker 2: stop 2s, restart 2s (total absent = 4s < 5s delay)
        client.succeed(stop_flicker)
        time.sleep(2)
        client.succeed(start_cmd)
        time.sleep(2)

        # Should still be throttled — flicker gaps didn't accumulate
        assert is_throttled(), "Should remain throttled after session flickers (contiguous absence < 5s)"

        # Genuine stop: absent for full stop delay
        client.succeed(stop_flicker)
        time.sleep(6)
        assert not is_throttled(), "Should unthrottle after contiguous 5s+ absence"

        # Restart after genuine stop
        client.succeed(start_cmd)
        time.sleep(2)
        assert is_throttled(), "Should re-throttle after restart"

        # Clean up
        client.succeed(stop_flicker)
        time.sleep(2)

    # Restore the fast-polling monitor for subsequent tests
    server.succeed("systemctl stop monitor-flicker || true")
    time.sleep(1)
    server.succeed(f"""
      systemd-run --unit=monitor-test \
        --setenv=JELLYFIN_URL=http://localhost:8096 \
        --setenv=JELLYFIN_API_KEY={token} \
        --setenv=QBITTORRENT_URL=http://localhost:8080 \
        --setenv=CHECK_INTERVAL=1 \
        --setenv=STREAMING_START_DELAY=1 \
        --setenv=STREAMING_STOP_DELAY=1 \
        --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
        --setenv=SERVICE_BUFFER=2000000 \
        --setenv=DEFAULT_STREAM_BITRATE=10000000 \
        --setenv=MIN_TORRENT_SPEED=100 \
        {python} {monitor}
    """)
    time.sleep(2)
    # === SERVICE RESTART TESTS ===

    with subtest("qBittorrent restart during throttled state re-applies throttling"):
        # Start external playback to trigger throttling
        playback_start = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-restart-1",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_start)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        time.sleep(2)
        assert is_throttled(), "Should be throttled before qBittorrent restart"

        # Restart qBittorrent (this resets alt_speed to its config default - disabled)
        server.succeed("systemctl restart qbittorrent.service")
        server.wait_for_unit("qbittorrent.service")
        server.wait_for_open_port(8080)
        server.wait_until_succeeds("curl -sf http://localhost:8080/api/v2/app/version", timeout=30)

        # qBittorrent restarted - alt_speed is now False (default on startup)
        # The monitor should detect this and re-apply throttling
        time.sleep(3)  # Give monitor time to detect and re-apply
        assert is_throttled(), "Monitor should re-apply throttling after qBittorrent restart"

        # Stop playback to clean up
        playback_stop = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-restart-1",
            "PositionTicks": 50000000,
        }
        stop_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' -d '{json.dumps(playback_stop)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(stop_cmd)
        time.sleep(2)

    with subtest("qBittorrent restart during unthrottled state stays unthrottled"):
        # Verify we're unthrottled (no active streams)
        assert not is_throttled(), "Should be unthrottled before test"

        # Restart qBittorrent
        server.succeed("systemctl restart qbittorrent.service")
        server.wait_for_unit("qbittorrent.service")
        server.wait_for_open_port(8080)
        server.wait_until_succeeds("curl -sf http://localhost:8080/api/v2/app/version", timeout=30)

        # Give monitor time to check state
        time.sleep(3)
        assert not is_throttled(), "Should remain unthrottled after qBittorrent restart with no streams"

    with subtest("Jellyfin restart during throttled state maintains throttling"):
        # Start external playback to trigger throttling
        playback_start = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-restart-2",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_start)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        time.sleep(2)
        assert is_throttled(), "Should be throttled before Jellyfin restart"

        # Restart Jellyfin
        server.succeed("systemctl restart jellyfin.service")
        server.wait_for_unit("jellyfin.service")
        server.wait_for_open_port(8096)
        server.wait_until_succeeds("curl -sf http://localhost:8096/health | grep -q Healthy", timeout=180)

        # During Jellyfin restart, monitor can't reach Jellyfin
        # After restart, sessions are cleared - monitor should eventually unthrottle
        # But during the unavailability window, throttling should be maintained (fail-safe)
        time.sleep(3)

        # Re-authenticate (old token invalid after restart)
        client_auth_result = json.loads(client.succeed(
            f"curl -sf -X POST 'http://{server_ip}:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}'"
        ))
        client_token = client_auth_result["AccessToken"]
        client_auth_result2 = json.loads(client.succeed(
            f"curl -sf -X POST 'http://{server_ip}:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth2}'"
        ))
        client_token2 = client_auth_result2["AccessToken"]

        # No active streams after Jellyfin restart, should eventually unthrottle
        time.sleep(3)
        assert not is_throttled(), "Should unthrottle after Jellyfin restart clears sessions"

    with subtest("Monitor recovers after Jellyfin temporary unavailability"):
        # Re-authenticate with fresh token
        client_auth_result = json.loads(client.succeed(
            f"curl -sf -X POST 'http://{server_ip}:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}'"
        ))
        client_token = client_auth_result["AccessToken"]
        client_auth_result2 = json.loads(client.succeed(
            f"curl -sf -X POST 'http://{server_ip}:8096/Users/AuthenticateByName' -d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth2}'"
        ))
        client_token2 = client_auth_result2["AccessToken"]

        # Start playback
        playback_start = {
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-restart-3",
            "CanSeek": True,
            "IsPaused": False,
        }
        start_cmd = f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' -d '{json.dumps(playback_start)}' -H 'Content-Type:application/json' -H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        client.succeed(start_cmd)
        time.sleep(2)
        assert is_throttled(), "Should be throttled"

        # Stop Jellyfin briefly (simulating temporary unavailability)
        server.succeed("systemctl stop jellyfin.service")
        time.sleep(2)

        # During unavailability, throttle state should be maintained (fail-safe)
        assert is_throttled(), "Should maintain throttle during Jellyfin unavailability"

        # Bring Jellyfin back
        server.succeed("systemctl start jellyfin.service")
        server.wait_for_unit("jellyfin.service")
        server.wait_for_open_port(8096)
        server.wait_until_succeeds("curl -sf http://localhost:8096/health | grep -q Healthy", timeout=180)

        # After Jellyfin comes back, sessions are gone - should unthrottle
        time.sleep(3)
        assert not is_throttled(), "Should unthrottle after Jellyfin returns with no sessions"

    # === SLSKD UPLOAD BANDWIDTH TESTS ===

    # Spin up a mock slskd API server that simulates active uploads. The
    # monitor should query it and subtract upload bandwidth from the torrent
    # budget alongside Jellyfin streaming.

    mock_slskd_port = 9999


    # Start a mock slskd server. It checks the X-API-Key header to verify
    # the monitor sends credentials correctly, and returns 401 if they mismatch.
    mock_slskd_key = "test-slskd-api-key"
    server.succeed(
        f"{pkgs.python3}/bin/python -c \""
        "import json, http.server, threading;"
        f"EXPECTED_KEY='{mock_slskd_key}';"
        "class H(http.server.BaseHTTPRequestHandler):"
        " uploads_response=[];"
        " def do_GET(s):"
        "  if s.path=='/api/v0/transfers/uploads':"
        "   if s.headers.get('X-API-Key')!=EXPECTED_KEY:"
        "    s.send_error(401);"
        "    return;"
        "   s.send_response(200);"
        "   s.send_header('Content-Type','application/json');"
        "   s.end_headers();"
        "   s.wfile.write(json.dumps(H.uploads_response).encode());"
        "  else: s.send_error(404);"
        " def log_message(*a): pass;"
        f"server=http.server.HTTPServer(('127.0.0.1',{mock_slskd_port}),H);"
        "threading.Thread(target=server.serve_forever,daemon=True).start();"
        "import signal; signal.pause()\" &"
    )
    server.wait_for_open_port(mock_slskd_port)
    server.succeed(f"curl -sf http://127.0.0.1:{mock_slskd_port}/api/v0/transfers/uploads")

    # Stop the normal monitor and start one pointed at the mock slskd
    server.succeed("systemctl stop monitor-test || true")
    time.sleep(1)
    server.succeed(f"""
      systemd-run --unit=monitor-slskd \
        --setenv=JELLYFIN_URL=http://localhost:8096 \
        --setenv=JELLYFIN_API_KEY={token} \
        --setenv=QBITTORRENT_URL=http://localhost:8080 \
        --setenv=CHECK_INTERVAL=1 \
        --setenv=STREAMING_START_DELAY=1 \
        --setenv=STREAMING_STOP_DELAY=1 \
        --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
        --setenv=SERVICE_BUFFER=2000000 \
        --setenv=DEFAULT_STREAM_BITRATE=10000000 \
        --setenv=MIN_TORRENT_SPEED=100 \
        --setenv=SLSKD_URL=http://127.0.0.1:{mock_slskd_port} \
        --setenv=SLSKD_API_KEY={mock_slskd_key} \
        {python} {monitor}
    """)
    time.sleep(2)
    assert not is_throttled(), "Should start unthrottled (slskd reports no uploads)"

    with subtest("Slskd uploads reduce torrent bandwidth budget"):
        # Re-authenticate to get fresh token after previous tests
        client_auth_result = json.loads(client.succeed(
            f"curl -sf -X POST 'http://{server_ip}:8096/Users/AuthenticateByName' "
            f"-d '@${jfLib.payloads.auth}' "
            "-H 'Content-Type:application/json' "
            f"-H 'X-Emby-Authorization:{client_auth}'"
        ))
        client_token = client_auth_result["AccessToken"]

        # Start a single Jellyfin stream to establish a baseline torrent limit
        playback_start = {{
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-slskd",
            "CanSeek": True,
            "IsPaused": False,
        }}
        start_cmd = (
            f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing' "
            f"-d '{json.dumps(playback_start)}' "
            "-H 'Content-Type:application/json' "
            f"-H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        )
        client.succeed(start_cmd)
        time.sleep(3)

        assert is_throttled(), "Should throttle with streaming"
        baseline_dl = get_alt_dl_limit()

        # Now simulate active slskd uploads consuming 15 Mbps (1,875,000 bytes/s)
        # by restarting the mock with two in-progress uploads at 937,500 bytes/s each.
        server.succeed("pkill -f 'H(http' || true")
        time.sleep(1)

        # Restart the mock server with upload data; still validates the API key.
        server.succeed(
            f"{pkgs.python3}/bin/python -c \""
            "import json, http.server, threading;"
            f"EXPECTED_KEY='{mock_slskd_key}';"
            "class H(http.server.BaseHTTPRequestHandler):"
            " uploads_response=[{'username':'peer1','directories':[{'directory':'music',"
            "'fileCount':2,'files':["
            "{'averageSpeed':937500,'state':'InProgress','filename':'t1.flac','size':30000000,'username':'peer1'},"
            "{'averageSpeed':937500,'state':'InProgress','filename':'t2.flac','size':25000000,'username':'peer1'}"
            "]}]];"
            " def do_GET(s):"
            "  if s.path=='/api/v0/transfers/uploads':"
            "   if s.headers.get('X-API-Key')!=EXPECTED_KEY:"
            "    s.send_error(401);"
            "    return;"
            "   s.send_response(200);"
            "   s.send_header('Content-Type','application/json');"
            "   s.end_headers();"
            "   s.wfile.write(json.dumps(H.uploads_response).encode());"
            "  else: s.send_error(404);"
            " def log_message(*a): pass;"
            f"server=http.server.HTTPServer(('127.0.0.1',{mock_slskd_port}),H);"
            "threading.Thread(target=server.serve_forever,daemon=True).start();"
            "import signal; signal.pause()\" &"
        )
        server.wait_for_open_port(mock_slskd_port)
        time.sleep(4)  # Let monitor poll and adjust

        slskd_dl = get_alt_dl_limit()
        # With 15 Mbps of slskd uploads consuming the shared budget, the
        # remaining bandwidth for torrents must be lower than baseline.
        assert slskd_dl < baseline_dl, \
            f"Slskd uploads should reduce torrent budget. baseline={baseline_dl}, with_slskd={slskd_dl}"

        # Stop Jellyfin playback
        playback_stop = {{
            "ItemId": movie_id,
            "MediaSourceId": media_source_id,
            "PlaySessionId": "test-play-session-slskd",
            "PositionTicks": 50000000,
        }}
        stop_cmd = (
            f"curl -sf -X POST 'http://{server_ip}:8096/Sessions/Playing/Stopped' "
            f"-d '{json.dumps(playback_stop)}' "
            "-H 'Content-Type:application/json' "
            f"-H 'X-Emby-Authorization:{client_auth}, Token={client_token}'"
        )
        client.succeed(stop_cmd)
        time.sleep(3)

    # Clean up: stop mock slskd and restore normal monitor
    server.succeed("pkill -f 'H(http' || true")
    server.succeed("systemctl stop monitor-slskd || true")
    time.sleep(1)
    server.succeed(f"""
      systemd-run --unit=monitor-test \
        --setenv=JELLYFIN_URL=http://localhost:8096 \
        --setenv=JELLYFIN_API_KEY={token} \
        --setenv=QBITTORRENT_URL=http://localhost:8080 \
        --setenv=CHECK_INTERVAL=1 \
        --setenv=STREAMING_START_DELAY=1 \
        --setenv=STREAMING_STOP_DELAY=1 \
        --setenv=TOTAL_BANDWIDTH_BUDGET=50000000 \
        --setenv=SERVICE_BUFFER=2000000 \
        --setenv=DEFAULT_STREAM_BITRATE=10000000 \
        --setenv=MIN_TORRENT_SPEED=100 \
        {python} {monitor}
    """)
    time.sleep(2)
  '';
}