diff --git a/include/fluent-bit/flb_conditionals.h b/include/fluent-bit/flb_conditionals.h
new file mode 100644
index 00000000000..d80afe666da
--- /dev/null
+++ b/include/fluent-bit/flb_conditionals.h
@@ -0,0 +1,91 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/* Fluent Bit
+ * ==========
+ * Copyright (C) 2015-2024 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_CONDITIONS_H
+#define FLB_CONDITIONS_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_regex.h>
+#include <fluent-bit/flb_mp.h>
+#include <fluent-bit/flb_record_accessor.h>
+#include <monkey/mk_core.h>
+#include <fluent-bit/flb_mp_chunk.h>
+
+/* Context types enum */
+enum record_context_type {
+    RECORD_CONTEXT_BODY    = 0,
+    RECORD_CONTEXT_METADATA    = 1
+};
+
+struct flb_condition;
+
+enum flb_condition_operator {
+    FLB_COND_OP_AND,
+    FLB_COND_OP_OR
+};
+
+enum flb_rule_operator {
+    FLB_RULE_OP_EQ,
+    FLB_RULE_OP_NEQ,
+    FLB_RULE_OP_GT,
+    FLB_RULE_OP_LT,
+    FLB_RULE_OP_REGEX,
+    FLB_RULE_OP_IN,
+    FLB_RULE_OP_NOT_IN
+};
+
+struct flb_condition_rule {
+    struct flb_cfl_record_accessor *ra;  /* Record accessor for the field */
+    enum record_context_type context;    /* Whether rule applies to body or metadata */
+    enum flb_rule_operator op;
+    union {
+        flb_sds_t str_val;
+        double num_val;
+        struct {
+            flb_sds_t *values;
+            int count;
+        } array;
+    } value;
+    struct flb_regex *regex;
+    struct mk_list _head;
+};
+
+struct flb_condition {
+    enum flb_condition_operator op;
+    struct mk_list rules;
+};
+
+/* Core condition functions */
+struct flb_condition *flb_condition_create(enum flb_condition_operator op);
+
+int flb_condition_add_rule(struct flb_condition *cond,
+                          const char *field,
+                          enum flb_rule_operator op,
+                          void *value,
+                          int value_count,
+                          enum record_context_type context);
+
+void flb_condition_destroy(struct flb_condition *cond);
+
+/* Evaluation function */
+int flb_condition_evaluate(struct flb_condition *cond,
+                          struct flb_mp_chunk_record *record);
+
+#endif /* FLB_CONDITIONS_H */
\ No newline at end of file
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 9ba91972ac0..b3df3ebbb91 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -88,6 +88,7 @@ set(src
   flb_lock.c
   flb_cfl_ra_key.c
   flb_cfl_record_accessor.c
+  flb_conditionals.c
   )
 
 # Config format
