.

README.md

@@ -43,3 +43,13 @@ openenv validate http://127.0.0.1:8000
 - `task_buffer`: pending tasks with estimated duration and hidden complexity
 - `flow_score`, `social_debt`, `calendar_churn`: core scoring metrics
 - `current_slot`, `current_time`, `recovery_state`, `mute_comms`: live execution state
+
+## Built-in benchmark tasks
+
+Set `TASK_NAME` to select a deterministic scenario before reset. Available tasks:
+
+- `quiet-morning`: high-noise start where muting comms early and protecting focus is rewarded
+- `meeting-surgery`: fragmented calendar where selective meeting moves should improve flow
+- `delivery-triage`: constrained delivery day with hidden task complexity and tighter tradeoffs
+
+Each task has a grader in [benchmark_tasks.py](/C:/Users/arshi/OneDrive/Desktop/idk/engineer-manager/benchmark_tasks.py:1). The environment also exposes task metadata and the current grader score in `observation.metadata.episode_metrics.grader_score`.

benchmark_tasks.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from focus_resource_env import DEEP_WORK, EMPTY, MEETING, FocusResourceEnv, Task
+
+
+StepRecord = dict[str, Any]
+TaskSetup = Callable[[FocusResourceEnv], None]
+TaskGrader = Callable[[list[StepRecord]], float]
+
+
+def _reset_state(env: FocusResourceEnv) -> None:
+    env.timeline[:] = EMPTY
+    env.meeting_meta = {}
+    env.task_buffer = []
+    env.current_slot = 0
+    env.current_work_streak_slots = 0
+    env.recovery_remaining = 0
+    env.mute_comms = False
+    env.social_debt = 0.0
+    env.calendar_churn = 0
+    env.flow_score = 0.0
+    env.last_executed_kind = EMPTY
+    env.interruptions = 0
+    env.invalid_actions = 0
+
+
+def _set_meeting(
+    env: FocusResourceEnv,
+    *,
+    start: int,
+    length: int,
+    priority: int,
+    meeting_id: int,
+) -> None:
+    env._place_meeting(start, length, priority, meeting_id)
+
+
+def _normalized_total_score(env: FocusResourceEnv) -> float:
+    max_score = max(1.0, (env.timeline_length * 0.5) ** 2)
+    return min(1.0, max(0.0, env._total_score() / max_score))
+
+
+def setup_quiet_morning(env: FocusResourceEnv) -> None:
+    _reset_state(env)
+    env.distraction_risk = 0.65
+    env.task_buffer = [
+        Task(duration=2, hidden_complexity=1.0),
+        Task(duration=3, hidden_complexity=1.0),
+        Task(duration=2, hidden_complexity=1.25),
+    ]
+    _set_meeting(env, start=5, length=1, priority=4, meeting_id=1)
+    _set_meeting(env, start=7, length=1, priority=3, meeting_id=2)
+
+
+def setup_meeting_surgery(env: FocusResourceEnv) -> None:
+    _reset_state(env)
+    env.distraction_risk = 0.10
+    env.task_buffer = [
+        Task(duration=2, hidden_complexity=1.0),
+        Task(duration=2, hidden_complexity=1.25),
+        Task(duration=1, hidden_complexity=1.0),
+    ]
+    _set_meeting(env, start=1, length=1, priority=2, meeting_id=1)
+    _set_meeting(env, start=3, length=1, priority=2, meeting_id=2)
+    _set_meeting(env, start=6, length=2, priority=8, meeting_id=3)
+
+
+def setup_delivery_triage(env: FocusResourceEnv) -> None:
+    _reset_state(env)
+    env.distraction_risk = 0.25
+    env.task_buffer = [
+        Task(duration=3, hidden_complexity=1.5),
+        Task(duration=2, hidden_complexity=1.0),
+        Task(duration=1, hidden_complexity=1.0),
+    ]
+    _set_meeting(env, start=4, length=1, priority=9, meeting_id=1)
+    _set_meeting(env, start=8, length=2, priority=7, meeting_id=2)
+
+
+def grade_quiet_morning(trajectory: list[StepRecord]) -> float:
+    if not trajectory:
+        return 0.0
+    first_action = int(trajectory[0]["action"]["operation"])
+    final = trajectory[-1]["observation"]
+    final_score = float(final["flow_score"])
+    transition_count = sum(1 for step in trajectory if step["info"]["transition_info"]["interrupted"])
+    scheduled = sum(1 for slot in final["timeline"] if int(slot) == DEEP_WORK)
+
+    score = 0.0
+    score += 0.25 if first_action == 3 else 0.0
+    score += min(0.45, final_score / 6.0)
+    score += 0.15 if transition_count == 0 else 0.0
+    score += min(0.15, scheduled / 6.0)
+    return min(1.0, round(score, 4))
+
+
+def grade_meeting_surgery(trajectory: list[StepRecord]) -> float:
+    if not trajectory:
+        return 0.0
+    final = trajectory[-1]["observation"]
+    flow = float(final["flow_score"])
+    debt = float(final["social_debt"])
+    churn = int(final["calendar_churn"])
+    reschedules = sum(
+        1
+        for step in trajectory
+        if step["info"].get("action_info", {}).get("status") == "meeting_rescheduled"
+    )
+
+    score = 0.0
+    score += min(0.40, flow / 5.0)
+    score += 0.20 if reschedules >= 1 else 0.0
+    score += 0.20 if 1 <= churn <= 2 else max(0.0, 0.20 - (0.10 * abs(churn - 1)))
+    score += max(0.0, 0.20 - (debt / 8.0))
+    return min(1.0, round(score, 4))
+
+
+def grade_delivery_triage(trajectory: list[StepRecord]) -> float:
+    if not trajectory:
+        return 0.0
+    final = trajectory[-1]["observation"]
+    total = float(final["flow_score"]) - float(final["social_debt"]) - float(final["calendar_churn"])
+    invalid_actions = sum(
+        1
+        for step in trajectory
+        if str(step["info"].get("action_info", {}).get("status", "")).startswith("invalid")
+    )
+    remaining_tasks = len(final["task_buffer"])
+    scheduled = sum(1 for slot in final["timeline"] if int(slot) == DEEP_WORK)
+
+    score = 0.0
+    score += min(0.45, max(0.0, total) / 6.0)
+    score += min(0.25, scheduled / 8.0)
+    score += 0.20 if remaining_tasks <= 1 else 0.10 if remaining_tasks == 2 else 0.0
+    score += max(0.0, 0.10 - (0.05 * invalid_actions))
+    return min(1.0, round(score, 4))
+
+
+@dataclass(frozen=True)
+class TaskSpec:
+    name: str
+    description: str
+    setup: TaskSetup
+    grader: TaskGrader
+
+
+TASK_SPECS: dict[str, TaskSpec] = {
+    "quiet-morning": TaskSpec(
+        name="quiet-morning",
+        description="High-noise morning where the agent should mute comms early and protect an uninterrupted work block.",
+        setup=setup_quiet_morning,
+        grader=grade_quiet_morning,
+    ),
+    "meeting-surgery": TaskSpec(
+        name="meeting-surgery",
+        description="A fragmented calendar where the agent should improve flow with limited, selective meeting moves.",
+        setup=setup_meeting_surgery,
+        grader=grade_meeting_surgery,
+    ),
+    "delivery-triage": TaskSpec(
+        name="delivery-triage",
+        description="A constrained day with hidden task complexity where the agent must schedule useful work without spiraling debt.",
+        setup=setup_delivery_triage,
+        grader=grade_delivery_triage,
+    ),
+}
+
+
+DEFAULT_TASK_NAME = "quiet-morning"
+
+
+def get_task_spec(task_name: str | None) -> TaskSpec:
+    normalized = (task_name or os.getenv("TASK_NAME") or DEFAULT_TASK_NAME).strip()
+    return TASK_SPECS.get(normalized, TASK_SPECS[DEFAULT_TASK_NAME])
+
+
+def apply_task(env: FocusResourceEnv, task_name: str | None) -> TaskSpec:
+    spec = get_task_spec(task_name)
+    spec.setup(env)
+    return spec
+
+
+def grade_trajectory(task_name: str, trajectory: list[StepRecord]) -> float:
+    spec = get_task_spec(task_name)
+    return spec.grader(trajectory)

