fix: backfill rollout status fields from logs when polling completes

The lightweight `/status` endpoint on the tracing gateway only returns the
status code; `Message`, `Details`, and `Extras` still live on the Logs
table. After PR #446 stopped reading from `/logs` on terminal status, the
SDK was constructing `Status(code=..., message="", details=[])` for every
completed rollout and `EvalProtocolError(message="")` for failures, which
broke `tests/remote_server/test_remote_fireworks_propagate_status.py`
(`assert row.rollout_status.message == "test error"`).

Restore the two-phase polling shape from the original PR: poll `/status`
for the code, and on a terminal (non-RUNNING) code do one
`async_search_logs` call to backfill `message`/`details`/`extras` from
the matching log row. This is still ~1000x cheaper on the Logs table than
the pre-#446 polling loop because the search runs once per rollout
completion instead of every poll interval.

Made-with: Cursor

Author: SunnySoldier357

eval_protocol/pytest/remote_rollout_processor.py

@@ -139,8 +139,28 @@ async def _process_row(row: EvaluationRow) -> EvaluationRow:
                         status_code,
                     )
 
-                    status_message = status.get("message", "") or ""
-                    status_details = status.get("details", []) or []
+                    # /status is a point-read on the Status table that only
+                    # carries the status code today. Backfill message/details/
+                    # extras from the Logs table with a single read once the
+                    # rollout has reached a terminal state.
+                    status_message: str = ""
+                    status_details: list = []
+                    status_extras: dict = {}
+                    completed_logs = await self._tracing_adapter.async_search_logs(
+                        session, tags=[f"rollout_id:{row.execution_metadata.rollout_id}"]
+                    )
+                    for log in completed_logs:
+                        sd = log.get("status")
+                        if sd and isinstance(sd, dict) and "code" in sd:
+                            status_message = sd.get("message", "") or ""
+                            status_details = sd.get("details", []) or []
+                            raw_extras = log.get("extras") or {}
+                            status_extras = {
+                                k: v
+                                for k, v in raw_extras.items()
+                                if k not in ("logger_name", "level", "timestamp")
+                            }
+                            break
 
                     exception = exception_for_status_code(status_code, status_message)
                     if exception is not None:
@@ -152,8 +172,7 @@ async def _process_row(row: EvaluationRow) -> EvaluationRow:
                         details=status_details,
                     )
 
-                    status_extras = (status_result or {}).get("extras")
-                    if isinstance(status_extras, dict):
+                    if status_extras:
                         if row.execution_metadata.extra:
                             row.execution_metadata.extra.update(status_extras)
                         else: