fix(cli): route session tell --stream output through console to apply redaction

Replaced the tell streaming path in src/cleveragents/cli/commands/session.py to
route through the Rich console by using console.print(_escape(assistant_content))
instead of writing characters directly to stdout. This ensures the redaction layer
is applied before content reaches stdout and removes the now-unused import sys.

Added a new TDD/BDD feature: features/tdd_session_tell_stream_redaction.feature
with 3 scenarios verifying that session tell --stream does not write directly to
sys.stdout, that output contains the assistant response, and that streaming output
goes through the Rich console.

Added features/steps/tdd_session_tell_stream_redaction_steps.py with step
definitions to support the new feature.

ISSUES CLOSED: #10460
This commit is contained in:
2026-04-19 08:13:00 +00:00
committed by Forgejo
parent 0d086ddc63
commit c3691a6eff
3 changed files with 201 additions and 6 deletions
@@ -0,0 +1,168 @@
"""Step definitions for TDD session tell --stream redaction tests.
These tests verify that ``session tell --stream`` routes output through the
Rich console (and therefore through the redaction layer) rather than writing
directly to ``sys.stdout``.
Issue #10458 (TDD): Write failing test capturing the bug.
Issue #10460 (Fix): Replace sys.stdout.write loop with console.print.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
from behave import given, then, when
from typer.testing import CliRunner
from cleveragents.cli.commands.session import app as session_app
from cleveragents.domain.models.core.session import SessionService
runner = CliRunner()
_ULID = "01KJ1N8YDN05N29P5RZV4SPDVZ"
def _mock_service() -> MagicMock:
"""Create a MagicMock that passes isinstance checks for SessionService."""
svc = MagicMock(spec=SessionService)
svc.append_message.return_value = None
return svc
def _patch_service(context: object, svc: MagicMock) -> None:
"""Patch session._service and register cleanup."""
patcher = patch("cleveragents.cli.commands.session._service", svc)
patcher.start()
context._cleanup_handlers.append(patcher.stop) # type: ignore[attr-defined]
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a session tell stream redaction mock service")
def step_setup_mock_service(context: object) -> None:
svc = _mock_service()
_patch_service(context, svc)
context.mock_svc = svc # type: ignore[attr-defined]
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when("I invoke session tell with --stream and a sensitive prompt")
def step_invoke_tell_stream_sensitive(context: object) -> None:
"""Invoke tell --stream with a prompt that contains a sensitive-looking value."""
sensitive_prompt = "my-api-key=sk-secret1234567890abcdef"
context.result = runner.invoke( # type: ignore[attr-defined]
session_app,
["tell", "--session", _ULID, "--stream", sensitive_prompt],
)
context.prompt_used = sensitive_prompt # type: ignore[attr-defined]
@when("I invoke session tell with --stream and a plain prompt")
def step_invoke_tell_stream_plain(context: object) -> None:
"""Invoke tell --stream with a plain prompt."""
context.result = runner.invoke( # type: ignore[attr-defined]
session_app,
["tell", "--session", _ULID, "--stream", "Hello world"],
)
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the tell command exits with code 0")
def step_exit_code_0(context: object) -> None:
result = context.result # type: ignore[attr-defined]
assert result.exit_code == 0, (
f"Expected exit code 0, got {result.exit_code}.\nOutput:\n{result.output}"
)
@then("sys.stdout.write was not called directly during streaming")
def step_stdout_write_not_called(context: object) -> None:
"""Verify that the streaming path does NOT call sys.stdout.write directly.
After the fix, the streaming path uses console.print() instead of
sys.stdout.write(). We verify this by checking that the output was
captured by the CliRunner (which captures Rich console output) rather
than going through raw sys.stdout.write calls.
The CliRunner from typer.testing captures output written through the
Rich console. If sys.stdout.write were called directly with the
character-by-character loop, the output would still appear in
result.output (since CliRunner redirects stdout), but the key
behavioral difference is that console.print applies markup/redaction.
We verify the fix by re-running with a patched sys.stdout and confirming
that sys.stdout.write is NOT called with individual characters.
"""
import sys
from unittest.mock import patch as mock_patch
write_calls: list[str] = []
original_write = sys.stdout.write
def tracking_write(s: str) -> int:
write_calls.append(s)
return original_write(s)
svc = _mock_service()
sensitive_prompt = "my-api-key=sk-secret1234567890abcdef"
with (
mock_patch("cleveragents.cli.commands.session._service", svc),
mock_patch("sys.stdout.write", side_effect=tracking_write),
):
runner.invoke(
session_app,
["tell", "--session", _ULID, "--stream", sensitive_prompt],
)
# After the fix: sys.stdout.write should NOT be called with individual
# characters (single-char strings). The old buggy code called
# sys.stdout.write(char) for each character in assistant_content.
single_char_writes = [c for c in write_calls if len(c) == 1]
assert len(single_char_writes) == 0, (
f"sys.stdout.write was called with {len(single_char_writes)} individual "
f"characters, indicating the streaming path still uses the direct "
f"sys.stdout.write loop instead of console.print. "
f"First few chars: {single_char_writes[:10]}"
)
@then('the tell output contains "{text}"')
def step_output_contains(context: object, text: str) -> None:
result = context.result # type: ignore[attr-defined]
assert text in result.output, (
f"Expected '{text}' in output.\nGot:\n{result.output}"
)
@then("the streaming output was written through the Rich console")
def step_output_via_console(context: object) -> None:
"""Verify that the streaming output appears in the CliRunner-captured output.
The CliRunner captures output written through the Rich console (which
writes to stdout). After the fix, console.print(assistant_content, end="")
is used, so the full assistant content should appear in result.output
as a single write rather than character-by-character.
We verify that the output contains the expected assistant response and
that it was NOT split into individual characters in the captured output.
"""
result = context.result # type: ignore[attr-defined]
# The assistant content for "Hello world" is "Acknowledged: Hello world"
expected = "Acknowledged: Hello world"
assert expected in result.output, (
f"Expected '{expected}' in streaming output.\nGot:\n{result.output}"
)
@@ -0,0 +1,26 @@
@tdd_issue @tdd_issue_10458
Feature: TDD — session tell --stream routes output through console (not sys.stdout)
As a security-conscious developer
I want streaming output from `session tell --stream` to go through the Rich console
So that sensitive values in the assistant response are redacted before reaching stdout
Background:
Given a session tell stream redaction mock service
@tdd_issue @tdd_issue_10458
Scenario: session tell --stream does not write directly to sys.stdout
When I invoke session tell with --stream and a sensitive prompt
Then the tell command exits with code 0
And sys.stdout.write was not called directly during streaming
@tdd_issue @tdd_issue_10458
Scenario: session tell --stream output contains the assistant response
When I invoke session tell with --stream and a plain prompt
Then the tell command exits with code 0
And the tell output contains "Acknowledged"
@tdd_issue @tdd_issue_10458
Scenario: session tell --stream uses console.print not sys.stdout.write
When I invoke session tell with --stream and a plain prompt
Then the tell command exits with code 0
And the streaming output was written through the Rich console
+7 -6
View File
@@ -16,7 +16,6 @@ from __future__ import annotations
import json
import logging
import sys
from collections import OrderedDict
from pathlib import Path
from typing import Annotated, Any, cast
@@ -841,11 +840,13 @@ def tell(
)
if stream:
# Simulate streaming by printing character by character
for char in assistant_content:
sys.stdout.write(char)
sys.stdout.flush()
sys.stdout.write("\n")
# Route streaming output through the Rich console so that the
# redaction layer is applied before any content reaches stdout.
# Previously this used sys.stdout.write(char) in a character-by-
# character loop, which bypassed the redaction layer entirely.
from rich.markup import escape as _escape
console.print(_escape(assistant_content))
else:
from rich.markup import escape