Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CLI-COMMANDS.md
Original file line number Diff line number Diff line change
Expand Up @@ -494,6 +494,8 @@ roboflow workflow-eval create --name "Answer accuracy" --spec <spec-id> --datase
# Bind a Workflow, run it, and read results.
roboflow workflow-eval binding suggest --spec <spec-id> --dataset <dataset-id> --workflow <workflow-id> --json
roboflow workflow-eval run start --eval <eval-id> --body @run.json --wait
# --wait gives up after 3600 s; Runs on a local inference server wait without limit. --timeout 0 = no limit.
roboflow workflow-eval run get <run-id> --eval <eval-id> --wait --timeout 0
roboflow workflow-eval execution overview <execution-id> --eval <eval-id> --run <run-id> --json
roboflow workflow-eval execution results <execution-id> --eval <eval-id> --run <run-id> --failed-check
roboflow workflow-eval compare --eval <eval-id> -x <execution-a> -x <execution-b> --json
Expand Down
85 changes: 78 additions & 7 deletions roboflow/cli/handlers/workflow_eval.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import json
import os
import sys
import time
from typing import Annotated, Any, Callable, Dict, List, Optional

import typer
Expand Down Expand Up @@ -894,23 +895,93 @@
# ---------------------------------------------------------------------------

_WaitOpt = Annotated[bool, typer.Option("--wait", help="Poll until the Run reaches a terminal state")]
_TimeoutOpt = Annotated[float, typer.Option("--timeout", help="Seconds to wait before giving up")]
_TimeoutOpt = Annotated[
Optional[float],
typer.Option(
"--timeout",
help="Seconds to wait before giving up; 0 = no limit "
"(default: no limit for Runs on a local inference server, 3600 otherwise)",
),
]
_IntervalOpt = Annotated[float, typer.Option("--interval", help="Seconds between polls")]

_SERVER_RUN_WAIT_TIMEOUT = 3600.0
_PROGRESS_EVERY = 10.0


def _run_id(admission: Dict[str, Any]) -> Optional[str]:
run = admission.get("run")
return run.get("id") if isinstance(run, dict) else admission.get("id")


def _then_wait(args: Any, eval_id: str, timeout: float, interval: float) -> Callable[[Any, Dict[str, Any]], Any]:
def _run_field(run: Dict[str, Any], key: str) -> Any:
nested = run.get("run")
return run.get(key) or (nested.get(key) if isinstance(nested, dict) else None)


def _progress_line(run_id: str, run: Dict[str, Any]) -> str:
parts = [f"Run {run_id}", str(_run_field(run, "state") or "unknown")]
if run.get("selectedCaseCount") is not None:
parts.append(f"{run['selectedCaseCount']} cases")
counts = run.get("childStateCounts")
if isinstance(counts, dict) and counts:
parts.append("executions " + ", ".join(f"{state} {count}" for state, count in sorted(counts.items())))
if run.get("lastActivityAt"):
parts.append(f"last activity {run['lastActivityAt']}")
return " · ".join(parts)


def _wait_for_run(
args: Any,
client: Any,
eval_id: str,
run_id: str,
timeout: Optional[float],
interval: float,
execution_mode: Optional[str] = None,
) -> Any:
"""Poll a Run to a terminal state, printing throttled progress on stderr.

Without an explicit ``timeout`` the limit is resolved from the Run's
``executionMode`` (from the admission payload, else the first poll): Runs on a
local inference server (``browser``) wait without limit, server Runs keep 3600 s.
"""
from roboflow.core.workflow_evals import RUN_TERMINAL_STATES

show_progress = not args.json and not args.quiet
started = time.monotonic()
limit = timeout
last_progress: Optional[float] = None

def on_poll(run: Dict[str, Any]) -> None:
nonlocal limit, last_progress
if limit is None:
mode = execution_mode or _run_field(run, "executionMode")
limit = 0 if mode == "browser" else _SERVER_RUN_WAIT_TIMEOUT
state = _run_field(run, "state")
if state in RUN_TERMINAL_STATES:
return
now = time.monotonic()
if show_progress and (last_progress is None or now - last_progress >= _PROGRESS_EVERY):
print(_progress_line(run_id, run), file=sys.stderr)
Comment thread
joaomarcoscrs marked this conversation as resolved.
Dismissed
last_progress = now
if limit and now - started >= limit:
raise TimeoutError(f"Run {run_id} is still {state} after {limit:g}s")

