diff --git a/plugins/out_logdna/logdna.c b/plugins/out_logdna/logdna.c
index ffff06b829a..0754be08a07 100644
--- a/plugins/out_logdna/logdna.c
+++ b/plugins/out_logdna/logdna.c
@@ -26,6 +26,12 @@
 
 #include "logdna.h"
 
+#define LOGDNA_META_KEY "meta"
+#define LOGDNA_LEVEL_KEY "level"
+#define LOGDNA_SEVERITY_KEY "severity"
+#define LOGDNA_FILE_KEY "file"
+#define LOGDNA_APP_KEY "app"
+
 static inline int primary_key_check(msgpack_object k, char *name, int len)
 {
     if (k.type != MSGPACK_OBJECT_STR) {
@@ -44,19 +50,26 @@ static inline int primary_key_check(msgpack_object k, char *name, int len)
 }
 
 /*
- * This function looks for the following keys and add them to the buffer
+ * This function looks for the following primary keys and promotes them to
+ * the top-level line object:
  *
  * - level or severity
  * - file
  * - app
  * - meta
+ *
+ * When line_pck is not NULL, non-primary keys are packed into it for use
+ * as the "line" body (excluding the promoted keys).
  */
 static int record_append_primary_keys(struct flb_logdna *ctx,
                                       msgpack_object *map,
-                                      msgpack_packer *mp_sbuf)
+                                      msgpack_packer *mp_sbuf,
+                                      msgpack_packer *line_pck)
 {
     int i;
     int c = 0;
+    int is_primary;
+    int line_count = 0;
     msgpack_object *level = NULL;
     msgpack_object *file = NULL;
     msgpack_object *app = NULL;
@@ -64,63 +77,94 @@ static int record_append_primary_keys(struct flb_logdna *ctx,
     msgpack_object k;
     msgpack_object v;
 
-    for (i = 0; i < map->via.array.size; i++) {
+    if (line_pck) {
+        for (i = 0; i < map->via.map.size; i++) {
+            k = map->via.map.ptr[i].key;
+            if (primary_key_check(k, LOGDNA_META_KEY, sizeof(LOGDNA_META_KEY) - 1) == FLB_TRUE ||
+                primary_key_check(k, LOGDNA_LEVEL_KEY, sizeof(LOGDNA_LEVEL_KEY) - 1) == FLB_TRUE ||
+                primary_key_check(k, LOGDNA_SEVERITY_KEY, sizeof(LOGDNA_SEVERITY_KEY) - 1) == FLB_TRUE ||
+                primary_key_check(k, LOGDNA_FILE_KEY, sizeof(LOGDNA_FILE_KEY) - 1) == FLB_TRUE ||
+                primary_key_check(k, LOGDNA_APP_KEY, sizeof(LOGDNA_APP_KEY) - 1) == FLB_TRUE) {
+                continue;
+            }
+            line_count++;
+        }
+        msgpack_pack_map(line_pck, line_count);
+    }
+
+    for (i = 0; i < map->via.map.size; i++) {
         k = map->via.map.ptr[i].key;
         v = map->via.map.ptr[i].val;
-
-        /* Level - optional */
-        if (!level &&
-            (primary_key_check(k, "level", 5) == FLB_TRUE ||
-             primary_key_check(k, "severity", 8) == FLB_TRUE)) {
-            level = &k;
-            msgpack_pack_str(mp_sbuf, 5);
-            msgpack_pack_str_body(mp_sbuf, "level", 5);
-            msgpack_pack_object(mp_sbuf, v);
-            c++;
+        is_primary = FLB_FALSE;
+
+        /* Level - optional (both "level" and "severity" are primary) */
+        if (primary_key_check(k, LOGDNA_LEVEL_KEY, sizeof(LOGDNA_LEVEL_KEY) - 1) == FLB_TRUE ||
+            primary_key_check(k, LOGDNA_SEVERITY_KEY, sizeof(LOGDNA_SEVERITY_KEY) - 1) == FLB_TRUE) {
+            is_primary = FLB_TRUE;
+            if (!level) {
+                level = &k;
+                msgpack_pack_str(mp_sbuf, sizeof(LOGDNA_LEVEL_KEY) - 1);
+                msgpack_pack_str_body(mp_sbuf, LOGDNA_LEVEL_KEY, sizeof(LOGDNA_LEVEL_KEY) - 1);
+                msgpack_pack_object(mp_sbuf, v);
+                c++;
+            }
         }
 
         /* Meta - optional */
-        if (!meta && primary_key_check(k, "meta", 4) == FLB_TRUE) {
-            meta = &k;
-            msgpack_pack_str(mp_sbuf, 4);
-            msgpack_pack_str_body(mp_sbuf, "meta", 4);
-            msgpack_pack_object(mp_sbuf, v);
-            c++;
+        if (primary_key_check(k, LOGDNA_META_KEY, sizeof(LOGDNA_META_KEY) - 1) == FLB_TRUE) {
+            is_primary = FLB_TRUE;
+            if (!meta) {
+                meta = &k;
+                msgpack_pack_str(mp_sbuf, sizeof(LOGDNA_META_KEY) - 1);
+                msgpack_pack_str_body(mp_sbuf, LOGDNA_META_KEY, sizeof(LOGDNA_META_KEY) - 1);
+                msgpack_pack_object(mp_sbuf, v);
+                c++;
+            }
         }
 
         /* File */
-        if (!file && primary_key_check(k, "file", 4) == FLB_TRUE) {
-            file = &k;
-            msgpack_pack_str(mp_sbuf, 4);
-            msgpack_pack_str_body(mp_sbuf, "file", 4);
-            msgpack_pack_object(mp_sbuf, v);
-            c++;
+        if (primary_key_check(k, LOGDNA_FILE_KEY, sizeof(LOGDNA_FILE_KEY) - 1) == FLB_TRUE) {
+            is_primary = FLB_TRUE;
+            if (!file) {
+                file = &k;
+                msgpack_pack_str(mp_sbuf, sizeof(LOGDNA_FILE_KEY) - 1);
+                msgpack_pack_str_body(mp_sbuf, LOGDNA_FILE_KEY, sizeof(LOGDNA_FILE_KEY) - 1);
+                msgpack_pack_object(mp_sbuf, v);
+                c++;
+            }
         }
 
         /* App */
-        if (primary_key_check(k, "app", 3) == FLB_TRUE) {
-            app = &k;
-            msgpack_pack_str(mp_sbuf, 3);
-            msgpack_pack_str_body(mp_sbuf, "app", 3);
-            msgpack_pack_object(mp_sbuf, v);
-            c++;
+        if (primary_key_check(k, LOGDNA_APP_KEY, sizeof(LOGDNA_APP_KEY) - 1) == FLB_TRUE) {
+            is_primary = FLB_TRUE;
+            if (!app) {
+                app = &k;
+                msgpack_pack_str(mp_sbuf, sizeof(LOGDNA_APP_KEY) - 1);
+                msgpack_pack_str_body(mp_sbuf, LOGDNA_APP_KEY, sizeof(LOGDNA_APP_KEY) - 1);
+                msgpack_pack_object(mp_sbuf, v);
+                c++;
+            }
+        }
+
+        if (line_pck && is_primary == FLB_FALSE) {
+            msgpack_pack_object(line_pck, k);
+            msgpack_pack_object(line_pck, v);
         }
     }
 
     /* Set the global file name if the record did not provided one */
     if (!file && ctx->file) {
-        msgpack_pack_str(mp_sbuf, 4);
-        msgpack_pack_str_body(mp_sbuf, "file", 4);
+        msgpack_pack_str(mp_sbuf, sizeof(LOGDNA_FILE_KEY) - 1);
+        msgpack_pack_str_body(mp_sbuf, LOGDNA_FILE_KEY, sizeof(LOGDNA_FILE_KEY) - 1);
         msgpack_pack_str(mp_sbuf, flb_sds_len(ctx->file));
         msgpack_pack_str_body(mp_sbuf, ctx->file, flb_sds_len(ctx->file));
         c++;
     }
 
-
     /* If no application name is set, set the default */
     if (!app) {
-        msgpack_pack_str(mp_sbuf, 3);
-        msgpack_pack_str_body(mp_sbuf, "app", 3);
+        msgpack_pack_str(mp_sbuf, sizeof(LOGDNA_APP_KEY) - 1);
+        msgpack_pack_str_body(mp_sbuf, LOGDNA_APP_KEY, sizeof(LOGDNA_APP_KEY) - 1);
         msgpack_pack_str(mp_sbuf, flb_sds_len(ctx->app));
         msgpack_pack_str_body(mp_sbuf, ctx->app, flb_sds_len(ctx->app));
         c++;
@@ -139,10 +183,14 @@ static flb_sds_t logdna_compose_payload(struct flb_logdna *ctx,
     int total_lines;
     int array_size = 0;
     off_t map_off;
+    size_t off;
     char *line_json;
     flb_sds_t json;
     msgpack_packer mp_pck;
     msgpack_sbuffer mp_sbuf;
+    msgpack_packer mp_line_pck;
+    msgpack_sbuffer mp_line_sbuf;
+    msgpack_unpacked mp_line_result;
     struct flb_log_event_decoder log_decoder;
     struct flb_log_event log_event;
 
@@ -178,10 +226,24 @@ static flb_sds_t logdna_compose_payload(struct flb_logdna *ctx,
         msgpack_pack_map(&mp_pck, array_size);
 
         /*
-         * Append primary keys found, the return values is the number of appended
+         * Append primary keys found, the return value is the number of appended
          * keys to the record, we use that to adjust the map header size.
+         *
+         * When exclude_promoted_keys is enabled, non-primary keys are packed
+         * into mp_line_sbuf for use as the "line" body.
          */
-        ret = record_append_primary_keys(ctx, log_event.body, &mp_pck);
+        if (ctx->exclude_promoted_keys) {
+            msgpack_sbuffer_init(&mp_line_sbuf);
+            msgpack_packer_init(&mp_line_pck, &mp_line_sbuf,
+                                msgpack_sbuffer_write);
+
+            ret = record_append_primary_keys(ctx, log_event.body,
+                                             &mp_pck, &mp_line_pck);
+        }
+        else {
+            ret = record_append_primary_keys(ctx, log_event.body,
+                                             &mp_pck, NULL);
+        }
         array_size += ret;
 
         /* Timestamp */
@@ -193,7 +255,23 @@ static flb_sds_t logdna_compose_payload(struct flb_logdna *ctx,
         msgpack_pack_str(&mp_pck, 4);
         msgpack_pack_str_body(&mp_pck, "line", 4);
 
-        line_json = flb_msgpack_to_json_str(1024, log_event.body, config->json_escape_unicode);
+        if (ctx->exclude_promoted_keys) {
+            msgpack_unpacked_init(&mp_line_result);
+            off = 0;
+            msgpack_unpack_next(&mp_line_result,
+                                mp_line_sbuf.data, mp_line_sbuf.size, &off);
+
+            line_json = flb_msgpack_to_json_str(1024, &mp_line_result.data,
+                                                config->json_escape_unicode);
+
+            msgpack_unpacked_destroy(&mp_line_result);
+            msgpack_sbuffer_destroy(&mp_line_sbuf);
+        }
+        else {
+            line_json = flb_msgpack_to_json_str(1024, log_event.body,
+                                                config->json_escape_unicode);
+        }
+
         len = strlen(line_json);
         msgpack_pack_str(&mp_pck, len);
         msgpack_pack_str_body(&mp_pck, line_json, len);
@@ -584,11 +662,39 @@ static struct flb_config_map config_map[] = {
      "Name of the application generating the data (optional)"
     },
 
+    {
+     FLB_CONFIG_MAP_BOOL, "exclude_promoted_keys", "false",
+     0, FLB_TRUE, offsetof(struct flb_logdna, exclude_promoted_keys),
+     "Exclude promoted keys (meta, level, severity (promoted as level), app, file) from the line body"
+    },
+
     /* EOF */
     {0}
 
 };
 
+static int cb_logdna_format_test(struct flb_config *config,
+                                 struct flb_input_instance *ins,
+                                 void *plugin_context,
+                                 void *flush_ctx,
+                                 int event_type,
+                                 const char *tag, int tag_len,
+                                 const void *data, size_t bytes,
+                                 void **out_data, size_t *out_size)
+{
+    flb_sds_t json;
+    struct flb_logdna *ctx = plugin_context;
+
+    json = logdna_compose_payload(ctx, data, bytes, tag, tag_len, config);
+    if (!json) {
+        return -1;
+    }
+
+    *out_data = json;
+    *out_size = flb_sds_len(json);
+    return 0;
+}
+
 /* Plugin reference */
 struct flb_output_plugin out_logdna_plugin = {
     .name        = "logdna",
@@ -597,5 +703,6 @@ struct flb_output_plugin out_logdna_plugin = {
     .cb_flush    = cb_logdna_flush,
     .cb_exit     = cb_logdna_exit,
     .config_map  = config_map,
+    .test_formatter.callback = cb_logdna_format_test,
     .flags       = FLB_OUTPUT_NET | FLB_IO_TLS,
 };
diff --git a/plugins/out_logdna/logdna.h b/plugins/out_logdna/logdna.h
index 7ec8e843d14..98d927825db 100644
--- a/plugins/out_logdna/logdna.h
+++ b/plugins/out_logdna/logdna.h
@@ -41,6 +41,7 @@ struct flb_logdna {
     flb_sds_t file;
     flb_sds_t app;
     struct mk_list *tags;
+    int exclude_promoted_keys;
 
     /* Internal */
     flb_sds_t _hostname;
diff --git a/tests/runtime/CMakeLists.txt b/tests/runtime/CMakeLists.txt
index 7bc6a7d7153..1508c6c81b7 100644
--- a/tests/runtime/CMakeLists.txt
+++ b/tests/runtime/CMakeLists.txt
@@ -231,6 +231,7 @@ if(FLB_IN_LIB)
   FLB_RT_TEST(FLB_OUT_HTTP             "out_http.c")
   FLB_RT_TEST(FLB_OUT_KAFKA            "out_kafka.c")
   FLB_RT_TEST(FLB_OUT_LIB              "out_lib.c")
+  FLB_RT_TEST(FLB_OUT_LOGDNA           "out_logdna.c")
   FLB_RT_TEST(FLB_OUT_LOKI             "out_loki.c")
   FLB_RT_TEST(FLB_OUT_NULL             "out_null.c")
   FLB_RT_TEST(FLB_OUT_PLOT             "out_plot.c")
diff --git a/tests/runtime/out_logdna.c b/tests/runtime/out_logdna.c
new file mode 100644
index 00000000000..6951c8a43aa
--- /dev/null
+++ b/tests/runtime/out_logdna.c
@@ -0,0 +1,756 @@
+/* -*- 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_sds.h>
+#include <fluent-bit/flb_time.h>
+#include <pthread.h>
+#include "flb_tests_runtime.h"
+
+/* Thread-safe callback invocation tracking */
+static pthread_mutex_t result_mutex = PTHREAD_MUTEX_INITIALIZER;
+static int num_output = 0;
+
+static void set_output_num(int num)
+{
+    pthread_mutex_lock(&result_mutex);
+    num_output = num;
+    pthread_mutex_unlock(&result_mutex);
+}
+
+static int get_output_num()
+{
+    int ret;
+    pthread_mutex_lock(&result_mutex);
+    ret = num_output;
+    pthread_mutex_unlock(&result_mutex);
+    return ret;
+}
+
+static void clear_output_num()
+{
+    set_output_num(0);
+}
+
+/*
+ * Test: primary keys (meta, level, app) are promoted to top-level fields
+ * and excluded from the "line" JSON string (no duplication).
+ */
+#define JSON_WITH_PRIMARY_KEYS \
+    "[12345678, {\"message\":\"hello world\"," \
+    "\"meta\":{\"source\":\"test\",\"env\":\"dev\"}," \
+    "\"level\":\"info\"," \
+    "\"app\":\"myapp\"}]"
+
+static void cb_check_non_duplication(void *ctx, int ffd, int res_ret,
+                                     void *res_data, size_t res_size,
+                                     void *data)
+{
+    flb_sds_t json = res_data;
+
+    /* Primary keys promoted at top level (unescaped) */
+    if (!TEST_CHECK(strstr(json, "\"meta\":") != NULL)) {
+        TEST_MSG("missing top-level meta: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"level\":") != NULL)) {
+        TEST_MSG("missing top-level level: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"app\":") != NULL)) {
+        TEST_MSG("missing top-level app: %s", json);
+    }
+
+    /* Primary keys must NOT appear inside the line value (escaped) */
+    if (!TEST_CHECK(strstr(json, "\\\"meta\\\":") == NULL)) {
+        TEST_MSG("meta duplicated in line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"level\\\":") == NULL)) {
+        TEST_MSG("level duplicated in line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"app\\\":") == NULL)) {
+        TEST_MSG("app duplicated in line: %s", json);
+    }
+
+    /* Non-primary key must be in line value (escaped) */
+    if (!TEST_CHECK(strstr(json, "\\\"message\\\":") != NULL)) {
+        TEST_MSG("message missing from line: %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_non_duplication()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   "exclude_promoted_keys", "true",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_non_duplication, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_WITH_PRIMARY_KEYS,
+                 sizeof(JSON_WITH_PRIMARY_KEYS) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: all non-primary keys are preserved in the "line" body;
+ * all primary keys are promoted. No data is lost.
+ */
+#define JSON_ALL_KEYS \
+    "[12345678, {\"message\":\"hello\"," \
+    "\"meta\":{\"foo\":\"bar\"}," \
+    "\"level\":\"info\"," \
+    "\"app\":\"myapp\"," \
+    "\"file\":\"test.log\"," \
+    "\"host\":\"server1\"," \
+    "\"custom\":\"data\"}]"
+
+static void cb_check_data_completeness(void *ctx, int ffd, int res_ret,
+                                       void *res_data, size_t res_size,
+                                       void *data)
+{
+    flb_sds_t json = res_data;
+
+    /* All primary keys promoted at top level */
+    if (!TEST_CHECK(strstr(json, "\"meta\":") != NULL)) {
+        TEST_MSG("missing top-level meta: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"level\":") != NULL)) {
+        TEST_MSG("missing top-level level: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"app\":") != NULL)) {
+        TEST_MSG("missing top-level app: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"file\":") != NULL)) {
+        TEST_MSG("missing top-level file: %s", json);
+    }
+
+    /* All non-primary keys in line body (escaped) */
+    if (!TEST_CHECK(strstr(json, "\\\"message\\\":") != NULL)) {
+        TEST_MSG("message missing from line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"host\\\":") != NULL)) {
+        TEST_MSG("host missing from line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"custom\\\":") != NULL)) {
+        TEST_MSG("custom missing from line: %s", json);
+    }
+
+    /* Primary keys not duplicated in line body */
+    if (!TEST_CHECK(strstr(json, "\\\"meta\\\":") == NULL)) {
+        TEST_MSG("meta duplicated in line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"file\\\":") == NULL)) {
+        TEST_MSG("file duplicated in line: %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_data_completeness()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   "exclude_promoted_keys", "true",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_data_completeness, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_ALL_KEYS,
+                 sizeof(JSON_ALL_KEYS) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: "severity" key is promoted as "level" in the output.
+ */
+#define JSON_SEVERITY \
+    "[12345678, {\"message\":\"hello\",\"severity\":\"warning\"}]"
+
+static void cb_check_severity(void *ctx, int ffd, int res_ret,
+                               void *res_data, size_t res_size,
+                               void *data)
+{
+    flb_sds_t json = res_data;
+
+    /* severity should be promoted as "level" */
+    if (!TEST_CHECK(strstr(json, "\"level\":\"warning\"") != NULL)) {
+        TEST_MSG("severity not promoted as level: %s", json);
+    }
+
+    /* severity should not appear in line body */
+    if (!TEST_CHECK(strstr(json, "\\\"severity\\\":") == NULL)) {
+        TEST_MSG("severity duplicated in line: %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_severity_promoted_as_level()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   "exclude_promoted_keys", "true",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_severity, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_SEVERITY,
+                 sizeof(JSON_SEVERITY) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: record with both "level" and "severity" — only first is promoted,
+ * neither appears in line body, and no msgpack corruption occurs.
+ */
+#define JSON_LEVEL_AND_SEVERITY \
+    "[12345678, {\"message\":\"hello\"," \
+    "\"level\":\"info\"," \
+    "\"severity\":\"warning\"," \
+    "\"host\":\"server1\"}]"
+
+static void cb_check_level_and_severity(void *ctx, int ffd, int res_ret,
+                                        void *res_data, size_t res_size,
+                                        void *data)
+{
+    flb_sds_t json = res_data;
+
+    /* First level/severity key promoted as "level" */
+    if (!TEST_CHECK(strstr(json, "\"level\":\"info\"") != NULL)) {
+        TEST_MSG("level not promoted: %s", json);
+    }
+
+    /* Neither level nor severity in line body */
+    if (!TEST_CHECK(strstr(json, "\\\"level\\\":") == NULL)) {
+        TEST_MSG("level duplicated in line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"severity\\\":") == NULL)) {
+        TEST_MSG("severity duplicated in line: %s", json);
+    }
+
+    /* Non-primary keys preserved in line */
+    if (!TEST_CHECK(strstr(json, "\\\"message\\\":") != NULL)) {
+        TEST_MSG("message missing from line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"host\\\":") != NULL)) {
+        TEST_MSG("host missing from line: %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_level_and_severity()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   "exclude_promoted_keys", "true",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_level_and_severity, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_LEVEL_AND_SEVERITY,
+                 sizeof(JSON_LEVEL_AND_SEVERITY) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: default app name is set when record has no "app" key.
+ */
+#define JSON_NO_APP \
+    "[12345678, {\"message\":\"hello\"}]"
+
+static void cb_check_default_app(void *ctx, int ffd, int res_ret,
+                                  void *res_data, size_t res_size,
+                                  void *data)
+{
+    flb_sds_t json = res_data;
+
+    if (!TEST_CHECK(strstr(json, "\"app\":\"Fluent Bit\"") != NULL)) {
+        TEST_MSG("default app not set: %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_default_app()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_default_app, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_NO_APP,
+                 sizeof(JSON_NO_APP) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: record with no primary keys — all fields go into line,
+ * only default app appears at top level.
+ */
+#define JSON_NO_PRIMARY_KEYS \
+    "[12345678, {\"message\":\"hello\",\"host\":\"server1\"}]"
+
+static void cb_check_no_primary_keys(void *ctx, int ffd, int res_ret,
+                                      void *res_data, size_t res_size,
+                                      void *data)
+{
+    flb_sds_t json = res_data;
+
+    /* Default app at top level */
+    if (!TEST_CHECK(strstr(json, "\"app\":\"Fluent Bit\"") != NULL)) {
+        TEST_MSG("default app not set: %s", json);
+    }
+
+    /* All keys in line body */
+    if (!TEST_CHECK(strstr(json, "\\\"message\\\":") != NULL)) {
+        TEST_MSG("message missing from line: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"host\\\":") != NULL)) {
+        TEST_MSG("host missing from line: %s", json);
+    }
+
+    /* No meta or level at top level (they aren't in the record) */
+    if (!TEST_CHECK(strstr(json, "\"meta\":") == NULL)) {
+        TEST_MSG("unexpected meta at top level: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"level\":") == NULL)) {
+        TEST_MSG("unexpected level at top level: %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_no_primary_keys()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_no_primary_keys, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_NO_PRIMARY_KEYS,
+                 sizeof(JSON_NO_PRIMARY_KEYS) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: basic payload structure contains required fields.
+ */
+static void cb_check_payload_structure(void *ctx, int ffd, int res_ret,
+                                        void *res_data, size_t res_size,
+                                        void *data)
+{
+    flb_sds_t json = res_data;
+
+    if (!TEST_CHECK(strstr(json, "\"lines\":") != NULL)) {
+        TEST_MSG("missing lines array: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"line\":") != NULL)) {
+        TEST_MSG("missing line field: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"timestamp\":") != NULL)) {
+        TEST_MSG("missing timestamp field: %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_payload_structure()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_payload_structure, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_NO_APP,
+                 sizeof(JSON_NO_APP) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: backward compatibility — when exclude_promoted_keys is not set
+ * (default false), promoted keys ARE present in the line body.
+ */
+static void cb_check_backward_compat(void *ctx, int ffd, int res_ret,
+                                      void *res_data, size_t res_size,
+                                      void *data)
+{
+    flb_sds_t json = res_data;
+
+    /* Primary keys at top level */
+    if (!TEST_CHECK(strstr(json, "\"meta\":") != NULL)) {
+        TEST_MSG("missing top-level meta: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"level\":") != NULL)) {
+        TEST_MSG("missing top-level level: %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\"app\":") != NULL)) {
+        TEST_MSG("missing top-level app: %s", json);
+    }
+
+    /* Primary keys ALSO in line body (escaped) — original behavior */
+    if (!TEST_CHECK(strstr(json, "\\\"meta\\\":") != NULL)) {
+        TEST_MSG("meta should be in line (backward compat): %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"level\\\":") != NULL)) {
+        TEST_MSG("level should be in line (backward compat): %s", json);
+    }
+    if (!TEST_CHECK(strstr(json, "\\\"app\\\":") != NULL)) {
+        TEST_MSG("app should be in line (backward compat): %s", json);
+    }
+
+    set_output_num(get_output_num() + 1);
+    flb_sds_destroy(json);
+}
+
+void flb_test_backward_compat()
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    clear_output_num();
+
+    ctx = flb_create();
+    flb_service_set(ctx, "flush", "1", "grace", "1",
+                    "log_level", "error", NULL);
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd,
+                   "match", "test",
+                   "api_key", "test-key",
+                   NULL);
+
+    ret = flb_output_set_test(ctx, out_ffd, "formatter",
+                              cb_check_backward_compat, NULL, NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    flb_lib_push(ctx, in_ffd,
+                 (char *) JSON_WITH_PRIMARY_KEYS,
+                 sizeof(JSON_WITH_PRIMARY_KEYS) - 1);
+
+    sleep(2);
+
+    if (!TEST_CHECK(get_output_num() > 0)) {
+        TEST_MSG("formatter callback was not invoked");
+    }
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+/*
+ * Test: repeated start/push/stop/destroy cycles to verify
+ * proper resource cleanup (no crashes from leaks or double-free).
+ */
+static void cb_lifecycle_noop(void *ctx, int ffd, int res_ret,
+                              void *res_data, size_t res_size,
+                              void *data)
+{
+    flb_sds_t json = res_data;
+    flb_sds_destroy(json);
+}
+
+void flb_test_lifecycle()
+{
+    int i;
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd, out_ffd;
+
+    for (i = 0; i < 3; i++) {
+        ctx = flb_create();
+        flb_service_set(ctx, "flush", "1", "grace", "1",
+                        "log_level", "error", NULL);
+
+        in_ffd = flb_input(ctx, (char *) "lib", NULL);
+        TEST_CHECK(in_ffd >= 0);
+        flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+        out_ffd = flb_output(ctx, (char *) "logdna", NULL);
+        TEST_CHECK(out_ffd >= 0);
+        flb_output_set(ctx, out_ffd,
+                       "match", "test",
+                       "api_key", "test-key",
+                       NULL);
+
+        flb_output_set_test(ctx, out_ffd, "formatter",
+                            cb_lifecycle_noop, NULL, NULL);
+
+        ret = flb_start(ctx);
+        TEST_CHECK(ret == 0);
+
+        flb_lib_push(ctx, in_ffd,
+                     (char *) JSON_WITH_PRIMARY_KEYS,
+                     sizeof(JSON_WITH_PRIMARY_KEYS) - 1);
+
+        sleep(1);
+        flb_stop(ctx);
+        flb_destroy(ctx);
+    }
+}
+
+TEST_LIST = {
+    {"non_duplication",            flb_test_non_duplication},
+    {"data_completeness",          flb_test_data_completeness},
+    {"severity_promoted_as_level", flb_test_severity_promoted_as_level},
+    {"level_and_severity",         flb_test_level_and_severity},
+    {"default_app",                flb_test_default_app},
+    {"no_primary_keys",            flb_test_no_primary_keys},
+    {"payload_structure",          flb_test_payload_structure},
+    {"backward_compat",            flb_test_backward_compat},
+    {"lifecycle",                  flb_test_lifecycle},
+    {NULL, NULL}
+};