add preview method

Author: sid-rl

src/runloop_api_client/sdk/__init__.py

@@ -7,6 +7,7 @@
 
 from .sync import AgentOps, DevboxOps, ScorerOps, RunloopSDK, ScenarioOps, SnapshotOps, BlueprintOps, StorageObjectOps
 from .agent import Agent
+from ._types import ScenarioPreview
 from .async_ import (
     AsyncAgentOps,
     AsyncDevboxOps,
@@ -75,6 +76,7 @@
     "AsyncScenarioRun",
     "ScenarioBuilder",
     "AsyncScenarioBuilder",
+    "ScenarioPreview",
     "Scorer",
     "AsyncScorer",
     "Snapshot",

src/runloop_api_client/sdk/_types.py

@@ -1,10 +1,13 @@
-from typing import Union, Callable, Optional
+from typing import Dict, Union, Callable, Optional
 from typing_extensions import TypedDict
 
 from .._types import Body, Query, Headers, Timeout, NotGiven
 from ..lib.polling import PollingConfig
 from ..types.devboxes import DiskSnapshotListParams, DiskSnapshotUpdateParams
 from ..types.scenarios import ScorerListParams, ScorerCreateParams, ScorerUpdateParams, ScorerValidateParams
+from ..types.input_context import InputContext
+from ..types.scenario_view import ScenarioView
+from ..types.scoring_contract import ScoringContract
 from ..types.agent_list_params import AgentListParams
 from ..types.devbox_list_params import DevboxListParams
 from ..types.object_list_params import ObjectListParams
@@ -186,3 +189,27 @@ class SDKScenarioRunAsyncParams(ScenarioStartRunBaseParams, LongRequestOptions):
 
 class SDKScenarioRunParams(ScenarioStartRunBaseParams, LongPollingRequestOptions):
     pass
+
+
+class InputContextPreview(InputContext):
+    problem_statement: Optional[str] = None  # type: ignore[assignment]
+    """The problem statement for the Scenario."""
+
+
+class ScenarioPreview(ScenarioView):
+    """Preview of scenario configuration with all fields optional."""
+
+    id: Optional[str] = None  # type: ignore[assignment]
+    """The ID of the Scenario."""
+
+    input_context: Optional[InputContextPreview] = None  # type: ignore[assignment]
+    """The input context for the Scenario."""
+
+    metadata: Optional[Dict[str, str]] = None  # type: ignore[assignment]
+    """User defined metadata to attach to the scenario for organization."""
+
+    name: Optional[str] = None  # type: ignore[assignment]
+    """The name of the Scenario."""
+
+    scoring_contract: Optional[ScoringContract] = None  # type: ignore[assignment]
+    """The scoring contract for the Scenario."""

src/runloop_api_client/sdk/async_scenario_builder.py

@@ -6,7 +6,7 @@
 from typing_extensions import Self, Unpack, Literal, override
 
 from ..types import ScenarioCreateParams, ScenarioEnvironmentParam
-from ._types import LongRequestOptions
+from ._types import ScenarioPreview, LongRequestOptions
 from .._client import AsyncRunloop
 from .async_scenario import AsyncScenario
 from .async_snapshot import AsyncSnapshot
@@ -387,6 +387,22 @@ def with_validation_type(self, validation_type: Literal["UNSPECIFIED", "FORWARD"
         self._validation_type = validation_type
         return self
 
+    def _build_normalized_scorers(self) -> List[ScoringFunctionParam]:
+        """Build normalized scorers list."""
+        total_weight = sum(s["weight"] for s in self._scorers)
+        return [{**s, "weight": s["weight"] / total_weight} for s in self._scorers]
+
+    def _build_environment_params(self) -> ScenarioEnvironmentParam:
+        """Build environment parameters."""
+        env_params: ScenarioEnvironmentParam = {}
+        if self._blueprint:
+            env_params["blueprint_id"] = self._blueprint.id
+        if self._snapshot:
+            env_params["snapshot_id"] = self._snapshot.id
+        if self._working_directory:
+            env_params["working_directory"] = self._working_directory
+        return env_params
+
     def _build_params(self) -> ScenarioCreateParams:
         """Build the scenario creation parameters.
 
@@ -405,51 +421,50 @@ def _build_params(self) -> ScenarioCreateParams:
                 "Call add_test_command_scorer(), add_bash_script_scorer(), or another scorer method first."
             )
 
-        # Normalize weights to sum to 1.0
-        total_weight = sum(s["weight"] for s in self._scorers)
-        normalized_scorers: List[ScoringFunctionParam] = [
-            {**s, "weight": s["weight"] / total_weight} for s in self._scorers
-        ]
-
-        params: ScenarioCreateParams = {
+        return {
             "name": self._name,
             "input_context": {
                 "problem_statement": self._problem_statement,
+                "additional_context": self._additional_context,
             },
             "scoring_contract": {
-                "scoring_function_parameters": normalized_scorers,
+                "scoring_function_parameters": self._build_normalized_scorers(),
             },
+            "environment_parameters": self._build_environment_params(),
+            "metadata": self._metadata,
+            "reference_output": self._reference_output,
+            "required_environment_variables": self._required_env_vars,
+            "required_secret_names": self._required_secrets,
+            "validation_type": self._validation_type,
         }
 
-        # Add additional context if set
-        if self._additional_context is not None:
-            params["input_context"]["additional_context"] = self._additional_context
+    def preview(self) -> ScenarioPreview:
+        """Preview the scenario configuration without pushing to the platform.
 
-        # Build environment parameters if any are set
-        env_params: ScenarioEnvironmentParam = {}
-        if self._blueprint:
-            env_params["blueprint_id"] = self._blueprint.id
-        if self._snapshot:
-            env_params["snapshot_id"] = self._snapshot.id
-        if self._working_directory:
-            env_params["working_directory"] = self._working_directory
+        Returns the current configuration state as a ScenarioPreview object.
+        Does not validate or raise errors for missing required fields.
 
-        if env_params:
-            params["environment_parameters"] = env_params
-
-        # Add optional fields
-        if self._metadata:
-            params["metadata"] = self._metadata
-        if self._reference_output:
-            params["reference_output"] = self._reference_output
-        if self._required_env_vars:
-            params["required_environment_variables"] = self._required_env_vars
-        if self._required_secrets:
-            params["required_secret_names"] = self._required_secrets
-        if self._validation_type:
-            params["validation_type"] = self._validation_type
-
-        return params
+        :return: Preview of the scenario configuration
+        :rtype: ScenarioPreview
+        """
+        return ScenarioPreview.model_validate(
+            {
+                "name": self._name,
+                "input_context": {
+                    "problem_statement": self._problem_statement,
+                    "additional_context": self._additional_context,
+                },
+                "scoring_contract": {
+                    "scoring_function_parameters": self._build_normalized_scorers(),
+                },
+                "environment": self._build_environment_params(),
+                "metadata": self._metadata,
+                "reference_output": self._reference_output,
+                "required_environment_variables": self._required_env_vars,
+                "required_secret_names": self._required_secrets,
+                "validation_type": self._validation_type,
+            }
+        )
 
     async def push(self, **options: Unpack[LongRequestOptions]) -> AsyncScenario:
         """Create the scenario on the platform.

src/runloop_api_client/sdk/scenario_builder.py

@@ -6,7 +6,7 @@
 from typing_extensions import Self, Unpack, Literal, override
 
 from ..types import ScenarioCreateParams, ScenarioEnvironmentParam
-from ._types import LongRequestOptions
+from ._types import ScenarioPreview, LongRequestOptions
 from .._client import Runloop
 from .scenario import Scenario
 from .snapshot import Snapshot
@@ -387,6 +387,22 @@ def with_validation_type(self, validation_type: Literal["UNSPECIFIED", "FORWARD"
         self._validation_type = validation_type
         return self
 
+    def _build_normalized_scorers(self) -> List[ScoringFunctionParam]:
+        """Build normalized scorers list."""
+        total_weight = sum(s["weight"] for s in self._scorers)
+        return [{**s, "weight": s["weight"] / total_weight} for s in self._scorers]
+
+    def _build_environment_params(self) -> ScenarioEnvironmentParam:
+        """Build environment parameters"""
+        env_params: ScenarioEnvironmentParam = {}
+        if self._blueprint:
+            env_params["blueprint_id"] = self._blueprint.id
+        if self._snapshot:
+            env_params["snapshot_id"] = self._snapshot.id
+        if self._working_directory:
+            env_params["working_directory"] = self._working_directory
+        return env_params
+
     def _build_params(self) -> ScenarioCreateParams:
         """Build the scenario creation parameters.
 
@@ -405,51 +421,50 @@ def _build_params(self) -> ScenarioCreateParams:
                 "Call add_test_command_scorer(), add_bash_script_scorer(), or another scorer method first."
             )
 
-        # Normalize weights to sum to 1.0
-        total_weight = sum(s["weight"] for s in self._scorers)
-        normalized_scorers: List[ScoringFunctionParam] = [
-            {**s, "weight": s["weight"] / total_weight} for s in self._scorers
-        ]
-
-        params: ScenarioCreateParams = {
+        return {
             "name": self._name,
             "input_context": {
                 "problem_statement": self._problem_statement,
+                "additional_context": self._additional_context,
             },
             "scoring_contract": {
-                "scoring_function_parameters": normalized_scorers,
+                "scoring_function_parameters": self._build_normalized_scorers(),
             },
+            "environment_parameters": self._build_environment_params(),
+            "metadata": self._metadata,
+            "reference_output": self._reference_output,
+            "required_environment_variables": self._required_env_vars,
+            "required_secret_names": self._required_secrets,
+            "validation_type": self._validation_type,
         }
 
-        # Add additional context if set
-        if self._additional_context is not None:
-            params["input_context"]["additional_context"] = self._additional_context
+    def preview(self) -> ScenarioPreview:
+        """Preview the scenario configuration without pushing to the platform.
 
-        # Build environment parameters if any are set
-        env_params: ScenarioEnvironmentParam = {}
-        if self._blueprint:
-            env_params["blueprint_id"] = self._blueprint.id
-        if self._snapshot:
-            env_params["snapshot_id"] = self._snapshot.id
-        if self._working_directory:
-            env_params["working_directory"] = self._working_directory
+        Returns the current configuration state as a ScenarioPreview object.
+        Does not validate or raise errors for missing required fields.
 
-        if env_params:
-            params["environment_parameters"] = env_params
-
-        # Add optional fields
-        if self._metadata:
-            params["metadata"] = self._metadata
-        if self._reference_output:
-            params["reference_output"] = self._reference_output
-        if self._required_env_vars:
-            params["required_environment_variables"] = self._required_env_vars
-        if self._required_secrets:
-            params["required_secret_names"] = self._required_secrets
-        if self._validation_type:
-            params["validation_type"] = self._validation_type
-
-        return params
+        :return: Preview of the scenario configuration
+        :rtype: ScenarioPreview
+        """
+        return ScenarioPreview.model_validate(
+            {
+                "name": self._name,
+                "input_context": {
+                    "problem_statement": self._problem_statement,
+                    "additional_context": self._additional_context,
+                },
+                "scoring_contract": {
+                    "scoring_function_parameters": self._build_normalized_scorers(),
+                },
+                "environment": self._build_environment_params(),
+                "metadata": self._metadata,
+                "reference_output": self._reference_output,
+                "required_environment_variables": self._required_env_vars,
+                "required_secret_names": self._required_secrets,
+                "validation_type": self._validation_type,
+            }
+        )
 
     def push(self, **options: Unpack[LongRequestOptions]) -> Scenario:
         """Create the scenario on the platform.