feat(sql-optim-env): add execution-grounded DuckDB SQL query optimization environment

docs/source/_toctree.yml

@@ -131,6 +131,8 @@
     title: OpenCode
   - local: environments/sophistry_bench_sprint
     title: Sophistry Bench Sprint
+  - local: environments/sql_optim
+    title: SQL Query Optimization
   title: Environments
 - isExpanded: false
   sections:

docs/source/environments.md

@@ -265,6 +265,13 @@ The OpenEnv community has built a catalog of ready-to-run environments that cove
         <a href="environments/sophistry_bench_sprint" class="!no-underline border dark:border-gray-700 px-3 py-1 rounded text-sm hover:shadow">📄 Docs</a>
       </div>
     </div>
+    <div class="border dark:border-gray-700 p-5 rounded-lg shadow">
+      <div class="font-bold mb-2">SQL Query Optimization</div>
+      <p class="text-sm"><code>sql_optim_env</code> rewards an agent for rewriting slow SQL: it executes both the original and optimized query against an in-memory DuckDB seeded with synthetic data and scores measured speedup plus result-correctness across five anti-pattern tasks.</p>
+      <div class="flex gap-2 mt-3">
+        <a href="environments/sql_optim" class="!no-underline border dark:border-gray-700 px-3 py-1 rounded text-sm hover:shadow">📄 Docs</a>
+      </div>
+    </div>
   </div>
 </div>
 

docs/source/environments/sql_optim.md

@@ -0,0 +1,95 @@
+<!-- openenv-source: sql_optim_env -->
+# SQL Query Optimization Environment
+
+An **execution-grounded** environment for training agents to write fast SQL. The
+agent is given a slow query plus its schema and must return a rewrite. Reward is
+**not** a keyword heuristic: the environment runs both the original and the
+optimized query against a real in-memory **DuckDB** database seeded with
+realistic synthetic data (10k users, 500k orders, 1k products, 1M events), then
+scores the agent on measured speedup and result-correctness.
+
+Five tasks scale from basic anti-patterns to expert window-function audits.
+
+## Quick Start
+
+```python
+from sql_optim_env import SQLOptimAction, SQLOptimEnv
+
+with SQLOptimEnv(base_url="http://localhost:8000").sync() as env:
+    obs = env.reset(task_id="task_1_basic_antipatterns").observation
+    print(obs.sql_query)          # the slow query to fix
+    print(obs.schema_info)        # tables, row counts
+
+    result = env.step(
+        SQLOptimAction(
+            suggestions=[
+                {"issue_type": "select_star", "line": 1,
+                 "description": "SELECT * reads every column", "severity": "high",
+                 "fix": "project only the needed columns"},
+            ],
+            optimized_query="SELECT id FROM orders WHERE customer_id = 5000",
+            summary="Dropped SELECT * and the non-sargable CAST.",
+            estimated_improvement="~10x faster",
+            approved=False,
+        )
+    )
+    print(result.reward, result.done)
+    print(result.metadata["reward_breakdown"])
+    print(result.metadata["execution"]["speedup"])   # real DuckDB speedup
+```
+
+Async usage mirrors the other environments:
+
+```python
+async with SQLOptimEnv(base_url="http://localhost:8000") as env:
+    result = await env.reset(task_id="task_1_basic_antipatterns")
+```
+
+## Action
+
+| Field | Type | Meaning |
+|-------|------|---------|
+| `suggestions` | `list[dict]` | Issues found: `issue_type`, `line`, `description`, `severity`, `fix` |
+| `optimized_query` | `str` | The rewritten SQL — executed for real |
+| `summary` | `str` | Overall analysis |
+| `estimated_improvement` | `str` | Agent's own speedup estimate |
+| `approved` | `bool` | `True` if the query is already optimal |
+
+## Observation
+
+`sql_query`, `schema_info`, `task_description`, `difficulty`, `step_count`,
+`max_steps`, `issues_found_so_far`, and `last_execution` (the previous step's
+real timing comparison, so the agent can refine its rewrite). The step reward is
+on `reward`; the composite `reward_breakdown`, `feedback`, and full `execution`
+comparison are under `metadata`.
+
+## Reward
+
+Composite score in `[0, 1]`, computed inside the environment:
+
+| Component | Weight | Source |
+|-----------|--------|--------|
+| Execution speedup | 35% | real DuckDB timing ratio |
+| Result correctness | 20% | do both queries return identical rows? |
+| Issue detection | 25% | flagged issues vs. ground truth |
+| Approval correctness | 8% | correctly judged the query bad/good |
+| Summary quality | 7% | thoroughness of the written analysis |
+| Severity labels | 5% | severities present |
+
+## Tasks
+
+Five tasks (`task_1_basic_antipatterns` … `task_5_*`) covering `SELECT *`,
+non-sargable predicates, functions on filter columns, join ordering, and window
+functions, spanning `easy` → `expert`.
+
+## Run the server locally
+
+```bash
+uvicorn server.app:app --host 0.0.0.0 --port 8000
+```
+
+## Tests
+
+```bash
+PYTHONPATH=src:envs uv run pytest tests/envs/test_sql_optim_environment.py -v
+```