diff --git a/src/flb_conditionals.c b/src/flb_conditionals.c
new file mode 100644
index 00000000000..ffc5fe8ffa5
--- /dev/null
+++ b/src/flb_conditionals.c
@@ -0,0 +1,365 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/* Fluent Bit
+ * ==========
+ * Copyright (C) 2015-2024 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/flb_info.h>
+#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_error.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_regex.h>
+#include <fluent-bit/flb_record_accessor.h>
+#include <fluent-bit/flb_mp.h>
+#include <fluent-bit/flb_conditionals.h>
+#include <fluent-bit/flb_cfl_record_accessor.h>
+
+/* Function to get the record variant based on context */
+static inline struct cfl_variant *get_record_variant(struct flb_mp_chunk_record *record,
+                                                    enum record_context_type context_type)
+{
+    if (!record) {
+        return NULL;
+    }
+
+    switch (context_type) {
+    case RECORD_CONTEXT_METADATA:
+        if (record->cobj_metadata) {
+            return record->cobj_metadata->variant;
+        }
+        break;
+
+    case RECORD_CONTEXT_BODY:
+        if (record->cobj_record) {
+            return record->cobj_record->variant;
+        }
+        break;
+    }
+
+    return NULL;
+}
+
+static struct flb_condition_rule *rule_create(const char *field,
+                                             enum flb_rule_operator op,
+                                             void *value,
+                                             int value_count,
+                                             enum record_context_type context)
+{
+    struct flb_condition_rule *rule;
+    int i,j;
+
+    if (!field || !value) {
+        return NULL;
+    }
+
+    switch (op) {
+    case FLB_RULE_OP_EQ:
+    case FLB_RULE_OP_NEQ:
+    case FLB_RULE_OP_REGEX:
+        if (!value || !((char *)value)[0]) {
+            return NULL;
+        }
+        break;
+    case FLB_RULE_OP_GT:
+    case FLB_RULE_OP_LT:
+        if (!value) {
+            return NULL;
+        }
+        break;
+
+    case FLB_RULE_OP_IN:
+    case FLB_RULE_OP_NOT_IN:
+        if (!value || value_count <= 0 || !((char **)value)[0]) {
+            return NULL;
+        }
+        for (i = 0; i < value_count; i++) {
+            if (!((char **)value)[i]) {
+                return NULL;
+            }
+        }
+        break;
+
+    default:
+        return NULL;
+    }
+
+    rule = flb_calloc(1, sizeof(struct flb_condition_rule));
+    if (!rule) {
+        cfl_errno();
+        return NULL;
+    }
+
+    rule->ra = flb_cfl_ra_create((char *)field, FLB_TRUE);
+    if (!rule->ra) {
+        flb_free(rule);
+        return NULL;
+    }
+
+    rule->context = context;
+    rule->op = op;
+
+    switch (op) {
+    case FLB_RULE_OP_NEQ:
+    case FLB_RULE_OP_EQ:
+        rule->value.str_val = flb_sds_create((char *)value);
+        if (!rule->value.str_val) {
+            flb_cfl_ra_destroy(rule->ra);
+            flb_free(rule);
+            return NULL;
+        }
+        break;
+
+    case FLB_RULE_OP_GT:
+    case FLB_RULE_OP_LT:
+        rule->value.num_val = *(double *)value;
+        break;
+
+    case FLB_RULE_OP_REGEX:
+        rule->regex = flb_regex_create((char *)value);
+        if (!rule->regex) {
+            flb_cfl_ra_destroy(rule->ra);
+            flb_free(rule);
+            return NULL;
+        }
+        break;
+
+    case FLB_RULE_OP_IN:
+    case FLB_RULE_OP_NOT_IN:
+        rule->value.array.values = flb_calloc(value_count, sizeof(flb_sds_t));
+        if (!rule->value.array.values) {
+            flb_cfl_ra_destroy(rule->ra);
+            flb_free(rule);
+            return NULL;
+        }
+
+        for (i = 0; i < value_count; i++) {
+            rule->value.array.values[i] = flb_sds_create(((char **)value)[i]);
+            if (!rule->value.array.values[i]) {
+                for (j = 0; j < i; j++) {
+                    flb_sds_destroy(rule->value.array.values[j]);
+                }
+                flb_free(rule->value.array.values);
+                flb_cfl_ra_destroy(rule->ra);
+                flb_free(rule);
+                return NULL;
+            }
+        }
+        rule->value.array.count = value_count;
+        break;
+    }
+
+    return rule;
+}
+
+static void rule_destroy(struct flb_condition_rule *rule)
+{
+    int i;
+
+    if (!rule) {
+        return;
+    }
+
+    if (rule->ra) {
+        flb_cfl_ra_destroy(rule->ra);
+    }
+
+    switch (rule->op) {
+    case FLB_RULE_OP_EQ:
+    case FLB_RULE_OP_NEQ:
+        if (rule->value.str_val) {
+            flb_sds_destroy(rule->value.str_val);
+        }
+        break;
+
+    case FLB_RULE_OP_REGEX:
+        if (rule->regex) {
+            flb_regex_destroy(rule->regex);
+        }
+        break;
+
+    case FLB_RULE_OP_IN:
+    case FLB_RULE_OP_NOT_IN:
+        for (i = 0; i < rule->value.array.count; i++) {
+            flb_sds_destroy(rule->value.array.values[i]);
+        }
+        flb_free(rule->value.array.values);
+        break;
+
+    case FLB_RULE_OP_GT:
+    case FLB_RULE_OP_LT:
+        break;
+
+    default:
+        break;
+    }
+
+    flb_free(rule);
+}
+
+struct flb_condition *flb_condition_create(enum flb_condition_operator op)
+{
+    struct flb_condition *cond;
+
+    cond = flb_calloc(1, sizeof(struct flb_condition));
+    if (!cond) {
+        cfl_errno();
+        return NULL;
+    }
+
+    cond->op = op;
+    mk_list_init(&cond->rules);
+
+    return cond;
+}
+
+int flb_condition_add_rule(struct flb_condition *cond,
+                           const char *field,
+                           enum flb_rule_operator op,
+                           void *value,
+                           int value_count,
+                           enum record_context_type context)
+{
+    struct flb_condition_rule *rule;
+
+    if (!cond || !field || !value) {
+        return FLB_FALSE;
+    }
+
+    rule = rule_create(field, op, value, value_count, context);
+    if (!rule) {
+        return FLB_FALSE;
+    }
+
+    mk_list_add(&rule->_head, &cond->rules);
+    return FLB_TRUE;
+}
+
+void flb_condition_destroy(struct flb_condition *cond)
+{
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct flb_condition_rule *rule;
+
+    if (!cond) {
+        return;
+    }
+
+    mk_list_foreach_safe(head, tmp, &cond->rules) {
+        rule = mk_list_entry(head, struct flb_condition_rule, _head);
+        mk_list_del(&rule->_head);
+        rule_destroy(rule);
+    }
+
+    flb_free(cond);
+}
+
+static int evaluate_rule(struct flb_condition_rule *rule,
+                         struct cfl_variant *record_variant)
+{
+    flb_sds_t str_val;
+    int i;
+    int result = FLB_FALSE;
+    double num_val;
+
+    if (!rule || !record_variant) {
+        return FLB_FALSE;
+    }
+
+    str_val = flb_cfl_ra_translate(rule->ra, NULL, 0, *record_variant, NULL);
+    if (!str_val) {
+        return FLB_FALSE;
+    }
+
+    switch (rule->op) {
+    case FLB_RULE_OP_EQ:
+        result = (strcmp(str_val, rule->value.str_val) == 0);
+        break;
+
+    case FLB_RULE_OP_NEQ:
+        result = (strcmp(str_val, rule->value.str_val) != 0);
+        break;
+
+    case FLB_RULE_OP_GT:
+        num_val = atof(str_val);
+        result = (num_val > rule->value.num_val);
+        break;
+
+    case FLB_RULE_OP_LT:
+        num_val = atof(str_val);
+        result = (num_val < rule->value.num_val);
+        break;
+
+    case FLB_RULE_OP_REGEX:
+        result = (flb_regex_match(rule->regex,
+                                  (unsigned char *)str_val,
+                                  flb_sds_len(str_val)) > 0);
+        break;
+
+    case FLB_RULE_OP_IN:
+    case FLB_RULE_OP_NOT_IN:
+        for (i = 0; i < rule->value.array.count; i++) {
+            if (strcmp(str_val, rule->value.array.values[i]) == 0) {
+                result = (rule->op == FLB_RULE_OP_IN);
+                break;
+            }
+        }
+        if (i == rule->value.array.count) {
+            result = (rule->op == FLB_RULE_OP_NOT_IN);
+        }
+        break;
+    }
+
+    flb_sds_destroy(str_val);
+    return result;
+}
+
+int flb_condition_evaluate(struct flb_condition *cond,
+                           struct flb_mp_chunk_record *record)
+{
+    struct mk_list *head;
+    struct flb_condition_rule *rule;
+    struct cfl_variant *record_variant;
+    int result;
+
+    if (!cond || !record) {
+        return FLB_TRUE;
+    }
+
+    if (mk_list_size(&cond->rules) == 0) {
+        return (cond->op == FLB_COND_OP_AND);
+    }
+
+    mk_list_foreach(head, &cond->rules) {
+        rule = mk_list_entry(head, struct flb_condition_rule, _head);
+
+        /* Get the variant for this rule's context */
+        record_variant = get_record_variant(record, rule->context);
+        if (!record_variant) {
+            continue;
+        }
+
+        result = evaluate_rule(rule, record_variant);
+
+        if (cond->op == FLB_COND_OP_AND && result == FLB_FALSE) {
+            return FLB_FALSE;
+        }
+        else if (cond->op == FLB_COND_OP_OR && result == FLB_TRUE) {
+            return FLB_TRUE;
+        }
+    }
+
+    return (cond->op == FLB_COND_OP_AND) ? FLB_TRUE : FLB_FALSE;
+}
diff --git a/tests/internal/CMakeLists.txt b/tests/internal/CMakeLists.txt
index c8225739b8c..ee6a086d958 100644
--- a/tests/internal/CMakeLists.txt
+++ b/tests/internal/CMakeLists.txt
@@ -43,6 +43,7 @@ set(UNIT_TESTS_FILES
   processor.c
   uri.c
   msgpack_append_message.c
+  flb_conditionals.c
   endianness
   )
 
