Forward-merge release/2.2 into develop

deploy/Dockerfile

@@ -76,7 +76,7 @@ COPY deploy/entrypoint.py deploy/start_web.py ./deploy/
 # import-fail at runtime. Syncing with the dev group installs those deps from the
 # frozen lock (still pinned, no re-resolve). TODO(deploy): move the source
 # packages to a dedicated non-dev group to keep test-only tooling out of the image.
-RUN uv sync --frozen --extra s3
+RUN uv sync --frozen --extra pii --extra s3
 
 # Install workspace packages (without CLI for base). The root install does not
 # need workspace source overrides because dependencies were synced above; avoid
@@ -90,11 +90,13 @@ RUN uv pip install --no-sources --no-deps -e . \
     && uv pip install --no-deps -e ./frontends/aiq_api \
     && uv pip install "psycopg[binary]>=3.0.0"
 
-RUN /app/.venv/bin/python -c "import aiq_api; import knowledge_layer; print('✓ Base packages installed')"
+RUN /app/.venv/bin/python -c \
+    "import asyncio; import aiq_api; import knowledge_layer; import presidio_analyzer; import presidio_anonymizer; import spacy; from nemoguardrails.library.sensitive_data_detection.actions import mask_sensitive_data; from nemoguardrails.rails.llm.config import RailsConfig; config = RailsConfig.from_content(yaml_content='rails:\n  config:\n    sensitive_data_detection:\n      output:\n        entities:\n          - EMAIL_ADDRESS\n'); assert spacy.util.is_package('en_core_web_lg'); assert asyncio.run(mask_sensitive_data(source='output', text='Contact customer@example.com', config=config)) == 'Contact <EMAIL_ADDRESS>'; print('✓ Base packages and PII runtime installed')"
 
+# Keep immutable application assets root-owned; only runtime data is writable by the service user.
 RUN chmod +x /app/deploy/start_web.py \
     && mkdir -p /app/data \
-    && chown -R 1000:1000 /app
+    && chown 1000:1000 /app/data
 
 # =============================================================================
 # Stage 2: Development (includes CLI)

docs/source/customization/guardrails.md

@@ -34,8 +34,20 @@ At each configured boundary, guardrails can make one of three decisions:
 | Modify | Replace the selected input or output field with the modified content returned by the rail. |
 | Block | Return the configured refusal response instead of continuing with the blocked content. |
 
-Input-rail evaluation exceptions are caught, logged, and converted to the middleware refusal response. Output-rail
-evaluation exceptions are not converted to a refusal; they propagate and fail the invocation.
+Input- and output-rail evaluation exceptions are caught, logged, and converted to the middleware refusal response.
+Output failures preserve the intercepted response schema and do not return the original unfiltered output.
+
+Buffered output streams are evaluated as one logical assistant response before any chunk is emitted. This includes
+streams that mix raw strings and structured response chunks. Modified output is redistributed across the buffered
+chunks, terminal workflow outcomes are synchronized with every rewritten structured chunk, and blocked or failed
+streams emit only a safe refusal.
+
+## PII Runtime Dependencies
+
+The built-in `sensitive_data_detection` action requires Presidio, spaCy, and a compatible spaCy language model. These
+large dependencies are available through the `pii` project extra rather than the base Python package. Install AI-Q with
+`--extra pii` when running PII rails. The release Docker image includes this extra and verifies during the build that the
+Presidio analyzer and anonymizer import, `en_core_web_lg` is installed, and email analysis succeeds.
 
 ## Configuration Shape
 

mcp/tests/test_config_and_packaging.py

@@ -150,6 +150,16 @@ def test_root_workspace_excludes_the_independent_mcp_project() -> None:
         if "registry" in source:
             assert source == {"registry": "https://pypi.org/simple"}
             continue
+        if "url" in source:
+            assert package["name"] == "en-core-web-lg"
+            assert package["version"] == "3.8.0"
+            assert source == {
+                "url": (
+                    "https://github.com/explosion/spacy-models/releases/download/"
+                    "en_core_web_lg-3.8.0/en_core_web_lg-3.8.0-py3-none-any.whl"
+                )
+            }
+            continue
         editable = source.get("editable")
         assert isinstance(editable, str)
         assert not Path(editable).is_absolute()

pyproject.toml

