diff --git a/cmake/plugins_options.cmake b/cmake/plugins_options.cmake
index e6e12e1aee9..a77281c3c3a 100644
--- a/cmake/plugins_options.cmake
+++ b/cmake/plugins_options.cmake
@@ -11,6 +11,7 @@ option(FLB_MINIMAL "Enable minimal build configuration" No)
 
 # Inputs (sources, data collectors)
 # =================================
+DEFINE_OPTION(FLB_IN_AEGISBPF                 "Enable AegisBPF input plugin"                 OFF)
 DEFINE_OPTION(FLB_IN_BLOB                     "Enable Blob input plugin"                     ON)
 DEFINE_OPTION(FLB_IN_CALYPTIA_FLEET           "Enable Calyptia Fleet input plugin"           ON)
 DEFINE_OPTION(FLB_IN_COLLECTD                 "Enable Collectd input plugin"                 ON)
diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt
index af43a8fc2e4..914094d485f 100644
--- a/plugins/CMakeLists.txt
+++ b/plugins/CMakeLists.txt
@@ -249,6 +249,7 @@ REGISTER_IN_PLUGIN("in_blob")
 
 # These plugins works only on Linux
 if(${CMAKE_SYSTEM_NAME} MATCHES "Linux")
+  REGISTER_IN_PLUGIN("in_aegisbpf")
   REGISTER_IN_PLUGIN("in_cpu")
   REGISTER_IN_PLUGIN("in_mem")
   REGISTER_IN_PLUGIN("in_thermal")