return client.wait_for_run(eval_id, run_id, timeout=0, interval=interval, on_poll=on_poll)


def _then_wait(
args: Any, eval_id: str, timeout: Optional[float], interval: float
) -> Callable[[Any, Dict[str, Any]], Any]:
def wait(client: Any, admission: Dict[str, Any]) -> Any:
run_id = _run_id(admission)
if not run_id:
return admission
if not args.json and not args.quiet:
print(f"Waiting for Run {run_id}…", file=sys.stderr)
return client.wait_for_run(eval_id, run_id, timeout=timeout, interval=interval)
mode = _run_field(admission, "executionMode")
return _wait_for_run(args, client, eval_id, run_id, timeout, interval, mode)

return wait

Expand Down Expand Up @@ -938,13 +1009,13 @@
run_id: Annotated[str, typer.Argument(help="Run id")],
eval_id: EvalOpt,
wait: _WaitOpt = False,
timeout: _TimeoutOpt = 3600,
timeout: _TimeoutOpt = None,
interval: _IntervalOpt = 5,
) -> None:
"""Show a Run's lifecycle, frozen setup and result availability."""
args = ctx_to_args(ctx)
if wait:
_invoke(args, lambda c: c.wait_for_run(eval_id, run_id, timeout=timeout, interval=interval))
_invoke(args, lambda c: _wait_for_run(args, c, eval_id, run_id, timeout, interval))
else:
_invoke(args, lambda c: c.get_run(eval_id, run_id))

Expand All @@ -960,7 +1031,7 @@
),
],
wait: _WaitOpt = False,
timeout: _TimeoutOpt = 3600,
timeout: _TimeoutOpt = None,
interval: _IntervalOpt = 5,
idempotency_key: IdempotencyOpt = None,
) -> None:
Expand Down Expand Up @@ -1103,7 +1174,7 @@
body: Annotated[
str,
typer.Option(
"--body", "-b", help="JSON {attemptCount, shardIndex, output, runtime?}: inline, @file, path, or -"
"--body", "-b", help="JSON {attemptCount, shardIndex, output, runtime}: inline, @file, path, or -"
),
],
) -> None:
Expand Down
30 changes: 22 additions & 8 deletions roboflow/core/workflow_evals.py
Original file line number Diff line number Diff line change
Expand Up @@ -505,24 +505,26 @@ def wait_for_run(
eval_id: str,
run_id: str,
*,
timeout: float = 3600,
timeout: Optional[float] = 3600,
interval: float = 5,
on_poll: Optional[Callable[[Dict[str, Any]], None]] = None,
) -> Dict[str, Any]:
"""Poll a Run until it reaches a terminal state and return it.

Stopping the wait never cancels the Run. Raises ``TimeoutError`` when
``timeout`` seconds elapse first.
``timeout`` seconds elapse first; ``None`` or ``0`` waits without a limit
(Runs on a local inference server can take hours). ``on_poll`` receives
every polled Run.
"""
deadline = time.monotonic() + timeout
deadline = time.monotonic() + timeout if timeout else None
while True:
run = self.get_run(eval_id, run_id)
if on_poll:
on_poll(run)
state = run.get("state") or (run.get("run") or {}).get("state")
if state in RUN_TERMINAL_STATES:
return run
if time.monotonic() >= deadline:
if deadline is not None and time.monotonic() >= deadline:
raise TimeoutError(f"Run {run_id} is still {state} after {timeout:g}s")
time.sleep(interval)

Expand Down Expand Up @@ -554,15 +556,27 @@ def list_execution_input_cases(
)