@@ -53,6 +53,10 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+pii = [
+    "nvidia-nat-security[defense]==1.8.0",
+    "en-core-web-lg @ https://github.com/explosion/spacy-models/releases/download/en_core_web_lg-3.8.0/en_core_web_lg-3.8.0-py3-none-any.whl",
+]
 s3 = [
     "boto3>=1.35.0,<2",
 ]

src/aiq_agent/guardrails/deep_agent/middleware.py

@@ -25,6 +25,7 @@
 
 from aiq_agent.agents.deep_researcher.models import DeepResearchAgentState
 from aiq_agent.guardrails.deep_agent.config import DeepAgentGuardrailsConfig
+from aiq_agent.guardrails.interface.middleware import _GUARDRAILS_FAILURE_REFUSAL
 from aiq_agent.guardrails.interface.middleware import GuardrailsMixin
 from nat.builder.builder import Builder
 from nat.middleware.middleware import InvocationContext
@@ -134,6 +135,14 @@ def _on_post_invoke_blocked(
         """Replace blocked deep-agent output content with the refusal."""
         return super()._on_post_invoke_blocked(context, block_message, original_output)
 
+    def _build_emergency_output_refusal(
+        self,
+        context: InvocationContext,
+        original_output: object,
+    ) -> DeepResearchAgentState:
+        """Return a minimal refusal state without traversing protected output."""
+        return DeepResearchAgentState(messages=[AIMessage(content=_GUARDRAILS_FAILURE_REFUSAL)])
+
     def _iter_targets_at_path(self, value: Any, path: str) -> Iterator[tuple[str, Callable[[str], None]]]:
         """Yield the latest deep-agent message for the inherited field-selection hook."""
         selected_message = self._extract_latest_message_text(value, path.split("."))

src/aiq_agent/guardrails/interface/middleware.py

@@ -49,8 +49,11 @@ async def pre_invoke(self, context: InvocationContext) -> InvocationContext | No
         """Run input rails and adapt blocked outputs for the intercepted boundary."""
         try:
             result = await super().pre_invoke(context)
-        except Exception:
-            logger.exception("Input Guardrails failed while evaluating selected fields; refusing request")
+        except Exception as exc:
+            logger.error(
+                "Input Guardrails failed while evaluating selected fields; refusing request; error_type=%s",
+                type(exc).__name__,
+            )
             context.output = self._on_pre_invoke_blocked(context, _GUARDRAILS_FAILURE_REFUSAL)
             return context
 
@@ -67,7 +70,20 @@ def _on_pre_invoke_blocked(self, context: InvocationContext, block_message: str)
 
     async def post_invoke(self, context: InvocationContext) -> InvocationContext | None:
         """Run output rails and adapt blocked outputs for the intercepted boundary."""
-        return await super().post_invoke(context)
+        try:
+            result = await super().post_invoke(context)
+            if result is not None:
+                self._synchronize_terminal_output(context)
+            return result
+        except Exception as exc:
+            logger.error(
+                "Output Guardrails failed while evaluating selected fields for function %s; "
+                "refusing response; error_type=%s",
+                context.function_context.name,
+                type(exc).__name__,
+            )
+            context.output = self._refuse_output_safely(context, context.output)
+            return context
 
     async def function_middleware_stream(
         self,
@@ -107,7 +123,18 @@ async def function_middleware_stream(
             yield ctx.output
             return
 
-        output, blocked = await self._apply_output_rails_to_structured_stream(ctx, buffered)
+        try:
+            output, blocked = await self._apply_output_rails_to_structured_stream(ctx, buffered)
+        except Exception as exc:
+            logger.error(
+                "Output Guardrails failed while evaluating buffered structured output for function %s; "
+                "refusing response; error_type=%s",
+                context.name,
+                type(exc).__name__,
+            )
+            ctx.output = self._refuse_output_safely(ctx, buffered[0])
+            yield ctx.output
+            return
         if blocked:
             yield output
             return
@@ -134,7 +161,7 @@ async def _apply_output_rails_to_structured_stream(
             input_text = getattr(raw, "input_message", None) or (raw if isinstance(raw, str) else str(raw))
 
         paths = self._resolve_guarded_targets_for_phase(context.function_context.name, "post_invoke")
-        selections = self._structured_stream_output_selections(buffered, paths)
+        selections = self._stream_output_selections(buffered, paths)
         if not selections:
             return buffered[0], False
 
@@ -161,20 +188,30 @@ async def _apply_output_rails_to_structured_stream(
             selections[0][1](result_text)
             for _text, apply_to_field in selections[1:]:
                 apply_to_field("")
+            self._synchronize_buffered_outputs(buffered, paths)
 
         return buffered[0], False
 
-    def _structured_stream_output_selections(
+    def _stream_output_selections(
         self,
         buffered: list[object],
         paths: list[str],
     ) -> list[tuple[str, Callable[[str], None]]]:
-        """Return selected text fields from buffered structured stream chunks."""
+        """Return writable text selections from string and structured stream chunks."""
         selections: list[tuple[str, Callable[[str], None]]] = []
-        for chunk in buffered:
-            selections.extend(self._gather_guardrail_inputs(chunk, paths, lambda _value: None))
+        for index, chunk in enumerate(buffered):
+            if isinstance(chunk, str):
+                selections.append((chunk, lambda value, index=index: buffered.__setitem__(index, value)))
+            else:
+                selections.extend(self._gather_guardrail_inputs(chunk, paths, lambda _value: None))
         return selections
 
+    def _synchronize_buffered_outputs(self, buffered: list[object], paths: list[str]) -> None:
+        """Let a concrete boundary align auxiliary fields with rewritten public output."""
+
+    def _synchronize_terminal_output(self, context: InvocationContext) -> None:
+        """Let a concrete boundary align auxiliary fields with guarded public output."""
+
     def on_post_invoke_blocked(self, context: InvocationContext, block_message: str) -> object:
         """Adapt blocked output before the intercepted result is returned."""
         return self._on_post_invoke_blocked(context, block_message, context.output)
@@ -188,7 +225,32 @@ def _on_post_invoke_blocked(
         """Adapt output-rail block output for the intercepted boundary."""
         if not isinstance(original_output, str):
             paths = self._resolve_guarded_targets_for_phase(context.function_context.name, "post_invoke")
+            modified = False
             for _text, apply_to_field in self._gather_guardrail_inputs(original_output, paths, lambda _value: None):
                 apply_to_field(block_message)
+                modified = True
+            if modified:
+                self._synchronize_blocked_output(original_output, block_message)
                 return original_output
+            return self._build_emergency_output_refusal(context, original_output)
         return block_message
+
+    def _synchronize_blocked_output(self, output: object, block_message: str) -> None:
+        """Let a concrete boundary align auxiliary fields with a blocked public output."""
+
+    def _refuse_output_safely(self, context: InvocationContext, original_output: object) -> object:
+        """Adapt a refusal without allowing a failing target traversal to escape."""
+        try:
+            return self._on_post_invoke_blocked(context, _GUARDRAILS_FAILURE_REFUSAL, original_output)
+        except Exception as exc:
+            logger.error(
+                "Output Guardrails failed while adapting refusal for function %s; "
+                "using emergency refusal; error_type=%s",
+                context.function_context.name,
+                type(exc).__name__,
+            )
+            return self._build_emergency_output_refusal(context, original_output)
+
+    def _build_emergency_output_refusal(self, context: InvocationContext, original_output: object) -> object:
+        """Return the traversal-independent refusal for this intercepted boundary."""
+        return _GUARDRAILS_FAILURE_REFUSAL

src/aiq_agent/guardrails/shallow_agent/middleware.py

@@ -24,6 +24,7 @@
 from langchain_core.messages import AIMessage
 
 from aiq_agent.agents.shallow_researcher.models import ShallowResearchAgentState
+from aiq_agent.guardrails.interface.middleware import _GUARDRAILS_FAILURE_REFUSAL
 from aiq_agent.guardrails.interface.middleware import GuardrailsMixin
 from aiq_agent.guardrails.shallow_agent.config import ShallowAgentGuardrailsConfig
 from nat.builder.builder import Builder
@@ -134,6 +135,14 @@ def _on_post_invoke_blocked(
         """Replace blocked shallow-agent output content with the refusal."""
         return super()._on_post_invoke_blocked(context, block_message, original_output)
 
+    def _build_emergency_output_refusal(
+        self,
+        context: InvocationContext,
+        original_output: object,
+    ) -> ShallowResearchAgentState:
+        """Return a minimal refusal state without traversing protected output."""
+        return ShallowResearchAgentState(messages=[AIMessage(content=_GUARDRAILS_FAILURE_REFUSAL)])
+
     def _iter_targets_at_path(self, value: Any, path: str) -> Iterator[tuple[str, Callable[[str], None]]]:
         """Yield the latest shallow-agent message for the inherited field-selection hook."""
         selected_message = self._extract_latest_message_text(value, path.split("."))