From 927e693b5787e7bbe28b89fc2067a14848be3225 Mon Sep 17 00:00:00 2001 From: MohammadHaroonAbuomar <40180927+MohammadHaroonAbuomar@users.noreply.github.com> Date: Fri, 7 Aug 2026 05:59:31 +0000 Subject: [PATCH] test(engine): port MAF review attack shapes into stream session tests Three shapes from the MAF agent-hooks review rounds, pinned against the section 18.1 accounting (analysis/streaming-acs-vs-maf.md, R9): - Retry re-invocation across sessions: an abandoned attempt settles failed on its uncleared residue while the clean resumed retry settles clean, and a retry that replays the abandoned attempt's clearance frontier fails closed rather than releasing a gap nothing evaluated. - Success-then-discard: a fully cleared, never emitted stream settles Complete, because clearance is permission and not a delivery record; the companion test pins that the accounting cannot distinguish an emitting host from a discarding one, which names the host obligation boundary instead of pretending to check it. - Streaming versus whole-snapshot asymmetry probe: deny-bearing content reaches the same terminal outcome through one section 18 whole snapshot and through section 18.1 segments, with the incremental path differing only in the bounded exposure of the cleared prefix; a clean stream is the negative control and lands clean both ways. Complements the resume-boundary straddle pair in PR #30; tests live in a new file so the two branches do not contend. Signed-off-by: MohammadHaroonAbuomar <40180927+MohammadHaroonAbuomar@users.noreply.github.com> --- engine/tests/stream_session_attack_shapes.rs | 425 +++++++++++++++++++ 1 file changed, 425 insertions(+) create mode 100644 engine/tests/stream_session_attack_shapes.rs diff --git a/engine/tests/stream_session_attack_shapes.rs b/engine/tests/stream_session_attack_shapes.rs new file mode 100644 index 0000000..f755a05 --- /dev/null +++ b/engine/tests/stream_session_attack_shapes.rs @@ -0,0 +1,425 @@ +//! Attack shapes ported from the MAF agent-hooks review. +//! +//! The MAF review rounds (see `analysis/streaming-acs-vs-maf.md`, R9) found +//! three shapes that defeated a first implementation of buffered gating: a +//! retry middleware re-invoking the stream so a second attempt escapes the +//! first attempt's verdict, a middleware draining a successful attempt and +//! discarding it, and divergence between the streamed and the non streamed +//! evaluation of the same content. These tests pin how the section 18.1 +//! accounting behaves under each shape. The session holds no stream text, so +//! where a guarantee is a host obligation rather than something the +//! accounting can check, the test asserts what the session can assert and +//! the comment names the boundary. +//! +//! Conventions follow `stream_session_mediation.rs`: a real `Runtime`, a +//! trigger policy that denies any target containing a banned term, cumulative +//! evaluation as section 18.1 obliges, and a negative control wherever the +//! shape has one. + +#![cfg(feature = "streaming")] + +use agent_control_spec::{ + AnnotatorDispatcher, AnnotatorInvocation, JsonValue, Manifest, PolicyDispatcher, + PreparedPolicyInvocation, Runtime, RuntimeError, SafetyLevel, SegmentOutcome, StreamEndReason, + StreamError, StreamSession, StreamSessionConfig, StreamSourceType, StreamSpan, StreamTrack, + Verdict, +}; +use serde_json::json; +use std::sync::Arc; + +const MANIFEST: &str = r#"agent_control_specification_version: 0.4.0-alpha.1 +policies: + p: + type: test +intervention_points: + post_model_call: + policy_target: $snap.completion + policy: + id: p +"#; + +/// Denies any policy target whose text contains the trigger substring. +struct TriggerPolicy { + trigger: &'static str, +} + +impl PolicyDispatcher for TriggerPolicy { + fn evaluate(&self, invocation: &PreparedPolicyInvocation) -> Result { + let rendered = invocation + .policy_input() + .map(JsonValue::to_string) + .unwrap_or_default(); + if rendered.contains(self.trigger) { + Ok(json!({"decision": "deny", "reason": "trigger"})) + } else { + Ok(json!({"decision": "allow"})) + } + } +} + +struct NoAnnotators; + +impl AnnotatorDispatcher for NoAnnotators { + fn dispatch( + &self, + annotator_name: &str, + _invocation: &AnnotatorInvocation, + _preliminary_policy_input: &JsonValue, + ) -> Result { + Err(RuntimeError::AnnotationFailed(annotator_name.to_string())) + } +} + +fn runtime(trigger: &'static str) -> Runtime { + let manifest = Manifest::from_yaml_str(MANIFEST).unwrap(); + Runtime::new( + manifest, + Arc::new(NoAnnotators), + Arc::new(TriggerPolicy { trigger }), + ) + .unwrap() +} + +/// Evaluate one policy target as an ordinary whole snapshot, which is the +/// section 18 path: one intervention point evaluation over the whole value. +fn evaluate(runtime: &Runtime, policy_target: &str) -> Verdict { + runtime + .evaluate_point( + StreamSourceType::ModelGenerated.interception_point(), + json!({"completion": {"text": policy_target}}), + ) + .verdict +} + +fn slice_runes(text: &str, start: u32, end: u32) -> String { + text.chars() + .skip(start as usize) + .take(end.saturating_sub(start) as usize) + .collect() +} + +/// A response only session over `tasks`, resuming at `resume_at`. +fn response_session(tasks: &[&str], resume_at: u32) -> StreamSession { + StreamSession::new(StreamSessionConfig { + safety_level: SafetyLevel::Blocking, + request_start_rune_offset: 0, + response_start_rune_offset: resume_at, + request_tasks: Vec::new(), + response_tasks: tasks.iter().map(|task| (*task).to_string()).collect(), + }) + .expect("config is valid") +} + +/// What one incrementally mediated run of `payloads` concluded. +struct IncrementalRun { + session: StreamSession, + /// Text the host emitted to the caller as the watermark advanced. Empty + /// when `emit` was false, because delivery is the host's act and not the + /// session's. + delivered: String, +} + +/// Drive a fresh single task session over `payloads`, evaluating each span +/// over the whole accumulated prefix as section 18.1 obliges. `emit` selects +/// whether the host delivers the cleared prefix as the watermark advances or +/// drains the stream without releasing anything, which is the discard shape. +fn drive_incremental(runtime: &Runtime, payloads: &[&str], emit: bool) -> IncrementalRun { + let mut session = response_session(&["harm"], 0); + let mut text = String::new(); + let mut delivered = String::new(); + let mut cursor = 0; + let mut emitted = 0; + for payload in payloads { + text.push_str(payload); + let end = session + .observe_text(StreamSourceType::ModelGenerated, payload) + .expect("observe"); + let span = + StreamSpan::new(StreamSourceType::ModelGenerated, cursor, end).expect("valid span"); + let verdict = evaluate(runtime, &text); + session + .record_verdict("harm", &span, &verdict) + .expect("the outcome records, a denial included"); + cursor = end; + if session.is_ended() { + break; + } + if let Some(safe) = session.advance(StreamTrack::Response) { + if emit { + delivered.push_str(&slice_runes(&text, emitted, safe)); + emitted = safe; + } + } + } + IncrementalRun { session, delivered } +} + +// --------------------------------------------------------------------------- +// Shape (a): retry re-invocation across sessions. +// +// MAF's finding was retry middleware re-invoking the stream so that a second +// attempt escaped the first attempt's verdict coverage. The ACS translation +// is two sessions over one caller visible stream: attempt one is abandoned +// mid stream, attempt two resumes at the offset the caller had reached. The +// accounting must keep the attempts isolated: the abandoned session settles +// on its own residue, the clean retry settles clean, and neither session's +// watermark leaks into the other. +// --------------------------------------------------------------------------- + +#[test] +fn an_abandoned_attempt_settles_failed_and_its_retry_settles_clean() { + let runtime = runtime("forbidden"); + + // Attempt one: two tasks mediate the response. Both clear the first + // payload, so "clean " is released. The second payload arrives and only + // one task clears it before the caller disconnects and the host abandons + // the attempt. + let mut first = response_session(&["harm", "pii"], 0); + let mut text = String::new(); + text.push_str("clean "); + first + .observe_text(StreamSourceType::ModelGenerated, "clean ") + .expect("observe"); + let span = StreamSpan::new(StreamSourceType::ModelGenerated, 0, 6).expect("valid span"); + first + .record_verdict("harm", &span, &evaluate(&runtime, &text)) + .expect("harm clears"); + first + .record_outcome("pii", &span, SegmentOutcome::Cleared) + .expect("pii clears"); + assert_eq!(first.advance(StreamTrack::Response), Some(6)); + let released_by_first = "clean "; + + text.push_str("and mo"); + first + .observe_text(StreamSourceType::ModelGenerated, "and mo") + .expect("observe"); + let span = StreamSpan::new(StreamSourceType::ModelGenerated, 6, 12).expect("valid span"); + first + .record_verdict("harm", &span, &evaluate(&runtime, &text)) + .expect("harm clears the tail"); + // pii never reports on [6, 12): the host abandons the attempt here. + + // Section 18.1: every session that is opened settles, an abandoned one + // included. The residue no task cleared fails the settlement closed; it + // is not silently dropped with the attempt. + let completion = first.finish(); + assert_eq!( + completion.reason, + StreamEndReason::Failed(StreamError::UnclearedResidue { + track: StreamTrack::Response, + pending: 6, + }) + ); + // The offsets stay readable for the audit record, and the failing + // settlement did not raise the release point over the residue. + assert_eq!(first.watermark(StreamTrack::Response).confirmed(), 6); + assert_eq!(first.watermark(StreamTrack::Response).received(), 12); + + // Attempt two resumes at the offset the caller actually received, not at + // the offset the abandoned attempt had evaluated. The retry re-generates + // from there; the abandoned residue "and mo" is gone, not inherited. The + // host retains the released tail and evaluates it under each value, per + // the retention obligation on resumed tracks. + let resume_at = 6; + let mut second = response_session(&["harm", "pii"], resume_at); + assert_eq!( + second.watermark(StreamTrack::Response).confirmed(), + resume_at, + "the retry's watermark starts at the resume offset, clean of the first attempt's state" + ); + let attempt = "fresh tail"; + let end = second + .observe_text(StreamSourceType::ModelGenerated, attempt) + .expect("observe"); + let span = + StreamSpan::new(StreamSourceType::ModelGenerated, resume_at, end).expect("valid span"); + let value = format!("{released_by_first}{attempt}"); + second + .record_verdict("harm", &span, &evaluate(&runtime, &value)) + .expect("harm clears"); + second + .record_outcome("pii", &span, SegmentOutcome::Cleared) + .expect("pii clears"); + assert_eq!(second.advance(StreamTrack::Response), Some(16)); + + // The clean retry settles clean: the abandoned attempt's failure does not + // bleed into it, and its own settlement claims nothing about attempt one. + assert_eq!(second.finish().reason, StreamEndReason::Complete); + let caller_sees = format!("{released_by_first}{attempt}"); + assert!( + !caller_sees.contains("and mo"), + "the abandoned attempt's uncleared residue never reached the caller" + ); +} + +#[test] +fn a_retry_cannot_inherit_the_abandoned_attempts_clearance_frontier() { + // The negative control for cross session watermark bleed. Attempt one + // cleared through rune 12 but the host had only emitted through rune 6 + // when it abandoned the attempt, so the retry resumes at 6. A host that + // treats the abandoned attempt's clearance frontier as carried over + // records its first retry span starting at 12, as if [6, 12) were still + // cleared. Clearance is per session: nothing in the retry evaluated + // [6, 12) of the retry's own text, so confirming that span would release + // a gap nothing evaluated, and the accounting fails it closed. + let mut first = response_session(&["harm"], 0); + first + .observe_text(StreamSourceType::ModelGenerated, "clean and mo") + .expect("observe"); + let span = StreamSpan::new(StreamSourceType::ModelGenerated, 0, 12).expect("valid span"); + first + .record_outcome("harm", &span, SegmentOutcome::Cleared) + .expect("harm clears through 12"); + assert_eq!(first.advance(StreamTrack::Response), Some(12)); + // The host emitted only through 6 before abandoning; delivery is the + // host's act, so the resume offset below is 6 regardless of the frontier. + + let mut second = response_session(&["harm"], 6); + second + .observe_text(StreamSourceType::ModelGenerated, "fresh tail") + .expect("observe"); + let stale_frontier = StreamSpan::new(StreamSourceType::ModelGenerated, 12, 16).expect("valid"); + assert_eq!( + second.record_outcome("harm", &stale_frontier, SegmentOutcome::Cleared), + Err(StreamError::NonContiguousOutcome { + task: "harm".to_string(), + expected: 6, + found: 12, + }), + "the first attempt's frontier does not clear the retry's gap" + ); + assert!(matches!( + second.end_reason(), + Some(StreamEndReason::Failed( + StreamError::NonContiguousOutcome { .. } + )) + )); +} + +// --------------------------------------------------------------------------- +// Shape (b): success-then-discard. +// +// MAF's shape was middleware draining a successful attempt, discarding it, +// and returning something else, with the drained attempt's side effects +// escaping the covering verdict. In the 18.1 topology a host can drain a +// stream through the accounting, clear every rune, and then abandon the +// response without releasing any of it. +// --------------------------------------------------------------------------- + +#[test] +fn a_fully_cleared_stream_the_host_never_emits_still_settles_complete() { + // The session records clearance, which is permission, not a delivery + // record. A fully cleared stream settles `Complete` whether or not the + // host ever emitted a rune, because `Complete` says "everything was + // evaluated and releasable" and claims nothing about delivery. + // + // Host obligation boundary: the session holds no text and no record of + // what the host actually wrote to its caller, so `safe_offset` against + // what was actually emitted is host side state the accounting cannot + // audit. What the session CAN assert is pinned here: the settlement is + // clean, the cleared extent is on the watermark for the audit record, + // and the settled session offers no release point afterwards, so a host + // that discarded the response cannot later hold up the settled session + // as live permission to emit the drained text. + let runtime = runtime("forbidden"); + let mut run = drive_incremental(&runtime, &["the ", "quick ", "brown "], false); + assert_eq!(run.delivered, "", "the host drained without releasing"); + let completion = run.session.finish(); + assert!(completion.reason.is_clean()); + assert!(!completion.transformed); + assert_eq!( + run.session.watermark(StreamTrack::Response).confirmed(), + 16, + "the cleared extent survives for the audit record" + ); + assert_eq!( + run.session.safe_offset(StreamTrack::Response), + None, + "a settled session offers no release point to emit the discarded text against" + ); + assert_eq!( + run.session + .observe_text(StreamSourceType::ModelGenerated, "more"), + Err(StreamError::SessionClosed), + "and it takes no further payload" + ); +} + +#[test] +fn settlement_records_clearance_not_delivery() { + // The boundary made concrete: two identical streams, one host emits every + // cleared prefix, the other drains and discards. The accounting cannot + // distinguish them, and must not pretend to: their settlements and their + // watermarks are identical, so any claim that discarded text was or was + // not delivered is the host's to make and the host's to prove. This is + // the file's convention of a negative control, applied to an assertion + // the session deliberately does not make. + let runtime = runtime("forbidden"); + let payloads = ["the ", "quick ", "brown "]; + let mut emitting = drive_incremental(&runtime, &payloads, true); + let mut discarding = drive_incremental(&runtime, &payloads, false); + assert_eq!(emitting.delivered, "the quick brown "); + assert_eq!(discarding.delivered, ""); + assert_eq!(emitting.session.finish(), discarding.session.finish()); + assert_eq!( + emitting.session.watermark(StreamTrack::Response), + discarding.session.watermark(StreamTrack::Response), + "the accounting carries no delivery record to tell the two hosts apart" + ); +} + +// --------------------------------------------------------------------------- +// Shape (c): streaming / whole-snapshot asymmetry probe. +// +// The metamorphic suite already pins that the incremental outcome does not +// depend on segmentation. This probe pins the cross path property: the same +// content evaluated as one section 18 whole snapshot and evaluated through +// section 18.1 segments reaches the same terminal outcome. The incremental +// path differs only in exposure, because a cleared prefix is released before +// the denial lands, which is the bounded exposure the profile trades for +// latency and not a divergent verdict. +// --------------------------------------------------------------------------- + +#[test] +fn a_deny_lands_on_both_the_whole_snapshot_and_the_incremental_path() { + let runtime = runtime("forbidden"); + let content = "clean forbidden text"; + + // Section 18 path: one evaluation over the whole assembled value. + let whole = evaluate(&runtime, content); + assert_eq!(whole.decision.as_str(), "deny"); + + // Section 18.1 path: the same runes, cut into payloads none of which + // contains the term on its own, evaluated cumulatively. + let mut run = drive_incremental(&runtime, &["clean ", "forbid", "den text"], true); + assert!(matches!( + run.session.end_reason(), + Some(StreamEndReason::Denied { .. }) + )); + assert!(!run.session.finish().reason.is_clean()); + + // Same terminal outcome; the paths differ only in exposure. The whole + // snapshot path emits nothing of a refused value, which is the host's + // act under section 18. The incremental path had already released the + // prefix cleared before the term completed, and nothing more: "forbid" + // cleared because no cumulative value containing it held the whole term + // yet, which is precisely the bounded exposure the profile documents. + assert_eq!(run.delivered, "clean forbid"); + assert!(!run.delivered.contains("forbidden")); +} + +#[test] +fn a_clean_stream_settles_the_same_on_both_paths() { + // The negative control: content the policy allows must land clean both + // ways, with the incremental path releasing everything it cleared. + let runtime = runtime("forbidden"); + let content = "clean text throughout"; + + let whole = evaluate(&runtime, content); + assert_eq!(whole.decision.as_str(), "allow"); + + let mut run = drive_incremental(&runtime, &["clean ", "text t", "hroughout"], true); + assert_eq!(run.session.finish().reason, StreamEndReason::Complete); + assert_eq!(run.delivered, content); +}