diff --git a/tests/internal/flb_conditionals.c b/tests/internal/flb_conditionals.c
new file mode 100644
index 00000000000..c746b8e3e46
--- /dev/null
+++ b/tests/internal/flb_conditionals.c
@@ -0,0 +1,1142 @@
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_log.h>
+#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_conditionals.h>
+#include <fluent-bit/flb_mp.h>
+#include <fluent-bit/flb_mp_chunk.h>
+#include <fluent-bit/flb_utils.h>
+
+#include "flb_tests_internal.h"
+
+struct test_record {
+    msgpack_object *obj;
+    msgpack_sbuffer sbuf;
+    msgpack_zone zone;
+    struct cfl_variant *variant;
+    struct cfl_kvlist *kvlist;
+    /* Stack objects */
+    struct cfl_variant stack_variant;
+    struct cfl_object stack_obj;
+    struct flb_mp_chunk_record chunk;
+};
+
+static struct test_record *create_test_record(const char *key, const char *value);
+static struct test_record *create_test_record_numeric(const char *key, double value);
+static struct test_record *create_test_record_with_meta(const char *key, const char *value,
+                                                      const char *meta_key, const char *meta_value);
+static void destroy_test_record(struct test_record *record);
+
+static struct test_record *create_test_record(const char *key, const char *value)
+{
+    struct test_record *record;
+    msgpack_packer pck;
+    char *key_copy;
+    char *value_copy;
+
+    record = flb_calloc(1, sizeof(struct test_record));
+    if (!record) {
+        return NULL;
+    }
+
+    /* Initialize buffers */
+    msgpack_sbuffer_init(&record->sbuf);
+    msgpack_zone_init(&record->zone, 2048);
+    msgpack_packer_init(&pck, &record->sbuf, msgpack_sbuffer_write);
+
+    /* Pack key-value */
+    msgpack_pack_map(&pck, 1);
+    msgpack_pack_str(&pck, strlen(key));
+    msgpack_pack_str_body(&pck, key, strlen(key));
+    msgpack_pack_str(&pck, strlen(value));
+    msgpack_pack_str_body(&pck, value, strlen(value));
+
+    /* Unpack to get the object */
+    record->obj = msgpack_zone_malloc(&record->zone, sizeof(msgpack_object));
+    if (!record->obj) {
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    msgpack_unpack(record->sbuf.data, record->sbuf.size, NULL,
+                  &record->zone, record->obj);
+
+    /* Create kvlist */
+    record->kvlist = cfl_kvlist_create();
+    if (!record->kvlist) {
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Create mutable copies of strings */
+    key_copy = strdup(key);
+    value_copy = strdup(value);
+    if (!key_copy || !value_copy) {
+        if (key_copy) free(key_copy);
+        if (value_copy) free(value_copy);
+        cfl_kvlist_destroy(record->kvlist);
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Insert the key-value pair */
+    if (cfl_kvlist_insert_string(record->kvlist, key_copy, value_copy) != 0) {
+        free(key_copy);
+        free(value_copy);
+        cfl_kvlist_destroy(record->kvlist);
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Create variant */
+    record->variant = cfl_variant_create();
+    if (!record->variant) {
+        cfl_kvlist_destroy(record->kvlist);
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Set the variant type and data */
+    record->variant->type = CFL_VARIANT_KVLIST;
+    record->variant->data.as_kvlist = record->kvlist;
+
+    /* Set up the stack objects */
+    record->stack_variant.type = CFL_VARIANT_KVLIST;
+    record->stack_variant.data.as_kvlist = record->kvlist;
+    record->stack_obj.type = CFL_VARIANT_KVLIST;
+    record->stack_obj.variant = &record->stack_variant;
+
+    /* Set up the chunk record */
+    record->chunk.event.body = record->obj;
+    record->chunk.cobj_record = &record->stack_obj;
+
+    return record;
+}
+
+struct test_record *create_test_record_with_meta(const char *key, const char *value, 
+                                               const char *meta_key, const char *meta_value)
+{
+    struct test_record *record;
+    char *meta_key_copy;
+    char *meta_value_copy;
+    struct cfl_kvlist *meta_kvlist;
+
+    /* Create regular record first */
+    record = create_test_record(key, value);
+    if (!record) {
+        return NULL;
+    }
+
+    /* Add metadata kvlist */
+    meta_kvlist = cfl_kvlist_create();
+    if (!meta_kvlist) {
+        destroy_test_record(record);
+        return NULL;
+    }
+
+    /* Create mutable copies for metadata */
+    meta_key_copy = strdup(meta_key);
+    meta_value_copy = strdup(meta_value);
+    if (!meta_key_copy || !meta_value_copy) {
+        if (meta_key_copy) free(meta_key_copy);
+        if (meta_value_copy) free(meta_value_copy);
+        cfl_kvlist_destroy(meta_kvlist);
+        destroy_test_record(record);
+        return NULL;
+    }
+
+    /* Insert metadata */
+    if (cfl_kvlist_insert_string(meta_kvlist, meta_key_copy, meta_value_copy) != 0) {
+        free(meta_key_copy);
+        free(meta_value_copy);
+        cfl_kvlist_destroy(meta_kvlist);
+        destroy_test_record(record);
+        return NULL;
+    }
+
+    /* Create metadata variant */
+    record->chunk.cobj_metadata = flb_calloc(1, sizeof(struct cfl_object));
+    if (!record->chunk.cobj_metadata) {
+        cfl_kvlist_destroy(meta_kvlist);
+        destroy_test_record(record);
+        return NULL;
+    }
+
+    record->chunk.cobj_metadata->variant = cfl_variant_create();
+    if (!record->chunk.cobj_metadata->variant) {
+        flb_free(record->chunk.cobj_metadata);
+        cfl_kvlist_destroy(meta_kvlist);
+        destroy_test_record(record);
+        return NULL;
+    }
+
+    record->chunk.cobj_metadata->type = CFL_VARIANT_KVLIST;
+    record->chunk.cobj_metadata->variant->type = CFL_VARIANT_KVLIST;
+    record->chunk.cobj_metadata->variant->data.as_kvlist = meta_kvlist;
+
+    return record;
+}
+
+static struct test_record *create_test_record_numeric(const char *key, double value)
+{
+    struct test_record *record;
+    msgpack_packer pck;
+    char *key_copy;
+
+    record = flb_calloc(1, sizeof(struct test_record));
+    if (!record) {
+        return NULL;
+    }
+
+    /* Initialize buffers */
+    msgpack_sbuffer_init(&record->sbuf);
+    msgpack_zone_init(&record->zone, 2048);
+    msgpack_packer_init(&pck, &record->sbuf, msgpack_sbuffer_write);
+
+    /* Pack key-value */
+    msgpack_pack_map(&pck, 1);
+    msgpack_pack_str(&pck, strlen(key));
+    msgpack_pack_str_body(&pck, key, strlen(key));
+    msgpack_pack_double(&pck, value);
+
+    /* Unpack to get the object */
+    record->obj = msgpack_zone_malloc(&record->zone, sizeof(msgpack_object));
+    if (!record->obj) {
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    msgpack_unpack(record->sbuf.data, record->sbuf.size, NULL,
+                  &record->zone, record->obj);
+
+    /* Create kvlist */
+    record->kvlist = cfl_kvlist_create();
+    if (!record->kvlist) {
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Create mutable copy of key */
+    key_copy = strdup(key);
+    if (!key_copy) {
+        cfl_kvlist_destroy(record->kvlist);
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Insert the key-value pair */
+    if (cfl_kvlist_insert_double(record->kvlist, key_copy, value) != 0) {
+        free(key_copy);
+        cfl_kvlist_destroy(record->kvlist);
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Create variant */
+    record->variant = cfl_variant_create();
+    if (!record->variant) {
+        cfl_kvlist_destroy(record->kvlist);
+        msgpack_zone_destroy(&record->zone);
+        msgpack_sbuffer_destroy(&record->sbuf);
+        flb_free(record);
+        return NULL;
+    }
+
+    /* Set the variant type and data */
+    record->variant->type = CFL_VARIANT_KVLIST;
+    record->variant->data.as_kvlist = record->kvlist;
+
+    /* Set up the stack objects */
+    record->stack_variant.type = CFL_VARIANT_KVLIST;
+    record->stack_variant.data.as_kvlist = record->kvlist;
+    record->stack_obj.type = CFL_VARIANT_KVLIST;
+    record->stack_obj.variant = &record->stack_variant;
+
+    /* Set up the chunk record */
+    record->chunk.event.body = record->obj;
+    record->chunk.cobj_record = &record->stack_obj;
+
+    return record;
+}
+
+static void destroy_test_record(struct test_record *record)
+{
+    if (!record) {
+        return;
+    }
+
+    if (record->variant) {
+        cfl_variant_destroy(record->variant);
+    }
+
+    msgpack_zone_destroy(&record->zone);
+    msgpack_sbuffer_destroy(&record->sbuf);
+    flb_free(record);
+}
+
+void test_condition_equals()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    /* Test matching equals condition */
+    record_data = create_test_record("level", "error");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ, 
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test non-matching equals condition */
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_numeric()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+    double val;
+
+    /* Test greater than */
+    record_data = create_test_record_numeric("count", 42.0);
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 40.0;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_GT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test less than */
+    record_data = create_test_record_numeric("count", 42.0);
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 50.0;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_LT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_not_equals()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    /* Test not equals - should match */
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NEQ, 
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test not equals with matching value - should not match */
+    record_data = create_test_record("level", "error");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NEQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_in()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+    const char *values[] = {"error", "warn", "fatal"};
+
+    /* Test value in array */
+    record_data = create_test_record("level", "error");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_IN,
+                                    (void *)values, 3, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test value not in array */
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_IN,
+                                    (void *)values, 3, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_not_in()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+    const char *values[] = {"error", "warn", "fatal"};
+
+    /* Test value not in array */
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NOT_IN,
+                                    (void *)values, 3, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test value in array */
+    record_data = create_test_record("level", "error");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NOT_IN,
+                                    (void *)values, 3, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_and()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    /* Test both conditions true */
+    record_data = create_test_record("level", "error");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NEQ,
+                                    "info", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test one condition false */
+    record_data = create_test_record("level", "warn");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NEQ,
+                                    "info", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_or()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    /* Test one condition true */
+    record_data = create_test_record("level", "error");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_OR);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "warn", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test both conditions false */
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_OR);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "warn", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_empty()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    /* Test empty AND condition */
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+
+    /* Test empty OR condition */
+    cond = flb_condition_create(FLB_COND_OP_OR);
+    TEST_CHECK(cond != NULL);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_regex()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    /* Test matching regex */
+    record_data = create_test_record("path", "/api/v1/users");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$path", FLB_RULE_OP_REGEX,
+                                    "^/api/.*$", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test non-matching regex */
+    record_data = create_test_record("path", "/other/endpoint");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$path", FLB_RULE_OP_REGEX,
+                                    "^/api/.*$", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test empty string */
+    record_data = create_test_record("path", "");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$path", FLB_RULE_OP_REGEX,
+                                    "^/api/.*$", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_invalid_expressions()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    /* Test NULL condition */
+    result = flb_condition_evaluate(NULL, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    /* Test NULL record */
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, NULL);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+
+    /* Test invalid record accessor */
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$[invalid", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+
+    /* Test invalid operator */
+    cond = flb_condition_create(999);  /* Invalid operator */
+    TEST_CHECK(cond != NULL);  /* Should still create but with default op */
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* Default AND behavior */
+
+    flb_condition_destroy(cond);
+
+    /* Test NULL key */
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, NULL, FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+
+    /* Test NULL value */
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    NULL, 0, RECORD_CONTEXT_BODY) == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+
+    /* Test invalid regex pattern */
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_REGEX,
+                                    "[invalid", 0, RECORD_CONTEXT_BODY) == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_metadata()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+
+    /* Test metadata match */
+    record_data = create_test_record_with_meta("message", "test log",
+                                             "streamName", "production");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$streamName", FLB_RULE_OP_EQ,
+                                    "production", 0, RECORD_CONTEXT_METADATA) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test metadata no match */
+    record_data = create_test_record_with_meta("message", "test log",
+                                             "streamName", "staging");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$streamName", FLB_RULE_OP_EQ,
+                                    "production", 0, RECORD_CONTEXT_METADATA) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test combination of metadata and body conditions */
+    record_data = create_test_record_with_meta("level", "error",
+                                             "streamName", "production");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$streamName", FLB_RULE_OP_EQ,
+                                    "production", 0, RECORD_CONTEXT_METADATA) == FLB_TRUE);
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "error", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_missing_values()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+    const char *values[] = {"error", "warn", "fatal"};
+
+    /* Test IN operator with missing body field */
+    record_data = create_test_record("other_field", "some_value");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$non_existent", FLB_RULE_OP_IN,
+                                    (void *)values, 3, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* Missing field should return false */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test NOT_IN operator with present field not in array */
+    record_data = create_test_record("level", "info");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NOT_IN,
+                                    (void *)values, 3, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* Present value not in array should return true */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test NOT_IN operator with present field in array */
+    record_data = create_test_record("level", "error");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_NOT_IN,
+                                    (void *)values, 3, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* Present value in array should return false */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_border_cases()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+    const char *values[] = {"error", "warn", "fatal", ""}; // Removed NULL
+    double val;
+
+    /* Test numeric comparison with non-numeric string */
+    record_data = create_test_record("count", "not_a_number");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 42.0;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_GT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* Non-numeric string should fail comparison */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test empty string in IN operator */
+    record_data = create_test_record("level", "");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_IN,
+                                    (void *)values, 4, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* Empty string should match empty string in array */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test regex with metacharacters */
+    record_data = create_test_record("path", "/api/v1/users[123]");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$path", FLB_RULE_OP_REGEX,
+                                    "^/api/v1/users\\[[0-9]+\\]$", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test combination of missing and empty fields with AND */
+    record_data = create_test_record("level", "");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    /* Use a non-empty string instead of empty string for comparison */
+    TEST_CHECK(flb_condition_add_rule(cond, "$level", FLB_RULE_OP_EQ,
+                                    "non-empty", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+    TEST_CHECK(flb_condition_add_rule(cond, "$non_existent", FLB_RULE_OP_EQ,
+                                    "non-empty", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* Should fail because both conditions are false */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test empty regex pattern - this should fail at rule creation */
+    record_data = create_test_record("path", "");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$path", FLB_RULE_OP_REGEX,
+                                    "", 0, RECORD_CONTEXT_BODY) == FLB_FALSE);  /* Should fail to create rule */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test metadata with non-existent field */
+    record_data = create_test_record_with_meta("message", "test",
+                                             "streamName", "production");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$non_existent", FLB_RULE_OP_EQ,
+                                    "production", 0, RECORD_CONTEXT_METADATA) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* Should fail because metadata field is missing */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+void test_condition_numeric_edge_cases()
+{
+    struct test_record *record_data;
+    struct flb_condition *cond;
+    int result;
+    double val;
+
+    /* Test non-numeric string */
+    record_data = create_test_record("count", "not_a_number");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 42.0;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_GT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* Non-numeric string should return false */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test numeric string */
+    record_data = create_test_record("count", "42.5");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 42.0;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_GT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* "42.5" > 42.0 should be true */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test infinity */
+    record_data = create_test_record("count", "inf");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 1e308;  /* Very large number */
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_GT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* inf should be greater than any finite number */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test negative infinity */
+    record_data = create_test_record("count", "-inf");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = -1e308;  /* Very large negative number */
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_LT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* -inf should be less than any finite number */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test NaN */
+    record_data = create_test_record("count", "NaN");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 0.0;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_GT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_FALSE);  /* NaN comparisons should return false */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test very large number */
+    record_data = create_test_record("count", "1e308");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 1e307;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_GT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* Very large number comparison */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test very small number */
+    record_data = create_test_record("count", "1e-308");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    val = 1e-307;
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_LT,
+                                    &val, 0, RECORD_CONTEXT_BODY) == FLB_TRUE);
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* Very small number comparison */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test zero */
+    record_data = create_test_record("count", "0");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    /* Test zero */
+    record_data = create_test_record("count", "0");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_EQ,
+                                    "0", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);  // Use string "0" instead of double
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* Zero comparison */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+
+    /* Test negative zero */
+    record_data = create_test_record("count", "-0");
+    TEST_CHECK(record_data != NULL);
+
+    cond = flb_condition_create(FLB_COND_OP_AND);
+    TEST_CHECK(cond != NULL);
+
+    TEST_CHECK(flb_condition_add_rule(cond, "$count", FLB_RULE_OP_EQ,
+                                    "-0", 0, RECORD_CONTEXT_BODY) == FLB_TRUE);  // Use string "-0" instead of double
+
+    result = flb_condition_evaluate(cond, &record_data->chunk);
+    TEST_CHECK(result == FLB_TRUE);  /* Negative zero should equal zero */
+
+    flb_condition_destroy(cond);
+    destroy_test_record(record_data);
+}
+
+TEST_LIST = {
+    {"equals", test_condition_equals},
+    {"not_equals", test_condition_not_equals},
+    {"numeric", test_condition_numeric},
+    {"numeric_edge_cases", test_condition_numeric_edge_cases},
+    {"in", test_condition_in},
+    {"not_in", test_condition_not_in},
+    {"regex", test_condition_regex},
+    {"and", test_condition_and},
+    {"or", test_condition_or},
+    {"empty", test_condition_empty},
+    {"invalid_expressions", test_condition_invalid_expressions},
+    {"metadata", test_condition_metadata},
+    {"missing_values", test_condition_missing_values},
+    {"border_cases", test_condition_border_cases},
+    {NULL, NULL}
+};
\ No newline at end of file