GuardrailsEngine.stream() and stream_full() bypass configured input scanning

Description

GuardrailsEngine.generate() scans input messages when scan_input=True, but GuardrailsEngine.stream() and GuardrailsEngine.stream_full() pass the original messages directly to the wrapped engine.

This causes streaming and non-streaming requests using the same guardrail configuration to behave differently:

• In BLOCK mode, generate() raises SecurityBlockError before entering the wrapped engine, while both streaming methods forward the sensitive input.
• In REDACT mode, generate() sends redacted input to the wrapped engine, while both streaming methods forward the original input.
• The streaming methods also appear to skip input-side WARN events for the same reason.

This report concerns input scanning before inference begins. It does not propose changing the existing post-hoc handling of model output during streaming.

Documentation evidence

The current behavior appears to contradict the documented streaming sequence.

The security architecture documentation shows this order:

GuardrailsEngine.stream(messages, model)
→ scan inputs before streaming
→ wrapped_engine.stream(messages, model)
→ yield tokens
→ scan accumulated output after the stream

The sequence diagram explicitly places:

G -> S: scan inputs (before streaming)
G -> E: stream(messages, model)

before any tokens are yielded.

The same section explains:

Because the scan is post-hoc, BLOCK mode cannot prevent delivery of streamed tokens — it only applies to the input side.

I understand “streamed tokens” here to refer to model output that has already been yielded to the caller. It does not prevent blocking the complete input, which is available before the wrapped engine begins inference.

The security user guide reinforces that distinction:

BLOCK mode only applies to the input side during streaming.

It also states:

SecurityBlockError can only be raised before the stream starts (for input scanning).

In Python async-generator terms, this would occur when the returned iterator is first advanced, before the wrapped engine receives the messages or yields its first result.

The user guide also documents scan_input as:

Whether to scan input messages

with a default of True, without limiting it to generate().

The documentation explicitly names stream(). I believe the same contract should also apply to stream_full() because it is the structured streaming variant of the same InferenceEngine interface, accepts the same input messages, and is overridden by GuardrailsEngine to provide streaming guardrail behavior. However, I would welcome maintainer confirmation on the intended stream_full() scope.

Implementation evidence

The current generate() implementation:

1. Copies the message sequence.
2. Scans each message containing text.
3. Applies _handle_findings().
4. Passes the resulting messages to the wrapped engine.

The current stream() implementation calls the wrapped engine directly with the original messages. It does not inspect _scan_input.

The current stream_full() implementation does the same.

Existing tests cover input blocking and redaction through generate() and streaming output scanning, but I could not find coverage for streaming input scanning.

Proposed Solution

The narrowest fix may be to extract the input-message processing currently implemented inside generate() into a private helper and invoke it from:

• generate()
• stream()
• stream_full()

The helper should preserve the current generate() behavior:

• Respect scan_input=False.
• Scan every message with textual content.
• Preserve all non-content Message fields.
• Create replacement messages when redaction is required.
• Avoid mutating the caller’s original message objects.
• Publish the existing input security events.
• Raise SecurityBlockError before the wrapped engine is entered in BLOCK mode.

The existing post-hoc streaming output behavior can remain unchanged.

Suggested regression tests:

1. Parameterize tests across stream() and stream_full().
2. Verify that BLOCK raises before the wrapped engine is entered.
3. Verify that REDACT sends sanitized input to the wrapped engine.
4. Verify input WARN events.
5. Verify that scan_input=False retains pass-through behavior.
6. Verify that the original messages and all non-content fields remain unchanged.

One compatibility consideration is that streaming callers using the default scan_input=True would begin receiving the already documented WARN, REDACT, or BLOCK behavior.

If maintainers confirm that this is unintended and agree with the proposed scope, I would be happy to comment take, implement the focused fix, and submit a PR with regression tests. I am open to adjusting the implementation approach, particularly if stream_full() is intended to have a different contract.

Steps to Reproduce

The following reproduction compares all three methods using the same sensitive input and an in-memory recording engine.

It performs no real inference and makes no network requests. scan_output=False is intentional: it isolates input handling from the documented post-hoc limitations of streaming output scanning.

import asyncio

from openjarvis.core.types import Message, Role
from openjarvis.engine._stubs import StreamChunk
from openjarvis.security.guardrails import (
    GuardrailsEngine,
    SecurityBlockError,
)
from openjarvis.security.types import RedactionMode


SECRET = "my key sk-abc123def456ghi789jkl012"
MESSAGES = [Message(role=Role.USER, content=SECRET)]


