Support OpenHands MAX and complete trainer trajectories

.agents/skills/benchflow-experiment-review/scripts/validate_run_artifacts.py

@@ -250,6 +250,7 @@ def validate_llm(
     issues: list[str] = []
     request_count = 0
     response_count = 0
+    completed_candidates: list[tuple[int, str, bool]] = []
     error_count = 0
     usage_count = 0
     for index, row in enumerate(rows, start=1):
@@ -266,6 +267,27 @@ def validate_llm(
             body = response.get("body")
             if isinstance(body, dict):
                 response_count += 1
+                response_status = body.get("status")
+                completed = (
+                    response.get("status_code") == 200
+                    and response_status in {None, "completed"}
+                    and not body.get("incomplete_details")
+                )
+                if completed:
+                    signature = json.dumps(
+                        (
+                            request.get("body")
+                            if isinstance(request, dict)
+                            and isinstance(request.get("body"), dict)
+                            else {}
+                        ),
+                        sort_keys=True,
+                        separators=(",", ":"),
+                        default=str,
+                    )
+                    completed_candidates.append(
+                        (index - 1, signature, has_token_usage(body))
+                    )
                 if has_token_usage(body):
                     usage_count += 1
             else:
@@ -282,14 +304,153 @@ def validate_llm(
         )
     if usage_count == 0:
         issues.append(f"{path}: no provider token usage in response bodies")
+    candidates_by_request: dict[str, list[tuple[int, bool]]] = {}
+    for index, signature, has_usage in completed_candidates:
+        candidates_by_request.setdefault(signature, []).append((index, has_usage))
+    request_bodies = [
+        json.dumps(
+            (
+                request.get("body")
+                if isinstance(request := row.get("request"), dict)
+                and isinstance(request.get("body"), dict)
+                else {}
+            ),
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+        )
+        for row in rows
+    ]
+    selected: list[tuple[int, bool]] = []
+    for candidates in candidates_by_request.values():
+        consumed = [
+            candidate
+            for candidate in candidates
+            if response_consumed_by_later_request(rows, request_bodies, candidate[0])
+        ]
+        if consumed:
+            selected.append(max(consumed))
+            continue
+        safe_unconsumed = [
+            candidate
+            for candidate in candidates
+            if response_safe_without_later_consumption(rows, candidate[0])
+        ]
+        if safe_unconsumed:
+            selected.append(max(safe_unconsumed))
+    selected.sort()
+    successful_exchange_indices = [index for index, _ in selected]
+    successful_response_count = len(selected)
+    successful_usage_count = sum(1 for _, has_usage in selected if has_usage)
+    if successful_response_count and successful_usage_count < successful_response_count:
+        issues.append(
+            f"{path}: completed responses with usage {successful_usage_count} < "
+            f"completed responses {successful_response_count}"
+        )
     return issues, {
         "requests": request_count,
         "responses": response_count,
+        "successful_responses": successful_response_count,
+        "successful_exchange_indices": successful_exchange_indices,
+        "successful_responses_with_usage": successful_usage_count,
+        "deduplicated_completed_responses": (
+            len(completed_candidates) - successful_response_count
+        ),
         "errors": error_count,
         "responses_with_usage": usage_count,
     }
 
 
+def response_consumed_by_later_request(
+    rows: list[dict[str, Any]],
+    request_bodies: list[str],
+    exchange_idx: int,
+) -> bool:
+    response = rows[exchange_idx].get("response")
+    body = response.get("body") if isinstance(response, dict) else {}
+    call_ids = response_call_ids(body if isinstance(body, dict) else {})
+    return bool(
+        call_ids
+        and any(
+            any(call_id in request_body for call_id in call_ids)
+            for request_body in request_bodies[exchange_idx + 1 :]
+        )
+    )
+
+
+def response_call_ids(body: dict[str, Any]) -> set[str]:
+    return {call_id for call_id, _ in response_tool_calls(body)}
+
+
+def response_safe_without_later_consumption(
+    rows: list[dict[str, Any]], exchange_idx: int
+) -> bool:
+    row = rows[exchange_idx]
+    response = row.get("response")
+    body = response.get("body") if isinstance(response, dict) else {}
+    calls = response_tool_calls(body if isinstance(body, dict) else {})
+    if not calls or all(name == "finish" for _, name in calls):
+        return True
+    return not any(
+        isinstance(request := later.get("request"), dict)
+        and isinstance(request.get("body"), dict)
+        for later in rows[exchange_idx + 1 :]
+    )
+
+
+def response_tool_calls(body: dict[str, Any]) -> list[tuple[str, str | None]]:
+    calls = [
+        (
+            str(item.get("call_id") or item.get("id")),
+            str(item.get("name")) if item.get("name") else None,
+        )
+        for item in body.get("output") or []
+        if isinstance(item, dict)
+        and item.get("type") in {"function_call", "tool_call"}
+        and (item.get("call_id") or item.get("id"))
+    ]
+    choices = body.get("choices")
+    if isinstance(choices, list):
+        for choice in choices:
+            message = choice.get("message") if isinstance(choice, dict) else None
+            if not isinstance(message, dict):
+                continue
+            calls.extend(
+                (
+                    str(call.get("id") or call.get("tool_call_id")),
+                    (
+                        str(function.get("name"))
+                        if isinstance(function := call.get("function"), dict)
+                        and function.get("name")
+                        else str(call.get("name"))
+                        if call.get("name")
+                        else None
+                    ),
+                )
+                for call in message.get("tool_calls") or []
+                if isinstance(call, dict)
+                and (call.get("id") or call.get("tool_call_id"))
+            )
+    message = body.get("message")
+    if isinstance(message, dict):
+        calls.extend(
+            (
+                str(call.get("id") or call.get("tool_call_id")),
+                (
+                    str(function.get("name"))
+                    if isinstance(function := call.get("function"), dict)
+                    and function.get("name")
+                    else str(call.get("name"))
+                    if call.get("name")
+                    else None
+                ),
+            )
+            for call in message.get("tool_calls") or []
+            if isinstance(call, dict) and (call.get("id") or call.get("tool_call_id"))
+        )
+    return calls
+
+
 def result_has_terminal_error(result: dict[str, Any]) -> bool:
     if result.get("error") or result.get("verifier_error"):
         return True
@@ -446,10 +607,20 @@ def validate_results_row(
                 )
         trajectory = row.get("trajectory")
         if isinstance(trajectory, list):
-            if llm_summary and llm_summary.get("responses", 0) and len(trajectory) == 0:
+            successful_responses = (
+                int(llm_summary.get("successful_responses", 0)) if llm_summary else 0
+            )
+            if successful_responses and len(trajectory) != successful_responses:
                 issues.append(
-                    f"{row_path}: no results trajectory steps from LLM exchanges"
+                    f"{row_path}: results trajectory steps {len(trajectory)} != "
+                    f"successful LLM responses {successful_responses}"
                 )
+            expected_indices = (
+                set(llm_summary.get("successful_exchange_indices", []))
+                if llm_summary
+                else set()
+            )
+            observed_indices: set[int] = set()
             for idx, step in enumerate(trajectory):
                 if not isinstance(step, dict):
                     issues.append(f"{row_path}: trajectory[{idx}] must be an object")
@@ -468,6 +639,34 @@ def validate_results_row(
                     issues.append(
                         f"{row_path}: trajectory[{idx}] source is not llm_trajectory"
                     )
+                if isinstance(extras, dict) and isinstance(
+                    extras.get("exchange_index"), int
+                ):
+                    observed_indices.add(extras["exchange_index"])
+                if step.get("is_truncated") is True:
+                    issues.append(
+                        f"{row_path}: trajectory[{idx}] is incorrectly truncated"
+                    )
+                step_messages = []
+                for field in ("prompt", "completion"):
+                    value = step.get(field)
+                    if isinstance(value, list):
+                        step_messages.extend(
+                            message for message in value if isinstance(message, dict)
+                        )
+                issues.extend(
+                    validate_training_messages(
+                        step_messages,
+                        tools=tools,
+                        row_path=f"{row_path}: trajectory[{idx}]",
+                    )
+                )
+            if expected_indices and observed_indices != expected_indices:
+                issues.append(
+                    f"{row_path}: results exchange indices "
+                    f"{sorted(observed_indices)} != successful LLM exchange indices "
+                    f"{sorted(expected_indices)}"
+                )
         token_usage = row.get("token_usage")
         if not isinstance(token_usage, dict):
             issues.append(f"{row_path}: missing object token_usage")

pyproject.toml

@@ -14,7 +14,7 @@ dependencies = [
     "rich>=13.0",
     # Dataset registry bench_version checks (PEP 440 specifier sets).
     "packaging>=24",
-    "litellm[proxy]==1.89.0",
+    "litellm[proxy]==1.91.0",
     "tomli-w>=1.0",
     "typer>=0.9",
 ]

src/benchflow/acp/runtime.py

@@ -25,6 +25,7 @@
 
 from benchflow.acp.client import ACPClient
 from benchflow.acp.container_transport import ContainerTransport
+from benchflow.acp.selection import selected_acp_transport
 from benchflow.acp.types import McpServerSpec
 from benchflow.agents.protocol import ACPSessionAdapter
 from benchflow.agents.providers import (
@@ -38,8 +39,13 @@
     AgentPromptTimeoutError,
     IdleTimeoutDiagnostic,
     IdleTimeoutError,
+    TransportClosedDiagnostic,
+    TransportClosedError,
+)
+from benchflow.sandbox.lockdown import (
+    build_priv_drop_cmd,
+    enforce_agent_egress_firewall,
 )
-from benchflow.sandbox.lockdown import build_priv_drop_cmd
 from benchflow.sandbox.process import DaytonaProcess, DaytonaPtyProcess, DockerProcess
 from benchflow.trajectories._capture import _capture_session_trajectory
 
@@ -51,6 +57,7 @@
     "IdleTimeoutError",
     "connect_acp",
     "execute_prompts",
+    "selected_acp_transport",
 ]
 
 logger = logging.getLogger(__name__)
@@ -59,6 +66,24 @@
 _ACP_CONNECT_MAX_RETRIES = 3
 _ACP_CONNECT_BASE_DELAY = 2.0
 _PROMPT_CANCEL_DRAIN_TIMEOUT_SEC = 0.25
+_ACP_HANDSHAKE_TIMEOUT_SEC = 60
+
+
+async def _wait_for_acp_handshake(awaitable, *, phase: str):
+    try:
+        return await asyncio.wait_for(awaitable, timeout=_ACP_HANDSHAKE_TIMEOUT_SEC)
+    except TimeoutError as e:
+        msg = (
+            f"ACP {phase} timed out after {_ACP_HANDSHAKE_TIMEOUT_SEC}s "
+            "before the first prompt"
+        )
+        raise TransportClosedError(
+            msg,
+            TransportClosedDiagnostic(
+                raw_message=msg,
+                transport_diagnosis=f"acp_{phase}_timeout",
+            ),
+        ) from e
 
 
 # models.dev provider inference — used when acp_model_format="provider/model"
@@ -484,9 +509,13 @@ async def connect_acp(
             if environment == "docker":
                 live_proc = DockerProcess.from_sandbox_env(env)
             elif environment == "daytona":
-                if agent == "gemini":
+                transport_name = selected_acp_transport(
+                    agent=agent,
+                    environment=environment,
+                )
+                if transport_name == "ssh":
                     live_proc = await DaytonaProcess.from_sandbox_env(env)
-                    logger.info("Using SSH transport for Gemini on Daytona")
+                    logger.info("Using SSH transport for %s on Daytona", agent)
                 else:
                     live_proc = await DaytonaPtyProcess.from_sandbox_env(env)
                     logger.info("Using PTY transport for Daytona sandbox")
@@ -511,15 +540,18 @@ async def connect_acp(
             acp_client = ACPClient(transport)
             await acp_client.connect()
 
-            init_result = await asyncio.wait_for(acp_client.initialize(), timeout=60)
+            init_result = await _wait_for_acp_handshake(
+                acp_client.initialize(),
+                phase="initialize",
+            )
             agent_name = (
                 init_result.agent_info.name if init_result.agent_info else agent
             )
             logger.info(f"ACP agent: {agent_name}")
 
-            session = await asyncio.wait_for(
+            session = await _wait_for_acp_handshake(
                 acp_client.session_new(cwd=agent_cwd, mcp_servers=mcp_servers),
-                timeout=60,
+                phase="session_new",
             )
             logger.info(f"Session: {session.session_id}")
             break
@@ -552,6 +584,7 @@ async def connect_acp(
             agent_env=agent_env,
             reasoning_effort=reasoning_effort,
         )
+        await enforce_agent_egress_firewall(env, sandbox_user, agent_env)
     except Exception:
         with contextlib.suppress(Exception):
             await acp_client.close()

src/benchflow/acp/selection.py

@@ -0,0 +1,30 @@
+"""Pure ACP transport selection helpers shared by runtime and artifacts."""
+
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+_DAYTONA_ACP_TRANSPORT_ENV = "BENCHFLOW_DAYTONA_ACP_TRANSPORT"
+_DAYTONA_ACP_TRANSPORTS = {"pty", "ssh"}
+
+
+def daytona_acp_transport() -> str:
+    value = os.environ.get(_DAYTONA_ACP_TRANSPORT_ENV, "pty").strip().lower()
+    if value in _DAYTONA_ACP_TRANSPORTS:
+        return value
+    logger.warning(
+        "Invalid %s=%r; using PTY transport",
+        _DAYTONA_ACP_TRANSPORT_ENV,
+        value,
+    )
+    return "pty"
+
+
+def selected_acp_transport(*, agent: str, environment: str) -> str:
+    """Return the concrete agent transport selected for artifact provenance."""
+    if environment == "docker":
+        return "docker-stdio"
+    if environment == "daytona":
+        return "ssh" if agent == "gemini" else daytona_acp_transport()
+    return "provider-default"