feat(agent): introduce generalized reverse proxy bidirectional streaming evaluation

[saurabh-net]

Summary

This PR introduces generalized reverse proxy bidirectional streaming evaluation into Evalbench, enabling multi-turn autonomous agents running in remote/sandboxed environments to connect via an outbound gRPC stream (rpc AgentInteract).

🏛️ Architectural Alignment: Extending Evalbench's Bidirectional Proxy Pattern

To ensure seamless integration and familiar conventions, this PR extends the proven in-memory streaming pattern already established in evalbench/generators/models/grpc_proxy.py (PROXY_QUEUES / rpc Interact) to autonomous multi-turn agents (AGENT_PROXY_QUEUES / rpc AgentInteract).

Feature / Pattern	Existing grpc_proxy.py	Proposed agentic_reverse_proxy.py	Alignment & Evolution
In-Memory Queue Storage	Global registry (PROXY_QUEUES[session_id])	Global registry (AGENT_PROXY_QUEUES[session_id])	Identical proven concurrency pattern
Streaming Servicer Loop	Inbound response reader + outbound queue consumer	Inbound response reader + outbound queue consumer	Identical streaming lifecycle
Memory Cleanup & Teardown	Cleaned up via finally blocks	Cleaned up via finally blocks	Zero memory leaks
Routing / Keying Model	Inboxes keyed by session_id	Multi-scenario correlation IDs (correlation_id)	Multiplexes multiple scenarios across threads
Message Envelopes	Overloads SQL prompt/query fields	Typed AgentStreamMessage oneof protobuf	Clean multi-turn tool calling & telemetry
Framework Base Class	QueryGenerator (Single-turn SQL)	AgentCliGenerator (Multi-turn conversations)	Native multi-turn evaluator support
Delegation Capabilities	Prompt generation only	3-Way Delegation (Turns + Scorers + Reporters)	Complete remote execution & reporting

---

Key Features

1. Typed Agent Streaming Envelopes (evalproto/eval_agent.proto & evalproto/eval_service.proto):

   • TurnRequest / TurnResponse: Pure agent-centric turn exchange delivering prompts, environment variables, working directories, timeout, and resume flags; collects natural language responses, multi-turn tool call trajectories (tool_name, parameters_json, output, duration_ms, status), token statistics, completion status, and error diagnostics.
   • ScoringRequest / ScoringResponse: Generic container for in-sandbox delegated domain scorers (ScorerSpec(scorer_name, config_json) $\rightarrow$ ScoreResult(scorer_name, score, success, result_json, error_message)).
   • ReportingRequest / ReportingResponse: Generic container for delegated workspace and telemetry archival (ReporterSpec(reporter_name, config_json) $\rightarrow$ ReporterResult(reporter_name, success, result_json, error_message)).
   • SessionSummaryMessage: Delivers final job score rollups and evaluation summaries to connected clients upon run completion.
   • rpc AgentInteract(stream AgentStreamMessage) returns (stream AgentStreamMessage): Dedicated streaming RPC endpoint.

2. AgenticReverseProxyGenerator (evalbench/generators/models/agentic_reverse_proxy.py):

   • Conforms to AgentCliGenerator to execute multi-turn agent conversations over the reverse stream.
   • Automatic MCP tool canonicalization (call_mcp_tool with ServerName/ToolName $\rightarrow$ <server>__<tool>).
   • Skill discovery & activation tracking (activate_skill $\rightarrow$ accumulated_skills).
   • Timeout support (timeout_seconds), token usage accounting, and trajectory building.

3. Transparent Delegated Scoring (evalbench/scorers/score.py & evalbench/scorers/remote_scorer.py):

   • Any scorer configured with delegated: true is automatically routed via RemoteScorerProxy across the reverse stream to execute inside the client environment.

4. Transparent Delegated Reporting (evalbench/reporting/__init__.py & evalbench/reporting/remote_reporter.py):

   • Any reporter configured with delegated: true delegates execution to the client via RemoteReporter.

