diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 11be906d..9910802c 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -494,6 +494,8 @@ roboflow workflow-eval create --name "Answer accuracy" --spec --datase # Bind a Workflow, run it, and read results. roboflow workflow-eval binding suggest --spec --dataset --workflow --json roboflow workflow-eval run start --eval --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 --eval --wait --timeout 0 roboflow workflow-eval execution overview --eval --run --json roboflow workflow-eval execution results --eval --run --failed-check roboflow workflow-eval compare --eval -x -x --json diff --git a/roboflow/cli/handlers/workflow_eval.py b/roboflow/cli/handlers/workflow_eval.py index 3a933e65..1a270743 100644 --- a/roboflow/cli/handlers/workflow_eval.py +++ b/roboflow/cli/handlers/workflow_eval.py @@ -18,6 +18,7 @@ import json import os import sys +import time from typing import Annotated, Any, Callable, Dict, List, Optional import typer @@ -894,23 +895,93 @@ def preparation_cases( # --------------------------------------------------------------------------- _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) + 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 @@ -938,13 +1009,13 @@ def get_run( 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)) @@ -960,7 +1031,7 @@ def start_run( ), ], wait: _WaitOpt = False, - timeout: _TimeoutOpt = 3600, + timeout: _TimeoutOpt = None, interval: _IntervalOpt = 5, idempotency_key: IdempotencyOpt = None, ) -> None: @@ -1103,7 +1174,7 @@ def submit_capture( 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: diff --git a/roboflow/core/workflow_evals.py b/roboflow/core/workflow_evals.py index 00fc842b..09fb65dd 100644 --- a/roboflow/core/workflow_evals.py +++ b/roboflow/core/workflow_evals.py @@ -505,16 +505,18 @@ 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: @@ -522,7 +524,7 @@ def wait_for_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) @@ -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( diff --git a/tests/cli/test_workflow_eval_handler.py b/tests/cli/test_workflow_eval_handler.py index c5d85a8f..db8c23e2 100644 --- a/tests/cli/test_workflow_eval_handler.py +++ b/tests/cli/test_workflow_eval_handler.py @@ -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 @@ -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: @@ -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() diff --git a/tests/test_workflow_evals.py b/tests/test_workflow_evals.py index 4aca881b..47b61ff7 100644 --- a/tests/test_workflow_evals.py +++ b/tests/test_workflow_evals.py @@ -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):