""" app.py — Executive Assistant OpenEnv Environment FastAPI server + environment logic for email and calendar management. """ import sys import os from pathlib import Path # Ensure server/ directory is on the path sys.path.insert(0, str(Path(__file__).parent)) from fastapi import FastAPI, HTTPException from models import AssistantAction, AssistantObservation, AssistantState from fastapi.responses import HTMLResponse from typing import Optional import statistics # Import scoring functions from data.py (teammate will implement these) from data import ( generate_scenario, compute_email_quality, check_scheduling_correctness, compute_conflict_resolution, apply_penalties, TASK_DEFINITIONS, ) # ============================================================ # THE ENVIRONMENT CLASS # ============================================================ class ExecAssistEnv: def __init__(self): self.current_scenario = None self.calendar_state = None self.episode_done = False self.steps_taken = 0 self.total_score = 0.0 self.current_task = None self.seed = 42 def reset(self, task: str = "easy"): """Start a new episode.""" if task not in TASK_DEFINITIONS: raise ValueError(f"Unknown task: {task}. Choose from: easy, medium, hard") self.current_task = task self.episode_done = False self.steps_taken = 0 self.total_score = 0.0 # Generate scenario (teammate implements this in data.py) self.current_scenario = generate_scenario(task, seed=self.seed) return { "observation": self._build_observation(), "reward": 0.0, "done": False, "info": { "task": task, "scenario_id": self.current_scenario.get("id", "unknown"), } } def step(self, action: dict): """Agent submits action — environment scores it.""" if self.episode_done: return { "observation": {"message": "Episode is done. Call /reset to start again."}, "reward": 0.0, "done": True, "info": {"total_score": self.total_score} } # Parse action try: assistant_action = AssistantAction(**action) except Exception as e: return { "observation": {"message": f"Invalid action format: {str(e)}"}, "reward": -0.5, "done": False, "info": {"error": "invalid_action_format"} } # Validate basic action structure if not assistant_action.email_reply or len(assistant_action.email_reply.strip()) == 0: return { "observation": {"message": "Empty email reply. Penalty applied."}, "reward": -0.3, "done": False, "info": {"error": "empty_email_reply"} } if assistant_action.calendar_action not in ["book", "propose_alternatives", "reschedule", "decline"]: return { "observation": {"message": f"Invalid calendar_action: {assistant_action.calendar_action}"}, "reward": -0.2, "done": False, "info": {"error": "invalid_calendar_action"} } # Compute rewards using teammate's functions email_score = compute_email_quality( assistant_action.email_reply, self.current_scenario ) # Convert meeting_details to dict if it exists meeting_details_dict = assistant_action.meeting_details.dict() if assistant_action.meeting_details else None scheduling_result = check_scheduling_correctness( meeting_details_dict, self.current_scenario ) conflict_score = compute_conflict_resolution( assistant_action.dict(), # ← Add .dict() here self.current_scenario ) penalty = apply_penalties(assistant_action.dict(), self.current_scenario) # Combine scores based on task difficulty task_def = TASK_DEFINITIONS[self.current_task] weights = task_def["reward_weights"] total_reward = ( weights["email"] * email_score + weights["scheduling"] * scheduling_result["score"] + weights["conflict"] * conflict_score ) total_reward = max(0.0, min(1.0, total_reward - penalty)) self.total_score = total_reward self.episode_done = True self.steps_taken += 1 return { "observation": self._build_completion_message(assistant_action, total_reward), "reward": round(total_reward, 4), "done": True, "info": { "email_score": round(email_score, 4), "scheduling_score": round(scheduling_result["score"], 4), "conflict_score": round(conflict_score, 4), "penalty": round(penalty, 4), "scheduling_checks": scheduling_result.get("checks", {}), } } def _build_observation(self) -> dict: """Build what the agent sees.""" scenario = self.current_scenario task_def = TASK_DEFINITIONS[self.current_task] obs = { "task": self.current_task, "description": task_def["description"], "emails": scenario["emails"], "calendar": scenario["calendar"], "contacts": scenario.get("contacts", {}), "action_required": task_def["action_required"], } return obs def _build_completion_message(self, action: AssistantAction, score: float) -> dict: """Build feedback message after step.""" if score >= 0.9: message = f"Excellent work! Score: {score:.2f}" elif score >= 0.7: message = f"Good response. Score: {score:.2f}" elif score >= 0.5: message = f"Acceptable. Score: {score:.2f}" else: message = f"Needs improvement. Score: {score:.2f}" return { "message": message, "email_sent": action.email_reply[:100] + "..." if len(action.email_reply) > 100 else action.email_reply, "calendar_action": action.calendar_action, } def state(self): """Return current state.""" return { "current_task": self.current_task, "emails_pending": len(self.current_scenario.get("emails", [])) if self.current_scenario else 0, "episode_done": self.episode_done, "steps_taken": self.steps_taken, "total_score": self.total_score, } # ============================================================ # FASTAPI SERVER # ============================================================ app = FastAPI( title="ExecAssist Environment", description=( "An OpenEnv environment where AI agents learn to manage email and calendar " "for a busy executive. Agents must draft professional replies, schedule meetings, " "and resolve conflicts." ), version="1.0.0" ) @app.get("/", response_class=HTMLResponse, include_in_schema=False) async def root(): """Friendly landing page for the bare URL.""" return """
An AI agent reads incoming emails, checks the calendar, drafts professional replies, and books meetings, that too without double-booking or breaking working hours. Trained with TRL GRPO on Qwen2.5-0.5B.
| Task | Baseline | Trained (GRPO) | Δ |
|---|---|---|---|
| Easy | 0.345 | 0.995 | +188% |
| Medium | 0.227 | 0.745 | +228% |
| Hard | 0.249 | 0.737 | +196% |
Built for the OpenEnv Hackathon · April 2026 · by @DevanshuDon
""" @app.get("/blog", response_class=HTMLResponse, include_in_schema=False) async def blog(): """Blog post about the hackathon submission.""" return """Let me start with the moment I thought I'd lost the hackathon.
It's around 3pm on day two. I've been training for an hour and a half on a free Colab T4, and the cell finally finishes. I run the eval. The plot pops up. Baseline bars in red, trained bars in green. Easy task: 0.493 to 0.200. Medium: 0.348 to 0.200. Hard: 0.331 to 0.186.
The trained model was worse. Identically worse on every task, by exactly the same number every time. I sat there for a full minute trying to figure out if I was reading the chart wrong.
Turns out I wasn't. This is what GRPO collapse looks like, and now I had a textbook example of it. The model had given up exploring, found a single response that scored exactly 0.2 against my reward function no matter what input you gave it, and locked itself in. All my training was technically optimization. It was just optimization toward "say the dumbest possible thing every time and never deviate."
Here's how I dug out of it.
I built ExecAssist, an OpenEnv environment that simulates an executive's morning inbox. The agent gets emails (sender, subject, body, priority), a calendar (existing meetings, working hours), and contacts. It has to spit out a JSON action: an email reply, a calendar action like book or propose_alternatives, and meeting details if it's booking something.
Three difficulty tiers:
Reward is a weighted blend of three independent graders. Email quality scores politeness markers, greeting and closing, sufficient detail. Scheduling correctness checks for double-booking, working hours, sensible duration. Conflict resolution looks for whether the agent recognized a clash and proposed real alternatives. Then four anti-hacking penalties on top: short emails, missing meeting details, generic phrasing, overly long responses.
I went with multiple independent graders because I'd read this line in the hackathon guide that stuck with me: "if you only have a single reward signal, it is easier for the model to hack it." Building it the right way from day one felt like the obvious move. I would later be very glad I did.
The first config looked reasonable on paper:
GRPOConfig(
learning_rate=5e-6,
num_train_epochs=1,
num_generations=4,
# no beta term
)
Looking at the training log afterward, I could see exactly what happened. Reward bounced between 0.0 and 0.4 for the first seven steps, peaked at 0.397, then crashed and pinned itself at 0.14 for the next 38 steps. The model had found a safe floor and stopped trying anything that might score higher because variance was punishing it.
The diagnosis came down to three problems. No KL penalty meant nothing was anchoring the trained policy to the base model, so it was free to drift to any degenerate local minimum. The learning rate was too aggressive, which compounded the drift. And one epoch wasn't enough runway to recover even if it wanted to.
The fix was three changes:
GRPOConfig(
learning_rate=1e-6, # 5x slower
num_train_epochs=3, # 3x longer
num_generations=8, # more variety per step
beta=0.1, # KL penalty, the critical one
)
I also dropped in a "reload clean model" cell before training started. The previous run had corrupted the weights, and I didn't want to start the next attempt from a broken policy. Then I hit Run All and walked away for 90 minutes.
Came back to find the cell still chugging along. 218 of 270 steps done. Trained weights were already in memory, so I ran the eval cells anyway and held my breath while the bars rendered.
Easy task: 0.345 to 0.995.
Medium: 0.227 to 0.745.
Hard: 0.249 to 0.737.
I made a noise. I'm not going to describe what kind.
Nine out of ten samples on the easy task hit a perfect 1.0. The model wasn't just lucky on those runs, it had learned the structure of the task. You could see it in the variance. Baseline scores ranged from 0.0 to 0.65 on the same prompt depending on how the dice rolled. Trained scores were tight: 0.95 to 1.0 on easy, 0.68 to 0.82 on medium, 0.65 to 0.80 on hard.
The training curve told the same story. First quartile mean reward: 0.390. Last quartile: 0.648. A 66% lift during training itself, on top of the much bigger gap between trained and untrained at evaluation time.
Because the reward had multiple independent components instead of one big scalar, I could see exactly how the model tried to game each one during training. Going through the early-step rollouts in order:
meeting_details block. The -0.40 penalty made that a non-strategy fast.Most submissions claim they have anti-hacking penalties. Showing the penalties firing during real training, on a real curve, is the rare part. It's the difference between saying a multi-grader rubric works and demonstrating that it does.
One sanity check from earlier in the day stuck with me. Same three tasks, same scoring, but evaluated against an untuned Nemotron 120B running through OpenRouter via the standard inference.py baseline. It averaged 0.337 across the three tasks.
After 90 minutes of GRPO, a model 240 times smaller is averaging 0.83 on the same environment. Free Colab T4. Zero-dollar cloud bill.
That's the point of training-environment design. A reward signal that's hard to game, a model that's allowed to actually train against it, and you get a result that beats a frontier model on its native task.
I think there's a research-shaped argument hiding in here. Frontier LLMs are notoriously bad at structured calendar reasoning. Try asking any production agent to find a 30-minute slot that doesn't conflict with your standups. ExecAssist isolates that specific failure mode into a tractable RL target. The result suggests that for a class of structured personal-task workflows, task-specific RL on a small model is a real alternative to scaling up. That feels like a workshop paper, maybe.
POST /reset?task=easy, then POST /step with your action. Swagger does most of the work.python inference.py reproduces the untrained scores around 0.32 average.The environment is hard in interesting ways. Try to break it. I'd be curious what the model learns to game next.
""" env = ExecAssistEnv() @app.post("/reset") def reset(task: str = "easy"): return env.reset(task) @app.post("/step") def step(action: AssistantAction): return env.step(action.dict()) @app.get("/state") def state(): return env.state() @app.get("/tasks") def tasks(): return { task_name: { "description": td["description"], "action_required": td["action_required"], "reward_weights": td["reward_weights"], } for task_name, td in TASK_DEFINITIONS.items() } @app.get("/health") def health(): return {"status": "healthy"} # ============================================================ # OPENENV REQUIRED ENDPOINTS # ============================================================ @app.get("/metadata") def metadata(): """Return environment name and description.""" return { "name": "exec-assist", "description": ( "Executive Assistant environment where AI agents learn to manage email " "and calendar for busy professionals. Agents must balance professionalism, " "scheduling correctness, and conflict resolution." ), "version": "1.0.0", "author": "Devanshu(Team PsPredator)", "tasks": ["easy", "medium", "hard"], } @app.get("/schema") def schema(): """Return action, observation, and state schemas.""" return { "action": { "type": "object", "properties": { "email_reply": {"type": "string"}, "calendar_action": {"type": "string", "enum": ["book", "propose_alternatives", "reschedule", "decline"]}, "meeting_details": {"type": "object"}, }, "required": ["email_reply", "calendar_action"], }, "observation": { "type": "object", "properties": { "task": {"type": "string"}, "emails": {"type": "array"}, "calendar": {"type": "object"}, "contacts": {"type": "object"}, }, }, "state": { "type": "object", "properties": { "current_task": {"type": "string"}, "emails_pending": {"type": "integer"}, "episode_done": {"type": "boolean"}, "steps_taken": {"type": "integer"}, "total_score": {"type": "number"}, }, }, } @app.post("/mcp") async def mcp_endpoint(request_body: dict = {}): """MCP JSON-RPC endpoint.""" method = request_body.get("method", "") req_id = request_body.get("id", 1) if method == "initialize": return { "jsonrpc": "2.0", "id": req_id, "result": { "protocolVersion": "2024-11-05", "serverInfo": {"name": "exec-assist", "version": "1.0.0"}, "capabilities": {"tools": {"listChanged": False}}, }, } elif method == "tools/list": return { "jsonrpc": "2.0", "id": req_id, "result": { "tools": [ { "name": "reset", "description": "Start new episode (easy/medium/hard)", "inputSchema": { "type": "object", "properties": {"task": {"type": "string", "enum": ["easy", "medium", "hard"]}}, }, }, { "name": "step", "description": "Submit email reply and calendar action", "inputSchema": { "type": "object", "properties": { "email_reply": {"type": "string"}, "calendar_action": {"type": "string"}, "meeting_details": {"type": "object"}, }, "required": ["email_reply", "calendar_action"], }, }, {"name": "state", "description": "Get current state", "inputSchema": {"type": "object"}}, ], }, } return {"jsonrpc": "2.0", "id": req_id, "result": {}} def main(): import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) if __name__ == "__main__": main()