5. Sample Model Config & Documentation:

   • Added datasets/model_configs/agentic_reverse_proxy_model.yaml.
   • Updated docs/agentic-evals.md supported agents reference.

---

Testing & Verification

1. Isolated Unit Tests (evalbench/test/agentic_reverse_proxy_test.py):

   • Generator turn execution and tool parsing (test_agentic_reverse_proxy_generator_turn).
   • Delegated scorer proxy dispatch and score extraction (test_remote_scorer_proxy_dispatch).
   • Delegated reporter dispatch and artifact response handling (test_remote_reporter_dispatch).

2. Hermetic End-to-End Integration Tests (evalbench/test/agentic_reverse_proxy_integration_test.py):

   • Runs a real in-process gRPC server bound dynamically to an ephemeral kernel port (127.0.0.1:0).
   • test_multi_turn_reverse_proxy_e2e: Full multi-turn conversation with skill activation (dataform_best_practices), MCP tool calls (dataform__compile), trajectory validation (TrajectoryMatcher), executable verification (executable), delegated in-sandbox scoring (dataform_compile), and delegated artifact reporting (gcs_artifacts).
   • test_concurrent_sessions_e2e: 3 concurrent independent client sessions running simultaneously without cross-talk or race conditions.
   • test_concurrent_multi_scenario_e2e: Concurrent scenario multiplexing across worker threads within a single session.

3. Verification Command:

   PYTHONPATH=evalbench:. python -m unittest evalbench/test/agentic_reverse_proxy_test.py evalbench/test/agentic_reverse_proxy_integration_test.py
   # Ran 6 tests in 1.828s -> OK
   pycodestyle evalbench --config=.pycodestyle
   # 0 lint errors

[saurabh-net]

🏛️ Architectural Alignment: Extending Evalbench's Bidirectional Proxy Pattern

To ensure seamless integration and familiar conventions for Evalbench maintainers, this PR extends the proven in-memory streaming pattern already established in evalbench/generators/models/grpc_proxy.py (PROXY_QUEUES / rpc Interact) to autonomous multi-turn agents (AGENT_PROXY_QUEUES / rpc AgentInteract).

📊 Side-by-Side Comparison

Feature / Pattern	Existing grpc_proxy.py (rpc Interact)	Proposed agentic_reverse_proxy.py (rpc AgentInteract)	Alignment & Evolution
In-Memory Queue Bus	Global registry: PROXY_QUEUES[session_id] = (inboxes_dict, out_queue)	Global registry: AGENT_PROXY_QUEUES[session_id] = (inboxes_dict, out_queue)	✅ 100% Identical Convention
Streaming Servicer Lifecycle	Async inbound reader task + async outbound yield loop in eval_service.py	Async inbound reader task + async outbound yield loop in eval_service.py	✅ 100% Identical Convention
Memory Cleanup & Teardown	Cleaned up via finally: PROXY_QUEUES.pop(session_id) and disconnect callbacks	Cleaned up via finally: AGENT_PROXY_QUEUES.pop(session_id) and disconnect callbacks	✅ 100% Identical Convention
Routing / Keying Mechanism	Inboxes keyed by conversation_id (single-turn per scenario)	Inboxes keyed by unique correlation_id = uuid4() tickets	🚀 Improved: Multiplexes concurrent turns, scorers, and reporters seamlessly
Message Enveloping	Overloads SQL proto (EvalInputRequest(nl_prompt, payload, generated_sql))	Typed AgentStreamMessage with typed oneof payloads	🚀 Improved: Clean typing without string hacking
Framework Base Class	QueryGenerator (single-turn SQL generation)	AgentCliGenerator (multi-turn autonomous agents)	🚀 Improved: Plugs natively into AgentEvaluator, SimulatedUser, and tool/skill extraction
Delegation Scope	Prompt generation only	3-Way Delegation: Turn generation + In-sandbox scoring + Workspace reporting	🚀 Improved: Enables delegated: true on any scorer or reporter

[prernakakkar-google]

/gcbrun

[prernakakkar-google]

/gcbrun

[prernakakkar-google]

/gcbrun