From ef808b4c73b8d6804bd1bafade421b8e2f1a5a2e Mon Sep 17 00:00:00 2001 From: Saurabh Maurya Date: Fri, 4 Sep 2026 12:56:22 +0000 Subject: [PATCH 1/2] feat(reporting): align delegated reporter protocol with ReportingContext and multi-type dispatch --- evalbench/evalproto/eval_agent.proto | 10 +++ evalbench/reporting/remote_reporter.py | 56 +++++++++---- evalbench/test/agent_grpc_proxy_test.py | 105 +++++++++++++++++++++++- 3 files changed, 155 insertions(+), 16 deletions(-) diff --git a/evalbench/evalproto/eval_agent.proto b/evalbench/evalproto/eval_agent.proto index 8e0ece74..e2388246 100644 --- a/evalbench/evalproto/eval_agent.proto +++ b/evalbench/evalproto/eval_agent.proto @@ -89,8 +89,17 @@ message ReporterSpec { float timeout_seconds = 3; } +message ReportingContext { + string job_id = 1; + string run_time = 2; // Timestamp of evaluation run + string store_type = 3; // "CONFIGS", "EVALS", "SCORES", "SUMMARY" + string results_json = 4; // Serialized DataFrame records as JSON string (orient="records") + string database = 5; // Optional target database name +} + message ReportingRequest { ReporterSpec reporter = 1; + ReportingContext context = 2; } message ReporterResult { @@ -98,6 +107,7 @@ message ReporterResult { bool success = 2; string result_json = 3; // Generic JSON payload returned by reporter (URIs, paths, metadata) string error_message = 4; + repeated string artifact_uris = 5; // Direct URIs produced (e.g. GCS URLs, archive paths) } message ReportingResponse { diff --git a/evalbench/reporting/remote_reporter.py b/evalbench/reporting/remote_reporter.py index 4102104a..ea4ee5bd 100644 --- a/evalbench/reporting/remote_reporter.py +++ b/evalbench/reporting/remote_reporter.py @@ -27,7 +27,8 @@ def __init__( self.reporter_name = reporter_name self.config = dict(reporting_config) if isinstance(reporting_config, dict) else {} self.timeout_seconds = float(self.config.get("timeout_seconds", 300.0)) - self._reported = False + self.database = str(self.config.get("database", "")) + self.artifact_uris: list[str] = [] logger.info( "Initialized RemoteReporter: name=%s, config=%s", self.reporter_name, @@ -36,18 +37,14 @@ def __init__( def store(self, results: pd.DataFrame, store_type: Any) -> None: type_name = getattr(store_type, "name", str(store_type)) - # Delegate once during evaluation results processing - if type_name != "EVALS": - return - if self._reported: - return session_id = rpc_id_var.get() if session_id not in AGENT_GRPC_PROXY_QUEUES: logger.warning( "RemoteReporter: session_id %s not in AGENT_GRPC_PROXY_QUEUES, " - "skipping delegated reporting", + "skipping delegated reporting for %s", session_id, + type_name, ) return @@ -62,15 +59,38 @@ def store(self, results: pd.DataFrame, store_type: Any) -> None: timeout_seconds=self.timeout_seconds, ) + results_json = "" + if isinstance(results, pd.DataFrame) and not results.empty: + try: + results_json = results.to_json(orient="records", date_format="iso") + except Exception as e: + logger.warning( + "RemoteReporter: Failed to serialize DataFrame to JSON: %s", + e, + ) + results_json = "" + + reporting_context = eval_agent_pb2.ReportingContext( + job_id=str(self.job_id or ""), + run_time=str(self.run_time or ""), + store_type=type_name, + results_json=results_json, + database=self.database, + ) + msg = eval_agent_pb2.AgentStreamMessage( session_id=session_id, correlation_id=correlation_id, - reporting_request=eval_agent_pb2.ReportingRequest(reporter=reporter_spec), + reporting_request=eval_agent_pb2.ReportingRequest( + reporter=reporter_spec, + context=reporting_context, + ), ) logger.info( - "[REVERSE_REPORTER] Dispatching ReportingRequest for '%s' (correlation_id=%s)", + "[REMOTE_REPORTER] Dispatching ReportingRequest for '%s' type=%s (correlation_id=%s)", self.reporter_name, + type_name, correlation_id, ) out_queue.put(msg) @@ -81,26 +101,34 @@ def store(self, results: pd.DataFrame, store_type: Any) -> None: res = resp_msg.reporting_response.result if res.success: logger.info( - "[REVERSE_REPORTER] Delegated reporter '%s' completed successfully: %s", + "[REMOTE_REPORTER] Delegated reporter '%s' (%s) completed successfully: %s", self.reporter_name, + type_name, res.result_json, ) + if res.artifact_uris: + self.artifact_uris.extend(res.artifact_uris) else: logger.error( - "[REVERSE_REPORTER] Delegated reporter '%s' failed: %s", + "[REMOTE_REPORTER] Delegated reporter '%s' (%s) failed: %s", self.reporter_name, + type_name, res.error_message, ) - self._reported = True else: logger.warning( - "[REVERSE_REPORTER] Received unexpected response on stream: %s", + "[REMOTE_REPORTER] Received unexpected response on stream: %s", resp_msg.WhichOneof("payload"), ) except queue.Empty: logger.error( - "[REVERSE_REPORTER] Timed out waiting for ReportingResponse for '%s'", + "[REMOTE_REPORTER] Timed out waiting for ReportingResponse for '%s' (%s)", self.reporter_name, + type_name, ) finally: inboxes.pop(correlation_id, None) + + def print_dashboard_links(self) -> None: + # TODO: Surface remote artifact URIs and dashboard links once UX format is finalized. + pass diff --git a/evalbench/test/agent_grpc_proxy_test.py b/evalbench/test/agent_grpc_proxy_test.py index 3ac53071..d41acb10 100644 --- a/evalbench/test/agent_grpc_proxy_test.py +++ b/evalbench/test/agent_grpc_proxy_test.py @@ -119,7 +119,7 @@ def answer_scorer(): def test_remote_reporter_success(self): reporter = RemoteReporter( "gcs_artifacts", - {"bucket": "test-bucket", "path_prefix": "runs"}, + {"bucket": "test-bucket", "path_prefix": "runs", "database": "bigquery"}, job_id="job_123", run_time="2026-08-18", ) @@ -131,6 +131,14 @@ def answer_reporter(): rep_spec = msg.reporting_request.reporter self.assertEqual(rep_spec.reporter_name, "gcs_artifacts") + ctx = msg.reporting_request.context + self.assertEqual(ctx.job_id, "job_123") + self.assertEqual(ctx.run_time, "2026-08-18") + self.assertEqual(ctx.store_type, "EVALS") + self.assertEqual(ctx.database, "bigquery") + parsed_data = json.loads(ctx.results_json) + self.assertEqual(parsed_data, [{"eval_id": "1"}]) + reply = eval_agent_pb2.AgentStreamMessage( session_id=self.session_id, correlation_id=corr_id, @@ -139,6 +147,7 @@ def answer_reporter(): reporter_name="gcs_artifacts", success=True, result_json='{"uri": "gs://test-bucket/runs/archive.zip"}', + artifact_uris=["gs://test-bucket/runs/archive.zip"], ) ), ) @@ -150,7 +159,99 @@ def answer_reporter(): df = pd.DataFrame({"eval_id": ["1"]}) reporter.store(df, STORETYPE.EVALS) t.join() - self.assertTrue(reporter._reported) + + self.assertEqual(reporter.artifact_uris, ["gs://test-bucket/runs/archive.zip"]) + reporter.print_dashboard_links() + + def test_remote_reporter_all_store_types(self): + reporter = RemoteReporter( + "csv", + {"output_directory": "/tmp/results"}, + job_id="job_456", + run_time="2026-08-18", + ) + + dispatched_types = [] + + def answer_reporter(): + for _ in range(4): + msg = self.out_queue.get(timeout=2.0) + corr_id = msg.correlation_id + ctx = msg.reporting_request.context + dispatched_types.append(ctx.store_type) + + reply = eval_agent_pb2.AgentStreamMessage( + session_id=self.session_id, + correlation_id=corr_id, + reporting_response=eval_agent_pb2.ReportingResponse( + result=eval_agent_pb2.ReporterResult( + reporter_name="csv", + success=True, + result_json=json.dumps({"written": ctx.store_type}), + ) + ), + ) + self.inboxes[corr_id].put(reply) + + t = threading.Thread(target=answer_reporter) + t.start() + + df = pd.DataFrame({"key": ["val"]}) + reporter.store(df, STORETYPE.CONFIGS) + reporter.store(df, STORETYPE.EVALS) + reporter.store(df, STORETYPE.SCORES) + reporter.store(df, STORETYPE.SUMMARY) + t.join() + + self.assertEqual( + dispatched_types, + ["CONFIGS", "EVALS", "SCORES", "SUMMARY"], + ) + + def test_remote_reporter_timeout(self): + reporter = RemoteReporter( + "slow_reporter", + {"timeout_seconds": 0.1}, + job_id="job_slow", + run_time="2026-08-18", + ) + df = pd.DataFrame({"eval_id": ["1"]}) + # Should complete without raising exception + reporter.store(df, STORETYPE.EVALS) + self.assertEqual(len(reporter.artifact_uris), 0) + + def test_remote_reporter_failure_response(self): + reporter = RemoteReporter( + "failing_reporter", + {"timeout_seconds": 5.0}, + job_id="job_fail", + run_time="2026-08-18", + ) + + def answer_failure(): + msg = self.out_queue.get(timeout=2.0) + corr_id = msg.correlation_id + + reply = eval_agent_pb2.AgentStreamMessage( + session_id=self.session_id, + correlation_id=corr_id, + reporting_response=eval_agent_pb2.ReportingResponse( + result=eval_agent_pb2.ReporterResult( + reporter_name="failing_reporter", + success=False, + error_message="GCS bucket not accessible", + ) + ), + ) + self.inboxes[corr_id].put(reply) + + t = threading.Thread(target=answer_failure) + t.start() + + df = pd.DataFrame({"eval_id": ["1"]}) + reporter.store(df, STORETYPE.EVALS) + t.join() + self.assertEqual(len(reporter.artifact_uris), 0) if __name__ == "__main__": From 2214ee659c8c0fb5befe2adb27f5f9f3aeaa80f2 Mon Sep 17 00:00:00 2001 From: Saurabh Maurya Date: Fri, 4 Sep 2026 13:03:22 +0000 Subject: [PATCH 2/2] refactor(reporting): remove artifact_uris and dashboard link hook --- evalbench/evalproto/eval_agent.proto | 1 - evalbench/reporting/remote_reporter.py | 7 ------- evalbench/test/agent_grpc_proxy_test.py | 6 ------ 3 files changed, 14 deletions(-) diff --git a/evalbench/evalproto/eval_agent.proto b/evalbench/evalproto/eval_agent.proto index e2388246..6444acb4 100644 --- a/evalbench/evalproto/eval_agent.proto +++ b/evalbench/evalproto/eval_agent.proto @@ -107,7 +107,6 @@ message ReporterResult { bool success = 2; string result_json = 3; // Generic JSON payload returned by reporter (URIs, paths, metadata) string error_message = 4; - repeated string artifact_uris = 5; // Direct URIs produced (e.g. GCS URLs, archive paths) } message ReportingResponse { diff --git a/evalbench/reporting/remote_reporter.py b/evalbench/reporting/remote_reporter.py index ea4ee5bd..7dda605f 100644 --- a/evalbench/reporting/remote_reporter.py +++ b/evalbench/reporting/remote_reporter.py @@ -28,7 +28,6 @@ def __init__( self.config = dict(reporting_config) if isinstance(reporting_config, dict) else {} self.timeout_seconds = float(self.config.get("timeout_seconds", 300.0)) self.database = str(self.config.get("database", "")) - self.artifact_uris: list[str] = [] logger.info( "Initialized RemoteReporter: name=%s, config=%s", self.reporter_name, @@ -106,8 +105,6 @@ def store(self, results: pd.DataFrame, store_type: Any) -> None: type_name, res.result_json, ) - if res.artifact_uris: - self.artifact_uris.extend(res.artifact_uris) else: logger.error( "[REMOTE_REPORTER] Delegated reporter '%s' (%s) failed: %s", @@ -128,7 +125,3 @@ def store(self, results: pd.DataFrame, store_type: Any) -> None: ) finally: inboxes.pop(correlation_id, None) - - def print_dashboard_links(self) -> None: - # TODO: Surface remote artifact URIs and dashboard links once UX format is finalized. - pass diff --git a/evalbench/test/agent_grpc_proxy_test.py b/evalbench/test/agent_grpc_proxy_test.py index d41acb10..910e0340 100644 --- a/evalbench/test/agent_grpc_proxy_test.py +++ b/evalbench/test/agent_grpc_proxy_test.py @@ -147,7 +147,6 @@ def answer_reporter(): reporter_name="gcs_artifacts", success=True, result_json='{"uri": "gs://test-bucket/runs/archive.zip"}', - artifact_uris=["gs://test-bucket/runs/archive.zip"], ) ), ) @@ -160,9 +159,6 @@ def answer_reporter(): reporter.store(df, STORETYPE.EVALS) t.join() - self.assertEqual(reporter.artifact_uris, ["gs://test-bucket/runs/archive.zip"]) - reporter.print_dashboard_links() - def test_remote_reporter_all_store_types(self): reporter = RemoteReporter( "csv", @@ -218,7 +214,6 @@ def test_remote_reporter_timeout(self): df = pd.DataFrame({"eval_id": ["1"]}) # Should complete without raising exception reporter.store(df, STORETYPE.EVALS) - self.assertEqual(len(reporter.artifact_uris), 0) def test_remote_reporter_failure_response(self): reporter = RemoteReporter( @@ -251,7 +246,6 @@ def answer_failure(): df = pd.DataFrame({"eval_id": ["1"]}) reporter.store(df, STORETYPE.EVALS) t.join() - self.assertEqual(len(reporter.artifact_uris), 0) if __name__ == "__main__":