graders.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import Any
+
+from benchmark_tasks import grade_trajectory
+
+
+def _coerce_trajectory(payload: Any) -> list[dict[str, Any]]:
+    if isinstance(payload, list):
+        return [dict(step) for step in payload]
+    if isinstance(payload, dict):
+        if isinstance(payload.get("trajectory"), list):
+            return [dict(step) for step in payload["trajectory"]]
+        if isinstance(payload.get("steps"), list):
+            return [dict(step) for step in payload["steps"]]
+    return []
+
+
+def _grade(task_name: str, payload: Any) -> dict[str, Any]:
+    trajectory = _coerce_trajectory(payload)
+    score = grade_trajectory(task_name, trajectory)
+    return {
+        "task_name": task_name,
+        "score": score,
+        "passed": score > 0.0,
+        "reward": score,
+    }
+
+
+def grade_task_0(payload: Any) -> dict[str, Any]:
+    return _grade("quiet-morning", payload)
+
+
+def grade_task_1(payload: Any) -> dict[str, Any]:
+    return _grade("meeting-surgery", payload)
+
+
+def grade_task_2(payload: Any) -> dict[str, Any]:
+    return _grade("delivery-triage", payload)

openenv.yaml

@@ -4,3 +4,79 @@ type: space
 runtime: fastapi
 app: server.app:app
 port: 8000
