diff --git a/plugins/in_blob/win32_glob.c b/plugins/in_blob/win32_glob.c
index 9bba4ca1e1c..f4bb62afb7a 100644
--- a/plugins/in_blob/win32_glob.c
+++ b/plugins/in_blob/win32_glob.c
@@ -112,6 +112,7 @@ static void globfree(glob_t *context)
         flb_free(context->gl_pathv);
         context->gl_pathv = NULL;
     }
+    context->gl_pathc = 0;
 
     flb_file_glob_clean(&context->inner_context);
 }
@@ -127,7 +128,7 @@ static int glob(const char *path,
 
     (void) unused;
 
-    result = flb_file_glob_start(context, path, flags);
+    result = flb_file_glob_start(&context->inner_context, path, flags);
 
     if (result == FLB_FILE_GLOB_ERROR_SUCCESS) {
         entries = cfl_list_size(&context->inner_context.inner_context->results);
@@ -150,6 +151,7 @@ static int glob(const char *path,
                 return result;
             }
         }
+        context->gl_pathc = entries;
     }
 
     return result;
@@ -157,7 +159,7 @@ static int glob(const char *path,
 
 static int is_directory(char *path, struct stat *fs_entry_metadata)
 {
-    return (fs_entry_metadata->st_mode & S_IFDIR != 0);
+    return (fs_entry_metadata->st_mode & S_IFMT) == S_IFDIR;
 }
 
 static void reset_errno()
@@ -604,4 +606,4 @@ int flb_file_glob_fetch(struct flb_file_glob_context *context,
     return FLB_FILE_GLOB_ERROR_SUCCESS;
 }
 
-#endif
\ No newline at end of file
+#endif
diff --git a/src/config_format/flb_cf_fluentbit.c b/src/config_format/flb_cf_fluentbit.c
index 8f3ee566161..ee193a3435d 100644
--- a/src/config_format/flb_cf_fluentbit.c
+++ b/src/config_format/flb_cf_fluentbit.c
@@ -191,11 +191,28 @@ static int read_glob(struct flb_cf *cf, struct local_ctx *ctx, const char *path,
     char *star, *p0, *p1;
     char pattern[MAX_PATH];
     char buf[MAX_PATH];
+    char full_path[MAX_PATH];
     int ret;
     struct stat st;
     HANDLE h;
     WIN32_FIND_DATA data;
 
+    if (ctx->root_path && path[0] != '/' && path[0] != '\\' &&
+        !(path[0] != '\0' && path[1] == ':')) {
+        ret = snprintf(buf, sizeof(buf), "%s\\%s", ctx->root_path, path);
+        if (ret < 0 || ret >= sizeof(buf)) {
+            return -1;
+        }
+        path = buf;
+    }
+
+    /* Resolve relative includes against the config directory and normalize
+     * forward slashes before splitting the Windows wildcard components. */
+    if (_fullpath(full_path, path, sizeof(full_path)) == NULL) {
+        return -1;
+    }
+    path = full_path;
+
     if (strlen(path) > MAX_PATH - 1) {
         return -1;
     }
@@ -351,7 +368,11 @@ static int is_file_included(struct local_ctx *ctx, const char *path)
 
     mk_list_foreach(head, &ctx->includes) {
         file = mk_list_entry(head, struct local_file, _head);
+#ifdef _WIN32
+        if (strcasecmp(file->path, path) == 0) {
+#else
         if (strcmp(file->path, path) == 0) {
+#endif
             return FLB_TRUE;
         }
     }
@@ -416,14 +437,15 @@ static int read_config(struct flb_cf *cf, struct local_ctx *ctx,
     int key_len;
     char *val = NULL;
     int val_len;
-    char *buf;
+    char *buf = NULL;
     char *fgets_ptr;
     size_t bufsize = FLB_DEFAULT_CF_BUF_SIZE;
     char tmp[PATH_MAX];
+    char *full_path = NULL;
     flb_sds_t section = NULL;
     flb_sds_t indent = NULL;
     struct stat st;
-    struct local_file *file;
+    struct local_file *file = NULL;
     struct flb_cf_meta *meta;
     struct flb_cf_section *current_section = NULL;
     struct flb_cf_group *current_group = NULL;
@@ -473,22 +495,46 @@ static int read_config(struct flb_cf *cf, struct local_ctx *ctx,
     }
 #endif
 
-    /* Check this file have not been included before */
+#if defined(_WIN32) && !defined(FLB_HAVE_STATIC_CONF)
+    full_path = realpath(cfg_file, NULL);
+    if (full_path == NULL) {
+        return -1;
+    }
+    cfg_file = full_path;
+#endif
+
+    /* Check both active and previously parsed files. */
     ret = is_file_included(ctx, cfg_file);
     if (ret) {
         flb_error("[config] file already included %s", cfg_file);
+        flb_free(full_path);
         return -1;
     }
-    ctx->level++;
 
 #ifndef FLB_HAVE_STATIC_CONF
     /* Open configuration file */
     if ((f = fopen(cfg_file, "rb")) == NULL) {
         flb_warn("[config] I cannot open %s file", cfg_file);
+        flb_free(full_path);
         return -1;
     }
 #endif
 
+    ctx->level++;
+    file = flb_malloc(sizeof(struct local_file));
+    if (!file) {
+        flb_errno();
+        goto error;
+    }
+    file->path = flb_sds_create(cfg_file);
+    if (!file->path) {
+        flb_free(file);
+        file = NULL;
+        goto error;
+    }
+    /* Register before parsing so recursive includes cannot re-enter this file. */
+    mk_list_add(&file->_head, &ctx->includes);
+
     /* Allocate temporal buffer to read file content */
     buf = flb_malloc(bufsize);
     if (!buf) {
@@ -744,20 +790,20 @@ static int read_config(struct flb_cf *cf, struct local_ctx *ctx,
     }
     flb_free(buf);
 
-    /* Append this file to the list */
-    file = flb_malloc(sizeof(struct local_file));
-    if (!file) {
-        flb_errno();
-        ctx->level--;
-        goto error;
-    }
-    file->path = flb_sds_create(cfg_file);
-    mk_list_add(&file->_head, &ctx->includes);
+    /* Keep successful files registered to reject later duplicates. */
+    flb_free(full_path);
     ctx->level--;
 
     return 0;
 
 error:
+    flb_free(full_path);
+    if (file) {
+        mk_list_del(&file->_head);
+        flb_sds_destroy(file->path);
+        flb_free(file);
+    }
+    ctx->level--;
     if (f) {
         fclose(f);
     }
diff --git a/src/config_format/flb_cf_yaml.c b/src/config_format/flb_cf_yaml.c
index d1c76d83a12..cbef7015dca 100644
--- a/src/config_format/flb_cf_yaml.c
+++ b/src/config_format/flb_cf_yaml.c
@@ -485,7 +485,11 @@ static int is_file_included(struct local_ctx *ctx, const char *path)
     mk_list_foreach(head, &ctx->includes) {
         entry = mk_list_entry(head, struct flb_slist_entry, _head);
 
+#ifdef _WIN32
+        if (strcasecmp(entry->str, path) == 0) {
+#else
         if (strcmp(entry->str, path) == 0) {
+#endif
             return FLB_TRUE;
         }
     }
@@ -574,11 +578,28 @@ static int read_glob(struct flb_cf *conf, struct local_ctx *ctx,
     char *star, *p0, *p1;
     char pattern[MAX_PATH];
     char buf[MAX_PATH];
+    char full_path[MAX_PATH];
     int ret;
     struct stat st;
     HANDLE hnd;
     WIN32_FIND_DATA data;
 
+    if (state->file->path && path[0] != '/' && path[0] != '\\' &&
+        !(path[0] != '\0' && path[1] == ':')) {
+        ret = snprintf(buf, sizeof(buf), "%s\\%s", state->file->path, path);
+        if (ret < 0 || ret >= sizeof(buf)) {
+            return -1;
+        }
+        path = buf;
+    }
+
+    /* Resolve relative includes against the config directory and normalize
+     * forward slashes before splitting the Windows wildcard components. */
+    if (_fullpath(full_path, path, sizeof(full_path)) == NULL) {
+        return -1;
+    }
+    path = full_path;
+
     if (strlen(path) > MAX_PATH - 1) {
         return -1;
     }
@@ -3131,6 +3152,9 @@ static int read_config(struct flb_cf *conf, struct local_ctx *ctx,
     yaml_event_t event;
     FILE *fh;
     struct file_state fstate;
+#ifdef _WIN32
+    char *full_path;
+#endif
 
     path_is_absolute = cfg_file[0] == '/';
 #ifdef _WIN32
@@ -3180,6 +3204,21 @@ static int read_config(struct flb_cf *conf, struct local_ctx *ctx,
         }
     }
 
+#ifdef _WIN32
+    /* Use the same absolute identity for direct and wildcard includes. */
+    full_path = realpath(include_dir, NULL);
+    if (full_path == NULL) {
+        flb_sds_destroy(include_dir);
+        return -1;
+    }
+    flb_sds_destroy(include_dir);
+    include_dir = flb_sds_create(full_path);
+    flb_free(full_path);
+    if (include_dir == NULL) {
+        return -1;
+    }
+#endif
+
     include_file = flb_sds_create(include_dir);
 
     if (include_file == NULL) {
diff --git a/src/flb_env.c b/src/flb_env.c
index 5293f01aea8..38a40a05ba1 100644
--- a/src/flb_env.c
+++ b/src/flb_env.c
@@ -118,6 +118,25 @@ void flb_env_destroy(struct flb_env *env)
     flb_free(env);
 }
 
+static const char *env_file_path(const char *uri)
+{
+    const char *path = uri;
+
+    if (strncmp(path, "file://", 7) == 0) {
+        path += 7;
+    }
+#ifdef FLB_SYSTEM_WINDOWS
+    /* A local drive URI uses file:///C:/path; the filesystem needs C:/path. */
+    if (path[0] == '/' &&
+        ((path[1] >= 'A' && path[1] <= 'Z') || (path[1] >= 'a' && path[1] <= 'z')) &&
+        path[2] == ':' && (path[3] == '/' || path[3] == '\\')) {
+        path++;
+    }
+#endif
+
+    return path;
+}
+
 int flb_env_set_extended(struct flb_env *env, const char *key, const char *val,
                          const char *uri, int refresh_interval)
 {
@@ -146,8 +165,7 @@ int flb_env_set_extended(struct flb_env *env, const char *key, const char *val,
     /* Check if the variable is a reference to a file */
     if (vlen > 7 && strncmp(value, "file://", 7) == 0) {
         orig_uri = value;
-        vlen -= 7;
-        value += 7;
+        value = env_file_path(value);
 
         if (access(value, R_OK) == -1) {
             flb_error("[env] file %s not found", value);
@@ -247,10 +265,7 @@ const char *flb_env_get(struct flb_env *env, const char *key)
                 now = time(NULL);
                 if (var->last_refresh == 0 ||
                     now - var->last_refresh >= var->refresh_interval) {
-                    file = var->uri;
-                    if (strncmp(file, "file://", 7) == 0) {
-                        file += 7;
-                    }
+                    file = env_file_path(var->uri);
 
                     if (access(file, R_OK) == -1) {
                         flb_error("[env] file %s not found", file);
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index b75fc5692a4..2c3de41d2a0 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -22,21 +22,9 @@
 from utils.fluent_bit_manager import fluent_bit_binary_version
 from utils.test_service import stop_active_services
 
-# Configure logging
-def configure_logging():
-    logger = logging.getLogger(__name__)
-    logger.setLevel(logging.INFO)
-
-    if not logger.handlers:
-        handler = logging.StreamHandler()
-        handler.setLevel(logging.INFO)
-        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
-        handler.setFormatter(formatter)
-        logger.addHandler(handler)
-
-    return logger
-
-logger = configure_logging()
+# Let pytest manage handlers so none retain a closed capture stream at shutdown.
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
 
 def load_global_config():
     config_file = os.path.abspath(os.path.join(os.path.dirname(__file__), 'config.yaml'))
diff --git a/tests/integration/requirements.txt b/tests/integration/requirements.txt
index 6da97e133bf..73a3aaa1749 100644
--- a/tests/integration/requirements.txt
+++ b/tests/integration/requirements.txt
@@ -9,7 +9,7 @@ Deprecated==1.2.14
 Flask==3.0.3
 google==3.0.0
 googleapis-common-protos==1.63.2
-grpcio==1.64.1
+grpcio==1.85.0
 h2==4.1.0
 idna==3.7
 importlib_metadata==7.1.0
diff --git a/tests/integration/scenarios/hot_reload_watch/tests/test_hot_reload_watch_001.py b/tests/integration/scenarios/hot_reload_watch/tests/test_hot_reload_watch_001.py
index 1e801470017..39dccd37d57 100644
--- a/tests/integration/scenarios/hot_reload_watch/tests/test_hot_reload_watch_001.py
+++ b/tests/integration/scenarios/hot_reload_watch/tests/test_hot_reload_watch_001.py
@@ -146,6 +146,7 @@ def test_hot_reload_watch_yaml_config_change():
         service.stop()
 
 
+@pytest.mark.skipif(os.name == "nt", reason="SIGHUP reload requires POSIX; Windows uses HTTP reload")
 def test_hot_reload_sighup_yaml_config_change():
     service = Service("fluent-bit-manual-before.yaml", "fluent-bit-manual-after.yaml")
 
diff --git a/tests/integration/scenarios/hot_reload_watch/tests/test_include_identity.py b/tests/integration/scenarios/hot_reload_watch/tests/test_include_identity.py
new file mode 100644
index 00000000000..ab2a3164740
--- /dev/null
+++ b/tests/integration/scenarios/hot_reload_watch/tests/test_include_identity.py
@@ -0,0 +1,69 @@
+"""Configuration includes must reject cycles and aliases before adding plugins."""
+
+import os
+import subprocess
+
+import pytest
+
+from utils.fluent_bit_manager import FluentBitManager
+
+
+def run_config(config, cwd, relative):
+    binary = FluentBitManager(str(config)).binary_absolute_path
+    argument = os.path.relpath(config, cwd) if relative else str(config)
+    return subprocess.run(
+        [binary, "--dry-run", "-c", argument], cwd=cwd,
+        capture_output=True, text=True, timeout=10,
+    )
+
+
+def include_text(config_format, paths):
+    if config_format == "conf":
+        return "".join(f"@INCLUDE {path}\n" for path in paths)
+    return "includes:\n" + "".join(f"  - '{path}'\n" for path in paths)
+
+
+@pytest.mark.parametrize("config_format", ["conf", "yaml"])
+@pytest.mark.parametrize("relative", [False, True])
+@pytest.mark.parametrize("indirect", [False, True])
+def test_include_cycle_is_rejected(tmp_path, config_format, relative, indirect):
+    directory = tmp_path / "config"
+    directory.mkdir()
+    main = directory / f"main.{config_format}"
+    if indirect:
+        main.write_text(include_text(config_format, [f"other.{config_format}"]), encoding="utf-8")
+        (directory / f"other.{config_format}").write_text(
+            include_text(config_format, [f"./main.{config_format}"]), encoding="utf-8"
+        )
+    else:
+        main.write_text(include_text(config_format, [f"*.{config_format}"]), encoding="utf-8")
+
+    result = run_config(main, tmp_path, relative)
+    assert result.returncode != 0
+    assert "already included" in result.stderr or "Read twice" in result.stderr, result.stderr
+
+
+@pytest.mark.parametrize("config_format", ["conf", "yaml"])
+@pytest.mark.parametrize("relative", [False, True])
+@pytest.mark.parametrize("duplicate", [False, True])
+def test_direct_and_wildcard_include_identity(tmp_path, config_format, relative, duplicate):
+    directory = tmp_path / "config"
+    partials = directory / "partials"
+    partials.mkdir(parents=True)
+    main = directory / f"main.{config_format}"
+    child = partials / f"input.{config_format}"
+    child.write_text(
+        "[INPUT]\n    Name dummy\n" if config_format == "conf"
+        else "pipeline:\n  inputs:\n    - name: dummy\n", encoding="utf-8"
+    )
+    paths = [f"partials/input.{config_format}"]
+    if duplicate:
+        paths.append(f"partials/*.{config_format}")
+    main.write_text(include_text(config_format, paths), encoding="utf-8")
+
+    result = run_config(main, tmp_path, relative)
+    if duplicate:
+        assert result.returncode != 0
+        assert "already included" in result.stderr or "Read twice" in result.stderr, result.stderr
+    else:
+        assert result.returncode == 0, result.stderr
diff --git a/tests/integration/scenarios/in_collectd/tests/test_in_collectd.py b/tests/integration/scenarios/in_collectd/tests/test_in_collectd.py
index 2fef8031f86..62ef764f780 100644
--- a/tests/integration/scenarios/in_collectd/tests/test_in_collectd.py
+++ b/tests/integration/scenarios/in_collectd/tests/test_in_collectd.py
@@ -6,6 +6,13 @@
 import pytest
 
 from utils.test_service import FluentBitTestService
+from utils.fluent_bit_manager import fluent_bit_input_supports_config_property
+
+
+@pytest.fixture(autouse=True, scope="module")
+def require_collectd_input():
+    if not fluent_bit_input_supports_config_property("collectd", "typesdb"):
+        pytest.skip("This Fluent Bit build does not include the collectd input")
 
 
 def part(kind, payload):
diff --git a/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py b/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py
index a8e81184ecf..25eb52c37d4 100644
--- a/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py
+++ b/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py
@@ -3,7 +3,7 @@
 import pytest
 
 from server.otlp_server import data_storage
-from utils.http_matrix import PROTOCOL_CASES, run_curl_request
+from utils.http_matrix import PROTOCOL_CASES, curl_for_http_mode, run_curl_request
 from utils.input_pause_resume import (
     assert_pause_resume_cycles,
     assert_shutdown_while_paused,
@@ -359,7 +359,7 @@ def test_in_elasticsearch_bulk_small_status_buffer_does_not_crash(case):
 
     trigger_payload = '{"index":{}}\n{}\n{"index":{}}\n{}\n'
     curl_command = [
-        "curl",
+        curl_for_http_mode(case["http_mode"]),
         "--silent",
         "--show-error",
         "--output",
diff --git a/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py b/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py
index 91959495f28..9f779e05d4a 100644
--- a/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py
+++ b/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py
@@ -581,6 +581,14 @@ def _send_tls_payload(port, payload, cafile):
     with socket.create_connection(("127.0.0.1", port), timeout=20) as raw_sock:
         with context.wrap_socket(raw_sock, server_hostname="localhost") as tls_sock:
             tls_sock.sendall(payload)
+            # Complete TLS shutdown before closing the socket. In particular,
+            # Windows can reset a connection with unread TLS session tickets.
+            try:
+                tls_sock.unwrap().close()
+            except (ConnectionResetError, ssl.SSLEOFError):
+                # Pause tests deliberately make the server drop this connection;
+                # callers still verify ingestion or the expected pause state.
+                pass
 
 
 def _create_tls_memory_bio_client(port, cafile):
@@ -1442,6 +1450,7 @@ def test_in_forward_forced_input_tag_overrides_incoming_tag():
     assert records[0]["message"] == "forced-tag"
 
 
+@pytest.mark.skipif(not hasattr(socket, "AF_UNIX"), reason="Python requires Unix-domain socket support")
 def test_in_forward_unix_socket_message_mode():
     service = Service("in_forward_unix.yaml", use_unix_socket=True)
     service.start()
@@ -1462,6 +1471,7 @@ def test_in_forward_unix_socket_message_mode():
     assert records[0]["message"] == "unix-socket"
 
 
+@pytest.mark.skipif(os.name == "nt", reason="Unix socket permissions require POSIX")
 def test_in_forward_unix_socket_permissions():
     service = Service("in_forward_unix_perm.yaml", use_unix_socket=True)
     service.start()
@@ -1619,6 +1629,7 @@ def test_in_forward_downstream_coro_eof_and_error_keep_listener_responsive(
     assert records[0]["message"] == "after-io-errors"
 
 
+@pytest.mark.skipif(os.name == "nt", reason="Checks POSIX errno diagnostics; Windows uses Winsock errors")
 def test_in_forward_tls_syscall_error_preserves_errno():
     service = Service("in_forward_tls.yaml")
     service.start()
@@ -1865,12 +1876,9 @@ def test_in_forward_tls_idle_connection_timeout_churn(monkeypatch):
                     pass
 
         # The listener must still accept a fresh connection and ingest data
-        context = ssl.create_default_context(cafile=cafile)
-        context.check_hostname = False
-        context.verify_mode = ssl.CERT_NONE
-        raw = socket.create_connection(("127.0.0.1", port), timeout=5)
-        with context.wrap_socket(raw, server_hostname="localhost") as tls:
-            tls.sendall(_message_mode_payload(TEST_TAG, {"message": "after-churn"}))
+        _send_tls_payload(
+            port, _message_mode_payload(TEST_TAG, {"message": "after-churn"}), cafile
+        )
 
         records = service.wait_for_record_count(1, timeout=15)
     finally:
@@ -2142,7 +2150,8 @@ def test_out_forward_metrics_signal_e2e_with_forward_receiver():
     assert record_payload["metrics"][0]["values"][0]["value_int64"] == 42
 
 
-def test_in_forward_ingests_all_metrics_contexts_from_single_payload():
+@pytest.mark.parametrize("input_workers", [1, 2])
+def test_in_forward_ingests_all_metrics_contexts_from_single_payload(input_workers):
     expected_names = {"forward_metric_a", "forward_metric_b"}
     timeout = 30 if memory_check_enabled() else 10
     metrics_payload = b"".join(_forward_metric_contexts(expected_names))
@@ -2155,35 +2164,34 @@ def test_in_forward_ingests_all_metrics_contexts_from_single_payload():
     )
     payload = _pack_obj(["test", metrics_payload, {"fluent_signal": 1}])
 
-    for input_workers in (1, 2):
-        receiver = FluentBitTestService(
-            receiver_config,
-            extra_env={"FORWARD_INPUT_WORKERS": input_workers},
-        )
-        receiver.start()
-        try:
-            with socket.create_connection(
-                ("127.0.0.1", receiver.flb_listener_port), timeout=5
-            ) as sock:
-                sock.sendall(payload)
-
-            output = receiver.wait_for_condition(
-                lambda: read_file(receiver.flb.log_file)
-                if all(
-                    metric_name in read_file(receiver.flb.log_file)
-                    for metric_name in expected_names
-                )
-                else None,
-                timeout=timeout,
-                interval=0.2,
-                description=(
-                    f"both forwarded cmetrics contexts with {input_workers} workers"
-                ),
+    receiver = FluentBitTestService(
+        receiver_config,
+        extra_env={"FORWARD_INPUT_WORKERS": input_workers},
+    )
+    receiver.start()
+    try:
+        with socket.create_connection(
+            ("127.0.0.1", receiver.flb_listener_port), timeout=5
+        ) as sock:
+            sock.sendall(payload)
+
+        output = receiver.wait_for_condition(
+            lambda: read_file(receiver.flb.log_file)
+            if all(
+                metric_name in read_file(receiver.flb.log_file)
+                for metric_name in expected_names
             )
-        finally:
-            receiver.stop()
+            else None,
+            timeout=timeout,
+            interval=0.2,
+            description=(
+                f"both forwarded cmetrics contexts with {input_workers} workers"
+            ),
+        )
+    finally:
+        receiver.stop()
 
-        assert all(metric_name in output for metric_name in expected_names)
+    assert all(metric_name in output for metric_name in expected_names)
 
 
 def test_in_forward_metrics_payload_append_is_atomic():
diff --git a/tests/integration/scenarios/in_mqtt/tests/test_in_mqtt_001.py b/tests/integration/scenarios/in_mqtt/tests/test_in_mqtt_001.py
index 5482bca6d35..60097d4de2a 100644
--- a/tests/integration/scenarios/in_mqtt/tests/test_in_mqtt_001.py
+++ b/tests/integration/scenarios/in_mqtt/tests/test_in_mqtt_001.py
@@ -3,9 +3,17 @@
 import time
 
 import requests
+import pytest
 
 from server.http_server import data_storage, http_server_run
 from utils.test_service import FluentBitTestService
+from utils.fluent_bit_manager import fluent_bit_input_supports_config_property
+
+
+@pytest.fixture(autouse=True, scope="module")
+def require_mqtt_input():
+    if not fluent_bit_input_supports_config_property("mqtt", "payload_key"):
+        pytest.skip("This Fluent Bit build does not include the MQTT input")
 
 
 MQTT_CONNECT_PACKET = bytes(
diff --git a/tests/integration/scenarios/in_nginx_exporter_metrics/tests/test_plus_schema.py b/tests/integration/scenarios/in_nginx_exporter_metrics/tests/test_plus_schema.py
index fc8b84b94cd..edd01db4308 100644
--- a/tests/integration/scenarios/in_nginx_exporter_metrics/tests/test_plus_schema.py
+++ b/tests/integration/scenarios/in_nginx_exporter_metrics/tests/test_plus_schema.py
@@ -129,6 +129,8 @@ def plus_endpoint(tmp_path, request):
     lock = threading.Lock()
 
     class Handler(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
         def do_GET(self):
             with lock:
                 if self.path == "/status/":
diff --git a/tests/integration/scenarios/in_opentelemetry/config/otlp_grpc_grouped.yaml b/tests/integration/scenarios/in_opentelemetry/config/otlp_grpc_grouped.yaml
index 3baf27c4c26..703d446ef17 100644
--- a/tests/integration/scenarios/in_opentelemetry/config/otlp_grpc_grouped.yaml
+++ b/tests/integration/scenarios/in_opentelemetry/config/otlp_grpc_grouped.yaml
@@ -9,7 +9,7 @@ pipeline:
     - name: opentelemetry
       port: ${FLUENT_BIT_TEST_LISTENER_PORT}
       http_server.buffer_max_size: 4M
-      http_server.workers: 4
+      http_server.workers: ${OTLP_TEST_HTTP_WORKERS}
 
   outputs:
     - name: opentelemetry
diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
index bd368d88654..64798439d8d 100644
--- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
+++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
@@ -50,6 +50,10 @@
 from server.otlp_server import configure_otlp_response, otlp_server_run, data_storage
 
 logger = logging.getLogger(__name__)
+requires_shared_listener = pytest.mark.skipif(
+    os.name == "nt", reason="HTTP shared listener ports are unavailable on Windows"
+)
+HTTP_WORKER_COUNTS = [1, pytest.param(4, marks=requires_shared_listener)]
 MOCK_VALID_JWT = (
     "eyJhbGciOiJSUzI1NiIsImtpZCI6InRlc3QiLCJ0eXAiOiJKV1QifQ."
     "eyJleHAiOjE4OTM0NTYwMDAsImlzcyI6Imlzc3VlciIsImF1ZCI6ImF1ZGllbmNlIiwiYXpwIjoiY2xpZW50MSJ9."
@@ -509,9 +513,7 @@ def generated_limit_service(tmp_path, workers, limit="16k"):
     return Service(str(config))
 
 
-@pytest.mark.parametrize("workers", [1, pytest.param(
-    2, marks=pytest.mark.skipif(os.name == "nt",
-                               reason="HTTP shared listener ports are unavailable on Windows"))])
+@pytest.mark.parametrize("workers", [1, pytest.param(2, marks=requires_shared_listener)])
 @pytest.mark.parametrize("encoding", ["json", "protobuf"])
 @pytest.mark.parametrize("attributes,scopes,records,resources,accepted", [
     (1, 32, 1, 1, True),
@@ -925,6 +927,7 @@ def test_in_opentelemetry_large_protobuf_logs():
     assert received_record_count == expected_record_count
 
 
+@requires_shared_listener
 def test_in_opentelemetry_batched_protobuf_logs_with_workers():
     service = Service(IN_OPENTELEMETRY_WORKER_PROTOCOL_CONFIGS["http1_cleartext"])
     source = ExportLogsServiceRequest()
@@ -1901,6 +1904,7 @@ def test_in_opentelemetry_http2_invalid_endpoint_returns_404():
     {"id": "http1_tls", "config_key": "http1_tls", "http_mode": "http1.1", "use_tls": True},
     {"id": "http2_tls_alpn", "config_key": "http2_tls", "http_mode": "http2", "use_tls": True},
 ], ids=lambda case: case["id"])
+@requires_shared_listener
 def test_in_opentelemetry_http_workers_mixed_signal_matrix(case):
     request_plan = [
         ("logs", "test_logs_001.in.json", "/v1/logs"),
@@ -1987,6 +1991,8 @@ def send_job(job):
     ],
 )
 def test_in_opentelemetry_pause_resume(config_file, signal_type, json_input, endpoint):
+    if os.name == "nt" and config_file == "otlp_pause_resume_workers.yaml":
+        pytest.skip("HTTP shared listener ports are unavailable on Windows")
     service = Service(config_file)
 
     try:
@@ -2025,6 +2031,8 @@ def open_active_connections():
     ids=["single_listener", "workers_4"],
 )
 def test_in_opentelemetry_shutdown_while_paused(config_file):
+    if os.name == "nt" and config_file == "otlp_pause_resume_workers.yaml":
+        pytest.skip("HTTP shared listener ports are unavailable on Windows")
     service = Service(config_file)
 
     try:
@@ -2048,6 +2056,7 @@ def test_in_opentelemetry_shutdown_while_paused(config_file):
         service.stop()
 
 
+@requires_shared_listener
 def test_in_opentelemetry_http_workers_export_ingress_queue_metrics():
     service = Service(IN_OPENTELEMETRY_WORKER_PROTOCOL_CONFIGS["http1_cleartext"])
     service.start()
@@ -2095,6 +2104,7 @@ def test_in_opentelemetry_http_workers_export_ingress_queue_metrics():
     ) == 0
 
 
+@requires_shared_listener
 def test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits():
     service = Service(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_CONFIG)
     service.start()
@@ -2145,6 +2155,7 @@ def test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits():
     ) == 0
 
 
+@requires_shared_listener
 def test_in_opentelemetry_http_workers_respect_ingress_queue_byte_limit():
     service = Service(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_BYTE_LIMIT_CONFIG)
     small_payload = service.build_otel_payload("test_logs_001.in.json", "logs")
@@ -2211,6 +2222,7 @@ def test_in_opentelemetry_http_workers_respect_ingress_queue_byte_limit():
     ) == 0
 
 
+@requires_shared_listener
 def test_in_opentelemetry_http_workers_account_metrics_payload_bytes():
     service = Service(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_BYTE_LIMIT_CONFIG)
     small_payload = service.build_otel_payload("test_metrics_001.in.json", "metrics")
@@ -2513,7 +2525,7 @@ def received_grpc_log_records():
     "gzip", "zstd", "zstd-unknown", "zstd-unknown-large-window",
     "snappy", "snappy-framed", "snappy-mixed",
 ])
-@pytest.mark.parametrize("workers", [1, 4])
+@pytest.mark.parametrize("workers", HTTP_WORKER_COUNTS)
 def test_in_opentelemetry_grpc_bounded_codecs(encoding, workers):
     service = Service("otlp_grpc_bounded.yaml")
     service.service.extra_env["OTLP_TEST_HTTP_WORKERS"] = workers
@@ -2525,6 +2537,12 @@ def test_in_opentelemetry_grpc_bounded_codecs(encoding, workers):
                                 ([4096, 1024], 8)]:
             frames = [(grpc_limit_payload(service, size), True) for size in sizes]
             assert send_grpc_limit_frames(service, frames, encoding=encoding) == expected
+        # Wait for delivery before stopping: Windows terminates the process
+        # without the final flush performed by a POSIX SIGTERM shutdown.
+        service.service.wait_for_condition(
+            lambda: len(received_grpc_log_records()) >= 16, timeout=30,
+            description="all records from accepted gRPC requests",
+        )
     finally:
         service.stop()
     # Every fixture has four records. Only the three accepted requests may ingest.
@@ -2536,9 +2554,10 @@ def test_in_opentelemetry_grpc_bounded_codecs(encoding, workers):
     "snappy", "snappy-framed", "snappy-mixed",
 ])
 @pytest.mark.parametrize("first_compressed", [False, True])
-def test_in_opentelemetry_grpc_rejected_request_is_not_ingested(encoding, first_compressed):
+@pytest.mark.parametrize("workers", HTTP_WORKER_COUNTS)
+def test_in_opentelemetry_grpc_rejected_request_is_not_ingested(encoding, first_compressed, workers):
     service = Service("otlp_grpc_bounded.yaml")
-    service.service.extra_env["OTLP_TEST_HTTP_WORKERS"] = 4
+    service.service.extra_env["OTLP_TEST_HTTP_WORKERS"] = workers
     try:
         service.start()
         first = grpc_limit_payload(service, 1024)
@@ -2556,9 +2575,24 @@ def test_in_opentelemetry_grpc_rejected_request_is_not_ingested(encoding, first_
         assert send_grpc_limit_frames(
             service, [(first, True)], encoding=encoding, suffix=b"\x01\x00\x00\x00\x01\xff"
         ) == 13
+        # An accepted marker ensures output has flushed before shutdown. This
+        # keeps an abrupt Windows stop from hiding partially ingested requests.
+        marker = ExportLogsServiceRequest()
+        scope = marker.resource_logs.add().scope_logs.add()
+        scope.log_records.add().body.string_value = "gRPC rejection verification marker"
+        assert send_grpc_limit_frames(
+            service, [(marker.SerializeToString(), False)], encoding=encoding
+        ) == 0
+        service.service.wait_for_condition(
+            lambda: any(record.body.string_value == "gRPC rejection verification marker"
+                        for record in received_grpc_log_records()),
+            timeout=30, description="accepted marker after rejected gRPC requests",
+        )
     finally:
         service.stop()
-    assert received_grpc_log_records() == []
+    records = received_grpc_log_records()
+    assert len(records) == 1
+    assert records[0].body.string_value == "gRPC rejection verification marker"
 
 
 @pytest.mark.parametrize("encoding,payload", [
@@ -2593,8 +2627,10 @@ def test_in_opentelemetry_grpc_rejects_advertised_size_before_decode(encoding, p
     "gzip", "zstd", "zstd-unknown", "zstd-unknown-large-window",
     "snappy", "snappy-framed", "snappy-mixed",
 ])
-def test_in_opentelemetry_grpc_preserves_groups_across_chunks(encoding):
+@pytest.mark.parametrize("workers", HTTP_WORKER_COUNTS)
+def test_in_opentelemetry_grpc_preserves_groups_across_chunks(encoding, workers):
     service = Service("otlp_grpc_grouped.yaml")
+    service.service.extra_env["OTLP_TEST_HTTP_WORKERS"] = workers
     frames = []
     expected = []
     for frame_index in range(2):
diff --git a/tests/integration/scenarios/in_syslog/tests/test_in_syslog_001.py b/tests/integration/scenarios/in_syslog/tests/test_in_syslog_001.py
index 615b4845399..a91ee3fa1ff 100644
--- a/tests/integration/scenarios/in_syslog/tests/test_in_syslog_001.py
+++ b/tests/integration/scenarios/in_syslog/tests/test_in_syslog_001.py
@@ -105,7 +105,7 @@ def test_in_syslog_tcp_tls():
         with socket.create_connection(("127.0.0.1", service.flb_listener_port), timeout=5) as raw_sock:
             with context.wrap_socket(raw_sock, server_hostname="localhost") as tls_sock:
                 tls_sock.sendall(TCP_RFC5424_MESSAGE)
-                tls_sock.shutdown(socket.SHUT_WR)
+                tls_sock.unwrap().close()
 
         payloads = service.read_forwarded_payloads()
     finally:
@@ -129,6 +129,7 @@ def test_in_syslog_udp_plaintext():
     _assert_message(payloads)
 
 
+@pytest.mark.skipif(not hasattr(socket, "AF_UNIX"), reason="Python requires Unix-domain socket support")
 def test_in_syslog_uds_stream_plaintext():
     service = Service("in_syslog_uds_stream_plaintext.yaml")
     service.start()
@@ -154,6 +155,7 @@ def test_in_syslog_uds_stream_plaintext():
 
 
 @pytest.mark.skipif(sys.platform == "darwin", reason="unix datagram syslog runtime-shell test is skipped on Darwin")
+@pytest.mark.skipif(not hasattr(socket, "AF_UNIX"), reason="Python requires Unix-domain socket support")
 def test_in_syslog_uds_dgram_plaintext():
     service = Service("in_syslog_uds_dgram_plaintext.yaml")
     service.start()
diff --git a/tests/integration/scenarios/in_syslog/tests/test_stream_framing.py b/tests/integration/scenarios/in_syslog/tests/test_stream_framing.py
index 97db9534c28..99320231416 100644
--- a/tests/integration/scenarios/in_syslog/tests/test_stream_framing.py
+++ b/tests/integration/scenarios/in_syslog/tests/test_stream_framing.py
@@ -22,6 +22,8 @@ def wait_for(predicate):
 
 @contextlib.contextmanager
 def daemon(tmp_path, mode, framing):
+    if mode == "unix_tcp" and not hasattr(socket, "AF_UNIX"):
+        pytest.skip("Python requires Unix-domain socket support")
     with socket.socket() as listener:
         listener.bind(("127.0.0.1", 0))
         port = listener.getsockname()[1]
@@ -52,7 +54,8 @@ def ready():
             wait_for(ready)
             yield address, log
         finally:
-            if process.poll() is None:
+            stopped_by_test = process.poll() is None
+            if stopped_by_test:
                 process.send_signal(signal.SIGTERM)
             try:
                 process.wait(timeout=30)
@@ -60,7 +63,8 @@ def ready():
                 process.kill()
                 process.wait()
                 pytest.fail("Fluent Bit did not shut down cleanly")
-            assert process.returncode == 0, log.read_text() + (memlog.read_text() if memory else "")
+            expected_code = 1 if os.name == "nt" and stopped_by_test else 0
+            assert process.returncode == expected_code, log.read_text() + (memlog.read_text() if memory else "")
             if memory:
                 assert "ERROR SUMMARY: 0 errors" in memlog.read_text(), memlog.read_text()
 
diff --git a/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records.py b/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records.py
index 4339be6e4ae..f18a7fe7f5f 100644
--- a/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records.py
+++ b/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records.py
@@ -27,11 +27,12 @@ def stop(self):
 
     def metrics(self, expected=None):
         url = f"http://127.0.0.1:{self.flb.http_monitoring_port}/api/v2/metrics/prometheus"
+        markers = [expected] if isinstance(expected, str) else (expected or [])
         return self.service.wait_for_condition(
             lambda: response.text
             if (
                 (response := requests.get(url, timeout=2)).status_code == 200
-                and (expected is None or expected in response.text)
+                and all(marker in response.text for marker in markers)
             )
             else None,
             timeout=10,
@@ -206,7 +207,7 @@ def test_input_tag_records_counter_accumulates():
 def test_input_tag_records_tracks_multiple_tags():
     metrics = _run_with_metrics(
         "tag_records_multi_tag.yaml",
-        expected='tag="tag.records.beta"',
+        expected=['tag="tag.records.alpha"', 'tag="tag.records.beta"'],
     )
 
     alpha = _metric_value(
@@ -239,7 +240,7 @@ def test_input_tag_records_can_be_enabled_per_input_only():
     # in must export tag record metrics; the other input must export nothing.
     metrics = _run_with_metrics(
         "tag_records_input_enable.yaml",
-        expected='name="tag_records_enabled_input"',
+        expected='fluentbit_input_logs_tag_records_total{name="tag_records_enabled_input"',
     )
 
     enabled = _metric_value(
@@ -265,7 +266,7 @@ def test_input_tag_records_nested_block_disables_input():
     # Service is enabled; the nested block disables one input only.
     metrics = _run_with_metrics(
         "tag_records_input_nested_disable.yaml",
-        expected='name="tag_records_nested_control"',
+        expected='fluentbit_input_logs_tag_records_total{name="tag_records_nested_control"',
     )
 
     control = _metric_value(
@@ -291,7 +292,7 @@ def test_input_tag_records_nested_block_enables_input():
     # nested block inside an input aborted startup.
     metrics = _run_with_metrics(
         "tag_records_input_nested_enable.yaml",
-        expected='name="tag_records_nested_enabled"',
+        expected='fluentbit_input_logs_tag_records_total{name="tag_records_nested_enabled"',
     )
 
     enabled = _metric_value(
diff --git a/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records_badconfig.py b/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records_badconfig.py
index 63f9704ef12..8bcf3cde5fa 100644
--- a/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records_badconfig.py
+++ b/tests/integration/scenarios/internal_metrics/tests/test_internal_metrics_tag_records_badconfig.py
@@ -30,11 +30,9 @@
 VALGRIND_STRICT = bool(os.environ.get("VALGRIND_STRICT"))
 
 CRASH_SIGNALS = {
-    signal.SIGSEGV,
-    signal.SIGABRT,
-    signal.SIGILL,
-    signal.SIGFPE,
-    signal.SIGBUS,
+    getattr(signal, name)
+    for name in ("SIGSEGV", "SIGABRT", "SIGILL", "SIGFPE", "SIGBUS")
+    if hasattr(signal, name)
 }
 
 # Configs that must be rejected cleanly (non-zero exit, no crash).
diff --git a/tests/integration/scenarios/networking/tests/test_upstream_timeout.py b/tests/integration/scenarios/networking/tests/test_upstream_timeout.py
index fdfea6ddf74..eb8f64ae221 100644
--- a/tests/integration/scenarios/networking/tests/test_upstream_timeout.py
+++ b/tests/integration/scenarios/networking/tests/test_upstream_timeout.py
@@ -13,6 +13,7 @@
 class SilentPeer(socketserver.BaseRequestHandler):
     def handle(self):
         # Accept TCP but never answer the TLS handshake.
+        self.server.accepted.set()
         self.server.release.wait(30)
 
 
@@ -25,6 +26,7 @@ class SilentServer(socketserver.ThreadingTCPServer):
 def test_timeout_logs_configured_host_and_port(tmp_path, log_error):
     server = SilentServer(("127.0.0.1", 0), SilentPeer)
     server.release = threading.Event()
+    server.accepted = threading.Event()
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
     port = server.server_address[1]
@@ -40,12 +42,16 @@ def test_timeout_logs_configured_host_and_port(tmp_path, log_error):
             "inputs": [{"name": "dummy", "samples": 1, "tag": "timeout"}],
             "outputs": [{
                 "name": "http",
+                "log_level": "debug",
                 "match": "*",
                 "host": "localhost",
                 "port": port,
                 "tls": "on",
                 "tls.verify": "off",
                 "net.dns.resolver": "LEGACY",
+                # Keep localhost in the diagnostic, but connect to the IPv4
+                # listener without spending the timeout on IPv6 fallback.
+                "net.dns.prefer_ipv4": "on",
                 "net.connect_timeout": 2,
                 "net.connect_timeout_log_error": "on" if log_error else "off",
                 "retry_limit": "no_retries",
@@ -63,6 +69,7 @@ def completed():
             return logs if "is not retried (no retry config)" in logs else None
 
         logs = service.wait_for_condition(completed, timeout=30, description="timed-out flush")
+        assert server.accepted.is_set(), "The silent IPv4 peer was never reached"
         timeout_lines = [line for line in logs.splitlines()
                          if "[upstream]" in line and "timed out" in line]
         assert timeout_lines, logs
diff --git a/tests/integration/scenarios/out_kafka/tests/test_out_kafka_schema_registry.py b/tests/integration/scenarios/out_kafka/tests/test_out_kafka_schema_registry.py
index 2fa7ef6052b..e067af7f510 100644
--- a/tests/integration/scenarios/out_kafka/tests/test_out_kafka_schema_registry.py
+++ b/tests/integration/scenarios/out_kafka/tests/test_out_kafka_schema_registry.py
@@ -11,6 +11,13 @@
 from server.schema_registry_server import data_storage as registry_data
 from test_out_kafka_001 import Service, _wait_for_log_text
 from utils.fluent_bit_manager import FluentBitStartupError
+from utils.fluent_bit_manager import fluent_bit_output_supports_config_property
+
+
+@pytest.fixture(autouse=True, scope="module")
+def require_protobuf_encoder():
+    if not fluent_bit_output_supports_config_property("kafka", "protobuf_message"):
+        pytest.skip("Kafka Protobuf Schema Registry requires FLB_PROTOBUF_ENCODER=On")
 
 
 DETAILS = '''syntax = "proto3"; package registry;
diff --git a/tests/integration/scenarios/out_stackdriver/tests/test_assertion_logging.py b/tests/integration/scenarios/out_stackdriver/tests/test_assertion_logging.py
index 1cd93e7e7ab..ada45cdea79 100644
--- a/tests/integration/scenarios/out_stackdriver/tests/test_assertion_logging.py
+++ b/tests/integration/scenarios/out_stackdriver/tests/test_assertion_logging.py
@@ -10,12 +10,15 @@
 import yaml
 
 from utils.fluent_bit_manager import FluentBitStartupError
+from utils.fluent_bit_manager import fluent_bit_output_supports_config_property
 from utils.test_service import FluentBitTestService
 
 
 @pytest.mark.parametrize("plugin", ["stackdriver", "bigquery", "chronicle"])
 @pytest.mark.parametrize("log_level", ["debug", "trace"])
 def test_assertion_not_logged(tmp_path, monkeypatch, plugin, log_level):
+    if not fluent_bit_output_supports_config_property(plugin, "google_service_credentials"):
+        pytest.skip(f"This Fluent Bit build does not include the {plugin} output")
     # Reuse a public test key, never an operator's service-account credentials.
     key = Path(__file__).resolve().parents[2] / "in_splunk/certificate/private_key.pem"
     credentials = tmp_path / "credentials.json"
diff --git a/tests/integration/scenarios/processor_content_modifier/tests/test_grouped_logs.py b/tests/integration/scenarios/processor_content_modifier/tests/test_grouped_logs.py
index e99111f4861..82453f8c618 100644
--- a/tests/integration/scenarios/processor_content_modifier/tests/test_grouped_logs.py
+++ b/tests/integration/scenarios/processor_content_modifier/tests/test_grouped_logs.py
@@ -142,6 +142,9 @@ def metrics_ready():
                 f"http://127.0.0.1:{service.flb.http_monitoring_port}/api/v2/metrics/prometheus",
                 timeout=10,
             )
+            if response.status_code == 404:
+                # The monitoring snapshot is published asynchronously.
+                return None
             response.raise_for_status()
             values = processor_metrics(response.text, location, owner="opentelemetry.0")
             for stage, (before, after) in enumerate(counts):
diff --git a/tests/integration/scenarios/processor_fts/tests/test_fts_pipeline.py b/tests/integration/scenarios/processor_fts/tests/test_fts_pipeline.py
index a598ef69acc..983d2ebc6c0 100644
--- a/tests/integration/scenarios/processor_fts/tests/test_fts_pipeline.py
+++ b/tests/integration/scenarios/processor_fts/tests/test_fts_pipeline.py
@@ -153,6 +153,8 @@ def test_output_policy_isolated_across_routes_and_retries(tmp_path, filesystem,
     blocked = {"value": replay}
 
     class Collector(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
         def do_POST(self):
             message = ExportLogsServiceRequest.FromString(self.rfile.read(int(self.headers["Content-Length"])))
             attempts.append(message)
diff --git a/tests/integration/scenarios/processor_fts/tests/test_fts_signals_pipeline.py b/tests/integration/scenarios/processor_fts/tests/test_fts_signals_pipeline.py
index 064a21806f4..85a230070c6 100644
--- a/tests/integration/scenarios/processor_fts/tests/test_fts_signals_pipeline.py
+++ b/tests/integration/scenarios/processor_fts/tests/test_fts_signals_pipeline.py
@@ -23,6 +23,8 @@ def test_signal_route_isolation_retry_and_backlog(tmp_path, signal, filesystem,
     blocked = {"value": replay}
 
     class Collector(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
         def do_POST(self):
             data = self.rfile.read(int(self.headers["Content-Length"]))
             attempts.append(data)
diff --git a/tests/integration/scenarios/processor_sql/tests/test_sql_grouped_logs.py b/tests/integration/scenarios/processor_sql/tests/test_sql_grouped_logs.py
index bba8630b8f7..f6334b8ee29 100644
--- a/tests/integration/scenarios/processor_sql/tests/test_sql_grouped_logs.py
+++ b/tests/integration/scenarios/processor_sql/tests/test_sql_grouped_logs.py
@@ -141,6 +141,9 @@ def metrics_ready():
                 f"http://127.0.0.1:{service.flb.http_monitoring_port}/api/v2/metrics/prometheus",
                 timeout=10,
             )
+            if response.status_code == 404:
+                # The monitoring snapshot is published asynchronously.
+                return None
             response.raise_for_status()
             values = processor_metrics(response.text, location, owner="opentelemetry.0")
             for stage, (before, after) in enumerate(counts):
diff --git a/tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py b/tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py
index 6557555cc6e..90aa0ca9bf0 100644
--- a/tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py
+++ b/tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py
@@ -9,6 +9,7 @@
 import pytest
 
 from utils.fluent_bit_manager import FluentBitManager
+from utils.fluent_bit_manager import fluent_bit_input_supports_config_property
 
 
 def wait_for(predicate, timeout=30):
@@ -26,6 +27,8 @@ def daemon(tmp_path, mode):
         listener.bind(("127.0.0.1", 0))
         port = listener.getsockname()[1]
     plugin = mode.split("-")[0]
+    if plugin == "mqtt" and not fluent_bit_input_supports_config_property("mqtt", "payload_key"):
+        pytest.skip("This Fluent Bit build does not include the MQTT input")
     address = str(tmp_path / "input.sock") if plugin == "unix_socket" else ("127.0.0.1", port)
     parser = tmp_path / "parsers.conf"
     parser.write_text("[PARSER]\n    Name json\n    Format json\n")
diff --git a/tests/integration/scenarios/security_protobuf/tests/test_security_protobuf.py b/tests/integration/scenarios/security_protobuf/tests/test_security_protobuf.py
index 415e778f77a..aa065dcf600 100644
--- a/tests/integration/scenarios/security_protobuf/tests/test_security_protobuf.py
+++ b/tests/integration/scenarios/security_protobuf/tests/test_security_protobuf.py
@@ -51,7 +51,8 @@ def ready():
             wait_for(ready)
             yield port, process, log
         finally:
-            if process.poll() is None:
+            stopped_by_test = process.poll() is None
+            if stopped_by_test:
                 process.send_signal(signal.SIGTERM)
             try:
                 process.wait(timeout=30)
@@ -59,7 +60,8 @@ def ready():
                 process.kill()
                 process.wait()
                 pytest.fail("Fluent Bit did not shut down cleanly")
-            assert process.returncode == 0, log.read_text() + (memlog.read_text() if memory else "")
+            expected_code = 1 if os.name == "nt" and stopped_by_test else 0
+            assert process.returncode == expected_code, log.read_text() + (memlog.read_text() if memory else "")
             if memory:
                 assert "ERROR SUMMARY: 0 errors" in memlog.read_text(), memlog.read_text()
 
diff --git a/tests/integration/scenarios/storage/tests/test_storage_down_chunk.py b/tests/integration/scenarios/storage/tests/test_storage_down_chunk.py
index a6d2bb4841d..7991ed45a22 100644
--- a/tests/integration/scenarios/storage/tests/test_storage_down_chunk.py
+++ b/tests/integration/scenarios/storage/tests/test_storage_down_chunk.py
@@ -35,6 +35,8 @@ def test_down_chunk_placement(tmp_path, limit, routes):
     lock = threading.Lock()
 
     class Sink(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
         def do_POST(self):
             records = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
             route = int(self.path[1:])
diff --git a/tests/integration/scenarios/storage/tests/test_trace_replay.py b/tests/integration/scenarios/storage/tests/test_trace_replay.py
index 7f76abe18a5..6d3dd6e923a 100644
--- a/tests/integration/scenarios/storage/tests/test_trace_replay.py
+++ b/tests/integration/scenarios/storage/tests/test_trace_replay.py
@@ -44,6 +44,8 @@ def trace_payload(codes, batch=0):
 
 def start_sink(release, received, lock):
     class Sink(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
         def do_POST(self):
             body = self.rfile.read(int(self.headers["Content-Length"]))
             if release.is_set():
@@ -318,6 +320,8 @@ def capture(path, headers, body):
         return True
 
     class HttpSink(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
         def do_POST(self):
             body = self.rfile.read(int(self.headers["Content-Length"]))
             if not capture(self.path, self.headers, body):
diff --git a/tests/integration/scenarios/syslog_parser_time/tests/test_syslog_parser_time.py b/tests/integration/scenarios/syslog_parser_time/tests/test_syslog_parser_time.py
index 125c8055679..0a4fd0ee620 100644
--- a/tests/integration/scenarios/syslog_parser_time/tests/test_syslog_parser_time.py
+++ b/tests/integration/scenarios/syslog_parser_time/tests/test_syslog_parser_time.py
@@ -57,7 +57,8 @@ def ready():
             wait_for(ready)
             yield address, process, log
         finally:
-            if process.poll() is None:
+            stopped_by_test = process.poll() is None
+            if stopped_by_test:
                 process.send_signal(signal.SIGTERM)
             try:
                 process.wait(timeout=30)
@@ -65,7 +66,9 @@ def ready():
                 process.kill()
                 process.wait()
                 pytest.fail("Fluent Bit did not shut down cleanly")
-            assert process.returncode == 0, log.read_text() + (memlog.read_text() if memory else "")
+            # Python implements SIGTERM with TerminateProcess(..., 1) on Windows.
+            expected_code = 1 if os.name == "nt" and stopped_by_test else 0
+            assert process.returncode == expected_code, log.read_text() + (memlog.read_text() if memory else "")
             if memory:
                 assert "ERROR SUMMARY: 0 errors" in memlog.read_text(), memlog.read_text()
 
diff --git a/tests/integration/scenarios/unix_socket_decode_status/tests/test_unix_socket_decode_status.py b/tests/integration/scenarios/unix_socket_decode_status/tests/test_unix_socket_decode_status.py
index 320a8e1b920..25dfd65a7ae 100644
--- a/tests/integration/scenarios/unix_socket_decode_status/tests/test_unix_socket_decode_status.py
+++ b/tests/integration/scenarios/unix_socket_decode_status/tests/test_unix_socket_decode_status.py
@@ -12,6 +12,9 @@
 import pytest
 
 
+pytestmark = pytest.mark.skipif(
+    not hasattr(socket, "AF_UNIX"), reason="Python does not support Unix-domain sockets on this platform"
+)
 
 def wait_for(predicate, timeout=30):
     deadline = time.monotonic() + timeout
diff --git a/tests/integration/src/utils/fluent_bit_manager.py b/tests/integration/src/utils/fluent_bit_manager.py
index 55a90bf18ca..8bc15a03719 100644
--- a/tests/integration/src/utils/fluent_bit_manager.py
+++ b/tests/integration/src/utils/fluent_bit_manager.py
@@ -23,10 +23,12 @@
 import signal
 import shutil
 import subprocess
+import sys
 import tempfile
 import time
 
 import requests
+import pytest
 
 from utils.network import find_available_port, wait_for_port_to_be_free
 from utils.leaks import assert_leaks_clean
@@ -51,6 +53,45 @@ class FluentBitStartupError(RuntimeError):
     pass
 
 
+def _console_options():
+    if platform.system() != "Windows":
+        return {}
+
+    # Reload broadcasts CTRL_BREAK_EVENT to its console. Give each instance a
+    # hidden console so a reload cannot interrupt pytest or sibling instances.
+    startup_info = subprocess.STARTUPINFO()
+    startup_info.dwFlags |= subprocess.STARTF_USESHOWWINDOW
+    startup_info.wShowWindow = subprocess.SW_HIDE
+    return {
+        "creationflags": subprocess.CREATE_NEW_CONSOLE,
+        "startupinfo": startup_info,
+    }
+
+
+def _send_windows_shutdown(pid):
+    # A helper attaches to the instance's isolated console. Attaching pytest
+    # itself would disrupt its caller's console and other parallel workers.
+    script = """
+import ctypes
+import sys
+kernel = ctypes.WinDLL('kernel32', use_last_error=True)
+kernel.FreeConsole()
+if not kernel.AttachConsole(int(sys.argv[1])):
+    raise ctypes.WinError(ctypes.get_last_error())
+try:
+    if not kernel.SetConsoleCtrlHandler(None, True):
+        raise ctypes.WinError(ctypes.get_last_error())
+    if not kernel.GenerateConsoleCtrlEvent(0, 0):
+        raise ctypes.WinError(ctypes.get_last_error())
+finally:
+    kernel.FreeConsole()
+"""
+    subprocess.run(
+        [sys.executable, "-c", script, str(pid)], check=True, timeout=5,
+        creationflags=subprocess.CREATE_NO_WINDOW,
+    )
+
+
 def _default_binary_path():
     repo_root = Path(__file__).resolve().parents[4]
     local_binary = repo_root / "build" / "bin" / "fluent-bit"
@@ -138,12 +179,20 @@ def fluent_bit_binary_supports_config_property(property_name, binary_path=None):
 
 
 def fluent_bit_input_supports_config_property(plugin_name, property_name, binary_path=None):
+    return _plugin_supports_config_property("-i", plugin_name, property_name, binary_path)
+
+
+def fluent_bit_output_supports_config_property(plugin_name, property_name, binary_path=None):
+    return _plugin_supports_config_property("-o", plugin_name, property_name, binary_path)
+
+
+def _plugin_supports_config_property(plugin_type, plugin_name, property_name, binary_path=None):
     resolved_path = _resolve_binary_path(binary_path)
     property_marker = property_name.lower()
 
     try:
         result = subprocess.run(
-            [resolved_path, "-i", plugin_name, "-h"],
+            [resolved_path, plugin_type, plugin_name, "-h"],
             capture_output=True,
             text=True,
             check=False,
@@ -216,8 +265,12 @@ def start(self):
         command = [
             self.binary_absolute_path,
             "-c", self.config_path,
-            "-l", self.log_file
         ]
+        # Windows CRT append handles do not protect against another handle's
+        # writes. Capture stderr and stdout through the same inherited handle
+        # instead of letting -l open a second writer to the capture file.
+        if platform.system() != "Windows":
+            command += ["-l", self.log_file]
 
         if valgrind_enabled():
             command = [
@@ -238,6 +291,7 @@ def start(self):
             command,
             stdout=self.output_handle,
             stderr=subprocess.STDOUT,
+            **_console_options(),
         )
 
         if leaks_enabled():
@@ -260,7 +314,13 @@ def stop(self):
         return_code = self.process.poll()
         supervisor_running = self.process.poll() is None
         if supervisor_running or (leaks_enabled() and self.target_pid):
-            self.send_signal(signal.SIGTERM)
+            if platform.system() == "Windows":
+                try:
+                    _send_windows_shutdown(pid)
+                except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
+                    logger.warning("Windows graceful shutdown failed for pid %s: %s", pid, exc)
+            else:
+                self.send_signal(signal.SIGTERM)
 
         if supervisor_running:
             try:
@@ -280,7 +340,10 @@ def stop(self):
             self.output_handle.close()
             self.output_handle = None
 
-        if self.http_monitoring_port:
+        # Windows closes the process's socket handles before wait() returns.
+        # Exclusive probes can still fail during TIME_WAIT; the allocator avoids
+        # reusing these ports, so do not delay every Windows test for that state.
+        if self.http_monitoring_port and platform.system() != "Windows":
             timeout = 10 if memory_check_enabled() else 5
             wait_for_port_to_be_free(int(self.http_monitoring_port), timeout=timeout)
 
@@ -422,6 +485,13 @@ def wait_for_fluent_bit(self, timeout=None):
         start_time = time.time()
         while time.time() - start_time < timeout:
             if self.process and self.process.poll() is not None:
+                if platform.system() == "Windows":
+                    try:
+                        log_text = Path(self.log_file).read_text(encoding="utf-8", errors="replace")
+                    except OSError:
+                        log_text = ""
+                    if "shared listener ports are not supported on this platform" in log_text:
+                        pytest.skip("Shared listener ports are unavailable on Windows")
                 raise FluentBitStartupError(
                     f"Fluent Bit exited early with code {self.process.returncode}. "
                     f"See log file {self.log_file}"
diff --git a/tests/integration/src/utils/http_matrix.py b/tests/integration/src/utils/http_matrix.py
index 7fcdf77fda1..ba9c6d88f0a 100644
--- a/tests/integration/src/utils/http_matrix.py
+++ b/tests/integration/src/utils/http_matrix.py
@@ -1,7 +1,10 @@
 import os
+import shutil
 import subprocess
 import tempfile
 
+import pytest
+
 
 PROTOCOL_CASES = [
     {
@@ -56,15 +59,25 @@
 ]
 
 
-def curl_supports_http2():
+def curl_supports_http2(curl_path=None):
     result = subprocess.run(
-        ["curl", "--version"],
+        [curl_path or shutil.which("curl") or "curl", "--version"],
         capture_output=True,
         text=True,
         check=True,
     )
-    first_line = result.stdout.splitlines()[0] if result.stdout else ""
-    return "HTTP2" in result.stdout or "HTTP2" in first_line
+    return any(
+        "HTTP2" in line.split()[1:]
+        for line in result.stdout.splitlines()
+        if line.startswith("Features:")
+    )
+
+
+def curl_for_http_mode(http_mode):
+    curl_path = shutil.which("curl") or "curl"
+    if http_mode in ("http2", "http2-prior-knowledge") and not curl_supports_http2(curl_path):
+        pytest.skip(f"Selected curl lacks HTTP2 support: {curl_path} ({http_mode})")
+    return curl_path
 
 
 def run_curl_request(
@@ -80,7 +93,7 @@ def run_curl_request(
     extra_args=None,
 ):
     command = [
-        "curl",
+        curl_for_http_mode(http_mode),
         "--silent",
         "--show-error",
         "--output",
diff --git a/tests/integration/src/utils/input_pause_resume.py b/tests/integration/src/utils/input_pause_resume.py
index b40ae319e8e..21f6d2ba18b 100644
--- a/tests/integration/src/utils/input_pause_resume.py
+++ b/tests/integration/src/utils/input_pause_resume.py
@@ -6,7 +6,7 @@
 
 import requests
 
-from utils.http_matrix import run_curl_request
+from utils.http_matrix import curl_for_http_mode, run_curl_request
 
 
 INGESTION_PAUSED_RE = re.compile(
@@ -338,7 +338,7 @@ def run_curl_without_check(
     ca_cert_path=None,
 ):
     command = [
-        "curl",
+        curl_for_http_mode(http_mode),
         "--silent",
         "--show-error",
         "--output",
diff --git a/tests/integration/src/utils/network.py b/tests/integration/src/utils/network.py
index 0e7771a8be3..6d4ad1e9b32 100644
--- a/tests/integration/src/utils/network.py
+++ b/tests/integration/src/utils/network.py
@@ -15,20 +15,48 @@
 #  limitations under the License.
 
 import socket
+import os
+import threading
 import time
 
+
+_allocated_ports = set()
+_allocation_lock = threading.Lock()
+
+
+def _configure_probe(sock):
+    if os.name == "nt":
+        # SO_REUSEADDR can bind an occupied port on Windows.
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
+    else:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+
 def find_available_port(starting_port=0):
+    with _allocation_lock:
+        return _find_available_port(starting_port)
+
+
+def _find_available_port(starting_port):
     if starting_port in (None, 0):
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            sock.bind(("127.0.0.1", 0))
-            return sock.getsockname()[1]
+        for _ in range(1000):
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                _configure_probe(sock)
+                sock.bind(("0.0.0.0", 0))
+                port = sock.getsockname()[1]
+                if os.name != "nt" or port not in _allocated_ports:
+                    _allocated_ports.add(port)
+                    return port
+        raise RuntimeError("Could not allocate an unused test port")
 
     for port in range(starting_port, 65535):
+        if os.name == "nt" and port in _allocated_ports:
+            continue
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
             try:
-                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-                sock.bind(("127.0.0.1", port))
+                _configure_probe(sock)
+                sock.bind(("0.0.0.0", port))
+                _allocated_ports.add(port)
                 return port
             except OSError:
                 continue
@@ -41,7 +69,7 @@ def wait_for_port_to_be_free(port, *, host="127.0.0.1", timeout=5, interval=0.1)
     while time.time() < deadline:
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
             try:
-                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                _configure_probe(sock)
                 sock.bind((host, port))
                 return True
             except OSError:
diff --git a/tests/integration/tests/test_fluent_bit_manager.py b/tests/integration/tests/test_fluent_bit_manager.py
index 7026acac397..34447b029c2 100644
--- a/tests/integration/tests/test_fluent_bit_manager.py
+++ b/tests/integration/tests/test_fluent_bit_manager.py
@@ -150,9 +150,10 @@ def test_send_signal_raises_when_process_is_missing():
     manager = FluentBitManager("/tmp/fluent-bit.yaml")
 
     with pytest.raises(RuntimeError, match="not running"):
-        manager.send_signal(signal.SIGHUP)
+        manager.send_signal(signal.SIGTERM)
 
 
+@pytest.mark.skipif(not hasattr(signal, "SIGHUP"), reason="SIGHUP requires POSIX")
 def test_send_sighup_forwards_signal_to_process():
     manager = FluentBitManager("/tmp/fluent-bit.yaml")
     manager.process = Mock()
@@ -163,7 +164,11 @@ def test_send_sighup_forwards_signal_to_process():
 
 
 @pytest.mark.parametrize("configured,expected", [(None, 10), (30, 30), (240, 240)])
-def test_stop_uses_configured_shutdown_timeout(monkeypatch, configured, expected):
+@pytest.mark.parametrize("system", ["Linux", "Windows"])
+def test_stop_uses_configured_shutdown_timeout(monkeypatch, configured, expected, system):
+    monkeypatch.setattr(manager_module.platform, "system", lambda: system)
+    windows_shutdown = Mock()
+    monkeypatch.setattr(manager_module, "_send_windows_shutdown", windows_shutdown)
     monkeypatch.delenv("VALGRIND", raising=False)
     monkeypatch.delenv("LEAKS", raising=False)
     manager = FluentBitManager("/tmp/fluent-bit.yaml", shutdown_timeout=configured)
@@ -173,7 +178,12 @@ def test_stop_uses_configured_shutdown_timeout(monkeypatch, configured, expected
 
     manager.stop()
 
-    process.send_signal.assert_called_once_with(signal.SIGTERM)
+    if system == "Windows":
+        windows_shutdown.assert_called_once_with(process.pid)
+        process.send_signal.assert_not_called()
+    else:
+        process.send_signal.assert_called_once_with(signal.SIGTERM)
+        windows_shutdown.assert_not_called()
     process.wait.assert_called_once_with(timeout=expected)
 
 
@@ -195,12 +205,46 @@ def fake_post(url, timeout):
     assert manager.trigger_http_reload() == {"reload": "done"}
 
 
+@pytest.mark.parametrize("helper_error", [
+    manager_module.subprocess.CalledProcessError(1, ["shutdown-helper"]),
+    manager_module.subprocess.TimeoutExpired(["shutdown-helper"], 5),
+])
+@pytest.mark.parametrize("wait_times_out", [False, True])
+def test_windows_shutdown_helper_failure_still_cleans_up(monkeypatch, helper_error, wait_times_out):
+    monkeypatch.setattr(manager_module.platform, "system", lambda: "Windows")
+    monkeypatch.delenv("VALGRIND", raising=False)
+    monkeypatch.delenv("LEAKS", raising=False)
+    monkeypatch.setattr(manager_module, "_send_windows_shutdown", Mock(side_effect=helper_error))
+    manager = FluentBitManager("/tmp/fluent-bit.yaml", shutdown_timeout=1)
+    process = Mock()
+    process.poll.return_value = None
+    if wait_times_out:
+        process.wait.side_effect = [manager_module.subprocess.TimeoutExpired("fluent-bit", 1), 0]
+    else:
+        process.wait.return_value = 0
+    output = Mock()
+    manager.process = process
+    manager.target_pid = 123
+    manager.output_handle = output
+
+    manager.stop()
+
+    assert process.wait.call_args_list[0].kwargs == {"timeout": 1}
+    assert process.kill.call_count == int(wait_times_out)
+    output.close.assert_called_once()
+    assert manager.process is None
+    assert manager.target_pid is None
+    assert manager.output_handle is None
+
+
 def test_start_uses_unique_valgrind_log_path(monkeypatch, tmp_path):
+    monkeypatch.setattr(manager_module.platform, "system", lambda: "Linux")
     config_path = tmp_path / "fluent-bit.yaml"
     config_path.write_text("service:\n  flush: 1\n", encoding="utf-8")
     popen_calls = []
 
     monkeypatch.setenv("VALGRIND", "1")
+    monkeypatch.setattr(manager_module.shutil, "which", lambda binary: binary)
     monkeypatch.setattr(manager_module.os.path, "exists", lambda path: True)
     monkeypatch.setattr(manager_module.os.path, "isfile", lambda path: True)
     monkeypatch.setattr(manager_module.os, "access", lambda path, mode: True)
@@ -227,7 +271,7 @@ def fake_create_results_directory(self, base_dir='results'):
     monkeypatch.setattr(FluentBitManager, "create_results_directory", fake_create_results_directory)
     monkeypatch.setattr(FluentBitManager, "get_version_info", lambda self: ("vtest", "commit"))
 
-    def fake_popen(command, stdout=None, stderr=None):
+    def fake_popen(command, stdout=None, stderr=None, **kwargs):
         popen_calls.append(command)
         return popen_result
 
@@ -249,6 +293,45 @@ def fake_popen(command, stdout=None, stderr=None):
     ]]
 
 
+@pytest.mark.parametrize("system", ["Windows", "Linux", "Darwin"])
+def test_console_options_isolate_windows_reload(monkeypatch, system):
+    monkeypatch.setattr(manager_module.platform, "system", lambda: system)
+    startup_info = Mock(dwFlags=0)
+    monkeypatch.setattr(manager_module.subprocess, "STARTUPINFO", lambda: startup_info, raising=False)
+    monkeypatch.setattr(manager_module.subprocess, "STARTF_USESHOWWINDOW", 1, raising=False)
+    monkeypatch.setattr(manager_module.subprocess, "SW_HIDE", 0, raising=False)
+    monkeypatch.setattr(manager_module.subprocess, "CREATE_NEW_CONSOLE", 16, raising=False)
+
+    options = manager_module._console_options()
+
+    if system == "Windows":
+        assert options == {"creationflags": 16, "startupinfo": startup_info}
+        assert startup_info.dwFlags == 1
+        assert startup_info.wShowWindow == 0
+    else:
+        assert options == {}
+
+
+@pytest.mark.parametrize("system,log_message,skipped", [
+    ("Windows", "shared listener ports are not supported on this platform", True),
+    ("Windows", "could not bind address", False),
+    ("Linux", "shared listener ports are not supported on this platform", False),
+])
+def test_startup_skips_only_unsupported_windows_listeners(monkeypatch, tmp_path,
+                                                        system, log_message, skipped):
+    monkeypatch.setattr(manager_module.platform, "system", lambda: system)
+    manager = FluentBitManager(str(tmp_path / "config.yaml"))
+    manager.http_monitoring_port = "2020"
+    manager.log_file = str(tmp_path / "fluent-bit.log")
+    (tmp_path / "fluent-bit.log").write_text(log_message, encoding="utf-8")
+    manager.process = Mock(returncode=1)
+    manager.process.poll.return_value = 1
+
+    expected = pytest.skip.Exception if skipped else manager_module.FluentBitStartupError
+    with pytest.raises(expected):
+        manager.wait_for_fluent_bit()
+
+
 @pytest.mark.parametrize("error", [requests.ConnectionError, requests.ReadTimeout])
 def test_startup_health_check_retries_transient_errors(monkeypatch, error):
     manager = FluentBitManager("/tmp/fluent-bit.yaml")
diff --git a/tests/integration/tests/test_http_matrix.py b/tests/integration/tests/test_http_matrix.py
new file mode 100644
index 00000000000..47ed313f475
--- /dev/null
+++ b/tests/integration/tests/test_http_matrix.py
@@ -0,0 +1,40 @@
+import subprocess
+from unittest.mock import Mock
+
+import pytest
+
+from utils import http_matrix
+from utils.input_pause_resume import run_curl_without_check
+
+
+@pytest.mark.parametrize("mode", ["http2", "http2-prior-knowledge"])
+@pytest.mark.parametrize("unchecked", [False, True])
+def test_http2_requests_skip_without_curl_capability(monkeypatch, mode, unchecked):
+    monkeypatch.setattr(http_matrix.shutil, "which", lambda name: "selected-curl")
+    run = Mock(return_value=subprocess.CompletedProcess(
+        [], 0, stdout="curl with HTTP2 in a description\nFeatures: SSL IPv6\n"
+    ))
+    monkeypatch.setattr(http_matrix.subprocess, "run", run)
+    with pytest.raises(pytest.skip.Exception, match="Selected curl lacks HTTP2 support: selected-curl"):
+        if unchecked:
+            run_curl_without_check("http://localhost/", b"{}", [], http_mode=mode, max_time=1)
+        else:
+            http_matrix.run_curl_request("http://localhost/", http_mode=mode)
+    run.assert_called_once_with(
+        ["selected-curl", "--version"], capture_output=True, text=True, check=True
+    )
+
+
+@pytest.mark.parametrize("mode", ["http1.1", "http2", "http2-prior-knowledge"])
+def test_supported_requests_preserve_curl_failures(monkeypatch, mode):
+    monkeypatch.setattr(http_matrix.shutil, "which", lambda name: "selected-curl")
+    failure = subprocess.CalledProcessError(7, ["selected-curl"], stderr=b"connect failed")
+    results = [failure]
+    if mode != "http1.1":
+        results.insert(0, subprocess.CompletedProcess([], 0, stdout="Features: SSL HTTP2\n"))
+    run = Mock(side_effect=results)
+    monkeypatch.setattr(http_matrix.subprocess, "run", run)
+    with pytest.raises(subprocess.CalledProcessError) as exc:
+        http_matrix.run_curl_request("http://localhost/", http_mode=mode)
+    assert exc.value is failure
+    assert run.call_args.args[0][0] == "selected-curl"
diff --git a/tests/integration/tests/test_network.py b/tests/integration/tests/test_network.py
new file mode 100644
index 00000000000..26fabd67e46
--- /dev/null
+++ b/tests/integration/tests/test_network.py
@@ -0,0 +1,22 @@
+import os
+import socket
+
+import pytest
+
+from utils.network import find_available_port, wait_for_port_to_be_free
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows port allocation regression")
+def test_windows_port_allocations_are_distinct():
+    ports = [find_available_port() for _ in range(100)]
+    assert len(set(ports)) == len(ports)
+
+
+def test_probe_rejects_a_live_listener():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
+        listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        listener.bind(("127.0.0.1", 0))
+        listener.listen()
+        port = listener.getsockname()[1]
+        assert not wait_for_port_to_be_free(port, timeout=0.1, interval=0.01)
+        assert find_available_port(port) != port
diff --git a/tests/internal/env.c b/tests/internal/env.c
index e28c6e7ae60..e9e6f670e45 100644
--- a/tests/internal/env.c
+++ b/tests/internal/env.c
@@ -566,7 +566,64 @@ void test_expand_default_hyphen_from_os()
 }
 
 
+#ifdef FLB_SYSTEM_WINDOWS
+void test_file_env_var_drive_uri(void)
+{
+    struct flb_env *env;
+    const char *test_file = "flb_test_drive_uri.txt";
+    const char *value;
+    char path[4096];
+    char uri[4104];
+    size_t index;
+    int fd;
+    int ret;
+
+    if (!TEST_CHECK(_fullpath(path, test_file, sizeof(path)) != NULL)) {
+        return;
+    }
+    for (index = 0; path[index] != '\0'; index++) {
+        if (path[index] == '\\') {
+            path[index] = '/';
+        }
+    }
+    snprintf(uri, sizeof(uri), "file:///%s", path);
+    fd = open(test_file, O_CREAT | O_WRONLY | O_TRUNC, 0644);
+    if (!TEST_CHECK(fd >= 0)) {
+        return;
+    }
+    TEST_CHECK(write(fd, "first", 5) == 5);
+    close(fd);
+    env = flb_env_create();
+    if (!TEST_CHECK(env != NULL)) {
+        unlink(test_file);
+        return;
+    }
+    ret = flb_env_set_extended(env, "DRIVE_URI_SECRET", NULL, uri, 1);
+    if (TEST_CHECK(ret >= 0)) {
+        value = flb_env_get(env, "DRIVE_URI_SECRET");
+        if (TEST_CHECK(value != NULL)) {
+            TEST_CHECK(strcmp(value, "first") == 0);
+        }
+        fd = open(test_file, O_WRONLY | O_TRUNC, 0644);
+        if (TEST_CHECK(fd >= 0)) {
+            TEST_CHECK(write(fd, "second", 6) == 6);
+            close(fd);
+            sleep(2);
+            value = flb_env_get(env, "DRIVE_URI_SECRET");
+            if (TEST_CHECK(value != NULL)) {
+                TEST_CHECK(strcmp(value, "second") == 0);
+            }
+        }
+    }
+    flb_env_destroy(env);
+    unlink(test_file);
+}
+#endif
+
 TEST_LIST = {
+#ifdef FLB_SYSTEM_WINDOWS
+    { "file_env_var_drive_uri"       , test_file_env_var_drive_uri},
+#endif
     { "translate_long_env"           , test_translate_long_env},
     { "file_env_var_basic"           , test_file_env_var_basic},
     { "file_env_var_refresh"         , test_file_env_var_refresh},