def submit_capture(
self, eval_id: str, run_id: str, execution_id: str, case_id: str, body: Dict[str, Any]
self,
eval_id: str,
run_id: str,
execution_id: str,
case_id: str,
body: Dict[str, Any],
*,
timeout: Optional[float] = None,
) -> Dict[str, Any]:
"""Submit one locally captured Workflow output for server-side scoring.

``body`` is ``{attemptCount, shardIndex, output, runtime?}`` or
``{attemptCount, shardIndex, resume: true}``.
``body`` is ``{attemptCount, shardIndex, output, runtime}`` where ``runtime``
is required and shaped ``{latencyMs?, error?, timedOut?, retryable?}``, or
``{attemptCount, shardIndex, resume: true}``. Scoring runs inside this
request, so ``timeout`` (seconds) overrides the adapter's default HTTP timeout.
"""
return self._call(
"POST", f"{self._execution_path(eval_id, run_id, execution_id)}/cases/{_seg(case_id)}/capture", body=body
"POST",
f"{self._execution_path(eval_id, run_id, execution_id)}/cases/{_seg(case_id)}/capture",
body=body,
**({"timeout": timeout} if timeout is not None else {}),
)

def get_results_overview(
Expand Down
81 changes: 79 additions & 2 deletions tests/cli/test_workflow_eval_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import os
import tempfile
import unittest
from unittest.mock import patch
from unittest.mock import ANY, patch

from typer.testing import CliRunner

Expand Down Expand Up @@ -272,7 +272,7 @@ def test_run_start_with_wait_polls_the_admitted_run(self, mock_start, mock_wait)
result = _invoke("--json", "workflow-eval", "run", "start", "-e", "e1", "--body", body, "--wait")
self.assertEqual(result.exit_code, 0, result.output)
self.assertEqual(json.loads(result.output)["state"], "completed")
mock_wait.assert_called_once_with("e1", "r1", timeout=3600, interval=5)
mock_wait.assert_called_once_with("e1", "r1", timeout=0, interval=5, on_poll=ANY)

@patch(f"{CLIENT}.list_case_results")
def test_execution_results_filters(self, mock_results) -> None:
Expand Down Expand Up @@ -341,5 +341,82 @@ def test_spec_ai_draft(self, mock_draft) -> None:
self.assertEqual(mock_draft.call_args.args, ("e1", "be strict"))


class TestWorkflowEvalRunWait(unittest.TestCase):
"""``run get/start --wait`` with a fake clock: ``sleep`` advances ``monotonic``."""

def _wait(self, *argv: str, runs: list, interval: str = "3600", timeout: str | None = None, admission=None):
clock = [0.0]

def sleep(seconds: float) -> None:
clock[0] += seconds

with (
patch(f"{CLIENT}.get_run", side_effect=runs) as get_run,
patch("time.monotonic", side_effect=lambda: clock[0]),
patch("time.sleep", side_effect=sleep),
patch(f"{CLIENT}.start_run", return_value=admission),
):
if admission is None:
command = ["workflow-eval", "run", "get", "r1", "-e", "e1"]
else:
command = ["workflow-eval", "run", "start", "-e", "e1", "--body", '{"executions": []}']
command += ["--wait", "--interval", interval, *(["--timeout", timeout] if timeout else [])]
result = _invoke(*argv, *command)
return result, get_run.call_count

@staticmethod
def _runs(mode: str, polls: int, **extra) -> list:
running = {"id": "r1", "state": "running", "executionMode": mode, "selectedCaseCount": 100, **extra}
return [running] * polls + [{"id": "r1", "state": "completed", "executionMode": mode}]

def test_browser_run_without_timeout_waits_past_the_old_limit(self) -> None:
result, polls = self._wait("--json", runs=self._runs("browser", 4))
self.assertEqual(result.exit_code, 0, result.output)
self.assertEqual(json.loads(result.stdout)["state"], "completed")
self.assertEqual(polls, 5)

def test_server_run_without_timeout_keeps_3600(self) -> None:
result, polls = self._wait("--json", runs=self._runs("server", 4))
self.assertEqual(result.exit_code, 1)
self.assertIn("still running after 3600s", result.stderr)
self.assertEqual(polls, 2)

def test_timeout_zero_waits_without_limit(self) -> None:
result, polls = self._wait("--json", runs=self._runs("server", 4), timeout="0")
self.assertEqual(result.exit_code, 0, result.output)
self.assertEqual(polls, 5)

def test_explicit_timeout_applies_to_browser_runs(self) -> None:
result, _ = self._wait("--json", runs=self._runs("browser", 4), interval="30", timeout="60")
self.assertEqual(result.exit_code, 1)
self.assertIn("still running after 60s", result.stderr)

def test_run_start_uses_the_admission_execution_mode(self) -> None:
admission = {"run": {"id": "r1", "state": "queued"}, "executionMode": "browser"}
runs = self._runs("", 4) # polled Runs lack executionMode: the admission decides
result, polls = self._wait("--json", runs=runs, admission=admission)
self.assertEqual(result.exit_code, 0, result.output)
self.assertEqual(polls, 5)

def test_progress_is_throttled_to_one_line_every_10s(self) -> None:
runs = self._runs("browser", 25, childStateCounts={"running": 1}, lastActivityAt="2026-10-22T10:00:00Z")
result, _ = self._wait(runs=runs, interval="1")
self.assertEqual(result.exit_code, 0, result.output)
lines = [line for line in result.stderr.splitlines() if line.startswith("Run r1")]
self.assertEqual(len(lines), 3) # t = 0, 10, 20
self.assertEqual(
lines[0], "Run r1 · running · 100 cases · executions running 1 · last activity 2026-10-22T10:00:00Z"
)

def test_progress_is_suppressed_with_json_and_quiet(self) -> None:
for flag in ("--json", "--quiet"):
with self.subTest(flag=flag):
result, _ = self._wait(flag, runs=self._runs("browser", 25), interval="1")
self.assertEqual(result.exit_code, 0, result.output)
self.assertNotIn("Run r1 ·", result.stderr)
if flag == "--json":
self.assertEqual(json.loads(result.stdout)["state"], "completed")


if __name__ == "__main__":
unittest.main()
35 changes: 35 additions & 0 deletions tests/test_workflow_evals.py
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,41 @@ def test_wait_for_run_polls_until_terminal(self, _sleep):
self.assertEqual(run["state"], "completed")
self.assertEqual(len(responses.calls), 2)

def _poll_hours(self, timeout, polls=4):
"""Wait on a Run that stays running for ``polls`` polls one hour apart."""
clock = [0.0]
runs = [{"id": R, "state": "running"}] * polls + [{"id": R, "state": "completed"}]
seen = []

def sleep(seconds):
clock[0] += seconds

with (
patch.object(self.client, "get_run", side_effect=runs),
patch("roboflow.core.workflow_evals.time.monotonic", side_effect=lambda: clock[0]),
patch("roboflow.core.workflow_evals.time.sleep", side_effect=sleep),
):
return self.client.wait_for_run(E, R, timeout=timeout, interval=3600, on_poll=seen.append), seen

def test_wait_for_run_without_limit_keeps_polling_until_terminal(self):
for timeout in (0, None):
with self.subTest(timeout=timeout):
run, seen = self._poll_hours(timeout)
self.assertEqual(run["state"], "completed")
self.assertEqual([r["state"] for r in seen], ["running"] * 4 + ["completed"])

def test_wait_for_run_positive_timeout_still_raises(self):
with self.assertRaisesRegex(TimeoutError, "still running after 7200s"):
self._poll_hours(7200)

def test_submit_capture_forwards_timeout(self):
body = {"attemptCount": 1, "shardIndex": 0, "output": {}, "runtime": {"latencyMs": 5}}
with patch("roboflow.core.workflow_evals.api.request", return_value={}) as request:
self.client.submit_capture(E, R, X, C, body, timeout=300)
self.assertEqual(request.call_args.kwargs["timeout"], 300)
self.client.submit_capture(E, R, X, C, body)
self.assertNotIn("timeout", request.call_args.kwargs)


class TestSseParser(unittest.TestCase):
def test_multiline_data_and_plain_text(self):
Expand Down
Loading