envs/sql_optim_env/README.md

@@ -0,0 +1,110 @@
+---
+title: SQL Query Optimization Environment
+emoji: 🗄️
+colorFrom: indigo
+colorTo: blue
+sdk: docker
+pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
+  - sql
+  - duckdb
+  - optimization
+---
+
+# SQL Query Optimization Environment
+
+An **execution-grounded** environment for training agents to write fast SQL. The
+agent is given a slow query plus its schema and must return a rewrite. Reward is
+**not** a keyword heuristic: the environment runs both the original and the
+optimized query against a real in-memory **DuckDB** database seeded with
+realistic synthetic data (10k users, 500k orders, 1k products, 1M events), then
+scores the agent on measured speedup and result-correctness.
+
+Five tasks scale from basic anti-patterns to expert window-function audits.
+
+## Quick Start
+
+```python
+from sql_optim_env import SQLOptimAction, SQLOptimEnv
+
+with SQLOptimEnv(base_url="http://localhost:8000").sync() as env:
+    obs = env.reset(task_id="task_1_basic_antipatterns").observation
+    print(obs.sql_query)          # the slow query to fix
+    print(obs.schema_info)        # tables, row counts
+
+    result = env.step(
+        SQLOptimAction(
+            suggestions=[
+                {"issue_type": "select_star", "line": 1,
+                 "description": "SELECT * reads every column", "severity": "high",
+                 "fix": "project only the needed columns"},
+            ],
+            optimized_query="SELECT id FROM orders WHERE customer_id = 5000",
+            summary="Dropped SELECT * and the non-sargable CAST.",
+            estimated_improvement="~10x faster",
+            approved=False,
+        )
+    )
+    print(result.reward, result.done)
+    print(result.metadata["reward_breakdown"])
+    print(result.metadata["execution"]["speedup"])   # real DuckDB speedup
+```
+
+Async usage mirrors the other environments:
+
+```python
+async with SQLOptimEnv(base_url="http://localhost:8000") as env:
+    result = await env.reset(task_id="task_1_basic_antipatterns")
+```
+
+## Action
+
+| Field | Type | Meaning |
+|-------|------|---------|
+| `suggestions` | `list[dict]` | Issues found: `issue_type`, `line`, `description`, `severity`, `fix` |
+| `optimized_query` | `str` | The rewritten SQL — executed for real |
+| `summary` | `str` | Overall analysis |
+| `estimated_improvement` | `str` | Agent's own speedup estimate |
+| `approved` | `bool` | `True` if the query is already optimal |
+
+## Observation
+
+`sql_query`, `schema_info`, `task_description`, `difficulty`, `step_count`,
+`max_steps`, `issues_found_so_far`, and `last_execution` (the previous step's
+real timing comparison, so the agent can refine its rewrite). The step reward is
+on `reward`; the composite `reward_breakdown`, `feedback`, and full `execution`
+comparison are under `metadata`.
+
+## Reward
+
+Composite score in `[0, 1]`, computed inside the environment:
+
+| Component | Weight | Source |
+|-----------|--------|--------|
+| Execution speedup | 35% | real DuckDB timing ratio |
+| Result correctness | 20% | do both queries return identical rows? |
+| Issue detection | 25% | flagged issues vs. ground truth |
+| Approval correctness | 8% | correctly judged the query bad/good |
+| Summary quality | 7% | thoroughness of the written analysis |
+| Severity labels | 5% | severities present |
+
+## Tasks
+
+Five tasks (`task_1_basic_antipatterns` … `task_5_*`) covering `SELECT *`,
+non-sargable predicates, functions on filter columns, join ordering, and window
+functions, spanning `easy` → `expert`.
+
+## Run the server locally
+
+```bash
+uvicorn server.app:app --host 0.0.0.0 --port 8000
+```
+
+## Tests
+
+```bash
+PYTHONPATH=src:envs uv run pytest tests/envs/test_sql_optim_environment.py -v
+```

