diff --git a/CHANGELOG.md b/CHANGELOG.md index a8738aa9..6ff67ecd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,11 +25,23 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/). - **BREAKING: `TokenRole::Workload` renamed to `TokenRole::CallerWorkload`.** A serde `alias = "workload"` keeps existing serialized config loading, but the Rust symbol is renamed — downstream Rust code must update. (#131) - **BREAKING: `DelegationMode::AsGateway` renamed to `AsThisWorkload`.** A serde `alias = "as_gateway"` keeps persisted values deserializing. (#131) - **BREAKING: `DelegationKey` is now `#[non_exhaustive]`** and gained a `client_id` field (partitioning the delegated-token cache per calling OAuth client, mirroring `workload_id`). Construct it via `DelegationKey::new(mode, audience, scopes)` + the `with_subject_id` / `with_workload_id` / `with_client_id` setters rather than a struct literal. (#131) +- **BREAKING: `PipelineResult` is now `#[non_exhaustive]`** and gained a `payload_modified` field. `modified_payload` is `Some` on every allowed pipeline, carrying the final payload whether or not a plugin touched it, so it never answered "did anything change?" — read the new flag for that. Construct via `allowed_with` / `denied` plus the `with_errors` / `with_payload_modified` builders rather than a struct literal; exhaustive destructuring must gain a `..` arm. (#151) +- **`payload_modified` is carried across the FFI to the Python and Go bindings.** `FfiPipelineResult`, `PyPipelineResult` (as a `payload_modified` property), and Go's `PipelineResult` / `TypedPipelineResult` all expose it, so non-Rust hosts can distinguish an accepted mutation from a payload the pipeline merely carried. Additive on the MessagePack wire, so the FFI ABI version is unchanged. The Go `Invoke` doc example no longer presents `ModifiedPayload != nil` as a mutation test — it is true on every allowed pipeline. (#151) +- **BREAKING: a pipeline field name reported to a plugin is now root-relative everywhere.** A `do:`-block field op passed `args.city` where the `args:` / `result:` sections passed `city`; both now pass `city`, with the phase selecting the root. The type of `PluginInvocation::Field.name` is unchanged, so this is a silent semantic change: any out-of-tree `PluginInvoker` that stripped an `args.` / `result.` prefix must drop that handling or it will mis-resolve the field. (#151) +- **A pipeline stage plugin that rewrites a field to the value it already held is treated as no change.** Previously any returned payload marked the field replaced. (#151) +- **`payload_modified` errs toward reporting a change.** It records that the executor *accepted* an edit, not that the bytes differ, so it trips on a plugin returning an untouched clone and on a field pipeline writing a field the value it already held. Both previously reported no change. Operators sizing this: inside the engine it costs a `Value` clone and a `Message` clone with no serialization, but a host that keys its wire re-encode off "did the payload change?" will now re-encode on somewhat more routes, and for FFI hosts the MessagePack step rides along. The direction is deliberate. A false positive costs one redundant re-encode; the false negative was the vulnerability fixed in this release, so a modest throughput shift on mutating routes is expected rather than a regression. (#151) ### Deprecated - The reserved `all` group and the `global.policies:` bundle location, in favor of the top-level `groups:` section. Both still load. (#131) +### Fixed + +- **Plugin payload mutations are no longer silently discarded.** A plugin that rewrote anything other than a message's text — a tool result, a tool call's arguments, a thinking block, an attachment — had its mutation dropped by the APL route handler, which decided "was this modified?" by comparing concatenated text content. Redaction and sanitisation plugins are precisely the ones that rewrite tool results, so the failure was fail-open on the path that matters most: the plugin reported a successful redaction and the host forwarded the original secret. Mutation is now reported by the executor at the point it accepts a plugin's payload (`PipelineResult.payload_modified`) and read from there, so it no longer depends on which part of the message changed. Plugins that appended a throwaway text part to force the old check to fire can drop that workaround. (#151) +- **A field pipeline no longer clobbers a plugin's edit to the same content part.** Folding an `args:` or `result:` pipeline's rewrite back into the message replaced the whole argument map / result content, discarding edits a plugin had made to other fields of it. Only the paths the pipeline actually changed are applied now, so a pipeline redacting one argument and a plugin scrubbing another both survive. (#151) +- **A plugin invoked as a pipeline stage now reports a value for the field it was pointed at.** It previously reported the message's concatenated text as the field's new value, which for a structured tool call meant an unrelated argument was overwritten with chat text. A plugin that rewrote some other part of the payload now reports no field change, and its mutation travels with the payload instead. The reported value is compared against the field as the payload held it before the plugin ran, so a `plugin(...)` stage that leaves the field alone can no longer undo an earlier `mask` / `redact` / `hash` stage in the same chain — those interim edits live only in the pipeline, never in the payload, and comparing against them handed the pre-redaction value back as if the plugin had produced it. (#151) +- **A field pipeline that conflicts with a plugin on the same path now logs the tie-break.** `args:` / `result:` pipeline edits take precedence over a plugin's edit to the same field (config-author-wins), a key the plugin removed comes back if the pipeline rewrote it, and a key the pipeline omitted goes even if the plugin had rewritten it. Every case warns with the field name instead of resolving silently. (#151) + ## [0.2.2] - 2026-07-15 ### Added diff --git a/bindings/python/python/cpex/_lib.pyi b/bindings/python/python/cpex/_lib.pyi index 6d93c0f3..987fe0eb 100644 --- a/bindings/python/python/cpex/_lib.pyi +++ b/bindings/python/python/cpex/_lib.pyi @@ -17,6 +17,8 @@ class PipelineResult: @property def modified_payload(self) -> Optional[dict]: ... @property + def payload_modified(self) -> bool: ... + @property def modified_extensions(self) -> Optional[dict]: ... @property def violation(self) -> Optional[dict]: ... diff --git a/bindings/python/src/result.rs b/bindings/python/src/result.rs index 07c6d16a..8d4cdc73 100644 --- a/bindings/python/src/result.rs +++ b/bindings/python/src/result.rs @@ -27,6 +27,7 @@ use crate::conversions::{json_value_to_pyobj, serialize_payload}; pub struct PyPipelineResult { pub continue_processing: bool, pub modified_payload: Option, + pub payload_modified: bool, pub modified_extensions: Option, pub violation: Option, pub errors: Vec, @@ -56,6 +57,17 @@ impl PyPipelineResult { } } + /// Whether a plugin's payload modification was accepted. + /// + /// `modified_payload` is set on every allowed pipeline, carrying the + /// final payload whether or not a plugin touched it, so it never + /// answered "did anything change?". Read this instead — comparing + /// payload contents cannot see mutations to non-text content parts. + #[getter] + fn payload_modified(&self) -> bool { + self.payload_modified + } + #[getter] fn modified_extensions<'py>(&self, py: Python<'py>) -> PyResult>> { match &self.modified_extensions { @@ -198,6 +210,7 @@ pub fn pipeline_result_to_py(mut result: PipelineResult) -> PyResult *args = true, @@ -1440,6 +1445,13 @@ pub async fn evaluate_pipeline( }); }, Stage::Plugin { name } => { + // Known limitation: `current` carries the edits earlier + // stages made, but the invoker's payload does not — a + // `mask` before this stage is visible here and invisible + // to the plugin, which reads the field from the payload. + // Pushing interim pipeline state into the payload would + // change what every later plugin in the request sees, so + // it's left alone deliberately. let invocation = PluginInvocation::Field { name: field_name, value: ¤t, @@ -3668,6 +3680,74 @@ mod tests { ); } + /// A plugin stage learns which field it's operating on from the + /// invocation's `name`. That name is relative to the args / result + /// root at every call site, so an invoker can look the field up in + /// its own payload projection without having to strip a prefix — + /// which it couldn't do safely anyway, since `args` is a legal + /// argument name. + #[tokio::test] + async fn field_op_reports_a_root_relative_field_name_to_plugins() { + /// Captures the field name the pipeline passed down. + struct NameRecorder { + seen: std::sync::Mutex>, + } + #[async_trait] + impl PluginInvoker for NameRecorder { + async fn invoke( + &self, + _name: &str, + _bag: &AttributeBag, + invocation: PluginInvocation<'_>, + ) -> Result { + if let PluginInvocation::Field { name, .. } = invocation { + self.seen.lock().unwrap().push(name.to_string()); + } + Ok(PluginOutcome::allow()) + } + } + + let recorder = Arc::new(NameRecorder { + seen: std::sync::Mutex::new(Vec::new()), + }); + let plugins: Arc = recorder.clone(); + + let rule = Rule { + condition: Expression::Always, + effects: vec![Effect::FieldOp { + path: "args.user.ssn".into(), + stages: vec![Stage::Plugin { + name: "scrubber".into(), + }], + }], + source: "demo.policy[0]".into(), + }; + let mut bag = AttributeBag::new(); + let mut payload = crate::route::RoutePayload::new(json!({ + "user": {"ssn": "123-45-6789"}, + })); + + let _ = evaluate_effects( + &[Effect::from(rule)], + &mut bag, + &(Arc::new(FakePdp { + decision: Decision::Allow, + }) as Arc), + &plugins, + &noop_delegations(), + &noop_elicitations(), + crate::step::DispatchPhase::Pre, + &mut payload, + ) + .await; + + assert_eq!( + recorder.seen.lock().unwrap().as_slice(), + ["user.ssn".to_string()], + "the `args.` prefix belongs to the config path, not the field name" + ); + } + #[tokio::test] async fn field_op_targeting_result_in_pre_phase_is_skipped() { // A `result.X | ...` op encountered during the Pre phase is a diff --git a/crates/apl-core/src/lib.rs b/crates/apl-core/src/lib.rs index 5844d59f..e817dc26 100644 --- a/crates/apl-core/src/lib.rs +++ b/crates/apl-core/src/lib.rs @@ -35,7 +35,9 @@ pub use pipeline::{FieldRule, Pipeline, ScanKind, Stage, TaintEvent, TaintScope, pub use plugin_decl::{ CapsView, EffectivePlugin, PluginDeclaration, PluginOverride, PluginRegistry, }; -pub use route::{evaluate_post, evaluate_pre, evaluate_route, RouteDecision, RoutePayload}; +pub use route::{ + evaluate_post, evaluate_pre, evaluate_route, get_dotted, RouteDecision, RoutePayload, +}; pub use rules::{ CompareOp, CompiledRoute, Condition, DenyResponse, Effect, Expression, Literal, Phase, PhaseSet, Rule, diff --git a/crates/apl-core/src/route.rs b/crates/apl-core/src/route.rs index 9159c6e9..83f07364 100644 --- a/crates/apl-core/src/route.rs +++ b/crates/apl-core/src/route.rs @@ -317,10 +317,12 @@ pub async fn evaluate_route( /// Read `root.a.b.c` from a JSON value via dot-separated path. Returns /// `None` if any segment is missing or the path crosses a non-object. -pub(crate) fn get_dotted<'a>( - root: &'a serde_json::Value, - path: &str, -) -> Option<&'a serde_json::Value> { +/// +/// Public because host bridges read fields back out of their own payload +/// projections — a plugin dispatched from a pipeline stage reports a new +/// value for the field it was pointed at, and finding that field has to +/// use the same path semantics the evaluator used to write it. +pub fn get_dotted<'a>(root: &'a serde_json::Value, path: &str) -> Option<&'a serde_json::Value> { let mut cur = root; for seg in path.split('.') { cur = cur.get(seg)?; diff --git a/crates/apl-core/src/step.rs b/crates/apl-core/src/step.rs index 6f6e65c8..c4351b8a 100644 --- a/crates/apl-core/src/step.rs +++ b/crates/apl-core/src/step.rs @@ -425,6 +425,11 @@ pub enum PluginInvocation<'a> { Step { phase: DispatchPhase }, /// Called inside an `args:` / `result:` pipe chain on one field. Field { + /// Dotted path to the field, relative to the args or result root + /// — `city`, `user.ssn`, never `args.city`. The phase says which + /// root it hangs off: Pre addresses args, Post addresses result. + /// Every call site uses this convention, so an invoker can read + /// the field back out of a payload without guessing. name: &'a str, value: &'a serde_json::Value, phase: DispatchPhase, @@ -845,6 +850,11 @@ pub struct PluginOutcome { /// args/result chain, it may rewrite the field value (e.g., a PII /// scrubber producing a redacted string). `None` means "leave value /// unchanged"; always `None` for policy / post_policy invocations. + /// + /// Scoped to the field named in [`PluginInvocation::Field`] and + /// nothing else. A plugin that rewrote some other part of the + /// payload reports `None` here — that mutation travels with the + /// payload instead, so it isn't lost. pub modified_value: Option, } diff --git a/crates/apl-cpex/src/cmf_invoker.rs b/crates/apl-cpex/src/cmf_invoker.rs index 568f3b16..b936eb3b 100644 --- a/crates/apl-cpex/src/cmf_invoker.rs +++ b/crates/apl-cpex/src/cmf_invoker.rs @@ -23,6 +23,13 @@ // [`persist_session`] after route evaluation. Session ID is pulled from // `extensions.agent.session_id`; absent → both ops are no-ops. // +// Alongside the payload, the invoker records *whether* a plugin ever +// handed back a payload ([`payload_was_modified`]). That flag is the +// authoritative answer for the host: a plugin mutation is only +// detectable at the moment it's accepted, not by comparing message +// content afterwards (content comparison can't see mutations to +// non-text parts, and equality isn't defined on the CMF payload types). +// // # Per-call taint extraction // // Each plugin invocation diffs `result.modified_extensions.security.labels` @@ -49,6 +56,7 @@ // invoker. use std::collections::HashSet; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use async_trait::async_trait; @@ -83,6 +91,16 @@ pub struct CmfPluginInvoker { /// `extensions` — accumulated text rewrites have to be visible to /// the next dispatch in the same request. payload: Arc>, + /// Set the moment a plugin's `modified_payload` is accepted into + /// `payload` above. Request-scoped and sticky: once any plugin in + /// the request mutates, it stays `true`. + /// + /// This is the *signal* the host reads to decide whether to forward + /// a modified payload. It exists because the fact is only knowable + /// here — a caller comparing message content afterwards sees text + /// parts only, so a redaction of a `ToolResult` (or any other + /// non-text part) looks identical to no mutation at all. + payload_modified: AtomicBool, /// Pre-resolved per-route plugin lineup. Built (or fetched from a /// shared `DispatchCache`) at request start by the host. plan: Arc, @@ -144,6 +162,7 @@ impl CmfPluginInvoker { manager, extensions: Arc::new(Mutex::new(extensions)), payload: Arc::new(Mutex::new(payload)), + payload_modified: AtomicBool::new(false), plan, session_id, session_store, @@ -158,6 +177,26 @@ impl CmfPluginInvoker { self.payload.lock().await.clone() } + /// Did any plugin in this request hand back a payload? + /// + /// `true` from the moment a `modified_payload` is accepted into the + /// request's payload, and never resets. The host uses this to decide + /// whether to forward [`current_payload`] downstream. Reported + /// independently of *what* changed: a plugin that rewrites a tool + /// result, a tool call's arguments, or a thinking block is as + /// visible here as one that rewrites text. + /// + /// Deliberately `false` when a plugin returned a payload of the + /// wrong concrete type — that mutation was dropped (with a warning), + /// so claiming it landed would forward an unmutated payload while + /// asserting it changed. + pub fn payload_was_modified(&self) -> bool { + // Pairs with the `Release` store in `invoke`: plugin branches + // can run on other tasks (`dispatch_parallel`), so the write + // has to be visible to this read. + self.payload_modified.load(Ordering::Acquire) + } + /// Snapshot the current extensions. Useful for hosts that need to /// inspect the post-evaluation extension state (audit, telemetry). pub async fn current_extensions(&self) -> Extensions { @@ -300,6 +339,23 @@ impl PluginInvoker for CmfPluginInvoker { // these become `PluginOutcome.taints`. let before_labels = snapshot_labels(¤t_extensions); + // Per-call field baseline for pipeline-stage dispatch: the field + // as *this payload* holds it right now. + // + // This is the only sound thing to compare a readback against. The + // pipeline's own `value` may already carry earlier stages' edits + // (`mask`, `redact`, `hash`) that were never pushed into the + // payload, so comparing against it would read the payload's + // untouched original as "the plugin's new value" and hand the + // pre-redaction plaintext back to the pipeline, undoing the + // earlier stage. + let field_before = match invocation { + PluginInvocation::Field { name, phase, .. } => { + field_value_from_message(¤t_payload.message, name, phase) + }, + PluginInvocation::Step { .. } => None, + }; + let (result, _bg) = self .manager .invoke_entries::( @@ -329,18 +385,43 @@ impl PluginInvoker for CmfPluginInvoker { // request payload. `PluginPayload` only exposes `as_any`, so we // downcast-ref and clone. `MessagePayload: Clone` makes this // cheap relative to the FFI/invoke cost. - let modified_value = if let Some(mp_boxed) = result.modified_payload.as_ref() { + // + // Gated on `payload_modified`, not on `modified_payload.is_some()`: + // the executor returns the final payload on every allowed + // pipeline, so `is_some()` is true even when the plugin never + // touched it. + let modified_value = if !result.payload_modified { + None + } else if let Some(mp_boxed) = result.modified_payload.as_ref() { match mp_boxed.as_any().downcast_ref::() { Some(modified) => { *self.payload.lock().await = modified.clone(); + // Record the mutation for the host. `Release` so the + // flag is visible to `payload_was_modified` even when + // this call ran on a `dispatch_parallel` branch task. + self.payload_modified.store(true, Ordering::Release); match invocation { - PluginInvocation::Field { .. } => Some(serde_json::Value::String( - modified.message.get_text_content(), - )), + PluginInvocation::Field { name, phase, .. } => { + let rewritten = + field_value_from_message(&modified.message, name, phase) + .filter(|new_value| field_before.as_ref() != Some(new_value)); + if rewritten.is_none() { + tracing::debug!( + plugin = %plugin_name, + field = %name, + "plugin mutated the payload but not this field; \ + leaving the field value alone" + ); + } + rewritten + }, PluginInvocation::Step { .. } => None, } }, None => { + // Left out of `payload_modified` on purpose: nothing + // was written, so the host must keep forwarding the + // payload it already has. tracing::warn!( plugin = %plugin_name, "CmfPluginInvoker: modified_payload was not MessagePayload \ @@ -386,6 +467,40 @@ impl PluginInvoker for CmfPluginInvoker { } } +/// Read the value of one pipeline field out of a message. +/// +/// A plugin dispatched from an `args:` / `result:` stage is handed the +/// whole message, not the field, so its new value for that field has to +/// be read back out. The projection matches what APL evaluated against: +/// Pre addresses args, Post addresses result. `field` is relative to +/// that root. +/// +/// Two shapes: +/// * object projection (a tool call's arguments, a structured tool +/// result) → look up `field` in it, `None` when absent. +/// * scalar projection (a text-only message, whose whole content is +/// the field) → the projection itself. +/// +/// The caller compares the result against the value the pipeline is +/// holding: equal, or `None` here, both mean "this plugin didn't change +/// this field". The plugin's payload mutation is recorded separately, so +/// reporting no field change never drops it. +fn field_value_from_message( + message: &cpex_core::cmf::Message, + field: &str, + phase: DispatchPhase, +) -> Option { + let projection = match phase { + DispatchPhase::Pre => crate::message_projection::extract_args_from_message(message), + DispatchPhase::Post => crate::message_projection::extract_result_from_message(message), + }; + if projection.is_object() { + apl_core::get_dotted(&projection, field).cloned() + } else { + Some(projection) + } +} + /// Snapshot `extensions.security.labels` as an owned `HashSet`. /// Empty when security is absent. fn snapshot_labels(extensions: &Extensions) -> HashSet { diff --git a/crates/apl-cpex/src/lib.rs b/crates/apl-cpex/src/lib.rs index 3cba5f32..2cb99923 100644 --- a/crates/apl-cpex/src/lib.rs +++ b/crates/apl-cpex/src/lib.rs @@ -37,6 +37,7 @@ pub mod cmf_invoker; pub mod delegation_invoker; pub mod dispatch_plan; pub mod elicitation_invoker; +mod message_projection; pub mod parallel_safety; pub mod pdp_router; pub mod register; diff --git a/crates/apl-cpex/src/message_projection.rs b/crates/apl-cpex/src/message_projection.rs new file mode 100644 index 00000000..ea567eec --- /dev/null +++ b/crates/apl-cpex/src/message_projection.rs @@ -0,0 +1,463 @@ +// Location: ./crates/apl-cpex/src/message_projection.rs +// Copyright 2026 +// SPDX-License-Identifier: Apache-2.0 +// Authors: Teryl Taylor, Fred Araujo +// +// Projections between a CMF `Message` and the flat JSON APL evaluates +// against (`RoutePayload.args` / `RoutePayload.result`), plus their +// inverses. +// +// APL reasons about `args.` and `result.`; CMF carries +// typed content parts. These functions are the only translation between +// the two, and both consumers depend on them agreeing: +// +// * `AplRouteHandler` projects before evaluation and writes back +// after, so pipeline edits reach the host's body re-serializer. +// * `CmfPluginInvoker` projects a plugin-mutated message to read back +// the field a pipeline stage was focused on. +// +// Phase decides the side: Pre projects args, Post projects result. Each +// `write_*` is the inverse of the matching `extract_*`, so +// extract → write round-trips a message unchanged. +// +// The projections are lossy by design: they surface the one part APL +// addresses (a tool call's arguments, a tool result's content) and +// ignore the rest. That makes them unfit for answering "did anything +// change?" about a whole message — a mutation to any part they don't +// read is invisible. Callers needing that answer read the mutation +// signal the executor reports instead. + +use serde_json::Value; + +use cpex_core::cmf::{ContentPart, Message}; + +/// Rewrite the first text part of `msg` with `new_text`. If there is no +/// text part, append one. Mirrors what `MessagePayload`'s normal +/// modify-path does for single-view v0. +pub(crate) fn rewrite_message_text(msg: &mut Message, new_text: &str) { + for part in msg.content.iter_mut() { + if let ContentPart::Text { text } = part { + *text = new_text.to_string(); + return; + } + } + msg.content.push(ContentPart::Text { + text: new_text.to_string(), + }); +} + +/// Extract `RoutePayload.args` from a CMF message. v0 maps: +/// * First `ContentPart::ToolCall` → `arguments` map (Object) +/// * First `ContentPart::PromptRequest` → `arguments` map (Object) +/// * Else (text / no entity parts) → JSON String of text content +/// +/// `args.` APL paths target tool / prompt arguments directly. +/// For text-only messages we fall back to the v0 "args = whole text" +/// shape so `args.text` predicates keep working. +pub(crate) fn extract_args_from_message(msg: &Message) -> Value { + for part in &msg.content { + match part { + ContentPart::ToolCall { content } => { + return Value::Object( + content + .arguments + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + ); + }, + ContentPart::PromptRequest { content } => { + return Value::Object( + content + .arguments + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + ); + }, + _ => {}, + } + } + Value::String(msg.get_text_content()) +} + +/// Inverse of [`extract_args_from_message`]: write `args` back into +/// `msg`'s first ToolCall / PromptRequest argument map, or — for +/// text payloads — into the first text part. +/// +/// Silently no-ops when the args shape doesn't match the message +/// content shape (e.g. operator pipeline produced a String for what +/// was originally a ToolCall). The mismatch path is recoverable — +/// the upstream just sees the original unmodified content rather +/// than a malformed rewrite. +pub(crate) fn write_args_back_to_message(msg: &mut Message, args: &Value) { + for part in msg.content.iter_mut() { + match part { + ContentPart::ToolCall { content } => { + if let Some(obj) = args.as_object() { + content.arguments = obj.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); + } + return; + }, + ContentPart::PromptRequest { content } => { + if let Some(obj) = args.as_object() { + content.arguments = obj.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); + } + return; + }, + _ => {}, + } + } + // Fall through: no structured entity part — treat as text. + if let Some(text) = args.as_str() { + rewrite_message_text(msg, text); + } +} + +/// Extract `RoutePayload.result` from a CMF message. Mirror of +/// [`extract_args_from_message`] for the Post phase. v0 maps: +/// * First `ContentPart::ToolResult` → its `content` JSON value +/// * Else (text / no structured result part) → JSON String of text +/// +/// `result.` APL paths target the structured result directly. +pub(crate) fn extract_result_from_message(msg: &Message) -> Value { + for part in &msg.content { + if let ContentPart::ToolResult { content } = part { + return content.content.clone(); + } + } + Value::String(msg.get_text_content()) +} + +/// Inverse of [`extract_result_from_message`]: write a mutated +/// `result` back into the message's first `ContentPart::ToolResult.content`, +/// or — for text-only messages — into the first text part. The praxis +/// filter's response-body re-serializer then lifts the new content +/// out of the ContentPart and folds it back into the JSON-RPC +/// `result.content[*].text` payload. +pub(crate) fn write_result_back_to_message(msg: &mut Message, result: &Value) { + for part in msg.content.iter_mut() { + if let ContentPart::ToolResult { content } = part { + content.content = result.clone(); + return; + } + } + if let Some(text) = result.as_str() { + rewrite_message_text(msg, text); + } +} + +/// Apply to `base` only what changed between `pre` and `post`. +/// +/// `pre` and `post` bracket one editor's work (an APL pipeline: the +/// projection it started from, and the projection it produced). `base` +/// is the same projection taken from a payload a *different* editor (a +/// plugin) has since rewritten. Copying `post` over `base` wholesale +/// would discard the plugin's edits to keys the pipeline never touched, +/// so instead each differing leaf, added key, and removed key is applied +/// individually. +/// +/// When nothing else edited the payload, `base` equals `pre` and the +/// result is exactly `post`. +/// +/// Objects merge key by key; arrays and scalars are single values, so a +/// change to one replaces it whole. That matches how APL writes fields: +/// its dotted paths only traverse objects. +/// +/// # Precedence when both editors touched the same path +/// +/// **The pipeline wins.** If a plugin rewrote a key and the pipeline +/// rewrote it too, the pipeline's value lands; if the plugin *removed* a +/// key the pipeline then rewrote, the key comes back with the pipeline's +/// value; if the pipeline *omitted* a key the plugin had rewritten, the +/// key goes and the plugin's edit with it. Every case logs a warning +/// naming the key, because the losing edit is usually also a redaction +/// and a silent tie-break in this path is exactly the class of bug this +/// function exists to prevent. +/// +/// The rule is config-author-wins: an `args:` / `result:` pipeline is +/// written by the operator deploying the policy, so it outranks a +/// plugin's own view when the two genuinely conflict. Neither ordering +/// leaks plaintext (both editors write redacted values), but a coarse +/// pipeline stage can beat a finer plugin redaction, which is why the +/// conflict is logged rather than resolved silently. +pub(crate) fn apply_changed_paths(base: &mut Value, pre: &Value, post: &Value) { + let (Some(pre_map), Some(post_map)) = (pre.as_object(), post.as_object()) else { + // Not a keyed shape at this level, so there's nothing to merge + // per-key: the value either changed or it didn't. + if pre != post { + *base = post.clone(); + } + return; + }; + let Some(base_map) = base.as_object_mut() else { + // The other editor replaced the keyed shape with something else + // entirely. There's no key to merge into, so take this editor's + // view rather than invent a reconciliation. + tracing::warn!( + "payload projection changed shape under a field pipeline; \ + applying the pipeline's view and discarding the other edit" + ); + *base = post.clone(); + return; + }; + + for (key, pre_value) in pre_map { + if !post_map.contains_key(key) { + // Silent when the plugin left the key alone, which is the + // common case for an `omit`. + if base_map + .get(key) + .is_some_and(|base_value| base_value != pre_value) + { + tracing::warn!( + field = %key, + "pipeline omitted this field, discarding a plugin's edit to it" + ); + } + base_map.remove(key); + } + } + + for (key, post_value) in post_map { + match pre_map.get(key) { + // Untouched by this editor — leave whatever `base` holds, + // which is the whole point. + Some(pre_value) if pre_value == post_value => {}, + Some(pre_value) => { + let nested = base_map + .get_mut(key) + .filter(|base_value| base_value.is_object()) + .filter(|_| pre_value.is_object() && post_value.is_object()); + match nested { + Some(base_value) => apply_changed_paths(base_value, pre_value, post_value), + None => { + warn_on_conflict(base_map.get(key), pre_value, key); + base_map.insert(key.clone(), post_value.clone()); + }, + } + }, + None => { + // Both editors added the same key. Same precedence, same + // reason to say so out loud. + if base_map.contains_key(key) { + tracing::warn!( + field = %key, + "pipeline and plugin both added this field; keeping the pipeline's value" + ); + } + base_map.insert(key.clone(), post_value.clone()); + }, + } + } +} + +/// Log the two ways a pipeline edit can override a plugin's work on the +/// same key: the plugin changed it too, or the plugin removed it. Silent +/// in the common case where the plugin left the key alone. +fn warn_on_conflict(base_value: Option<&Value>, pre_value: &Value, key: &str) { + match base_value { + Some(base_value) if base_value != pre_value => tracing::warn!( + field = %key, + "pipeline edit overrides a plugin's edit to the same field; \ + keeping the pipeline's value" + ), + Some(_) => {}, + None => tracing::warn!( + field = %key, + "pipeline edit reinstates a field the plugin removed" + ), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use cpex_core::cmf::enums::Role; + use cpex_core::cmf::{ToolCall, ToolResult}; + + fn tool_call_message() -> Message { + Message::with_content( + Role::User, + vec![ContentPart::ToolCall { + content: ToolCall { + tool_call_id: "tc_001".to_string(), + name: "get_weather".to_string(), + arguments: [ + ("city".to_string(), serde_json::json!("London")), + ("units".to_string(), serde_json::json!("metric")), + ] + .into_iter() + .collect(), + namespace: None, + }, + }], + ) + } + + fn tool_result_message() -> Message { + Message::with_content( + Role::Tool, + vec![ContentPart::ToolResult { + content: ToolResult { + tool_call_id: "tc_001".to_string(), + tool_name: "get_weather".to_string(), + content: serde_json::json!({"temp": 12, "sky": "grey"}), + is_error: false, + }, + }], + ) + } + + #[test] + fn args_round_trip_leaves_a_tool_call_unchanged() { + let mut msg = tool_call_message(); + let args = extract_args_from_message(&msg); + write_args_back_to_message(&mut msg, &args); + assert_eq!(extract_args_from_message(&msg), args); + } + + #[test] + fn result_round_trip_leaves_a_tool_result_unchanged() { + let mut msg = tool_result_message(); + let result = extract_result_from_message(&msg); + write_result_back_to_message(&mut msg, &result); + assert_eq!(extract_result_from_message(&msg), result); + } + + #[test] + fn text_message_projects_to_its_whole_text_both_ways() { + let mut msg = Message::text(Role::User, "hello"); + assert_eq!( + extract_args_from_message(&msg), + serde_json::json!("hello"), + "a text-only message has no structured args, so args are the text" + ); + write_args_back_to_message(&mut msg, &serde_json::json!("goodbye")); + assert_eq!(msg.get_text_content(), "goodbye"); + } + + #[test] + fn changed_paths_apply_over_an_untouched_base() { + // Nobody else edited the payload, so base == pre and applying + // the changes must land exactly on post. + let pre = serde_json::json!({"city": "London", "units": "metric"}); + let post = serde_json::json!({"city": "[REDACTED]", "units": "metric"}); + let mut base = pre.clone(); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!(base, post); + } + + #[test] + fn changed_paths_preserve_another_editors_keys() { + let pre = serde_json::json!({"city": "London", "token": "sk-secret"}); + // The pipeline rewrote `city` only. + let post = serde_json::json!({"city": "[REDACTED]", "token": "sk-secret"}); + // Meanwhile a plugin rewrote `token`. + let mut base = serde_json::json!({"city": "London", "token": "[SCRUBBED]"}); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!( + base, + serde_json::json!({"city": "[REDACTED]", "token": "[SCRUBBED]"}), + "both editors' work must survive" + ); + } + + #[test] + fn changed_paths_apply_removals() { + let pre = serde_json::json!({"city": "London", "debug": true}); + let post = serde_json::json!({"city": "London"}); + let mut base = serde_json::json!({"city": "Paris", "debug": true}); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!(base, serde_json::json!({"city": "Paris"})); + } + + /// The removal counterpart of the same-key conflict: a pipeline `omit` + /// of a key the plugin had just rewritten still drops the key, and the + /// plugin's edit with it. Safe in either ordering (the key is gone + /// whoever wins), so this pins the outcome rather than a leak. + #[test] + fn changed_paths_omit_a_key_the_plugin_rewrote() { + let pre = serde_json::json!({"city": "London", "ssn": "123-45-6789"}); + let post = serde_json::json!({"city": "London"}); + let mut base = serde_json::json!({"city": "London", "ssn": "[REDACTED]"}); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!( + base, + serde_json::json!({"city": "London"}), + "an omitted key must not survive because a plugin rewrote it" + ); + } + + #[test] + fn changed_paths_recurse_into_nested_objects() { + let pre = serde_json::json!({"user": {"name": "ada", "ssn": "123-45-6789"}}); + let post = serde_json::json!({"user": {"name": "ada", "ssn": "[REDACTED]"}}); + let mut base = serde_json::json!({"user": {"name": "ADA", "ssn": "123-45-6789"}}); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!( + base, + serde_json::json!({"user": {"name": "ADA", "ssn": "[REDACTED]"}}), + "a sibling edit inside the same object must not be clobbered" + ); + } + + /// Documented precedence: when both editors rewrote the same key, the + /// pipeline's value lands. Pinned so a future change to the rule is + /// deliberate rather than incidental. + #[test] + fn changed_paths_let_the_pipeline_win_a_same_key_conflict() { + let pre = serde_json::json!({"ssn": "123-45-6789"}); + let post = serde_json::json!({"ssn": "***-**-6789"}); + // The plugin redacted the same key more aggressively. + let mut base = serde_json::json!({"ssn": "[REDACTED]"}); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!(base, serde_json::json!({"ssn": "***-**-6789"})); + } + + /// A key the plugin removed comes back when the pipeline rewrote it, + /// carrying the pipeline's (redacted) value. + #[test] + fn changed_paths_reinstate_a_key_the_plugin_removed() { + let pre = serde_json::json!({"ssn": "123-45-6789", "name": "Ada"}); + let post = serde_json::json!({"ssn": "[REDACTED]", "name": "Ada"}); + let mut base = serde_json::json!({"name": "Ada"}); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!( + base, + serde_json::json!({"ssn": "[REDACTED]", "name": "Ada"}), + "the pipeline's redaction lands even though the plugin dropped the key" + ); + } + + #[test] + fn changed_paths_replace_a_scalar_projection_whole() { + let mut base = serde_json::json!("plugin text"); + apply_changed_paths( + &mut base, + &serde_json::json!("original"), + &serde_json::json!("pipeline text"), + ); + assert_eq!(base, serde_json::json!("pipeline text")); + } + + #[test] + fn changed_paths_take_the_pipeline_view_when_base_shape_differs() { + let pre = serde_json::json!({"city": "London"}); + let post = serde_json::json!({"city": "[REDACTED]"}); + let mut base = serde_json::json!("no longer an object"); + apply_changed_paths(&mut base, &pre, &post); + assert_eq!(base, post); + } + + #[test] + fn shape_mismatch_leaves_a_tool_call_untouched() { + let mut msg = tool_call_message(); + let before = extract_args_from_message(&msg); + // A pipeline that produced a string where the message holds + // structured arguments: better to forward the original than a + // malformed rewrite. + write_args_back_to_message(&mut msg, &serde_json::json!("not an object")); + assert_eq!(extract_args_from_message(&msg), before); + } +} diff --git a/crates/apl-cpex/src/route_handler.rs b/crates/apl-cpex/src/route_handler.rs index 7af3bcb3..fa7085b8 100644 --- a/crates/apl-cpex/src/route_handler.rs +++ b/crates/apl-cpex/src/route_handler.rs @@ -55,6 +55,10 @@ use crate::cmf_invoker::CmfPluginInvoker; use crate::delegation_invoker::DelegationPluginInvoker; use crate::dispatch_plan::DispatchCache; use crate::elicitation_invoker::ElicitationPluginInvoker; +use crate::message_projection::{ + apply_changed_paths, extract_args_from_message, extract_result_from_message, + write_args_back_to_message, write_result_back_to_message, +}; use crate::pdp_router::PdpRouter; use crate::session_store::SessionStore; @@ -439,47 +443,90 @@ impl AnyHookHandler for AplRouteHandler { let final_payload = invoker.current_payload().await; let final_extensions = invoker.current_extensions().await; - // Detect whether the args pipeline mutated the payload by - // re-extracting from the pre-eval message (msg_payload is - // still borrowed) and comparing against the post-eval - // route_payload.args. Re-extraction allocates but mirrors the - // surrounding pattern and avoids holding a pre-eval clone. - let pre_args = extract_args_from_message(&msg_payload.message); - // For Post phase, also detect result mutations from `result:` - // pipelines. Pre routes don't carry a result so this is None. + // The pre-evaluation projections. No longer used to *detect* + // pipeline edits (the decision reports those) — they're the + // baseline for folding those edits back in below, which needs to + // know which paths the pipeline touched. + // + // Each side is projected only in the phase that can edit it: + // `evaluate_pre` never sets `result_modified` and `evaluate_post` + // never sets `args_modified`, so the other projection would be + // unread work on every request. + let pre_args = match self.phase { + Phase::Pre => Some(extract_args_from_message(&msg_payload.message)), + Phase::Post => None, + }; let pre_result = match self.phase { Phase::Pre => None, Phase::Post => Some(extract_result_from_message(&msg_payload.message)), }; - let modified_payload: Option> = if route_payload.args != pre_args { + // Which of the three sources changed the payload, in precedence + // order. Each condition is a signal from the code that performed + // the change: the decision's flags are set when a pipeline's + // `set_dotted` / `remove_dotted` actually writes, and the + // invoker's flag is set when a plugin's payload is accepted. + // Nothing here infers a change by comparing values. + let modified_payload: Option> = if decision.args_modified { // An args pipeline (Pre) rewrote a field. Fold the new // args back into a fresh MessagePayload so downstream // readers (the host's body re-serializer) see the // change. + // + // Only the paths the pipeline touched are applied. A plugin + // may have rewritten other arguments on the same tool call, + // and those edits aren't in `route_payload.args` (it was + // projected before any plugin ran), so writing it wholesale + // would silently drop them. + // + // Without a pre-projection there is no way to tell which + // paths the pipeline changed, so write nothing rather than + // fold in an unattributable diff — a wholesale write is + // exactly the clobbering this merge exists to prevent. Only + // the Pre phase sets `args_modified`, and only the Pre phase + // projects `pre_args`, so this holds by construction. let mut updated = final_payload.clone(); - write_args_back_to_message(&mut updated.message, &route_payload.args); + if let Some(pre) = pre_args.as_ref() { + let mut merged = extract_args_from_message(&updated.message); + apply_changed_paths(&mut merged, pre, &route_payload.args); + write_args_back_to_message(&mut updated.message, &merged); + } Some(Box::new(updated) as Box) - } else if matches!(self.phase, Phase::Post) - && pre_result - .as_ref() - .zip(route_payload.result.as_ref()) - .map(|(prev, current)| prev != current) - .unwrap_or(false) - { + } else if decision.result_modified { // A `result:` pipeline rewrote a field in the upstream // response. Fold the new result back into the message // so the host's response body re-serializer can write - // it out before forwarding downstream. + // it out before forwarding downstream. Only the Post phase + // can set this — a Pre route has no result to rewrite. + // + // Same per-path merge as the args branch above, for the same + // reason: a plugin may have redacted a different part of the + // same tool result. + // Same "no pre-projection, no write" rule as the args branch + // above, for the same reason. let mut updated = final_payload.clone(); - if let Some(result_value) = route_payload.result.as_ref() { - write_result_back_to_message(&mut updated.message, result_value); + if let (Some(result_value), Some(pre)) = + (route_payload.result.as_ref(), pre_result.as_ref()) + { + let mut merged = extract_result_from_message(&updated.message); + apply_changed_paths(&mut merged, pre, result_value); + write_result_back_to_message(&mut updated.message, &merged); } Some(Box::new(updated) as Box) - } else if msg_payload.message.get_text_content() != final_payload.message.get_text_content() - { - // A `pre_invocation:` plugin mutated the message directly via - // `modify_payload` (not through a field pipeline). Pass - // the invoker's view through unchanged. + } else if invoker.payload_was_modified() { + // A plugin mutated the message directly via `modify_payload` + // (not through a field pipeline). Pass the invoker's view + // through unchanged. + // + // The invoker records this when it accepts the mutation, + // which is the only point it can be known. Comparing message + // content here instead would read text parts only, so a + // redacted tool result, a rewritten tool call, or an edited + // thinking block would look identical to no mutation and get + // dropped. + tracing::debug!( + route = %self.route.route_key, + "plugin mutated the payload directly; forwarding the mutated view" + ); Some(Box::new(final_payload) as Box) } else { None @@ -661,126 +708,6 @@ fn decorate_denial_response(violation: &mut PluginViolation, response: Option<&D } } -/// Rewrite the first text part of `msg` with `new_text`. If there is no -/// text part, append one. Mirrors what `MessagePayload`'s normal -/// modify-path does for single-view v0. -fn rewrite_message_text(msg: &mut cpex_core::cmf::Message, new_text: &str) { - for part in msg.content.iter_mut() { - if let cpex_core::cmf::ContentPart::Text { text } = part { - *text = new_text.to_string(); - return; - } - } - msg.content.push(cpex_core::cmf::ContentPart::Text { - text: new_text.to_string(), - }); -} - -/// Extract `RoutePayload.args` from a CMF message. v0 maps: -/// * First `ContentPart::ToolCall` → `arguments` map (Object) -/// * First `ContentPart::PromptRequest` → `arguments` map (Object) -/// * Else (text / no entity parts) → JSON String of text content -/// -/// `args.` APL paths target tool / prompt arguments directly. -/// For text-only messages we fall back to the v0 "args = whole text" -/// shape so `args.text` predicates keep working. -fn extract_args_from_message(msg: &cpex_core::cmf::Message) -> Value { - use cpex_core::cmf::ContentPart; - for part in &msg.content { - match part { - ContentPart::ToolCall { content } => { - return Value::Object( - content - .arguments - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(), - ); - }, - ContentPart::PromptRequest { content } => { - return Value::Object( - content - .arguments - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(), - ); - }, - _ => {}, - } - } - Value::String(msg.get_text_content()) -} - -/// Inverse of [`extract_args_from_message`]: write `args` back into -/// `msg`'s first ToolCall / PromptRequest argument map, or — for -/// text payloads — into the first text part. -/// -/// Silently no-ops when the args shape doesn't match the message -/// content shape (e.g. operator pipeline produced a String for what -/// was originally a ToolCall). The mismatch path is recoverable — -/// the upstream just sees the original unmodified content rather -/// than a malformed rewrite. -fn write_args_back_to_message(msg: &mut cpex_core::cmf::Message, args: &Value) { - use cpex_core::cmf::ContentPart; - for part in msg.content.iter_mut() { - match part { - ContentPart::ToolCall { content } => { - if let Some(obj) = args.as_object() { - content.arguments = obj.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); - } - return; - }, - ContentPart::PromptRequest { content } => { - if let Some(obj) = args.as_object() { - content.arguments = obj.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); - } - return; - }, - _ => {}, - } - } - // Fall through: no structured entity part — treat as text. - if let Some(text) = args.as_str() { - rewrite_message_text(msg, text); - } -} - -/// Extract `RoutePayload.result` from a CMF message. Mirror of -/// [`extract_args_from_message`] for the Post phase. v0 maps: -/// * First `ContentPart::ToolResult` → its `content` JSON value -/// * Else (text / no structured result part) → JSON String of text -/// -/// `result.` APL paths target the structured result directly. -fn extract_result_from_message(msg: &cpex_core::cmf::Message) -> Value { - use cpex_core::cmf::ContentPart; - for part in &msg.content { - if let ContentPart::ToolResult { content } = part { - return content.content.clone(); - } - } - Value::String(msg.get_text_content()) -} - -/// Inverse of [`extract_result_from_message`]: write a mutated -/// `result` back into the message's first `ContentPart::ToolResult.content`, -/// or — for text-only messages — into the first text part. The praxis -/// filter's response-body re-serializer then lifts the new content -/// out of the ContentPart and folds it back into the JSON-RPC -/// `result.content[*].text` payload. -fn write_result_back_to_message(msg: &mut cpex_core::cmf::Message, result: &Value) { - use cpex_core::cmf::ContentPart; - for part in msg.content.iter_mut() { - if let ContentPart::ToolResult { content } = part { - content.content = result.clone(); - return; - } - } - if let Some(text) = result.as_str() { - rewrite_message_text(msg, text); - } -} - /// Cheap pointer-equality check across the few mutable extension slots /// the executor would care about. False positives (claiming a change /// when there isn't one) are cheap — the executor re-validates anyway. diff --git a/crates/apl-cpex/tests/cmf_invoker_dispatch.rs b/crates/apl-cpex/tests/cmf_invoker_dispatch.rs index 766b8ea4..512445da 100644 --- a/crates/apl-cpex/tests/cmf_invoker_dispatch.rs +++ b/crates/apl-cpex/tests/cmf_invoker_dispatch.rs @@ -197,6 +197,69 @@ impl PluginFactory for ModifyPluginFactory { } } +/// Redactor that rewrites **only** `ToolResult.content`, leaving every +/// Text part byte-identical. This is the shape of a real PII/secret +/// redactor, and the shape whose mutation used to vanish: nothing about +/// the message's text changes, so text-based change detection reports +/// "unmodified" and the unredacted result gets forwarded. +struct RedactToolResultPlugin { + cfg: PluginConfig, +} + +#[async_trait] +impl Plugin for RedactToolResultPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } +} + +impl HookHandler for RedactToolResultPlugin { + async fn handle( + &self, + payload: &MessagePayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let new_content: Vec = payload + .message + .content + .iter() + .map(|part| match part { + ContentPart::ToolResult { content } => { + let mut redacted = content.clone(); + redacted.content = serde_json::Value::String("[REDACTED]".to_string()); + ContentPart::ToolResult { content: redacted } + }, + other => other.clone(), + }) + .collect(); + PluginResult::modify_payload(MessagePayload { + message: Message { + schema_version: payload.message.schema_version.clone(), + role: payload.message.role, + content: new_content, + channel: payload.message.channel, + }, + }) + } +} + +struct RedactToolResultPluginFactory; +impl PluginFactory for RedactToolResultPluginFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = Arc::new(RedactToolResultPlugin { + cfg: config.clone(), + }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![( + "cmf.tool_pre_invoke", + Arc::new(TypedHandlerAdapter::::new(plugin)), + )], + }) + } +} + // --------------------------------------------------------------------- // Test helpers // --------------------------------------------------------------------- @@ -207,6 +270,38 @@ fn payload_with_text(text: &str) -> MessagePayload { } } +/// A message carrying a tool result alongside a text part. The text part +/// is what any text-based comparison would see; the secret lives in the +/// tool result, where only a structural reader finds it. +fn payload_with_tool_result(text: &str, result: &str) -> MessagePayload { + MessagePayload { + message: Message::with_content( + Role::Tool, + vec![ + ContentPart::Text { + text: text.to_string(), + }, + ContentPart::ToolResult { + content: cpex_core::cmf::ToolResult { + tool_call_id: "tc_001".to_string(), + tool_name: "get_secret".to_string(), + content: serde_json::Value::String(result.to_string()), + is_error: false, + }, + }, + ], + ), + } +} + +/// Read the first `ToolResult.content` out of a payload. +fn tool_result_content(payload: &MessagePayload) -> Option<&serde_json::Value> { + payload.message.content.iter().find_map(|part| match part { + ContentPart::ToolResult { content } => Some(&content.content), + _ => None, + }) +} + fn empty_bag() -> AttributeBag { AttributeBag::new() } @@ -384,6 +479,456 @@ async fn current_payload_reflects_accumulated_mutations() { assert_eq!(final_payload.message.get_text_content(), "hello [MODIFIED]"); } +// --------------------------------------------------------------------- +// Mutation signalling — the invoker reports whether a plugin handed back +// a payload, so the host never has to guess from message content. A +// guess based on text can't see a rewritten tool result, tool call, +// thinking block, image, or any other non-text part. +// +// Not covered here: a plugin returning a payload of the wrong concrete +// type (the downcast-failure path that warns and drops). `HookHandler` +// is typed on `PluginResult`, so a foreign payload can't +// be constructed through the typed dispatch path these tests use — it +// would take a hand-rolled `AnyHookHandler` bypassing the adapter. +// --------------------------------------------------------------------- + +#[tokio::test] +async fn no_mutation_reported_before_any_dispatch() { + let mgr = build_manager("allow-plugin", Box::new(AllowPluginFactory)).await; + let plan = plan_for(&mgr, "allow-plugin"); + let invoker = CmfPluginInvoker::for_request( + mgr, + Extensions::default(), + payload_with_text("hello"), + plan, + Arc::new(MemorySessionStore::new()), + ) + .await + .expect("for_request"); + + assert!( + !invoker.payload_was_modified(), + "a fresh invoker has dispatched nothing, so nothing can have mutated" + ); +} + +#[tokio::test] +async fn plugin_that_allows_without_mutating_reports_no_mutation() { + let mgr = build_manager("allow-plugin", Box::new(AllowPluginFactory)).await; + let plan = plan_for(&mgr, "allow-plugin"); + let invoker = CmfPluginInvoker::for_request( + mgr, + Extensions::default(), + payload_with_text("hello"), + plan, + Arc::new(MemorySessionStore::new()), + ) + .await + .expect("for_request"); + + let bag = empty_bag(); + let _ = invoker + .invoke( + "allow-plugin", + &bag, + PluginInvocation::Step { + phase: apl_core::step::DispatchPhase::Pre, + }, + ) + .await + .expect("invoke"); + + assert!( + !invoker.payload_was_modified(), + "a plain allow carries no payload; reporting a mutation here would \ + make every request look modified" + ); +} + +#[tokio::test] +async fn text_mutation_is_reported() { + let mgr = build_manager("modify-plugin", Box::new(ModifyPluginFactory)).await; + let plan = plan_for(&mgr, "modify-plugin"); + let invoker = CmfPluginInvoker::for_request( + mgr, + Extensions::default(), + payload_with_text("hello"), + plan, + Arc::new(MemorySessionStore::new()), + ) + .await + .expect("for_request"); + + let bag = empty_bag(); + let value = serde_json::Value::String("hello".to_string()); + let _ = invoker + .invoke( + "modify-plugin", + &bag, + PluginInvocation::Field { + name: "content", + value: &value, + phase: apl_core::step::DispatchPhase::Pre, + }, + ) + .await + .expect("invoke"); + + assert!(invoker.payload_was_modified()); +} + +/// The reported bug, at the invoker layer: a redactor rewrites only +/// `ToolResult.content`, so the message's text is byte-identical before +/// and after. The mutation must still be reported. +#[tokio::test] +async fn tool_result_only_mutation_is_reported() { + let mgr = build_manager("redact-plugin", Box::new(RedactToolResultPluginFactory)).await; + let plan = plan_for(&mgr, "redact-plugin"); + let original = payload_with_tool_result("here is the result", "sk-secret-value"); + let text_before = original.message.get_text_content(); + let invoker = CmfPluginInvoker::for_request( + mgr, + Extensions::default(), + original, + plan, + Arc::new(MemorySessionStore::new()), + ) + .await + .expect("for_request"); + + let bag = empty_bag(); + let _ = invoker + .invoke( + "redact-plugin", + &bag, + PluginInvocation::Step { + phase: apl_core::step::DispatchPhase::Pre, + }, + ) + .await + .expect("invoke"); + + let final_payload = invoker.current_payload().await; + assert_eq!( + tool_result_content(&final_payload), + Some(&serde_json::Value::String("[REDACTED]".to_string())), + "the redaction must land in the shared payload" + ); + assert_eq!( + final_payload.message.get_text_content(), + text_before, + "fixture sanity: the text is untouched, so text comparison sees no change" + ); + assert!( + invoker.payload_was_modified(), + "the mutation is invisible to text comparison but must still be reported" + ); +} + +// --------------------------------------------------------------------- +// Field-stage dispatch — a plugin invoked from an `args:` / `result:` +// pipeline is handed the whole message, so its new value for the field +// in focus has to be read back out of the part that field came from. +// Reporting the message's concatenated text instead would overwrite a +// structured argument with unrelated content. +// --------------------------------------------------------------------- + +/// Rewrites one named tool-call argument, leaving other arguments and +/// all text parts alone. +struct ArgRewritePlugin { + cfg: PluginConfig, + arg: &'static str, +} + +#[async_trait] +impl Plugin for ArgRewritePlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } +} + +impl HookHandler for ArgRewritePlugin { + async fn handle( + &self, + payload: &MessagePayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let content: Vec = payload + .message + .content + .iter() + .map(|part| match part { + ContentPart::ToolCall { content } => { + let mut next = content.clone(); + next.arguments.insert( + self.arg.to_string(), + serde_json::Value::String("[REDACTED]".to_string()), + ); + ContentPart::ToolCall { content: next } + }, + other => other.clone(), + }) + .collect(); + PluginResult::modify_payload(MessagePayload { + message: Message { + schema_version: payload.message.schema_version.clone(), + role: payload.message.role, + content, + channel: payload.message.channel, + }, + }) + } +} + +struct ArgRewriteFactory { + arg: &'static str, +} + +impl PluginFactory for ArgRewriteFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = Arc::new(ArgRewritePlugin { + cfg: config.clone(), + arg: self.arg, + }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![( + "cmf.field_redact", + Arc::new(TypedHandlerAdapter::::new(plugin)), + )], + }) + } +} + +/// Rewrites one field inside an object-shaped tool result. The Post-phase +/// counterpart to `ArgRewritePlugin`. +struct ResultFieldRewritePlugin { + cfg: PluginConfig, + field: &'static str, +} + +#[async_trait] +impl Plugin for ResultFieldRewritePlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } +} + +impl HookHandler for ResultFieldRewritePlugin { + async fn handle( + &self, + payload: &MessagePayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let content: Vec = payload + .message + .content + .iter() + .map(|part| match part { + ContentPart::ToolResult { content } => { + let mut next = content.clone(); + if let Some(obj) = next.content.as_object_mut() { + obj.insert( + self.field.to_string(), + serde_json::Value::String("[REDACTED]".to_string()), + ); + } + ContentPart::ToolResult { content: next } + }, + other => other.clone(), + }) + .collect(); + PluginResult::modify_payload(MessagePayload { + message: Message { + schema_version: payload.message.schema_version.clone(), + role: payload.message.role, + content, + channel: payload.message.channel, + }, + }) + } +} + +struct ResultFieldRewriteFactory { + field: &'static str, +} + +impl PluginFactory for ResultFieldRewriteFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = Arc::new(ResultFieldRewritePlugin { + cfg: config.clone(), + field: self.field, + }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![( + "cmf.field_redact", + Arc::new(TypedHandlerAdapter::::new(plugin)), + )], + }) + } +} + +fn payload_with_tool_call(city: &str, note: &str) -> MessagePayload { + MessagePayload { + message: Message::with_content( + Role::User, + vec![ + ContentPart::Text { + text: note.to_string(), + }, + ContentPart::ToolCall { + content: cpex_core::cmf::ToolCall { + tool_call_id: "tc_001".to_string(), + name: "get_weather".to_string(), + arguments: [("city".to_string(), serde_json::json!(city))] + .into_iter() + .collect(), + namespace: None, + }, + }, + ], + ), + } +} + +/// The field in focus is `city`, and the plugin rewrites `city`. The new +/// value must be the redacted city, not the message's text. +#[tokio::test] +async fn field_dispatch_reports_the_field_the_plugin_rewrote() { + let mgr = build_manager("arg-redactor", Box::new(ArgRewriteFactory { arg: "city" })).await; + let plan = plan_for(&mgr, "arg-redactor"); + let invoker = CmfPluginInvoker::for_request( + mgr, + Extensions::default(), + payload_with_tool_call("London", "unrelated chatter"), + plan, + Arc::new(MemorySessionStore::new()), + ) + .await + .expect("for_request"); + + let bag = empty_bag(); + let value = serde_json::json!("London"); + let outcome = invoker + .invoke( + "arg-redactor", + &bag, + PluginInvocation::Field { + name: "city", + value: &value, + phase: apl_core::step::DispatchPhase::Pre, + }, + ) + .await + .expect("invoke"); + + assert_eq!( + outcome.modified_value, + Some(serde_json::json!("[REDACTED]")) + ); +} + +/// The field in focus is `city`, but the plugin rewrote `token`. The +/// pipeline must be told the field is unchanged — and the payload +/// mutation must still be recorded, so the rewrite isn't lost. +#[tokio::test] +async fn field_dispatch_reports_no_change_when_another_field_was_rewritten() { + let mgr = build_manager("arg-redactor", Box::new(ArgRewriteFactory { arg: "token" })).await; + let plan = plan_for(&mgr, "arg-redactor"); + let invoker = CmfPluginInvoker::for_request( + mgr, + Extensions::default(), + payload_with_tool_call("London", "unrelated chatter"), + plan, + Arc::new(MemorySessionStore::new()), + ) + .await + .expect("for_request"); + + let bag = empty_bag(); + let value = serde_json::json!("London"); + let outcome = invoker + .invoke( + "arg-redactor", + &bag, + PluginInvocation::Field { + name: "city", + value: &value, + phase: apl_core::step::DispatchPhase::Pre, + }, + ) + .await + .expect("invoke"); + + assert_eq!( + outcome.modified_value, None, + "the field in focus is untouched, so the pipeline must leave it alone" + ); + assert!( + invoker.payload_was_modified(), + "the rewrite of another field still has to reach the host" + ); +} + +/// Post-phase dispatch reads the field out of the *result* projection, +/// not the args one. A `result:` pipeline stage that rewrites one field +/// of a structured tool result must get that field back. +#[tokio::test] +async fn post_phase_field_dispatch_reads_the_result_projection() { + let mgr = build_manager( + "result-redactor", + Box::new(ResultFieldRewriteFactory { field: "ssn" }), + ) + .await; + let plan = plan_for(&mgr, "result-redactor"); + let payload = MessagePayload { + message: Message::with_content( + Role::Tool, + vec![ContentPart::ToolResult { + content: cpex_core::cmf::ToolResult { + tool_call_id: "tc_001".to_string(), + tool_name: "get_employee".to_string(), + content: serde_json::json!({"name": "Ada", "ssn": "123-45-6789"}), + is_error: false, + }, + }], + ), + }; + let invoker = CmfPluginInvoker::for_request( + mgr, + Extensions::default(), + payload, + plan, + Arc::new(MemorySessionStore::new()), + ) + .await + .expect("for_request"); + + let bag = empty_bag(); + let value = serde_json::json!("123-45-6789"); + let outcome = invoker + .invoke( + "result-redactor", + &bag, + PluginInvocation::Field { + name: "ssn", + value: &value, + phase: apl_core::step::DispatchPhase::Post, + }, + ) + .await + .expect("invoke"); + + assert_eq!( + outcome.modified_value, + Some(serde_json::json!("[REDACTED]")), + "Post phase must read the field back out of the tool result, not the args" + ); +} + // --------------------------------------------------------------------- // Capability gating — APL route override of `capabilities:` materializes // a derived PluginRef wrapping the same plugin Arc with a merged diff --git a/crates/apl-cpex/tests/payload_mutation_propagation.rs b/crates/apl-cpex/tests/payload_mutation_propagation.rs new file mode 100644 index 00000000..97a7872d --- /dev/null +++ b/crates/apl-cpex/tests/payload_mutation_propagation.rs @@ -0,0 +1,1151 @@ +// Location: ./crates/apl-cpex/tests/payload_mutation_propagation.rs +// Copyright 2026 +// SPDX-License-Identifier: Apache-2.0 +// Authors: Fred Araujo +// +// A plugin mutation has to survive the whole way to the host, whichever +// part of the message it touched. These tests drive the real +// `AplRouteHandler` through `invoke_named::` and assert on the +// payload the host would forward. +// +// The failure these guard against is silent and fails open: a redactor +// rewrites `ToolResult.content`, reports success, and the host forwards +// the original secret anyway. Redaction and sanitisation plugins are +// exactly the ones that mutate non-text parts, so "only text mutations +// survive" is worst in the case that matters most. +// +// Text parts are left byte-identical on purpose in most fixtures below. +// Any check that infers "was this modified?" from message text passes +// them through unchanged, which is the bug. + +use std::sync::Arc; + +use async_trait::async_trait; +use cpex_core::cmf::enums::Role; +use cpex_core::cmf::{CmfHook, ContentPart, Message, MessagePayload, ToolCall, ToolResult}; +use cpex_core::context::PluginContext; +use cpex_core::error::PluginError as CoreError; +use cpex_core::factory::{PluginFactory, PluginInstance}; +use cpex_core::hooks::adapter::TypedHandlerAdapter; +use cpex_core::hooks::payload::Extensions; +use cpex_core::hooks::trait_def::{HookHandler, PluginResult}; +use cpex_core::manager::PluginManager; +use cpex_core::plugin::{Plugin, PluginConfig}; + +use apl_cpex::{register_apl, AplOptions, DispatchCache, MemorySessionStore}; + +// --------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------- + +/// Rewrites one content part and leaves the rest alone. Which part it +/// touches is chosen per instance so a single fixture covers several +/// `ContentPart` variants. +#[derive(Clone, Copy)] +enum Target { + /// Replace the whole tool result content. + ToolResultContent, + /// Replace one field inside an object-shaped tool result content, + /// leaving its siblings alone. + ToolResultField(&'static str), + /// Replace the `city` argument of a tool call. + ToolCallArguments, + /// Replace one named argument of a tool call. + ToolCallArgument(&'static str), + /// Replace a key inside an object-valued tool call argument. + NestedToolCallArgument(&'static str, &'static str), + Thinking, + Text, +} + +struct RewritePlugin { + cfg: PluginConfig, + target: Target, +} + +#[async_trait] +impl Plugin for RewritePlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } +} + +impl HookHandler for RewritePlugin { + async fn handle( + &self, + payload: &MessagePayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let content: Vec = payload + .message + .content + .iter() + .map(|part| match (self.target, part) { + (Target::ToolResultContent, ContentPart::ToolResult { content }) => { + let mut next = content.clone(); + next.content = serde_json::Value::String("[REDACTED]".to_string()); + ContentPart::ToolResult { content: next } + }, + (Target::ToolResultField(field), ContentPart::ToolResult { content }) => { + let mut next = content.clone(); + if let Some(obj) = next.content.as_object_mut() { + obj.insert( + field.to_string(), + serde_json::Value::String("[REDACTED]".to_string()), + ); + } + ContentPart::ToolResult { content: next } + }, + (Target::ToolCallArguments, ContentPart::ToolCall { content }) => { + let mut next = content.clone(); + next.arguments.insert( + "city".to_string(), + serde_json::Value::String("[REDACTED]".to_string()), + ); + ContentPart::ToolCall { content: next } + }, + (Target::ToolCallArgument(arg), ContentPart::ToolCall { content }) => { + let mut next = content.clone(); + next.arguments.insert( + arg.to_string(), + serde_json::Value::String("[REDACTED]".to_string()), + ); + ContentPart::ToolCall { content: next } + }, + (Target::NestedToolCallArgument(arg, key), ContentPart::ToolCall { content }) => { + let mut next = content.clone(); + if let Some(obj) = next.arguments.get_mut(arg).and_then(|v| v.as_object_mut()) { + obj.insert( + key.to_string(), + serde_json::Value::String("[REDACTED]".to_string()), + ); + } + ContentPart::ToolCall { content: next } + }, + (Target::Thinking, ContentPart::Thinking { .. }) => ContentPart::Thinking { + text: "[REDACTED]".to_string(), + }, + (Target::Text, ContentPart::Text { .. }) => ContentPart::Text { + text: "[REDACTED]".to_string(), + }, + (_, other) => other.clone(), + }) + .collect(); + + PluginResult::modify_payload(MessagePayload { + message: Message { + schema_version: payload.message.schema_version.clone(), + role: payload.message.role, + content, + channel: payload.message.channel, + }, + }) + } +} + +struct RewriteFactory { + target: Target, + hook: &'static str, +} + +impl PluginFactory for RewriteFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = Arc::new(RewritePlugin { + cfg: config.clone(), + target: self.target, + }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![( + self.hook, + Arc::new(TypedHandlerAdapter::::new(plugin)), + )], + }) + } +} + +/// Allows without returning a payload — the baseline for "nothing +/// changed, so nothing should be forwarded as modified". +struct NoopPlugin { + cfg: PluginConfig, +} + +#[async_trait] +impl Plugin for NoopPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } +} + +impl HookHandler for NoopPlugin { + async fn handle( + &self, + _payload: &MessagePayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + PluginResult::allow() + } +} + +/// Denies, so a route can mutate and then refuse in one request. +struct DenyPlugin { + cfg: PluginConfig, +} + +#[async_trait] +impl Plugin for DenyPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } +} + +impl HookHandler for DenyPlugin { + async fn handle( + &self, + _payload: &MessagePayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + PluginResult::deny(cpex_core::error::PluginViolation::new( + "policy.forbidden", + "test fixture denied this call", + )) + } +} + +struct DenyFactory; + +impl PluginFactory for DenyFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = Arc::new(DenyPlugin { + cfg: config.clone(), + }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![( + "cmf.tool_pre_invoke", + Arc::new(TypedHandlerAdapter::::new(plugin)), + )], + }) + } +} + +struct NoopFactory { + hook: &'static str, +} + +impl PluginFactory for NoopFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = Arc::new(NoopPlugin { + cfg: config.clone(), + }); + Ok(PluginInstance { + plugin: plugin.clone(), + handlers: vec![( + self.hook, + Arc::new(TypedHandlerAdapter::::new(plugin)), + )], + }) + } +} + +// --------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------- + +/// Wire one plugin behind an APL route on `get_weather`, with the route +/// phase and hook chosen by the caller. +async fn manager_with( + kind: &'static str, + factory: Box, + hook: &str, + phase: &str, +) -> Arc { + let mgr = Arc::new(PluginManager::default()); + mgr.register_factory(kind, factory); + register_apl( + &mgr, + AplOptions { + dispatch_cache: Arc::new(DispatchCache::new()), + session_store: Arc::new(MemorySessionStore::new()), + pdps: Vec::new(), + pdp_factories: Vec::new(), + session_store_factories: Vec::new(), + base_capabilities: None, + }, + ); + let yaml = format!( + r#" +plugins: + - name: {kind} + kind: {kind} + hooks: [{hook}] +routes: + - tool: get_weather + apl: + {phase}: + - "plugin({kind})" +"# + ); + mgr.load_config_yaml(&yaml).expect("load_config_yaml"); + mgr.initialize().await.expect("initialize"); + mgr +} + +/// Wire one plugin behind a route that *also* runs a field pipeline, so +/// two editors touch the same content part in one request. +async fn manager_with_yaml( + kind: &'static str, + factory: Box, + yaml: &str, +) -> Arc { + let mgr = Arc::new(PluginManager::default()); + mgr.register_factory(kind, factory); + register_apl( + &mgr, + AplOptions { + dispatch_cache: Arc::new(DispatchCache::new()), + session_store: Arc::new(MemorySessionStore::new()), + pdps: Vec::new(), + pdp_factories: Vec::new(), + session_store_factories: Vec::new(), + base_capabilities: None, + }, + ); + mgr.load_config_yaml(yaml).expect("load_config_yaml"); + mgr.initialize().await.expect("initialize"); + mgr +} + +/// Routes match on the request's entity type + name, so a request needs +/// tool meta for the `tool: get_weather` handler to fire at all. +fn tool_meta() -> Extensions { + let mut meta = cpex_core::extensions::MetaExtension::default(); + meta.entity_type = Some("tool".to_string()); + meta.entity_name = Some("get_weather".to_string()); + Extensions { + meta: Some(Arc::new(meta)), + ..Default::default() + } +} + +fn tool_call_part(city: &str) -> ContentPart { + ContentPart::ToolCall { + content: ToolCall { + tool_call_id: "tc_001".to_string(), + name: "get_weather".to_string(), + arguments: [("city".to_string(), serde_json::json!(city))] + .into_iter() + .collect(), + namespace: None, + }, + } +} + +fn tool_result_part(content: &str) -> ContentPart { + ContentPart::ToolResult { + content: ToolResult { + tool_call_id: "tc_001".to_string(), + tool_name: "get_weather".to_string(), + content: serde_json::Value::String(content.to_string()), + is_error: false, + }, + } +} + +fn payload_of(role: Role, parts: Vec) -> MessagePayload { + MessagePayload { + message: Message::with_content(role, parts), + } +} + +/// The payload the host would forward, downcast back to CMF. +fn forwarded(result: &cpex_core::executor::PipelineResult) -> MessagePayload { + result + .modified_payload + .as_ref() + .expect("an allowed pipeline always carries the final payload") + .as_any() + .downcast_ref::() + .expect("cmf hooks carry MessagePayload") + .clone() +} + +fn tool_result_of(payload: &MessagePayload) -> Option<&serde_json::Value> { + payload.message.content.iter().find_map(|part| match part { + ContentPart::ToolResult { content } => Some(&content.content), + _ => None, + }) +} + +fn tool_arg_of(payload: &MessagePayload, key: &str) -> Option { + payload.message.content.iter().find_map(|part| match part { + ContentPart::ToolCall { content } => content.arguments.get(key).cloned(), + _ => None, + }) +} + +// --------------------------------------------------------------------- +// Direct mutations reach the host, whichever part they touched +// --------------------------------------------------------------------- + +/// The reported failure: a redactor rewrites only `ToolResult.content`, +/// so the message's text is untouched. The redaction must reach the host. +#[tokio::test] +async fn tool_result_redaction_reaches_the_host() { + let mgr = manager_with( + "redactor", + Box::new(RewriteFactory { + target: Target::ToolResultContent, + hook: "cmf.tool_pre_invoke", + }), + "cmf.tool_pre_invoke", + "pre_invocation", + ) + .await; + + let payload = payload_of( + Role::Tool, + vec![ + ContentPart::Text { + text: "here is the result".to_string(), + }, + tool_result_part("sk-secret-value"), + ], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing, "route should allow"); + assert!( + result.payload_modified, + "the plugin returned a mutation, so the pipeline must report one" + ); + let out = forwarded(&result); + assert_eq!( + tool_result_of(&out), + Some(&serde_json::Value::String("[REDACTED]".to_string())), + "the host must receive the redacted tool result, not the original secret" + ); + assert_eq!( + out.message.get_text_content(), + "here is the result", + "fixture sanity: text is untouched, so text comparison sees no change" + ); +} + +/// Same shape, on the arguments of an outgoing tool call. +#[tokio::test] +async fn tool_call_argument_rewrite_reaches_the_host() { + let mgr = manager_with( + "arg-redactor", + Box::new(RewriteFactory { + target: Target::ToolCallArguments, + hook: "cmf.tool_pre_invoke", + }), + "cmf.tool_pre_invoke", + "pre_invocation", + ) + .await; + + let payload = payload_of(Role::User, vec![tool_call_part("London")]); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + assert_eq!( + tool_arg_of(&forwarded(&result), "city"), + Some(serde_json::json!("[REDACTED]")) + ); +} + +/// A thinking block carries no text part at all, so this is the cheapest +/// proof the fix is variant-agnostic rather than a tool-result special case. +#[tokio::test] +async fn thinking_rewrite_reaches_the_host() { + let mgr = manager_with( + "thought-redactor", + Box::new(RewriteFactory { + target: Target::Thinking, + hook: "cmf.tool_pre_invoke", + }), + "cmf.tool_pre_invoke", + "pre_invocation", + ) + .await; + + let payload = payload_of( + Role::Assistant, + vec![ContentPart::Thinking { + text: "the user's SSN is 123-45-6789".to_string(), + }], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + assert_eq!( + forwarded(&result).message.get_thinking_content().as_deref(), + Some("[REDACTED]") + ); +} + +/// The case that always worked. Kept so a future change can't fix the +/// others by breaking this one. +#[tokio::test] +async fn text_rewrite_still_reaches_the_host() { + let mgr = manager_with( + "text-redactor", + Box::new(RewriteFactory { + target: Target::Text, + hook: "cmf.tool_pre_invoke", + }), + "cmf.tool_pre_invoke", + "pre_invocation", + ) + .await; + + let payload = payload_of( + Role::User, + vec![ContentPart::Text { + text: "my card is 4111 1111 1111 1111".to_string(), + }], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + assert_eq!(forwarded(&result).message.get_text_content(), "[REDACTED]"); +} + +/// Post phase carries its own handler instance, so it needs its own proof. +#[tokio::test] +async fn tool_result_redaction_reaches_the_host_in_post_phase() { + let mgr = manager_with( + "redactor", + Box::new(RewriteFactory { + target: Target::ToolResultContent, + hook: "cmf.tool_post_invoke", + }), + "cmf.tool_post_invoke", + "post_invocation", + ) + .await; + + let payload = payload_of(Role::Tool, vec![tool_result_part("sk-secret-value")]); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_post_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + assert_eq!( + tool_result_of(&forwarded(&result)), + Some(&serde_json::Value::String("[REDACTED]".to_string())) + ); +} + +// --------------------------------------------------------------------- +// A pipeline edit and a plugin edit in the same request +// +// Both write to the same content part: the pipeline through +// `route_payload.args` / `.result`, the plugin through the payload +// itself. Folding the pipeline's view back in has to be per-path, or +// whichever editor is folded last wins and the other's work vanishes. +// --------------------------------------------------------------------- + +/// An `args:` pipeline redacts `city` while a plugin rewrites `token` on +/// the same tool call. Both edits must reach the host. +#[tokio::test] +async fn args_pipeline_and_plugin_edits_both_survive() { + const YAML: &str = r#" +plugins: + - name: token-scrubber + kind: token-scrubber + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_weather + apl: + args: + city: "str | redact" + pre_invocation: + - "plugin(token-scrubber)" +"#; + + let mgr = manager_with_yaml( + "token-scrubber", + Box::new(RewriteFactory { + target: Target::ToolCallArgument("token"), + hook: "cmf.tool_pre_invoke", + }), + YAML, + ) + .await; + + let payload = payload_of( + Role::User, + vec![ContentPart::ToolCall { + content: ToolCall { + tool_call_id: "tc_001".to_string(), + name: "get_weather".to_string(), + arguments: [ + ("city".to_string(), serde_json::json!("London")), + ("token".to_string(), serde_json::json!("sk-secret")), + ] + .into_iter() + .collect(), + namespace: None, + }, + }], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing, "route should allow"); + let out = forwarded(&result); + assert_eq!( + tool_arg_of(&out, "city"), + Some(serde_json::json!("[REDACTED]")), + "the pipeline's redaction must reach the host" + ); + assert_eq!( + tool_arg_of(&out, "token"), + Some(serde_json::json!("[REDACTED]")), + "the plugin's edit must not be clobbered by folding the pipeline's \ + args back in" + ); +} + +/// The mirror case on the response side: a `result:` pipeline masks one +/// field while a plugin redacts another in the same tool result. +#[tokio::test] +async fn result_pipeline_and_plugin_edits_both_survive() { + const YAML: &str = r#" +plugins: + - name: ssn-redactor + kind: ssn-redactor + hooks: [cmf.tool_post_invoke] +routes: + - tool: get_weather + apl: + result: + employee_id: "str | mask(2)" + post_invocation: + - "plugin(ssn-redactor)" +"#; + + let mgr = manager_with_yaml( + "ssn-redactor", + Box::new(RewriteFactory { + target: Target::ToolResultField("ssn"), + hook: "cmf.tool_post_invoke", + }), + YAML, + ) + .await; + + let payload = payload_of( + Role::Tool, + vec![ContentPart::ToolResult { + content: ToolResult { + tool_call_id: "tc_001".to_string(), + tool_name: "get_weather".to_string(), + content: serde_json::json!({ + "employee_id": "E12345", + "ssn": "123-45-6789", + }), + is_error: false, + }, + }], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_post_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing, "route should allow"); + let out = forwarded(&result); + let content = tool_result_of(&out).expect("tool result present").clone(); + assert_eq!( + content.get("employee_id"), + Some(&serde_json::json!("****45")), + "the pipeline's mask must reach the host" + ); + assert_eq!( + content.get("ssn"), + Some(&serde_json::json!("[REDACTED]")), + "the plugin's redaction must not be clobbered by folding the \ + pipeline's result back in" + ); +} + +/// A plugin invoked as a pipeline stage on `city` rewrites `city`. The +/// redaction must land on that argument and nowhere else — in +/// particular, the message's unrelated text must not be copied over it. +#[tokio::test] +async fn plugin_stage_rewrites_only_the_field_it_was_pointed_at() { + const YAML: &str = r#" +plugins: + - name: city-scrubber + kind: city-scrubber + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_weather + apl: + args: + city: "str | plugin(city-scrubber)" +"#; + + let mgr = manager_with_yaml( + "city-scrubber", + Box::new(RewriteFactory { + target: Target::ToolCallArgument("city"), + hook: "cmf.tool_pre_invoke", + }), + YAML, + ) + .await; + + let payload = payload_of( + Role::User, + vec![ + ContentPart::Text { + text: "chatter that must not become an argument".to_string(), + }, + tool_call_part("London"), + ], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing, "route should allow"); + assert_eq!( + tool_arg_of(&forwarded(&result), "city"), + Some(serde_json::json!("[REDACTED]")) + ); +} + +/// Two plugins, each rewriting a different part. Mutations accumulate +/// across a route's plugin chain, so the last one to run must not be the +/// only one that survives. +#[tokio::test] +async fn mutations_from_several_plugins_accumulate() { + const YAML: &str = r#" +plugins: + - name: result-redactor + kind: result-redactor + hooks: [cmf.tool_pre_invoke] + - name: thought-redactor + kind: thought-redactor + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_weather + apl: + pre_invocation: + - "plugin(result-redactor)" + - "plugin(thought-redactor)" +"#; + + let mgr = Arc::new(PluginManager::default()); + mgr.register_factory( + "result-redactor", + Box::new(RewriteFactory { + target: Target::ToolResultContent, + hook: "cmf.tool_pre_invoke", + }), + ); + mgr.register_factory( + "thought-redactor", + Box::new(RewriteFactory { + target: Target::Thinking, + hook: "cmf.tool_pre_invoke", + }), + ); + register_apl( + &mgr, + AplOptions { + dispatch_cache: Arc::new(DispatchCache::new()), + session_store: Arc::new(MemorySessionStore::new()), + pdps: Vec::new(), + pdp_factories: Vec::new(), + session_store_factories: Vec::new(), + base_capabilities: None, + }, + ); + mgr.load_config_yaml(YAML).expect("load_config_yaml"); + mgr.initialize().await.expect("initialize"); + + let payload = payload_of( + Role::Assistant, + vec![ + ContentPart::Thinking { + text: "ssn is 123-45-6789".to_string(), + }, + tool_result_part("sk-secret-value"), + ], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + let out = forwarded(&result); + assert_eq!( + tool_result_of(&out), + Some(&serde_json::Value::String("[REDACTED]".to_string())), + "the first plugin's redaction must survive the second plugin's dispatch" + ); + assert_eq!( + out.message.get_thinking_content().as_deref(), + Some("[REDACTED]"), + "the second plugin's redaction must land too" + ); +} + +/// A route that mutates and then denies must forward nothing. Half- +/// applying a denied request is worse than either outcome. +#[tokio::test] +async fn a_denied_route_forwards_no_payload() { + const YAML: &str = r#" +plugins: + - name: redactor + kind: redactor + hooks: [cmf.tool_pre_invoke] + - name: denier + kind: denier + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_weather + apl: + pre_invocation: + - "plugin(redactor)" + - "plugin(denier)" +"#; + + let mgr = Arc::new(PluginManager::default()); + mgr.register_factory( + "redactor", + Box::new(RewriteFactory { + target: Target::ToolResultContent, + hook: "cmf.tool_pre_invoke", + }), + ); + mgr.register_factory("denier", Box::new(DenyFactory)); + register_apl( + &mgr, + AplOptions { + dispatch_cache: Arc::new(DispatchCache::new()), + session_store: Arc::new(MemorySessionStore::new()), + pdps: Vec::new(), + pdp_factories: Vec::new(), + session_store_factories: Vec::new(), + base_capabilities: None, + }, + ); + mgr.load_config_yaml(YAML).expect("load_config_yaml"); + mgr.initialize().await.expect("initialize"); + + let payload = payload_of(Role::Tool, vec![tool_result_part("sk-secret-value")]); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(!result.continue_processing, "the route must deny"); + assert!( + result.modified_payload.is_none(), + "a denied pipeline carries no payload forward" + ); + assert!(!result.payload_modified); +} + +/// An audit-mode plugin cannot modify: the executor drops whatever +/// payload it returns. Nothing may report that mutation as applied, or +/// the host would forward a payload the framework deliberately rejected. +#[tokio::test] +async fn a_mutation_the_executor_rejects_is_not_reported() { + const YAML: &str = r#" +plugins: + - name: redactor + kind: redactor + hooks: [cmf.tool_pre_invoke] + mode: audit +routes: + - tool: get_weather + apl: + pre_invocation: + - "plugin(redactor)" +"#; + + let mgr = manager_with_yaml( + "redactor", + Box::new(RewriteFactory { + target: Target::ToolResultContent, + hook: "cmf.tool_pre_invoke", + }), + YAML, + ) + .await; + + let payload = payload_of(Role::Tool, vec![tool_result_part("sk-secret-value")]); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + assert!( + !result.payload_modified, + "an audit-mode plugin's payload is discarded, so no mutation was applied" + ); + assert_eq!( + tool_result_of(&forwarded(&result)), + Some(&serde_json::Value::String("sk-secret-value".to_string())), + "and the original must be what's forwarded" + ); +} + +/// An `omit` stage drops a field while a plugin rewrites another. The +/// removal has to apply to the plugin's payload, not replace it. +#[tokio::test] +async fn omit_stage_and_plugin_edit_both_survive() { + const YAML: &str = r#" +plugins: + - name: token-scrubber + kind: token-scrubber + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_weather + apl: + args: + debug: "str | omit" + pre_invocation: + - "plugin(token-scrubber)" +"#; + + let mgr = manager_with_yaml( + "token-scrubber", + Box::new(RewriteFactory { + target: Target::ToolCallArgument("token"), + hook: "cmf.tool_pre_invoke", + }), + YAML, + ) + .await; + + let payload = payload_of( + Role::User, + vec![ContentPart::ToolCall { + content: ToolCall { + tool_call_id: "tc_001".to_string(), + name: "get_weather".to_string(), + arguments: [ + ("city".to_string(), serde_json::json!("London")), + ("debug".to_string(), serde_json::json!("verbose")), + ("token".to_string(), serde_json::json!("sk-secret")), + ] + .into_iter() + .collect(), + namespace: None, + }, + }], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + let out = forwarded(&result); + assert_eq!(tool_arg_of(&out, "debug"), None, "the omit must apply"); + assert_eq!( + tool_arg_of(&out, "token"), + Some(serde_json::json!("[REDACTED]")), + "the plugin's edit must not be undone by applying the omit" + ); + assert_eq!( + tool_arg_of(&out, "city"), + Some(serde_json::json!("London")), + "an argument nobody touched must be left alone" + ); +} + +/// Nested arguments: the pipeline redacts `user.ssn`, a plugin rewrites +/// `user.name`. Merging has to descend into the object rather than +/// replace it. +#[tokio::test] +async fn nested_pipeline_and_plugin_edits_both_survive() { + const YAML: &str = r#" +plugins: + - name: name-scrubber + kind: name-scrubber + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_weather + apl: + args: + user.ssn: "str | redact" + pre_invocation: + - "plugin(name-scrubber)" +"#; + + let mgr = manager_with_yaml( + "name-scrubber", + Box::new(RewriteFactory { + target: Target::NestedToolCallArgument("user", "name"), + hook: "cmf.tool_pre_invoke", + }), + YAML, + ) + .await; + + let payload = payload_of( + Role::User, + vec![ContentPart::ToolCall { + content: ToolCall { + tool_call_id: "tc_001".to_string(), + name: "get_weather".to_string(), + arguments: [( + "user".to_string(), + serde_json::json!({"name": "Ada Lovelace", "ssn": "123-45-6789"}), + )] + .into_iter() + .collect(), + namespace: None, + }, + }], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + let user = tool_arg_of(&forwarded(&result), "user").expect("user argument present"); + assert_eq!( + user.get("ssn"), + Some(&serde_json::json!("[REDACTED]")), + "the pipeline's nested redaction must reach the host" + ); + assert_eq!( + user.get("name"), + Some(&serde_json::json!("[REDACTED]")), + "the plugin's edit to a sibling key must survive the merge" + ); +} + +/// A plugin stage must not undo an earlier stage in its own chain. +/// +/// The payload never sees a pipeline's interim edits, so the field a +/// plugin stage reads back still holds the pre-`mask` value. Treating +/// that as "the plugin's new value" hands the plaintext back to the +/// pipeline and the mask is forwarded undone — the redaction path +/// failing open, which is the whole point of this code. +#[tokio::test] +async fn a_plugin_stage_does_not_undo_an_earlier_mask_in_the_same_chain() { + const YAML: &str = r#" +plugins: + - name: token-scrubber + kind: token-scrubber + hooks: [cmf.tool_pre_invoke] +routes: + - tool: get_weather + apl: + args: + city: "str | mask(2) | plugin(token-scrubber)" + pre_invocation: + - "plugin(token-scrubber)" +"#; + + // The plugin rewrites `token`, never `city`. It must therefore report + // no change for `city` and leave the mask standing. + let mgr = manager_with_yaml( + "token-scrubber", + Box::new(RewriteFactory { + target: Target::ToolCallArgument("token"), + hook: "cmf.tool_pre_invoke", + }), + YAML, + ) + .await; + + let payload = payload_of( + Role::User, + vec![ContentPart::ToolCall { + content: ToolCall { + tool_call_id: "tc_001".to_string(), + name: "get_weather".to_string(), + arguments: [ + ("city".to_string(), serde_json::json!("London")), + ("token".to_string(), serde_json::json!("sk-secret")), + ] + .into_iter() + .collect(), + namespace: None, + }, + }], + ); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing, "route should allow"); + let out = forwarded(&result); + assert_eq!( + tool_arg_of(&out, "city"), + Some(serde_json::json!("****on")), + "the mask stage's output must survive the plugin stage that followed it" + ); + assert_eq!( + tool_arg_of(&out, "token"), + Some(serde_json::json!("[REDACTED]")), + "and the plugin's own edit must still reach the host" + ); +} + +/// A route whose plugin allows without mutating, and which has no +/// pipelines, must not report a modification. Reporting one is harmless +/// for correctness but makes every request look modified, so the signal +/// stops meaning anything. +#[tokio::test] +async fn allow_without_mutation_reports_no_modification() { + let mgr = manager_with( + "noop", + Box::new(NoopFactory { + hook: "cmf.tool_pre_invoke", + }), + "cmf.tool_pre_invoke", + "pre_invocation", + ) + .await; + + let payload = payload_of(Role::User, vec![tool_call_part("London")]); + + let (result, _bg) = mgr + .invoke_named::("cmf.tool_pre_invoke", payload, tool_meta(), None) + .await; + + assert!(result.continue_processing); + assert!(!result.payload_modified); + assert_eq!( + tool_arg_of(&forwarded(&result), "city"), + Some(serde_json::json!("London")), + "an untouched request must arrive untouched" + ); +} diff --git a/crates/cpex-core/src/cmf/message.rs b/crates/cpex-core/src/cmf/message.rs index 4eb7678c..a8818b4d 100644 --- a/crates/cpex-core/src/cmf/message.rs +++ b/crates/cpex-core/src/cmf/message.rs @@ -79,6 +79,12 @@ impl Message { /// Extract all text content from the message. /// /// Concatenates text from all `Text` content parts. + /// + /// Reads `Text` parts and nothing else, so it is not a stand-in for + /// message equality or change detection: two messages differing only + /// in a tool call, tool result, thinking block, or attachment return + /// the same string. Callers asking "did this change?" need a signal + /// from whatever performed the change. pub fn get_text_content(&self) -> String { let mut texts = Vec::new(); for part in &self.content { diff --git a/crates/cpex-core/src/executor.rs b/crates/cpex-core/src/executor.rs index 364379bf..721e6d13 100644 --- a/crates/cpex-core/src/executor.rs +++ b/crates/cpex-core/src/executor.rs @@ -70,7 +70,15 @@ impl Default for ExecutorConfig { /// /// Background tasks are returned separately as [`BackgroundTasks`] /// to keep the policy result immutable. +/// +/// `#[non_exhaustive]`: this result type keeps gaining fields as the +/// engine grows, so it is sealed against external struct-literal +/// construction and exhaustive destructuring — hosts read it, they don't +/// build it. Construct via [`Self::allowed_with`] / [`Self::denied`] plus +/// the `with_*` builders. New fields can then be added without breaking +/// downstream readers. #[derive(Debug)] +#[non_exhaustive] pub struct PipelineResult { /// Whether the pipeline should continue processing. /// `false` means a plugin denied — the pipeline was halted. @@ -78,8 +86,30 @@ pub struct PipelineResult { /// The final payload after all modifications (type-erased). /// `None` if the pipeline was denied before any modifications. + /// + /// Note this is `Some` on **every** allowed pipeline, carrying the + /// final payload whether or not a plugin touched it. To learn + /// whether anything actually changed, read [`Self::payload_modified`] + /// — do not compare payload contents, and do not read `is_some()` as + /// "was modified". pub modified_payload: Option>, + /// Whether any plugin's payload modification was accepted into + /// `modified_payload` above. + /// + /// Set by the phases that can modify (sequential, transform) at the + /// moment a handler's payload replaces the current one, so it + /// reflects what the executor actually applied: a plugin lacking the + /// modify capability, or one running in a read-only phase, does not + /// set it. + /// + /// This exists because the fact is knowable only here. A caller + /// comparing payloads afterwards cannot: the payload types are + /// type-erased with no equality, and content-shaped comparisons + /// (e.g. a message's text) are blind to whichever parts they don't + /// read. + pub payload_modified: bool, + /// The final extensions after all modifications. /// `None` if no plugin modified extensions. pub modified_extensions: Option, @@ -114,6 +144,7 @@ impl PipelineResult { Self { continue_processing: true, modified_payload: Some(payload), + payload_modified: false, modified_extensions: Some(extensions), violation: None, errors: Vec::new(), @@ -122,6 +153,14 @@ impl PipelineResult { } } + /// Record that a plugin's payload modification was applied. Chained + /// off [`Self::allowed_with`] by the executor, mirroring + /// [`Self::with_errors`]. + pub fn with_payload_modified(mut self, modified: bool) -> Self { + self.payload_modified = modified; + self + } + /// Pipeline was denied by a plugin. pub fn denied( violation: crate::error::PluginViolation, @@ -131,6 +170,7 @@ impl PipelineResult { Self { continue_processing: false, modified_payload: None, + payload_modified: false, modified_extensions: Some(extensions), violation: Some(violation), errors: Vec::new(), @@ -288,6 +328,10 @@ impl Executor { // observable. Halt-condition errors (Fail, deny) skip this and // become the violation directly. let mut errors: Vec = Vec::new(); + // Sticky across both modifying phases: true once any handler's + // payload has been accepted. Reported on the result so callers + // read an exact signal instead of comparing payload contents. + let mut payload_modified = false; if let Some(v) = self .run_serial_phase( @@ -299,6 +343,7 @@ impl Executor { true, // can_modify "SEQUENTIAL", &mut errors, + &mut payload_modified, ) .await { @@ -319,6 +364,7 @@ impl Executor { true, // can_modify "TRANSFORM", &mut errors, + &mut payload_modified, ) .await; @@ -362,7 +408,8 @@ impl Executor { ( PipelineResult::allowed_with(current_payload, current_extensions, ctx_table) - .with_errors(errors), + .with_errors(errors) + .with_payload_modified(payload_modified), BackgroundTasks::from_handles(bg_handles), ) } @@ -374,6 +421,11 @@ impl Executor { /// a borrow and clone only if they modify. Modified payloads in /// the result replace the current payload. /// + /// `payload_modified` is set to `true` when a handler's payload is + /// accepted, and never cleared — this is the only place that fact is + /// observable, so it's reported out rather than left to be guessed + /// from the resulting payload's contents. + /// /// Each plugin's context is looked up in the context table (preserving /// `local_state` from previous hooks) or created fresh. After execution, /// `global_state` changes are merged back so the next plugin sees them. @@ -388,6 +440,7 @@ impl Executor { can_modify: bool, phase_label: &str, errors: &mut Vec, + payload_modified: &mut bool, ) -> Option { for entry in entries { // Borrow names/ids on the happy path — allocate only when @@ -449,6 +502,7 @@ impl Executor { if can_modify { if let Some(mp) = erased.modified_payload { *payload = mp; + *payload_modified = true; } if let Some(owned) = erased.modified_extensions { // Pointer-equality gate on the truly-immutable @@ -1116,6 +1170,10 @@ mod tests { assert!(result.continue_processing); assert!(result.modified_payload.is_some()); assert!(result.violation.is_none()); + assert!( + !result.payload_modified, + "carrying a payload is not the same as a plugin having changed it" + ); } #[test] diff --git a/crates/cpex-core/src/manager.rs b/crates/cpex-core/src/manager.rs index b752f7c0..da5bb66c 100644 --- a/crates/cpex-core/src/manager.rs +++ b/crates/cpex-core/src/manager.rs @@ -2733,6 +2733,10 @@ mod tests { .await; assert!(result.continue_processing); + assert!( + result.payload_modified, + "the transform accepted a new payload, so the result must say so" + ); let final_payload = result.modified_payload.unwrap(); let typed = final_payload .as_any() @@ -2741,6 +2745,34 @@ mod tests { assert_eq!(typed.value, "original_transformed"); } + /// `modified_payload` is `Some` on every allowed pipeline, carrying + /// the final payload whether or not a plugin touched it. Only + /// `payload_modified` distinguishes the two, so callers deciding + /// whether to forward a rewritten payload must read that. + #[tokio::test] + async fn allow_without_mutation_reports_payload_unmodified() { + let mgr = PluginManager::default(); + let config = make_config("allow-plugin", 10, PluginMode::Sequential); + let plugin = Arc::new(AllowPlugin { + cfg: config.clone(), + }); + + mgr.register_handler::(plugin, config).unwrap(); + mgr.initialize().await.unwrap(); + + let payload = TestPayload { + value: "original".into(), + }; + + let (result, _) = mgr + .invoke::(payload, Extensions::default(), None) + .await; + + assert!(result.continue_processing); + assert!(result.modified_payload.is_some()); + assert!(!result.payload_modified); + } + /// Transform phase is documented `can_block: No` (plugin.rs PluginMode /// table). An `on_error: Fail` plugin error or timeout in Transform must /// NOT halt the pipeline — non-blocking is non-blocking, regardless of diff --git a/crates/cpex-ffi/src/lib.rs b/crates/cpex-ffi/src/lib.rs index ad62bdfb..4d4d765a 100644 --- a/crates/cpex-ffi/src/lib.rs +++ b/crates/cpex-ffi/src/lib.rs @@ -875,6 +875,7 @@ pub unsafe extern "C" fn cpex_invoke( metadata: result.metadata, payload_type: result_payload_type, modified_payload: modified_payload_bytes, + payload_modified: result.payload_modified, modified_extensions: modified_extensions_bytes, }; @@ -1122,6 +1123,7 @@ unsafe fn finish_pipeline_result( metadata: result.metadata, payload_type: result_payload_type, modified_payload: modified_payload_bytes, + payload_modified: result.payload_modified, modified_extensions: modified_extensions_bytes, }; @@ -1276,10 +1278,19 @@ struct FfiPipelineResult { metadata: Option, /// Payload type ID — tells the Go caller how to deserialize. payload_type: u8, - /// Modified payload as raw MessagePack bytes (if a plugin modified it). + /// Modified payload as raw MessagePack bytes. Present on every + /// allowed pipeline, carrying the final payload whether or not a + /// plugin touched it — read `payload_modified` to learn whether + /// anything actually changed. #[serde(skip_serializing_if = "Option::is_none")] #[serde(with = "serde_bytes_opt")] modified_payload: Option>, + /// Whether a plugin's payload modification was accepted. The exact + /// signal; `modified_payload.is_some()` is not a substitute, and + /// comparing payload contents across the boundary cannot see + /// mutations to non-text parts. + #[serde(default)] + payload_modified: bool, /// Modified extensions as raw MessagePack bytes (if a plugin modified them). #[serde(skip_serializing_if = "Option::is_none")] #[serde(with = "serde_bytes_opt")] diff --git a/docs/plans/2026-08-06-001-fix-modify-payload-propagation-plan.md b/docs/plans/2026-08-06-001-fix-modify-payload-propagation-plan.md new file mode 100644 index 00000000..44f0f33f --- /dev/null +++ b/docs/plans/2026-08-06-001-fix-modify-payload-propagation-plan.md @@ -0,0 +1,375 @@ +--- +title: "fix: propagate plugin payload mutations exactly instead of inferring them" +type: fix +status: implemented +date: 2026-08-06 +origin: https://github.com/contextforge-org/cpex/issues/151 +--- + +## Implementation Note + +One premise in this plan was wrong, and the fix is larger than planned because of it. + +The plan (following the issue) assumed `CmfPluginInvoker` already knew whether a plugin returned a mutation, because `result.modified_payload` is `Some` in that case. It isn't a signal: `PipelineResult::allowed_with` sets `modified_payload: Some(payload)` on **every** allowed pipeline, carrying the final payload whether or not any plugin touched it. Reading `is_some()` reports a mutation on every request. + +Acceptance is only observable inside `Executor::run_serial_phase`, at `*payload = mp`. So the signal is recorded there and surfaced as a new `PipelineResult::payload_modified` field, which the invoker reads. That contradicts this plan's "no change to the executor" scope boundary; the alternative was comparing serialized payloads, which is both slower and back to inferring. Hosts get the same exact signal as a side effect. + +This also exposed a fourth bug the plan didn't predict: because the old code keyed off `modified_payload.is_some()`, **every** `plugin(...)` stage in a field pipeline reported a new field value, mutation or not — so a non-mutating plugin stage overwrote its field with the message's concatenated text. Fixed by the same gate. + +Two smaller deviations: + +- U4 and U5 landed in one commit. Switching the branch conditions to the decision's flags leaves `pre_args` / `pre_result` unused until the merge consumes them, so splitting them would mean committing code that doesn't compile. +- The field readback reports no change when the field's value is unchanged, not only when the field is absent. Returning the current value would mark the field replaced on every mutating dispatch, which sets `args_modified` and forces a needless write-back. + +# fix: propagate plugin payload mutations exactly instead of inferring them + +## Summary + +`AplRouteHandler::invoke` decides whether to emit `modified_payload` by diffing `Message::get_text_content()` before and after route evaluation (`crates/apl-cpex/src/route_handler.rs:478`). That accessor reads only `ContentPart::Text`, so a plugin that mutates a `ToolResult`, `ToolCall`, `Resource`, `Thinking`, or any other non-text variant produces a payload the handler classifies as unchanged and drops. The plugin reports success and the host forwards the original bytes. + +The reported bug is one instance of a pattern that repeats three more times in the same twenty lines: mutation is *inferred* from values rather than *signalled* by the code that performed it. This plan fixes all four instances. + +1. Direct `modify_payload` mutations are detected by a text diff, so non-text mutations are dropped. (the reported bug) +2. The `args:` write-back replaces `ToolCall.arguments` wholesale, so a pipeline edit silently clobbers a plugin's argument mutation. The `result:` write-back has the same shape. +3. Pipeline modification is detected by re-extracting and diffing, even though `RouteDecision` already carries `args_modified` / `result_modified` and the handler ignores them. +4. A plugin invoked as a pipeline field stage reports its new field value as `Value::String(get_text_content())`, which is wrong for any structured field: a per-key stage on `args.city` gets back the concatenation of every text part in the message. + +Every fix replaces an inference with the signal that already exists at the site that knows the answer. + +--- + +## Documentation and Traceability Constraint + +**Applies to every unit below, without exception.** + +Identifiers from this plan and from any requirements document (`R1`, `U2`, `D3`, `AE4`, and any similar scheme) must not appear in: + +- source code or code comments +- rustdoc / doc comments +- test names or test comments +- commit messages +- the PR title, description, or review replies +- `CHANGELOG.md` + +Doc IDs are planning artifacts. They are noise to a future reader of the code and they rot the moment this document changes or is deleted. Describe the behavior in plain terms instead. If traceability matters, express the behavior itself clearly and let the issue number (`#151`) carry the external link. The `Requirements:` annotations inside the units below live in this document only; they do not travel into the diff. + +--- + +## Problem Frame + +Verified against `main` at `d21593c`. Branch `fix/payload_mutations` is currently identical to `main`, so nothing is started. + +### The reported defect + +The decision chain in `AplRouteHandler::invoke` has three branches (`crates/apl-cpex/src/route_handler.rs:454-484`): + +1. `route_payload.args != pre_args`: an APL `args:` pipeline rewrote a field. +2. Post phase only, `pre_result != route_payload.result`: a `result:` pipeline rewrote the upstream response. +3. `msg_payload.message.get_text_content() != final_payload.message.get_text_content()`: intended to catch a plugin calling `modify_payload` directly. + +Branch 3 is broken. `Message::get_text_content()` (`crates/cpex-core/src/cmf/message.rs:82-92`) matches only `ContentPart::Text`, while `ContentPart` (`crates/cpex-core/src/cmf/content.rs:227-275`) has twelve variants: `Text`, `Thinking`, `ToolCall`, `ToolResult`, `Resource`, `ResourceRef`, `PromptRequest`, `PromptResult`, `Image`, `Video`, `Audio`, `Document`. Eleven of twelve can be mutated with no effect on the branch condition. + +Severity is highest for redaction and sanitization plugins, which are exactly the plugins that rewrite `ToolResult.content`. The failure is fail-open on the security-relevant path: the plugin's telemetry says it redacted, and the unredacted value goes downstream. + +### The write-back clobber + +`pre_args` and `route_payload.args` are both projections of the *pre-evaluation* message. When an `args:` pipeline changes any key, branch 1 calls `write_args_back_to_message(&mut updated.message, &route_payload.args)` (`crates/apl-cpex/src/route_handler.rs:454-462`, helper at `:724-747`), which replaces `ToolCall.arguments` wholesale with the pipeline's args object. A plugin's own rewrite of `ToolCall.arguments` is not in that object, so it is overwritten. Fixing branch 3 does not help: branch 1 wins the `if/else` chain, and its write is total rather than differential. + +`write_result_back_to_message` (`:771-782`) has the identical shape against `ToolResult.content`, so a `result:` pipeline edit clobbers a plugin's tool-result redaction in the Post phase. That is the same user-visible symptom as the reported bug, reached by a different path. + +### The ignored pipeline signals + +`RouteDecision` already carries `args_modified` and `result_modified` (`crates/apl-core/src/route.rs:76-79`), set by the only code that can rewrite those values: the section pipelines (`crates/apl-core/src/route.rs:110-155`, `:202-240`) and `do:`-block field ops (`crates/apl-core/src/evaluator.rs:1112-1190`), all of which set the flag exactly when `set_dotted` / `remove_dotted` reports a write. `AplRouteHandler` never reads either flag. It re-extracts `pre_args` / `pre_result` from the pre-evaluation message and diffs. Two allocations per request to recompute something the decision already states. + +### The field-stage value projection + +For `PluginInvocation::Field`, `CmfPluginInvoker::invoke` returns `modified_value = Some(Value::String(modified.message.get_text_content()))` (`crates/apl-cpex/src/cmf_invoker.rs:336-339`). The evaluator assigns that straight onto the field (`crates/apl-core/src/evaluator.rs:1442-1460`). + +For a text-shaped message with a whole-message field, that is coherent: the projection *is* the field. For a structured tool call it is wrong in a way that is worse than lossy. A route with `args: city | plugin(scrubber)` sets `args.city` to the concatenated text of the entire message. Nothing warns. + +A second inconsistency sits underneath it: the `name` handed to `PluginInvocation::Field` is root-relative from section pipelines (`rule.field`, e.g. `city`) but prefixed from `do:`-block field ops (the full `path`, e.g. `args.city`, `crates/apl-core/src/evaluator.rs:1163`). Any invoker that wants to read a field back by name has to guess which convention it got, and stripping an `args.` prefix defensively would corrupt a legitimate argument named `args`. + +### Two facts that shape the fixes + +- The ground truth for a plugin mutation already exists. `CmfPluginInvoker::invoke` downcasts `result.modified_payload` and, on success, writes it into `self.payload` (`crates/apl-cpex/src/cmf_invoker.rs:332-341`). That is the only site in the crate that mutates the shared payload: `DelegationPluginInvoker` and `ElicitationPluginInvoker` share the extensions `Mutex` but never the payload. A flag set there is complete. +- The diff-based alternative is expensive. `Message`, `ContentPart`, `ToolCall`, `ToolResult`, `Resource`, and the other nine payload structs derive only `Debug, Clone, Serialize, Deserialize`. Comparing content vectors means adding `PartialEq` across all of them and committing to structural equality as a public API property of the CMF types. + +--- + +## Requirements + +**Direct mutation propagation** +- R1. A plugin mutation delivered via `PluginResult::modify_payload` reaches `ErasedResultFields.modified_payload` regardless of which `ContentPart` variant it touched. +- R2. Detection is an explicit signal recorded where the mutation is accepted, not a value diff computed later. +- R3. A dropped mutation (downcast failure) still warns and still reports "not modified". The invoker never claims a mutation it did not accept. + +**Composition with pipelines** +- R4. A plugin mutation and a pipeline mutation in the same request both survive, including when both target `ToolCall.arguments` (Pre) or `ToolResult.content` (Post). +- R5. Pipeline write-back applies only what the pipeline changed, leaving the rest of the plugin-mutated message intact. +- R6. Pipeline modification is read from `RouteDecision.args_modified` / `result_modified` rather than re-derived by diffing. + +**Field-stage dispatch** +- R7. A plugin invoked as a pipeline field stage reports a new value for *that field*, not the message's concatenated text. +- R8. `PluginInvocation::Field.name` has one documented convention across all call sites: a path relative to the args or result root, with the phase selecting the root. +- R9. When the field cannot be located in the mutated payload, the invoker reports no field change. The payload mutation still propagates via R1, so nothing is lost. +- R10. Existing behavior is preserved for text-shaped messages, where the whole projection is the field value. + +**Hygiene** +- R11. `Message::get_text_content()` carries a doc note that it is a text accessor and not a change detector, so the next caller does not repeat this. + +--- + +## Scope Boundaries + +- No change to `PluginResult`, `ErasedResultFields`, the executor, or the FFI boundary. +- No `PartialEq` derives on CMF payload types. +- No change to `Message::get_text_content()` behavior. Docs only. +- No new hook, no new capability, no config surface. +- Python-side `CopyOnWriteList` / `CopyOnWriteDict` equality bugs (#135, #54) are the same defect family but a different codebase and mechanism. Not touched. + +### Known Limitation, Documented Not Fixed + +Field pipelines and the shared request payload are not synchronized mid-pipeline. Earlier stages (`mask`, `redact`, `hash`) mutate `route_payload.args` only, so a later `plugin(...)` stage in the same chain is handed a payload whose arguments still hold the *original* values. After this plan the readback is field-precise and the payload mutation propagates, so no data is lost, but a field-stage plugin still does not see its own pipeline's in-progress edits. + +Fixing that means writing interim pipeline state into the shared payload before dispatch, which changes what every downstream `pre_invocation:` plugin sees. That is a semantic decision with its own blast radius. File it as a separate issue; do not fold it in here. Add a `// Known limitation:` comment at the dispatch site describing the behavior in plain terms, with no reference to this document. + +--- + +## Key Technical Decisions + +**D1. Explicit flag on the invoker, not a content diff.** Matches the issue's recommendation and the maintainer's confirmation. Exact instead of approximate, and it removes the content-shape dependency rather than widening it. The diff alternative would need `PartialEq` on twelve payload structs and would still be a guess about intent. + +**D2. `AtomicBool`, not `Mutex`.** The read happens in `AplRouteHandler::invoke` after evaluation and must not introduce an `await` inside the existing branch chain. Store with `Ordering::Release` at the mutation site, load with `Ordering::Acquire` in the accessor, so a mutation written from a `dispatch_parallel` branch task is visible to the reader. + +**D3. Set the flag whenever a mutation is accepted, even if the returned payload is byte-identical.** The flag answers "did a plugin return a payload?", which is the question the handler needs. A false positive costs one redundant body re-serialization; a false negative is the bug being fixed. `extensions_changed` already documents the same fail-safe tradeoff (`crates/apl-cpex/src/route_handler.rs:784-786`). + +**D4. Replace branch 3's condition outright rather than OR-ing the flag with the text diff.** The text diff can only be true when a plugin returned a payload, which is exactly when the flag is true. Keeping both leaves a dead heuristic reading as if it were load-bearing. + +**D5. Differential write-back, computed as a three-way merge.** Base is the args projected from the plugin-mutated payload; `pre_args` and `route_payload.args` bracket what the pipeline changed. Walk the pre/post pair recursively, apply only differing leaves and removals onto the base. With no plugin mutation the base equals `pre_args`, so the result is byte-identical to today's wholesale write, which makes existing tests the regression guard. + +**D6. Fix the field-name convention in apl-core rather than normalizing defensively in the invoker.** Stripping an `args.` prefix in the invoker would corrupt a legitimate argument named `args`. One line in `dispatch_field_op` makes both call sites root-relative, and the phase already tells the invoker which root to project. The name is informational to CMF plugins today (the invoker is its only structural consumer), so the change is low risk. + +**D7. Field readback projects the mutated message the same way APL projected the original.** Pre uses the args projection, Post uses the result projection. If the projection is an object, read the field's dotted path; if it is a scalar (a text-shaped message), the projection itself is the field value. That second rule is what preserves R10 and keeps the existing text tests green. + +**D8. Ship as one PR, in the unit order below, one commit per unit.** The units are individually revertable, and the first two carry the reported fix. If the fix needs cherry-picking to a patch release, U1 and U2 alone are sufficient and self-contained. + +--- + +## Implementation Units + +Reminder: no plan or requirement identifiers in code, comments, rustdoc, test names, or commit messages. See "Documentation and Traceability Constraint" above. + +### U1. Record payload mutation on `CmfPluginInvoker` + +**Goal:** the invoker exposes whether any plugin in this request returned a payload mutation it accepted. + +**Requirements:** R2, R3  ·  **Dependencies:** none + +**Files:** `crates/apl-cpex/src/cmf_invoker.rs` + +**Approach:** +- Add `payload_modified: AtomicBool` next to `payload` (`:85`), initialized `false` in `for_request`. Document it as request-scoped, sticky once set, and the authoritative answer to "did a plugin mutate the payload", so callers never infer it from content. +- In the `Some(modified)` arm (`:333-341`), immediately after `*self.payload.lock().await = modified.clone();`, store `true` with `Ordering::Release`. +- Do **not** set it in the downcast-failure arm. That path already warns and drops the mutation; claiming "modified" there would forward an unmutated payload while asserting it changed. +- Add `pub fn payload_was_modified(&self) -> bool` loading with `Ordering::Acquire`. Sync, not async, so the caller uses it inside the existing branch chain without restructuring. +- Extend the module-level request-scoped-state docs (`:16-24`) with one line on the flag. + +**Patterns to follow:** accessor shape of `current_payload` / `current_extensions` (`:157-166`). + +**Test scenarios** (`crates/apl-cpex/tests/cmf_invoker_dispatch.rs`, which already has `ModifyPluginFactory` and `payload_with_text`): +- Fresh invoker reports `false` before any dispatch. +- After dispatch to a plugin returning `modify_payload`, reports `true`. +- After dispatch to a plugin returning a plain allow, stays `false`. +- A plugin whose `modified_payload` is not a `MessagePayload` leaves the flag `false`. If the typed hook signature makes a foreign payload unconstructible from a test, say so in a test-module comment instead of leaving the case silently uncovered. + +**Verification:** `cargo test -p apl-cpex --test cmf_invoker_dispatch`. + +--- + +### U2. Consult the flag in `AplRouteHandler` + +**Goal:** the reported bug is fixed. Mutations to any `ContentPart` variant propagate. + +**Requirements:** R1  ·  **Dependencies:** U1 + +**Files:** `crates/apl-cpex/src/route_handler.rs` + +**Approach:** +- Replace the condition at `:478` with `invoker.payload_was_modified()`. Rewrite the branch comment: a plugin mutated the payload directly, the invoker recorded it, pass the invoker's view through. State plainly why a text diff cannot detect this, since the existing comment is correct about intent while the code is not. +- Add a `tracing::debug!` on that branch with the route key. The two pipeline branches are inferable from their inputs; this one was invisible. +- Leave branch ordering alone. U5 fixes the composition problem inside branches 1 and 2. + +**Test scenarios** (`crates/apl-cpex/tests/end_to_end_route.rs`, using the `register_apl` + `invoke_named::` pattern at `:660-728`, which returns a typed `PluginResult` whose `modified_payload` is directly assertable): +- Regression, the reported bug: a `pre_invocation` plugin rewrites only `ContentPart::ToolResult.content` to `[REDACTED]` and leaves every `Text` part alone. `modified_payload` is `Some` and carries the redacted value. Fails before this unit, passes after. +- Same for `ContentPart::ToolCall.arguments` on a pre route. +- Same for `ContentPart::Thinking`, the cheapest proof the fix is variant-agnostic rather than a `ToolResult` special case. +- Text-only mutation still propagates. +- A plugin that allows without mutating, on a route with no pipelines, yields `modified_payload: None`, so the flag has not made every request look modified. +- Post phase: a plugin rewriting `ToolResult.content` in `post_invocation` propagates. + +**Verification:** `cargo test -p apl-cpex`. + +--- + +### U3. Extract a shared message-projection module + +**Goal:** one home for the message-to-JSON projections, so the invoker and the handler cannot drift apart on field semantics. + +**Requirements:** enabler for R5, R7  ·  **Dependencies:** none (land after U2 to keep the fix commit small) + +**Files:** +- Create: `crates/apl-cpex/src/message_projection.rs` +- Modify: `crates/apl-cpex/src/route_handler.rs`, `crates/apl-cpex/src/lib.rs` + +**Approach:** +- Move `extract_args_from_message`, `extract_result_from_message`, `write_args_back_to_message`, `write_result_back_to_message`, and `rewrite_message_text` out of `route_handler.rs` (`:712-782` and the text helper) into the new module, `pub(crate)`, keeping their existing doc comments. +- Module doc states the contract both consumers depend on: which `ContentPart` each projection reads, that Pre projects args and Post projects result, and that the write-back functions are the inverse of the extractors. +- Pure move. No behavior change in this unit. + +**Test scenarios:** +- Unit tests for round-tripping each projection: tool-call args extract then write back yields the original message; tool-result likewise; a text-only message falls through to the text path. +- Existing `apl-cpex` suite passes unchanged. + +**Verification:** `cargo test -p apl-cpex`, `make lint`. + +--- + +### U4. Read pipeline modification from the decision + +**Goal:** branches 1 and 2 stop re-deriving what `RouteDecision` already states. + +**Requirements:** R6  ·  **Dependencies:** U2 + +**Files:** `crates/apl-cpex/src/route_handler.rs` + +**Approach:** +- Gate branch 1 on `decision.args_modified` and branch 2 on `matches!(self.phase, Phase::Post) && decision.result_modified`. +- Keep the `pre_args` / `pre_result` extraction. It stops being the detector and becomes the merge input U5 needs. Retitle the comment accordingly, so the next reader does not think it is still driving the decision. +- Note in the branch comment that these flags are set by `set_dotted` / `remove_dotted` succeeding, which is the only way those values change. + +**Behavior note:** a pipeline that writes a field back to the value it already held sets `args_modified` while the old diff saw equality. Such a request now emits `modified_payload` where it previously emitted `None`. Fail-safe, and consistent with D3. Call it out in the PR description. + +**Test scenarios:** +- Existing args-pipeline and result-pipeline tests pass unchanged. +- A `redact` stage whose condition is false leaves `modified_payload` as `None` (the flag is not set when no write happened). +- A pipeline stage that writes an identical value emits `modified_payload: Some`, asserted deliberately so the behavior note is pinned by a test rather than by prose. + +**Verification:** `cargo test -p apl-cpex`. + +--- + +### U5. Differential write-back for args and result + +**Goal:** a pipeline edit no longer clobbers a plugin's mutation to the same content part. + +**Requirements:** R4, R5  ·  **Dependencies:** U3, U4 + +**Files:** `crates/apl-cpex/src/route_handler.rs`, `crates/apl-cpex/src/message_projection.rs` + +**Approach:** +- Add `apply_changed_paths(base: &mut Value, pre: &Value, post: &Value)` to the projection module. Walk `pre` and `post` in parallel: for a differing or added leaf, write it into `base` at that path; for a key present in `pre` and absent in `post`, remove it from `base`. Objects recurse; arrays and scalars are leaves (whole-value replacement), matching `set_dotted`'s object-only path semantics (`crates/apl-core/src/route.rs:334-361`). +- Branch 1 becomes: project args from the plugin-mutated `final_payload` as the base, apply the changed paths from `pre_args` to `route_payload.args`, write the merged object back with `write_args_back_to_message`. +- Branch 2 does the same against the result projection. +- Keep the existing non-object fallbacks (`args.as_str()` to `rewrite_message_text`) untouched. +- With no plugin mutation the base equals `pre_args`, so the merged output is byte-identical to today's wholesale write. State that invariant in the function's doc comment, in plain terms. + +**Test scenarios:** +- An `args:` stage rewrites key `a` while a plugin rewrites key `b` on `ToolCall.arguments`: both survive. Fails before this unit. +- A `result:` stage rewrites one field while a plugin redacts a different field of `ToolResult.content`: both survive. +- Pipeline-only route: outcome identical to pre-change behavior. +- Plugin-only route: branch 3 still handles it, and the merge does not run. +- An `omit` stage removes a key while a plugin mutates another: the key is gone and the mutation survives. +- Nested dotted field (`args.user.name`) merges without disturbing sibling keys. +- Bare-string args with no structured entity part: text fallback still applies. + +**Verification:** `cargo test -p apl-cpex`, and confirm no existing pipeline test needed an expectation change. Any test whose expectation moves is a finding to explain in the PR, not to silently update. + +--- + +### U6. Field-precise `modified_value` for pipeline stage plugins + +**Goal:** a plugin invoked as a field stage reports a new value for that field, not the message's concatenated text. + +**Requirements:** R7, R8, R9, R10  ·  **Dependencies:** U3 + +**Files:** +- Modify: `crates/apl-core/src/evaluator.rs`, `crates/apl-core/src/step.rs`, `crates/apl-core/src/route.rs` +- Modify: `crates/apl-cpex/src/cmf_invoker.rs` + +**Approach, apl-core side:** +- In `dispatch_field_op`, pass `subpath` (root-relative) to `evaluate_pipeline` instead of the prefixed `path` (`crates/apl-core/src/evaluator.rs:1163`), matching what the section pipelines already pass (`crates/apl-core/src/route.rs:110-155`, `:202-240`). The `path` stays prefixed for deny messages and diagnostics. +- Document the convention on `PluginInvocation::Field.name` (`crates/apl-core/src/step.rs:408-432`): a dotted path relative to the args or result root, with `phase` selecting the root. Document on `PluginOutcome.modified_value` (`:848`) that the value replaces that field only. +- Make `get_dotted` `pub` (`crates/apl-core/src/route.rs:318-329`) so the host bridge reads fields with the same semantics the evaluator writes them. Leave `set_dotted` / `remove_dotted` crate-private; U5's merge does not need them. + +**Approach, apl-cpex side:** +- Replace the `PluginInvocation::Field` arm at `crates/apl-cpex/src/cmf_invoker.rs:336-339`. Project the mutated message per phase (Pre: args, Post: result) using the U3 module, then: + - projection is an object: `get_dotted(&projection, name)`, `Some(value.clone())` when found, `None` when absent. + - projection is a scalar: the projection itself is the field value, so return it. This is the text-shaped-message case and preserves existing behavior. +- On `None`, log a `tracing::debug!` naming the field, so "plugin mutated something other than this field" is observable rather than silent. The payload mutation still propagates through U1 and U2, which is what makes `None` safe here. +- Add the `// Known limitation:` comment described under Scope Boundaries, in plain terms, no doc IDs. + +**Test scenarios:** +- Structured args, the defect: field `city`, a plugin that rewrites `ToolCall.arguments["city"]`, message also carrying text parts. `modified_value` is the new city, not the message text. Fails before this unit. +- Structured args, plugin mutates a different key than the field in focus: `modified_value` is `None` and the pipeline leaves the field alone, while the payload mutation still reaches `modified_payload`. +- Text-shaped message: existing assertions at `crates/apl-cpex/tests/cmf_invoker_dispatch.rs:300-385` pass unchanged. +- Post phase, field on `ToolResult.content`: readback uses the result projection. +- apl-core unit test: a `do:`-block field op on `result.x` reaches the invoker with name `x`, not `result.x`. Assert via a recording `PluginInvoker` test double, the pattern already used at `crates/apl-core/src/evaluator.rs:2637`. +- End to end: a route with `args: city | plugin(scrubber)` where the scrubber redacts the city produces a forwarded payload whose `ToolCall.arguments["city"]` is redacted and whose other arguments are untouched. + +**Verification:** `cargo test -p apl-core -p apl-cpex`. + +--- + +### U7. Docs and changelog + +**Goal:** the accessor's limits are documented, and the fixes are visible to users on 0.2.2. + +**Requirements:** R11  ·  **Dependencies:** none + +**Files:** `crates/cpex-core/src/cmf/message.rs`, `CHANGELOG.md` + +**Approach:** +- Extend the rustdoc on `get_text_content` (`:80`) to state it reads only `Text` parts and is therefore not a change-detection or equality signal for a `Message`. +- Changelog entries under `## [0.2.3] - unreleased`, `### Fixed`, matching the existing entry voice. Name the user-visible symptoms: plugin mutations to non-text content parts silently discarded (worst case, an unredacted tool result forwarded after a redaction plugin reported success); a pipeline edit clobbering a plugin's mutation to the same content part; a field-stage plugin's new field value reported as the message's concatenated text. Reference `#151`. No plan or requirement identifiers. +- Under `### Changed`, note the two behavior shifts operators could observe: requests where a pipeline rewrites a field to its existing value now carry a modified payload, and `PluginInvocation::Field.name` is now root-relative from `do:`-block field ops. + +**Verification:** `make lint` (rustdoc must survive `-D warnings`). + +--- + +## System-Wide Impact + +- `apl-cpex` carries most of the change. `apl-core` gets one line in `dispatch_field_op`, one visibility change, and doc comments. `cpex-core` gets a doc comment. +- No public API removals. `CmfPluginInvoker` gains one method and one private field; `for_request` keeps its signature. `get_dotted` widens from `pub(crate)` to `pub`. +- Behavior visible to hosts: requests where a plugin returned a payload, or where a pipeline wrote a field to its existing value, now carry `modified_payload: Some(..)` where they previously carried `None`. Hosts re-serialize the body in that case, so expect slightly more re-serialization on routes with mutating plugins. +- Behavior visible to plugin authors: a field-stage plugin's `modified_value` is now scoped to the field in focus, and `PluginInvocation::Field.name` is root-relative everywhere. Plugins that ignore `name` and mutate the payload directly are unaffected. +- The FFI path (`crates/cpex-ffi/src/lib.rs:845, 1095`) serializes `modified_payload` when present. Non-Rust plugins were subject to the same drop, since the drop was on the handler side, and are fixed by the same change. + +## Risks + +- **Widest-reaching unit is U5.** It rewrites a merge path that existing args-pipeline tests depend on. The "pipeline-only route is byte-identical" test is the guard, and any existing test whose expectation moves must be explained rather than updated. +- **U6 touches apl-core.** The field-name convention change is only observable to invokers that read `name`; `CmfPluginInvoker` is the only one in-tree. Grep for other `PluginInvocation::Field` consumers before landing. +- **Unconditional flag setting (D3)** means a plugin that always returns an untouched clone marks every request modified. Fail-safe, cost is re-serialization. If profiling later shows it matters, the narrowing move is a cheap comparison inside the invoker, not a return to content diffing in the handler. +- **Atomic ordering** is the only concurrency-sensitive detail. `dispatch_parallel` runs plugin branches on separate tasks; Release/Acquire plus the existing `Mutex` around the payload covers visibility. +- **Scope creep.** U1 and U2 alone close the reported issue. If review pressure builds, land those two and split U3 through U6 into a follow-up PR rather than letting the fix sit. + +## Verification + +``` +cargo test -p apl-core -p apl-cpex +make lint +make test # full workspace +``` + +Manual confirmation of the reported symptom, matching the issue's reproduction: a plugin that redacts only `ToolResult.content` on a `get_weather`-style route, dispatched through `invoke_named::`, yields a `modified_payload` carrying the redacted content with no throwaway `Text` part appended. The workaround in the issue (appending a dummy `Text` part so the text diff fires) becomes unnecessary; call that out in the PR description so the reporter can delete it. + +## Sources & References + +- Issue: https://github.com/contextforge-org/cpex/issues/151, plus the maintainer confirmation comment on `main`. +- `crates/apl-cpex/src/route_handler.rs:454-484` (branch chain), `:712-782` (projections and write-backs), `:784-786` (fail-safe precedent). +- `crates/apl-cpex/src/cmf_invoker.rs:332-341` (mutation acceptance), `:336-339` (field projection), `:76-104` (struct), `:157-166` (accessors). +- `crates/apl-core/src/route.rs:67-86` (`RouteDecision` flags), `:110-155` and `:202-240` (section pipelines), `:318-380` (dotted helpers). +- `crates/apl-core/src/evaluator.rs:1112-1190` (`dispatch_field_op`), `:1442-1460` (`Stage::Plugin`). +- `crates/apl-core/src/step.rs:408-432` (`PluginInvocation`), `:848` (`modified_value`). +- `crates/cpex-core/src/cmf/message.rs:82-92` (`get_text_content`), `crates/cpex-core/src/cmf/content.rs:227-275` (`ContentPart` variants). +- Test patterns: `crates/apl-cpex/tests/cmf_invoker_dispatch.rs:300-385`, `crates/apl-cpex/tests/end_to_end_route.rs:660-760`. diff --git a/go/cpex/manager.go b/go/cpex/manager.go index 00d7fcf3..72f7bdcb 100644 --- a/go/cpex/manager.go +++ b/go/cpex/manager.go @@ -546,9 +546,13 @@ func (m *PluginManager) InvokeResolved( // mgr, "cmf.tool_pre_invoke", cpex.PayloadCMFMessage, // payload, ext, nil, // ) -// if !result.IsDenied() && result.ModifiedPayload != nil { +// if !result.IsDenied() && result.PayloadModified { // fmt.Println(result.ModifiedPayload.Message.Role) // } +// +// ModifiedPayload is non-nil on every allowed pipeline, carrying the +// final payload whether or not a plugin touched it. Test PayloadModified +// to learn whether a plugin actually changed it. func Invoke[P any]( m *PluginManager, hookName string, @@ -568,6 +572,7 @@ func Invoke[P any]( Errors: raw.Errors, Metadata: raw.Metadata, PayloadType: raw.PayloadType, + PayloadModified: raw.PayloadModified, } // Deserialize modified payload if present diff --git a/go/cpex/types.go b/go/cpex/types.go index 04cd0adc..1db1f663 100644 --- a/go/cpex/types.go +++ b/go/cpex/types.go @@ -265,8 +265,15 @@ type PipelineResult struct { Metadata map[string]any `msgpack:"metadata,omitempty"` // Payload type ID — tells the caller how to deserialize ModifiedPayload. PayloadType uint8 `msgpack:"payload_type"` - // Modified payload as raw MessagePack bytes. + // Modified payload as raw MessagePack bytes. Present on every allowed + // pipeline, carrying the final payload whether or not a plugin touched + // it — check PayloadModified to learn whether anything changed. ModifiedPayload []byte `msgpack:"modified_payload,omitempty"` + // PayloadModified reports whether a plugin's payload modification was + // accepted. This is the signal to test; a non-empty ModifiedPayload + // only means the pipeline carried a payload, and comparing payload + // contents cannot see mutations to non-text content parts. + PayloadModified bool `msgpack:"payload_modified"` // Modified extensions as raw MessagePack bytes. ModifiedExtensions []byte `msgpack:"modified_extensions,omitempty"` } @@ -280,6 +287,9 @@ type TypedPipelineResult[P any] struct { Metadata map[string]any PayloadType uint8 ModifiedPayload *P + // PayloadModified reports whether a plugin actually changed the + // payload. See PipelineResult.PayloadModified. + PayloadModified bool ModifiedExtensions *Extensions }