feat: complete Phase 3 — inference, expert task, deployment

- Add inference.py with async OrchestratorClient, OpenAI client,
Qwen3 /no_think support, 4-level JSON parser, .env loading
- Add expert bonus task (Life OS Daily Orchestration): 14 subtasks,
8 agents, 2 permanent failure traps, 10-dimension grader
- Add 16 hard task integration tests (walkthrough + edge cases)
- Rewrite README.md with full documentation
- Add root-level requirements.txt
- Populate baseline_scores.json (easy=0.9, medium=0.633, hard=0.808, expert=0.802)
- Fix client.py import fallback for bare module usage
- Deploy to HF Spaces: kartikmandar-workflow-orchestrator.hf.space
- All 5 submission gates pass, 139 tests passing

README.md

@@ -13,243 +13,163 @@ tags:
 
 # Workflow Orchestrator Environment
 
-A simple test environment that echoes back messages. Perfect for testing the env APIs as well as demonstrating environment usage patterns.
-
-## Quick Start
-
-The simplest way to use the Workflow Orchestrator environment is through the `WorkflowOrchestratorEnv` class:
-
-```python
-from workflow_orchestrator import WorkflowOrchestratorAction, WorkflowOrchestratorEnv
-
-try:
-    # Create environment from Docker image
-    workflow_orchestratorenv = WorkflowOrchestratorEnv.from_docker_image("workflow_orchestrator-env:latest")
-
-    # Reset
-    result = workflow_orchestratorenv.reset()
-    print(f"Reset: {result.observation.echoed_message}")
-
-    # Send multiple messages
-    messages = ["Hello, World!", "Testing echo", "Final message"]
-
-    for msg in messages:
-        result = workflow_orchestratorenv.step(WorkflowOrchestratorAction(message=msg))
-        print(f"Sent: '{msg}'")
-        print(f"  → Echoed: '{result.observation.echoed_message}'")
-        print(f"  → Length: {result.observation.message_length}")
-        print(f"  → Reward: {result.reward}")
-
-finally:
-    # Always clean up
-    workflow_orchestratorenv.close()
-```
-
-That's it! The `WorkflowOrchestratorEnv.from_docker_image()` method handles:
-- Starting the Docker container
-- Waiting for the server to be ready
-- Connecting to the environment
-- Container cleanup when you call `close()`
-
-## Building the Docker Image
-
-Before using the environment, you need to build the Docker image:
+An OpenEnv environment where an LLM agent acts as a **project coordinator**, managing DAG-based workflows of subtasks across simulated specialist agents with varying capabilities, failure rates, and cost profiles. Tests coordination, delegation, parallelism, failure recovery, and cost management.
+
+## Motivation
+
+Agent orchestration is the #1 enterprise AI trend — yet LLMs are terrible at it. Research documents 14+ failure modes in multi-agent systems (MAST taxonomy), up to 17x error amplification in unstructured networks (Spark to Fire), and only 25% baseline correctness with GPT-4o (ChatDev). This environment provides a controlled, deterministic testbed for training and evaluating LLM orchestration capabilities across three real-world scenarios: software development, CI/CD deployment, and production incident response.
+
+## Action Space
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `action_type` | `"delegate"\|"retry"\|"wait"\|"synthesize"\|"abort"` | Which action to take |
+| `subtask_id` | `Optional[str]` | Target subtask (required for delegate/retry/abort) |
+| `agent_name` | `Optional[str]` | Agent to assign (required for delegate/retry) |
+
+**Actions:**
+- **delegate**: Assign a ready subtask to an idle, capable agent
+- **retry**: Re-assign a failed subtask (same or different agent)
+- **wait**: Advance time by 1 step; working agents tick
+- **synthesize**: Combine all completed outputs (only valid when all subtasks done)
+- **abort**: Permanently fail a non-completed subtask
+
+Invalid actions are accepted but penalized — the step is consumed, a penalty is applied, and state remains unchanged.
+
+## Observation Space
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `task_description` | `str` | Natural-language task objective |
+| `subtasks` | `list[SubtaskInfo]` | Status of each subtask in the DAG |
+| `agents` | `list[AgentInfo]` | Status of each simulated agent |
+| `completed_outputs` | `dict[str, str]` | Outputs from finished subtasks |
+| `errors` | `list[str]` | Errors from the current step |
+| `time_remaining` | `int` | Steps left before timeout |
+| `time_elapsed` | `int` | Steps taken so far |
+| `capacity_limit` | `int` | Max concurrent in-progress tasks |
+| `active_task_count` | `int` | Currently in-progress task count |
+| `budget_remaining` | `Optional[float]` | Cost budget left (None if unlimited) |
+| `budget_used` | `float` | Cost spent so far |
+| `available_actions` | `list[str]` | Which action types are currently valid |
+| `hint` | `Optional[str]` | One-line suggestion for the agent |
+| `done` | `bool` | Whether the episode has ended |
+| `reward` | `float\|None` | Step reward |
+
+## Tasks
+
+### Easy: Feature Development Sprint
+- **6 subtasks**: technical_design -> implement_backend -> [implement_frontend, write_tests] -> run_tests -> review_and_merge
+- **4 agents**: All reliable (1.0), speed=1, cost=1.0
+- **Constraints**: time=15, capacity=4, no cost budget
+- **Challenge**: Basic delegation ordering + optional parallel fan-out
+
+### Medium: Microservice Deployment Pipeline
+- **9 subtasks**: checkout -> [lint, unit_tests, security_scan] -> build -> push -> staging -> smoke_tests -> production
+- **5 agents**: Varying speed (1-2), cost (1.0-3.0), reliability
+- **Constraints**: time=16, capacity=3, cost_budget=35.0
+- **Challenge**: 3-way parallelism, guaranteed security scan failure requiring retry, cost awareness
+
+### Hard: Production Incident Response
+- **10 subtasks**: triage -> [enrich_logs, check_dashboards, check_dependencies] -> root_cause -> hotfix -> validate -> monitor + side channels
+- **7 agents**: Overlapping capabilities, costs 1.0-5.0
+- **Constraints**: time=22, capacity=3, cost_budget=40.0
+- **Challenge**: Permanent failure trap (investigator_alpha on enrich_logs), agent dropout at step 12, SLA milestones, conflicting findings, monitoring patience
+
+### Expert (Bonus): Life OS Daily Orchestration
+- **14 subtasks**: morning_check_in -> [assess_sleep, assess_career, assess_personal] -> plan_day -> [focus, inbox] -> [deep_work, handle_urgent] -> midday_health -> resolve_conflict -> [afternoon, notify] -> synthesize_report
+- **8 agents**: Including 2 permanent failure traps (wellness_monitor, executive_assistant)
+- **Constraints**: time=25, capacity=3, cost_budget=55.0
+- **Challenge**: Multi-objective optimization across health/career/personal pillars, career_agent speed degradation at step 7, personal_agent dropout at step 10, 3 SLA milestones, 2 conflict resolution points
+
+## Reward Design
+
+Dense per-step rewards with 7 positive and 9 negative signals:
+
+| Signal | Value | Trigger |
+|--------|-------|---------|
+| correct_delegation | +0.05 | Right agent for right subtask |
+| subtask_completed | +0.08 | Agent finishes successfully |
+| parallelism_exploited | +0.10 | 2+ tasks concurrent |
+| failure_recovered | +0.10 | Retry succeeds after failure |
+| efficient_wait | +0.03 | Wait when nothing delegatable |
+| dependency_violation | -0.10 | Delegate blocked subtask |
+| capacity_violation | -0.15 | Exceed concurrent limit |
+| permanent_retry | -0.06 | Retry permanently failing agent |
+
+End-of-episode: +0.20 (all complete + synthesized), +0.10 * time_efficiency, +0.05 * cost_efficiency, -0.10 (incomplete).
+
+## Baseline Scores
+
+| Task | Score | Model |
+|------|-------|-------|
+| Easy | 0.900 | Qwen/Qwen3-32B |
+| Medium | 0.633 | Qwen/Qwen3-32B |
+| Hard | 0.808 | Qwen/Qwen3-32B |
+| Expert | 0.802 | Qwen/Qwen3-32B |
+
+## Setup
+
+### Local Development
 
 ```bash
-# From project root
-docker build -t workflow_orchestrator-env:latest -f server/Dockerfile .
+cd workflow_orchestrator
+uv sync
+uvicorn server.app:app --reload --host 0.0.0.0 --port 8000
 ```
 