envs/sql_optim_env/__init__.py

@@ -0,0 +1,34 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+SQL Query Optimization Environment for OpenEnv.
+
+An execution-grounded RL environment: the agent rewrites slow SQL and is
+rewarded by real DuckDB execution speedup plus result-correctness across five
+tasks that scale from basic anti-patterns to expert window-function audits.
+
+Examples:
+
+```python
+from envs.sql_optim_env import SQLOptimEnv, SQLOptimAction
+
+with SQLOptimEnv(base_url="http://localhost:8000").sync() as env:
+    obs = env.reset(task_id="task_1_basic_antipatterns").observation
+    result = env.step(SQLOptimAction(optimized_query="SELECT id FROM orders ..."))
+    print(result.reward, result.done)
+```
+"""
+
+from .client import SQLOptimEnv
+from .models import SQLOptimAction, SQLOptimObservation, SQLOptimState
+
+__all__ = [
+    "SQLOptimEnv",
+    "SQLOptimAction",
+    "SQLOptimObservation",
+    "SQLOptimState",
+]

envs/sql_optim_env/client.py

@@ -0,0 +1,72 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Client for the SQL Query Optimization environment.
+
+Maintains a persistent WebSocket connection to the environment server for
+efficient multi-step episodes. Async by default; use `.sync()` for a
+synchronous wrapper.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from openenv.core.client_types import StepResult
+from openenv.core.env_client import EnvClient
+
+from .models import SQLOptimAction, SQLOptimObservation, SQLOptimState
+
+
+class SQLOptimEnv(EnvClient[SQLOptimAction, SQLOptimObservation, SQLOptimState]):
+    """
+    Client for the SQL Query Optimization environment.
+
+    The agent receives a slow SQL query plus its schema, returns a
+    [`SQLOptimAction`][envs.sql_optim_env.models.SQLOptimAction] with a
+    rewritten `optimized_query`, and is rewarded by real DuckDB execution
+    speedup and result-correctness.
+
+    Examples:
+
+    ```python
+    with SQLOptimEnv(base_url="http://localhost:8000").sync() as env:
+        obs = env.reset(task_id="task_1_basic_antipatterns").observation
+        result = env.step(
+            SQLOptimAction(
+                optimized_query="SELECT id FROM orders WHERE customer_id = 5000",
+                summary="Dropped SELECT * and the non-sargable CAST.",
+            )
+        )
+        print(result.reward, result.done)
+    ```
+    """
+
+    def _step_payload(self, action: SQLOptimAction) -> Dict[str, Any]:
+        """Serialize a [`SQLOptimAction`] into the step request payload."""
+        return {
+            "suggestions": action.suggestions,
+            "optimized_query": action.optimized_query,
+            "summary": action.summary,
+            "estimated_improvement": action.estimated_improvement,
+            "approved": action.approved,
+        }
+
+    def _parse_result(self, payload: Dict[str, Any]) -> StepResult[SQLOptimObservation]:
+        """Parse a server step/reset response into a `StepResult`."""
+        obs_data = payload.get("observation", {})
+        observation = SQLOptimObservation(**obs_data)
+        return StepResult(
+            observation=observation,
+            reward=payload.get("reward", observation.reward),
+            done=payload.get("done", observation.done),
+            metadata=payload.get("metadata", observation.metadata),
+        )
+
+    def _parse_state(self, payload: Dict[str, Any]) -> SQLOptimState:
+        """Parse a server state response into a [`SQLOptimState`]."""
+        return SQLOptimState(**payload)