diff --git a/plugins/in_aegisbpf/CMakeLists.txt b/plugins/in_aegisbpf/CMakeLists.txt
new file mode 100644
index 00000000000..dc0ef6d679e
--- /dev/null
+++ b/plugins/in_aegisbpf/CMakeLists.txt
@@ -0,0 +1,4 @@
+set(src
+  in_aegisbpf.c)
+
+FLB_PLUGIN(in_aegisbpf "${src}" "")
diff --git a/plugins/in_aegisbpf/in_aegisbpf.c b/plugins/in_aegisbpf/in_aegisbpf.c
new file mode 100644
index 00000000000..5d6cf1c17a2
--- /dev/null
+++ b/plugins/in_aegisbpf/in_aegisbpf.c
@@ -0,0 +1,488 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+/* AegisBPF input plugin
+ * ---------------------
+ * Streams runtime-security events from a co-located AegisBPF agent into the
+ * Fluent Bit pipeline. AegisBPF (https://github.com/ErenAri/Aegis-BPF) is a
+ * BPF-LSM enforcement agent that exposes an opt-in, root-only Unix control
+ * socket; a "GET /events" request turns the connection into a newline-delimited
+ * stream of JSON (OCSF) security events. This plugin connects out to that
+ * socket, forwards each event as a record, and reconnects if the agent restarts.
+ *
+ * The agent drops slow readers (its broadcast uses non-blocking sends), so the
+ * plugin drains the socket in an event-driven collector rather than polling.
+ */
+
+#include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_config_map.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_log_event_encoder.h>
+
+#include <errno.h>
+#include <fcntl.h>
+#include <string.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include "in_aegisbpf.h"
+
+static void aegisbpf_disconnect(struct flb_in_aegisbpf *ctx)
+{
+    if (ctx->coll_fd_read >= 0) {
+        flb_input_collector_delete(ctx->coll_fd_read, ctx->ins);
+        ctx->coll_fd_read = -1;
+    }
+    if (ctx->fd >= 0) {
+        close(ctx->fd);
+        ctx->fd = -1;
+    }
+    ctx->connected = 0;
+    ctx->handshake_done = 0;
+    ctx->skipping_line = 0;
+    ctx->buf_len = 0;
+}
+
+static int write_all(int fd, const char *buf, size_t len)
+{
+    size_t off = 0;
+    ssize_t w;
+
+    while (off < len) {
+        w = write(fd, buf + off, len - off);
+        if (w < 0) {
+            if (errno == EINTR) {
+                continue;
+            }
+            return -1;
+        }
+        off += (size_t) w;
+    }
+    return 0;
+}
+
+/* Encode every complete newline-terminated JSON line held in ctx->buf into the
+ * event encoder, then compact the leftover partial line to the front. */
+static void aegisbpf_process_lines(struct flb_in_aegisbpf *ctx)
+{
+    size_t start = 0;
+    size_t i;
+    char *nl;
+    char *line;
+    size_t line_len;
+    char *mp;
+    size_t mp_size;
+    int root_type;
+    size_t consumed;
+    int ret;
+    int r;
+    struct flb_time tm;
+
+    /* If a previous read dropped an oversized line, discard bytes up to and
+     * including the next newline so the tail of that line is never parsed as a
+     * (truncated) event. */
+    if (ctx->skipping_line) {
+        nl = memchr(ctx->buf, '\n', ctx->buf_len);
+        if (nl == NULL) {
+            ctx->buf_len = 0;
+            return;
+        }
+        ctx->skipping_line = 0;
+        start = (size_t) (nl - ctx->buf) + 1;
+    }
+
+    for (i = start; i < ctx->buf_len; i++) {
+        if (ctx->buf[i] != '\n') {
+            continue;
+        }
+
+        line = ctx->buf + start;
+        line_len = i - start;
+
+        /* strip a trailing CR if present */
+        if (line_len > 0 && line[line_len - 1] == '\r') {
+            line_len--;
+        }
+
+        start = i + 1;
+
+        /* The agent's first line is the streaming ack, not an event. */
+        if (!ctx->handshake_done) {
+            ctx->handshake_done = 1;
+            continue;
+        }
+        if (line_len == 0) {
+            continue;
+        }
+
+        mp = NULL;
+        mp_size = 0;
+        root_type = 0;
+        consumed = 0;
+        ret = flb_pack_json(line, line_len, &mp, &mp_size, &root_type, &consumed);
+        /* Accept only a single, whole JSON object per line: reject parse errors,
+         * arrays/scalars, and any trailing bytes after the object (which
+         * flb_pack_json would otherwise pack as extra roots). */
+        if (ret != 0 || mp == NULL ||
+            root_type != FLB_PACK_JSON_OBJECT || consumed != line_len) {
+            flb_plg_debug(ctx->ins,
+                          "skipping line: not a single JSON object (%zu bytes)",
+                          line_len);
+            if (mp != NULL) {
+                flb_free(mp);
+            }
+            continue;
+        }
+
+        if (flb_log_event_encoder_begin_record(ctx->encoder) ==
+                FLB_EVENT_ENCODER_SUCCESS) {
+            flb_time_get(&tm);
+            flb_log_event_encoder_set_timestamp(ctx->encoder, &tm);
+            r = flb_log_event_encoder_set_body_from_raw_msgpack(ctx->encoder,
+                                                                mp, mp_size);
+            if (r == FLB_EVENT_ENCODER_SUCCESS) {
+                flb_log_event_encoder_commit_record(ctx->encoder);
+            }
+            else {
+                flb_log_event_encoder_rollback_record(ctx->encoder);
+            }
+        }
+        flb_free(mp);
+    }
+
+    if (start > 0) {
+        if (start < ctx->buf_len) {
+            memmove(ctx->buf, ctx->buf + start, ctx->buf_len - start);
+        }
+        ctx->buf_len -= start;
+    }
+}
+
+/* Socket collector: drain all currently-available bytes, then flush records. */
+static int in_aegisbpf_read(struct flb_input_instance *ins,
+                            struct flb_config *config, void *data)
+{
+    struct flb_in_aegisbpf *ctx = data;
+    int disconnected = 0;
+    size_t drained = 0;
+    ssize_t n;
+    size_t new_size;
+    char *tmp;
+
+    (void) config;
+
+    flb_log_event_encoder_reset(ctx->encoder);
+
+    while (1) {
+        if (ctx->buf_len == ctx->buf_size) {
+            if (ctx->buf_size >= FLB_IN_AEGISBPF_BUF_MAX) {
+                /* A single line exceeded the cap; drop the buffered head and mark
+                 * the line for skipping so its remaining tail (still in the
+                 * socket) is discarded up to the next newline rather than parsed
+                 * as a truncated event. */
+                flb_plg_warn(ins, "line exceeded %d bytes, dropping",
+                             FLB_IN_AEGISBPF_BUF_MAX);
+                ctx->buf_len = 0;
+                ctx->skipping_line = 1;
+            }
+            else {
+                new_size = ctx->buf_size * 2;
+                if (new_size > FLB_IN_AEGISBPF_BUF_MAX) {
+                    new_size = FLB_IN_AEGISBPF_BUF_MAX;
+                }
+                tmp = flb_realloc(ctx->buf, new_size);
+                if (tmp == NULL) {
+                    flb_errno();
+                    break;
+                }
+                ctx->buf = tmp;
+                ctx->buf_size = new_size;
+            }
+        }
+
+        n = recv(ctx->fd, ctx->buf + ctx->buf_len,
+                 ctx->buf_size - ctx->buf_len, 0);
+        if (n > 0) {
+            ctx->buf_len += (size_t) n;
+            aegisbpf_process_lines(ctx);
+            /* Bound work per wake so a continuously-writing agent can't hold the
+             * engine thread or grow the append arbitrarily large. The socket
+             * collector re-arms and continues on the next wake. */
+            drained += (size_t) n;
+            if (drained >= FLB_IN_AEGISBPF_DRAIN_MAX) {
+                break;
+            }
+            continue;
+        }
+        else if (n == 0) {
+            flb_plg_info(ins, "agent closed the connection");
+            disconnected = 1;
+            break;
+        }
+        else {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                break; /* drained */
+            }
+            if (errno == EINTR) {
+                continue;
+            }
+            flb_plg_warn(ins, "read error: %s", strerror(errno));
+            disconnected = 1;
+            break;
+        }
+    }
+
+    if (ctx->encoder->output_length > 0) {
+        flb_input_log_append(ins, NULL, 0,
+                             ctx->encoder->output_buffer,
+                             ctx->encoder->output_length);
+    }
+
+    if (disconnected) {
+        aegisbpf_disconnect(ctx);
+    }
+
+    return 0;
+}
+
+static int aegisbpf_connect(struct flb_in_aegisbpf *ctx,
+                            struct flb_config *config)
+{
+    struct sockaddr_un addr;
+    int fd;
+    int flags;
+    static const char req[] = "GET /events\n";
+
+    if (flb_sds_len(ctx->socket_path) >= sizeof(addr.sun_path)) {
+        flb_plg_error(ctx->ins, "socket_path too long: %s", ctx->socket_path);
+        return -1;
+    }
+
+    fd = socket(AF_UNIX, SOCK_STREAM, 0);
+    if (fd < 0) {
+        flb_errno();
+        return -1;
+    }
+
+    memset(&addr, 0, sizeof(addr));
+    addr.sun_family = AF_UNIX;
+    strncpy(addr.sun_path, ctx->socket_path, sizeof(addr.sun_path) - 1);
+
+    if (connect(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
+        flb_plg_debug(ctx->ins, "connect(%s) failed: %s",
+                      ctx->socket_path, strerror(errno));
+        close(fd);
+        return -1;
+    }
+
+    /* Request the event stream (blocking write; the request is tiny). */
+    if (write_all(fd, req, sizeof(req) - 1) < 0) {
+        flb_plg_warn(ctx->ins, "failed to send stream request: %s",
+                     strerror(errno));
+        close(fd);
+        return -1;
+    }
+
+    /* Non-blocking reads so the collector never stalls the engine. */
+    flags = fcntl(fd, F_GETFL, 0);
+    if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {
+        flb_errno();
+        close(fd);
+        return -1;
+    }
+
+    ctx->fd = fd;
+    ctx->connected = 1;
+    ctx->handshake_done = 0;
+    ctx->buf_len = 0;
+
+    ctx->coll_fd_read = flb_input_set_collector_socket(ctx->ins,
+                                                       in_aegisbpf_read,
+                                                       fd, config);
+    if (ctx->coll_fd_read < 0) {
+        flb_plg_error(ctx->ins, "could not register read collector");
+        aegisbpf_disconnect(ctx);
+        return -1;
+    }
+    if (flb_input_collector_start(ctx->coll_fd_read, ctx->ins) < 0) {
+        flb_plg_error(ctx->ins, "could not start read collector");
+        aegisbpf_disconnect(ctx);
+        return -1;
+    }
+
+    flb_plg_info(ctx->ins, "connected to AegisBPF at %s", ctx->socket_path);
+    return 0;
+}
+
+/* Time collector: (re)establish the connection while disconnected. */
+static int in_aegisbpf_reconnect(struct flb_input_instance *ins,
+                                 struct flb_config *config, void *data)
+{
+    struct flb_in_aegisbpf *ctx = data;
+
+    (void) ins;
+
+    if (ctx->connected) {
+        return 0;
+    }
+    aegisbpf_connect(ctx, config);
+    return 0;
+}
+
+static int in_aegisbpf_init(struct flb_input_instance *in,
+                            struct flb_config *config, void *data)
+{
+    struct flb_in_aegisbpf *ctx;
+    int ret;
+
+    (void) data;
+
+    ctx = flb_calloc(1, sizeof(struct flb_in_aegisbpf));
+    if (ctx == NULL) {
+        flb_errno();
+        return -1;
+    }
+    ctx->ins = in;
+    ctx->fd = -1;
+    ctx->coll_fd_read = -1;
+    ctx->coll_fd_reconnect = -1;
+
+    flb_input_set_context(in, ctx);
+
+    if (flb_input_config_map_set(in, (void *) ctx) < 0) {
+        flb_plg_error(in, "unable to load configuration");
+        flb_free(ctx);
+        return -1;
+    }
+
+    if (ctx->reconnect_sec <= 0) {
+        ctx->reconnect_sec = FLB_IN_AEGISBPF_DEFAULT_RECONN;
+    }
+
+    ctx->buf_size = FLB_IN_AEGISBPF_BUF_INIT;
+    ctx->buf = flb_malloc(ctx->buf_size);
+    if (ctx->buf == NULL) {
+        flb_errno();
+        flb_free(ctx);
+        return -1;
+    }
+
+    ctx->encoder = flb_log_event_encoder_create(FLB_LOG_EVENT_FORMAT_DEFAULT);
+    if (ctx->encoder == NULL) {
+        flb_plg_error(in, "could not initialize event encoder");
+        flb_free(ctx->buf);
+        flb_free(ctx);
+        return -1;
+    }
+
+    /* Drive (re)connection from a timer; the read collector is registered once
+     * a connection is established. */
+    ret = flb_input_set_collector_time(in, in_aegisbpf_reconnect,
+                                       ctx->reconnect_sec, 0, config);
+    if (ret < 0) {
+        flb_plg_error(in, "could not register reconnect collector");
+        flb_log_event_encoder_destroy(ctx->encoder);
+        flb_free(ctx->buf);
+        flb_free(ctx);
+        return -1;
+    }
+    ctx->coll_fd_reconnect = ret;
+
+    return 0;
+}
+
+static int in_aegisbpf_exit(void *data, struct flb_config *config)
+{
+    struct flb_in_aegisbpf *ctx = data;
+
+    (void) config;
+
+    if (ctx == NULL) {
+        return 0;
+    }
+    aegisbpf_disconnect(ctx);
+    if (ctx->encoder != NULL) {
+        flb_log_event_encoder_destroy(ctx->encoder);
+    }
+    if (ctx->buf != NULL) {
+        flb_free(ctx->buf);
+    }
+    flb_free(ctx);
+    return 0;
+}
+
+static void in_aegisbpf_pause(void *data, struct flb_config *config)
+{
+    struct flb_in_aegisbpf *ctx = data;
+
+    (void) config;
+
+    if (ctx->coll_fd_reconnect >= 0) {
+        flb_input_collector_pause(ctx->coll_fd_reconnect, ctx->ins);
+    }
+    if (ctx->coll_fd_read >= 0) {
+        flb_input_collector_pause(ctx->coll_fd_read, ctx->ins);
+    }
+}
+
+static void in_aegisbpf_resume(void *data, struct flb_config *config)
+{
+    struct flb_in_aegisbpf *ctx = data;
+
+    (void) config;
+
+    if (ctx->coll_fd_reconnect >= 0) {
+        flb_input_collector_resume(ctx->coll_fd_reconnect, ctx->ins);
+    }
+    if (ctx->coll_fd_read >= 0) {
+        flb_input_collector_resume(ctx->coll_fd_read, ctx->ins);
+    }
+}
+
+static struct flb_config_map config_map[] = {
+    {
+        FLB_CONFIG_MAP_STR, "socket_path", FLB_IN_AEGISBPF_DEFAULT_SOCKET,
+        0, FLB_TRUE, offsetof(struct flb_in_aegisbpf, socket_path),
+        "Path to the AegisBPF control socket (root-only Unix stream socket)."
+    },
+    {
+        FLB_CONFIG_MAP_INT, "reconnect_sec", "2",
+        0, FLB_TRUE, offsetof(struct flb_in_aegisbpf, reconnect_sec),
+        "Interval in seconds between reconnection attempts."
+    },
+    /* EOF */
+    {0}
+};
+
+struct flb_input_plugin in_aegisbpf_plugin = {
+    .name         = "aegisbpf",
+    .description  = "AegisBPF runtime-security events",
+    .cb_init      = in_aegisbpf_init,
+    .cb_pre_run   = NULL,
+    .cb_collect   = in_aegisbpf_reconnect,
+    .cb_flush_buf = NULL,
+    .config_map   = config_map,
+    .cb_pause     = in_aegisbpf_pause,
+    .cb_resume    = in_aegisbpf_resume,
+    .cb_exit      = in_aegisbpf_exit
+};
diff --git a/plugins/in_aegisbpf/in_aegisbpf.h b/plugins/in_aegisbpf/in_aegisbpf.h
new file mode 100644
index 00000000000..f1fbc6e87f0
--- /dev/null
+++ b/plugins/in_aegisbpf/in_aegisbpf.h
@@ -0,0 +1,55 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_IN_AEGISBPF_H
+#define FLB_IN_AEGISBPF_H
+
+#include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/flb_log_event_encoder.h>
+
+#define FLB_IN_AEGISBPF_DEFAULT_SOCKET  "/var/run/aegisbpf/aegisbpf.sock"
+#define FLB_IN_AEGISBPF_DEFAULT_RECONN  2      /* seconds */
+#define FLB_IN_AEGISBPF_BUF_INIT        16384  /* initial line-assembly buffer */
+#define FLB_IN_AEGISBPF_BUF_MAX         (1024 * 1024) /* cap: drop a pathological line */
+#define FLB_IN_AEGISBPF_DRAIN_MAX       (4 * 1024 * 1024) /* max bytes drained per collector wake */
+
+struct flb_in_aegisbpf {
+    /* config */
+    flb_sds_t socket_path;   /* AegisBPF control socket path */
+    int reconnect_sec;       /* reconnect interval */
+
+    /* connection state */
+    int fd;                  /* stream socket fd, -1 when disconnected */
+    int connected;
+    int handshake_done;      /* the agent's first line is a streaming ack; skip it */
+    int skipping_line;       /* discarding the tail of an over-length line */
+    int coll_fd_reconnect;   /* time collector: (re)connect */
+    int coll_fd_read;        /* socket collector: drain events */
+
+    /* line assembly */
+    char *buf;
+    size_t buf_size;
+    size_t buf_len;
+
+    struct flb_log_event_encoder *encoder;
+    struct flb_input_instance *ins;
+};
+
+#endif
diff --git a/tests/runtime/CMakeLists.txt b/tests/runtime/CMakeLists.txt
index 34652aaa03a..ccc6ba01504 100644
--- a/tests/runtime/CMakeLists.txt
+++ b/tests/runtime/CMakeLists.txt
@@ -48,6 +48,7 @@ if(FLB_OUT_LIB)
     FLB_RT_TEST(FLB_IN_NETIF          "in_netif.c")
     FLB_RT_TEST(FLB_IN_PODMAN_METRICS "in_podman_metrics.c")
     FLB_RT_TEST(FLB_IN_DOCKER        "in_docker.c")
+    FLB_RT_TEST(FLB_IN_AEGISBPF      "in_aegisbpf.c")
   endif()
   FLB_RT_TEST(FLB_IN_HEAD          "in_head.c")
   FLB_RT_TEST(FLB_IN_DUMMY         "in_dummy.c")
diff --git a/tests/runtime/in_aegisbpf.c b/tests/runtime/in_aegisbpf.c
new file mode 100644
index 00000000000..028fc1bcad3
--- /dev/null
+++ b/tests/runtime/in_aegisbpf.c
@@ -0,0 +1,357 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit.h>
+#include <fluent-bit/flb_time.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <poll.h>
+#include <pthread.h>
+#include "flb_tests_runtime.h"
+
+/*
+ * These tests exercise the in_aegisbpf plugin without a real AegisBPF agent:
+ * a dummy Unix-domain socket server plays the agent, serving a scripted event
+ * stream (a handshake line the plugin skips, valid single-object OCSF events,
+ * and malformed lines that must be rejected). We then assert, through the lib
+ * output callback, that exactly the valid events are forwarded as records.
+ */
+
+/* Scripted stream the dummy agent serves once the plugin connects. The first
+ * line is the streaming ack (skipped by the plugin); [1,2,3] and "scalar" are
+ * not single JSON objects and must be dropped; the three objects are records. */
+#define AEGIS_ACK        "{\"ok\":true}\n"
+#define AEGIS_EVENT_1    "{\"class_uid\":1001,\"activity_name\":\"Open\"}\n"
+#define AEGIS_BAD_ARRAY  "[1,2,3]\n"
+#define AEGIS_BAD_SCALAR "\"scalar\"\n"
+#define AEGIS_EVENT_2    "{\"class_uid\":4001,\"activity_name\":\"Connect\"}\n"
+#define AEGIS_EVENT_3    "{\"class_uid\":1001,\"disposition\":\"Blocked\"}\n"
+
+#define AEGIS_STREAM \
+    AEGIS_ACK AEGIS_EVENT_1 AEGIS_BAD_ARRAY AEGIS_BAD_SCALAR \
+    AEGIS_EVENT_2 AEGIS_EVENT_3
+
+#define AEGIS_EXPECTED_RECORDS 3
+
+struct test_ctx {
+    flb_ctx_t *flb;
+    int i_ffd;
+    int o_ffd;
+};
+
+/* dummy-agent server state */
+struct dummy_agent {
+    char path[108];
+    int listen_fd;
+    pthread_t thread;
+    pthread_mutex_t lock;
+    int stop;             /* protected by lock; signals the server to exit */
+};
+
+static int agent_should_stop(struct dummy_agent *agent)
+{
+    int s;
+    pthread_mutex_lock(&agent->lock);
+    s = agent->stop;
+    pthread_mutex_unlock(&agent->lock);
+    return s;
+}
+
+/* result accounting, shared with the output callback */
+pthread_mutex_t result_mutex = PTHREAD_MUTEX_INITIALIZER;
+int num_output = 0;
+
+static int get_output_num()
+{
+    int ret;
+    pthread_mutex_lock(&result_mutex);
+    ret = num_output;
+    pthread_mutex_unlock(&result_mutex);
+    return ret;
+}
+
+static void set_output_num(int num)
+{
+    pthread_mutex_lock(&result_mutex);
+    num_output = num;
+    pthread_mutex_unlock(&result_mutex);
+}
+
+/* Each forwarded record must be a JSON object carrying a class_uid; arrays and
+ * scalars from the stream must never reach here. */
+static int cb_check_record(void *record, size_t size, void *data)
+{
+    char *result = (char *) record;
+
+    set_output_num(get_output_num() + 1);
+
+    if (!TEST_CHECK(strstr(result, "\"class_uid\"") != NULL)) {
+        TEST_MSG("record missing class_uid: %s", result);
+    }
+    if (!TEST_CHECK(result[0] == '[')) {
+        /* lib/json output wraps each record as [timestamp, {...}] */
+        TEST_MSG("unexpected record framing: %s", result);
+    }
+
+    flb_free(record);
+    return 0;
+}
+
+/* Serve the scripted stream to the first client, then keep the connection open
+ * until the test tears the server down. accept() is driven through poll() with
+ * a stop flag so the thread always exits (and pthread_join never stalls) even
+ * if the plugin never connects. */
+static void *dummy_agent_run(void *arg)
+{
+    struct dummy_agent *agent = (struct dummy_agent *) arg;
+    struct pollfd pfd;
+    struct timespec ts;
+    char reqbuf[64];
+    const char *p = AEGIS_STREAM;
+    size_t remaining = sizeof(AEGIS_STREAM) - 1;
+    ssize_t w;
+    ssize_t r;
+    int fd = -1;
+    int ret;
+
+    ts.tv_sec = 0;
+    ts.tv_nsec = 20 * 1000 * 1000; /* 20ms */
+
+    /* Wait for the plugin to connect without blocking accept() indefinitely. */
+    pfd.fd = agent->listen_fd;
+    pfd.events = POLLIN;
+    while (!agent_should_stop(agent)) {
+        ret = poll(&pfd, 1, 100);
+        if (ret > 0 && (pfd.revents & POLLIN)) {
+            fd = accept(agent->listen_fd, NULL, NULL);
+            break;
+        }
+    }
+    if (fd < 0) {
+        return NULL;
+    }
+
+    /* The plugin turns the connection into an event stream with a
+     * "GET /events\n" request; validate it before serving the stream. Read it
+     * through poll() so a client that connects but never sends can't wedge the
+     * thread (and hence pthread_join) at teardown. */
+    pfd.fd = fd;
+    pfd.events = POLLIN;
+    r = 0;
+    while (!agent_should_stop(agent)) {
+        ret = poll(&pfd, 1, 100);
+        if (ret > 0 && (pfd.revents & POLLIN)) {
+            r = read(fd, reqbuf, sizeof(reqbuf) - 1);
+            break;
+        }
+    }
+    if (r > 0) {
+        reqbuf[r] = '\0';
+        if (!TEST_CHECK(strstr(reqbuf, "GET /events") != NULL)) {
+            TEST_MSG("unexpected request from plugin: %s", reqbuf);
+        }
+    }
+
+    while (remaining > 0) {
+        w = write(fd, p, remaining);
+        if (w <= 0) {
+            break;
+        }
+        p += w;
+        remaining -= (size_t) w;
+    }
+
+    /* Hold the connection open so the plugin drains what we sent rather than
+     * seeing an immediate EOF/reconnect; exit once teardown sets the stop flag. */
+    while (!agent_should_stop(agent)) {
+        nanosleep(&ts, NULL);
+    }
+
+    close(fd);
+    return NULL;
+}
+
+static int dummy_agent_start(struct dummy_agent *agent)
+{
+    struct sockaddr_un addr;
+    int fd;
+
+    agent->stop = 0;
+    pthread_mutex_init(&agent->lock, NULL);
+
+    /* Unique, short socket path under the runtime temp dir. */
+    snprintf(agent->path, sizeof(agent->path),
+             "/tmp/flb-aegisbpf-test-%d.sock", (int) getpid());
+    unlink(agent->path);
+
+    fd = socket(AF_UNIX, SOCK_STREAM, 0);
+    if (!TEST_CHECK(fd >= 0)) {
+        pthread_mutex_destroy(&agent->lock);
+        return -1;
+    }
+
+    memset(&addr, 0, sizeof(addr));
+    addr.sun_family = AF_UNIX;
+    strncpy(addr.sun_path, agent->path, sizeof(addr.sun_path) - 1);
+
+    if (!TEST_CHECK(bind(fd, (struct sockaddr *) &addr, sizeof(addr)) == 0)) {
+        close(fd);
+        pthread_mutex_destroy(&agent->lock);
+        return -1;
+    }
+    if (!TEST_CHECK(listen(fd, 1) == 0)) {
+        close(fd);
+        unlink(agent->path);
+        pthread_mutex_destroy(&agent->lock);
+        return -1;
+    }
+
+    agent->listen_fd = fd;
+    if (!TEST_CHECK(pthread_create(&agent->thread, NULL,
+                                   dummy_agent_run, agent) == 0)) {
+        close(fd);
+        agent->listen_fd = -1;
+        unlink(agent->path);
+        pthread_mutex_destroy(&agent->lock);
+        return -1;
+    }
+
+    return 0;
+}
+
+static void dummy_agent_stop(struct dummy_agent *agent)
+{
+    /* Signal the server thread, wait for it to exit, then close the listener.
+     * Closing only after the join keeps listen_fd valid for the poll() loop and
+     * avoids a data race on it. */
+    pthread_mutex_lock(&agent->lock);
+    agent->stop = 1;
+    pthread_mutex_unlock(&agent->lock);
+
+    pthread_join(agent->thread, NULL);
+
+    if (agent->listen_fd >= 0) {
+        close(agent->listen_fd);
+        agent->listen_fd = -1;
+    }
+    unlink(agent->path);
+    pthread_mutex_destroy(&agent->lock);
+}
+
+static struct test_ctx *test_ctx_create(struct dummy_agent *agent,
+                                         struct flb_lib_out_cb *data)
+{
+    struct test_ctx *ctx;
+    int ret;
+
+    ctx = flb_malloc(sizeof(struct test_ctx));
+    if (!TEST_CHECK(ctx != NULL)) {
+        return NULL;
+    }
+
+    ctx->flb = flb_create();
+    TEST_CHECK(ctx->flb != NULL);
+
+    flb_service_set(ctx->flb,
+                    "Flush", "0.200000000",
+                    "Grace", "1",
+                    "Log_Level", "error",
+                    NULL);
+
+    ctx->i_ffd = flb_input(ctx->flb, (char *) "aegisbpf", NULL);
+    TEST_CHECK(ctx->i_ffd >= 0);
+    ret = flb_input_set(ctx->flb, ctx->i_ffd,
+                        "socket_path", agent->path,
+                        "reconnect_sec", "1",
+                        NULL);
+    TEST_CHECK(ret == 0);
+
+    ctx->o_ffd = flb_output(ctx->flb, (char *) "lib", (void *) data);
+    TEST_CHECK(ctx->o_ffd >= 0);
+    ret = flb_output_set(ctx->flb, ctx->o_ffd,
+                         "match", "*",
+                         "format", "json",
+                         NULL);
+    TEST_CHECK(ret == 0);
+
+    return ctx;
+}
+
+static void test_ctx_destroy(struct test_ctx *ctx)
+{
+    TEST_CHECK(ctx != NULL);
+    flb_stop(ctx->flb);
+    flb_destroy(ctx->flb);
+    flb_free(ctx);
+}
+
+/* End-to-end: the plugin connects to the dummy agent, skips the handshake,
+ * forwards the three valid OCSF objects, and drops the array and scalar. */
+static void flb_test_aegisbpf_stream()
+{
+    struct dummy_agent agent;
+    struct test_ctx *ctx;
+    struct flb_lib_out_cb cb;
+    int ret;
+    int i;
+
+    set_output_num(0);
+    memset(&agent, 0, sizeof(agent));
+    agent.listen_fd = -1;
+
+    if (!TEST_CHECK(dummy_agent_start(&agent) == 0)) {
+        return;
+    }
+
+    cb.cb = cb_check_record;
+    cb.data = NULL;
+
+    ctx = test_ctx_create(&agent, &cb);
+    if (!TEST_CHECK(ctx != NULL)) {
+        dummy_agent_stop(&agent);
+        return;
+    }
+
+    ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+
+    /* Wait (bounded) for the three records to be forwarded. */
+    for (i = 0; i < 50; i++) {
+        if (get_output_num() >= AEGIS_EXPECTED_RECORDS) {
+            break;
+        }
+        flb_time_msleep(100);
+    }
+
+    if (!TEST_CHECK(get_output_num() == AEGIS_EXPECTED_RECORDS)) {
+        TEST_MSG("expected %d records, got %d",
+                 AEGIS_EXPECTED_RECORDS, get_output_num());
+    }
+
+    test_ctx_destroy(ctx);
+    dummy_agent_stop(&agent);
+}
+
+TEST_LIST = {
+    {"stream", flb_test_aegisbpf_stream},
+    {NULL, NULL}
+};