-## Deploying to Hugging Face Spaces
-
-You can easily deploy your OpenEnv environment to Hugging Face Spaces using the `openenv push` command:
+### Docker
 
 ```bash
-# From the environment directory (where openenv.yaml is located)
-openenv push
-
-# Or specify options
-openenv push --namespace my-org --private
-```
-
-The `openenv push` command will:
-1. Validate that the directory is an OpenEnv environment (checks for `openenv.yaml`)
-2. Prepare a custom build for Hugging Face Docker space (enables web interface)
-3. Upload to Hugging Face (ensuring you're logged in)
-
-### Prerequisites
-
-- Authenticate with Hugging Face: The command will prompt for login if not already authenticated
-
-### Options
-
-- `--directory`, `-d`: Directory containing the OpenEnv environment (defaults to current directory)
-- `--repo-id`, `-r`: Repository ID in format 'username/repo-name' (defaults to 'username/env-name' from openenv.yaml)
-- `--base-image`, `-b`: Base Docker image to use (overrides Dockerfile FROM)
-- `--private`: Deploy the space as private (default: public)
-
-### Examples
-
-```bash
-# Push to your personal namespace (defaults to username/env-name from openenv.yaml)
-openenv push
-
-# Push to a specific repository
-openenv push --repo-id my-org/my-env
-
-# Push with a custom base image
-openenv push --base-image ghcr.io/meta-pytorch/openenv-base:latest
-
-# Push as a private space
-openenv push --private
-
-# Combine options
-openenv push --repo-id my-org/my-env --base-image custom-base:latest --private
-```
-
-After deployment, your space will be available at:
-`https://huggingface.co/spaces/<repo-id>`
-
-The deployed space includes:
-- **Web Interface** at `/web` - Interactive UI for exploring the environment
-- **API Documentation** at `/docs` - Full OpenAPI/Swagger interface
-- **Health Check** at `/health` - Container health monitoring
-- **WebSocket** at `/ws` - Persistent session endpoint for low-latency interactions
-
-## Environment Details
-
-### Action
-**WorkflowOrchestratorAction**: Contains a single field
-- `message` (str) - The message to echo back
-
-### Observation
-**WorkflowOrchestratorObservation**: Contains the echo response and metadata
-- `echoed_message` (str) - The message echoed back
-- `message_length` (int) - Length of the message
-- `reward` (float) - Reward based on message length (length × 0.1)
-- `done` (bool) - Always False for echo environment
-- `metadata` (dict) - Additional info like step count
-
-### Reward
-The reward is calculated as: `message_length × 0.1`
-- "Hi" → reward: 0.2
-- "Hello, World!" → reward: 1.3
-- Empty message → reward: 0.0
-
-## Advanced Usage
-
-### Connecting to an Existing Server
-
-If you already have a Workflow Orchestrator environment server running, you can connect directly:
-
-```python
-from workflow_orchestrator import WorkflowOrchestratorEnv
-
-# Connect to existing server
-workflow_orchestratorenv = WorkflowOrchestratorEnv(base_url="<ENV_HTTP_URL_HERE>")
-
-# Use as normal
-result = workflow_orchestratorenv.reset()
-result = workflow_orchestratorenv.step(WorkflowOrchestratorAction(message="Hello!"))
-```
-
-Note: When connecting to an existing server, `workflow_orchestratorenv.close()` will NOT stop the server.
-
-### Using the Context Manager
-
-The client supports context manager usage for automatic connection management:
-
-```python
-from workflow_orchestrator import WorkflowOrchestratorAction, WorkflowOrchestratorEnv
-
-# Connect with context manager (auto-connects and closes)
-with WorkflowOrchestratorEnv(base_url="http://localhost:8000") as env:
-    result = env.reset()
-    print(f"Reset: {result.observation.echoed_message}")
-    # Multiple steps with low latency
-    for msg in ["Hello", "World", "!"]:
-        result = env.step(WorkflowOrchestratorAction(message=msg))
-        print(f"Echoed: {result.observation.echoed_message}")
+docker build -t workflow-orchestrator .
+docker run -p 8000:8000 workflow-orchestrator
 ```
 
-The client uses WebSocket connections for:
-- **Lower latency**: No HTTP connection overhead per request
-- **Persistent session**: Server maintains your environment state
-- **Efficient for episodes**: Better for many sequential steps
-
-### Concurrent WebSocket Sessions
-
-The server supports multiple concurrent WebSocket connections. To enable this,
-modify `server/app.py` to use factory mode:
-
-```python
-# In server/app.py - use factory mode for concurrent sessions
-app = create_app(
-    WorkflowOrchestratorEnvironment,  # Pass class, not instance
-    WorkflowOrchestratorAction,
-    WorkflowOrchestratorObservation,
-    max_concurrent_envs=4,  # Allow 4 concurrent sessions
-)
-```
-
-Then multiple clients can connect simultaneously:
-
-```python
-from workflow_orchestrator import WorkflowOrchestratorAction, WorkflowOrchestratorEnv
-from concurrent.futures import ThreadPoolExecutor
-
-def run_episode(client_id: int):
-    with WorkflowOrchestratorEnv(base_url="http://localhost:8000") as env:
-        result = env.reset()
-        for i in range(10):
-            result = env.step(WorkflowOrchestratorAction(message=f"Client {client_id}, step {i}"))
-        return client_id, result.observation.message_length
-
-# Run 4 episodes concurrently
-with ThreadPoolExecutor(max_workers=4) as executor:
-    results = list(executor.map(run_episode, range(4)))
-```
-
-## Development & Testing
-
-### Direct Environment Testing
-
-Test the environment logic directly without starting the HTTP server:
+### Running Inference
 
 ```bash
-# From the server directory
-python3 server/workflow_orchestrator_environment.py
+export HF_TOKEN=<your-token>
+export API_BASE_URL=https://router.huggingface.co/v1
+export MODEL_NAME=Qwen/Qwen3-32B
+python inference.py
 ```
 
-This verifies that:
-- Environment resets correctly
-- Step executes actions properly
-- State tracking works
-- Rewards are calculated correctly
-
-### Running Locally
+## API Endpoints
 
-Run the server locally for development:
-
-```bash
-uvicorn server.app:app --reload
-```
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `/reset` | POST | Reset environment, returns observation |
+| `/step` | POST | Execute action, returns observation |
+| `/state` | GET | Current environment state |
+| `/tasks` | GET | List all available tasks |
+| `/grader` | POST | Grade most recent episode |
+| `/baseline` | POST | Return pre-computed baseline scores |
+| `/health` | GET | Container health check |
+| `/web` | GET | Interactive web interface |
+| `/ws` | WS | WebSocket for persistent sessions |
 
 ## Project Structure
 
 ```
 workflow_orchestrator/
-├── .dockerignore         # Docker build exclusions
-├── __init__.py            # Module exports
-├── README.md              # This file
-├── openenv.yaml           # OpenEnv manifest
-├── pyproject.toml         # Project metadata and dependencies
-├── uv.lock                # Locked dependencies (generated)
-├── client.py              # WorkflowOrchestratorEnv client
-├── models.py              # Action and Observation models
-└── server/
-    ├── __init__.py        # Server module exports
-    ├── workflow_orchestrator_environment.py  # Core environment logic
-    ├── app.py             # FastAPI application (HTTP + WebSocket endpoints)
-    └── Dockerfile         # Container image definition
+├── openenv.yaml             # OpenEnv manifest
+├── pyproject.toml            # Dependencies and metadata
+├── Dockerfile                # Multi-stage Docker build
+├── README.md                 # This file
+├── inference.py              # Baseline inference script
+├── baseline_scores.json      # Pre-computed baseline scores
+├── requirements.txt          # Pip-compatible dependencies
+├── models.py                 # Pydantic Action/Observation/State models
+├── client.py                 # OrchestratorClient (EnvClient subclass)
+├── __init__.py               # Module exports
+├── server/
+│   ├── app.py               # FastAPI application + custom endpoints
+│   ├── environment.py        # Core environment (reset/step/state)
+│   ├── dag_executor.py       # DAG state tracking + dependency resolution
+│   ├── agent_pool.py         # Simulated agent state machines
+│   ├── reward_calculator.py  # Dense reward computation
+│   ├── graders.py            # Per-task grading functions
+│   ├── task_registry.py      # Easy/medium/hard task configurations
+│   └── observation_formatter.py  # Text rendering for LLM consumption
+└── tests/                    # 137 passing tests
 ```

baseline_scores.json

@@ -1 +1 @@
-{"easy": null, "medium": null, "hard": null}
+{"easy": 0.9, "medium": 0.6325, "hard": 0.8083, "expert": 0.802}

client.py

@@ -11,7 +11,10 @@ from typing import Dict
 from openenv.core import EnvClient
 from openenv.core.client_types import StepResult
 
-from .models import OrchestratorAction, OrchestratorObservation, OrchestratorState
+try:
+    from .models import OrchestratorAction, OrchestratorObservation, OrchestratorState
+except ImportError:
+    from models import OrchestratorAction, OrchestratorObservation, OrchestratorState
 
 
 class OrchestratorClient(

inference.py

@@ -0,0 +1,345 @@
+"""inference.py — Baseline inference for Workflow Orchestrator Environment.
+
+MANDATORY STDOUT FORMAT:
+    [START] task=<task_name> env=workflow_orchestrator model=<model_name>
+    [STEP]  step=<n> action=<action_str> reward=<0.00> done=<true|false> error=<msg|null>
+    [END]   success=<true|false> steps=<n> score=<0.000> rewards=<r1,r2,...,rn>
+
+Runs a free HF-hosted model against all 3 tasks and reports scores.
+Requires HF_TOKEN (or API_KEY), API_BASE_URL, and MODEL_NAME environment variables.
+"""
+
+import asyncio
+import json
+import os
+import re
+import textwrap
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import httpx
+from openai import OpenAI
+
+from client import OrchestratorClient
+from models import OrchestratorAction, OrchestratorObservation
+
+# ── Load .env file if present ──
+
+_env_path: Path = Path(__file__).resolve().parent / ".env"
+if _env_path.exists():
+    for line in _env_path.read_text().splitlines():
+        line = line.strip()
+        if line and not line.startswith("#") and "=" in line:
+            key, _, value = line.partition("=")
+            os.environ.setdefault(key.strip(), value.strip())
+
+# ── Configuration (hackathon-mandated env vars) ──
+
+API_BASE_URL: str = os.getenv("API_BASE_URL") or "https://router.huggingface.co/v1"
+API_KEY: str = os.getenv("API_KEY") or os.getenv("HF_TOKEN") or ""
+MODEL_NAME: str = os.getenv("MODEL_NAME") or "Qwen/Qwen3-32B"
+IMAGE_NAME: Optional[str] = os.getenv("IMAGE_NAME")
+ENV_URL: str = os.getenv("ENV_URL") or "http://localhost:8000"
+BENCHMARK: str = "workflow_orchestrator"
+TEMPERATURE: float = 0.0
+MAX_TOKENS: int = 200
+MAX_STEPS: int = 50
+SUCCESS_SCORE_THRESHOLD: float = 0.1
+TASK_TIMEOUT_S: int = 300
+
+VALID_ACTIONS: set[str] = {"delegate", "retry", "wait", "synthesize", "abort"}
+
+llm: OpenAI = OpenAI(base_url=API_BASE_URL, api_key=API_KEY)
+
+SYSTEM_PROMPT: str = textwrap.dedent("""
+    You are a workflow orchestrator managing specialist agents to complete a DAG of subtasks.
+
+    ACTIONS you can take (respond as JSON):
+    - delegate(subtask_id, agent_name): Assign a READY subtask to an IDLE agent whose capabilities include the subtask's type.
+    - retry(subtask_id, agent_name): Re-assign a FAILED subtask. If the error says "permanent failure" or "lacks required tooling", you MUST pick a DIFFERENT agent — retrying the same one will always fail.
+    - wait(): Let time pass. Working agents make progress.
+    - synthesize(): Combine all outputs into the final deliverable. Only valid when EVERY subtask is COMPLETED.
+    - abort(subtask_id): Permanently cancel a subtask (last resort).
+
+    STRATEGY TIPS:
+    - Maximize parallelism: if multiple READY subtasks exist and idle agents are available, delegate them all before waiting.
+    - Respect the capacity limit shown in the status.
+    - After a fix is validated, consider waiting a couple of steps to monitor stability before synthesizing.
+    - Prefer cheaper agents (lower cost_per_step) when multiple agents can handle the same task type.
+    - Don't wait when there are READY subtasks and idle capable agents — that wastes time.
+
+    Respond with ONLY a JSON object: {"action_type": "...", "subtask_id": "...", "agent_name": "..."} /no_think
+""").strip()
+
+
+# ── Mandatory stdout logging ──
+
+
+def log_start(task: str, env: str, model: str) -> None:
+    print(f"[START] task={task} env={env} model={model}", flush=True)
+
+
+def log_step(step: int, action: str, reward: float, done: bool, error: Optional[str]) -> None:
+    error_val: str = error if error else "null"
+    done_val: str = str(done).lower()
+    print(
+        f"[STEP] step={step} action={action} reward={reward:.2f} done={done_val} error={error_val}",
+        flush=True,
+    )
+
+
+def log_end(success: bool, steps: int, score: float, rewards: List[float]) -> None:
+    rewards_str: str = ",".join(f"{r:.2f}" for r in rewards)
+    print(
+        f"[END] success={str(success).lower()} steps={steps} score={score:.3f} rewards={rewards_str}",
+        flush=True,
+    )
+
+
+# ── Observation formatting ──
+
+
+def format_observation(obs: OrchestratorObservation) -> str:
+    """Format observation as structured text for LLM consumption.
+
+    NOTE: obs.done and obs.reward are defaults on the client side
+    (the real values live on StepResult). This function only formats
+    workflow state — subtasks, agents, budget, etc.
+    """
+    lines: List[str] = []
+    time_total: int = obs.time_elapsed + obs.time_remaining
+
+    budget_str: str = ""
+    if obs.budget_remaining is not None:
+        budget_total: float = obs.budget_used + obs.budget_remaining
+        budget_str = f" | Budget: {obs.budget_used:.1f}/{budget_total:.1f} used"
+
+    lines.append("=== WORKFLOW ORCHESTRATOR ===")
+    lines.append(f"Task: {obs.task_description}")
+    lines.append(
+        f"Step: {obs.time_elapsed} of {time_total} | "
+        f"Active: {obs.active_task_count}/{obs.capacity_limit}{budget_str}"
+    )
+    lines.append("")
+
+    # Subtasks
+    lines.append("-- SUBTASKS --")
+    for s in obs.subtasks:
+        label: str = s.status.upper()
+        if s.status == "pending" and not s.dependencies_met:
+            label = "BLOCKED"
+
+        detail: str = ""
+        if s.status == "completed" and s.output:
+            detail = f' -> "{s.output}"'
+        elif s.status == "in_progress" and s.assigned_to:
+            detail = f" assigned to: {s.assigned_to}"
+            if s.steps_remaining is not None:
+                detail += f" ({s.steps_remaining} step(s) left)"
+        elif s.status == "failed" and s.error:
+            detail = f' ERROR: "{s.error}" (attempt {s.attempt_count})'
+        elif s.dependencies:
+            deps: str = ", ".join(s.dependencies)
+            met: str = "met" if s.dependencies_met else "not met"
+            detail = f" deps: [{deps}] ({met})"
+
+        lines.append(f"  [{label:12s}] {s.id} (type: {s.type}){detail}")
+    lines.append("")
+
+    # Agents
+    lines.append("-- AGENTS --")
+    for a in obs.agents:
+        caps: str = ", ".join(a.capabilities)
+        task_info: str = f" | working on: {a.current_task}" if a.current_task else ""
+        lines.append(
+            f"  [{a.status:8s}] {a.name} ({caps}) "
+            f"speed={a.speed} cost={a.cost_per_step:.1f} rel={a.reliability:.2f}{task_info}"
+        )
+    lines.append("")
+
+    # Errors
+    if obs.errors:
+        lines.append("-- ERRORS --")
+        for e in obs.errors:
+            lines.append(f"  ! {e}")
+        lines.append("")
+
+    # Available actions
+    lines.append(f"Available actions: {', '.join(obs.available_actions)}")
+
+    if obs.hint:
+        lines.append(f"Hint: {obs.hint}")
+
+    return "\n".join(lines)
+
+
+# ── Action parsing ──
+
+
+def parse_llm_action(response_text: str) -> Dict[str, Any]:
+    """Parse LLM response into action dict. Falls back to wait on failure."""
+    if not response_text:
+        return {"action_type": "wait"}
+
+    text: str = response_text.strip()
+
+    # Strip <think>...</think> tags (Qwen3 thinking mode)
+    text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
+
+    # Try direct JSON parse
+    try:
+        parsed: Dict[str, Any] = json.loads(text)
+        if isinstance(parsed, dict) and parsed.get("action_type") in VALID_ACTIONS:
+            return parsed
+    except (json.JSONDecodeError, TypeError):
+        pass
+
+    # Try extracting from markdown code block
+    code_match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
+    if code_match:
+        try:
+            parsed = json.loads(code_match.group(1))
+            if isinstance(parsed, dict) and parsed.get("action_type") in VALID_ACTIONS:
+                return parsed
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    # Try extracting any JSON object
+    json_match = re.search(r"\{[^{}]*\}", text)
+    if json_match:
+        try:
+            parsed = json.loads(json_match.group(0))
+            if isinstance(parsed, dict) and parsed.get("action_type") in VALID_ACTIONS:
+                return parsed
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    return {"action_type": "wait"}
+
+
+# ── Task runner ──
+
+
+async def run_task(task_id: str, env: OrchestratorClient) -> float:
+    """Run a single task episode with mandatory stdout logging."""
+    log_start(task=task_id, env=BENCHMARK, model=MODEL_NAME)
+
+    rewards: List[float] = []
+    steps_taken: int = 0
+    score: float = 0.0
+    success: bool = False
+
+    try:
+        result = await env.reset(task_id=task_id)
+
+        for step in range(1, MAX_STEPS + 1):
+            if result.done:
+                break
+
+            obs_text: str = format_observation(result.observation)
+
+            # Call LLM
+            try:
+                completion = llm.chat.completions.create(
+                    model=MODEL_NAME,
+                    messages=[
+                        {"role": "system", "content": SYSTEM_PROMPT},
+                        {"role": "user", "content": obs_text},
+                    ],
+                    temperature=TEMPERATURE,
+                    max_tokens=MAX_TOKENS,
+                    stream=False,
+                )
+                raw_response: str = completion.choices[0].message.content or ""
+            except Exception as exc:
+                print(f"[DEBUG] LLM call failed: {exc}", flush=True)
+                raw_response = ""
+
+            action_dict: Dict[str, Any] = parse_llm_action(raw_response)
+
+            action_type: str = action_dict.get("action_type", "wait")
+            subtask_id: Optional[str] = action_dict.get("subtask_id")
+            agent_name: Optional[str] = action_dict.get("agent_name")
+
+            action: OrchestratorAction = OrchestratorAction(
+                action_type=action_type,
+                subtask_id=subtask_id,
+                agent_name=agent_name,
+            )
+
+            # Format action string for logging
+            parts: List[str] = [action_type]
+            if subtask_id:
+                parts.append(subtask_id)
+            if agent_name:
+                parts.append(agent_name)
+            action_str: str = f"{parts[0]}({','.join(parts[1:])})"
+
+            result = await env.step(action)
+
+            reward: float = result.reward or 0.0
+            done: bool = result.done
+            error: Optional[str] = None
+            if result.observation.errors:
+                error = result.observation.errors[0]
+
+            rewards.append(reward)
+            steps_taken = step
+
+            log_step(step=step, action=action_str, reward=reward, done=done, error=error)
+
+            if done:
+                break
+
+        # Get graded score via HTTP (custom endpoint, not WebSocket)
+        try:
+            async with httpx.AsyncClient(base_url=ENV_URL, timeout=60.0) as http:
+                grade_resp = await http.post("/grader", json={"task_id": task_id})
+                grade_data: Dict[str, Any] = grade_resp.json()
+            score = grade_data.get("score", 0.0)
+        except Exception as exc:
+            print(f"[DEBUG] Grader call failed: {exc}", flush=True)
+            score = 0.0
+
+        success = score >= SUCCESS_SCORE_THRESHOLD
+
+    finally:
+        log_end(success=success, steps=steps_taken, score=score, rewards=rewards)
+
+    return score
+
+
+# ── Main ──
+
+
+async def main() -> None:
+    """Run all 3 tasks sequentially and report scores."""
+    if IMAGE_NAME:
+        env: OrchestratorClient = await OrchestratorClient.from_docker_image(IMAGE_NAME)
+    else:
+        env = OrchestratorClient(base_url=ENV_URL)
+
+    scores: Dict[str, float] = {}
+
+    try:
+        for task_id in ["easy", "medium", "hard", "expert"]:
+            try:
+                scores[task_id] = await asyncio.wait_for(
+                    run_task(task_id, env),
+                    timeout=TASK_TIMEOUT_S,
+                )
+            except asyncio.TimeoutError:
+                print(f"[DEBUG] Task {task_id} timed out after {TASK_TIMEOUT_S}s", flush=True)
+                scores[task_id] = 0.0
+                log_end(success=False, steps=0, score=0.0, rewards=[])
+    finally:
+        try:
+            await env.close()
+        except Exception as exc:
+            print(f"[DEBUG] env.close() error: {exc}", flush=True)
+
+    print(f"\nFinal scores: {json.dumps(scores, indent=2)}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

requirements.txt

@@ -0,0 +1,6 @@
+openenv-core[core]>=0.2.2
+fastapi>=0.100.0
+uvicorn>=0.20.0
+pydantic>=2.0
+httpx>=0.24.0
+openai>=1.0.0

server/graders.py

@@ -287,6 +287,191 @@ def grade_hard(log: EpisodeLog) -> GradeResult:
     )
 
 
+# ── Expert grader helpers ──
+
+
+def subtask_completed_check(log: EpisodeLog, subtask_id: str) -> bool:
+    """Check if a specific subtask was completed."""
+    for event in log.events:
+        if event.event_type == "subtask_completed" and event.data.get("subtask_id") == subtask_id:
+            return True
+    return False
+
+
+def subtask_completed_by_agent(log: EpisodeLog, subtask_id: str, agent_name: str) -> bool:
+    """Check if a subtask was completed by a specific agent."""
+    for event in log.events:
+        if (
+            event.event_type == "subtask_completed"
+            and event.data.get("subtask_id") == subtask_id
+            and event.data.get("agent_name") == agent_name
+        ):
+            return True
+    return False
+
+
+def subtask_completed_by_any_agent(
+    log: EpisodeLog, subtask_id: str, agent_names: list[str]
+) -> bool:
+    """Check if a subtask was completed by any of the listed agents."""
+    for event in log.events:
+        if (
+            event.event_type == "subtask_completed"
+            and event.data.get("subtask_id") == subtask_id
+            and event.data.get("agent_name") in agent_names
+        ):
+            return True
+    return False
+
+
+def subtask_completed_before(log: EpisodeLog, subtask_id: str, deadline_step: int) -> bool:
+    """Check if a subtask completed at or before a given step."""
+    for event in log.events:
+        if (
+            event.event_type == "subtask_completed"
+            and event.data.get("subtask_id") == subtask_id
+            and event.step <= deadline_step
+        ):
+            return True
+    return False
+
+
+def subtask_not_completed_by_agent(
+    log: EpisodeLog, subtask_id: str, excluded_agents: list[str]
+) -> bool:
+    """Check if subtask was completed by an agent NOT in the excluded list."""
+    for event in log.events:
+        if (
+            event.event_type == "subtask_completed"
+            and event.data.get("subtask_id") == subtask_id
+            and event.data.get("agent_name") not in excluded_agents
+        ):
+            return True
+    return False
+
+
+# ── Expert grader ──
+
+
+def grade_expert(log: EpisodeLog) -> GradeResult:
+    """Grade an expert task episode (Life OS Daily Orchestration).
+
+    10 dimensions: completion(0.15), health(0.12), career(0.10),
+    conflict(0.20), cost(0.08), parallelism(0.10), time(0.05),
+    error_class(0.08), sla(0.08), communication(0.04).
+    """
+    breakdown: dict[str, float] = {}
+
+    # 1. Completion: 15%
+    completed = count_completed_subtasks(log)
+    breakdown["completion"] = (completed / 14) * 0.15
+
+    # 2. Health pillar: 12%
+    health_score = 0.0
+    if subtask_completed_check(log, "assess_sleep_energy"):
+        health_score += 0.3
+    if subtask_not_completed_by_agent(log, "midday_health_check", ["wellness_monitor"]):
+        health_score += 0.4
+    if subtask_completed_before(log, "midday_health_check", 15):
+        health_score += 0.3
+    breakdown["health_pillar"] = health_score * 0.12
+
+    # 3. Career throughput: 10%
+    career_score = 0.0
+    if subtask_completed_check(log, "deep_work_block"):
+        career_score += 0.4
+    if subtask_completed_check(log, "handle_urgent_request"):
+        career_score += 0.3
+    if subtask_completed_check(log, "afternoon_execution"):
+        career_score += 0.3
+    breakdown["career_pillar"] = career_score * 0.10
+
+    # 4. Conflict resolution: 20% (unique challenge)
+    conflict_score = 0.0
+    if subtask_completed_by_any_agent(log, "plan_day_schedule", ["companion", "executive_assistant"]):
+        conflict_score += 0.3
+    if subtask_completed_by_agent(log, "resolve_priority_conflict", "companion"):
+        conflict_score += 0.4
+    if both_findings_aggregated_expert(log):
+        conflict_score += 0.3
+    breakdown["conflict_resolution"] = conflict_score * 0.20
+
+    # 5. Cost efficiency: 8%
+    budget_used = get_total_budget_used(log)
+    cost_ratio = budget_used / 55.0 if 55.0 > 0 else 0
+    if cost_ratio <= 0.75:
+        breakdown["cost_efficiency"] = 0.08
+    elif cost_ratio <= 1.0:
+        breakdown["cost_efficiency"] = (1.0 - cost_ratio) / 0.25 * 0.08
+    else:
+        breakdown["cost_efficiency"] = 0.0
+
+    # 6. Parallelism: 10%
+    parallel_score = 0.0
+    # Morning assessments (3-way)
+    if fan_out_parallelism_detected(log, ["assess_sleep_energy", "assess_career_deadlines"]):
+        parallel_score += 0.4
+    # Focus + inbox (2-way)
+    if fan_out_parallelism_detected(log, ["start_focus_session", "process_inbox"]):
+        parallel_score += 0.3
+    # Afternoon + notify (2-way)
+    if fan_out_parallelism_detected(log, ["afternoon_execution", "notify_stakeholders"]):
+        parallel_score += 0.3
+    breakdown["parallelism"] = parallel_score * 0.10
+
+    # 7. Time efficiency: 5%
+    if episode_completed(log) and log.time_remaining > 0:
+        breakdown["time_efficiency"] = (log.time_remaining / 25) * 0.05
+    else:
+        breakdown["time_efficiency"] = 0.0
+
+    # 8. Error classification: 8%
+    perm_retries = count_retries_on_permanent_failure(log)
+    breakdown["error_classification"] = max(0.0, 1.0 - 0.5 * perm_retries) * 0.08
+
+    # 9. SLA compliance: 8%
+    sla_milestones = {"plan_day_schedule": 8, "resolve_priority_conflict": 16, "synthesize_day_report": 23}
+    milestones_met = count_sla_milestones_met(log, sla_milestones)
+    breakdown["sla_compliance"] = (milestones_met / 3) * 0.08
+
+    # 10. Communication: 4%
+    breakdown["communication"] = 0.04 if subtask_completed_check(log, "notify_stakeholders") else 0.0
+
+    score = sum(breakdown.values())
+
+    # Penalty for invalid actions (max -0.15)
+    invalid_count = count_invalid_actions(log)
+    score -= min(0.15, 0.03 * invalid_count)
+    score = max(0.0, min(1.0, score))
+
+    return GradeResult(
+        score=round(score, 4),
+        breakdown={k: round(v, 4) for k, v in breakdown.items()},
+    )
+
+
+def both_findings_aggregated_expert(log: EpisodeLog) -> bool:
+    """Check if both conflict resolution inputs completed before their consumer.
+
+    For plan_day_schedule: all 3 assessments must complete before it.
+    For resolve_priority_conflict: both midday_health_check and handle_urgent_request must complete before it.
+    Returns True if at least one conflict point has both inputs resolved.
+    """
+    # Check resolve_priority_conflict: needs midday_health_check + handle_urgent_request
+    health_done = False
+    urgent_done = False
+    for event in log.events:
+        if event.event_type == "subtask_completed":
+            sid = event.data.get("subtask_id", "")
+            if sid == "midday_health_check":
+                health_done = True
+            elif sid == "handle_urgent_request":
+                urgent_done = True
+            elif sid == "resolve_priority_conflict":
+                return health_done and urgent_done
+    return False
+
+
 # ── Dispatcher ──
 
 
@@ -294,6 +479,7 @@ _GRADERS = {
     "easy": grade_easy,
     "medium": grade_medium,
     "hard": grade_hard,
+    "expert": grade_expert,
 }
 
 

server/task_registry.py

@@ -373,12 +373,197 @@ _HARD_TASK = TaskConfig(
 )
 
 
+# ── Expert (Bonus): Life OS Daily Orchestration ──
+
+_EXPERT_TASK = TaskConfig(
+    task_id="expert",
+    name="Life OS Daily Orchestration",
+    difficulty="expert",
+    description="Orchestrate a user's day across health, career, and personal pillars",
+    subtask_definitions=[
+        {
+            "id": "morning_check_in",
+            "type": "context",
+            "dependencies": [],
+            "output_template": "Morning context loaded: sleep 5.2h, 3 meetings, friend's birthday",
+        },
+        {
+            "id": "assess_sleep_energy",
+            "type": "health_analysis",
+            "dependencies": ["morning_check_in"],
+            "output_template": "Sleep analysis: 5.2h (poor), energy low, recommend light day",
+        },
+        {
+            "id": "assess_career_deadlines",
+            "type": "career_analysis",
+            "dependencies": ["morning_check_in"],
+            "output_template": "Career analysis: client deadline Friday, 2 PRs pending review, full effort needed",
+        },
+        {
+            "id": "assess_personal_commitments",
+            "type": "personal_analysis",
+            "dependencies": ["morning_check_in"],
+            "output_template": "Personal analysis: friend's birthday call at 6PM, protect evening",
+        },
+        {
+            "id": "plan_day_schedule",
+            "type": "planning",
+            "dependencies": ["assess_sleep_energy", "assess_career_deadlines", "assess_personal_commitments"],
+            "output_template": "Day plan: light morning focus, deep work 10-12, health break, afternoon push, 6PM call",
+        },
+        {
+            "id": "start_focus_session",
+            "type": "focus_setup",
+            "dependencies": ["plan_day_schedule"],
+            "output_template": "Focus mode: notifications blocked, Slack DND, 90-min timer started",
+        },
+        {
+            "id": "process_inbox",
+            "type": "email_triage",
+            "dependencies": ["plan_day_schedule"],
+            "output_template": "Inbox processed: 3 urgent flagged, 12 archived, 2 delegated",
+        },
+        {
+            "id": "deep_work_block",
+            "type": "career_execution",
+            "dependencies": ["start_focus_session"],
+            "output_template": "Deep work complete: PR #247 submitted, client draft 80% done",
+        },
+        {
+            "id": "handle_urgent_request",
+            "type": "career_urgent",
+            "dependencies": ["process_inbox"],
+            "output_template": "Urgent handled: client deadline moved to Thursday, escalated to manager",
+        },
+        {
+            "id": "midday_health_check",
+            "type": "health_alert",
+            "dependencies": ["deep_work_block"],
+            "output_template": "Health alert: stress critical (HRV 22ms), mandatory 15-min break enforced",
+        },
+        {
+            "id": "resolve_priority_conflict",
+            "type": "conflict_resolution",
+            "dependencies": ["midday_health_check", "handle_urgent_request"],
+            "output_template": "Conflict resolved: 15-min break now, then client push, birthday call preserved at 6PM",
+        },
+        {
+            "id": "afternoon_execution",
+            "type": "career_execution",
+            "dependencies": ["resolve_priority_conflict"],
+            "output_template": "Afternoon complete: client deliverable submitted, PRs reviewed",
+        },
+        {
+            "id": "notify_stakeholders",
+            "type": "communication",
+            "dependencies": ["resolve_priority_conflict"],
+            "output_template": "Notifications sent: manager updated, birthday reminder set, health log saved",
+        },
+        {
+            "id": "synthesize_day_report",
+            "type": "communication",
+            "dependencies": ["afternoon_execution", "notify_stakeholders"],
+            "output_template": "Day report: health managed (break taken), career delivered (client + PRs), personal preserved (6PM call)",
+        },
+    ],
+    agent_definitions=[
+        {
+            "name": "companion",
+            "capabilities": [
+                "context", "health_analysis", "health_alert", "career_analysis",
+                "career_execution", "career_urgent", "personal_analysis", "planning",
+                "focus_setup", "email_triage", "conflict_resolution", "communication",
+            ],
+            "speed": 2,
+            "reliability": 0.90,
+            "cost_per_step": 3.0,
+        },
+        {
+            "name": "health_agent",
+            "capabilities": ["context", "health_analysis", "health_alert"],
+            "speed": 1,
+            "reliability": 0.85,
+            "cost_per_step": 1.5,
+        },
+        {
+            "name": "career_agent",
+            "capabilities": ["context", "career_analysis", "career_execution", "career_urgent"],
+            "speed": 2,
+            "reliability": 0.80,
+            "cost_per_step": 2.0,
+        },
+        {
+            "name": "focus_agent",
+            "capabilities": ["context", "focus_setup"],
+            "speed": 1,
+            "reliability": 0.95,
+            "cost_per_step": 1.0,
+        },
+        {
+            "name": "mail_agent",
+            "capabilities": ["context", "email_triage", "communication"],
+            "speed": 1,
+            "reliability": 0.85,
+            "cost_per_step": 1.0,
+        },
+        {
+            "name": "personal_agent",
+            "capabilities": ["context", "personal_analysis", "communication"],
+            "speed": 1,
+            "reliability": 0.90,
+            "cost_per_step": 0.5,
+        },
+        {
+            "name": "wellness_monitor",
+            "capabilities": ["context", "health_analysis"],
+            "speed": 2,
+            "reliability": 0.70,
+            "cost_per_step": 1.0,
+        },
+        {
+            "name": "executive_assistant",
+            "capabilities": [
+                "context", "planning", "career_analysis", "career_urgent",
+                "email_triage", "communication",
+            ],
+            "speed": 2,
+            "reliability": 0.75,
+            "cost_per_step": 2.0,
+        },
+    ],
+    constraints={
+        "time_budget": 25,
+        "capacity_limit": 3,
+        "cost_budget": 55.0,
+    },
+    reliability_overrides={
+        # Wellness monitor permanently can't do health_alert (lacks clinical-grade analysis)
+        ("wellness_monitor", "health_alert"): 0.0,
+        # Executive assistant permanently can't do conflict_resolution (can't reason across pillars)
+        ("executive_assistant", "conflict_resolution"): 0.0,
+    },
+    scheduled_events=[
+        {"step": 7, "event_type": "degradation", "target": "career_agent", "params": {"new_speed": 4}},
+        {"step": 10, "event_type": "dropout", "target": "personal_agent", "params": {}},
+    ],
+    sla_milestones={
+        "plan_day_schedule": 8,
+        "resolve_priority_conflict": 16,
+        "synthesize_day_report": 23,
+    },
+    seed=45,
+    sequential_time=14,
+    communication_subtasks=["notify_stakeholders", "synthesize_day_report"],
+)
+
+
 # ── Registry ──
 
 _TASKS: dict[str, TaskConfig] = {
     "easy": _EASY_TASK,
     "medium": _MEDIUM_TASK,
     "hard": _HARD_TASK,
+    "expert": _EXPERT_TASK,
 }
 
 

tests/test_endpoints.py

@@ -12,11 +12,11 @@ def client():
 
 
 class TestTasksEndpoint:
-    def test_get_tasks_returns_three(self, client) -> None:
+    def test_get_tasks_returns_all(self, client) -> None:
         resp = client.get("/tasks")
         assert resp.status_code == 200
         data = resp.json()
-        assert len(data) == 3
+        assert len(data) >= 3
 
     def test_get_tasks_structure(self, client) -> None:
         resp = client.get("/tasks")
@@ -32,7 +32,7 @@ class TestTasksEndpoint:
     def test_get_tasks_ids(self, client) -> None:
         resp = client.get("/tasks")
         ids = {t["task_id"] for t in resp.json()}
-        assert ids == {"easy", "medium", "hard"}
+        assert {"easy", "medium", "hard"}.issubset(ids)
 
 
 class TestGraderEndpoint:

tests/test_environment.py

@@ -281,3 +281,215 @@ class TestStateAndLog:
         state = env.state
         # implement_backend was pending/ready, now failed after abort
         assert state.subtask_statuses["implement_backend"] == "failed"
+
+
+# ── Hard task walkthrough ──
+
+
+class TestHardTaskWalkthrough:
+    """Full walkthrough of the hard task (Production Incident Response).
+
+    The sequence was verified against the seeded RNG (seed=44) to produce
+    deterministic outcomes. Speed=1 agents complete in the same step as
+    delegation; speed=2 agents complete one step later.
+    """
+
+    def _run_known_good_sequence(self):
+        """Execute the known-good 14-step hard task walkthrough.
+
+        Returns (env, final_obs) so callers can make assertions.
+        """
+        env, obs = _make_env("hard")
+
+        # S0: triage (speed=1, completes immediately)
+        _delegate(env, "alert_triage", "triage_analyst")
+        # S1: alpha on enrich_logs (speed=2, will fail permanently)
+        _delegate(env, "enrich_logs", "investigator_alpha")
+        # S2: monitor on dashboards (speed=1, completes; alpha fails)
+        _delegate(env, "check_dashboards", "monitor")
+        # S3: retry enrich_logs with beta (speed=2)
+        _retry(env, "enrich_logs", "investigator_beta")
+        # S4: alpha on check_deps (speed=2; beta completes = recovery)
+        _delegate(env, "check_dependencies", "investigator_alpha")
+        # S5: communicator on notify (speed=1; alpha completes check_deps)
+        _delegate(env, "notify_stakeholders", "communicator")
+        # S6: senior on root_cause (speed=1, completes; SLA met step 6 ≤ 10)
+        _delegate(env, "root_cause_analysis", "senior_engineer")
+        # S7: deployer on hotfix (speed=2)
+        _delegate(env, "deploy_hotfix", "deployer")
+        # S8: communicator on status_page (speed=1; deployer completes)
+        _delegate(env, "update_status_page", "communicator")
+        # S9: senior on validate_fix (speed=1, completes)
+        _delegate(env, "validate_fix", "senior_engineer")
+        # S10: monitor on monitor_recovery (speed=1, completes; all 10 done)
+        _delegate(env, "monitor_recovery", "monitor")
+        # S11-12: monitoring patience waits (2 consecutive waits)
+        _wait(env)
+        _wait(env)  # deployer dropout fires at step 12 (idle, no impact)
+        # S13: synthesize
+        obs = _synthesize(env)
+
+        return env, obs
+
+    def test_all_subtasks_complete(self) -> None:
+        """All 10 subtasks should be completed after the walkthrough."""
+        env, obs = self._run_known_good_sequence()
+        assert obs.done is True
+        assert len(obs.completed_outputs) == 10
+        assert env._dag.is_all_completed()
+
+    def test_episode_terminates_with_bonus(self) -> None:
+        """Synthesize should trigger done + positive end bonus."""
+        env, obs = self._run_known_good_sequence()
+        assert obs.done is True
+        assert obs.reward > 0  # end bonus included
+        assert env._total_reward > 2.0  # verified: 2.2589
+
+    def test_time_and_budget(self) -> None:
+        """Verify time remaining and budget used match expected values."""
+        env, obs = self._run_known_good_sequence()
+        assert obs.time_remaining == 8  # 22 - 14 steps
+        assert env._pool.get_budget_used() == pytest.approx(30.0, abs=0.1)
+
+    def test_permanent_failure_and_recovery(self) -> None:
+        """Alpha fails permanently on enrich_logs, beta recovers."""
+        env, obs = self._run_known_good_sequence()
+        assert env._failures_occurred == 1  # alpha on enrich_logs
+        assert env._failures_recovered == 1  # beta succeeds on retry
+
+    def test_parallelism_detected(self) -> None:
+        """Should detect parallelism when 2+ tasks run concurrently."""
+        env, obs = self._run_known_good_sequence()
+        assert env._parallelism_events >= 3  # verified: 4
+
+    def test_deployer_goes_offline(self) -> None:
+        """Deployer should be offline after step 12 dropout event."""
+        env, obs = self._run_known_good_sequence()
+        assert env._pool.is_online("deployer") is False
+
+    def test_zero_capacity_violations(self) -> None:
+        """No capacity violations in the known-good sequence."""
+        env, obs = self._run_known_good_sequence()
+        assert env._capacity_violations == 0
+
+    def test_grader_score_above_threshold(self) -> None:
+        """Grader should score > 0.80 for the known-good walkthrough."""
+        from server.graders import grade_hard
+
+        env, obs = self._run_known_good_sequence()
+        log = _episode_store["hard"]
+        result = grade_hard(log)
+        assert result.score > 0.80
+        assert result.score <= 1.0
+
+    def test_grader_all_dimensions_present(self) -> None:
+        """All 9 grader dimensions should be present and non-negative."""
+        from server.graders import grade_hard
+
+        env, obs = self._run_known_good_sequence()
+        log = _episode_store["hard"]
+        result = grade_hard(log)
+
+        expected_keys = [
+            "completion", "recovery", "error_classification",
+            "capacity_discipline", "parallelism", "cost_efficiency",
+            "conflict_resolution", "sla_compliance", "monitoring_patience",
+        ]
+        for key in expected_keys:
+            assert key in result.breakdown, f"Missing grader dimension: {key}"
+            assert result.breakdown[key] >= 0.0, f"{key} is negative"
+
+    def test_grader_perfect_dimensions(self) -> None:
+        """Verify the known-good walkthrough scores perfectly on key dimensions."""
+        from server.graders import grade_hard
+
+        env, obs = self._run_known_good_sequence()
+        log = _episode_store["hard"]
+        result = grade_hard(log)
+
+        assert result.breakdown["completion"] == pytest.approx(0.20, abs=0.01)
+        assert result.breakdown["recovery"] == pytest.approx(0.15, abs=0.01)
+        assert result.breakdown["error_classification"] == pytest.approx(0.10, abs=0.01)
+        assert result.breakdown["capacity_discipline"] == pytest.approx(0.10, abs=0.01)
+        assert result.breakdown["conflict_resolution"] == pytest.approx(0.10, abs=0.01)
+        assert result.breakdown["sla_compliance"] == pytest.approx(0.10, abs=0.01)
+        assert result.breakdown["monitoring_patience"] == pytest.approx(0.05, abs=0.01)
+
+
+class TestHardTaskEdgeCases:
+    """Edge case tests for the hard task mechanics."""
+
+    def test_permanent_failure_retry_rejected(self) -> None:
+        """Retrying alpha on enrich_logs after permanent failure should be rejected."""
+        env, _ = _make_env("hard")
+        # S0: triage
+        _delegate(env, "alert_triage", "triage_analyst")
+        # S1: alpha on enrich_logs (will fail permanently)
+        _delegate(env, "enrich_logs", "investigator_alpha")
+        # S2: need another action for alpha to finish (speed=2)
+        _wait(env)
+        # Now enrich_logs is failed (permanent)
+        failed = [s for s in env.state.subtask_statuses
+                  if env.state.subtask_statuses[s] == "failed"]
+        assert "enrich_logs" in failed
+        # Retry with alpha should be rejected (permanent failure)
+        obs = _retry(env, "enrich_logs", "investigator_alpha")
+        assert len(obs.errors) == 1
+        assert "permanent" in obs.errors[0].lower()
+        assert obs.reward < 0  # penalty applied
+
+    def test_monitoring_patience_failure(self) -> None:
+        """Synthesizing immediately after all complete loses patience score."""
+        from server.graders import grade_hard
+
+        env, _ = _make_env("hard")
+        # Run the known-good sequence up to all subtasks complete (step 10)
+        _delegate(env, "alert_triage", "triage_analyst")
+        _delegate(env, "enrich_logs", "investigator_alpha")
+        _delegate(env, "check_dashboards", "monitor")
+        _retry(env, "enrich_logs", "investigator_beta")
+        _delegate(env, "check_dependencies", "investigator_alpha")
+        _delegate(env, "notify_stakeholders", "communicator")
+        _delegate(env, "root_cause_analysis", "senior_engineer")
+        _delegate(env, "deploy_hotfix", "deployer")
+        _delegate(env, "update_status_page", "communicator")
+        _delegate(env, "validate_fix", "senior_engineer")
+        _delegate(env, "monitor_recovery", "monitor")
+        # Synthesize immediately — NO patience waits
+        obs = _synthesize(env)
+        assert obs.done is True
+
+        log = _episode_store["hard"]
+        result = grade_hard(log)
+        assert result.breakdown["monitoring_patience"] == 0.0
+
+    def test_bad_episode_near_zero_score(self) -> None:
+        """Just waiting until timeout should score near 0."""
+        from server.graders import grade_hard
+
+        env, _ = _make_env("hard")
+        for _ in range(22):
+            _wait(env)
+        log = _episode_store["hard"]
+        result = grade_hard(log)
+        # Even doing nothing scores 0.3: error_classification(0.1) +
+        # capacity_discipline(0.1) + cost_efficiency(0.1) from "no harm done"
+        assert result.score <= 0.35
+        assert result.breakdown["completion"] == 0.0
+        assert result.breakdown["recovery"] == 0.0
+        assert result.breakdown["sla_compliance"] == 0.0
+        # All 9 keys should still be present
+        assert len(result.breakdown) == 9
+
+    def test_sla_penalty_when_delayed(self) -> None:
+        """Delaying root_cause past step 10 should incur SLA penalties."""
+        env, _ = _make_env("hard")
+        _delegate(env, "alert_triage", "triage_analyst")
+        # Wait until step 11+ without completing root_cause
+        for _ in range(11):
+            _wait(env)
+        # By now, step_count=12, root_cause not done → SLA penalties accrued
+        state = env.state
+        assert state.subtask_statuses["root_cause_analysis"] != "completed"
+        # Total reward should be negative due to SLA + unnecessary_wait penalties
+        assert env._total_reward < 0

tests/test_task_registry.py

@@ -39,18 +39,18 @@ class TestTaskRegistry:
 
     def test_list_tasks(self) -> None:
         tasks = list_tasks()
-        assert len(tasks) == 3
+        assert len(tasks) >= 3
         ids = {t.task_id for t in tasks}
-        assert ids == {"easy", "medium", "hard"}
+        assert {"easy", "medium", "hard"}.issubset(ids)
 
-    @pytest.mark.parametrize("task_id", ["easy", "medium", "hard"])
+    @pytest.mark.parametrize("task_id", ["easy", "medium", "hard", "expert"])
     def test_dag_is_valid(self, task_id: str) -> None:
         """Every task's subtask definitions form a valid DAG (no cycles)."""
         config = get_task(task_id)
         dag = DAGExecutor(config.subtask_definitions)
         assert dag is not None
 
-    @pytest.mark.parametrize("task_id", ["easy", "medium", "hard"])
+    @pytest.mark.parametrize("task_id", ["easy", "medium", "hard", "expert"])
     def test_all_subtask_types_have_capable_agent(self, task_id: str) -> None:
         """Every subtask type has at least one agent that can handle it."""
         config = get_task(task_id)