diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 2a7667af..2767ae68 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -457,6 +457,42 @@ roboflow eval compare --project my-project --version 3 --frontier-metric mAP5095 Reads existing evaluations; does not start new ones. Requires `model-eval:read` and workspace Model Evaluation access. +### Evaluate Workflows (Workflow Evals) + +```bash +# Discover the engine catalog and what your key may do. +roboflow workflow-eval capabilities --json +roboflow workflow-eval evaluator list +roboflow workflow-eval schema get spec --json + +# Author: Spec, Eval Dataset, Cases, Eval. +roboflow workflow-eval spec create --body @spec.json --name "Boolean answer" +roboflow workflow-eval dataset create --body @dataset.json +roboflow workflow-eval case upload ./image.png --dataset # prints artifactId +roboflow workflow-eval case add --dataset --body @case.json +roboflow workflow-eval case import --dataset --input-field image --from-dataset my-project:valid +roboflow workflow-eval create --name "Answer accuracy" --spec --dataset + +# 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 +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 +roboflow workflow-eval export start --eval --run --format csv --wait + +# Agent guidance served by the installed engine. +roboflow workflow-eval agent manifest --json +roboflow workflow-eval agent skill skill:create-eval +``` + +`--body` accepts inline JSON, `@file.json`, a file path, or `-` for stdin. Commands that create +resources or start work send a fresh `Idempotency-Key` (override with `--idempotency-key` to retry +safely); updates take the resource's current `--revision`. Deleting an Eval or Run shows its impact +and asks for confirmation (`--yes` to skip). Requires the Workflow Evals feature and the +`workflow-evals:read|write|run|export` scopes; running saved Workflows also needs `workflow:read` +and `model:infer`. The same API is available in Python via `rf.workspace().workflow_evals()`. + ### Workspace stats and billing ```bash @@ -560,6 +596,7 @@ Version numbers are always numeric — that's how `x/y` is disambiguated between | `device` | List, get, create, and observe RFDM devices (v2 deployment API) | | `eval` | Inspect model evaluation runs (mAP, confusion matrix, recommendations, ...) | | `workflow` | Manage workflows | +| `workflow-eval` | Evaluate Workflows: Specs, Eval Datasets, Cases, Runs, results, exports | | `folder` | Manage workspace folders | | `annotation` | Annotation batches and jobs | | `autolabel` | Auto-label batches with hosted foundation or Roboflow models | diff --git a/roboflow/adapters/workflowevalsapi.py b/roboflow/adapters/workflowevalsapi.py new file mode 100644 index 00000000..74c75e87 --- /dev/null +++ b/roboflow/adapters/workflowevalsapi.py @@ -0,0 +1,263 @@ +"""Transport for the public Workflow Evals API. + +Every route lives under ``{API_URL}/workspaces/{workspace}/workflow-evals``. +This module owns the HTTP mechanics shared by all of them — bearer auth, +``Idempotency-Key`` / ``If-Match`` / ``Deletion-Key`` headers, the +Workflow Evals error envelope, SSE parsing for AI drafting and signed +uploads. The per-endpoint surface lives in +:class:`roboflow.core.workflow_evals.WorkflowEvals`. +""" + +from __future__ import annotations + +import json +import uuid +from typing import Any, Dict, Iterator, Optional +from urllib.parse import quote + +import requests + +from roboflow.adapters.rfapi import RoboflowError +from roboflow.config import API_URL + +DEFAULT_TIMEOUT = 60 + + +class WorkflowEvalError(RoboflowError): + """Error returned by the Workflow Evals API. + + The server responds with ``{"error": {code, category, retryable, + requestId, message, details?, resource?}}``. Authentication failures may + use the platform ``{"error": {message, type, hint?}}`` envelope instead, + in which case only ``message`` (and ``hint``) are populated. + """ + + def __init__( + self, + message: str, + status_code: Optional[int] = None, + *, + code: Optional[str] = None, + category: Optional[str] = None, + retryable: Optional[bool] = None, + request_id: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + resource: Optional[Dict[str, Any]] = None, + hint: Optional[str] = None, + body: Any = None, + ) -> None: + super().__init__(message, status_code=status_code) + self.message = message + self.code = code + self.category = category + self.retryable = retryable + self.request_id = request_id + self.details = details or {} + self.resource = resource + self.hint = hint + self.body = body + + def to_dict(self) -> Dict[str, Any]: + """Return the error as the CLI's ``{"message": ..., ...}`` payload.""" + payload: Dict[str, Any] = {"message": self.message} + for key, value in ( + ("status", self.status_code), + ("code", self.code), + ("category", self.category), + ("retryable", self.retryable), + ("requestId", self.request_id), + ("resource", self.resource), + ("hint", self.hint), + ): + if value is not None: + payload[key] = value + if self.details: + payload["details"] = self.details + return payload + + +def error_from_response(response: requests.Response) -> WorkflowEvalError: + """Translate a non-2xx response into a :class:`WorkflowEvalError`.""" + body: Any = None + try: + body = response.json() + except ValueError: + pass + error = body.get("error") if isinstance(body, dict) else None + if isinstance(error, dict): + return WorkflowEvalError( + str(error.get("message") or error.get("code") or response.reason or "Request failed"), + status_code=response.status_code, + code=error.get("code"), + category=error.get("category") or error.get("type"), + retryable=error.get("retryable"), + request_id=error.get("requestId"), + details=error.get("details") if isinstance(error.get("details"), dict) else None, + resource=error.get("resource"), + hint=error.get("hint"), + body=body, + ) + if isinstance(error, str): + message = body.get("message") or error + return WorkflowEvalError(str(message), status_code=response.status_code, code=error, body=body) + text = (response.text or "").strip() or response.reason or "Request failed" + return WorkflowEvalError(text, status_code=response.status_code, body=body) + + +def new_idempotency_key() -> str: + """Return a fresh UUID v4, the format the API requires for ``Idempotency-Key``.""" + return str(uuid.uuid4()) + + +def base_url(workspace_url: str) -> str: + return f"{API_URL}/workspaces/{quote(workspace_url, safe='')}/workflow-evals" + + +def segment(value: Any) -> str: + """Percent-encode one path segment (resource IDs like ``skill:create-eval``).""" + return quote(str(value), safe="") + + +def _headers( + api_key: str, + *, + idempotency_key: Optional[str] = None, + revision: Optional[Any] = None, + deletion_key: Optional[str] = None, + accept: str = "application/json", +) -> Dict[str, str]: + headers = {"Authorization": f"Bearer {api_key}", "Accept": accept} + if idempotency_key: + headers["Idempotency-Key"] = idempotency_key + if revision is not None: + headers["If-Match"] = str(revision) + if deletion_key: + headers["Deletion-Key"] = deletion_key + return headers + + +def _clean_params(params: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + if not params: + return None + cleaned: Dict[str, Any] = {} + for key, value in params.items(): + if value is None: + continue + if isinstance(value, bool): + cleaned[key] = "true" if value else "false" + elif isinstance(value, (list, tuple)): + if value: + cleaned[key] = list(value) + else: + cleaned[key] = value + return cleaned or None + + +def request( + api_key: str, + workspace_url: str, + method: str, + path: str = "", + *, + params: Optional[Dict[str, Any]] = None, + body: Any = None, + idempotency_key: Optional[str] = None, + revision: Optional[Any] = None, + deletion_key: Optional[str] = None, + timeout: float = DEFAULT_TIMEOUT, +) -> Any: + """Call one Workflow Evals route and return the decoded JSON body. + + Returns ``None`` for ``204 No Content``. Raises :class:`WorkflowEvalError` + for any non-2xx response. + """ + response = requests.request( + method.upper(), + f"{base_url(workspace_url)}{path}", + params=_clean_params(params), + json=body, + headers=_headers(api_key, idempotency_key=idempotency_key, revision=revision, deletion_key=deletion_key), + timeout=timeout, + ) + if not 200 <= response.status_code < 300: + raise error_from_response(response) + if response.status_code == 204 or not response.content: + return None + try: + return response.json() + except ValueError: + raise WorkflowEvalError( + f"Expected JSON from {method.upper()} {path or '/'}", status_code=response.status_code + ) from None + + +def stream_events( + api_key: str, + workspace_url: str, + path: str, + *, + body: Any, + idempotency_key: Optional[str] = None, + timeout: float = 600, +) -> Iterator[Dict[str, Any]]: + """POST to an SSE route and yield ``{"event": name, "data": payload}`` dicts. + + ``data`` is decoded as JSON when possible, otherwise left as text. The + stream ends after the server's terminal ``done`` event. + """ + response = requests.post( + f"{base_url(workspace_url)}{path}", + json=body, + headers=_headers(api_key, idempotency_key=idempotency_key, accept="text/event-stream"), + stream=True, + timeout=timeout, + ) + if not 200 <= response.status_code < 300: + raise error_from_response(response) + try: + # Iterate raw bytes: the server omits a charset, so requests would decode the + # stream as ISO-8859-1. parse_sse decodes each line as UTF-8. + yield from parse_sse(response.iter_lines()) + finally: + response.close() + + +def parse_sse(lines: Any) -> Iterator[Dict[str, Any]]: + """Parse ``text/event-stream`` lines into event dicts.""" + event = "message" + data_lines: list = [] + for raw in lines: + line = raw.decode("utf-8") if isinstance(raw, bytes) else raw + if line is None: + continue + if line == "": + if data_lines: + yield {"event": event, "data": _decode_data("\n".join(data_lines))} + event, data_lines = "message", [] + continue + if line.startswith(":"): + continue + field, _, value = line.partition(":") + value = value[1:] if value.startswith(" ") else value + if field == "event": + event = value + elif field == "data": + data_lines.append(value) + if data_lines: + yield {"event": event, "data": _decode_data("\n".join(data_lines))} + + +def _decode_data(data: str) -> Any: + try: + return json.loads(data) + except ValueError: + return data + + +def put_signed_upload(upload_url: str, data: bytes, headers: Dict[str, str], *, timeout: float = 300) -> None: + """PUT bytes to a signed upload URL with exactly the headers the API returned.""" + response = requests.put(upload_url, data=data, headers=headers, timeout=timeout) + if not 200 <= response.status_code < 300: + raise WorkflowEvalError( + f"Signed upload failed ({response.status_code}): {response.text}", status_code=response.status_code + ) diff --git a/roboflow/cli/__init__.py b/roboflow/cli/__init__.py index 2c2610be..d8588ad5 100644 --- a/roboflow/cli/__init__.py +++ b/roboflow/cli/__init__.py @@ -207,6 +207,7 @@ def _walk(group: Any, prefix: str = "") -> None: from roboflow.cli.handlers.video import video_app # noqa: E402 from roboflow.cli.handlers.vision_events import vision_events_app # noqa: E402 from roboflow.cli.handlers.workflow import workflow_app # noqa: E402 +from roboflow.cli.handlers.workflow_eval import workflow_eval_app # noqa: E402 from roboflow.cli.handlers.workspace import workspace_app # noqa: E402 # Register ALL commands in alphabetical order for clean --help output @@ -239,6 +240,7 @@ def _walk(group: Any, prefix: str = "") -> None: app.add_typer(video_app, name="video") app.add_typer(vision_events_app, name="vision-events") app.add_typer(workflow_app, name="workflow") +app.add_typer(workflow_eval_app, name="workflow-eval") app.add_typer(workspace_app, name="workspace") # Hidden aliases (loaded last — still functional but not in --help) diff --git a/roboflow/cli/handlers/workflow_eval.py b/roboflow/cli/handlers/workflow_eval.py new file mode 100644 index 00000000..3a933e65 --- /dev/null +++ b/roboflow/cli/handlers/workflow_eval.py @@ -0,0 +1,1392 @@ +"""Workflow Evals commands. + +Wraps the public ``/workspaces/{workspace}/workflow-evals`` API: Evals, +Specs, Eval Datasets and Cases, binding suggestion/validation, Case +preparation, grouped Runs, Execution results, comparisons, exports, +embedding analyses, the engine catalog and agent guidance. + +Request bodies follow the API's JSON contracts. Options named ``--body`` +(and ``--subject`` / ``--binding-set``) accept inline JSON, ``@path.json``, +a path to a JSON file, or ``-`` for stdin. Explicit flags are merged on top +of ``--body``. Commands that create resources or start work send a fresh +``Idempotency-Key`` unless ``--idempotency-key`` is given; updates require the +resource's current ``--revision``. +""" + +from __future__ import annotations + +import json +import os +import sys +from typing import Annotated, Any, Callable, Dict, List, Optional + +import typer + +from roboflow.cli._compat import SortedGroup, ctx_to_args + +workflow_eval_app = typer.Typer( + cls=SortedGroup, + help="Evaluate Workflows against golden datasets (Workflow Evals)", + no_args_is_help=True, +) +spec_app = typer.Typer(cls=SortedGroup, help="Author and validate evaluation Specs", no_args_is_help=True) +dataset_app = typer.Typer(cls=SortedGroup, help="Manage Eval Datasets", no_args_is_help=True) +case_app = typer.Typer(cls=SortedGroup, help="Manage Eval Dataset Cases", no_args_is_help=True) +binding_app = typer.Typer(cls=SortedGroup, help="Suggest and validate BindingSets", no_args_is_help=True) +preparation_app = typer.Typer(cls=SortedGroup, help="Prepare candidate outputs for Case labeling", no_args_is_help=True) +run_app = typer.Typer(cls=SortedGroup, help="Start and manage evaluation Runs", no_args_is_help=True) +execution_app = typer.Typer(cls=SortedGroup, help="Inspect Run Executions and results", no_args_is_help=True) +export_app = typer.Typer(cls=SortedGroup, help="Export Run results", no_args_is_help=True) +embedding_app = typer.Typer(cls=SortedGroup, help="Image embedding analyses for Executions", no_args_is_help=True) +evaluator_app = typer.Typer(cls=SortedGroup, help="Browse the evaluator catalog", no_args_is_help=True) +schema_app = typer.Typer(cls=SortedGroup, help="Read authoring JSON Schemas", no_args_is_help=True) +agent_app = typer.Typer(cls=SortedGroup, help="Read engine guidance for agents and MCP clients", no_args_is_help=True) + +for _sub_app, _name in ( + (agent_app, "agent"), + (binding_app, "binding"), + (case_app, "case"), + (dataset_app, "dataset"), + (embedding_app, "embedding"), + (evaluator_app, "evaluator"), + (execution_app, "execution"), + (export_app, "export"), + (preparation_app, "preparation"), + (run_app, "run"), + (schema_app, "schema"), + (spec_app, "spec"), +): + workflow_eval_app.add_typer(_sub_app, name=_name) + + +# --------------------------------------------------------------------------- +# Shared option types +# --------------------------------------------------------------------------- + +EvalOpt = Annotated[str, typer.Option("--eval", "-e", help="Eval id")] +RunOpt = Annotated[str, typer.Option("--run", "-r", help="Run id")] +ExecutionOpt = Annotated[str, typer.Option("--execution", "-x", help="Execution id")] +DatasetOpt = Annotated[str, typer.Option("--dataset", "-d", help="Eval Dataset id")] +RevisionOpt = Annotated[str, typer.Option("--revision", help="Current revision of the resource (sent as If-Match)")] +LimitOpt = Annotated[Optional[int], typer.Option("--limit", "-n", help="Page size")] +CursorOpt = Annotated[Optional[str], typer.Option("--cursor", help="Cursor from a previous page's nextCursor")] +BodyOpt = Annotated[ + Optional[str], typer.Option("--body", "-b", help="JSON request body: inline JSON, @file.json, file path, or -") +] +IdempotencyOpt = Annotated[ + Optional[str], + typer.Option("--idempotency-key", help="UUID v4 to reuse when retrying (default: generated)"), +] +YesOpt = Annotated[bool, typer.Option("--yes", "-y", help="Skip the confirmation prompt")] +SliceOpt = Annotated[Optional[List[str]], typer.Option("--slice", help="Filter by slice (repeatable)")] +SliceMatchOpt = Annotated[Optional[str], typer.Option("--slice-match", help="Slice match mode: any or all")] + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _client(args: Any) -> Any: + from roboflow.cli._resolver import resolve_ws_and_key + from roboflow.core.workflow_evals import WorkflowEvals + + resolved = resolve_ws_and_key(args) + if not resolved: + return None + workspace_url, api_key = resolved + return WorkflowEvals(api_key, workspace_url) + + +def _error_hint(exc: Any) -> Optional[str]: + status, code = exc.status_code, (exc.code or "") + if code == "FEATURE_DISABLED": + return "Workflow Evals is not enabled for this workspace." + if code == "deletion_confirmation_required": + return "Re-run the delete command to confirm, or pass --yes." + if status == 428: + return "Pass the resource's current --revision (shown by the matching get command)." + if status == 412: + return "The resource changed since you read it. Fetch it again and retry with the new --revision." + if status == 401: + return "Check your API key, or run 'roboflow auth login'." + if status == 403: + return ( + "The API key needs the workflow-evals:read/write/run/export scope for this operation " + "and cannot be folder-scoped." + ) + if status == 422: + return "See error.details.diagnostics for the validation failures." + return exc.hint + + +def _exit_code(exc: Any) -> int: + if exc.status_code in (401, 403): + return 2 + if exc.status_code == 404: + return 3 + return 1 + + +def _fail(args: Any, exc: Exception) -> None: + from roboflow.adapters.workflowevalsapi import WorkflowEvalError + from roboflow.cli._output import output_error + + if isinstance(exc, WorkflowEvalError): + payload = exc.to_dict() + payload.pop("hint", None) + output_error(args, json.dumps({"error": payload}), hint=_error_hint(exc), exit_code=_exit_code(exc)) + else: + output_error(args, str(exc)) + + +def _invoke( + args: Any, + call: Callable[[Any], Any], + text: Optional[Callable[[Any], Optional[str]]] = None, + *, + interactive: bool = False, +) -> None: + """Resolve the client, run *call*, and print the result (or a structured error). + + ``interactive`` keeps stdout attached so confirmation prompts stay visible. + """ + import contextlib + + from roboflow.cli._output import output, suppress_sdk_output + + client = _client(args) + if client is None: + return + try: + with contextlib.nullcontext() if interactive else suppress_sdk_output(args): + result = call(client) + except Exception as exc: # noqa: BLE001 — every failure becomes a structured CLI error + _fail(args, exc) + return + output(args, result, text=text(result) if text and not getattr(args, "json", False) else None) + + +def _load_json(args: Any, value: Optional[str], flag: str = "--body") -> Any: + """Parse a JSON option: inline JSON, ``@path``, an existing file path, or ``-`` for stdin.""" + from roboflow.cli._output import output_error + + if value is None: + return None + try: + if value == "-": + return json.load(sys.stdin) + path = value[1:] if value.startswith("@") else value + if value.startswith("@") or (not value.lstrip().startswith(("{", "[")) and os.path.isfile(path)): + with open(path, encoding="utf-8") as handle: + return json.load(handle) + return json.loads(value) + except (OSError, ValueError) as exc: + output_error( + args, + f"{flag} is not valid JSON: {exc}", + hint=f"Pass inline JSON, @file.json, or - to read {flag} from stdin.", + ) + return None + + +def _load_object(args: Any, value: Optional[str], flag: str = "--body") -> Dict[str, Any]: + from roboflow.cli._output import output_error + + data = _load_json(args, value, flag) + if data is None: + return {} + if not isinstance(data, dict): + output_error(args, f"{flag} must be a JSON object.") + return data + + +def _require(args: Any, body: Dict[str, Any], flag: str = "--body") -> Dict[str, Any]: + from roboflow.cli._output import output_error + + if not body: + output_error(args, f"{flag} is required.", hint="Run the command with --help for the expected JSON.") + return body + + +def _merge(body: Dict[str, Any], **fields: Any) -> Dict[str, Any]: + return {**body, **{key: value for key, value in fields.items() if value is not None}} + + +def _subject( + args: Any, + subject_json: Optional[str], + workflow: Optional[str], + workflow_version: Optional[str], + subject_key: Optional[str], +) -> Dict[str, Any]: + from roboflow.cli._output import output_error + + if subject_json: + return _load_object(args, subject_json, "--subject") + if not workflow: + output_error( + args, + "A Workflow subject is required.", + hint="Pass --workflow (optionally --workflow-version) or --subject .", + ) + subject: Dict[str, Any] = {"kind": "savedWorkflow", "workflowId": workflow, "subjectKey": subject_key or workflow} + if workflow_version: + subject.update(kind="savedWorkflowVersion", workflowVersionId=workflow_version) + return subject + + +def _pick(item: Dict[str, Any], *keys: str) -> Any: + for key in keys: + value = item.get(key) + if value not in (None, ""): + return value + return "" + + +def _page_text(columns: Dict[str, tuple]) -> Callable[[Any], str]: + """Render ``{"items": [...]}`` pages as a table; *columns* maps header -> candidate keys.""" + from roboflow.cli._table import format_table + + def render(page: Any) -> str: + items = page.get("items", []) if isinstance(page, dict) else [] + rows = [{header: _pick(item, *keys) for header, keys in columns.items()} for item in items] + table = format_table(rows, list(columns)) + if isinstance(page, dict) and page.get("nextCursor"): + table += f"\n\nMore results: --cursor {page['nextCursor']}" + return table + + return render + + +def _confirm_delete(args: Any, client: Any, request: Callable[[Optional[str]], Any], what: str) -> Any: + """Run the two-step deletion protocol, asking for confirmation in between.""" + from roboflow.adapters.workflowevalsapi import WorkflowEvalError + from roboflow.cli._output import confirm_destructive + + try: + return request(None) + except WorkflowEvalError as exc: + key = exc.details.get("deletionKey") if exc.code == "deletion_confirmation_required" else None + if not key: + raise + impact = exc.details.get("impact") + if impact and not getattr(args, "yes", False) and not getattr(args, "json", False): + print(f"Deletion impact: {json.dumps(impact, default=str)}", file=sys.stderr) + if not confirm_destructive(args, f"Delete {what}? This cannot be undone."): + sys.exit(0) # confirm_destructive already reported the cancellation + return request(key) + + +_EVAL_COLUMNS = { + "ID": ("id",), + "NAME": ("name",), + "STATE": ("state",), + "RUNS": ("runCount",), + "LAST RUN": ("lastRunAt",), +} +_RUN_COLUMNS = { + "ID": ("id",), + "STATE": ("state",), + "MODE": ("executionMode",), + "CASES": ("selectedCaseCount",), + "EXECUTIONS": ("executionCount",), + "CREATED": ("createdAt",), +} +_NAMED_COLUMNS = {"ID": ("id",), "NAME": ("name",), "REVISION": ("revision",), "UPDATED": ("updatedAt",)} + + +# --------------------------------------------------------------------------- +# Evals +# --------------------------------------------------------------------------- + + +@workflow_eval_app.command("list") +def list_evals( + ctx: typer.Context, + limit: LimitOpt = None, + cursor: CursorOpt = None, + search: Annotated[Optional[str], typer.Option("--search", help="Filter by name")] = None, + spec_summary: Annotated[bool, typer.Option("--spec-summary", help="Include a summary of each Spec")] = False, +) -> None: + """List Evals in the workspace (newest first).""" + args = ctx_to_args(ctx) + _invoke( + args, + lambda c: c.list(limit=limit, cursor=cursor, search=search, include_spec_summary=spec_summary), + _page_text(_EVAL_COLUMNS), + ) + + +@workflow_eval_app.command("get") +def get_eval( + ctx: typer.Context, + eval_id: Annotated[str, typer.Argument(help="Eval id (or readable URL with --by-url)")], + by_url: Annotated[bool, typer.Option("--by-url", help="Treat the argument as the Eval's readable URL")] = False, +) -> None: + """Show an Eval with its setup state, diagnostics and recent Runs.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_by_url(eval_id) if by_url else c.get(eval_id)) + + +@workflow_eval_app.command("create") +def create_eval( + ctx: typer.Context, + name: Annotated[Optional[str], typer.Option("--name", help="Eval name")] = None, + description: Annotated[Optional[str], typer.Option("--description", help="Eval description")] = None, + spec: Annotated[Optional[str], typer.Option("--spec", "-s", help="Spec id")] = None, + dataset: Annotated[Optional[str], typer.Option("--dataset", "-d", help="Eval Dataset id")] = None, + body: BodyOpt = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Create an Eval (an empty draft when no options are given).""" + args = ctx_to_args(ctx) + payload = _merge(_load_object(args, body), name=name, description=description, specId=spec, evalDatasetId=dataset) + _invoke(args, lambda c: c.create(payload, idempotency_key=idempotency_key)) + + +@workflow_eval_app.command("update") +def update_eval( + ctx: typer.Context, + eval_id: Annotated[str, typer.Argument(help="Eval id")], + revision: RevisionOpt, + name: Annotated[Optional[str], typer.Option("--name", help="New name")] = None, + description: Annotated[Optional[str], typer.Option("--description", help="New description")] = None, + spec: Annotated[Optional[str], typer.Option("--spec", "-s", help="Attach this Spec id")] = None, + dataset: Annotated[Optional[str], typer.Option("--dataset", "-d", help="Attach this Eval Dataset id")] = None, + body: BodyOpt = None, +) -> None: + """Update an Eval's metadata, Spec/Dataset, subjects, bindings or server configuration.""" + args = ctx_to_args(ctx) + payload = _require( + args, + _merge(_load_object(args, body), name=name, description=description, specId=spec, evalDatasetId=dataset), + "--body or a field option", + ) + _invoke(args, lambda c: c.update(eval_id, payload, revision=revision)) + + +@workflow_eval_app.command("delete") +def delete_eval( + ctx: typer.Context, + eval_id: Annotated[str, typer.Argument(help="Eval id")], + yes: YesOpt = False, +) -> None: + """Delete an Eval with its Runs, results and exclusively owned artifacts.""" + args = ctx_to_args(ctx, yes=yes) + _invoke( + args, + lambda c: _confirm_delete(args, c, lambda key: c.delete(eval_id, deletion_key=key), f"Eval {eval_id}"), + interactive=True, + ) + + +@workflow_eval_app.command("capabilities") +def capabilities(ctx: typer.Context) -> None: + """List the public operations and what this credential may do.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.capabilities()) + + +@workflow_eval_app.command("openapi") +def openapi(ctx: typer.Context) -> None: + """Print the Workflow Evals OpenAPI 3.1 contract.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.openapi()) + + +@workflow_eval_app.command("operation") +def operation( + ctx: typer.Context, + operation_id: Annotated[str, typer.Argument(help="Operation id from a deletion response")], +) -> None: + """Show the status of a deletion operation.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_operation(operation_id)) + + +@workflow_eval_app.command("compare") +def compare( + ctx: typer.Context, + eval_id: EvalOpt, + execution: Annotated[ + Optional[List[str]], typer.Option("--execution", "-x", help="Completed Execution id (repeat 2+ times)") + ] = None, + body: BodyOpt = None, +) -> None: + """Compare two or more completed Executions (per-check and per-Case deltas).""" + args = ctx_to_args(ctx) + payload = _require(args, _merge(_load_object(args, body), executionIds=execution or None), "--execution") + _invoke(args, lambda c: c.compare(eval_id, payload)) + + +# --------------------------------------------------------------------------- +# Specs +# --------------------------------------------------------------------------- + + +@spec_app.command("list") +def list_specs(ctx: typer.Context, limit: LimitOpt = None, cursor: CursorOpt = None) -> None: + """List Specs in the workspace.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.list_specs(limit=limit, cursor=cursor), _page_text(_NAMED_COLUMNS)) + + +@spec_app.command("get") +def get_spec(ctx: typer.Context, spec_id: Annotated[str, typer.Argument(help="Spec id")]) -> None: + """Show the current live Spec.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_spec(spec_id)) + + +@spec_app.command("create") +def create_spec( + ctx: typer.Context, + body: Annotated[ + str, typer.Option("--body", "-b", help="JSON {name?, description?, spec}: inline, @file.json, path, or -") + ], + name: Annotated[Optional[str], typer.Option("--name", help="Spec name")] = None, + description: Annotated[Optional[str], typer.Option("--description", help="Spec description")] = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Validate, compile and save a Spec.""" + args = ctx_to_args(ctx) + payload = _merge(_require(args, _load_object(args, body)), name=name, description=description) + _invoke(args, lambda c: c.create_spec(payload, idempotency_key=idempotency_key)) + + +@spec_app.command("update") +def update_spec( + ctx: typer.Context, + spec_id: Annotated[str, typer.Argument(help="Spec id")], + revision: RevisionOpt, + name: Annotated[Optional[str], typer.Option("--name", help="New name")] = None, + description: Annotated[Optional[str], typer.Option("--description", help="New description")] = None, + body: BodyOpt = None, +) -> None: + """Replace the live Spec's policy and/or metadata.""" + args = ctx_to_args(ctx) + payload = _require( + args, _merge(_load_object(args, body), name=name, description=description), "--body or a field option" + ) + _invoke(args, lambda c: c.update_spec(spec_id, payload, revision=revision)) + + +@spec_app.command("validate") +def validate_spec( + ctx: typer.Context, + body: Annotated[str, typer.Option("--body", "-b", help="Spec JSON (or {spec}): inline, @file.json, path, or -")], +) -> None: + """Validate an unsaved Spec and show diagnostics and ground-truth requirements.""" + args = ctx_to_args(ctx) + spec = _require(args, _load_object(args, body)) + _invoke(args, lambda c: c.validate_spec(spec)) + + +@spec_app.command("compatible-datasets") +def spec_compatible_datasets( + ctx: typer.Context, + spec_id: Annotated[str, typer.Argument(help="Spec id")], + limit: LimitOpt = None, + cursor: CursorOpt = None, +) -> None: + """List Eval Datasets that satisfy this Spec.""" + args = ctx_to_args(ctx) + _invoke( + args, + lambda c: c.list_compatible_datasets(spec_id, limit=limit, cursor=cursor), + _page_text(_NAMED_COLUMNS), + ) + + +@spec_app.command("ai-draft") +def ai_draft( + ctx: typer.Context, + eval_id: EvalOpt, + instructions: Annotated[ + Optional[str], typer.Option("--instructions", "-i", help="Evaluation intent or requested improvements") + ] = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Draft a Spec and bindings with AI for an Eval's Subject (consumes AI credits; not saved).""" + args = ctx_to_args(ctx) + show_progress = not args.json and not args.quiet + + def on_event(event: Dict[str, Any]) -> None: + if not show_progress: + return + data = event.get("data") + if event["event"] == "status" and isinstance(data, dict) and data.get("message"): + print(f"… {data['message']}", file=sys.stderr) + + _invoke(args, lambda c: c.ai_draft(eval_id, instructions, idempotency_key=idempotency_key, on_event=on_event)) + + +# --------------------------------------------------------------------------- +# Eval Datasets +# --------------------------------------------------------------------------- + + +@dataset_app.command("list") +def list_datasets(ctx: typer.Context, limit: LimitOpt = None, cursor: CursorOpt = None) -> None: + """List Eval Datasets in the workspace.""" + args = ctx_to_args(ctx) + columns = {**_NAMED_COLUMNS, "CASES": ("caseCount",)} + _invoke(args, lambda c: c.list_datasets(limit=limit, cursor=cursor), _page_text(columns)) + + +@dataset_app.command("get") +def get_dataset(ctx: typer.Context, dataset_id: Annotated[str, typer.Argument(help="Eval Dataset id")]) -> None: + """Show an Eval Dataset and its derived summary.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_dataset(dataset_id)) + + +@dataset_app.command("create") +def create_dataset( + ctx: typer.Context, + body: Annotated[ + str, + typer.Option( + "--body", "-b", help="JSON {name, description?, inputs, groundTruthContract}: inline, @file, path, or -" + ), + ], + name: Annotated[Optional[str], typer.Option("--name", help="Dataset name")] = None, + description: Annotated[Optional[str], typer.Option("--description", help="Dataset description")] = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Create an empty Eval Dataset with declared input and ground-truth contracts.""" + args = ctx_to_args(ctx) + payload = _merge(_require(args, _load_object(args, body)), name=name, description=description) + _invoke(args, lambda c: c.create_dataset(payload, idempotency_key=idempotency_key)) + + +@dataset_app.command("update") +def update_dataset( + ctx: typer.Context, + dataset_id: Annotated[str, typer.Argument(help="Eval Dataset id")], + revision: RevisionOpt, + name: Annotated[Optional[str], typer.Option("--name", help="New name")] = None, + description: Annotated[Optional[str], typer.Option("--description", help="New description")] = None, +) -> None: + """Update an Eval Dataset's name or description.""" + args = ctx_to_args(ctx) + payload = _require(args, _merge({}, name=name, description=description), "--name or --description") + _invoke(args, lambda c: c.update_dataset(dataset_id, payload, revision=revision)) + + +@dataset_app.command("update-contract") +def update_dataset_contract( + ctx: typer.Context, + dataset_id: Annotated[str, typer.Argument(help="Eval Dataset id")], + revision: RevisionOpt, + body: Annotated[ + str, typer.Option("--body", "-b", help="JSON {inputs?, groundTruthContract?}: inline, @file, path, or -") + ], +) -> None: + """Replace the declared input and/or ground-truth contracts (may rewrite Cases in the background).""" + args = ctx_to_args(ctx) + payload = _require(args, _load_object(args, body)) + _invoke(args, lambda c: c.mutate_dataset_contract(dataset_id, payload, revision=revision)) + + +@dataset_app.command("compatible-specs") +def dataset_compatible_specs( + ctx: typer.Context, + dataset_id: Annotated[str, typer.Argument(help="Eval Dataset id")], + limit: LimitOpt = None, + cursor: CursorOpt = None, +) -> None: + """List Specs this Eval Dataset satisfies.""" + args = ctx_to_args(ctx) + _invoke( + args, + lambda c: c.list_compatible_specs(dataset_id, limit=limit, cursor=cursor), + _page_text(_NAMED_COLUMNS), + ) + + +# --------------------------------------------------------------------------- +# Cases +# --------------------------------------------------------------------------- + + +@case_app.command("list") +def list_cases( + ctx: typer.Context, + dataset_id: DatasetOpt, + limit: LimitOpt = None, + cursor: CursorOpt = None, + slice_: SliceOpt = None, + slice_match: SliceMatchOpt = None, + ground_truth: Annotated[ + Optional[str], typer.Option("--ground-truth", help="Filter by ground truth: complete or incomplete") + ] = None, +) -> None: + """List Cases in an Eval Dataset.""" + args = ctx_to_args(ctx) + columns = {"ID": ("id",), "NAME": ("name",), "GROUND TRUTH": ("groundTruthState",), "SLICES": ("slices",)} + _invoke( + args, + lambda c: c.list_cases( + dataset_id, + limit=limit, + cursor=cursor, + slices=slice_, + slice_match=slice_match, + ground_truth_completeness=ground_truth, + ), + _page_text(columns), + ) + + +@case_app.command("get") +def get_case( + ctx: typer.Context, case_id: Annotated[str, typer.Argument(help="Case id")], dataset_id: DatasetOpt +) -> None: + """Show one Case with short-lived read URLs for its assets.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_case(dataset_id, case_id)) + + +@case_app.command("add") +def add_case( + ctx: typer.Context, + dataset_id: DatasetOpt, + body: Annotated[ + str, + typer.Option("--body", "-b", help="JSON {name, inputs, groundTruth, slices?, ...}: inline, @file, path, or -"), + ], + name: Annotated[Optional[str], typer.Option("--name", help="Case name")] = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Add one Case to an Eval Dataset.""" + args = ctx_to_args(ctx) + payload = _merge(_require(args, _load_object(args, body)), name=name) + _invoke(args, lambda c: c.add_case(dataset_id, payload, idempotency_key=idempotency_key)) + + +@case_app.command("replace") +def replace_case( + ctx: typer.Context, + case_id: Annotated[str, typer.Argument(help="Case id")], + dataset_id: DatasetOpt, + revision: RevisionOpt, + body: Annotated[ + str, typer.Option("--body", "-b", help="JSON {inputs, groundTruth, name?, slices?}: inline, @file, path, or -") + ], +) -> None: + """Replace one Case atomically.""" + args = ctx_to_args(ctx) + payload = _require(args, _load_object(args, body)) + _invoke(args, lambda c: c.replace_case(dataset_id, case_id, payload, revision=revision)) + + +@case_app.command("delete") +def delete_case( + ctx: typer.Context, + case_id: Annotated[str, typer.Argument(help="Case id")], + dataset_id: DatasetOpt, + revision: RevisionOpt, + yes: YesOpt = False, +) -> None: + """Remove a Case from the live Dataset (Runs that froze it are unaffected).""" + from roboflow.cli._output import confirm_destructive + + args = ctx_to_args(ctx, yes=yes) + if not confirm_destructive(args, f"Delete Case {case_id}?"): + return + + def call(c: Any) -> Dict[str, Any]: + c.delete_case(dataset_id, case_id, revision=revision) + return {"deleted": True, "caseId": case_id} + + _invoke(args, call, lambda _: f"Deleted Case {case_id}.") + + +@case_app.command("import") +def import_cases( + ctx: typer.Context, + dataset_id: DatasetOpt, + input_field: Annotated[ + Optional[str], typer.Option("--input-field", help="Dataset image input that receives each item") + ] = None, + source: Annotated[ + Optional[List[str]], typer.Option("--source", help="Platform image Source id (repeatable)") + ] = None, + from_dataset: Annotated[ + Optional[List[str]], + typer.Option("--from-dataset", help="Platform Dataset id, optionally id:train|valid|test (repeatable)"), + ] = None, + artifact: Annotated[ + Optional[List[str]], typer.Option("--artifact", help="Uploaded Case asset artifact id (repeatable)") + ] = None, + slice_: Annotated[Optional[List[str]], typer.Option("--slice", help="Slice for every imported Case")] = None, + body: BodyOpt = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Import images as incomplete Cases in the background.""" + args = ctx_to_args(ctx) + items: List[Dict[str, Any]] = [{"kind": "source", "sourceId": value} for value in source or []] + for value in from_dataset or []: + dataset, _, split = value.partition(":") + items.append({"kind": "dataset", "datasetId": dataset, **({"split": split} if split else {})}) + items += [{"kind": "artifact", "artifactId": value} for value in artifact or []] + payload = _merge(_load_object(args, body), inputField=input_field, slices=slice_ or None) + if items: + payload["items"] = payload.get("items", []) + items + payload = _require(args, payload, "--source/--from-dataset/--artifact or --body") + _invoke(args, lambda c: c.import_cases(dataset_id, payload, idempotency_key=idempotency_key)) + + +@case_app.command("import-status") +def import_status( + ctx: typer.Context, + intake_id: Annotated[str, typer.Argument(help="Intake id from `case import`")], + dataset_id: DatasetOpt, +) -> None: + """Show a Case import's status and progress.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_case_import(dataset_id, intake_id)) + + +@case_app.command("upload") +def upload_case_asset( + ctx: typer.Context, + path: Annotated[str, typer.Argument(help="Local image file (JPEG, PNG, WebP or GIF)")], + dataset_id: DatasetOpt, + content_type: Annotated[ + Optional[str], typer.Option("--content-type", help="MIME type (default: inferred from the extension)") + ] = None, + role: Annotated[Optional[str], typer.Option("--role", help="Optional asset role")] = None, +) -> None: + """Upload an image as a Case asset and print its artifactId.""" + args = ctx_to_args(ctx) + if not os.path.isfile(path): + from roboflow.cli._output import output_error + + output_error(args, f"File not found: {path}", exit_code=3) + _invoke( + args, + lambda c: c.upload_case_asset(dataset_id, path, content_type=content_type, role=role), + lambda asset: f"Uploaded {path} → artifactId {asset.get('artifactId')}", + ) + + +# --------------------------------------------------------------------------- +# Bindings +# --------------------------------------------------------------------------- + +_SubjectOpt = Annotated[Optional[str], typer.Option("--subject", help="WorkflowSubject JSON (overrides --workflow)")] +_WorkflowOpt = Annotated[Optional[str], typer.Option("--workflow", help="Saved Workflow id")] +_WorkflowVersionOpt = Annotated[Optional[str], typer.Option("--workflow-version", help="Saved WorkflowVersion id")] +_SubjectKeyOpt = Annotated[Optional[str], typer.Option("--subject-key", help="Subject key (default: workflow id)")] + + +@binding_app.command("suggest") +def suggest_bindings( + ctx: typer.Context, + spec_id: Annotated[str, typer.Option("--spec", "-s", help="Spec id")], + dataset_id: DatasetOpt, + workflow: _WorkflowOpt = None, + workflow_version: _WorkflowVersionOpt = None, + subject_key: _SubjectKeyOpt = None, + subject: _SubjectOpt = None, +) -> None: + """Suggest a BindingSet that maps a Workflow to the Spec and Dataset.""" + args = ctx_to_args(ctx) + resolved = _subject(args, subject, workflow, workflow_version, subject_key) + _invoke(args, lambda c: c.suggest_bindings(spec_id, dataset_id, resolved)) + + +@binding_app.command("validate") +def validate_bindings( + ctx: typer.Context, + spec_id: Annotated[str, typer.Option("--spec", "-s", help="Spec id")], + dataset_id: DatasetOpt, + binding_set: Annotated[str, typer.Option("--binding-set", help="BindingSet JSON: inline, @file.json, path, or -")], + workflow: _WorkflowOpt = None, + workflow_version: _WorkflowVersionOpt = None, + subject_key: _SubjectKeyOpt = None, + subject: _SubjectOpt = None, +) -> None: + """Validate a BindingSet against the Spec, Dataset and Workflow.""" + args = ctx_to_args(ctx) + resolved = _subject(args, subject, workflow, workflow_version, subject_key) + bindings = _require(args, _load_object(args, binding_set, "--binding-set"), "--binding-set") + _invoke(args, lambda c: c.validate_bindings(spec_id, dataset_id, resolved, bindings)) + + +# --------------------------------------------------------------------------- +# Case preparations +# --------------------------------------------------------------------------- + + +@preparation_app.command("start") +def start_preparation( + ctx: typer.Context, + eval_id: EvalOpt, + runtime: Annotated[str, typer.Option("--runtime", help="serverless or dedicated")] = "serverless", + url: Annotated[Optional[str], typer.Option("--url", help="Dedicated deployment base URL")] = None, + case: Annotated[ + Optional[List[str]], typer.Option("--case", help="Prepare only these Case ids (repeatable)") + ] = None, + case_import: Annotated[ + Optional[str], typer.Option("--case-import", help="Follow a running Case import's async task id") + ] = None, + body: BodyOpt = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Run the Eval's Workflow over Cases to produce candidate outputs for labeling.""" + args = ctx_to_args(ctx) + payload = _load_object(args, body) + payload.setdefault("runtime", _merge({"kind": runtime}, url=url)) + if case: + payload["caseSelection"] = {"kind": "explicit", "caseIds": case} + payload = _merge(payload, caseImportId=case_import) + _invoke(args, lambda c: c.start_case_preparation(eval_id, payload, idempotency_key=idempotency_key)) + + +@preparation_app.command("current") +def current_preparation(ctx: typer.Context, eval_id: EvalOpt) -> None: + """Show the latest preparation for the Eval's current setup.""" + args = ctx_to_args(ctx) + _invoke( + args, + lambda c: c.get_current_case_preparation(eval_id), + lambda result: None if result else "No current case preparation.", + ) + + +@preparation_app.command("get") +def get_preparation( + ctx: typer.Context, preparation_id: Annotated[str, typer.Argument(help="Preparation id")], eval_id: EvalOpt +) -> None: + """Show a preparation's state and progress.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_case_preparation(eval_id, preparation_id)) + + +@preparation_app.command("resume") +def resume_preparation( + ctx: typer.Context, preparation_id: Annotated[str, typer.Argument(help="Preparation id")], eval_id: EvalOpt +) -> None: + """Resume a preparation, requeueing only failed Cases.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.resume_case_preparation(eval_id, preparation_id)) + + +@preparation_app.command("cases") +def preparation_cases( + ctx: typer.Context, + preparation_id: Annotated[str, typer.Argument(help="Preparation id")], + eval_id: EvalOpt, + limit: LimitOpt = None, + cursor: CursorOpt = None, +) -> None: + """List per-Case candidate outputs of a preparation.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.list_case_preparation_cases(eval_id, preparation_id, limit=limit, cursor=cursor)) + + +# --------------------------------------------------------------------------- +# Runs +# --------------------------------------------------------------------------- + +_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")] +_IntervalOpt = Annotated[float, typer.Option("--interval", help="Seconds between polls")] + + +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 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) + + return wait + + +@run_app.command("list") +def list_runs( + ctx: typer.Context, + eval_id: EvalOpt, + limit: LimitOpt = None, + cursor: CursorOpt = None, + check_count: Annotated[bool, typer.Option("--check-count", help="Include each Run's frozen Check count")] = False, +) -> None: + """List an Eval's Runs (newest first).""" + args = ctx_to_args(ctx) + _invoke( + args, + lambda c: c.list_runs(eval_id, limit=limit, cursor=cursor, include_check_count=check_count), + _page_text(_RUN_COLUMNS), + ) + + +@run_app.command("get") +def get_run( + ctx: typer.Context, + run_id: Annotated[str, typer.Argument(help="Run id")], + eval_id: EvalOpt, + wait: _WaitOpt = False, + timeout: _TimeoutOpt = 3600, + 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)) + else: + _invoke(args, lambda c: c.get_run(eval_id, run_id)) + + +@run_app.command("start") +def start_run( + ctx: typer.Context, + eval_id: EvalOpt, + body: Annotated[ + str, + typer.Option( + "--body", "-b", help="JSON {executions: [{subject, bindingSet}], caseSelection?, ...}: inline, @file, or -" + ), + ], + wait: _WaitOpt = False, + timeout: _TimeoutOpt = 3600, + interval: _IntervalOpt = 5, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Start a grouped Run of one or more Workflows over the Eval's Dataset (may consume credits).""" + args = ctx_to_args(ctx) + payload = _require(args, _load_object(args, body)) + then = _then_wait(args, eval_id, timeout, interval) + + def call(c: Any) -> Any: + admission = c.start_run(eval_id, payload, idempotency_key=idempotency_key) + return then(c, admission) if wait else admission + + _invoke(args, call) + + +@run_app.command("cancel") +def cancel_run(ctx: typer.Context, run_id: Annotated[str, typer.Argument(help="Run id")], eval_id: EvalOpt) -> None: + """Request cancellation of a queued or running Run.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.cancel_run(eval_id, run_id)) + + +@run_app.command("retry") +def retry_run( + ctx: typer.Context, + run_id: Annotated[str, typer.Argument(help="Run id")], + eval_id: EvalOpt, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Retry missing or operationally failed Cases in the same Run.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.retry_run(eval_id, run_id, idempotency_key=idempotency_key)) + + +@run_app.command("replay") +def replay_run( + ctx: typer.Context, + run_id: Annotated[str, typer.Argument(help="Source Run id")], + eval_id: EvalOpt, + body: BodyOpt = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Create a new Run from a previous one, optionally replacing Subjects and bindings.""" + args = ctx_to_args(ctx) + payload = _load_object(args, body) + _invoke(args, lambda c: c.replay_run(eval_id, run_id, payload, idempotency_key=idempotency_key)) + + +@run_app.command("rescore") +def rescore_run( + ctx: typer.Context, + run_id: Annotated[str, typer.Argument(help="Source Run id")], + eval_id: EvalOpt, + body: Annotated[ + str, typer.Option("--body", "-b", help="JSON {specId, executions: [...]}: inline, @file, path, or -") + ], + idempotency_key: IdempotencyOpt = None, +) -> None: + """Rescore a Run's retained captures with the Eval's current saved Spec.""" + args = ctx_to_args(ctx) + payload = _require(args, _load_object(args, body)) + _invoke(args, lambda c: c.rescore_run(eval_id, run_id, payload, idempotency_key=idempotency_key)) + + +@run_app.command("delete") +def delete_run( + ctx: typer.Context, + run_id: Annotated[str, typer.Argument(help="Run id")], + eval_id: EvalOpt, + yes: YesOpt = False, +) -> None: + """Delete a Run with its Executions, results and artifacts.""" + args = ctx_to_args(ctx, yes=yes) + _invoke( + args, + lambda c: _confirm_delete( + args, c, lambda key: c.delete_run(eval_id, run_id, deletion_key=key), f"Run {run_id}" + ), + interactive=True, + ) + + +@run_app.command("config") +def run_config(ctx: typer.Context, run_id: Annotated[str, typer.Argument(help="Run id")], eval_id: EvalOpt) -> None: + """Show a Run's frozen shared configuration.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_run_configuration(eval_id, run_id)) + + +# --------------------------------------------------------------------------- +# Executions and results +# --------------------------------------------------------------------------- + +_ExecutionArg = Annotated[str, typer.Argument(help="Execution id")] + + +@execution_app.command("list") +def list_executions(ctx: typer.Context, eval_id: EvalOpt, run_id: RunOpt) -> None: + """List the Executions of a Run.""" + args = ctx_to_args(ctx) + columns = {"ID": ("id",), "SUBJECT": ("subjectKey", "name"), "STATE": ("state",)} + _invoke(args, lambda c: c.list_executions(eval_id, run_id), _page_text(columns)) + + +@execution_app.command("get") +def get_execution(ctx: typer.Context, execution_id: _ExecutionArg, eval_id: EvalOpt, run_id: RunOpt) -> None: + """Show one Execution's lifecycle and result availability.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_execution(eval_id, run_id, execution_id)) + + +@execution_app.command("config") +def execution_config(ctx: typer.Context, execution_id: _ExecutionArg, eval_id: EvalOpt, run_id: RunOpt) -> None: + """Show an Execution's frozen Subject, BindingSet and linked plan.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_execution_configuration(eval_id, run_id, execution_id)) + + +@execution_app.command("input-cases") +def execution_input_cases( + ctx: typer.Context, + execution_id: _ExecutionArg, + eval_id: EvalOpt, + run_id: RunOpt, + limit: LimitOpt = None, + cursor: CursorOpt = None, +) -> None: + """List frozen Case inputs and checkpoints for a locally executed Run.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.list_execution_input_cases(eval_id, run_id, execution_id, limit=limit, cursor=cursor)) + + +@execution_app.command("capture") +def submit_capture( + ctx: typer.Context, + execution_id: _ExecutionArg, + eval_id: EvalOpt, + run_id: RunOpt, + case_id: Annotated[str, typer.Option("--case", "-c", help="Case id")], + body: Annotated[ + str, + typer.Option( + "--body", "-b", help="JSON {attemptCount, shardIndex, output, runtime?}: inline, @file, path, or -" + ), + ], +) -> None: + """Submit a locally captured Workflow output for server-side scoring.""" + args = ctx_to_args(ctx) + payload = _require(args, _load_object(args, body)) + _invoke(args, lambda c: c.submit_capture(eval_id, run_id, execution_id, case_id, payload)) + + +@execution_app.command("overview") +def results_overview( + ctx: typer.Context, + execution_id: _ExecutionArg, + eval_id: EvalOpt, + run_id: RunOpt, + slice_: SliceOpt = None, + slice_match: SliceMatchOpt = None, +) -> None: + """Show aggregate Check performance and runtime for an Execution.""" + args = ctx_to_args(ctx) + _invoke( + args, lambda c: c.get_results_overview(eval_id, run_id, execution_id, slices=slice_, slice_match=slice_match) + ) + + +@execution_app.command("results") +def list_case_results( + ctx: typer.Context, + execution_id: _ExecutionArg, + eval_id: EvalOpt, + run_id: RunOpt, + limit: LimitOpt = None, + cursor: CursorOpt = None, + slice_: SliceOpt = None, + slice_match: SliceMatchOpt = None, + state: Annotated[ + Optional[str], typer.Option("--state", help="running, completed, failed, cancelled or skipped") + ] = None, + case_id: Annotated[Optional[str], typer.Option("--case", "-c", help="Only this Case id")] = None, + check_id: Annotated[Optional[str], typer.Option("--check", help="Filter by Check id")] = None, + judgment: Annotated[Optional[str], typer.Option("--judgment", help="Filter by judgment (pass/fail)")] = None, + failed_check: Annotated[ + Optional[bool], typer.Option("--failed-check/--no-failed-check", help="Only Cases with a failed Check") + ] = None, + order: Annotated[Optional[str], typer.Option("--order", help="Sort by case or latency")] = None, +) -> None: + """List per-Case results of an Execution.""" + args = ctx_to_args(ctx) + columns = {"CASE": ("caseId", "id"), "STATE": ("state",), "LATENCY MS": ("latencyMs",)} + _invoke( + args, + lambda c: c.list_case_results( + eval_id, + run_id, + execution_id, + limit=limit, + cursor=cursor, + slices=slice_, + slice_match=slice_match, + order=order, + state=state, + case_id=case_id, + check_id=check_id, + judgment=judgment, + failed_check=failed_check, + ), + _page_text(columns), + ) + + +@execution_app.command("result") +def get_case_result( + ctx: typer.Context, + execution_id: _ExecutionArg, + eval_id: EvalOpt, + run_id: RunOpt, + case_id: Annotated[str, typer.Option("--case", "-c", help="Case id")], +) -> None: + """Show one Case's detailed result in an Execution.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_case_result(eval_id, run_id, execution_id, case_id)) + + +# --------------------------------------------------------------------------- +# Exports +# --------------------------------------------------------------------------- + + +@export_app.command("start") +def start_export( + ctx: typer.Context, + eval_id: EvalOpt, + run_id: RunOpt, + export_format: Annotated[ + Optional[str], typer.Option("--format", "-f", help="csv, json or xlsx (default: json)") + ] = None, + execution: Annotated[ + Optional[List[str]], typer.Option("--execution", "-x", help="Limit to these Execution ids (repeatable)") + ] = None, + body: BodyOpt = None, + wait: Annotated[bool, typer.Option("--wait", help="Poll until the export finishes")] = False, + timeout: _TimeoutOpt = 1800, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Export a stable Run's results.""" + args = ctx_to_args(ctx) + payload = _merge(_load_object(args, body), format=export_format, executionIds=execution or None) + payload.setdefault("format", "json") + + def call(c: Any) -> Any: + admission = c.start_export(eval_id, run_id, payload, idempotency_key=idempotency_key) + if not wait or not admission.get("asyncTaskId"): + return admission + return c.wait_for_export(eval_id, run_id, admission["asyncTaskId"], timeout=timeout) + + _invoke(args, call) + + +@export_app.command("status") +def export_status( + ctx: typer.Context, + task_id: Annotated[str, typer.Argument(help="Export asyncTaskId")], + eval_id: EvalOpt, + run_id: RunOpt, +) -> None: + """Show an export job's state and result.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_export_job(eval_id, run_id, task_id)) + + +@export_app.command("download-url") +def export_download_url( + ctx: typer.Context, + artifact_id: Annotated[str, typer.Argument(help="Export artifact id")], + eval_id: EvalOpt, + run_id: RunOpt, +) -> None: + """Mint a fresh short-lived download URL for a completed export.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_export_download(eval_id, run_id, artifact_id)) + + +# --------------------------------------------------------------------------- +# Embedding analyses +# --------------------------------------------------------------------------- + + +@embedding_app.command("start") +def start_embeddings( + ctx: typer.Context, + eval_id: EvalOpt, + run_id: RunOpt, + execution_id: ExecutionOpt, + provider: Annotated[str, typer.Option("--provider", "-p", help="clip, dinov2, dinov3 or siglip2")], + input_key: Annotated[Optional[str], typer.Option("--input-key", help="Image input to embed")] = None, + retry: Annotated[bool, typer.Option("--retry", help="Retry a failed or cancelled job")] = False, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Generate (or reuse) image embeddings for an Execution's Case inputs.""" + args = ctx_to_args(ctx) + _invoke( + args, + lambda c: c.start_embeddings( + eval_id, + run_id, + execution_id, + provider, + input_key=input_key, + retry=retry or None, + idempotency_key=idempotency_key, + ), + ) + + +@embedding_app.command("get") +def get_embeddings( + ctx: typer.Context, + eval_id: EvalOpt, + run_id: RunOpt, + execution_id: ExecutionOpt, + provider: Annotated[Optional[str], typer.Option("--provider", "-p", help="Discover the job for a provider")] = None, + task_id: Annotated[Optional[str], typer.Option("--task", help="Read a specific job attempt")] = None, +) -> None: + """Show an embedding analysis job and its projection result.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_embeddings(eval_id, run_id, execution_id, provider=provider, async_task_id=task_id)) + + +@embedding_app.command("cancel") +def cancel_embeddings( + ctx: typer.Context, + task_id: Annotated[str, typer.Argument(help="Embedding job asyncTaskId")], + eval_id: EvalOpt, + run_id: RunOpt, + execution_id: ExecutionOpt, +) -> None: + """Cancel an embedding analysis job.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.cancel_embeddings(eval_id, run_id, execution_id, task_id)) + + +# --------------------------------------------------------------------------- +# Engine catalog and agent guidance +# --------------------------------------------------------------------------- + + +@evaluator_app.command("list") +def list_evaluators( + ctx: typer.Context, + semantic_type: Annotated[Optional[str], typer.Option("--semantic-type", help="Filter by semantic type")] = None, + family: Annotated[Optional[str], typer.Option("--family", help="Filter by evaluator family")] = None, +) -> None: + """List the installed engine's evaluators.""" + args = ctx_to_args(ctx) + columns = {"TYPE": ("type",), "SEMANTIC TYPE": ("semanticType",), "FAMILY": ("family",)} + _invoke(args, lambda c: c.list_evaluators(semantic_type=semantic_type, family=family), _page_text(columns)) + + +@evaluator_app.command("get") +def get_evaluator(ctx: typer.Context, evaluator_type: Annotated[str, typer.Argument(help="Evaluator type")]) -> None: + """Show one evaluator and its configuration schema.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_evaluator(evaluator_type)) + + +@schema_app.command("get") +def get_schema( + ctx: typer.Context, + schema_name: Annotated[ + str, typer.Argument(help="spec, eval-dataset, case, binding-set, subject or evaluator-config") + ], +) -> None: + """Print an authoring JSON Schema.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_schema(schema_name)) + + +_RESOURCE_COLUMNS = {"ID": ("id",), "PATH": ("path",)} + + +@agent_app.command("manifest") +def agent_manifest(ctx: typer.Context) -> None: + """Show the engine's agent manifest (entry skill, skills, bundles, assets).""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.agent_manifest()) + + +@agent_app.command("skills") +def agent_skills(ctx: typer.Context) -> None: + """List agent skills.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.list_agent_skills(), _page_text(_RESOURCE_COLUMNS)) + + +@agent_app.command("skill") +def agent_skill( + ctx: typer.Context, resource_id: Annotated[str, typer.Argument(help="Skill id, e.g. skill:create-eval")] +) -> None: + """Print an agent skill's Markdown.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_agent_skill(resource_id), lambda r: r.get("content")) + + +@agent_app.command("bundles") +def agent_bundles(ctx: typer.Context) -> None: + """List agent bundles.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.list_agent_bundles(), _page_text(_RESOURCE_COLUMNS)) + + +@agent_app.command("bundle") +def agent_bundle( + ctx: typer.Context, resource_id: Annotated[str, typer.Argument(help="Bundle id, e.g. bundle:create-eval")] +) -> None: + """Show a parsed agent bundle.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_agent_bundle(resource_id)) + + +@agent_app.command("asset") +def agent_asset( + ctx: typer.Context, resource_id: Annotated[str, typer.Argument(help="Asset id from a bundle's assets")] +) -> None: + """Print a registered bundle asset's original content.""" + args = ctx_to_args(ctx) + _invoke(args, lambda c: c.get_agent_asset(resource_id), lambda r: r.get("content")) diff --git a/roboflow/core/workflow_evals.py b/roboflow/core/workflow_evals.py new file mode 100644 index 00000000..00fc842b --- /dev/null +++ b/roboflow/core/workflow_evals.py @@ -0,0 +1,688 @@ +"""Workflow Evals — wraps the public ``/workspaces/{workspace}/workflow-evals`` API. + +A :class:`WorkflowEvals` client is bound to one workspace and exposes one +method per public endpoint, returning the server's JSON as plain dicts so new +response fields work without an SDK update. Get one from +:meth:`roboflow.core.workspace.Workspace.workflow_evals`:: + + import roboflow + + evals = roboflow.Roboflow().workspace().workflow_evals() + draft = evals.create(name="Boolean answer") + spec = evals.create_spec({"name": "Boolean answer", "spec": {...}}) + +Conventions shared by every method: + +* Commands that create resources or start background work send an + ``Idempotency-Key``. One is generated when ``idempotency_key`` is omitted; + pass your own to safely retry a request whose response was lost. +* Updates take the resource's current numeric ``revision`` (sent as + ``If-Match``). A stale revision fails with HTTP 412. +* List methods return ``{"items": [...], "nextCursor": ...}``; pass + ``cursor`` to read the next page. +* Errors raise :class:`roboflow.adapters.workflowevalsapi.WorkflowEvalError`. + +The workspace needs the Workflow Evals feature, and the API key needs the +``workflow-evals:read|write|run|export`` scopes for the operations it calls. +""" + +from __future__ import annotations + +import mimetypes +import os +import time +from typing import Any, Callable, Dict, Iterator, List, Optional + +from roboflow.adapters import workflowevalsapi as api +from roboflow.adapters.workflowevalsapi import WorkflowEvalError + +RUN_TERMINAL_STATES = frozenset({"completed", "partially_completed", "failed", "cancelled"}) +TASK_TERMINAL_STATES = frozenset({"completed", "partially_completed", "failed", "cancelled"}) + + +def _seg(value: Any) -> str: + return api.segment(value) + + +def _body(**fields: Any) -> Dict[str, Any]: + return {key: value for key, value in fields.items() if value is not None} + + +class WorkflowEvals: + """Client for one workspace's Workflow Evals.""" + + def __init__(self, api_key: str, workspace_url: str) -> None: + self._api_key = api_key + self.workspace_url = workspace_url + + # -- transport ---------------------------------------------------------- + + def _call(self, method: str, path: str = "", **kwargs: Any) -> Any: + return api.request(self._api_key, self.workspace_url, method, path, **kwargs) + + def _command(self, path: str, body: Any = None, idempotency_key: Optional[str] = None, **kwargs: Any) -> Any: + return self._call( + "POST", + path, + body=body if body is not None else {}, + idempotency_key=idempotency_key or api.new_idempotency_key(), + **kwargs, + ) + + @staticmethod + def _run_path(eval_id: str, run_id: str) -> str: + return f"/{_seg(eval_id)}/runs/{_seg(run_id)}" + + @classmethod + def _execution_path(cls, eval_id: str, run_id: str, execution_id: str) -> str: + return f"{cls._run_path(eval_id, run_id)}/executions/{_seg(execution_id)}" + + # -- discovery ---------------------------------------------------------- + + def capabilities(self) -> Dict[str, Any]: + """Describe the public operations and the caller's effective actions.""" + return self._call("GET", "/capabilities") + + def openapi(self) -> Dict[str, Any]: + """Return the machine-readable OpenAPI 3.1 contract.""" + return self._call("GET", "/openapi.json") + + def list_evaluators(self, *, semantic_type: Optional[str] = None, family: Optional[str] = None) -> Dict[str, Any]: + """List the installed engine's evaluator catalog.""" + return self._call("GET", "/evaluators", params={"semanticType": semantic_type, "family": family}) + + def get_evaluator(self, evaluator_type: str) -> Dict[str, Any]: + """Return one evaluator with its configuration schema.""" + return self._call("GET", f"/evaluators/{_seg(evaluator_type)}") + + def get_schema(self, schema_name: str) -> Dict[str, Any]: + """Return an authoring JSON Schema. + + ``schema_name`` is one of ``spec``, ``eval-dataset``, ``case``, + ``binding-set``, ``subject`` or ``evaluator-config``. + """ + return self._call("GET", f"/schemas/{_seg(schema_name)}") + + def get_operation(self, operation_id: str) -> Dict[str, Any]: + """Read a deletion operation's status receipt.""" + return self._call("GET", f"/operations/{_seg(operation_id)}") + + # -- agent guidance ----------------------------------------------------- + + def agent_manifest(self) -> Dict[str, Any]: + """Return ``{engineVersion, manifest}`` for MCP/agent clients.""" + return self._call("GET", "/agent/manifest") + + def list_agent_skills(self) -> Dict[str, Any]: + return self._call("GET", "/agent/skills") + + def get_agent_skill(self, resource_id: str) -> Dict[str, Any]: + """Return a skill's Markdown, e.g. ``skill:create-eval``.""" + return self._call("GET", f"/agent/skills/{_seg(resource_id)}") + + def list_agent_bundles(self) -> Dict[str, Any]: + return self._call("GET", "/agent/bundles") + + def get_agent_bundle(self, resource_id: str) -> Dict[str, Any]: + """Return a parsed bundle, e.g. ``bundle:create-eval``.""" + return self._call("GET", f"/agent/bundles/{_seg(resource_id)}") + + def get_agent_asset(self, resource_id: str) -> Dict[str, Any]: + """Return the original content of a registered bundle dependency.""" + return self._call("GET", f"/agent/assets/{_seg(resource_id)}") + + # -- evals -------------------------------------------------------------- + + def list( + self, + *, + limit: Optional[int] = None, + cursor: Optional[str] = None, + include_spec_summary: bool = False, + search: Optional[str] = None, + ) -> Dict[str, Any]: + """List Evals, newest first.""" + params = { + "limit": limit, + "cursor": cursor, + "include": "specSummary" if include_spec_summary else None, + "search": search, + } + return self._call("GET", "", params=params) + + def create( + self, + body: Optional[Dict[str, Any]] = None, + *, + name: Optional[str] = None, + description: Optional[str] = None, + spec_id: Optional[str] = None, + eval_dataset_id: Optional[str] = None, + idempotency_key: Optional[str] = None, + ) -> Dict[str, Any]: + """Create an Eval. Every field is optional; ``{}`` creates an untitled draft. + + ``body`` may carry any other accepted field (``subject``, + ``subjectBindings``, ``serverConfiguration``); keyword arguments win. + """ + payload = { + **(body or {}), + **_body(name=name, description=description, specId=spec_id, evalDatasetId=eval_dataset_id), + } + return self._command("", payload, idempotency_key) + + def get(self, eval_id: str) -> Dict[str, Any]: + return self._call("GET", f"/{_seg(eval_id)}") + + def get_by_url(self, eval_url: str) -> Dict[str, Any]: + return self._call("GET", f"/by-url/{_seg(eval_url)}") + + def update(self, eval_id: str, body: Dict[str, Any], *, revision: Any) -> Dict[str, Any]: + """Patch an Eval (name, description, specId, evalDatasetId, subject(s), bindings, server config).""" + return self._call("PATCH", f"/{_seg(eval_id)}", body=body, revision=revision) + + def delete(self, eval_id: str, *, deletion_key: Optional[str] = None) -> Dict[str, Any]: + """Delete an Eval and everything it exclusively owns. + + Without ``deletion_key`` the server answers ``409 + deletion_confirmation_required`` with the impact and a short-lived key + in ``error.details``; resend with that key to confirm. See + :meth:`delete_confirmed` for the two-step flow in one call. + """ + return self._call("DELETE", f"/{_seg(eval_id)}", deletion_key=deletion_key) + + def delete_confirmed(self, eval_id: str) -> Dict[str, Any]: + """Request the deletion key and immediately confirm the Eval deletion.""" + return self._delete_confirmed(lambda key: self.delete(eval_id, deletion_key=key)) + + @staticmethod + def _delete_confirmed(send: Callable[[Optional[str]], Dict[str, Any]]) -> Dict[str, Any]: + try: + return send(None) + except WorkflowEvalError as exc: + key = exc.details.get("deletionKey") if exc.code == "deletion_confirmation_required" else None + if not key: + raise + return send(key) + + # -- specs -------------------------------------------------------------- + + def list_specs(self, *, limit: Optional[int] = None, cursor: Optional[str] = None) -> Dict[str, Any]: + return self._call("GET", "/specs", params={"limit": limit, "cursor": cursor}) + + def create_spec(self, body: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: + """Validate, compile and save a Spec: ``{name?, description?, spec}``.""" + return self._command("/specs", body, idempotency_key) + + def get_spec(self, spec_id: str) -> Dict[str, Any]: + return self._call("GET", f"/specs/{_seg(spec_id)}") + + def update_spec(self, spec_id: str, body: Dict[str, Any], *, revision: Any) -> Dict[str, Any]: + return self._call("PATCH", f"/specs/{_seg(spec_id)}", body=body, revision=revision) + + def validate_spec(self, spec: Dict[str, Any]) -> Dict[str, Any]: + """Validate an unsaved Spec. Accepts the engine Spec or ``{"spec": ...}``.""" + body = spec if "spec" in spec else {"spec": spec} + return self._call("POST", "/specs/validate", body=body) + + def list_compatible_datasets( + self, spec_id: str, *, limit: Optional[int] = None, cursor: Optional[str] = None + ) -> Dict[str, Any]: + return self._call( + "GET", f"/specs/{_seg(spec_id)}/compatible-eval-datasets", params={"limit": limit, "cursor": cursor} + ) + + def iter_ai_draft( + self, eval_id: str, instructions: Optional[str] = None, *, idempotency_key: Optional[str] = None + ) -> Iterator[Dict[str, Any]]: + """Stream AI drafting events (``status``, ``delta``, ``draft``, ``error``, ``done``). + + Consumes AI credits. Reusing a completed ``idempotency_key`` replays the + final draft without another model call. + """ + return api.stream_events( + self._api_key, + self.workspace_url, + "/specs/ai-draft", + body=_body(evalId=eval_id, instructions=instructions), + idempotency_key=idempotency_key or api.new_idempotency_key(), + ) + + def ai_draft( + self, + eval_id: str, + instructions: Optional[str] = None, + *, + idempotency_key: Optional[str] = None, + on_event: Optional[Callable[[Dict[str, Any]], None]] = None, + ) -> Dict[str, Any]: + """Draft a Spec and bindings with AI and return the terminal ``draft`` payload. + + The draft is not persisted; save the Spec and confirm bindings + separately. + """ + draft: Optional[Dict[str, Any]] = None + for event in self.iter_ai_draft(eval_id, instructions, idempotency_key=idempotency_key): + if on_event: + on_event(event) + if event["event"] == "draft": + draft = event["data"] + elif event["event"] == "error": + data = event["data"] + message = data.get("message") if isinstance(data, dict) else str(data) + raise WorkflowEvalError(message or "AI draft failed", code="ai_draft_failed") + if draft is None: + raise WorkflowEvalError("AI draft stream ended without a draft", code="ai_draft_incomplete") + return draft + + # -- eval datasets ------------------------------------------------------ + + def list_datasets(self, *, limit: Optional[int] = None, cursor: Optional[str] = None) -> Dict[str, Any]: + return self._call("GET", "/eval-datasets", params={"limit": limit, "cursor": cursor}) + + def create_dataset(self, body: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: + """Create an empty Eval Dataset: ``{name, description?, inputs, groundTruthContract}``.""" + return self._command("/eval-datasets", body, idempotency_key) + + def get_dataset(self, dataset_id: str) -> Dict[str, Any]: + return self._call("GET", f"/eval-datasets/{_seg(dataset_id)}") + + def update_dataset(self, dataset_id: str, body: Dict[str, Any], *, revision: Any) -> Dict[str, Any]: + """Update Dataset metadata (``name``, ``description``).""" + return self._call("PATCH", f"/eval-datasets/{_seg(dataset_id)}", body=body, revision=revision) + + def mutate_dataset_contract(self, dataset_id: str, body: Dict[str, Any], *, revision: Any) -> Dict[str, Any]: + """Replace declared ``inputs`` and/or ``groundTruthContract``.""" + return self._call("POST", f"/eval-datasets/{_seg(dataset_id)}/contract-mutations", body=body, revision=revision) + + def list_compatible_specs( + self, dataset_id: str, *, limit: Optional[int] = None, cursor: Optional[str] = None + ) -> Dict[str, Any]: + return self._call( + "GET", f"/eval-datasets/{_seg(dataset_id)}/compatible-specs", params={"limit": limit, "cursor": cursor} + ) + + # -- cases -------------------------------------------------------------- + + def list_cases( + self, + dataset_id: str, + *, + limit: Optional[int] = None, + cursor: Optional[str] = None, + slices: Optional[List[str]] = None, + slice_match: Optional[str] = None, + ground_truth_completeness: Optional[str] = None, + ) -> Dict[str, Any]: + params = { + "limit": limit, + "cursor": cursor, + "slice": slices, + "sliceMatch": slice_match, + "groundTruthCompleteness": ground_truth_completeness, + } + return self._call("GET", f"/eval-datasets/{_seg(dataset_id)}/cases", params=params) + + def add_case( + self, dataset_id: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None + ) -> Dict[str, Any]: + """Create one Case: ``{name, inputs, groundTruth, slices?, evidenceRefs?, url?}``.""" + return self._command(f"/eval-datasets/{_seg(dataset_id)}/cases", body, idempotency_key) + + def get_case(self, dataset_id: str, case_id: str) -> Dict[str, Any]: + return self._call("GET", f"/eval-datasets/{_seg(dataset_id)}/cases/{_seg(case_id)}") + + def replace_case(self, dataset_id: str, case_id: str, body: Dict[str, Any], *, revision: Any) -> Dict[str, Any]: + """Replace one Case atomically.""" + return self._call( + "PUT", f"/eval-datasets/{_seg(dataset_id)}/cases/{_seg(case_id)}", body=body, revision=revision + ) + + def delete_case(self, dataset_id: str, case_id: str, *, revision: Any) -> None: + """Remove a Case from the live Dataset. Runs that froze it are unaffected.""" + self._call("DELETE", f"/eval-datasets/{_seg(dataset_id)}/cases/{_seg(case_id)}", revision=revision) + + def import_cases( + self, dataset_id: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None + ) -> Dict[str, Any]: + """Import platform Sources, Datasets or uploaded artifacts as incomplete Cases. + + ``body`` is ``{inputField, items: [{kind: "source"|"dataset"|"artifact", ...}], slices?}``. + Returns ``{intakeId, asyncTaskId, status, pollUrl}``. + """ + return self._command(f"/eval-datasets/{_seg(dataset_id)}/cases/import", body, idempotency_key) + + def get_case_import(self, dataset_id: str, intake_id: str) -> Dict[str, Any]: + return self._call("GET", f"/eval-datasets/{_seg(dataset_id)}/case-imports/{_seg(intake_id)}") + + def prepare_case_asset_upload( + self, + dataset_id: str, + *, + asset_name: str, + content_type: str, + expected_size: int, + role: Optional[str] = None, + idempotency_key: Optional[str] = None, + ) -> Dict[str, Any]: + """Request a signed upload URL for a Case image (JPEG, PNG, WebP or GIF).""" + body = _body(assetName=asset_name, contentType=content_type, expectedSize=expected_size, role=role) + return self._command(f"/eval-datasets/{_seg(dataset_id)}/case-assets/uploads", body, idempotency_key) + + def complete_case_asset_upload(self, dataset_id: str, upload_id: str) -> Dict[str, Any]: + """Verify an uploaded object and publish it as a Case asset artifact.""" + return self._call("POST", f"/eval-datasets/{_seg(dataset_id)}/case-assets/uploads/{_seg(upload_id)}/complete") + + def upload_case_asset( + self, + dataset_id: str, + path: str, + *, + content_type: Optional[str] = None, + role: Optional[str] = None, + asset_name: Optional[str] = None, + ) -> Dict[str, Any]: + """Upload a local image and return the completed asset (with ``artifactId``). + + Use ``{"artifactId": ...}`` as a Case input value afterwards. + """ + content_type = content_type or mimetypes.guess_type(path)[0] + if not content_type: + raise ValueError(f"Cannot infer the content type of {path}; pass content_type explicitly") + with open(path, "rb") as handle: + data = handle.read() + intent = self.prepare_case_asset_upload( + dataset_id, + asset_name=asset_name or os.path.basename(path), + content_type=content_type, + expected_size=len(data), + role=role, + ) + api.put_signed_upload(intent["uploadUrl"], data, intent.get("requiredHeaders") or {}) + return self.complete_case_asset_upload(dataset_id, intent["uploadId"]) + + # -- bindings ----------------------------------------------------------- + + def suggest_bindings(self, spec_id: str, eval_dataset_id: str, subject: Dict[str, Any]) -> Dict[str, Any]: + """Suggest a BindingSet for one WorkflowSubject. Nothing is persisted.""" + body = {"specId": spec_id, "evalDatasetId": eval_dataset_id, "subject": subject} + return self._call("POST", "/bindings/suggest", body=body) + + def validate_bindings( + self, spec_id: str, eval_dataset_id: str, subject: Dict[str, Any], binding_set: Dict[str, Any] + ) -> Dict[str, Any]: + """Link a BindingSet against the Spec, Dataset and Subject and return diagnostics.""" + body = {"specId": spec_id, "evalDatasetId": eval_dataset_id, "subject": subject, "bindingSet": binding_set} + return self._call("POST", "/bindings/validate", body=body) + + # -- case preparations -------------------------------------------------- + + def start_case_preparation( + self, eval_id: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None + ) -> Dict[str, Any]: + """Run the Eval's Subject over Cases to produce candidate outputs for labeling. + + ``body`` is ``{runtime: {kind: "serverless"|"dedicated", url?}, caseSelection?, caseImportId?}``. + """ + return self._command(f"/{_seg(eval_id)}/case-preparations", body, idempotency_key) + + def get_current_case_preparation(self, eval_id: str) -> Optional[Dict[str, Any]]: + """Return the latest preparation for the current setup, or ``None``.""" + return self._call("GET", f"/{_seg(eval_id)}/case-preparations/current") + + def get_case_preparation(self, eval_id: str, preparation_id: str) -> Dict[str, Any]: + return self._call("GET", f"/{_seg(eval_id)}/case-preparations/{_seg(preparation_id)}") + + def resume_case_preparation(self, eval_id: str, preparation_id: str) -> Dict[str, Any]: + return self._call("POST", f"/{_seg(eval_id)}/case-preparations/{_seg(preparation_id)}/resume") + + def list_case_preparation_cases( + self, eval_id: str, preparation_id: str, *, limit: Optional[int] = None, cursor: Optional[str] = None + ) -> Dict[str, Any]: + return self._call( + "GET", + f"/{_seg(eval_id)}/case-preparations/{_seg(preparation_id)}/cases", + params={"limit": limit, "cursor": cursor}, + ) + + # -- runs --------------------------------------------------------------- + + def list_runs( + self, + eval_id: str, + *, + limit: Optional[int] = None, + cursor: Optional[str] = None, + include_check_count: bool = False, + ) -> Dict[str, Any]: + """List Runs newest first; ``limit=1`` returns the latest Run.""" + params = {"limit": limit, "cursor": cursor, "include": "checkCount" if include_check_count else None} + return self._call("GET", f"/{_seg(eval_id)}/runs", params=params) + + def start_run(self, eval_id: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: + """Start one grouped Run: ``{executions: [{subject, bindingSet}], caseSelection?, ...}``.""" + return self._command(f"/{_seg(eval_id)}/runs", body, idempotency_key) + + def get_run(self, eval_id: str, run_id: str) -> Dict[str, Any]: + return self._call("GET", self._run_path(eval_id, run_id)) + + def get_run_configuration(self, eval_id: str, run_id: str) -> Dict[str, Any]: + return self._call("GET", f"{self._run_path(eval_id, run_id)}/configuration") + + def cancel_run(self, eval_id: str, run_id: str) -> Dict[str, Any]: + return self._call("POST", f"{self._run_path(eval_id, run_id)}/cancel") + + def retry_run(self, eval_id: str, run_id: str, *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: + """Retry missing or operationally failed Cases in the same Run.""" + return self._command(f"{self._run_path(eval_id, run_id)}/retry-failed", {}, idempotency_key) + + def replay_run( + self, + eval_id: str, + run_id: str, + body: Optional[Dict[str, Any]] = None, + *, + idempotency_key: Optional[str] = None, + ) -> Dict[str, Any]: + """Create a derived Run, optionally with replacement Subjects and bindings.""" + return self._command(f"{self._run_path(eval_id, run_id)}/replay", body or {}, idempotency_key) + + def rescore_run( + self, eval_id: str, run_id: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None + ) -> Dict[str, Any]: + """Rescore retained captures with the Eval's current saved Spec: ``{specId, executions}``.""" + return self._command(f"{self._run_path(eval_id, run_id)}/rescore", body, idempotency_key) + + def delete_run(self, eval_id: str, run_id: str, *, deletion_key: Optional[str] = None) -> Dict[str, Any]: + """Delete a Run. Same two-step confirmation protocol as :meth:`delete`.""" + return self._call("DELETE", self._run_path(eval_id, run_id), deletion_key=deletion_key) + + def delete_run_confirmed(self, eval_id: str, run_id: str) -> Dict[str, Any]: + return self._delete_confirmed(lambda key: self.delete_run(eval_id, run_id, deletion_key=key)) + + def wait_for_run( + self, + eval_id: str, + run_id: str, + *, + timeout: 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. + """ + deadline = time.monotonic() + timeout + 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: + raise TimeoutError(f"Run {run_id} is still {state} after {timeout:g}s") + time.sleep(interval) + + # -- executions and results --------------------------------------------- + + def list_executions(self, eval_id: str, run_id: str) -> Dict[str, Any]: + return self._call("GET", f"{self._run_path(eval_id, run_id)}/executions") + + def get_execution(self, eval_id: str, run_id: str, execution_id: str) -> Dict[str, Any]: + return self._call("GET", self._execution_path(eval_id, run_id, execution_id)) + + def get_execution_configuration(self, eval_id: str, run_id: str, execution_id: str) -> Dict[str, Any]: + return self._call("GET", f"{self._execution_path(eval_id, run_id, execution_id)}/configuration") + + def list_execution_input_cases( + self, + eval_id: str, + run_id: str, + execution_id: str, + *, + limit: Optional[int] = None, + cursor: Optional[str] = None, + ) -> Dict[str, Any]: + """Read frozen Case inputs for a local (client-run) Execution.""" + return self._call( + "GET", + f"{self._execution_path(eval_id, run_id, execution_id)}/input-cases", + params={"limit": limit, "cursor": cursor}, + ) + + def submit_capture( + self, eval_id: str, run_id: str, execution_id: str, case_id: str, body: Dict[str, Any] + ) -> Dict[str, Any]: + """Submit one locally captured Workflow output for server-side scoring. + + ``body`` is ``{attemptCount, shardIndex, output, runtime?}`` or + ``{attemptCount, shardIndex, resume: true}``. + """ + return self._call( + "POST", f"{self._execution_path(eval_id, run_id, execution_id)}/cases/{_seg(case_id)}/capture", body=body + ) + + def get_results_overview( + self, + eval_id: str, + run_id: str, + execution_id: str, + *, + slices: Optional[List[str]] = None, + slice_match: Optional[str] = None, + ) -> Dict[str, Any]: + return self._call( + "GET", + f"{self._execution_path(eval_id, run_id, execution_id)}/results/overview", + params={"slice": slices, "sliceMatch": slice_match}, + ) + + def list_case_results( + self, + eval_id: str, + run_id: str, + execution_id: str, + *, + limit: Optional[int] = None, + cursor: Optional[str] = None, + slices: Optional[List[str]] = None, + slice_match: Optional[str] = None, + order: Optional[str] = None, + state: Optional[str] = None, + case_id: Optional[str] = None, + check_id: Optional[str] = None, + judgment: Optional[str] = None, + failed_check: Optional[bool] = None, + ) -> Dict[str, Any]: + params = { + "limit": limit, + "cursor": cursor, + "slice": slices, + "sliceMatch": slice_match, + "order": order, + "state": state, + "caseId": case_id, + "checkId": check_id, + "judgment": judgment, + "failedCheck": failed_check, + } + return self._call("GET", f"{self._execution_path(eval_id, run_id, execution_id)}/cases", params=params) + + def get_case_result(self, eval_id: str, run_id: str, execution_id: str, case_id: str) -> Dict[str, Any]: + return self._call("GET", f"{self._execution_path(eval_id, run_id, execution_id)}/cases/{_seg(case_id)}") + + def compare(self, eval_id: str, body: Dict[str, Any]) -> Dict[str, Any]: + """Compare two or more completed Executions: ``{executionIds, limit?, cursor?, ...}``.""" + return self._call("POST", f"/{_seg(eval_id)}/comparisons", body=body) + + # -- exports ------------------------------------------------------------ + + def start_export( + self, eval_id: str, run_id: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None + ) -> Dict[str, Any]: + """Start a ``csv``/``json``/``xlsx`` export of a stable Run's results.""" + return self._command(f"{self._run_path(eval_id, run_id)}/exports", body, idempotency_key) + + def get_export_job(self, eval_id: str, run_id: str, async_task_id: str) -> Dict[str, Any]: + return self._call("GET", f"{self._run_path(eval_id, run_id)}/export-jobs/{_seg(async_task_id)}") + + def get_export_download(self, eval_id: str, run_id: str, artifact_id: str) -> Dict[str, Any]: + """Mint a fresh short-lived download URL for a completed export.""" + return self._call("GET", f"{self._run_path(eval_id, run_id)}/exports/{_seg(artifact_id)}") + + def wait_for_export( + self, eval_id: str, run_id: str, async_task_id: str, *, timeout: float = 1800, interval: float = 3 + ) -> Dict[str, Any]: + """Poll an export job until it finishes and return its final status.""" + deadline = time.monotonic() + timeout + while True: + job = self.get_export_job(eval_id, run_id, async_task_id) + if job.get("state") in TASK_TERMINAL_STATES: + return job + if time.monotonic() >= deadline: + raise TimeoutError(f"Export {async_task_id} is still {job.get('state')} after {timeout:g}s") + time.sleep(interval) + + # -- embedding analyses ------------------------------------------------- + + def start_embeddings( + self, + eval_id: str, + run_id: str, + execution_id: str, + provider: str, + *, + input_key: Optional[str] = None, + retry: Optional[bool] = None, + idempotency_key: Optional[str] = None, + ) -> Dict[str, Any]: + """Generate (or reuse) image embeddings for an Execution's Case inputs.""" + body = _body(provider=provider, inputKey=input_key, retry=retry) + return self._command( + f"{self._execution_path(eval_id, run_id, execution_id)}/analyses/embeddings", body, idempotency_key + ) + + def get_embeddings( + self, + eval_id: str, + run_id: str, + execution_id: str, + *, + provider: Optional[str] = None, + async_task_id: Optional[str] = None, + ) -> Dict[str, Any]: + """Discover the job for ``provider`` or read a specific ``async_task_id`` attempt.""" + return self._call( + "GET", + f"{self._execution_path(eval_id, run_id, execution_id)}/analyses/embeddings", + params={"provider": provider, "asyncTaskId": async_task_id}, + ) + + def cancel_embeddings(self, eval_id: str, run_id: str, execution_id: str, async_task_id: str) -> Dict[str, Any]: + return self._call( + "POST", + f"{self._execution_path(eval_id, run_id, execution_id)}/analyses/embeddings/{_seg(async_task_id)}/cancel", + ) diff --git a/roboflow/core/workspace.py b/roboflow/core/workspace.py index d4e3a634..ebaf20dc 100644 --- a/roboflow/core/workspace.py +++ b/roboflow/core/workspace.py @@ -21,6 +21,7 @@ if TYPE_CHECKING: from roboflow.core.device import Device from roboflow.core.model_eval import ModelEval + from roboflow.core.workflow_evals import WorkflowEvals class Workspace: @@ -1661,6 +1662,17 @@ def eval(self, eval_id: str) -> "ModelEval": info = rfapi.get_model_eval(self.__api_key, self.url, eval_id) return ModelEval(self.__api_key, self.url, info.get("id", eval_id), info=info) + def workflow_evals(self) -> "WorkflowEvals": + """Return a client for this workspace's Workflow Evals. + + Example: + >>> evals = rf.workspace().workflow_evals() + >>> evals.list(limit=10)["items"] + """ + from roboflow.core.workflow_evals import WorkflowEvals + + return WorkflowEvals(self.__api_key, self.url) + def trash(self) -> dict: """ List items currently in the workspace Trash. diff --git a/tests/cli/test_workflow_eval_handler.py b/tests/cli/test_workflow_eval_handler.py new file mode 100644 index 00000000..c5d85a8f --- /dev/null +++ b/tests/cli/test_workflow_eval_handler.py @@ -0,0 +1,345 @@ +"""Tests for the Workflow Evals CLI handler (`roboflow workflow-eval ...`).""" + +from __future__ import annotations + +import json +import os +import tempfile +import unittest +from unittest.mock import patch + +from typer.testing import CliRunner + +from roboflow.adapters.workflowevalsapi import WorkflowEvalError +from roboflow.cli import app + +runner = CliRunner() + +COMMANDS = { + "": [ + "list", + "get", + "create", + "update", + "delete", + "capabilities", + "openapi", + "operation", + "compare", + ], + "spec": ["list", "get", "create", "update", "validate", "compatible-datasets", "ai-draft"], + "dataset": ["list", "get", "create", "update", "update-contract", "compatible-specs"], + "case": ["list", "get", "add", "replace", "delete", "import", "import-status", "upload"], + "binding": ["suggest", "validate"], + "preparation": ["start", "current", "get", "resume", "cases"], + "run": ["list", "get", "start", "cancel", "retry", "replay", "rescore", "delete", "config"], + "execution": ["list", "get", "config", "input-cases", "capture", "overview", "results", "result"], + "export": ["start", "status", "download-url"], + "embedding": ["start", "get", "cancel"], + "evaluator": ["list", "get"], + "schema": ["get"], + "agent": ["manifest", "skills", "skill", "bundles", "bundle", "asset"], +} + +CLIENT = "roboflow.core.workflow_evals.WorkflowEvals" + + +def _invoke(*argv: str, input: str | None = None): + return runner.invoke(app, ["--workspace", "ws", "--api-key", "key", *argv], input=input) + + +class TestWorkflowEvalRegistration(unittest.TestCase): + def test_every_subcommand_has_help(self) -> None: + for group, commands in COMMANDS.items(): + for command in commands: + argv = ["workflow-eval", *([group] if group else []), command, "--help"] + with self.subTest(command=" ".join(argv)): + result = runner.invoke(app, argv) + self.assertEqual(result.exit_code, 0, result.output) + + +class TestWorkflowEvalCommands(unittest.TestCase): + @patch(f"{CLIENT}.list") + def test_list_renders_table_and_cursor(self, mock_list) -> None: + mock_list.return_value = { + "items": [{"id": "e1", "name": "First", "state": "ready", "runCount": 2, "lastRunAt": None}], + "nextCursor": "next", + } + result = _invoke("workflow-eval", "list", "--limit", "5") + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("First", result.output) + self.assertIn("--cursor next", result.output) + mock_list.assert_called_once_with(limit=5, cursor=None, search=None, include_spec_summary=False) + + @patch(f"{CLIENT}.list") + def test_list_json_is_raw_page(self, mock_list) -> None: + page = {"items": [{"id": "e1"}]} + mock_list.return_value = page + result = _invoke("--json", "workflow-eval", "list") + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(json.loads(result.output), page) + + @patch(f"{CLIENT}.create") + def test_create_merges_body_and_flags(self, mock_create) -> None: + mock_create.return_value = {"id": "e1"} + body = json.dumps({"name": "from body", "subject": {"kind": "savedWorkflow"}}) + result = _invoke("--json", "workflow-eval", "create", "--body", body, "--name", "flag", "--spec", "s1") + self.assertEqual(result.exit_code, 0, result.output) + payload = mock_create.call_args.args[0] + self.assertEqual(payload, {"name": "flag", "subject": {"kind": "savedWorkflow"}, "specId": "s1"}) + + @patch(f"{CLIENT}.update") + def test_update_sends_revision(self, mock_update) -> None: + mock_update.return_value = {"id": "e1", "revision": 4} + result = _invoke("--json", "workflow-eval", "update", "e1", "--revision", "3", "--dataset", "d1") + self.assertEqual(result.exit_code, 0, result.output) + mock_update.assert_called_once_with("e1", {"evalDatasetId": "d1"}, revision="3") + + def test_update_requires_revision(self) -> None: + result = _invoke("workflow-eval", "update", "e1", "--name", "x") + self.assertNotEqual(result.exit_code, 0) + + @patch(f"{CLIENT}.create_spec") + def test_spec_create_reads_body_from_file(self, mock_create) -> None: + mock_create.return_value = {"id": "s1"} + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "spec.json") + with open(path, "w") as handle: + json.dump({"spec": {"checks": []}}, handle) + result = _invoke("--json", "workflow-eval", "spec", "create", "--body", f"@{path}", "--name", "S") + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(mock_create.call_args.args[0], {"spec": {"checks": []}, "name": "S"}) + + @patch(f"{CLIENT}.validate_spec") + def test_spec_validate_reads_stdin(self, mock_validate) -> None: + mock_validate.return_value = {"valid": True} + result = _invoke("--json", "workflow-eval", "spec", "validate", "--body", "-", input='{"checks": []}') + self.assertEqual(result.exit_code, 0, result.output) + mock_validate.assert_called_once_with({"checks": []}) + + def test_invalid_json_body_is_a_structured_error(self) -> None: + result = _invoke("--json", "workflow-eval", "spec", "validate", "--body", "{not json") + self.assertEqual(result.exit_code, 1) + self.assertIn("--body is not valid JSON", result.output) + + @patch(f"{CLIENT}.get") + def test_api_errors_map_to_exit_codes_and_json(self, mock_get) -> None: + mock_get.side_effect = WorkflowEvalError( + "Workflow Eval not found", status_code=404, code="NOT_FOUND", request_id="req-1", retryable=False + ) + result = _invoke("--json", "workflow-eval", "get", "missing") + self.assertEqual(result.exit_code, 3) + error = json.loads(result.output.strip().splitlines()[-1])["error"] + self.assertEqual(error["code"], "NOT_FOUND") + self.assertEqual(error["requestId"], "req-1") + self.assertEqual(error["message"], "Workflow Eval not found") + + @patch(f"{CLIENT}.list") + def test_scope_error_exits_with_auth_code_and_hint(self, mock_list) -> None: + mock_list.side_effect = WorkflowEvalError("Missing scope", status_code=403, code="FORBIDDEN") + result = _invoke("workflow-eval", "list") + self.assertEqual(result.exit_code, 2) + self.assertIn("workflow-evals", result.output) + + @patch(f"{CLIENT}.get_by_url") + def test_get_by_url(self, mock_get) -> None: + mock_get.return_value = {"id": "e1"} + result = _invoke("--json", "workflow-eval", "get", "my-eval", "--by-url") + self.assertEqual(result.exit_code, 0, result.output) + mock_get.assert_called_once_with("my-eval") + + @patch(f"{CLIENT}.delete") + def test_delete_confirms_with_deletion_key(self, mock_delete) -> None: + mock_delete.side_effect = [ + WorkflowEvalError( + "Deletion confirmation is required", + status_code=409, + code="deletion_confirmation_required", + details={"impact": {"runs": 1}, "deletionKey": "dk"}, + ), + {"asyncTaskId": "t1", "statusUrl": "/ops/t1"}, + ] + result = _invoke("--json", "workflow-eval", "delete", "e1", "--yes") + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(json.loads(result.output)["asyncTaskId"], "t1") + self.assertEqual(mock_delete.call_args_list[0].kwargs, {"deletion_key": None}) + self.assertEqual(mock_delete.call_args_list[1].kwargs, {"deletion_key": "dk"}) + + @patch(f"{CLIENT}.delete_run") + def test_run_delete_without_yes_refuses_when_not_a_tty(self, mock_delete) -> None: + mock_delete.side_effect = WorkflowEvalError( + "Deletion confirmation is required", + status_code=409, + code="deletion_confirmation_required", + details={"deletionKey": "dk"}, + ) + result = _invoke("workflow-eval", "run", "delete", "r1", "--eval", "e1") + self.assertEqual(result.exit_code, 1) + self.assertEqual(mock_delete.call_count, 1) + + @patch(f"{CLIENT}.import_cases") + def test_case_import_builds_items(self, mock_import) -> None: + mock_import.return_value = {"intakeId": "i1"} + result = _invoke( + "--json", + "workflow-eval", + "case", + "import", + "--dataset", + "d1", + "--input-field", + "image", + "--source", + "src-1", + "--from-dataset", + "proj:valid", + "--from-dataset", + "proj2", + "--artifact", + "art-1", + "--slice", + "night", + ) + self.assertEqual(result.exit_code, 0, result.output) + dataset_id, payload = mock_import.call_args.args + self.assertEqual(dataset_id, "d1") + self.assertEqual( + payload, + { + "inputField": "image", + "slices": ["night"], + "items": [ + {"kind": "source", "sourceId": "src-1"}, + {"kind": "dataset", "datasetId": "proj", "split": "valid"}, + {"kind": "dataset", "datasetId": "proj2"}, + {"kind": "artifact", "artifactId": "art-1"}, + ], + }, + ) + + @patch(f"{CLIENT}.delete_case") + def test_case_delete_with_yes(self, mock_delete) -> None: + result = _invoke("--json", "workflow-eval", "case", "delete", "c1", "-d", "d1", "--revision", "2", "--yes") + self.assertEqual(result.exit_code, 0, result.output) + mock_delete.assert_called_once_with("d1", "c1", revision="2") + self.assertEqual(json.loads(result.output), {"deleted": True, "caseId": "c1"}) + + @patch(f"{CLIENT}.suggest_bindings") + def test_binding_suggest_builds_saved_workflow_version_subject(self, mock_suggest) -> None: + mock_suggest.return_value = {"bindingSet": {}} + result = _invoke( + "--json", + "workflow-eval", + "binding", + "suggest", + "--spec", + "s1", + "--dataset", + "d1", + "--workflow", + "wf", + "--workflow-version", + "v2", + ) + self.assertEqual(result.exit_code, 0, result.output) + mock_suggest.assert_called_once_with( + "s1", + "d1", + {"kind": "savedWorkflowVersion", "workflowId": "wf", "subjectKey": "wf", "workflowVersionId": "v2"}, + ) + + def test_binding_suggest_requires_a_subject(self) -> None: + result = _invoke("workflow-eval", "binding", "suggest", "--spec", "s1", "--dataset", "d1") + self.assertEqual(result.exit_code, 1) + self.assertIn("--workflow", result.output) + + @patch(f"{CLIENT}.start_case_preparation") + def test_preparation_start_defaults_and_explicit_cases(self, mock_start) -> None: + mock_start.return_value = {"id": "p1"} + result = _invoke("--json", "workflow-eval", "preparation", "start", "-e", "e1", "--case", "c1", "--case", "c2") + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + mock_start.call_args.args[1], + {"runtime": {"kind": "serverless"}, "caseSelection": {"kind": "explicit", "caseIds": ["c1", "c2"]}}, + ) + + @patch(f"{CLIENT}.wait_for_run") + @patch(f"{CLIENT}.start_run") + def test_run_start_with_wait_polls_the_admitted_run(self, mock_start, mock_wait) -> None: + mock_start.return_value = {"run": {"id": "r1", "state": "queued"}, "executions": []} + mock_wait.return_value = {"id": "r1", "state": "completed"} + body = json.dumps({"executions": [{"subject": {}, "bindingSet": {}}]}) + 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) + + @patch(f"{CLIENT}.list_case_results") + def test_execution_results_filters(self, mock_results) -> None: + mock_results.return_value = {"items": []} + result = _invoke( + "workflow-eval", + "execution", + "results", + "x1", + "-e", + "e1", + "-r", + "r1", + "--slice", + "a", + "--failed-check", + "--state", + "failed", + ) + self.assertEqual(result.exit_code, 0, result.output) + kwargs = mock_results.call_args.kwargs + self.assertEqual(kwargs["slices"], ["a"]) + self.assertTrue(kwargs["failed_check"]) + self.assertEqual(kwargs["state"], "failed") + + @patch(f"{CLIENT}.compare") + def test_compare_collects_execution_ids(self, mock_compare) -> None: + mock_compare.return_value = {"checks": []} + result = _invoke("--json", "workflow-eval", "compare", "-e", "e1", "-x", "a", "-x", "b") + self.assertEqual(result.exit_code, 0, result.output) + mock_compare.assert_called_once_with("e1", {"executionIds": ["a", "b"]}) + + @patch(f"{CLIENT}.wait_for_export") + @patch(f"{CLIENT}.start_export") + def test_export_start_with_wait(self, mock_start, mock_wait) -> None: + mock_start.return_value = {"asyncTaskId": "t1", "pollUrl": "/p"} + mock_wait.return_value = {"asyncTaskId": "t1", "state": "completed"} + result = _invoke("--json", "workflow-eval", "export", "start", "-e", "e1", "-r", "r1", "-f", "csv", "--wait") + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(mock_start.call_args.args[2], {"format": "csv"}) + mock_wait.assert_called_once_with("e1", "r1", "t1", timeout=1800) + + @patch(f"{CLIENT}.start_export") + def test_export_start_keeps_body_format_and_defaults_to_json(self, mock_start) -> None: + mock_start.return_value = {"asyncTaskId": "t1"} + result = _invoke( + "--json", "workflow-eval", "export", "start", "-e", "e1", "-r", "r1", "--body", '{"format": "xlsx"}' + ) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(mock_start.call_args.args[2], {"format": "xlsx"}) + _invoke("--json", "workflow-eval", "export", "start", "-e", "e1", "-r", "r1") + self.assertEqual(mock_start.call_args.args[2], {"format": "json"}) + + @patch(f"{CLIENT}.get_agent_skill") + def test_agent_skill_prints_markdown_in_text_mode(self, mock_skill) -> None: + mock_skill.return_value = {"engineVersion": "1", "id": "skill:x", "path": "x.md", "content": "# Skill"} + result = _invoke("workflow-eval", "agent", "skill", "skill:x") + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(result.output.strip(), "# Skill") + + @patch(f"{CLIENT}.ai_draft") + def test_spec_ai_draft(self, mock_draft) -> None: + mock_draft.return_value = {"spec": {}, "workflowBindings": {}} + result = _invoke("--json", "workflow-eval", "spec", "ai-draft", "-e", "e1", "-i", "be strict") + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(mock_draft.call_args.args, ("e1", "be strict")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workflow_evals.py b/tests/test_workflow_evals.py new file mode 100644 index 00000000..4aca881b --- /dev/null +++ b/tests/test_workflow_evals.py @@ -0,0 +1,395 @@ +import json +import os +import tempfile +import unittest +import uuid +from unittest.mock import patch + +import responses + +from roboflow.adapters import workflowevalsapi +from roboflow.adapters.workflowevalsapi import WorkflowEvalError +from roboflow.config import API_URL +from roboflow.core.workflow_evals import WorkflowEvals + +WS = "test-ws" +KEY = "test_key" +BASE = f"{API_URL}/workspaces/{WS}/workflow-evals" + +E, R, X, D, S, C = "eval-1", "run-1", "exec-1", "ds-1", "spec-1", "case-1" +SUBJECT = {"kind": "savedWorkflow", "workflowId": "wf", "subjectKey": "wf"} + +# (client method, args, kwargs, HTTP method, path relative to BASE, sends Idempotency-Key, If-Match value) +# One row per public JSON route in the Workflow Evals OpenAPI registry. The 71st route, +# POST /specs/ai-draft, streams SSE and is covered by the ai_draft tests below. +ROUTES = [ + ("capabilities", (), {}, "GET", "/capabilities", False, None), + ("openapi", (), {}, "GET", "/openapi.json", False, None), + ("list_evaluators", (), {}, "GET", "/evaluators", False, None), + ("get_evaluator", ("bool-match",), {}, "GET", "/evaluators/bool-match", False, None), + ("get_schema", ("spec",), {}, "GET", "/schemas/spec", False, None), + ("get_operation", ("op-1",), {}, "GET", "/operations/op-1", False, None), + ("agent_manifest", (), {}, "GET", "/agent/manifest", False, None), + ("list_agent_skills", (), {}, "GET", "/agent/skills", False, None), + ("get_agent_skill", ("skill:create-eval",), {}, "GET", "/agent/skills/skill%3Acreate-eval", False, None), + ("list_agent_bundles", (), {}, "GET", "/agent/bundles", False, None), + ("get_agent_bundle", ("bundle:create-eval",), {}, "GET", "/agent/bundles/bundle%3Acreate-eval", False, None), + ("get_agent_asset", ("schema:spec",), {}, "GET", "/agent/assets/schema%3Aspec", False, None), + ("list", (), {}, "GET", "", False, None), + ("create", (), {"name": "n"}, "POST", "", True, None), + ("get", (E,), {}, "GET", f"/{E}", False, None), + ("get_by_url", ("my-eval",), {}, "GET", "/by-url/my-eval", False, None), + ("update", (E, {"name": "n"}), {"revision": 3}, "PATCH", f"/{E}", False, "3"), + ("delete", (E,), {"deletion_key": "dk"}, "DELETE", f"/{E}", False, None), + ("list_specs", (), {}, "GET", "/specs", False, None), + ("create_spec", ({"spec": {}},), {}, "POST", "/specs", True, None), + ("get_spec", (S,), {}, "GET", f"/specs/{S}", False, None), + ("update_spec", (S, {"name": "n"}), {"revision": 2}, "PATCH", f"/specs/{S}", False, "2"), + ("validate_spec", ({"checks": []},), {}, "POST", "/specs/validate", False, None), + ("list_compatible_datasets", (S,), {}, "GET", f"/specs/{S}/compatible-eval-datasets", False, None), + ("list_datasets", (), {}, "GET", "/eval-datasets", False, None), + ("create_dataset", ({"name": "d"},), {}, "POST", "/eval-datasets", True, None), + ("get_dataset", (D,), {}, "GET", f"/eval-datasets/{D}", False, None), + ("update_dataset", (D, {"name": "d"}), {"revision": 1}, "PATCH", f"/eval-datasets/{D}", False, "1"), + ( + "mutate_dataset_contract", + (D, {"inputs": []}), + {"revision": 1}, + "POST", + f"/eval-datasets/{D}/contract-mutations", + False, + "1", + ), + ("list_compatible_specs", (D,), {}, "GET", f"/eval-datasets/{D}/compatible-specs", False, None), + ("list_cases", (D,), {}, "GET", f"/eval-datasets/{D}/cases", False, None), + ("add_case", (D, {"name": "c"}), {}, "POST", f"/eval-datasets/{D}/cases", True, None), + ("get_case", (D, C), {}, "GET", f"/eval-datasets/{D}/cases/{C}", False, None), + ("replace_case", (D, C, {"inputs": {}}), {"revision": 4}, "PUT", f"/eval-datasets/{D}/cases/{C}", False, "4"), + ("delete_case", (D, C), {"revision": 4}, "DELETE", f"/eval-datasets/{D}/cases/{C}", False, "4"), + ("import_cases", (D, {"items": []}), {}, "POST", f"/eval-datasets/{D}/cases/import", True, None), + ("get_case_import", (D, "in-1"), {}, "GET", f"/eval-datasets/{D}/case-imports/in-1", False, None), + ( + "prepare_case_asset_upload", + (D,), + {"asset_name": "a.png", "content_type": "image/png", "expected_size": 3}, + "POST", + f"/eval-datasets/{D}/case-assets/uploads", + True, + None, + ), + ( + "complete_case_asset_upload", + (D, "up-1"), + {}, + "POST", + f"/eval-datasets/{D}/case-assets/uploads/up-1/complete", + False, + None, + ), + ("suggest_bindings", (S, D, SUBJECT), {}, "POST", "/bindings/suggest", False, None), + ("validate_bindings", (S, D, SUBJECT, {}), {}, "POST", "/bindings/validate", False, None), + ("start_case_preparation", (E, {"runtime": {}}), {}, "POST", f"/{E}/case-preparations", True, None), + ("get_current_case_preparation", (E,), {}, "GET", f"/{E}/case-preparations/current", False, None), + ("get_case_preparation", (E, "p-1"), {}, "GET", f"/{E}/case-preparations/p-1", False, None), + ("resume_case_preparation", (E, "p-1"), {}, "POST", f"/{E}/case-preparations/p-1/resume", False, None), + ("list_case_preparation_cases", (E, "p-1"), {}, "GET", f"/{E}/case-preparations/p-1/cases", False, None), + ("list_runs", (E,), {}, "GET", f"/{E}/runs", False, None), + ("start_run", (E, {"executions": []}), {}, "POST", f"/{E}/runs", True, None), + ("get_run", (E, R), {}, "GET", f"/{E}/runs/{R}", False, None), + ("get_run_configuration", (E, R), {}, "GET", f"/{E}/runs/{R}/configuration", False, None), + ("cancel_run", (E, R), {}, "POST", f"/{E}/runs/{R}/cancel", False, None), + ("retry_run", (E, R), {}, "POST", f"/{E}/runs/{R}/retry-failed", True, None), + ("replay_run", (E, R), {}, "POST", f"/{E}/runs/{R}/replay", True, None), + ("rescore_run", (E, R, {"specId": S}), {}, "POST", f"/{E}/runs/{R}/rescore", True, None), + ("delete_run", (E, R), {}, "DELETE", f"/{E}/runs/{R}", False, None), + ("list_executions", (E, R), {}, "GET", f"/{E}/runs/{R}/executions", False, None), + ("get_execution", (E, R, X), {}, "GET", f"/{E}/runs/{R}/executions/{X}", False, None), + ( + "get_execution_configuration", + (E, R, X), + {}, + "GET", + f"/{E}/runs/{R}/executions/{X}/configuration", + False, + None, + ), + ("list_execution_input_cases", (E, R, X), {}, "GET", f"/{E}/runs/{R}/executions/{X}/input-cases", False, None), + ( + "submit_capture", + (E, R, X, C, {"attemptCount": 1, "shardIndex": 0, "output": None}), + {}, + "POST", + f"/{E}/runs/{R}/executions/{X}/cases/{C}/capture", + False, + None, + ), + ("get_results_overview", (E, R, X), {}, "GET", f"/{E}/runs/{R}/executions/{X}/results/overview", False, None), + ("list_case_results", (E, R, X), {}, "GET", f"/{E}/runs/{R}/executions/{X}/cases", False, None), + ("get_case_result", (E, R, X, C), {}, "GET", f"/{E}/runs/{R}/executions/{X}/cases/{C}", False, None), + ("compare", (E, {"executionIds": [X]}), {}, "POST", f"/{E}/comparisons", False, None), + ("start_export", (E, R, {"format": "json"}), {}, "POST", f"/{E}/runs/{R}/exports", True, None), + ("get_export_job", (E, R, "t-1"), {}, "GET", f"/{E}/runs/{R}/export-jobs/t-1", False, None), + ("get_export_download", (E, R, "a-1"), {}, "GET", f"/{E}/runs/{R}/exports/a-1", False, None), + ( + "start_embeddings", + (E, R, X, "clip"), + {}, + "POST", + f"/{E}/runs/{R}/executions/{X}/analyses/embeddings", + True, + None, + ), + ("get_embeddings", (E, R, X), {}, "GET", f"/{E}/runs/{R}/executions/{X}/analyses/embeddings", False, None), + ( + "cancel_embeddings", + (E, R, X, "t-1"), + {}, + "POST", + f"/{E}/runs/{R}/executions/{X}/analyses/embeddings/t-1/cancel", + False, + None, + ), +] + + +def _is_uuid4(value): + try: + return uuid.UUID(value).version == 4 + except (TypeError, ValueError): + return False + + +class TestWorkflowEvalsRoutes(unittest.TestCase): + def setUp(self): + self.client = WorkflowEvals(KEY, WS) + + def test_covers_every_public_json_route(self): + self.assertEqual(len({(row[3], row[4]) for row in ROUTES}), 70) + + @responses.activate + def test_each_method_calls_its_route(self): + for name, args, kwargs, method, path, idempotent, if_match in ROUTES: + with self.subTest(method=name): + responses.reset() + responses.add(method, f"{BASE}{path}", json={"ok": True}, status=200) + getattr(self.client, name)(*args, **kwargs) + request = responses.calls[0].request + self.assertEqual(request.method, method) + self.assertEqual(request.url.split("?")[0], f"{BASE}{path}") + self.assertEqual(request.headers["Authorization"], f"Bearer {KEY}") + self.assertEqual(_is_uuid4(request.headers.get("Idempotency-Key")), idempotent) + self.assertEqual(request.headers.get("If-Match"), if_match) + + +class TestWorkflowEvalsClient(unittest.TestCase): + def setUp(self): + self.client = WorkflowEvals(KEY, WS) + + @responses.activate + def test_create_merges_body_and_keyword_fields(self): + responses.add(responses.POST, BASE, json={"id": E}, status=201) + key = str(uuid.uuid4()) + self.client.create({"subject": SUBJECT, "name": "old"}, name="new", spec_id=S, idempotency_key=key) + request = responses.calls[0].request + self.assertEqual(json.loads(request.body), {"subject": SUBJECT, "name": "new", "specId": S}) + self.assertEqual(request.headers["Idempotency-Key"], key) + + @responses.activate + def test_list_query_parameters(self): + responses.add(responses.GET, BASE, json={"items": []}, status=200) + self.client.list(limit=10, cursor="c", include_spec_summary=True) + self.assertEqual(responses.calls[0].request.params, {"limit": "10", "cursor": "c", "include": "specSummary"}) + + @responses.activate + def test_case_results_filters_repeat_slices_and_encode_booleans(self): + url = f"{BASE}/{E}/runs/{R}/executions/{X}/cases" + responses.add(responses.GET, url, json={"items": []}, status=200) + self.client.list_case_results(E, R, X, slices=["a", "b"], failed_check=True, state="failed") + query = responses.calls[0].request.url.split("?", 1)[1] + self.assertIn("slice=a&slice=b", query) + self.assertIn("failedCheck=true", query) + self.assertIn("state=failed", query) + + @responses.activate + def test_validate_spec_wraps_bare_spec(self): + responses.add(responses.POST, f"{BASE}/specs/validate", json={"valid": True}, status=200) + self.client.validate_spec({"checks": []}) + self.assertEqual(json.loads(responses.calls[0].request.body), {"spec": {"checks": []}}) + + @responses.activate + def test_no_content_returns_none(self): + responses.add(responses.GET, f"{BASE}/{E}/case-preparations/current", status=204) + self.assertIsNone(self.client.get_current_case_preparation(E)) + + @responses.activate + def test_error_envelope_is_parsed(self): + body = { + "error": { + "code": "INVALID_SPEC", + "category": "validation", + "retryable": False, + "requestId": "req-1", + "message": "Spec is invalid", + "details": {"diagnostics": [{"path": "$.checks"}]}, + } + } + responses.add(responses.POST, f"{BASE}/specs", json=body, status=422) + with self.assertRaises(WorkflowEvalError) as ctx: + self.client.create_spec({"spec": {}}) + error = ctx.exception + self.assertEqual(error.status_code, 422) + self.assertEqual(error.code, "INVALID_SPEC") + self.assertFalse(error.retryable) + self.assertEqual(error.request_id, "req-1") + self.assertEqual(error.details["diagnostics"], [{"path": "$.checks"}]) + self.assertEqual(str(error), "Spec is invalid") + + @responses.activate + def test_platform_auth_error_envelope(self): + body = {"error": {"message": "Unauthorized", "type": "OAuthException", "hint": "check key"}} + responses.add(responses.GET, BASE, json=body, status=401) + with self.assertRaises(WorkflowEvalError) as ctx: + self.client.list() + self.assertEqual(ctx.exception.status_code, 401) + self.assertEqual(ctx.exception.category, "OAuthException") + self.assertEqual(ctx.exception.hint, "check key") + + @responses.activate + def test_delete_confirmed_resends_with_deletion_key(self): + confirm = { + "error": { + "code": "deletion_confirmation_required", + "category": "conflict", + "retryable": False, + "requestId": "r", + "message": "Deletion confirmation is required", + "details": {"impact": {"runs": 2}, "deletionKey": "dk-1"}, + } + } + responses.add(responses.DELETE, f"{BASE}/{E}", json=confirm, status=409) + responses.add(responses.DELETE, f"{BASE}/{E}", json={"asyncTaskId": "t", "statusUrl": "/s"}, status=202) + result = self.client.delete_confirmed(E) + self.assertEqual(result["asyncTaskId"], "t") + self.assertNotIn("Deletion-Key", responses.calls[0].request.headers) + self.assertEqual(responses.calls[1].request.headers["Deletion-Key"], "dk-1") + + @responses.activate + def test_delete_confirmed_reraises_other_conflicts(self): + body = {"error": {"code": "CONFLICT", "message": "busy", "retryable": True, "requestId": "r"}} + responses.add(responses.DELETE, f"{BASE}/{E}/runs/{R}", json=body, status=409) + with self.assertRaises(WorkflowEvalError): + self.client.delete_run_confirmed(E, R) + self.assertEqual(len(responses.calls), 1) + + @responses.activate + def test_ai_draft_streams_events_and_returns_draft(self): + stream = ( + 'event: status\ndata: {"message": "Reading Eval"}\n\n' + 'event: delta\ndata: {"delta": "..."}\n\n' + ": keep-alive\n\n" + 'event: draft\ndata: {"spec": {"id": "s"}, "workflowBindings": {}}\n\n' + "event: done\ndata: {}\n\n" + ) + responses.add( + responses.POST, + f"{BASE}/specs/ai-draft", + body=stream, + status=200, + content_type="text/event-stream", + ) + events = [] + draft = self.client.ai_draft(E, "be strict", on_event=events.append) + self.assertEqual(draft, {"spec": {"id": "s"}, "workflowBindings": {}}) + self.assertEqual([event["event"] for event in events], ["status", "delta", "draft", "done"]) + request = responses.calls[0].request + self.assertEqual(json.loads(request.body), {"evalId": E, "instructions": "be strict"}) + self.assertTrue(_is_uuid4(request.headers["Idempotency-Key"])) + self.assertEqual(request.headers["Accept"], "text/event-stream") + + @responses.activate + def test_ai_draft_decodes_utf8_without_charset(self): + stream = 'event: draft\ndata: {"name": "Avaliação → Ångström 中文"}\n\nevent: done\ndata: {}\n\n' + responses.add( + responses.POST, + f"{BASE}/specs/ai-draft", + body=stream.encode("utf-8"), + status=200, + content_type="text/event-stream", + ) + self.assertEqual(self.client.ai_draft(E), {"name": "Avaliação → Ångström 中文"}) + + @responses.activate + def test_ai_draft_error_event_raises(self): + stream = 'event: error\ndata: {"message": "No subject"}\n\nevent: done\ndata: {}\n\n' + responses.add(responses.POST, f"{BASE}/specs/ai-draft", body=stream, status=200) + with self.assertRaises(WorkflowEvalError) as ctx: + self.client.ai_draft(E) + self.assertIn("No subject", str(ctx.exception)) + + @responses.activate + def test_upload_case_asset_runs_the_three_steps(self): + upload_url = "https://storage.googleapis.com/bucket/object?sig=1" + responses.add( + responses.POST, + f"{BASE}/eval-datasets/{D}/case-assets/uploads", + json={ + "uploadId": "up-1", + "artifactId": "art-1", + "uploadUrl": upload_url, + "requiredHeaders": {"Content-Type": "image/png", "x-goog-meta": "1"}, + "expiresInSeconds": 600, + }, + status=201, + ) + responses.add(responses.PUT, upload_url, status=200) + responses.add( + responses.POST, + f"{BASE}/eval-datasets/{D}/case-assets/uploads/up-1/complete", + json={"artifactId": "art-1"}, + status=200, + ) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "pixel.png") + with open(path, "wb") as handle: + handle.write(b"png") + asset = self.client.upload_case_asset(D, path) + self.assertEqual(asset, {"artifactId": "art-1"}) + intent = json.loads(responses.calls[0].request.body) + self.assertEqual(intent, {"assetName": "pixel.png", "contentType": "image/png", "expectedSize": 3}) + put = responses.calls[1].request + self.assertEqual(put.body, b"png") + self.assertEqual(put.headers["x-goog-meta"], "1") + self.assertNotIn("Authorization", put.headers) + + @responses.activate + @patch("roboflow.core.workflow_evals.time.sleep") + def test_wait_for_run_polls_until_terminal(self, _sleep): + url = f"{BASE}/{E}/runs/{R}" + responses.add(responses.GET, url, json={"id": R, "state": "running"}, status=200) + responses.add(responses.GET, url, json={"id": R, "state": "completed"}, status=200) + run = self.client.wait_for_run(E, R, interval=0) + self.assertEqual(run["state"], "completed") + self.assertEqual(len(responses.calls), 2) + + +class TestSseParser(unittest.TestCase): + def test_multiline_data_and_plain_text(self): + lines = ["event: delta", "data: line one", "data: line two", "", 'data: {"a": 1}', ""] + events = list(workflowevalsapi.parse_sse(lines)) + self.assertEqual(events[0], {"event": "delta", "data": "line one\nline two"}) + self.assertEqual(events[1], {"event": "message", "data": {"a": 1}}) + + +class TestWorkspaceAccessor(unittest.TestCase): + def test_workspace_returns_bound_client(self): + from roboflow.core.workspace import Workspace + + info = {"workspace": {"name": "Test", "url": WS, "projects": [], "members": []}} + workspace = Workspace(info, api_key=KEY, default_workspace=WS, model_format="yolov8") + client = workspace.workflow_evals() + self.assertIsInstance(client, WorkflowEvals) + self.assertEqual(client.workspace_url, WS) + + +if __name__ == "__main__": + unittest.main()