feat: implement zero-latency streaming interceptor for LLM token generation (#246)#389
Conversation
|
Warning Review limit reached
Next review available in: 54 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughAdds a streaming token interception pipeline (OpenAI, Anthropic, Google) with per-token safety blocking and a WebSocket broadcast manager. Introduces a BFT multi-auditor consensus system with provider-specific auditors, a consensus engine supporting majority/weighted/Byzantine algorithms, and Redis-backed trust scoring. Also updates ChangesStreaming Interception Pipeline
BFT Multi-Auditor Consensus System
Type-check syntax cleanup and test formatting
Sequence Diagram(s)sequenceDiagram
participant Agent
participant StreamInterceptor
participant BaseStreamInterceptor
participant StreamWebSocketManager
Agent->>StreamInterceptor: intercept_stream(prompt)
loop Each token from LLM provider
StreamInterceptor->>BaseStreamInterceptor: process_token(token)
BaseStreamInterceptor->>BaseStreamInterceptor: _detect_dangerous_patterns / _detect_suspicious_patterns
BaseStreamInterceptor-->>StreamInterceptor: SafetyResult(status)
alt BLOCKED
StreamInterceptor->>StreamWebSocketManager: broadcast_status(session_id, BLOCKED)
StreamInterceptor-->>Agent: TokenChunk([BLOCKED: ...])
Note over StreamInterceptor: stop streaming
else SAFE / SUSPICIOUS
StreamInterceptor->>StreamWebSocketManager: broadcast_token(session_id, TokenChunk)
StreamInterceptor-->>Agent: TokenChunk
end
end
sequenceDiagram
participant BFTAuditor
participant OpenAIAuditor
participant AnthropicAuditor
participant ConsensusEngine
participant TrustScorer
BFTAuditor->>OpenAIAuditor: audit_step(step_number, step_data)
BFTAuditor->>AnthropicAuditor: audit_step(step_number, step_data)
OpenAIAuditor-->>BFTAuditor: AuditResult(score, rationale)
AnthropicAuditor-->>BFTAuditor: AuditResult(score, rationale)
BFTAuditor->>ConsensusEngine: run_consensus(results, BYZANTINE)
ConsensusEngine->>ConsensusEngine: _detect_byzantine outliers
ConsensusEngine->>TrustScorer: get_trust_score(provider, model)
TrustScorer-->>ConsensusEngine: trust weight
ConsensusEngine-->>BFTAuditor: ConsensusResult(final_score, agreement_level)
BFTAuditor->>TrustScorer: update_score(provider, model, was_correct)
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested labels
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 3❌ Failed checks (3 warnings)
✅ Passed checks (2 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Hey @SakethSumanBathini @sreerevanth! 👋 This PR (#246) is ready for review. I've implemented the zero-latency streaming interceptor with:
Could you please:
Happy to make any changes requested! 🚀 Closes #246 |
|
Caution Failed to replace (edit) comment. This is likely due to insufficient permissions or the comment being deleted. Error details |
🧪 PR Test Results
Python 3.12 · commit 4635d70 |
|
@BhakktiGautam join the discord and also please check the pr |
|
@BhakktiGautam update or reply !! |
|
@BhakktiGautam final call for fixing the ci |
|
@sreerevanth - I'm on it! 🙏 I'll fix all the CI issues today and get this PR ready for merge. |
|
@sreerevanth - CI is 99% green! Only |
- Add base auditor interface with provider-agnostic design - Implement OpenAI, Anthropic, Google auditors - Add consensus engine with majority, weighted, Byzantine algorithms - Add trust scoring system with Redis persistence - Implement Byzantine fault detection and exclusion - Add comprehensive tests Closes sreerevanth#244
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 17
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🟡 Minor comments (4)
agentwatch/interceptors/google.py-19-28 (1)
19-28: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winHonor the per-call
modeloverride or drop the parameter.
intercept_stream(..., model=...)never usesmodel; it always streams with theGenerativeModelcreated in__init__. That silently ignores caller intent and breaks the shared interceptor contract.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agentwatch/interceptors/google.py` around lines 19 - 28, The Google interceptor’s intercept_stream method accepts a per-call model override but never uses it, so either honor the model argument when calling generate_content_async or remove the parameter from the shared interceptor contract. Update the logic in intercept_stream and, if needed, the GenerativeModel usage in __init__ so the requested model is actually selected for each call.agentwatch/auditors/consensus.py-180-182 (1)
180-182: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winFallback median is wrong for even-sized result sets.
scores[len(scores) // 2]picks the upper-middle element, not the median, when there are an even number of auditors. Usestatistics.median()or average the two middle scores.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agentwatch/auditors/consensus.py` around lines 180 - 182, The fallback median calculation in consensus.py is incorrect for even-sized auditor result sets because picking scores[len(scores) // 2] returns the upper-middle value instead of the true median. Update the logic in the consensus scoring path that builds scores and assigns median so it uses statistics.median() or explicitly averages the two middle scores when the list length is even, while keeping the existing results/score flow intact.agentwatch/auditors/openai.py-41-47 (1)
41-47: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winRecovered auditors stay
UNRESPONSIVEforever.The error paths mark this auditor unresponsive, but the success path never calls
mark_healthy(). Sinceagentwatch/auditors/bft_auditor.py, Lines 189-191 exposesa.status.valuedirectly, one transient failure leaves status reporting stale until restart.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agentwatch/auditors/openai.py` around lines 41 - 47, The OpenAI auditor success path in the return from AuditResult leaves status stale after a prior failure because it never clears the unresponsive state. Update the successful execution path in the openai auditor class to call mark_healthy() before returning AuditResult, so a recovered auditor transitions back from UNRESPONSIVE. Use the existing provider/model audit flow and the auditor status methods already defined in the auditor implementation to locate the fix.agentwatch/auditors/anthropic.py-59-65 (1)
59-65: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
health_check()should make a real Anthropic request
This always returnsTrue, so bad credentials or API outages still look healthy. Use a minimal authenticated call and returnFalseon request errors.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agentwatch/auditors/anthropic.py` around lines 59 - 65, The AnthropicAuditor.health_check() method currently returns True without contacting the API, so it never detects invalid credentials or outages. Update health_check() to perform a minimal authenticated Anthropic request using the existing client in anthropic.py, and keep the try/except so request failures return False. Use the AnthropicAuditor.health_check symbol to locate the method and ensure the check is lightweight but real.
🧹 Nitpick comments (6)
tests/test_interceptors/test_openai.py (2)
13-18: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExercise
process_token()in these tests instead of mutatingbufferdirectly.Writing
bufferby hand skips the hot-path bookkeeping inBaseStreamInterceptor.process_token()—especiallytotal_tokensand overflow trimming. That means these tests won't catch regressions inblocked_at_indexor incremental stream state even though the runtime path depends on them.Also applies to: 27-29, 37-40, 48-51
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_interceptors/test_openai.py` around lines 13 - 18, The OpenAI interceptor tests are bypassing the real stream-processing path by appending directly to buffer, so they miss BaseStreamInterceptor.process_token() bookkeeping like total_tokens, overflow trimming, and incremental state updates. Update the affected tests to feed each token through process_token() on the interceptor instance before calling _safety_check(), and keep assertions focused on the resulting blocked_at_index and stream state rather than manual buffer contents.
8-8: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDrop the
@pytest.mark.asynciodecorators.asyncio_mode = "auto"is already enabled, so these markers are redundant here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_interceptors/test_openai.py` at line 8, Remove the redundant async test marker from the OpenAI interceptor tests: since asyncio_mode is already set to auto, drop the `@pytest.mark.asyncio` decorator from the affected test definitions in the test_openai module and keep the tests as plain async functions.Source: Coding guidelines
agentwatch/interceptors/_init_.py (1)
1-18: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRename or remove this duplicate initializer module.
_init_.pyis imported asagentwatch.interceptors._init_, not as the package initializer. Normalimport agentwatch.interceptorsstill uses the existingagentwatch/interceptors/__init__.py, so this export surface never takes effect.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agentwatch/interceptors/_init_.py` around lines 1 - 18, The _init_.py module is a duplicate initializer that is not used for package imports, so its exports never affect agentwatch.interceptors. Rename or remove this file and, if these re-exports are meant to be public, move them into the actual package initializer in agentwatch.interceptors/__init__.py, keeping the existing symbols like BaseStreamInterceptor, TokenChunk, SafetyResult, TokenStatus, OpenAIStreamInterceptor, AnthropicStreamInterceptor, GoogleStreamInterceptor, and StreamInterceptorFactory consistent.tests/test_interceptors/test_factory.py (1)
23-26: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a regression test for the
geminialias.
StreamInterceptorFactoryexposes both"google"and"gemini", but only the"google"path is asserted here. One extra test would keep that public alias from regressing silently.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_interceptors/test_factory.py` around lines 23 - 26, Add a regression test in test_factory for the StreamInterceptorFactory gemini alias, since only the google path is currently covered. Create a new test alongside test_factory_creates_google that calls StreamInterceptorFactory.create with "gemini" and asserts it returns a GoogleStreamInterceptor, mirroring the existing google assertion so both public aliases are protected.tests/test_bft_auditor/test_consensus.py (1)
6-7: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDrop the redundant
@pytest.mark.asynciomarkers.These tests already run under
asyncio_mode = 'auto', so the decorators add noise and violate the repo’s test convention. As per coding guidelines, "Use asyncio_mode = 'auto' in pytest configuration — no need for@pytest.mark.asyncioon async tests."Also applies to: 21-22, 36-37
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_bft_auditor/test_consensus.py` around lines 6 - 7, Remove the redundant asyncio test decorators from the async test functions in test_consensus.py, since pytest is already configured with asyncio_mode = 'auto'. Update the affected tests such as test_majority_consensus and the other async tests mentioned in the comment to rely on auto mode instead of `@pytest.mark.asyncio`, keeping the test definitions otherwise unchanged.Source: Coding guidelines
tests/test_bft_auditor/test_bft_auditor.py (1)
9-10: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRemove the
pytest.mark.asynciodecorators here.Repo pytest config already runs async tests with
asyncio_mode = 'auto', so these decorators are redundant, and Line 10 can be a plaindefbecause it never awaits anything. As per coding guidelines,tests/**/*.py:Use asyncio_mode = 'auto' in pytest configuration — no need for@pytest.mark.asyncioon async tests.Proposed fix
-@pytest.mark.asyncio -async def test_bft_auditor_init(): +def test_bft_auditor_init(): @@ -@pytest.mark.asyncio async def test_bft_auditor_audit():Also applies to: 21-22
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_bft_auditor/test_bft_auditor.py` around lines 9 - 10, Remove the redundant pytest.mark.asyncio decorators from the BFT auditor tests and make the non-awaiting test functions plain def. Update test_bft_auditor_init and the other affected test in test_bft_auditor.py so they rely on pytest’s asyncio_mode=auto configuration instead of explicit async markers.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@agentwatch/api/websocket_stream.py`:
- Around line 28-47: The broadcast methods in websocket_stream should not
swallow send failures while keeping dead sockets in active_connections. In both
send paths that call send_text for a session_id, catch the exception and remove
that session_id from active_connections so a reconnect can establish a fresh
WebSocket instead of repeatedly failing on the stale one.
In `@agentwatch/auditors/_init_.py`:
- Around line 1-22: Rename the module to __init__.py so the agentwatch.auditors
package executes these re-exports on import; _init_.py is treated as a normal
module and won’t expose the intended API. Move the existing re-export block and
__all__ definition into agentwatch.auditors.__init__ so imports like
agentwatch.auditors resolve BaseAuditor, OpenAIAuditor, ConsensusEngine,
TrustScorer, and BFTAuditor correctly.
In `@agentwatch/auditors/bft_auditor.py`:
- Around line 72-86: The timeout fallback in BFTAuditor._audit_step cannot
collect partial results because tasks is built from bare coroutines, so the
timeout handler later calls .done() on objects that do not support it and would
also wait on pending work. Update the task creation in BFTAuditor._audit_step to
schedule each _audit_with_timeout call as asyncio Tasks, and in the
asyncio.TimeoutError path only inspect already-finished tasks to gather their
results without awaiting pending ones. Keep the behavior localized to the
tasks/results handling around asyncio.gather and the timeout fallback.
In `@agentwatch/auditors/consensus.py`:
- Around line 119-127: The consensus output is collapsing multiple auditors that
share the same provider because individual_scores is keyed only by provider.
Update ConsensusResult construction in the consensus logic to key
individual_scores by the full auditor identity used elsewhere in the stack,
matching the provider:model format from the auditor result objects (for example
in the majority/weighted/unanimous paths). Make sure the key is derived from
each result’s unique auditor identifier rather than r.provider so distinct
models from the same provider do not overwrite each other.
- Around line 77-85: The outlier detection in consensus.py’s byzantine selection
logic relies only on mean/std in the result-scan loop, which misses small-quorum
cases like 0.9/0.9/0.1. Update the detection in the same consensus path to use a
small-quorum fallback or a median/MAD-style rule instead of only `> 3 * std`,
and keep the `byzantine_indices` computation aligned with the new criterion so
the 3-auditor test case is flagged correctly.
- Around line 94-97: The “more than 50% Byzantine” path in
`Consensus.run_consensus()` is not actually using the median fallback because
returning `(results, 0)` keeps the normal consensus flow active. Update the
branch around `byzantine_indices` handling so it returns a sentinel or directly
invokes `_fallback_consensus()`, and make sure `run_consensus()` detects that
signal and bypasses `_majority_consensus()` / `_weighted_consensus()`
accordingly.
In `@agentwatch/auditors/google.py`:
- Around line 10-13: The GoogleAuditor initializer is mutating global SDK state
via genai.configure(), which can cause later GoogleAuditor instances to reuse
the wrong credentials. Update GoogleAuditor.__init__ to avoid process-wide
configuration by using a per-instance client/API setup or another isolated
authentication mechanism, and ensure self.model_instance is created from that
isolated client rather than relying on shared global state.
In `@agentwatch/auditors/openai.py`:
- Line 63: The OpenAI auditor health check is using an unsupported `limit`
argument and never clears a stale unhealthy state on success. Update
`OpenAIAuditor.health_check()` to call `self.client.models.list()` without the
invalid parameter, and after a successful call make sure the auditor’s `status`
is reset to healthy before returning true so transient failures do not leave it
stuck unhealthy.
In `@agentwatch/auditors/trust.py`:
- Around line 109-114: Time decay is only applied in update(), so reads still
return stale persisted scores after reloads or idle time. Update
get_trust_score() and get_ranking() in trust.py to recalculate the current
decayed value from each score’s last_update using _calculate_score() before
returning or sorting, so ConsensusEngine._weighted_consensus() sees fresh trust
values.
- Around line 46-55: The trust calculation in the score method is scaling the
base value down too aggressively, so a correct audit can lower trust below the
neutral baseline. Update the scoring logic in TrustAuditor’s score/update path
so a correct audit moves trust upward from 0.5 instead of multiplying it down by
confidence; keep the confidence and decay factors, but apply them in a way that
preserves or increases trust for `was_correct=True` and matches the expectations
in `tests/test_bft_auditor/test_trust.py`.
In `@agentwatch/core/token_stream.py`:
- Around line 31-35: The TokenStream.clear() method only resets the buffer and
total_tokens, so reused streams keep stale lifecycle state. Update clear() in
TokenStream to also reset start_time, _dangerous_detected, and _blocked_at so a
fresh session starts cleanly and properties like is_dangerous and token_rate
reflect the new stream state.
In `@agentwatch/interceptors/anthropic.py`:
- Around line 32-40: The stream handling in AnthropicInterceptor is checking the
wrong event type, so streamed content is never passed through process_token.
Update the async loop in AnthropicInterceptor to use the Anthropic SDK stream
event types emitted by messages.stream() (for example text and input_json)
instead of content_block_delta, and route both text tokens and tool-use argument
chunks through the existing safety-check path so process_token still runs.
In `@agentwatch/interceptors/base.py`:
- Around line 39-56: The interceptor state is only partially cleared between
streams, so previous tokens can leak into the next request and skew blocking
indexes. Add a reset helper on the base interceptor (for example in the class
that defines process_token, buffer, total_tokens, and is_blocked) to clear
buffer, reset total_tokens, and unblock state, then call it at the start of each
provider intercept_stream() implementation such as the OpenAI interceptor before
processing a new stream.
In `@agentwatch/interceptors/factory.py`:
- Around line 3-6: The Google interceptor is imported unconditionally in the
interceptor factory, which breaks imports on clean installs when
google.generativeai is unavailable. Update agentwatch/interceptors/factory.py
and agentwatch/interceptors/__init__.py so GoogleStreamInterceptor is loaded
lazily or wrapped in an optional-dependency guard, while keeping
AnthropicStreamInterceptor, OpenAIStreamInterceptor, and BaseStreamInterceptor
importable without Google installed.
In `@agentwatch/interceptors/google.py`:
- Around line 32-39: The streaming loop in GoogleInterceptor is only passing
chunk.text into process_token(), so Gemini tool/function-call parts are being
skipped. Update the async iteration in the interceptor to detect and extract
non-text parts from each chunk (including function/tool-call arguments) and feed
those payloads through process_token() alongside normal text. Use the existing
Google interceptor flow and process_token method to ensure streamed call data is
reconstructed instead of ignored.
In `@agentwatch/interceptors/openai.py`:
- Around line 19-20: `BaseStreamInterceptor.process_token()` is only resetting
`self.is_blocked`, so reused interceptor instances keep stale `self.buffer` and
`self.total_tokens` across streams. Update the stream-start setup in
`process_token()` to fully reset the buffered state before processing a new
response, including clearing the buffer and zeroing the token count along with
`self.is_blocked`, so each stream starts cleanly.
- Around line 32-34: intercept_stream() currently only forwards
chunk.choices[0].delta.content into process_token(), so streamed tool-call
arguments are skipped and never inspected; update the OpenAI stream handling in
OpenAIInterceptor.intercept_stream to also detect delta.tool_calls and pass each
streamed tool-call function.arguments fragment through process_token() with the
same safety checks as content tokens.
---
Minor comments:
In `@agentwatch/auditors/anthropic.py`:
- Around line 59-65: The AnthropicAuditor.health_check() method currently
returns True without contacting the API, so it never detects invalid credentials
or outages. Update health_check() to perform a minimal authenticated Anthropic
request using the existing client in anthropic.py, and keep the try/except so
request failures return False. Use the AnthropicAuditor.health_check symbol to
locate the method and ensure the check is lightweight but real.
In `@agentwatch/auditors/consensus.py`:
- Around line 180-182: The fallback median calculation in consensus.py is
incorrect for even-sized auditor result sets because picking scores[len(scores)
// 2] returns the upper-middle value instead of the true median. Update the
logic in the consensus scoring path that builds scores and assigns median so it
uses statistics.median() or explicitly averages the two middle scores when the
list length is even, while keeping the existing results/score flow intact.
In `@agentwatch/auditors/openai.py`:
- Around line 41-47: The OpenAI auditor success path in the return from
AuditResult leaves status stale after a prior failure because it never clears
the unresponsive state. Update the successful execution path in the openai
auditor class to call mark_healthy() before returning AuditResult, so a
recovered auditor transitions back from UNRESPONSIVE. Use the existing
provider/model audit flow and the auditor status methods already defined in the
auditor implementation to locate the fix.
In `@agentwatch/interceptors/google.py`:
- Around line 19-28: The Google interceptor’s intercept_stream method accepts a
per-call model override but never uses it, so either honor the model argument
when calling generate_content_async or remove the parameter from the shared
interceptor contract. Update the logic in intercept_stream and, if needed, the
GenerativeModel usage in __init__ so the requested model is actually selected
for each call.
---
Nitpick comments:
In `@agentwatch/interceptors/_init_.py`:
- Around line 1-18: The _init_.py module is a duplicate initializer that is not
used for package imports, so its exports never affect agentwatch.interceptors.
Rename or remove this file and, if these re-exports are meant to be public, move
them into the actual package initializer in agentwatch.interceptors/__init__.py,
keeping the existing symbols like BaseStreamInterceptor, TokenChunk,
SafetyResult, TokenStatus, OpenAIStreamInterceptor, AnthropicStreamInterceptor,
GoogleStreamInterceptor, and StreamInterceptorFactory consistent.
In `@tests/test_bft_auditor/test_bft_auditor.py`:
- Around line 9-10: Remove the redundant pytest.mark.asyncio decorators from the
BFT auditor tests and make the non-awaiting test functions plain def. Update
test_bft_auditor_init and the other affected test in test_bft_auditor.py so they
rely on pytest’s asyncio_mode=auto configuration instead of explicit async
markers.
In `@tests/test_bft_auditor/test_consensus.py`:
- Around line 6-7: Remove the redundant asyncio test decorators from the async
test functions in test_consensus.py, since pytest is already configured with
asyncio_mode = 'auto'. Update the affected tests such as test_majority_consensus
and the other async tests mentioned in the comment to rely on auto mode instead
of `@pytest.mark.asyncio`, keeping the test definitions otherwise unchanged.
In `@tests/test_interceptors/test_factory.py`:
- Around line 23-26: Add a regression test in test_factory for the
StreamInterceptorFactory gemini alias, since only the google path is currently
covered. Create a new test alongside test_factory_creates_google that calls
StreamInterceptorFactory.create with "gemini" and asserts it returns a
GoogleStreamInterceptor, mirroring the existing google assertion so both public
aliases are protected.
In `@tests/test_interceptors/test_openai.py`:
- Around line 13-18: The OpenAI interceptor tests are bypassing the real
stream-processing path by appending directly to buffer, so they miss
BaseStreamInterceptor.process_token() bookkeeping like total_tokens, overflow
trimming, and incremental state updates. Update the affected tests to feed each
token through process_token() on the interceptor instance before calling
_safety_check(), and keep assertions focused on the resulting blocked_at_index
and stream state rather than manual buffer contents.
- Line 8: Remove the redundant async test marker from the OpenAI interceptor
tests: since asyncio_mode is already set to auto, drop the `@pytest.mark.asyncio`
decorator from the affected test definitions in the test_openai module and keep
the tests as plain async functions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 315a13e5-691f-4304-86f1-a11c97c1104c
📒 Files selected for processing (28)
agentwatch/api/websocket_stream.pyagentwatch/auditors/_init_.pyagentwatch/auditors/anthropic.pyagentwatch/auditors/base.pyagentwatch/auditors/bft_auditor.pyagentwatch/auditors/consensus.pyagentwatch/auditors/google.pyagentwatch/auditors/openai.pyagentwatch/auditors/trust.pyagentwatch/core/token_stream.pyagentwatch/governance/gdpr.pyagentwatch/interceptors/__init__.pyagentwatch/interceptors/_init_.pyagentwatch/interceptors/anthropic.pyagentwatch/interceptors/base.pyagentwatch/interceptors/factory.pyagentwatch/interceptors/google.pyagentwatch/interceptors/openai.pyagentwatch/memory/nlquery.pyagentwatch/security/owasp.pytests/test_bft_auditor/test_bft_auditor.pytests/test_bft_auditor/test_consensus.pytests/test_bft_auditor/test_trust.pytests/test_channels.pytests/test_cli_export.pytests/test_cost.pytests/test_interceptors/test_factory.pytests/test_interceptors/test_openai.py
| """Multi-auditor consensus system for AgentWatch""" | ||
| from .base import BaseAuditor, AuditResult | ||
| from .openai import OpenAIAuditor | ||
| from .anthropic import AnthropicAuditor | ||
| from .google import GoogleAuditor | ||
| from .consensus import ConsensusEngine, ConsensusResult, ConsensusAlgorithm | ||
| from .trust import TrustScorer, TrustScore | ||
| from .bft_auditor import BFTAuditor | ||
|
|
||
| __all__ = [ | ||
| "BaseAuditor", | ||
| "AuditResult", | ||
| "OpenAIAuditor", | ||
| "AnthropicAuditor", | ||
| "GoogleAuditor", | ||
| "ConsensusEngine", | ||
| "ConsensusResult", | ||
| "ConsensusAlgorithm", | ||
| "TrustScorer", | ||
| "TrustScore", | ||
| "BFTAuditor", | ||
| ] |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
Rename this file to __init__.py.
_init_.py is just a normal module, so import agentwatch.auditors will never execute these re-exports. That breaks the package API surface this PR is introducing. As per coding guidelines, agentwatch/**/__init__.py: __init__.py files use wildcard re-exports — suppress ruff F401/F403 rules per-file.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/auditors/_init_.py` around lines 1 - 22, Rename the module to
__init__.py so the agentwatch.auditors package executes these re-exports on
import; _init_.py is treated as a normal module and won’t expose the intended
API. Move the existing re-export block and __all__ definition into
agentwatch.auditors.__init__ so imports like agentwatch.auditors resolve
BaseAuditor, OpenAIAuditor, ConsensusEngine, TrustScorer, and BFTAuditor
correctly.
Source: Coding guidelines
| async for chunk in response: | ||
| if self.is_blocked: | ||
| break | ||
|
|
||
| if chunk.text: | ||
| token = chunk.text | ||
|
|
||
| safety_result = await self.process_token(token) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
Handle Gemini function/tool-call parts too. chunk.text skips non-text parts, so tool-call arguments never reach process_token(), and streamed call payloads won’t be reconstructed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/interceptors/google.py` around lines 32 - 39, The streaming loop
in GoogleInterceptor is only passing chunk.text into process_token(), so Gemini
tool/function-call parts are being skipped. Update the async iteration in the
interceptor to detect and extract non-text parts from each chunk (including
function/tool-call arguments) and feed those payloads through process_token()
alongside normal text. Use the existing Google interceptor flow and
process_token method to ensure streamed call data is reconstructed instead of
ignored.
sreerevanth
left a comment
sreerevanth
left a comment
There was a problem hiding this comment.
Thanks for the huge amount of work you've put into this PR—there are some really interesting ideas here. However, I can't merge it in its current form.
There are several blocking issues that need to be addressed first:
Fix the incorrect init.py package file (init.py is required).
Resolve the timeout/task handling in the BFT auditor.
Address the major correctness issues raised in the review (stream state reset, optional dependencies, tool-call interception, WebSocket cleanup, consensus logic).
Get all CI checks passing.
I'd also strongly recommend splitting the streaming interceptor and the multi-auditor consensus system into separate PRs. They are substantial features on their own and will be much easier to review independently.
|
@BhakktiGautam THE CI FAILS IN BANDIT PR AND LINT ALONG WITH TESTS PPLEASE FIX THIS ASAP |
- Add streaming interceptors to public API - Add BFT multi-auditor system to public API - Add WebSocket manager to public API - Add token stream buffer to public API - Make Google imports optional - Clean up __all__ list Closes sreerevanth#246
|
@BhakktiGautam is attempting to deploy a commit to the sreerevanth's projects Team on Vercel. A member of the Team first needs to authorize it. |
BhakktiGautam
force-pushed
the
feature/streaming-interceptor-246
branch
from
612d39e to
4516088
Compare
Keep only streaming interceptor code as requested by maintainer Closes sreerevanth#246
BhakktiGautam
force-pushed
the
feature/streaming-interceptor-246
branch
from
d3040ea to
b300352
Compare
2 participants
Closes #246
🏷️ Labels
enhancementfeatureperformancesecurityssoc/social-summer-of-codeneeds-reviewThis PR is submitted as part of Social Summer of Code (SSoC) program. 🚀
Description (Short Version):