class RecordingEngine:
    """Minimal fake backend that records every request it receives."""

    engine_id = "recording"

    def __init__(self):
        self.calls = []

    def generate(self, messages, **kwargs):
        self.calls.append(
            ("generate", [message.content for message in messages])
        )
        return {"content": "ok"}

    async def stream(self, messages, **kwargs):
        self.calls.append(
            ("stream", [message.content for message in messages])
        )
        yield "ok"

    async def stream_full(self, messages, **kwargs):
        self.calls.append(
            ("stream_full", [message.content for message in messages])
        )
        yield StreamChunk(content="ok")
        yield StreamChunk(finish_reason="stop")


def make_guarded_engine(mode):
    backend = RecordingEngine()
    guarded = GuardrailsEngine(
        backend,
        mode=mode,
        scan_input=True,
        scan_output=False,
    )
    return guarded, backend


def exercise_generate(mode):
    guarded, backend = make_guarded_engine(mode)

    try:
        result = guarded.generate(MESSAGES, model="test")
        status = f"ALLOWED: {result}"
    except SecurityBlockError as exc:
        status = f"BLOCKED: {exc}"

    print("generate()")
    print("  result:", status)
    print("  backend calls:", backend.calls)


async def exercise_stream(mode):
    guarded, backend = make_guarded_engine(mode)

    try:
        tokens = [
            token
            async for token in guarded.stream(MESSAGES, model="test")
        ]
        status = f"ALLOWED: {tokens}"
    except SecurityBlockError as exc:
        status = f"BLOCKED: {exc}"

    print("stream()")
    print("  result:", status)
    print("  backend calls:", backend.calls)


async def exercise_stream_full(mode):
    guarded, backend = make_guarded_engine(mode)

    try:
        chunks = [
            {
                "content": chunk.content,
                "finish_reason": chunk.finish_reason,
            }
            async for chunk in guarded.stream_full(
                MESSAGES,
                model="test",
            )
        ]
        status = f"ALLOWED: {chunks}"
    except SecurityBlockError as exc:
        status = f"BLOCKED: {exc}"

    print("stream_full()")
    print("  result:", status)
    print("  backend calls:", backend.calls)


async def main():
    for mode in (RedactionMode.BLOCK, RedactionMode.REDACT):
        print(f"\n=== {mode.value.upper()} ===")
        exercise_generate(mode)
        await exercise_stream(mode)
        await exercise_stream_full(mode)


asyncio.run(main())

Run with:

uv run python scratch/reproduce_guardrails_stream_input.py

Expected Behavior

With scan_input=True, all three entry points should apply the configured input policy before invoking their corresponding wrapped-engine method.

BLOCK mode

• generate(), stream(), and stream_full() should raise SecurityBlockError.
• The recording engine’s call list should remain empty.
• For async streaming methods, the exception should occur when iteration begins, before entering the wrapped engine or yielding output.

REDACT mode

• All three methods should invoke the recording engine.
• The recording engine should receive:

my key [REDACTED:openai_key]

• The caller’s original Message should remain unchanged.

The documented post-hoc limitation for streaming model output should remain unchanged.

Actual Behavior

BLOCK mode

generate()
  result: BLOCKED: Security scan blocked input: 1 finding(s) detected
  backend calls: []

stream()
  result: ALLOWED: ['ok']
  backend calls: [('stream', ['my key sk-abc123def456ghi789jkl012'])]

stream_full()
  result: ALLOWED: [
      {'content': 'ok', 'finish_reason': None},
      {'content': None, 'finish_reason': 'stop'}
  ]
  backend calls: [
      ('stream_full', ['my key sk-abc123def456ghi789jkl012'])
  ]

REDACT mode

generate()
  result: ALLOWED: {'content': 'ok'}
  backend calls: [
      ('generate', ['my key [REDACTED:openai_key]'])
  ]

stream()
  result: ALLOWED: ['ok']
  backend calls: [
      ('stream', ['my key sk-abc123def456ghi789jkl012'])
  ]

stream_full()
  result: ALLOWED: [
      {'content': 'ok', 'finish_reason': None},
      {'content': None, 'finish_reason': 'stop'}
  ]
  backend calls: [
      ('stream_full', ['my key sk-abc123def456ghi789jkl012'])
  ]

The streaming methods therefore forward the original sensitive input despite scan_input=True.

Operating System

Windows

Python Version

3.12

Hardware

CPU only

Engine

Other

Logs / Traceback