+version: 0.1.0
+entry_point: server.engineer_manager_environment:EngineerManagerEnvironment
+api:
+  base_url: /
+  endpoints:
+    reset:
+      method: POST
+      path: /reset
+    step:
+      method: POST
+      path: /step
+    state:
+      method: GET
+      path: /state
+    tasks:
+      method: GET
+      path: /tasks
+    grader:
+      method: POST
+      path: /grader
+tasks:
+  - id: engineer_manager_task_0
+    task_id: quiet-morning
+    name: quiet-morning
+    difficulty: easy
+    description: High-noise morning where the agent should mute comms early and protect an uninterrupted work block.
+    max_steps: 32
+    reset_params:
+      task_name: quiet-morning
+    action_schema:
+      target_slot: integer slot index within the workday
+      operation: 0=idle, 1=schedule work, 2=reschedule meeting, 3=toggle mute comms
+    grader: graders:grade_task_0
+    graders:
+      - graders:grade_task_0
+    reward_range:
+      - 0.0
+      - 1.0
+  - id: engineer_manager_task_1
+    task_id: meeting-surgery
+    name: meeting-surgery
+    difficulty: medium
+    description: Fragmented calendar where selective meeting moves should improve flow.
+    max_steps: 32
+    reset_params:
+      task_name: meeting-surgery
+    action_schema:
+      target_slot: integer slot index within the workday
+      operation: 0=idle, 1=schedule work, 2=reschedule meeting, 3=toggle mute comms
+    grader: graders:grade_task_1
+    graders:
+      - graders:grade_task_1
+    reward_range:
+      - 0.0
+      - 1.0
+  - id: engineer_manager_task_2
+    task_id: delivery-triage
+    name: delivery-triage
+    difficulty: hard
+    description: Constrained delivery day with hidden task complexity and tighter tradeoffs.
+    max_steps: 32
+    reset_params:
+      task_name: delivery-triage
+    action_schema:
+      target_slot: integer slot index within the workday
+      operation: 0=idle, 1=schedule work, 2=reschedule meeting, 3=toggle mute comms
+    grader: graders:grade_task_2
+    graders:
+      - graders:grade_task_2
+    reward_range:
+      - 0.0
+      - 1.0
+graders:
+  - graders:grade_task_0
+  - graders:grade_task_1
+  - graders:grade_task_2

server/app.py

@@ -7,6 +7,10 @@ from textwrap import dedent
 import uvicorn
 from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse, Response
 from openenv.core.env_server.http_server import create_fastapi_app
