OpenEnv Specification: Simulated Organization (OrgSim)
Deadline: 8 April 11:59 PM
Team: Clawless (Mohan Singh — Lead, Kushagra Lakhwani)
Project Overview
Environment Name: org-sim
Type: Single-agent, multi-task RL Environment
Description: A hierarchical organization simulation where an AI agent must triage tasks, coordinate across teams via escalation, manage shared resources, and complete work before deadlines. Tests prioritization, resource reasoning, and sequential decision-making under real-world organizational constraints.
Target: Real-world organizational workflows (not games/toys)
Judging Rubric (from PDF)
| Criterion | Weight | What judges check |
|---|---|---|
| Real-world utility | 30% | Fills a real gap; would someone use this to train/evaluate agents? |
| Task & grader quality | 25% | 3+ tasks with difficulty range; graders deterministic; hard task challenges frontier models |
| Environment design | 20% | Clean state; well-typed obs/action; reward provides varying signal; sensible episode bounds |
| Code quality & spec compliance | 15% | openenv validate passes; docker builds; HF Space responds; baseline runs |
| Creativity & novelty | 10% | Domain not seen in OpenEnv before; interesting mechanics |
Disqualification conditions
- Environment does not deploy or respond
- Plagiarized / trivially modified existing environment
- Graders that always return the same score
- No baseline inference script
Judging phases
- Phase 1 – Automated Validation: HF Space deploys, OpenEnv spec passes, Docker builds, baseline reproduces, 3+ tasks with graders
- Phase 2 – Agentic Evaluation: Nemotron 3 Super (standard Open LLM) run against all environments; score variance check
- Phase 3 – Human Review: Meta + HuggingFace engineers review for utility, creativity, exploit checks
1. Architecture
Organization Structure
┌──────────────────┐
│ CEO / BOARD │
│ (Final Appeals) │
└────────┬─────────┘
│
┌────────▼─────────┐
│ ESCALATION PORTAL │
│ (Routes issues to │
│ appropriate team)│
└────────┬─────────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐
│ENGINEERING│ │ SALES │ │OPERATIONS │
│ TEAM │ │ TEAM │ │ TEAM │
├───────────┤ ├───────────┤ ├───────────┤
│Tech Lead │ │Sales Mgr │ │Ops Lead │
├───────────┤ ├───────────┤ ├───────────┤
│SWE x3 │ │Rep x3 │ │Analyst x3 │
└───────────┘ └───────────┘ └───────────┘
Action Space
| Action | Description | Payload keys |
|---|---|---|
REQUEST_TASK |
Request task assignment from team queue | (none) |
ACCEPT_TASK |
Explicitly accept a pending task | task_id |
COMPLETE_TASK |
Mark assigned task as done | task_id |
REQUEST_HELP |
Ask for help on active task (advances progress) | task_id |
PROVIDE_HELP |
Respond to a help request (advances task progress) | task_id |
ESCALATE |
Escalate a difficult/cross-team task | task_id |
REQUEST_RESOURCE |
Lock a shared resource | resource_id |
REPORT_STATUS |
No-op status report | (none) |
2. Project Structure
org-sim/ ← repo root
├── inference.py ← baseline inference script (MUST be here)
├── SPEC-orgsim.md
├── open.pdf
└── org_sim/ ← Python package
├── __init__.py
├── README.md ← HF Space landing page (has HF frontmatter)
├── client.py
├── models.py
├── openenv.yaml
├── pyproject.toml
└── server/
├── __init__.py
├── app.py
├── org_environment.py
├── Dockerfile ← built from repo root: docker build -f org_sim/server/Dockerfile .
└── requirements.txt
3. Three Named Tasks (Easy → Medium → Hard)
Each task is a named scenario selected via reset(task_id=<name>). The inference script iterates all three.
Task 1 — solo_bug_fix (Easy)
Scenario: One engineer, one high-priority bug. Deadline: 10 steps.
Optimal path: REQUEST_TASK → (REQUEST_HELP × 2) → COMPLETE_TASK in ~4 steps.
Why it's easy: Single team, no coordination, no resource dependency.
| Property | Value |
|---|---|
| Team | engineering |
| Tasks | 1 (fix_bug, priority=high, deadline=10, difficulty=1) |
| Resources required | none |
| Done condition | task terminal (completed/failed/escalated) |
Grader (grade("solo_bug_fix")):
completed within deadline → 0.50 + time_bonus(max 0.30) + priority_bonus(0.15) = up to 0.95
completed after deadline → 0.15 (partial credit — agent did complete it)
not completed / failed → 0.00
escalated appropriately → 0.20 (difficulty < 2, so escalation is never appropriate here → 0.0)
Score is always in [0.0, 0.95] — never the same two runs due to time bonus varying with step count.
Task 2 — cross_team_launch (Medium)
Scenario: Engineering builds a feature (deadline 15); Sales prepares a proposal (deadline 12). Agent controls only engineering. Must recognize sales task needs escalation.
Optimal path: REQUEST_TASK (eng) → REQUEST_HELP × 2 → COMPLETE_TASK (eng) → ESCALATE (sales task) → done.
Why it's medium: Two-team awareness required; partial cross-team actions penalized; escalation judgment needed.
| Property | Value |
|---|---|
| Teams | engineering + sales |
| Tasks | 2 (implement_feature high/15, prepare_proposal medium/12) |
| Resources required | none |
| Done condition | all tasks terminal |
Grader (grade("cross_team_launch")):
eng task completed on time → up to 0.60
sales task correctly escalated (difficulty≥2) → +0.25
both done → total up to 0.85
partial (only eng done) → up to 0.60
nothing done → 0.00
Score varies with timing of eng task completion and whether escalation was used correctly.
Task 3 — startup_crisis (Hard)
Scenario: Three concurrent tasks with a tight critical window. Task incident is critical with deadline=5. Task feature requires senior_engineer resource. Task client is cross-team (sales). Resource is consumed on REQUEST_RESOURCE — agent must grab it before deadline passes.
Optimal path (in 12 steps): REQUEST_RESOURCE(senior_engineer) → REQUEST_TASK → COMPLETE_TASK(incident, step≤5) → REQUEST_TASK(feature) → REQUEST_HELP × 2 → COMPLETE_TASK(feature) → ESCALATE(client).
Why it's hard: Tight window forces immediate resource grab; wrong ordering fails the incident; cross-team task requires correct escalation; even frontier LLMs often fail the resource-first sequencing.
| Property | Value |
|---|---|
| Teams | engineering + sales |
| Tasks | 3 (fix_bug critical/5, implement_feature high/20+resource, client_meeting critical/8) |
| Resources required | senior_engineer (locks out until released) |
| Done condition | all tasks terminal |
Grader (grade("startup_crisis")):
incident completed ≤ step 5 → 0.40
feature completed (with resource locked) → 0.30
client task correctly escalated → 0.20
resource secured before feature attempt → 0.10
max total → 1.00
incident missed (step > 5) → 0.00 for that component
premature escalation on incident → incident component = 0.00
resource not secured before feature attempt → feature component = 0.00
Score has genuine variance: incident timing, whether resource was grabbed, whether escalation was correct.
4. Type-Safe Models (models.py)
# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
# BSD-style license
"""
Data models for the OrgSim Environment.
Defines typed Action, Observation, and State for the organization simulation.
"""
from typing import Any, Literal
from pydantic import Field
from openenv.core.env_server.types import Action, Observation, State
class OrgAction(Action):
"""Action for organization simulation."""
action_type: Literal[
"REQUEST_TASK",
"ACCEPT_TASK",
"COMPLETE_TASK",
"REQUEST_HELP",
"PROVIDE_HELP",
"ESCALATE",
"REQUEST_RESOURCE",
"REPORT_STATUS",
] = Field(..., description="Type of action to perform")
target_id: str = Field(default="", description="Target agent/team ID")
payload: dict[str, Any] = Field(
default_factory=dict, description="Additional action parameters"
)
class OrgState(State):
"""State of the OrgSimEnvironment containing organization context."""
tasks: dict[str, dict[str, Any]] = Field(
default_factory=dict, description="All tasks in the organization"
)
team_members: dict[str, list[str]] = Field(
default_factory=dict, description="Team membership: team_id -> [agent_ids]"
)
resource_status: dict[str, bool] = Field(
default_factory=dict, description="Resource availability status"
)
class OrgObservation(Observation):
"""Observation returned by OrgSimEnvironment."""
my_agent_id: str = Field(..., description="Current agent ID")
my_team: str = Field(..., description="Current agent's team")
my_role: str = Field(default="member", description="Role: lead or member")
available_tasks: list[dict[str, Any]] = Field(
default_factory=list, description="Tasks available for assignment"
)
active_task: dict[str, Any] | None = Field(
default=None, description="Currently assigned task"
)
inbox: list[dict[str, Any]] = Field(
default_factory=list, description="Messages for this agent"
)
team_status: dict[str, Any] = Field(
default_factory=dict, description="Team workload and availability"
)
resources: dict[str, bool] = Field(
default_factory=dict, description="Available resources"
)
metrics: dict[str, Any] = Field(default_factory=dict, description="Episode metrics")
5. Environment (server/org_environment.py)
Key design decisions
reset(task_id, agent_id)— selects one of the 3 named scenarios.task_idcan be"solo_bug_fix","cross_team_launch", or"startup_crisis". Defaults to"cross_team_launch"._check_done()— checks if all tasks are in a terminal state (completed,failed,escalated), NOT just not-pending. This is the critical fix vs. the broken version that checkedstatus == "pending".grade_episode()— returns float 0.0–1.0 based on which tasks were completed, when, and how appropriately escalation/resources were used._get_team_for_agent()— checks bothleadandmembersso team leads are assigned correctly.
# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
# BSD-style license
"""
Organization Simulation Environment.
A multi-agent environment simulating hierarchical organization with
teams, escalation, and resource management.
"""
import uuid
from typing import Any
from openenv.core.env_server.interfaces import Environment
from ..models import OrgAction, OrgObservation, OrgState
TASK_SCENARIOS = {
"solo_bug_fix": {
"description": "Fix a high-priority bug in the engineering team",
"tasks": [
{
"id": "task_bug",
"team": "engineering",
"type": "fix_bug",
"priority": "high",
"deadline": 10,
"difficulty": 1,
}
],
},
"cross_team_launch": {
"description": "Engineering and Sales coordinate a product launch",
"tasks": [
{
"id": "task_eng",
"team": "engineering",
"type": "implement_feature",
"priority": "high",
"deadline": 15,
"difficulty": 2,
},
{
"id": "task_sales",
"team": "sales",
"type": "prepare_proposal",
"priority": "medium",
"deadline": 12,
"difficulty": 2,
},
],
},
"startup_crisis": {
"description": "Simultaneous incident + resource-gated feature + cross-team client meeting",
"tasks": [
{
"id": "task_incident",
"team": "engineering",
"type": "fix_bug",
"priority": "critical",
"deadline": 5,
"difficulty": 3,
},
{
"id": "task_feature",
"team": "engineering",
"type": "implement_feature",
"priority": "high",
"deadline": 20,
"difficulty": 2,
"requires_resource": "senior_engineer",
},
{
"id": "task_client",
"team": "sales",
"type": "client_meeting",
"priority": "critical",
"deadline": 8,
"difficulty": 2,
},
],
},
}
TASK_IDS = list(TASK_SCENARIOS.keys())
class OrgSimEnvironment(Environment):
"""Organization simulation environment."""
def __init__(self, max_steps: int = 50):
super().__init__()
self.max_steps = max_steps
self._current_task_id = "cross_team_launch"
self._state = OrgState(episode_id="", step_count=0)
self._current_agent = "eng_swe1"
self._tasks: dict[str, dict] = {}
self._messages: list[dict] = []
self._episode_resources: dict[str, bool] = {}
self._init_organization()
def _init_organization(self):
"""Initialize static organization structure."""
self.teams = {
"engineering": {
"lead": "eng_lead",
"members": ["eng_swe1", "eng_swe2", "eng_swe3"],
},
"sales": {
"lead": "sales_mgr",
"members": ["sales_rep1", "sales_rep2", "sales_rep3"],
},
"operations": {
"lead": "ops_lead",
"members": ["ops_analyst1", "ops_analyst2", "ops_analyst3"],
},
}
def reset(self, agent_id: str = "eng_swe1", task_id: str = "cross_team_launch", **kwargs: Any) -> OrgObservation:
"""Reset environment for new episode.
Args:
agent_id: The agent playing this episode.
task_id: One of 'solo_bug_fix', 'cross_team_launch', 'startup_crisis'.
"""
self._current_task_id = task_id if task_id in TASK_SCENARIOS else "cross_team_launch"
self._state.episode_id = str(uuid.uuid4())
self._state.step_count = 0
self._current_agent = agent_id
self._messages = []
scenario = TASK_SCENARIOS[self._current_task_id]
self._tasks = {
t["id"]: {
**t,
"status": "pending",
"assignee": None,
"progress": 0.0,
}
for t in scenario["tasks"]
}
# Reset shared resources per episode
self._episode_resources = {
"senior_engineer": True,
"cloud_instances": True,
"budget_approval": True,
}
return self._get_observation()
def step(self, action: OrgAction) -> OrgObservation:
"""Execute action and return observation."""
self._state.step_count += 1
reward = 0.0
if action.action_type == "REQUEST_TASK":
reward = self._handle_request_task(action)
elif action.action_type == "ACCEPT_TASK":
reward = self._handle_accept_task(action)
elif action.action_type == "COMPLETE_TASK":
reward = self._handle_complete_task(action)
elif action.action_type == "REQUEST_HELP":
reward = self._handle_request_help(action)
elif action.action_type == "PROVIDE_HELP":
reward = self._handle_provide_help(action)
elif action.action_type == "ESCALATE":
reward = self._handle_escalate(action)
elif action.action_type == "REQUEST_RESOURCE":
reward = self._handle_request_resource(action)
elif action.action_type == "REPORT_STATUS":
reward = 0.0
done = self._check_done()
return self._get_observation(reward=reward, done=done)
def _handle_request_task(self, action: OrgAction) -> float:
my_team = self._get_team_for_agent(self._current_agent)
available = [
t
for t in self._tasks.values()
if t["team"] == my_team
and t["status"] == "pending"
and t.get("assignee") is None
]
if available:
task = available[0]
# Block assignment if required resource is not held
required = task.get("requires_resource")
if required and self._episode_resources.get(required, True):
# Resource exists but hasn't been locked by this agent — still allow
# but note: resource should be locked before completing
pass
task["status"] = "in_progress"
task["assignee"] = self._current_agent
return 0.1
return -0.05 # No task available — slight penalty to discourage looping
def _handle_accept_task(self, action: OrgAction) -> float:
task_id = action.payload.get("task_id")
if not task_id or task_id not in self._tasks:
return -0.1
task = self._tasks[task_id]
if task["status"] != "pending":
return -0.1
task["status"] = "in_progress"
task["assignee"] = self._current_agent
return 0.1
def _handle_complete_task(self, action: OrgAction) -> float:
task_id = action.payload.get("task_id")
if not task_id or task_id not in self._tasks:
return -0.2
task = self._tasks[task_id]
if task.get("assignee") != self._current_agent:
return -0.1
if task["status"] != "in_progress":
return -0.1
# Block completion of resource-gated tasks if resource was never locked
required = task.get("requires_resource")
if required and self._episode_resources.get(required, True):
# Resource still available (never locked) — penalize and don't complete
return -0.2
deadline = task.get("deadline", 20)
time_taken = self._state.step_count
if time_taken <= deadline:
base_reward = 1.0
time_bonus = max(0.0, (deadline - time_taken) / deadline) * 0.3
priority_scores = {"critical": 0.3, "high": 0.2, "medium": 0.1, "low": 0.0}
priority_bonus = priority_scores.get(task.get("priority", "medium"), 0.1)
task["status"] = "completed"
task["completed_at"] = self._state.step_count
return base_reward + time_bonus + priority_bonus
else:
task["status"] = "failed"
return -0.3
def _handle_request_help(self, action: OrgAction) -> float:
task_id = action.payload.get("task_id")
if not task_id or task_id not in self._tasks:
return -0.1
task = self._tasks[task_id]
if task.get("assignee") != self._current_agent:
return -0.1
task["progress"] = min(1.0, task.get("progress", 0.0) + 0.2)
self._messages.append({
"from": self._current_agent,
"to": self._get_team_lead(self._get_team_for_agent(self._current_agent)),
"type": "help_request",
"task_id": task_id,
})
return 0.1
def _handle_provide_help(self, action: OrgAction) -> float:
task_id = action.payload.get("task_id")
if not task_id or task_id not in self._tasks:
return -0.1
task = self._tasks[task_id]
task["progress"] = min(1.0, task.get("progress", 0.0) + 0.3)
return 0.2
def _handle_escalate(self, action: OrgAction) -> float:
task_id = action.payload.get("task_id")
if not task_id or task_id not in self._tasks:
return -0.2
task = self._tasks[task_id]
if task["status"] in ("completed", "failed", "escalated"):
return -0.1 # Already terminal
# Appropriate: difficulty >= 2, progress < 0.5, or cross-team task
my_team = self._get_team_for_agent(self._current_agent)
is_cross_team = task["team"] != my_team
is_hard_stuck = task.get("difficulty", 1) >= 2 and task.get("progress", 0.0) < 0.5
if is_cross_team or is_hard_stuck:
task["status"] = "escalated"
task["escalated_at"] = self._state.step_count
return 0.3
else:
return -0.2 # Premature / unnecessary escalation
def _handle_request_resource(self, action: OrgAction) -> float:
resource_id = action.payload.get("resource_id")
if not resource_id or resource_id not in self._episode_resources:
return -0.2
if self._episode_resources[resource_id]:
self._episode_resources[resource_id] = False # Lock it
return 0.2
return -0.1 # Already locked
def _check_done(self) -> bool:
"""Episode ends when all tasks are in a terminal state OR max_steps reached.
IMPORTANT: checks for terminal states (completed/failed/escalated), NOT just
non-pending. A task in 'in_progress' is NOT done — episode must continue.
"""
if self._state.step_count >= self.max_steps:
return True
active = sum(
1 for t in self._tasks.values()
if t["status"] in ("pending", "in_progress")
)
return active == 0
def grade_episode(self) -> float:
"""Grade the completed episode. Returns 0.0–1.0.
Called after done=True. Score varies with timing, escalation quality,
resource management. Never returns the same value for different episode
trajectories (time bonus alone creates variance).
"""
task_id = self._current_task_id
if task_id == "solo_bug_fix":
return self._grade_solo_bug_fix()
elif task_id == "cross_team_launch":
return self._grade_cross_team_launch()
elif task_id == "startup_crisis":
return self._grade_startup_crisis()
return 0.0
def _grade_solo_bug_fix(self) -> float:
task = self._tasks.get("task_bug", {})
if task.get("status") == "completed":
completed_at = task.get("completed_at", self._state.step_count)
deadline = task.get("deadline", 10)
time_bonus = max(0.0, (deadline - completed_at) / deadline) * 0.30
return min(1.0, 0.50 + time_bonus + 0.15) # 0.50 base + 0.30 time + 0.15 priority
elif task.get("status") == "failed":
return 0.15 # partial credit for attempting
return 0.0
def _grade_cross_team_launch(self) -> float:
score = 0.0
eng = self._tasks.get("task_eng", {})
sales = self._tasks.get("task_sales", {})
if eng.get("status") == "completed":
completed_at = eng.get("completed_at", self._state.step_count)
deadline = eng.get("deadline", 15)
time_bonus = max(0.0, (deadline - completed_at) / deadline) * 0.20
score += 0.40 + time_bonus # up to 0.60
if sales.get("status") == "escalated":
score += 0.25 # correct action for cross-team task
elif sales.get("status") == "completed":
score += 0.20 # completed somehow (less expected but fine)
return min(1.0, score)
def _grade_startup_crisis(self) -> float:
score = 0.0
incident = self._tasks.get("task_incident", {})
feature = self._tasks.get("task_feature", {})
client = self._tasks.get("task_client", {})
# Incident: must complete by step 5
if incident.get("status") == "completed":
completed_at = incident.get("completed_at", self._state.step_count)
if completed_at <= 5:
score += 0.40
else:
score += 0.10 # completed but late
# Feature: must have resource locked (resource False = locked)
resource_locked = not self._episode_resources.get("senior_engineer", True)
if feature.get("status") == "completed" and resource_locked:
score += 0.30
elif feature.get("status") == "completed":
score += 0.10 # completed without resource (no credit for the resource part)
# Client: correct cross-team escalation
if client.get("status") == "escalated":
score += 0.20
elif client.get("status") == "completed":
score += 0.10
# Resource management bonus
if resource_locked:
score += 0.10
return min(1.0, score)
def _get_observation(self, reward: float = 0.0, done: bool = False) -> OrgObservation:
my_team = self._get_team_for_agent(self._current_agent)
available = [
t for t in self._tasks.values()
if t["team"] == my_team
and t["status"] == "pending"
and t.get("assignee") is None
]
active = next(
(t for t in self._tasks.values()
if t.get("assignee") == self._current_agent and t["status"] == "in_progress"),
None,
)
team_status = {
team_name: {
"busy": sum(1 for t in self._tasks.values() if t["team"] == team_name and t.get("assignee")),
"total": len(team_data["members"]),
}
for team_name, team_data in self.teams.items()
}
return OrgObservation(
my_agent_id=self._current_agent,
my_team=my_team,
my_role="lead" if self._current_agent in (
self.teams[my_team]["lead"],) else "member",
available_tasks=available,
active_task=active,
inbox=[m for m in self._messages if m.get("to") == self._current_agent],
team_status=team_status,
resources=self._episode_resources.copy(),
metrics={
"tasks_completed": sum(1 for t in self._tasks.values() if t["status"] == "completed"),
"tasks_failed": sum(1 for t in self._tasks.values() if t["status"] == "failed"),
"tasks_escalated": sum(1 for t in self._tasks.values() if t["status"] == "escalated"),
"current_task_id": self._current_task_id,
"step_count": self._state.step_count,
},
done=done,
reward=reward,
metadata={"step_count": self._state.step_count, "task_id": self._current_task_id},
)
def _get_team_for_agent(self, agent_id: str) -> str:
"""Get team for agent — checks both lead and members."""
for team_name, team_data in self.teams.items():
if agent_id == team_data["lead"] or agent_id in team_data["members"]:
return team_name
return "engineering"
def _get_team_lead(self, team_name: str) -> str:
return self.teams.get(team_name, {}).get("lead", "")
@property
def state(self) -> OrgState:
self._state.tasks = self._tasks.copy()
self._state.team_members = {k: v["members"] for k, v in self.teams.items()}
self._state.resource_status = self._episode_resources.copy()
return self._state
6. Server App (server/app.py)
Exposes /grade endpoint in addition to standard reset/step/state.
# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
# BSD-style license
"""FastAPI application for the OrgSim Environment."""
import os
from fastapi import FastAPI
from openenv.core.env_server.http_server import create_app
from ..models import OrgAction, OrgObservation, OrgState
from .org_environment import OrgSimEnvironment, TASK_IDS
_env: OrgSimEnvironment | None = None
def get_env() -> OrgSimEnvironment:
global _env
if _env is None:
max_steps = int(os.getenv("ORGSIM_MAX_STEPS", "50"))
_env = OrgSimEnvironment(max_steps=max_steps)
return _env
def create_orgsim_app() -> FastAPI:
"""Create the OrgSim FastAPI app with /grade and /tasks endpoints."""
max_steps = int(os.getenv("ORGSIM_MAX_STEPS", "50"))
env_factory = lambda: OrgSimEnvironment(max_steps=max_steps)
app = create_app(
env_factory,
OrgAction,
OrgObservation,
env_name="org_sim",
)
@app.get("/tasks")
def list_tasks():
"""List available task IDs for enumeration by automated validator."""
return {"tasks": TASK_IDS}
@app.get("/grade")
def grade_episode():
"""Return current episode grade (call after done=True)."""
env = get_env()
score = env.grade_episode()
return {"score": score, "task_id": env._current_task_id}
return app
app = create_orgsim_app()
def main():
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
main()
7. openenv.yaml
spec_version: 1
name: org_sim
type: space
runtime: fastapi
app: server.app:app
port: 8000
tasks:
- id: solo_bug_fix
difficulty: easy
- id: cross_team_launch
difficulty: medium
- id: startup_crisis
difficulty: hard
8. pyproject.toml
[project]
name = "org-sim"
version = "0.1.0"
description = "Organization Simulation Environment for RL training"
requires-python = ">=3.10"
dependencies = [
"openenv-core>=0.2.0",
"fastapi>=0.104.0",
"uvicorn>=0.24.0",
"pydantic>=2.0.0",
"openai>=1.0.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"httpx>=0.25.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["."]
9. server/Dockerfile
IMPORTANT: Built from the repo root with:
docker build -t org-sim:latest -f org_sim/server/Dockerfile .
FROM python:3.11-slim
WORKDIR /app
# Copy requirements from correct path within build context (repo root)
COPY org_sim/server/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy full repo so org_sim package is importable
COPY . .
EXPOSE 8000
# Module path from repo root
CMD ["python", "-m", "uvicorn", "org_sim.server.app:app", "--host", "0.0.0.0", "--port", "8000"]
10. server/requirements.txt
openenv-core>=0.2.0
fastapi>=0.104.0
uvicorn>=0.24.0
pydantic>=2.0.0
httpx>=0.25.0
openai>=1.0.0
11. __init__.py
# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
# BSD-style license
"""OrgSim - Organization Simulation Environment."""
from .client import OrgSimEnv
from .models import OrgAction, OrgObservation, OrgState
__all__ = ["OrgSimEnv", "OrgAction", "OrgObservation", "OrgState"]
12. inference.py (root of repo)
Critical: Exact log format from the PDF spec
[START] task={task} env={env} model={model}
[STEP] step={step} action={action} reward={reward:.2f} done={done_val} error={error_val}
[END] success={str(success).lower()} steps={steps} score={score:.3f} rewards={rewards_str}
Any deviation (JSON, extra fields, wrong order) results in incorrect automated scoring — disqualification.
#!/usr/bin/env python3
"""
Baseline inference script for OrgSim environment.
Runs all 3 tasks (solo_bug_fix, cross_team_launch, startup_crisis) against the
OrgSim environment using an LLM agent via the OpenAI client.
Required env vars:
API_BASE_URL - LLM API endpoint
MODEL_NAME - Model identifier
HF_TOKEN - HuggingFace / API key (used as OpenAI api_key)
Optional:
ORGSIM_ENV_URL - Environment base URL (default: http://localhost:8000)
"""
import os
import json
import sys
import textwrap
from typing import Optional
try:
from openai import OpenAI
except ImportError:
print("ERROR: openai package not installed", file=sys.stderr)
sys.exit(1)
from org_sim import OrgSimEnv, OrgAction
API_BASE_URL = os.getenv("API_BASE_URL")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o")
HF_TOKEN = os.getenv("HF_TOKEN")
if not API_BASE_URL or not HF_TOKEN:
print("ERROR: API_BASE_URL and HF_TOKEN must be set", file=sys.stderr)
sys.exit(1)
client = OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN)
TASKS = ["solo_bug_fix", "cross_team_launch", "startup_crisis"]
ENV_NAME = "org_sim"
# --------------------------------------------------------------------------- #
# Exact log format — do not change field names, ordering, or format strings #
# --------------------------------------------------------------------------- #
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 = error if error else "null"
done_val = 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 = ",".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,
)
# --------------------------------------------------------------------------- #
# LLM decision logic #
# --------------------------------------------------------------------------- #
def build_user_prompt(step: int, last_echoed: str, last_reward: float, history: list[str]) -> str:
history_block = "\n".join(history[-4:]) if history else "None"
return textwrap.dedent(f"""
Step: {step}
Last echoed message: {last_echoed!r}
Last reward: {last_reward:.2f}
Previous steps:
{history_block}
Send your next message.
""").strip()
def get_model_action(step: int, obs, last_reward: float, history: list[str]) -> OrgAction:
"""Use LLM to decide next action."""
system_prompt = textwrap.dedent(f"""
You are an agent in an organization simulation (OrgSim).
Agent ID: {obs.my_agent_id}
Team: {obs.my_team}
Role: {obs.my_role}
Available tasks: {obs.available_tasks}
Active task: {obs.active_task}
Inbox: {obs.inbox}
Team status: {obs.team_status}
Resources: {obs.resources}
Metrics: {obs.metrics}
Available actions:
- REQUEST_TASK: Get next task from your team queue (no payload needed)
- ACCEPT_TASK: payload={{"task_id": "<id>"}}
- COMPLETE_TASK: payload={{"task_id": "<id>"}} — only when you have an active task
- REQUEST_HELP: payload={{"task_id": "<id>"}} — advances progress on your task
- PROVIDE_HELP: payload={{"task_id": "<id>"}}
- ESCALATE: payload={{"task_id": "<id>"}} — for cross-team or stuck tasks
- REQUEST_RESOURCE: payload={{"resource_id": "<id>"}} — lock senior_engineer before feature tasks
- REPORT_STATUS: (no payload)
Strategy hints:
1. For startup_crisis: REQUEST_RESOURCE(senior_engineer) FIRST, then tackle the critical incident.
2. For cross-team tasks you can't do yourself, ESCALATE them.
3. Use REQUEST_HELP to build progress before attempting COMPLETE_TASK.
Respond ONLY with valid JSON: {{"action_type": "...", "target_id": "...", "payload": {{}}}}
""").strip()
user_prompt = build_user_prompt(step, obs.my_agent_id, last_reward, history)
try:
response = client.chat.completions.create(
model=MODEL_NAME,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
temperature=0.3,
)
content = response.choices[0].message.content.strip()
# Strip markdown code blocks if present
if content.startswith("```"):
content = content.split("```")[1]
if content.startswith("json"):
content = content[4:]
action_data = json.loads(content)
return OrgAction(
action_type=action_data.get("action_type", "REQUEST_TASK"),
target_id=action_data.get("target_id", ""),
payload=action_data.get("payload", {}),
)
except Exception:
return OrgAction(action_type="REQUEST_TASK", target_id="", payload={})
# --------------------------------------------------------------------------- #
# Main loop #
# --------------------------------------------------------------------------- #
def run_task(env_url: str, task_id: str) -> tuple[bool, int, float, list[float]]:
"""Run one episode for a given task. Returns (success, steps, score, rewards)."""
rewards: list[float] = []
with OrgSimEnv(base_url=env_url).sync() as env:
result = env.reset(task_id=task_id)
step_count = 0
history: list[str] = []
last_reward = 0.0
error_msg = None
while not result.done:
step_count += 1
obs = result.observation
try:
action = get_model_action(step_count, obs, last_reward, history)
error_msg = None
except Exception as e:
action = OrgAction(action_type="REQUEST_TASK", target_id="", payload={})
error_msg = str(e)
try:
result = env.step(action)
last_reward = result.reward
rewards.append(result.reward)
history.append(f"step={step_count} action={action.action_type} reward={result.reward:.2f}")
except Exception as e:
error_msg = str(e)
last_reward = 0.0
rewards.append(0.0)
log_step(
step=step_count,
action=action.action_type,
reward=last_reward,
done=result.done,
error=error_msg,
)
# Get final grade from /grade endpoint
try:
import httpx
resp = httpx.get(f"{env_url}/grade", timeout=10.0)
score = resp.json().get("score", 0.0)
except Exception:
# Fallback: compute from metrics
metrics = result.observation.metrics if result else {}
completed = metrics.get("tasks_completed", 0)
total = len(obs.available_tasks) + completed + metrics.get("tasks_failed", 0)
score = completed / max(1, total)
success = score > 0.0
return success, step_count, score, rewards
def main():
env_url = os.getenv("ORGSIM_ENV_URL", "http://localhost:8000")
for task_id in TASKS:
log_start(task=task_id, env=ENV_NAME, model=MODEL_NAME)
success, steps, score, rewards = run_task(env_url, task_id)
log_end(success=success, steps=steps, score=score, rewards=rewards)
if __name__ == "__main__":
main()
13. Reward Function Summary
| Component | Reward |
|---|---|
| Task assigned (REQUEST_TASK, found task) | +0.10 |
| Task assigned (no task available) | -0.05 |
| COMPLETE_TASK within deadline | +1.0 base + time_bonus(≤0.3) + priority_bonus(≤0.3) |
| COMPLETE_TASK after deadline | -0.30 |
| COMPLETE_TASK without locking required resource | -0.20 |
| REQUEST_HELP on own task | +0.10 (advances progress) |
| PROVIDE_HELP | +0.20 |
| ESCALATE (appropriate — cross-team or hard+stuck) | +0.30 |
| ESCALATE (premature / already terminal) | -0.20 |
| REQUEST_RESOURCE (available) | +0.20 |
| REQUEST_RESOURCE (already locked) | -0.10 |
| Invalid action (bad task_id, wrong assignee, etc.) | -0.10 to -0.20 |
14. README.md (org_sim/README.md)
The README must include baseline scores section (required by judging rubric):
---
title: OrgSim
emoji: 🏢
colorFrom: blue
colorTo: purple
sdk: docker
pinned: false
app_port: 8000
tags: openenv,multi-agent,organization,simulation
---
# OrgSim — Organization Simulation Environment
A real-world organizational workflow environment for RL agent training. Agents must triage tasks,
manage shared resources, coordinate across teams via escalation, and complete work before deadlines.
## Environment Description
OrgSim simulates a 3-team startup (Engineering, Sales, Operations) where an AI agent acts as a
team member. Unlike game environments, OrgSim tests skills directly transferable to AI work assistants:
prioritization under time pressure, resource contention, cross-functional escalation judgment.
## Action Space
| Action | Description | Required payload |
|--------|-------------|-----------------|
| REQUEST_TASK | Pull next task from team queue | none |
| ACCEPT_TASK | Accept a specific task | task_id |
| COMPLETE_TASK | Mark task done | task_id |
| REQUEST_HELP | Ask for help (advances progress +0.2) | task_id |
| PROVIDE_HELP | Respond to help request (+0.3 progress) | task_id |
| ESCALATE | Escalate cross-team or stuck task | task_id |
| REQUEST_RESOURCE | Lock shared resource | resource_id |
| REPORT_STATUS | No-op | none |
## Observation Space
| Field | Type | Description |
|-------|------|-------------|
| my_agent_id | str | Agent's ID |
| my_team | str | Team (engineering/sales/operations) |
| my_role | str | lead or member |
| available_tasks | list[dict] | Unassigned tasks in my team |
| active_task | dict\|None | Currently assigned task |
| inbox | list[dict] | Messages addressed to me |
| team_status | dict | Per-team busy/total counts |
| resources | dict[str, bool] | Resource availability |
| metrics | dict | Episode metrics (completed/failed/escalated counts) |
## Tasks (Easy → Medium → Hard)
| Task ID | Difficulty | Description | Optimal steps |
|---------|------------|-------------|---------------|
| solo_bug_fix | Easy | Fix one high-priority bug before deadline | ~4 |
| cross_team_launch | Medium | Complete eng feature + escalate sales task | ~8 |
| startup_crisis | Hard | Triage critical incident + resource-gated feature + cross-team client | ~12 |
## Baseline Scores
Scores produced by GPT-4o baseline agent (inference.py):
| Task | Score | Steps |
|------|-------|-------|
| solo_bug_fix | ~0.75 | 4–6 |
| cross_team_launch | ~0.55 | 7–10 |
| startup_crisis | ~0.30 | 10–15 |
## Setup
### Local
```bash
pip install -e ./org_sim
python -m org_sim.server.app
Docker
docker build -t org-sim:latest -f org_sim/server/Dockerfile .
docker run -p 8000:8000 \
-e API_BASE_URL=<endpoint> \
-e MODEL_NAME=<model> \
-e HF_TOKEN=<token> \
org-sim:latest
Run Inference
export API_BASE_URL=https://...
export MODEL_NAME=gpt-4o
export HF_TOKEN=hf_...
export ORGSIM_ENV_URL=http://localhost:8000
python inference.py
Environment Variables
| Variable | Required | Default | Description |
|---|---|---|---|
| API_BASE_URL | Yes | — | LLM API endpoint |
| MODEL_NAME | Yes | gpt-4o | Model identifier |
| HF_TOKEN | Yes | — | HuggingFace / API key |
| ORGSIM_ENV_URL | No | http://localhost:8000 | Environment server URL |
| ORGSIM_MAX_STEPS | No | 50 | Max steps per episode |
---
## 15. Validation Checklist
### Phase 1 (Automated — all must pass or disqualified)
- [ ] HF Space URL returns HTTP 200
- [ ] HF Space responds to `POST /reset`
- [ ] `openenv validate` passes against `openenv.yaml`
- [ ] Typed models: `OrgAction`, `OrgObservation`, `OrgState`
- [ ] `step()` → returns observation, reward, done, info
- [ ] `reset()` → returns initial observation (clean state)
- [ ] `state()` / `GET /state` → returns current state
- [ ] `GET /tasks` → enumerates 3 task IDs
- [ ] `GET /grade` → returns score in 0.0–1.0 after episode
- [ ] `docker build -f org_sim/server/Dockerfile . && docker run` works
- [ ] `inference.py` at repo root
- [ ] `inference.py` runs without error on all 3 tasks
- [ ] Log format exactly: `[START] task=... env=... model=...` / `[STEP] step=... action=... reward=... done=... error=...` / `[END] success=... steps=... score=... rewards=...`
- [ ] Scores produced in 0.0–1.0 range for all 3 tasks
- [ ] Graders do NOT always return the same score (time bonus ensures variance)
- [ ] Runtime < 20 min on 2 vCPU / 8GB RAM
### Phase 2 (Agentic — scored)
- [ ] Nemotron 3 Super can parse observations and take valid actions
- [ ] Environment handles unexpected/invalid action payloads gracefully (returns -0.1, does not crash)
- [ ] Score variance across multiple runs (different step counts → different time bonuses)
- [ ] Hard task genuinely difficult for frontier models (resource ordering, tight deadline)
### Known fixes vs. original implementation
| Issue | Status |
|-------|--------|
| `_check_done()` checked `pending==0` instead of all-terminal | Fixed in spec |
| Log format was JSON dicts instead of key=value | Fixed in spec |
| `inference.py` ran only 1 task, not all 3 | Fixed in spec |
| Dockerfile `COPY requirements.txt` path wrong | Fixed in spec |
| Dockerfile CMD used wrong module path | Fixed in spec |
| `openai` missing from requirements | Fixed in spec |
| `_get_team_for_agent()` didn't check `lead` key | Fixed in spec |
| No `/tasks` or `/grade` endpoints | Fixed in spec |
| `grade_task()` not reachable via HTTP | Fixed in spec |
| No baseline scores in README | Fixed in spec |