Skip to content

Unified CEL context and on_reply become Tasks #377

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
adam-cattermole wants to merge 12 commits into
base: task-dependencies
Choose a base branch
from
Draft

Unified CEL context and on_reply become Tasks #377

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
fd9455e
Remove implicit sequential dependency from typed action
adam-cattermole May 29, 2026
68416b1
Add task_id to action type tasks
adam-cattermole Jun 9, 2026
24960a8
Task id is no longer optional
adam-cattermole Jun 9, 2026
b23c9ee
Add cel_types to task trait
adam-cattermole Jun 9, 2026
5efd4ab
ReqRespCtx provides cel Context
adam-cattermole Jun 9, 2026
49ff5b4
All tasks leverage ctx.cel.new_ctx
adam-cattermole Jun 9, 2026
eda890d
Retain hierarchical bindings
adam-cattermole Jun 9, 2026
2394d94
on_reply actions are now Tasks
adam-cattermole Jun 9, 2026
73ae484
Add response var binding to context
adam-cattermole Jun 9, 2026
66e05aa
Ensure predicates test use correct cel::Context
adam-cattermole Jun 9, 2026
1ff92da
Add FailTask
adam-cattermole Jun 9, 2026
49318ae
Unify send and headers to have single CEL based impl
adam-cattermole Jun 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 48 additions & 26 deletions src/data/cel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -534,16 +534,7 @@ impl Predicate {
})
}

pub fn test(&self, req_ctx: &ReqRespCtx) -> PredicateResult {
let mut cel_ctx = Context::default();
self.test_with_ctx(req_ctx, &mut cel_ctx)
}