+from pydantic import BaseModel
+
+from graders import grade_task_0, grade_task_1, grade_task_2
+from tasks import TASKS
 
 try:
     from ..models import EngineerManagerAction, EngineerManagerObservation
@@ -29,6 +33,11 @@ app = create_fastapi_app(
     max_concurrent_envs=2,
 )
 
+
+class GraderRequest(BaseModel):
+    task_id: str
+    trajectory: list[dict]
+
 WEB_CSS = dedent(
     """\
     :root {
@@ -453,6 +462,27 @@ def manifest() -> JSONResponse:
     )
 
 
+@app.get("/tasks", include_in_schema=False)
+def tasks() -> JSONResponse:
+    return JSONResponse({"tasks": TASKS})
+
+
+@app.post("/grader", include_in_schema=False)
+def grader(request: GraderRequest) -> JSONResponse:
+    graders = {
+        "quiet-morning": grade_task_0,
+        "meeting-surgery": grade_task_1,
+        "delivery-triage": grade_task_2,
+    }
+    grader_fn = graders.get(request.task_id)
+    if grader_fn is None:
+        return JSONResponse(
+            {"error": f"Unknown task_id: {request.task_id}", "score": 0.0, "passed": False},
+            status_code=400,
+        )
+    return JSONResponse(grader_fn({"trajectory": request.trajectory}))
+
+
 def run(host: str = "0.0.0.0", port: int = 8000) -> None:
     """Run the OpenEnv HTTP server."""
     uvicorn.run(app, host=host, port=port)

server/engineer_manager_environment.py

@@ -3,10 +3,12 @@
 from __future__ import annotations
 
 from uuid import uuid4
+import os
 
 from openenv.core.env_server.interfaces import Environment, EnvironmentMetadata
 from openenv.core.env_server.types import State
 
+from benchmark_tasks import TASK_SPECS, apply_task, grade_trajectory
 from focus_resource_env import FocusResourceEnv
 
 try:
@@ -28,14 +30,17 @@ class EngineerManagerEnvironment(
         end_hour: str = "17:00",
         distraction_risk: float = 0.15,
         seed: int | None = 7,
+        task_name: str | None = None,
     ) -> None:
         super().__init__()
         self._start_hour = start_hour
         self._end_hour = end_hour
         self._distraction_risk = distraction_risk
         self._seed = seed
+        self._task_name = task_name or os.getenv("TASK_NAME")
         self._step_count = 0
         self._episode_id = str(uuid4())
+        self._trajectory: list[dict[str, object]] = []
         self._env = FocusResourceEnv(
             start_hour=start_hour,
             end_hour=end_hour,
@@ -47,18 +52,23 @@ class EngineerManagerEnvironment(
         self,
         seed: int | None = None,
         episode_id: str | None = None,
+        task_name: str | None = None,
         **_: object,
     ) -> EngineerManagerObservation:
         self._seed = self._seed if seed is None else seed
+        self._task_name = task_name or self._task_name or os.getenv("TASK_NAME")
         self._episode_id = episode_id or str(uuid4())
         self._step_count = 0
+        self._trajectory = []
         self._env = FocusResourceEnv(
             start_hour=self._start_hour,
             end_hour=self._end_hour,
             distraction_risk=self._distraction_risk,
             seed=self._seed,
         )
-        return self._to_observation(self._env.reset(), reward=0.0, done=False)
+        self._env.reset()
+        apply_task(self._env, self._task_name)
+        return self._to_observation(self._env._observation(), reward=0.0, done=False)
 
     def step(
         self,
@@ -71,6 +81,15 @@ class EngineerManagerEnvironment(
             (action.target_slot, action.operation)
         )
         self._step_count += 1
+        self._trajectory.append(
+            {
+                "action": {"target_slot": int(action.target_slot), "operation": int(action.operation)},
+                "observation": observation,
+                "reward": float(reward),
+                "done": bool(done),
+                "info": info,
+            }
+        )
         return self._to_observation(observation, reward=reward, done=done, info=info)
 
     @property
@@ -87,7 +106,8 @@ class EngineerManagerEnvironment(
             name="Engineer Manager",
             description=(
                 "Manage a workday by scheduling deep work, rescheduling meetings, "
-                "and controlling communication noise."
+                "and controlling communication noise. "
+                f"Available tasks: {', '.join(sorted(TASK_SPECS))}."
             ),
             version="0.1.0",
         )
@@ -103,5 +123,20 @@ class EngineerManagerEnvironment(
         payload = dict(observation)
         payload["reward"] = reward
         payload["done"] = done
-        payload["metadata"] = info or {}
+        metadata = dict(info or {})
+        metadata["task_name"] = self._task_name
+        metadata["episode_metrics"] = {
+            "interruptions": int(self._env.interruptions),
+            "invalid_actions": int(self._env.invalid_actions),
+            "remaining_tasks": len(self._env.task_buffer),
+            "scheduled_work_slots": sum(1 for slot in self._env.timeline if int(slot) == 1),
+            "successful_reschedules": sum(
+                1
+                for step in self._trajectory
+                if step["info"].get("action_info", {}).get("status") == "meeting_rescheduled"
+            ),
+            "total_score": float(self._env._total_score()),
+            "grader_score": grade_trajectory(self._task_name or "", self._trajectory) if self._trajectory else 0.0,
+        }
+        payload["metadata"] = metadata
         return EngineerManagerObservation.model_validate(payload)

tasks.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from benchmark_tasks import TASK_SPECS
+
+
+TASKS = [
+    {
+        "id": "engineer_manager_task_0",
+        "task_id": "quiet-morning",
+        "name": "quiet-morning",
+        "difficulty": "easy",
+        "description": TASK_SPECS["quiet-morning"].description,
+        "max_steps": 32,
+        "reset_params": {"task_name": "quiet-morning"},
+        "action_schema": {
+            "target_slot": "integer slot index within the workday",
+            "operation": "0=idle, 1=schedule work, 2=reschedule meeting, 3=toggle mute comms",
+        },
+        "grader": "graders:grade_task_0",
+        "graders": ["graders:grade_task_0"],
+        "reward_range": [0.0, 1.0],
+    },
+    {
+        "id": "engineer_manager_task_1",
+        "task_id": "meeting-surgery",
+        "name": "meeting-surgery",
+        "difficulty": "medium",
+        "description": TASK_SPECS["meeting-surgery"].description,
+        "max_steps": 32,
+        "reset_params": {"task_name": "meeting-surgery"},
+        "action_schema": {
+            "target_slot": "integer slot index within the workday",
+            "operation": "0=idle, 1=schedule work, 2=reschedule meeting, 3=toggle mute comms",
+        },
+        "grader": "graders:grade_task_1",
+        "graders": ["graders:grade_task_1"],
+        "reward_range": [0.0, 1.0],
+    },
+    {
+        "id": "engineer_manager_task_2",
+        "task_id": "delivery-triage",
+        "name": "delivery-triage",
+        "difficulty": "hard",
+        "description": TASK_SPECS["delivery-triage"].description,
+        "max_steps": 32,
+        "reset_params": {"task_name": "delivery-triage"},
+        "action_schema": {
+            "target_slot": "integer slot index within the workday",
+            "operation": "0=idle, 1=schedule work, 2=reschedule meeting, 3=toggle mute comms",
+        },
+        "grader": "graders:grade_task_2",
+        "graders": ["graders:grade_task_2"],
+        "reward_range": [0.0, 1.0],
+    },
+]
+
+
+TASK_ID_TO_INDEX = {task["task_id"]: index for index, task in enumerate(TASKS)}
+
+
+TASK_GRADER_PAIRS = [
+    ("engineer_manager_task_0", "graders:grade_task_0"),
+    ("engineer_manager_task_1", "graders:grade_task_1"),
+    ("engineer_manager_task_2", "graders:grade_task_2"),
+]
+
+
+__all__ = ["TASKS", "TASK_ID_TO_INDEX", "TASK_GRADER_PAIRS"]