pub fn test_with_ctx(
&self,
req_ctx: &ReqRespCtx,
cel_ctx: &mut Context<'_>,
) -> PredicateResult {
pub fn test(&self, req_ctx: &ReqRespCtx, cel_ctx: &mut Context<'_>) -> PredicateResult {
match self.expression.eval(req_ctx, cel_ctx) {
Ok(AttributeState::Pending) => Ok(AttributeState::Pending),
Ok(AttributeState::Available(value)) => match value {
Expand Down Expand Up @@ -592,8 +583,9 @@ impl PredicateVec for Vec<Predicate> {
.collect();
req_ctx.ensure_attributes(&paths);

let mut cel_ctx = Context::default();
for predicate in self.iter() {
match predicate.test(req_ctx)? {
match predicate.test(req_ctx, &mut cel_ctx)? {
Comment on lines +586 to +588

@coderabbitai coderabbitai Bot Jun 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Do not reuse one mutable CEL Context across different predicates.

Expression::eval() mutates the passed context by adding functions and variable bindings, and nothing removes them afterwards. Reusing a single Context here makes predicate evaluation order-dependent: an earlier predicate can leave queryMap or other bindings behind for a later predicate that should have been evaluated in a clean context.

Suggested fix 
-        let mut cel_ctx = Context::default();
         for predicate in self.iter() {
-            match predicate.test(req_ctx, &mut cel_ctx)? {
+            let mut cel_ctx = Context::default();
+            match predicate.test(req_ctx, &mut cel_ctx)? {
                 AttributeState::Pending => {
                     return Ok(AttributeState::Pending);
                 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
let mut cel_ctx = Context::default();
for predicate in self.iter() {
match predicate.test(req_ctx)? {
match predicate.test(req_ctx, &mut cel_ctx)? {
for predicate in self.iter() {
let mut cel_ctx = Context::default();
match predicate.test(req_ctx, &mut cel_ctx)? {
AttributeState::Pending => {
return Ok(AttributeState::Pending);
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/data/cel.rs` around lines 586 - 588, The code is reusing a single mutable
CEL Context across multiple predicate evaluations which allows earlier
evaluations to leave bindings (e.g., queryMap) that affect later predicates;
change the loop so each predicate gets a fresh Context::default() (i.e., create
a new Context inside the for loop before calling predicate.test) instead of
creating one outside the loop, ensuring predicate.test(req_ctx, &mut cel_ctx) is
invoked with a newly created context for each predicate.

AttributeState::Pending => {
return Ok(AttributeState::Pending);
}
Expand Down Expand Up @@ -1091,7 +1083,7 @@ mod tests {
use crate::kuadrant::MockWasmHost;
use crate::kuadrant::ReqRespCtx;
use cel::objects::ValueType;
use cel::Value;
use cel::{Context, Value};
use std::collections::HashMap;
use std::sync::Arc;

Expand All @@ -1102,7 +1094,9 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));
let predicate = Predicate::new("source.port == 65432").expect("This is valid CEL!");
assert_eq!(
predicate.test(&ctx).expect("This must evaluate properly!"),
predicate
.test(&ctx, &mut Context::default())
.expect("This must evaluate properly!"),
AttributeState::Available(true)
);
}
Expand Down Expand Up @@ -1239,7 +1233,9 @@ mod tests {
)
.expect("This is valid!");
assert_eq!(
predicate.test(&ctx).expect("This must evaluate properly!"),
predicate
.test(&ctx, &mut Context::default())
.expect("This must evaluate properly!"),
AttributeState::Available(true)
);

Expand All @@ -1253,7 +1249,9 @@ mod tests {
)
.expect("This is valid!");
assert_eq!(
predicate.test(&ctx).expect("This must evaluate properly!"),
predicate
.test(&ctx, &mut Context::default())
.expect("This must evaluate properly!"),
AttributeState::Available(true)
);

Expand All @@ -1263,7 +1261,9 @@ mod tests {
let predicate =
Predicate::route_rule("queryMap(request.query) == {'👾': ''}").expect("This is valid!");
assert_eq!(
predicate.test(&ctx).expect("This must evaluate properly!"),
predicate
.test(&ctx, &mut Context::default())
.expect("This must evaluate properly!"),
AttributeState::Available(true)
);
}
Expand Down Expand Up @@ -1394,7 +1394,10 @@ mod tests {
"'👾' in queryMap(request.query) ? queryMap(request.query)['👾'] == '123' : false",
)
.expect("This is valid!");
assert_eq!(predicate.test(&ctx), Ok(AttributeState::Available(true)));
assert_eq!(
predicate.test(&ctx, &mut Context::default()),
Ok(AttributeState::Available(true))
);

let headers = vec![
("X-Auth".to_string(), "kuadrant".to_string()),
Expand All @@ -1404,7 +1407,10 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));
let predicate =
Predicate::route_rule("request.headers.exists(h, h.lowerAscii() == 'x-auth' && request.headers[h] == 'kuadrant')").expect("This is valid!");
assert_eq!(predicate.test(&ctx), Ok(AttributeState::Available(true)));
assert_eq!(
predicate.test(&ctx, &mut Context::default()),
Ok(AttributeState::Available(true))
);
}

#[test]
Expand Down Expand Up @@ -1474,7 +1480,9 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));

let predicate = Predicate::new("source.port == 65432").expect("This is valid CEL!");
let result = predicate.test(&ctx).expect("Test should succeed");
let result = predicate
.test(&ctx, &mut Context::default())
.expect("Test should succeed");

assert_eq!(result, AttributeState::Pending);
}
Expand Down Expand Up @@ -1603,7 +1611,9 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));
let predicate = Predicate::new("request.grpc.service == 'UserService'").expect("valid CEL");
assert_eq!(
predicate.test(&ctx).expect("must evaluate"),
predicate
.test(&ctx, &mut Context::default())
.expect("must evaluate"),
AttributeState::Available(true)
);
}
Expand All @@ -1623,7 +1633,9 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));
let predicate = Predicate::new("request.grpc.method == 'GetUser'").expect("valid CEL");
assert_eq!(
predicate.test(&ctx).expect("must evaluate"),
predicate
.test(&ctx, &mut Context::default())
.expect("must evaluate"),
AttributeState::Available(true)
);
}
Expand All @@ -1640,7 +1652,9 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));
let predicate = Predicate::new("has(request.grpc)").expect("valid CEL");
assert_eq!(
predicate.test(&ctx).expect("must evaluate"),
predicate
.test(&ctx, &mut Context::default())
.expect("must evaluate"),
AttributeState::Available(true)
);
}
Expand All @@ -1654,7 +1668,9 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));
let predicate = Predicate::new("has(request.grpc)").expect("valid CEL");
assert_eq!(
predicate.test(&ctx).expect("must evaluate"),
predicate
.test(&ctx, &mut Context::default())
.expect("must evaluate"),
AttributeState::Available(false)
);
}
Expand All @@ -1668,7 +1684,9 @@ mod tests {
let ctx = ReqRespCtx::new(Arc::new(mock_host));
let predicate = Predicate::new("has(request.grpc)").expect("valid CEL");
assert_eq!(
predicate.test(&ctx).expect("must evaluate"),
predicate
.test(&ctx, &mut Context::default())
.expect("must evaluate"),
AttributeState::Available(false)
);
}
Expand All @@ -1687,7 +1705,9 @@ mod tests {
Predicate::new("has(request.grpc) && request.grpc.service == 'UserService'")
.expect("valid CEL");
assert_eq!(
predicate.test(&ctx).expect("must evaluate"),
predicate
.test(&ctx, &mut Context::default())
.expect("must evaluate"),
AttributeState::Available(true)
);
}
Expand All @@ -1703,7 +1723,9 @@ mod tests {
Predicate::new("has(request.grpc) && request.grpc.service == 'UserService'")
.expect("valid CEL");
assert_eq!(
predicate.test(&ctx).expect("must evaluate"),
predicate
.test(&ctx, &mut Context::default())
.expect("must evaluate"),
AttributeState::Available(false)
);
}
Expand Down
131 changes: 127 additions & 4 deletions src/kuadrant/context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
use cel::Value;
use cel::{Context, Env, Value};
use std::cell::OnceCell;
use std::collections::{BTreeMap, HashMap};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use tracing::{debug, warn};
use tracing::{debug, error, warn};

use crate::data::attribute::{wasm_prop, AttributeError, AttributeState, AttributeValue, Path};
use crate::data::{Expression, Headers};
use crate::kuadrant::cache::{AttributeCache, CachedValue};
use crate::kuadrant::pipeline::tasks::Task;
use crate::kuadrant::resolver::{AttributeResolver, ProxyWasmHost};
use crate::services::ServiceError;
use tracing_opentelemetry::OpenTelemetrySpanExt;
Expand All @@ -28,6 +29,7 @@ pub struct ReqRespCtx {
tracker: Tracker,
stored_values: BTreeMap<String, Value>,
pub barrier: Barrier,
pub cel: CelScope,
}

impl Default for ReqRespCtx {
Expand All @@ -49,6 +51,7 @@ impl ReqRespCtx {
tracker: Tracker::default(),
stored_values: BTreeMap::new(),
barrier: Barrier::default(),
cel: CelScope::default(),
}
}

Expand Down Expand Up @@ -467,7 +470,7 @@ impl Barrier {
match self.count.checked_sub(1) {
Some(new_value) => self.count = new_value,
None => {
tracing::error!(
error!(
"Attempted to lower upstream barrier when count is already 0 - mismatched raise/lower pairs"
);
}
Expand Down Expand Up @@ -514,6 +517,64 @@ impl BodyContext {
}
}

pub struct CelScope {
env: Arc<Env>,
registered: HashSet<String>,
bindings: BTreeMap<String, Vec<(String, Value)>>,
}

impl Default for CelScope {
fn default() -> Self {
Self {
env: Arc::new(Env::stdlib()),
registered: HashSet::new(),
bindings: BTreeMap::new(),
}
}
}

impl CelScope {
pub fn new_ctx(&mut self, task: &dyn Task) -> Context<'static> {
let task_id = task.id();

if !self.registered.contains(task_id) {
let types = task.cel_types();
if !types.is_empty() {
match Arc::get_mut(&mut self.env) {
Some(env) => {
for type_def in types {
env.add_struct(type_def);
}
}
None => error!("Failed to add CEL types: Arc refcount > 1"),
}
}
self.registered.insert(task_id.to_string());
Comment on lines +540 to +552

@coderabbitai coderabbitai Bot Jun 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Do not mark a task as registered when CEL type registration failed.

If Arc::get_mut(&mut self.env) returns None, this path logs the error but still inserts task_id into registered. That turns a transient failure into a permanent one: every later new_ctx() call for the same task id will skip cel_types() registration, so tasks that depend on those structs will keep evaluating against an incomplete CEL environment.

Suggested fix 
         if !self.registered.contains(task_id) {
             let types = task.cel_types();
+            let mut registration_succeeded = true;
             if !types.is_empty() {
                 match Arc::get_mut(&mut self.env) {
                     Some(env) => {
                         for type_def in types {
                             env.add_struct(type_def);
                         }
                     }
-                    None => error!("Failed to add CEL types: Arc refcount > 1"),
+                    None => {
+                        error!("Failed to add CEL types: Arc refcount > 1");
+                        registration_succeeded = false;
+                    }
                 }
             }
-            self.registered.insert(task_id.to_string());
+            if registration_succeeded {
+                self.registered.insert(task_id.to_string());
+            }
         }
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/kuadrant/context.rs` around lines 540 - 552, Currently you insert task_id
into self.registered even when Arc::get_mut(&mut self.env) fails; change the
logic so task_id is only inserted when CEL type registration actually succeeds.
Specifically, for the block that checks !self.registered.contains(task_id) and
then calls task.cel_types(), if types is empty insert immediately; otherwise
attempt Arc::get_mut(&mut self.env) and only after successfully iterating and
calling env.add_struct(...) for each type_def insert
self.registered.insert(task_id.to_string()); do not insert when Arc::get_mut
returns None (log and bail out so future new_ctx() calls will retry
registration).

}

let mut ctx = Context::with_env(Arc::clone(&self.env));
for (scope_id, scope_bindings) in &self.bindings {
if is_ancestor(scope_id, task_id) {
for (name, value) in scope_bindings {
ctx.add_variable_from_value(name, value.clone());
}
}
}
ctx
}

pub fn add_scoped_binding(&mut self, task_id: &str, name: String, val: Value) {
self.bindings
.entry(task_id.to_string())
.or_default()
.push((name, val));
}
}

fn is_ancestor(scope_id: &str, task_id: &str) -> bool {
task_id == scope_id || task_id.starts_with(&format!("{}.", scope_id))
}

#[cfg(test)]
mod tests {
use super::*;
Expand Down Expand Up @@ -754,4 +815,66 @@ mod tests {
Ok(AttributeState::Available(Some(ref s))) if s == "external-user-id"
));
}

#[test]
fn test_cel_scope_hierarchical_bindings() {
use crate::kuadrant::pipeline::tasks::{Task, TaskOutcome};

struct MockTask {
id: String,
}
impl Task for MockTask {
fn id(&self) -> &str {
&self.id
}
fn apply(self: Box<Self>, _ctx: &mut ReqRespCtx) -> TaskOutcome {
TaskOutcome::Done
}
}

let mut scope = CelScope::default();

// Task "0" gets a binding
scope.add_scoped_binding(
"0",
"my_response".to_string(),
Value::String(Arc::new("user123".to_string())),
);

// Task "0.0" should see "my_response" from parent "0"
let task_0_0 = MockTask {
id: "0.0".to_string(),
};
let ctx_0_0 = scope.new_ctx(&task_0_0);
assert!(ctx_0_0.get_variable("my_response").is_some());

// Task "1" should NOT see "my_response" (different branch)
let task_1 = MockTask {
id: "1".to_string(),
};
let ctx_1 = scope.new_ctx(&task_1);
assert!(ctx_1.get_variable("my_response").is_none());
}

#[test]
fn test_is_ancestor() {
// Same task
assert!(is_ancestor("0", "0"));

// Direct child
assert!(is_ancestor("0", "0.0"));

// Nested child
assert!(is_ancestor("0", "0.0.1"));

// Different branch
assert!(!is_ancestor("0", "1"));
assert!(!is_ancestor("0", "1.0"));

// Sibling
assert!(!is_ancestor("0.0", "0.1"));

// Parent relationship is not symmetric
assert!(!is_ancestor("0.0", "0"));
}
}
Loading
Loading